risingwavelabs/risingwave · error · SinkError

Columns mismatch. RisingWave schema has

Error message

Columns mismatch. RisingWave schema has {} fields, DeltaLake schema has {} fields

What it means

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.

Solutions

  1. Align the sink query column list with the Delta table schema (select exactly the table's columns)
  2. Recreate/replace the Delta table to match the RisingWave schema
  3. Use a fresh location for the sink so a new table is created from the RisingWave schema
  4. Check for recent ALTER/evolution on the Delta table

Example fix

// before
CREATE SINK s FROM mv WITH (connector='deltalake', location='s3://bkt/existing_table')
-- existing_table has 5 cols, mv has 4
// after
CREATE SINK s FROM (SELECT a,b,c,d,e FROM mv) WITH (connector='deltalake', location='s3://bkt/existing_table')
Defensive patterns

Strategy: validation

Validate before calling

-- Compare column counts before sinking
SELECT count(*) FROM information_schema.columns WHERE table_name = 'mv';
-- vs. inspect the Delta table schema (e.g. via delta-rs or Spark DESCRIBE)

Prevention

When it happens

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

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

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/ec6b1f60d5979776. Report an issue: GitHub.

Appendix: source

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

    }

    async fn validate(&self) -> Result<()> {
        if self.config.r#type != SINK_TYPE_APPEND_ONLY
            && self.config.r#type != SINK_USER_FORCE_APPEND_ONLY_OPTION
        {
            return Err(SinkError::Config(anyhow!(
                "only append-only delta lake sink is supported",
            )));
        }
        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 {:?}",

View on GitHub (pinned to 6469eb736d)