risingwavelabs/risingwave · error · SinkError::Coordinator

Only AddColumns schema change is supported for Snowflake sin

Error message

Only AddColumns schema change is supported for Snowflake sink

What it means

Guard in SnowflakeSinkWriter::commit_schema_change: RisingWave only propagates AddColumns schema changes to the Snowflake target; any other operation kind in the schema-change payload is rejected with this coordinator error. Fires when an evolving downstream schema change other than adding columns reaches the sink.

Source

Thrown at src/connector/src/sink/snowflake_redshift/snowflake.rs:794

        }
        Ok(())
    }

    async fn commit_data(&mut self, _epoch: u64, _metadata: Vec<SinkMetadata>) -> Result<()> {
        Ok(())
    }

    async fn commit_schema_change(
        &mut self,
        _epoch: u64,
        schema_change: PbSinkSchemaChange,
    ) -> Result<()> {
        use risingwave_pb::stream_plan::sink_schema_change::PbOp as SinkSchemaChangeOp;
        let schema_change_op = schema_change
            .op
            .ok_or_else(|| SinkError::Coordinator(anyhow!("Invalid schema change operation")))?;
        let SinkSchemaChangeOp::AddColumns(add_columns) = schema_change_op else {
            return Err(SinkError::Coordinator(anyhow!(
                "Only AddColumns schema change is supported for Snowflake sink"
            )));
        };
        let client = self.client.as_mut().ok_or_else(|| {
            SinkError::Config(anyhow!("Snowflake sink committer is not initialized."))
        })?;
        client
            .execute_alter_add_columns(
                &add_columns
                    .fields
                    .into_iter()
                    .map(|f| {
                        let dt = DataType::from(f.data_type.unwrap());
                        Ok((f.name, convert_snowflake_data_type(&dt)?))
                    })
                    .collect::<Result<Vec<_>>>()?,
            )
            .await

View on GitHub (pinned to 6469eb736d)

Solutions

1. Restrict upstream DDL on the sink's source/table to ADD COLUMN operations
2. Recreate the sink if an unsupported change (drop/alter) is needed
3. Upgrade RisingWave if support for other ops has been added in a newer release

Example fix

-- before (unsupported)
ALTER TABLE mv DROP COLUMN extra_col; -- sink cannot propagate
-- after (supported)
ALTER TABLE mv ADD COLUMN new_col INT;

let supported = matches!(schema_change.op, Some(PbOp::AddColumns(_)));
if !supported { return Err("only AddColumns is supported for Snowflake sink"); }

fn is_add_columns(c: &PbSinkSchemaChange) -> bool {
    matches!(c.op, Some(risingwave_pb::stream_plan::sink_schema_change::PbOp::AddColumns(_)))
}

When it happens

Trigger: `commit_schema_change` receiving a schema change whose op is a variant other than `AddColumns` (e.g. DropColumns, AlterColumnType as the pb feature set grows).

Common situations: Dropping or altering columns upstream and expecting the sink to propagate; newer meta emitting ops the sink code cannot handle (version skew).

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

• Snowflake catalog only supports iceberg sources
• LanceDB sink does not support schema change
• Invalid schema change operation
• Snowflake sink committer is not initialized.
• feature {feature:?} is not available based on your license

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