{"record":{"id":"ae5335efefcc23e8","repo":"risingwavelabs/risingwave","slug":"invalid-schema-change-operation","errorCode":null,"errorMessage":"Invalid schema change operation","messagePattern":"Invalid schema change operation","errorType":"validation","errorClass":"SinkError::Coordinator","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/snowflake_redshift/redshift.rs","lineNumber":707,"sourceCode":"            }\n            tracing::info!(\n                \"Manifest file written to S3 for sink id {} at epoch {}\",\n                self.sink_id,\n                epoch\n            );\n        }\n        Ok(())\n    }\n\n    async fn commit_schema_change(\n        &mut self,\n        _epoch: u64,\n        schema_change: PbSinkSchemaChange,\n    ) -> Result<()> {\n        use risingwave_pb::stream_plan::sink_schema_change::PbOp as SinkSchemaChangeOp;\n        let schema_change_op = schema_change\n            .op\n            .ok_or_else(|| SinkError::Coordinator(anyhow!(\"Invalid schema change operation\")))?;\n        let SinkSchemaChangeOp::AddColumns(add_columns) = schema_change_op else {\n            return Err(SinkError::Coordinator(anyhow!(\n                \"Only AddColumns schema change is supported for Redshift sink\"\n            )));\n        };\n        if let Some(shutdown_sender) = &self.shutdown_sender {\n            // Send shutdown signal to the periodic task before altering the table\n            shutdown_sender\n                .send(())\n                .map_err(|e| SinkError::Config(anyhow!(e)))?;\n        }\n        let sql = build_alter_add_column_sql(\n            self.config.schema.as_deref(),\n            &self.config.table,\n            &add_columns\n                .fields\n                .iter()\n                .map(|f| {","sourceCodeStart":689,"sourceCodeEnd":725,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/snowflake_redshift/redshift.rs#L689-L725","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Set the `op` field on the `PbSinkSchemaChange` at the caller (e.g. `op: Some(PbOp::AddColumns(...))`) before dispatching the schema change.","Check which code path builds the schema-change message (frontend `ALTER` handling / meta coordinator) and ensure it produces a concrete op.","Upgrade or patch the code so unsupported/empty ops are rejected upstream instead of reaching the sink."],"exampleFix":"// before\nlet change = PbSinkSchemaChange { op: None };\n// after\nlet change = PbSinkSchemaChange {\n    op: Some(risingwave_pb::stream_plan::sink_schema_change::PbOp::AddColumns(\n        PbAddColumns { columns },\n    )),\n};","handlingStrategy":"validation","validationCode":"fn has_schema_change_op(change: &PbSinkSchemaChange) -> bool { change.op.is_some() }\nif !has_schema_change_op(&change) { return Err(anyhow!(\"schema change has no op\")); }","typeGuard":"fn as_add_columns(op: &Option<PbOp>) -> Option<&PbAddColumns> {\n    if let Some(PbOp::AddColumns(c)) = op { Some(c) } else { None }\n}","tryCatchPattern":null,"preventionTips":["Always construct PbSinkSchemaChange with an explicit op variant before dispatch.","Add a compile-time constructor helper that enforces op population."],"tags":["rust","sink","schema-change","protobuf"],"backgroundTag":"null-argument","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}