{"record":{"id":"685d7b548d4d5685","repo":"risingwavelabs/risingwave","slug":"invalid-schema-change-operation-685d7b","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/snowflake.rs","lineNumber":792,"sourceCode":"            client.execute_create_pipe().await?;\n            client.execute_create_merge_into_task().await?;\n        }\n        Ok(())\n    }\n\n    async fn commit_data(&mut self, _epoch: u64, _metadata: Vec<SinkMetadata>) -> Result<()> {\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 Snowflake sink\"\n            )));\n        };\n        let client = self.client.as_mut().ok_or_else(|| {\n            SinkError::Config(anyhow!(\"Snowflake sink committer is not initialized.\"))\n        })?;\n        client\n            .execute_alter_add_columns(\n                &add_columns\n                    .fields\n                    .into_iter()\n                    .map(|f| {\n                        let dt = DataType::from(f.data_type.unwrap());\n                        Ok((f.name, convert_snowflake_data_type(&dt)?))\n                    })\n                    .collect::<Result<Vec<_>>>()?,","sourceCodeStart":774,"sourceCodeEnd":810,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/snowflake_redshift/snowflake.rs#L774-L810","documentation":"`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.","triggerScenarios":"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).","commonSituations":"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.","solutions":["Ensure the component producing `SinkSchemaChange` sets a valid `op` (currently only AddColumns)","Upgrade meta/frontend so schema changes are built with the op populated","If the event is corrupt, recreate the sink rather than retrying the same event"],"exampleFix":"// before\nlet change = PbSinkSchemaChange { op: None, ..Default::default() };\n// after\nlet change = PbSinkSchemaChange {\n    op: Some(PbOp::AddColumns(PbAddColumns { fields: new_fields })),\n    ..Default::default()\n};","handlingStrategy":"try-catch","validationCode":"if schema_change.op.is_none() {\n    return Err(\"schema change has no op set\");\n}","typeGuard":"fn has_op(c: &PbSinkSchemaChange) -> bool { c.op.is_some() }","tryCatchPattern":"match sink.commit_schema_change(epoch, change).await {\n    Err(e) if e.to_string().contains(\"Invalid schema change\") => {\n        log::warn!(\"dropping corrupt schema change: {e}\");\n    }\n    r => r?,\n}","preventionTips":["Always construct SinkSchemaChange with an explicit op","Keep meta/frontend and connector versions aligned"],"tags":["snowflake","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"}