What causes "column {} not found in the downstream SQL Server table {}" in risingwavelabs/risingwave?

ALTER TABLE on SQL Server to add the missing column (with a compatible type)

risingwavelabs/risingwave · error · SinkError::SqlServer

column not found in the downstream SQL Server table

Error message

column {} not found in the downstream SQL Server table {}

What it means

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.

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

Example fix

// before
CREATE SINK s AS SELECT id, name, email FROM mv INTO sqlserver...; -- 'email' not in table
// after
ALTER TABLE dbo.target ADD email NVARCHAR(255);
-- or
CREATE SINK s AS SELECT id, name FROM mv INTO sqlserver...;

Defensive patterns

Strategy: validation

Validate before calling

-- before creating the sink, verify every sink column exists downstream
SELECT c.name FROM sys.columns c
JOIN sys.tables t ON c.object_id = t.object_id
WHERE t.name = 'target'
-- compare result with the sink SELECT column list

Try / catch

match sink.validate().await {
    Err(e) if e.to_string().contains("not found in the downstream SQL Server table") => reconcile_schemas_and_recreate_sink(),
    other => other,
}

Prevention

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

When it happens

Trigger: 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).

Common situations: 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.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/6d9f9908f2c42447. Report an issue: GitHub.

Appendix: source

Thrown at src/connector/src/sink/sqlserver.rs:215

        validate_sql_server_write_permission(&mut sql_client, &self.config, self.is_append_only)
            .await?;
        let sql_server_table_metadata =
            query_sql_server_table_metadata(&mut sql_client, &self.config).await?;
        let sql_server_pk_count = sql_server_table_metadata
            .iter()
            .filter(|metadata| metadata.is_pk)
            .count();
        let sql_server_table_metadata = sql_server_table_metadata
            .into_iter()
            .map(|metadata| (metadata.name.clone(), metadata))
            .collect::<HashMap<_, _>>();

        // Validate Column name, Primary Key and data type.
        for (idx, col) in self.schema.fields().iter().enumerate() {
            let rw_is_pk = self.pk_indices.contains(&idx);
            match sql_server_table_metadata.get(&normalize_sql_server_column_name(&col.name)) {
                None => {
                    return Err(SinkError::SqlServer(anyhow!(format!(
                        "column {} not found in the downstream SQL Server table {}",
                        col.name,
                        self.config.full_object_path()
                    ))));
                }
                Some(sql_server_col) => {
                    validate_data_type_compatibility(
                        &col.name,
                        &col.data_type,
                        &sql_server_col.data_type,
                    )?;
                    if self.is_append_only {
                        continue;
                    }
                    if rw_is_pk && !sql_server_col.is_pk {
                        return Err(SinkError::SqlServer(anyhow!(format!(
                            "column {} specified in primary_key mismatches with the downstream SQL Server table {} PK",
                            col.name,

View on GitHub (pinned to 6469eb736d)