risingwavelabs/risingwave · error · SinkError::Coordinator

Invalid schema change operation

Error message

Invalid schema change operation

What it means

The Redshift sink's `commit_schema_change` requires a schema-change operation payload (`schema_change.op`) to be present, but the protobuf `PbSinkSchemaChange` arrived with `op` unset. RisingWave throws this as a Coordinator error because it cannot interpret an empty schema-change instruction. It is an internal invariant: callers should always populate `op` before invoking schema evolution on the sink.

Source

Thrown at src/connector/src/sink/snowflake_redshift/redshift.rs:707

            }
            tracing::info!(
                "Manifest file written to S3 for sink id {} at epoch {}",
                self.sink_id,
                epoch
            );
        }
        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 Redshift sink"
            )));
        };
        if let Some(shutdown_sender) = &self.shutdown_sender {
            // Send shutdown signal to the periodic task before altering the table
            shutdown_sender
                .send(())
                .map_err(|e| SinkError::Config(anyhow!(e)))?;
        }
        let sql = build_alter_add_column_sql(
            self.config.schema.as_deref(),
            &self.config.table,
            &add_columns
                .fields
                .iter()
                .map(|f| {

View on GitHub (pinned to 6469eb736d)

Solutions

1. Set the `op` field on the `PbSinkSchemaChange` at the caller (e.g. `op: Some(PbOp::AddColumns(...))`) before dispatching the schema change.
2. Check which code path builds the schema-change message (frontend `ALTER` handling / meta coordinator) and ensure it produces a concrete op.
3. Upgrade or patch the code so unsupported/empty ops are rejected upstream instead of reaching the sink.

Example fix

// before
let change = PbSinkSchemaChange { op: None };
// after
let change = PbSinkSchemaChange {
    op: Some(risingwave_pb::stream_plan::sink_schema_change::PbOp::AddColumns(
        PbAddColumns { columns },
    )),
};

fn has_schema_change_op(change: &PbSinkSchemaChange) -> bool { change.op.is_some() }
if !has_schema_change_op(&change) { return Err(anyhow!("schema change has no op")); }

fn as_add_columns(op: &Option<PbOp>) -> Option<&PbAddColumns> {
    if let Some(PbOp::AddColumns(c)) = op { Some(c) } else { None }
}

When it happens

Trigger: Calling `commit_schema_change` on the Redshift sink with a `PbSinkSchemaChange` whose oneof `op` field is `None`, e.g. an empty or default-constructed `SinkSchemaChange` message sent from the frontend/meta during schema evolution.

Common situations: Upstream code constructs a `SinkSchemaChange` but forgets to set the operation; a new schema-change op type was added but not plumbed through; a protobuf message round-trip drops the unset oneof field.

Related errors

• Only AddColumns schema change is supported for Redshift sink
• Invalid schema change operation
• query_epoch not set in distributed lookup join
• Failed to decode prost: field not found `{}`
• Invalid datetime: seconds: {secs}, nanoseconds: {nsecs}

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