risingwavelabs/risingwave · error · SinkError::Coordinator

Invalid schema change operation

Error message

Invalid schema change operation

What it means

`commit_schema_change` handles sink schema evolution; the protobuf `PbSinkSchemaChange` must carry an `op`. When `schema_change.op` is None, the committer cannot interpret the change and returns this Coordinator error.

Source

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

            client.execute_create_pipe().await?;
            client.execute_create_merge_into_task().await?;
        }
        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<_>>>()?,

View on GitHub (pinned to 6469eb736d)

Solutions

  1. Ensure the component producing `SinkSchemaChange` sets a valid `op` (currently only AddColumns)
  2. Upgrade meta/frontend so schema changes are built with the op populated
  3. If the event is corrupt, recreate the sink rather than retrying the same event

Example fix

// before
let change = PbSinkSchemaChange { op: None, ..Default::default() };
// after
let change = PbSinkSchemaChange {
    op: Some(PbOp::AddColumns(PbAddColumns { fields: new_fields })),
    ..Default::default()
};
Defensive patterns

Strategy: try-catch

Validate before calling

if schema_change.op.is_none() {
    return Err("schema change has no op set");
}

Type guard

fn has_op(c: &PbSinkSchemaChange) -> bool { c.op.is_some() }

Try / catch

match sink.commit_schema_change(epoch, change).await {
    Err(e) if e.to_string().contains("Invalid schema change") => {
        log::warn!("dropping corrupt schema change: {e}");
    }
    r => r?,
}

Prevention

When it happens

Trigger: Calling `commit_schema_change` with a `PbSinkSchemaChange` message whose `op` field was never set (an empty/invalid schema change from the stream plan or meta coordinator).

Common situations: Meta node emitting a schema-change event with no op set due to a bug or partial protobuf construction; manually replaying messages; version skew between frontend/meta producing old pb messages.

Related errors

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