risingwavelabs/risingwave · error · SinkError::Coordinator

Only AddColumns schema change is supported for Snowflake…

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.

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

• Invalid schema change operation
• LanceDB sink does not support schema change
• Snowflake catalog only supports iceberg sources
• Snowflake sink committer is not initialized.
• ambiguous auth: multiple auth options provided; remove one…

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

Appendix: 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)