{"record":{"id":"ef2255fb23e3b6d7","repo":"risingwavelabs/risingwave","slug":"only-addcolumns-schema-change-is-supported-for-sno","errorCode":null,"errorMessage":"Only AddColumns schema change is supported for Snowflake sink","messagePattern":"Only AddColumns schema change is supported for Snowflake sink","errorType":"validation","errorClass":"SinkError::Coordinator","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/snowflake_redshift/snowflake.rs","lineNumber":794,"sourceCode":"        }\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<_>>>()?,\n            )\n            .await","sourceCodeStart":776,"sourceCodeEnd":812,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/snowflake_redshift/snowflake.rs#L776-L812","documentation":"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.","triggerScenarios":"`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).","commonSituations":"Dropping or altering columns upstream and expecting the sink to propagate; newer meta emitting ops the sink code cannot handle (version skew).","solutions":["Restrict upstream DDL on the sink's source/table to ADD COLUMN operations","Recreate the sink if an unsupported change (drop/alter) is needed","Upgrade RisingWave if support for other ops has been added in a newer release"],"exampleFix":"-- before (unsupported)\nALTER TABLE mv DROP COLUMN extra_col; -- sink cannot propagate\n-- after (supported)\nALTER TABLE mv ADD COLUMN new_col INT;","handlingStrategy":"validation","validationCode":"let supported = matches!(schema_change.op, Some(PbOp::AddColumns(_)));\nif !supported { return Err(\"only AddColumns is supported for Snowflake sink\"); }","typeGuard":"fn is_add_columns(c: &PbSinkSchemaChange) -> bool {\n    matches!(c.op, Some(risingwave_pb::stream_plan::sink_schema_change::PbOp::AddColumns(_)))\n}","tryCatchPattern":null,"preventionTips":["Limit upstream DDL to ADD COLUMN while a Snowflake sink is attached","Handle drop/alter changes by recreating the sink instead"],"tags":["snowflake","sink","schema-change","unsupported"],"backgroundTag":"unsupported-operation","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"}