{"record":{"id":"3b86a37c07332c17","repo":"risingwavelabs/risingwave","slug":"reload-iceberg-table-3b86a3","errorCode":null,"errorMessage":"reload iceberg table","messagePattern":"reload iceberg table","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/commit_retry.rs","lineNumber":100,"sourceCode":"    /// `Transaction::commit` (or its `apply`) failed. Retriable — likely a\n    /// commit conflict or transient network error.\n    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.","sourceCodeStart":82,"sourceCodeEnd":118,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/commit_retry.rs#L82-L118","documentation":"The commit_retry helper `reload_table` reloads the table from the catalog and then asserts the current schema id equals the schema id the data files were written with; a mismatch means schema evolution happened externally and iceberg sink commits with the old schema are not allowed. The load itself or the schema check failing surfaces through this function in run_with_retry.","triggerScenarios":"Another writer (Spark/Flink/manual DDL) added/changed columns on the Iceberg table after the RisingWave sink cached its schema id, so `table.metadata().current_schema_id() != schema_id`; also triggered when load_table itself fails (catalog outage, table dropped).","commonSituations":"Concurrent schema evolution by an external engine; a RisingWave schema-change path updated the table but the sink retry still carries the old schema id; migration of catalogs yielding different schema ids.","solutions":["Restart or recreate the RisingWave sink so it picks up the table's new current schema id.","Ensure schema changes flow only through RisingWave's sink schema-change path so ids stay in sync.","If load_table failed, fix catalog connectivity/credentials and retry.","Coordinate external writers to not evolve the schema while the sink is running."],"exampleFix":"// before\nif table.metadata().current_schema_id() != schema_id {\n    bail!(\"iceberg sink: schema evolution not supported; expect schema id {}, got {}\", schema_id, table.metadata().current_schema_id());\n}\n// after: reload and reconcile via the schema-change path instead of failing\nif table.metadata().current_schema_id() != schema_id {\n    return Err(SinkError::Iceberg(anyhow!(\n        \"schema id mismatch: sink expects {}, table is at {}; run commit_schema_change to reconcile\",\n        schema_id, table.metadata().current_schema_id()\n    )));\n}","handlingStrategy":"try-catch","validationCode":"// After any external DDL, verify the sink's expected schema id:\nlet table = catalog.load_table(&ident).await?;\nif table.metadata().current_schema_id() != sink_schema_id {\n    // reconcile: restart sink or run commit_schema_change\n}","typeGuard":"fn schema_in_sync(table: &Table, sink_schema_id: i32) -> bool {\n    table.metadata().current_schema_id() == sink_schema_id\n}","tryCatchPattern":"match run_with_retry(...).await {\n    Err(e) if msg_contains(&e, \"schema evolution not supported\") => reconcile_sink_schema().await,\n    Err(e) => alert(&e),\n    Ok(_) => {}\n}","preventionTips":["Route all schema evolution for sink-managed tables through RisingWave.","Restart the sink promptly after any accepted schema change.","Freeze external writers' DDL during sink uptime."],"tags":["iceberg","schema-evolution","concurrency","retry"],"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"}