risingwavelabs/risingwave · error · SinkError::Config

Snowflake sink committer is not initialized.

Error message

Snowflake sink committer is not initialized.

What it means

`commit_schema_change` requires the sink's JDBC `client` (committer) to have been initialized. When `self.client` is `None` — e.g. the sink instance is not the designated committer or initialization failed — it returns this Config error before executing the ALTER.

Source

Thrown at src/connector/src/sink/snowflake_redshift/snowflake.rs:799

        Ok(())
    }

    async fn commit_schema_change(
        &mut self,
        _epoch: u64,
        schema_change: PbSinkSchemaChange,
    ) -> Result<()> {
        use risingwave_pb::stream_plan::sink_schema_change::PbOp as SinkSchemaChangeOp;
        let schema_change_op = schema_change
            .op
            .ok_or_else(|| SinkError::Coordinator(anyhow!("Invalid schema change operation")))?;
        let SinkSchemaChangeOp::AddColumns(add_columns) = schema_change_op else {
            return Err(SinkError::Coordinator(anyhow!(
                "Only AddColumns schema change is supported for Snowflake sink"
            )));
        };
        let client = self.client.as_mut().ok_or_else(|| {
            SinkError::Config(anyhow!("Snowflake sink committer is not initialized."))
        })?;
        client
            .execute_alter_add_columns(
                &add_columns
                    .fields
                    .into_iter()
                    .map(|f| {
                        let dt = DataType::from(f.data_type.unwrap());
                        Ok((f.name, convert_snowflake_data_type(&dt)?))
                    })
                    .collect::<Result<Vec<_>>>()?,
            )
            .await
    }
}

impl Drop for SnowflakeSinkCommitter {
    fn drop(&mut self) {

View on GitHub (pinned to 6469eb736d)

Solutions

1. Ensure schema changes are applied through the sink's committer instance where `client` is initialized
2. Check earlier logs for JDBC client initialization failures and fix connectivity/config first
3. Restart/resync the sink so the committer re-initializes its client

Example fix

// before: calling commit_schema_change on a plain writer
writer.commit_schema_change(epoch, change).await?; // client is None
// after: route through the committer that owns the initialized JDBC client
committer.commit_schema_change(epoch, change).await?;

if sink.client().is_none() {
    return Err("snowflake committer not initialized; route schema changes to the committer");
}

fn committer_ready(sink: &SnowflakeSinkCommitter) -> bool { sink.client.is_some() }

When it happens

Trigger: Calling `commit_schema_change` on a Snowflake sink instance whose `client: Option<JdbcClient>` was never populated (non-committer writer instance, or earlier client initialization failure).

Common situations: Schema change routed to a writer instance instead of the committer; JDBC client initialization failure earlier swallowed; S3 mode where the JDBC client lifecycle differs.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Invalid schema change operation
• Only AddColumns schema change is supported for Snowflake sin
• table {} is not active
• feature {feature:?} is not available based on your license
• Snowflake catalog only supports iceberg sources

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