risingwavelabs/risingwave · error · CommitError::Commit

apply iceberg fast_append

Error message

apply iceberg fast_append

What it means

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.

Source

Thrown at src/connector/src/sink/iceberg/commit.rs:765

                            epoch.to_string(),
                        )]))
                        .add_data_files(data_files);

                    let tx = append_action.apply(txn).map_err(|err| {
                        let err: IcebergError = err.into();
                        tracing::error!(
                            iceberg_component = "sink_committer",
                            iceberg_operation = "commit",
                            sink_id = %sink_id,
                            table = %table_name,
                            epoch,
                            snapshot_id,
                            branch = %target_branch,
                            data_file_count,
                            error = %err.as_report(),
                            "iceberg_sink_commit_fast_append_apply_failed",
                        );
                        CommitError::Commit(anyhow!(err).context("apply iceberg fast_append"))
                    })?;

                    let table = tx.commit(catalog.as_ref()).await.map_err(|err| {
                        let err: IcebergError = err.into();
                        tracing::error!(
                            iceberg_component = "sink_committer",
                            iceberg_operation = "commit",
                            sink_id = %sink_id,
                            table = %table_name,
                            epoch,
                            snapshot_id,
                            branch = %target_branch,
                            data_file_count,
                            error = %err.as_report(),
                            "iceberg_sink_commit_transaction_failed",
                        );
                        CommitError::Commit(anyhow!(err).context("commit iceberg transaction"))
                    })?;

View on GitHub (pinned to 6469eb736d)

Solutions

1. 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).
2. Verify the target branch exists in the table or remove branch targeting if using the main branch.
3. Ensure no concurrent writers change schema/partition spec mid-epoch; align the sink schema with the table.
4. Retry the epoch commit; run_with_retry already retries, so check whether the failure is transient (catalog 4xx/5xx).

Example fix

// before: hard to diagnose
CommitError::Commit(anyhow!(err).context("apply iceberg fast_append"))
// after: include branch and file counts in the wrapped error for diagnosis
CommitError::Commit(anyhow!(err).context(format!(
    "apply iceberg fast_append (branch={branch}, files={data_file_count}, schema_id={expect_schema_id})"
)))

// Validate branch and schema before appending
let branch_ok = table.metadata().branches().map(|b| b.contains(&target_branch)).unwrap_or(target_branch == "main");
let schema_ok = table.metadata().schema_by_id(expect_schema_id).is_some();
if !(branch_ok && schema_ok) { return Err(/* abort before apply */); }

fn branch_exists(table: &Table, branch: &str) -> bool {
    table.metadata().branches().map(|bs| bs.iter().any(|b| b == branch)).unwrap_or(false)
}

match fast_append_result {
    Err(e) if is_transient_catalog_error(&e) => schedule_retry(epoch),
    Err(e) => { log_error(&e); alert_sink_stalled(epoch); }
    Ok(_) => {}
}

When it happens

Trigger: 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).

Common situations: 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.

Related errors

• Can't find schema by id {}
• apply iceberg pk-index sink overwrite_files action
• support negative scale for arrow decimal
• `warehouse.path` must be set
• Unsupported scheme: {}

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