{"record":{"id":"6d9f9908f2c42447","repo":"risingwavelabs/risingwave","slug":"column-not-found-in-the-downstream-sql-server-t","errorCode":null,"errorMessage":"column {} not found in the downstream SQL Server table {}","messagePattern":"column (.+?) not found in the downstream SQL Server table (.+?)","errorType":"validation","errorClass":"SinkError::SqlServer","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/sqlserver.rs","lineNumber":215,"sourceCode":"        validate_sql_server_write_permission(&mut sql_client, &self.config, self.is_append_only)\n            .await?;\n        let sql_server_table_metadata =\n            query_sql_server_table_metadata(&mut sql_client, &self.config).await?;\n        let sql_server_pk_count = sql_server_table_metadata\n            .iter()\n            .filter(|metadata| metadata.is_pk)\n            .count();\n        let sql_server_table_metadata = sql_server_table_metadata\n            .into_iter()\n            .map(|metadata| (metadata.name.clone(), metadata))\n            .collect::<HashMap<_, _>>();\n\n        // Validate Column name, Primary Key and data type.\n        for (idx, col) in self.schema.fields().iter().enumerate() {\n            let rw_is_pk = self.pk_indices.contains(&idx);\n            match sql_server_table_metadata.get(&normalize_sql_server_column_name(&col.name)) {\n                None => {\n                    return Err(SinkError::SqlServer(anyhow!(format!(\n                        \"column {} not found in the downstream SQL Server table {}\",\n                        col.name,\n                        self.config.full_object_path()\n                    ))));\n                }\n                Some(sql_server_col) => {\n                    validate_data_type_compatibility(\n                        &col.name,\n                        &col.data_type,\n                        &sql_server_col.data_type,\n                    )?;\n                    if self.is_append_only {\n                        continue;\n                    }\n                    if rw_is_pk && !sql_server_col.is_pk {\n                        return Err(SinkError::SqlServer(anyhow!(format!(\n                            \"column {} specified in primary_key mismatches with the downstream SQL Server table {} PK\",\n                            col.name,","sourceCodeStart":197,"sourceCodeEnd":233,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/sqlserver.rs#L197-L233","documentation":"During validate(), every column of the RisingWave sink schema is looked up (case-normalized) in the downstream SQL Server table's metadata. If a sink column does not exist in the actual SQL Server table, the sink fails with this error naming the column and target table path.","triggerScenarios":"Creating a SQL Server sink whose schema contains a column absent from the pre-created downstream table (checked via sql_server_table_metadata lookup after normalize_sql_server_column_name).","commonSituations":"Downstream table was created/modified after writing the CREATE SINK statement; column renamed in SQL Server; case-sensitivity mismatch beyond normalization; extra SELECT column added to the sink query.","solutions":["ALTER TABLE on SQL Server to add the missing column (with a compatible type)","Change the sink query to only select columns that exist in the downstream table","Drop and recreate the sink after aligning schemas; verify column names and case match"],"exampleFix":"// before\nCREATE SINK s AS SELECT id, name, email FROM mv INTO sqlserver...; -- 'email' not in table\n// after\nALTER TABLE dbo.target ADD email NVARCHAR(255);\n-- or\nCREATE SINK s AS SELECT id, name FROM mv INTO sqlserver...;","handlingStrategy":"validation","validationCode":"-- before creating the sink, verify every sink column exists downstream\nSELECT c.name FROM sys.columns c\nJOIN sys.tables t ON c.object_id = t.object_id\nWHERE t.name = 'target'\n-- compare result with the sink SELECT column list","typeGuard":null,"tryCatchPattern":"match sink.validate().await {\n    Err(e) if e.to_string().contains(\"not found in the downstream SQL Server table\") => reconcile_schemas_and_recreate_sink(),\n    other => other,\n}","preventionTips":["Create the sink query from the actual downstream table's column list","Re-validate after any SQL Server table DDL change","Use consistent naming/casing conventions between RW and SQL Server"],"tags":["sqlserver","sink","schema-mismatch","validation"],"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"}