{"record":{"id":"441f6ccc8475cac0","repo":"risingwavelabs/risingwave","slug":"iceberg-sink-schema-evolution-not-supported-expe","errorCode":null,"errorMessage":"iceberg sink: schema evolution not supported; expect schema id {}, got {}","messagePattern":"iceberg sink: schema evolution not supported; expect schema id (.+?), got (.+?)","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/commit_retry.rs","lineNumber":102,"sourceCode":"    Commit(anyhow::Error),\n}\n\n/// Reload the iceberg table from the catalog and assert that its current\n/// `schema_id` and `default_partition_spec_id` still match the values the\n/// caller computed against. Schema or partition evolution mid-commit is\n/// surfaced as a non-retriable error by the call sites.\npub async fn reload_table(\n    catalog: &dyn Catalog,\n    table_ident: &TableIdent,\n    schema_id: i32,\n    partition_spec_id: i32,\n) -> Result<Table> {\n    let table = catalog\n        .load_table(table_ident)\n        .await\n        .map_err(|e| anyhow!(e).context(\"reload iceberg table\"))?;\n    if table.metadata().current_schema_id() != schema_id {\n        bail!(\n            \"iceberg sink: schema evolution not supported; expect schema id {}, got {}\",\n            schema_id,\n            table.metadata().current_schema_id(),\n        );\n    }\n    if table.metadata().default_partition_spec_id() != partition_spec_id {\n        bail!(\n            \"iceberg sink: partition evolution not supported; expect partition spec id {}, got {}\",\n            partition_spec_id,\n            table.metadata().default_partition_spec_id(),\n        );\n    }\n    Ok(table)\n}\n\n/// Run a commit-action against the given iceberg table with retry.\n/// 1. Calls `reload_table` before each commit attempt to get the latest metadata\n/// 2. If `reload_table` fails (table not exists/schema/partition mismatch), stops retrying immediately","sourceCodeStart":84,"sourceCodeEnd":120,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/commit_retry.rs#L84-L120","documentation":"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.","triggerScenarios":"`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.","commonSituations":"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).","solutions":["Stop or recreate the RisingWave iceberg sink so it picks up the new schema id at creation time","Revert the external schema change (restore the original schema) so the table's current_schema_id matches the sink's expected id","Point the sink at a separate Iceberg table dedicated to RisingWave output to avoid external schema edits","Check catalog audit logs to find which process performed the schema change and coordinate schema evolution windows with sink maintenance"],"exampleFix":"// before: table evolved externally, sink commits keep failing\n//   expect schema id 0, got 1\n// after: recreate the sink against the current table schema\nDROP SINK my_iceberg_sink;\nCREATE SINK my_iceberg_sink AS\n  SELECT ... FROM mv\n  WITH (\n    connector = 'iceberg',\n    -- table now has the new column; sink binds to current schema id\n    table.name = 'db.my_table'\n  );","handlingStrategy":"validation","validationCode":"// Rust, before committing via the sink's retry loop\nlet table = catalog.load_table(&table_ident).await?;\nif table.metadata().current_schema_id() != expected_schema_id {\n    return Err(anyhow!(\n        \"iceberg table schema changed (expect {}, got {}); recreate the sink\",\n        expected_schema_id, table.metadata().current_schema_id()\n    ));\n}","typeGuard":"fn schema_matches(table: &Table, expected_schema_id: i32) -> bool {\n    table.metadata().current_schema_id() == expected_schema_id\n}","tryCatchPattern":"match run_with_retry(...).await {\n    Err(e) if e.to_string().contains(\"schema evolution not supported\") => {\n        // schema changed externally: recreate the sink, do not blind-retry\n        recreate_sink().await?;\n    }\n    Err(e) => return Err(e),\n    Ok(_) => {}\n}","preventionTips":["Do not let external engines (Spark/Flink) ALTER the schema of a table a RisingWave sink writes to","Dedicate an Iceberg table to the RisingWave sink output","Freeze schema changes during sink commit windows; evolve schema only during planned sink recreation","Monitor the table's current_schema_id in the catalog and alert on changes"],"tags":["iceberg","schema-evolution","sink","rust","config-conflict"],"backgroundTag":"schema-validation-failed","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}