{"record":{"id":"c631a15999cbc253","repo":"risingwavelabs/risingwave","slug":"apply-iceberg-fast-append","errorCode":null,"errorMessage":"apply iceberg fast_append","messagePattern":"apply iceberg fast_append","errorType":"exception","errorClass":"CommitError::Commit","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/commit.rs","lineNumber":765,"sourceCode":"                            epoch.to_string(),\n                        )]))\n                        .add_data_files(data_files);\n\n                    let tx = append_action.apply(txn).map_err(|err| {\n                        let err: IcebergError = err.into();\n                        tracing::error!(\n                            iceberg_component = \"sink_committer\",\n                            iceberg_operation = \"commit\",\n                            sink_id = %sink_id,\n                            table = %table_name,\n                            epoch,\n                            snapshot_id,\n                            branch = %target_branch,\n                            data_file_count,\n                            error = %err.as_report(),\n                            \"iceberg_sink_commit_fast_append_apply_failed\",\n                        );\n                        CommitError::Commit(anyhow!(err).context(\"apply iceberg fast_append\"))\n                    })?;\n\n                    let table = tx.commit(catalog.as_ref()).await.map_err(|err| {\n                        let err: IcebergError = err.into();\n                        tracing::error!(\n                            iceberg_component = \"sink_committer\",\n                            iceberg_operation = \"commit\",\n                            sink_id = %sink_id,\n                            table = %table_name,\n                            epoch,\n                            snapshot_id,\n                            branch = %target_branch,\n                            data_file_count,\n                            error = %err.as_report(),\n                            \"iceberg_sink_commit_transaction_failed\",\n                        );\n                        CommitError::Commit(anyhow!(err).context(\"commit iceberg transaction\"))\n                    })?;","sourceCodeStart":747,"sourceCodeEnd":783,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/commit.rs#L747-L783","documentation":"Wraps an error from iceberg-rust's `Transaction::fast_append().apply(txn)` inside a CommitError during the sink commit. Applying the fast-append action validates the appended data files against the table's current schema/partition spec and snapshot state; a failure means the transaction could not be built, before any catalog commit happens.","triggerScenarios":"Data files reference a partition spec or schema that no longer matches the reloaded table (commit_data_impl reloads the table and expects write_results' schema_id/partition_spec_id to exist); partition type resolution mismatch; invalid snapshot id/branch set on the append action (e.g., target branch does not exist).","commonSituations":"Concurrent schema evolution by another writer changed the table between write and commit; sink configured with a branch that doesn't exist (write_mode branch); partition spec changed by an external engine; iceberg-rust version incompatibilities with the catalog.","solutions":["Read the inner iceberg error in the logs (iceberg_sink_commit_fast_append_apply_failed) to identify the exact mismatch (schema, partition spec, or branch).","Verify the target branch exists in the table or remove branch targeting if using the main branch.","Ensure no concurrent writers change schema/partition spec mid-epoch; align the sink schema with the table.","Retry the epoch commit; run_with_retry already retries, so check whether the failure is transient (catalog 4xx/5xx)."],"exampleFix":"// before: hard to diagnose\nCommitError::Commit(anyhow!(err).context(\"apply iceberg fast_append\"))\n// after: include branch and file counts in the wrapped error for diagnosis\nCommitError::Commit(anyhow!(err).context(format!(\n    \"apply iceberg fast_append (branch={branch}, files={data_file_count}, schema_id={expect_schema_id})\"\n)))","handlingStrategy":"retry","validationCode":"// Validate branch and schema before appending\nlet branch_ok = table.metadata().branches().map(|b| b.contains(&target_branch)).unwrap_or(target_branch == \"main\");\nlet schema_ok = table.metadata().schema_by_id(expect_schema_id).is_some();\nif !(branch_ok && schema_ok) { return Err(/* abort before apply */); }","typeGuard":"fn branch_exists(table: &Table, branch: &str) -> bool {\n    table.metadata().branches().map(|bs| bs.iter().any(|b| b == branch)).unwrap_or(false)\n}","tryCatchPattern":"match fast_append_result {\n    Err(e) if is_transient_catalog_error(&e) => schedule_retry(epoch),\n    Err(e) => { log_error(&e); alert_sink_stalled(epoch); }\n    Ok(_) => {}\n}","preventionTips":["Keep the sink's target branch created before starting the sink.","Avoid concurrent external schema/partition spec changes mid-epoch.","Pin compatible iceberg-rust versions across writers."],"tags":["iceberg","transaction","fast-append","sink-commit"],"backgroundTag":"database-write-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"}