{"record":{"id":"cf040a568354f923","repo":"risingwavelabs/risingwave","slug":"iceberg-sink-partition-evolution-not-supported-e","errorCode":null,"errorMessage":"iceberg sink: partition evolution not supported; expect partition spec id {}, got {}","messagePattern":"iceberg sink: partition evolution not supported; expect partition spec id (.+?), got (.+?)","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/commit_retry.rs","lineNumber":109,"sourceCode":"pub 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\n/// 3. If commit fails, retries with backoff up to `retry_num` times.\n///\n/// Strategy: exponential backoff 10ms→60s with jitter, up to `retry_num` retries.\npub async fn run_with_retry<F, Fut, Out>(\n    catalog: Arc<dyn Catalog>,\n    table_ident: TableIdent,\n    schema_id: i32,","sourceCodeStart":91,"sourceCodeEnd":127,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/commit_retry.rs#L91-L127","documentation":"Analogous to the schema check, the Iceberg sink compares the table's default partition spec id against the one recorded when the sink was created. If the partition spec changed externally (partition evolution, e.g. day->hour bucketing), the sink aborts the commit because this library does not support partition evolution. This prevents writing data files with an old partitioning layout into a table with a new spec.","triggerScenarios":"`reload_table` during `run_with_retry` finds `table.metadata().default_partition_spec_id() != partition_spec_id`, i.e. the table's default partition spec was changed by an external tool between sink creation and this commit.","commonSituations":"Someone re-partitions the Iceberg table with Spark (REPLACE ... PARTITION) or replaces the table with a different partitioning; a table drop-and-recreate resets/changes the spec id; concurrent writers from different engines disagree on partitioning.","solutions":["Recreate the RisingWave iceberg sink so it binds to the table's current partition spec id","Revert the external partition-spec change so default_partition_spec_id matches the sink's expected id","Point the sink at a dedicated table not touched by other engines' partition evolution","Coordinate partition changes with sink downtime windows"],"exampleFix":"// before: partition spec changed externally\n//   expect partition spec id 0, got 1\n// after: recreate the sink\nDROP SINK my_iceberg_sink;\nCREATE SINK my_iceberg_sink AS ... WITH (\n  connector = 'iceberg',\n  table.name = 'db.my_table'  -- binds to current partition spec\n);","handlingStrategy":"validation","validationCode":"let table = catalog.load_table(&table_ident).await?;\nif table.metadata().default_partition_spec_id() != expected_partition_spec_id {\n    return Err(anyhow!(\n        \"iceberg table partition spec changed (expect {}, got {}); recreate the sink\",\n        expected_partition_spec_id, table.metadata().default_partition_spec_id()\n    ));\n}","typeGuard":"fn partition_spec_matches(table: &Table, expected_spec_id: i32) -> bool {\n    table.metadata().default_partition_spec_id() == expected_spec_id\n}","tryCatchPattern":"match run_with_retry(...).await {\n    Err(e) if e.to_string().contains(\"partition evolution not supported\") => {\n        // partition spec changed externally: recreate the sink\n        recreate_sink().await?;\n    }\n    Err(e) => return Err(e),\n    Ok(_) => {}\n}","preventionTips":["Avoid REPLACE/REWRITE operations that change the partition spec on tables written by the sink","Use a dedicated table for RisingWave output","Coordinate partition re-bucketing with sink maintenance windows","Track default_partition_spec_id changes via catalog metadata monitoring"],"tags":["iceberg","partition-evolution","sink","rust"],"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-14T16:17:12.679Z"}