{"record":{"id":"163580c9501579f0","repo":"risingwavelabs/risingwave","slug":"schema-change-is-not-implemented-for-single-phase","errorCode":null,"errorMessage":"Schema change is not implemented for single-phase commit coordinator {}","messagePattern":"Schema change is not implemented for single-phase commit coordinator (.+?)","errorType":"exception","errorClass":"SinkError::Coordinator","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mod.rs","lineNumber":975,"sourceCode":"    TwoPhase(BoxTwoPhaseCoordinator),\n}\n\n#[async_trait]\npub trait SinglePhaseCommitCoordinator {\n    /// Initialize the sink committer coordinator.\n    async fn init(&mut self) -> Result<()>;\n\n    /// Commit data directly using single-phase strategy.\n    async fn commit_data(&mut self, epoch: u64, metadata: Vec<SinkMetadata>) -> Result<()>;\n\n    /// Idempotent implementation is required, because `commit_schema_change` in the same epoch could be called multiple\n    /// times.\n    async fn commit_schema_change(\n        &mut self,\n        _epoch: u64,\n        _schema_change: PbSinkSchemaChange,\n    ) -> Result<()> {\n        Err(SinkError::Coordinator(anyhow!(\n            \"Schema change is not implemented for single-phase commit coordinator {}\",\n            std::any::type_name::<Self>()\n        )))\n    }\n}\n\n#[async_trait]\npub trait TwoPhaseCommitCoordinator {\n    /// Initialize the sink committer coordinator.\n    async fn init(&mut self) -> Result<()>;\n\n    /// Return serialized commit metadata to be passed to `commit`.\n    async fn pre_commit(\n        &mut self,\n        epoch: u64,\n        metadata: Vec<SinkMetadata>,\n        schema_change: Option<PbSinkSchemaChange>,\n    ) -> Result<Option<Vec<u8>>>;","sourceCodeStart":957,"sourceCodeEnd":993,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mod.rs#L957-L993","documentation":"Default trait method `commit_schema_change` on the single-phase sink commit coordinator always fails with this error. It signals that the sink connector has not implemented schema-change handling (e.g. adding/replacing columns mid-flight) in its commit coordinator, so a schema evolution event cannot be committed for this sink type.","triggerScenarios":"A schema change event (`PbSinkSchemaChange`) is delivered to a sink's commit coordinator whose type only has the default `commit_schema_change` implementation; typically triggered when the upstream table/mv schema evolves while a sink with schema-change support expectations is running.","commonSituations":"Altering the source table (ADD COLUMN etc.) that a sink reads from, while the sink connector's coordinator lacks schema evolution support; enabling schema-change handling on sinks that don't implement it.","solutions":["Avoid schema changes on the upstream relation while this sink is active, or recreate the sink after the change","Use a connector whose commit coordinator implements `commit_schema_change` (e.g. Iceberg) if schema evolution is required","If authoring the connector, override `commit_schema_change` to handle the `PbSinkSchemaChange`"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// avoid schema changes on relations feeding sinks that lack schema-change support\n-- check sink type first: SELECT * FROM rw_sinks WHERE name = 'my_sink';","typeGuard":null,"tryCatchPattern":"match err {\n    SinkError::Coordinator(e) if e.to_string().contains(\"Schema change is not implemented\") => {\n        // recreate sink or switch connector; surface actionable message to user\n    }\n    e => return Err(e.into()),\n}","preventionTips":["Do not ALTER tables/mviews feeding sinks on connectors without schema evolution support","Plan schema migrations with sink recreation as a documented step"],"tags":["sink","schema-change","coordinator"],"backgroundTag":"method-not-implemented","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"}