risingwavelabs/risingwave · error · anyhow::Error

iceberg sink: schema evolution not supported; expect schema

Error message

iceberg sink: schema evolution not supported; expect schema id {}, got {}

What it means

The Iceberg sink reloads the target table from the catalog before each commit and refuses to proceed if the table's current schema id differs from the schema id the sink was created with. This library deliberately does not implement schema evolution, so any external schema change (add/drop/alter column) invalidates the sink's cached schema and the commit aborts with this error. It is a safety guard to prevent writing data with a stale schema into an evolved table.

Source

Thrown at src/connector/src/sink/iceberg/commit_retry.rs:102

    Commit(anyhow::Error),
}

/// Reload the iceberg table from the catalog and assert that its current
/// `schema_id` and `default_partition_spec_id` still match the values the
/// caller computed against. Schema or partition evolution mid-commit is
/// surfaced as a non-retriable error by the call sites.
pub async fn reload_table(
    catalog: &dyn Catalog,
    table_ident: &TableIdent,
    schema_id: i32,
    partition_spec_id: i32,
) -> Result<Table> {
    let table = catalog
        .load_table(table_ident)
        .await
        .map_err(|e| anyhow!(e).context("reload iceberg table"))?;
    if table.metadata().current_schema_id() != schema_id {
        bail!(
            "iceberg sink: schema evolution not supported; expect schema id {}, got {}",
            schema_id,
            table.metadata().current_schema_id(),
        );
    }
    if table.metadata().default_partition_spec_id() != partition_spec_id {
        bail!(
            "iceberg sink: partition evolution not supported; expect partition spec id {}, got {}",
            partition_spec_id,
            table.metadata().default_partition_spec_id(),
        );
    }
    Ok(table)
}

/// Run a commit-action against the given iceberg table with retry.
/// 1. Calls `reload_table` before each commit attempt to get the latest metadata
/// 2. If `reload_table` fails (table not exists/schema/partition mismatch), stops retrying immediately

View on GitHub (pinned to 6469eb736d)

Solutions

1. Stop or recreate the RisingWave iceberg sink so it picks up the new schema id at creation time
2. Revert the external schema change (restore the original schema) so the table's current_schema_id matches the sink's expected id
3. Point the sink at a separate Iceberg table dedicated to RisingWave output to avoid external schema edits
4. Check catalog audit logs to find which process performed the schema change and coordinate schema evolution windows with sink maintenance

Example fix

// before: table evolved externally, sink commits keep failing
//   expect schema id 0, got 1
// after: recreate the sink against the current table schema
DROP SINK my_iceberg_sink;
CREATE SINK my_iceberg_sink AS
  SELECT ... FROM mv
  WITH (
    connector = 'iceberg',
    -- table now has the new column; sink binds to current schema id
    table.name = 'db.my_table'
  );

// Rust, before committing via the sink's retry loop
let table = catalog.load_table(&table_ident).await?;
if table.metadata().current_schema_id() != expected_schema_id {
    return Err(anyhow!(
        "iceberg table schema changed (expect {}, got {}); recreate the sink",
        expected_schema_id, table.metadata().current_schema_id()
    ));
}

fn schema_matches(table: &Table, expected_schema_id: i32) -> bool {
    table.metadata().current_schema_id() == expected_schema_id
}

match run_with_retry(...).await {
    Err(e) if e.to_string().contains("schema evolution not supported") => {
        // schema changed externally: recreate the sink, do not blind-retry
        recreate_sink().await?;
    }
    Err(e) => return Err(e),
    Ok(_) => {}
}

When it happens

Trigger: `reload_table(schema_id, partition_spec_id)` in commit_retry.rs detects `table.metadata().current_schema_id() != schema_id` during `run_with_retry` — i.e. an external actor (Spark, Flink, another engine, or manual ALTER via the catalog) changed the Iceberg table schema between sink creation and a commit attempt.

Common situations: A data engineering team evolves the Iceberg table (adds a column) while a RisingWave sink is still committing to it; concurrent pipelines owned by different tools write to the same table; the sink was pointed at a table that was dropped and recreated with a new schema (schema id resets/changes).

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• schema_id and partition_spec_id should be the same in all wr
• Unsupported sink schema change op in iceberg sink: {:?}
• Current iceberg schema does not match either original_schema
• iceberg sink: partition evolution not supported; expect part
• Invalid order key item `{item}`: `NULLS` must be followed by

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