{"record":{"id":"ec6b1f60d5979776","repo":"risingwavelabs/risingwave","slug":"columns-mismatch-risingwave-schema-has-fields","errorCode":null,"errorMessage":"Columns mismatch. RisingWave schema has {} fields, DeltaLake schema has {} fields","messagePattern":"Columns mismatch\\. RisingWave schema has (.+?) fields, DeltaLake schema has (.+?) fields","errorType":"validation","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/deltalake.rs","lineNumber":438,"sourceCode":"    }\n\n    async fn validate(&self) -> Result<()> {\n        if self.config.r#type != SINK_TYPE_APPEND_ONLY\n            && self.config.r#type != SINK_USER_FORCE_APPEND_ONLY_OPTION\n        {\n            return Err(SinkError::Config(anyhow!(\n                \"only append-only delta lake sink is supported\",\n            )));\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 {:?}\",","sourceCodeStart":420,"sourceCodeEnd":456,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/deltalake.rs#L420-L456","documentation":"During validate, the DeltaLake table's schema field count is compared to the RisingWave sink schema field count. A mismatch means the existing Delta table and the sink's RisingWave schema describe different column sets, so validation aborts with this DeltaLake error.","triggerScenarios":"`CREATE SINK` into an existing DeltaLake table whose field count differs from the sink query's schema — e.g. extra or missing columns in the table, schema evolution on the table, or a SELECT that projects fewer/more columns.","commonSituations":"Pointing a sink at a pre-existing Delta table created by another pipeline; the table gained columns via schema evolution; sink query uses SELECT * on a table that changed shape since planning.","solutions":["Align the sink query column list with the Delta table schema (select exactly the table's columns)","Recreate/replace the Delta table to match the RisingWave schema","Use a fresh location for the sink so a new table is created from the RisingWave schema","Check for recent ALTER/evolution on the Delta table"],"exampleFix":"// before\nCREATE SINK s FROM mv WITH (connector='deltalake', location='s3://bkt/existing_table')\n-- existing_table has 5 cols, mv has 4\n// after\nCREATE SINK s FROM (SELECT a,b,c,d,e FROM mv) WITH (connector='deltalake', location='s3://bkt/existing_table')","handlingStrategy":"validation","validationCode":"-- Compare column counts before sinking\nSELECT count(*) FROM information_schema.columns WHERE table_name = 'mv';\n-- vs. inspect the Delta table schema (e.g. via delta-rs or Spark DESCRIBE)","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Match the sink SELECT column list to the existing Delta table exactly","Prefer sinking to a fresh location and letting RisingWave create the table","Re-check table schema after any external schema evolution"],"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"}