risingwavelabs/risingwave · error · SinkError::Coordinator

Schema change is not implemented for single-phase commit coo

Error message

Schema change is not implemented for single-phase commit coordinator {}

What it means

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.

Source

Thrown at src/connector/src/sink/mod.rs:975

    TwoPhase(BoxTwoPhaseCoordinator),
}

#[async_trait]
pub trait SinglePhaseCommitCoordinator {
    /// Initialize the sink committer coordinator.
    async fn init(&mut self) -> Result<()>;

    /// Commit data directly using single-phase strategy.
    async fn commit_data(&mut self, epoch: u64, metadata: Vec<SinkMetadata>) -> Result<()>;

    /// Idempotent implementation is required, because `commit_schema_change` in the same epoch could be called multiple
    /// times.
    async fn commit_schema_change(
        &mut self,
        _epoch: u64,
        _schema_change: PbSinkSchemaChange,
    ) -> Result<()> {
        Err(SinkError::Coordinator(anyhow!(
            "Schema change is not implemented for single-phase commit coordinator {}",
            std::any::type_name::<Self>()
        )))
    }
}

#[async_trait]
pub trait TwoPhaseCommitCoordinator {
    /// Initialize the sink committer coordinator.
    async fn init(&mut self) -> Result<()>;

    /// Return serialized commit metadata to be passed to `commit`.
    async fn pre_commit(
        &mut self,
        epoch: u64,
        metadata: Vec<SinkMetadata>,
        schema_change: Option<PbSinkSchemaChange>,
    ) -> Result<Option<Vec<u8>>>;

View on GitHub (pinned to 6469eb736d)

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`

Defensive patterns

Strategy: try-catch

Validate before calling

// avoid schema changes on relations feeding sinks that lack schema-change support
-- check sink type first: SELECT * FROM rw_sinks WHERE name = 'my_sink';

Try / catch

match err {
    SinkError::Coordinator(e) if e.to_string().contains("Schema change is not implemented") => {
        // recreate sink or switch connector; surface actionable message to user
    }
    e => return Err(e.into()),
}

Prevention

Do not ALTER tables/mviews feeding sinks on connectors without schema evolution support
Plan schema migrations with sink recreation as a documented step

When it happens

Trigger: 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.

Common situations: 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.

Related errors

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