{"record":{"id":"869b71091fda5f7b","repo":"risingwavelabs/risingwave","slug":"column-not-found-in-deltalake-table","errorCode":null,"errorMessage":"column {} not found in deltalake table","messagePattern":"column (.+?) not found in deltalake table","errorType":"validation","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/deltalake.rs","lineNumber":446,"sourceCode":"            )));\n        }\n        let table = self.config.common.create_deltalake_client().await?;\n        let snapshot = table.snapshot()?;\n        let delta_schema = snapshot.schema();\n        let deltalake_fields: HashMap<&String, &DeltaLakeDataType> = delta_schema\n            .fields()\n            .map(|f| (f.name(), f.data_type()))\n            .collect();\n        if deltalake_fields.len() != self.param.schema().fields().len() {\n            return Err(SinkError::DeltaLake(anyhow!(\n                \"Columns mismatch. RisingWave schema has {} fields, DeltaLake schema has {} fields\",\n                self.param.schema().fields().len(),\n                deltalake_fields.len()\n            )));\n        }\n        for field in self.param.schema().fields() {\n            if !deltalake_fields.contains_key(&field.name) {\n                return Err(SinkError::DeltaLake(anyhow!(\n                    \"column {} not found in deltalake table\",\n                    field.name\n                )));\n            }\n            let deltalake_field_type = deltalake_fields.get(&field.name).ok_or_else(|| {\n                SinkError::DeltaLake(anyhow!(\"cannot find field type for {}\", field.name))\n            })?;\n            if !check_field_type(&field.data_type, deltalake_field_type)? {\n                return Err(SinkError::DeltaLake(anyhow!(\n                    \"column '{}' type mismatch: deltalake type is {:?}, RisingWave type is {:?}\",\n                    field.name,\n                    deltalake_field_type,\n                    field.data_type\n                )));\n            }\n        }\n        if self.config.common.commit_checkpoint_interval == 0 {\n            return Err(SinkError::Config(anyhow!(","sourceCodeStart":428,"sourceCodeEnd":464,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/deltalake.rs#L428-L464","documentation":"For each RisingWave sink field, validate checks that a column of the same name exists in the DeltaLake table schema. If a sink field's name has no counterpart in the table, this DeltaLake error names the missing column.","triggerScenarios":"Sink schema and Delta table have equal field counts (passing the earlier check) but names differ — e.g. column renamed in RisingWave, aliases in the SELECT differing from the table's column names, or case-sensitive name mismatch.","commonSituations":"Renamed columns in the source materialized view; SELECT with aliases that don't match the Delta table; case sensitivity differences between RisingWave and Delta column names; pointing sink at a table created with different column names.","solutions":["Alias the sink query columns to match the Delta table's column names exactly","Drop and recreate the Delta table from the RisingWave schema","Verify column name spelling and casing on both sides","Use a new sink location so the table is created with the current schema"],"exampleFix":"// before\nCREATE SINK s FROM (SELECT user_id AS uid, name FROM mv) WITH (connector='deltalake', location='s3://bkt/t')\n-- t has column 'user_id'\n// after\nCREATE SINK s FROM (SELECT user_id, name FROM mv) WITH (connector='deltalake', location='s3://bkt/t')","handlingStrategy":"validation","validationCode":"// Ensure sink column names match the Delta table\n-- SELECT with explicit aliases:\n// SELECT col1 AS \"table_col1\", col2 AS \"table_col2\" FROM mv","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Alias sink columns to the Delta table's exact names and casing","Avoid renaming columns in upstream materialized views feeding sinks","Diff RisingWave schema vs Delta schema before creating/updating sinks"],"tags":["deltalake","schema","validation","columns"],"backgroundTag":"schema-validation-failed","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}