risingwavelabs/risingwave · error · SinkError

column {} not found in deltalake table

Error message

column {} not found in deltalake table

What it means

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.

Source

Thrown at src/connector/src/sink/deltalake.rs:446

            )));
        }
        let table = self.config.common.create_deltalake_client().await?;
        let snapshot = table.snapshot()?;
        let delta_schema = snapshot.schema();
        let deltalake_fields: HashMap<&String, &DeltaLakeDataType> = delta_schema
            .fields()
            .map(|f| (f.name(), f.data_type()))
            .collect();
        if deltalake_fields.len() != self.param.schema().fields().len() {
            return Err(SinkError::DeltaLake(anyhow!(
                "Columns mismatch. RisingWave schema has {} fields, DeltaLake schema has {} fields",
                self.param.schema().fields().len(),
                deltalake_fields.len()
            )));
        }
        for field in self.param.schema().fields() {
            if !deltalake_fields.contains_key(&field.name) {
                return Err(SinkError::DeltaLake(anyhow!(
                    "column {} not found in deltalake table",
                    field.name
                )));
            }
            let deltalake_field_type = deltalake_fields.get(&field.name).ok_or_else(|| {
                SinkError::DeltaLake(anyhow!("cannot find field type for {}", field.name))
            })?;
            if !check_field_type(&field.data_type, deltalake_field_type)? {
                return Err(SinkError::DeltaLake(anyhow!(
                    "column '{}' type mismatch: deltalake type is {:?}, RisingWave type is {:?}",
                    field.name,
                    deltalake_field_type,
                    field.data_type
                )));
            }
        }
        if self.config.common.commit_checkpoint_interval == 0 {
            return Err(SinkError::Config(anyhow!(

View on GitHub (pinned to 6469eb736d)

Solutions

1. Alias the sink query columns to match the Delta table's column names exactly
2. Drop and recreate the Delta table from the RisingWave schema
3. Verify column name spelling and casing on both sides
4. Use a new sink location so the table is created with the current schema

Example fix

// before
CREATE SINK s FROM (SELECT user_id AS uid, name FROM mv) WITH (connector='deltalake', location='s3://bkt/t')
-- t has column 'user_id'
// after
CREATE SINK s FROM (SELECT user_id, name FROM mv) WITH (connector='deltalake', location='s3://bkt/t')

// Ensure sink column names match the Delta table
-- SELECT with explicit aliases:
// SELECT col1 AS "table_col1", col2 AS "table_col2" FROM mv

When it happens

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

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

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

• Columns mismatch. RisingWave schema has {} fields, DeltaLake
• Type {:?} is not supported for DeltaLake sink.
• column '{}' type mismatch: deltalake type is {:?}, RisingWav
• HTTP sink with multiple columns only supports payload and ur
• schema invalid, record type required at top level of the sch

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