{"record":{"id":"0c3987900993a580","repo":"risingwavelabs/risingwave","slug":"column-type-mismatch-deltalake-type-is","errorCode":null,"errorMessage":"column '{}' type mismatch: deltalake type is {:?}, RisingWave type is {:?}","messagePattern":"column '(.+?)' type mismatch: deltalake type is (.+?), RisingWave type is (.+?)","errorType":"validation","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/deltalake.rs","lineNumber":455,"sourceCode":"        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!(\n                \"`commit_checkpoint_interval` must be greater than 0\"\n            )));\n        }\n        Ok(())\n    }\n\n    fn is_coordinated_sink(&self) -> bool {\n        true\n    }","sourceCodeStart":437,"sourceCodeEnd":473,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/deltalake.rs#L437-L473","documentation":"After locating the column in both schemas, `check_field_type` compares the RisingWave data type to the DeltaLake data type. If they are not compatible (recursively for structs/lists), validation fails with this error reporting both types and the column name.","triggerScenarios":"CREATE SINK into an existing DeltaLake table where a same-named column has an incompatible type — e.g. RisingWave INTEGER vs Delta long, VARCHAR vs struct, or a nested struct/list with a differing element type.","commonSituations":"Sinking into a table created by another system with wider/different types; RisingWave type changed upstream; a nested struct field added on the RisingWave side but absent in the Delta struct type.","solutions":["Cast the offending column in the sink query to match the Delta table type (e.g. col::VARCHAR)","Recreate the Delta table with types matching the RisingWave schema","Check nested struct/list element types match on both sides","Review the supported type mapping between RisingWave and DeltaLake"],"exampleFix":"// before\nCREATE SINK s FROM (SELECT id, ts FROM mv) -- ts is TIMESTAMP, table column is STRING\n// after\nCREATE SINK s FROM (SELECT id, ts::VARCHAR AS ts FROM mv)","handlingStrategy":"validation","validationCode":"-- Cast columns to match Delta table types before sinking\n-- SELECT id, amount::DECIMAL AS amount, created_at::VARCHAR AS created_at FROM mv","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Map RisingWave types to Delta types explicitly in your pipeline design","Cast with :: in the sink SELECT when the table pre-exists with different types","Verify nested struct/list element types align on both sides"],"tags":["deltalake","schema","type-mismatch","validation"],"backgroundTag":"type-mismatch","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"}