risingwavelabs/risingwave · error · SinkError::Iceberg

reload iceberg table

Error message

reload iceberg table

What it means

Raised in is_snapshot_id_in_iceberg when reloading the Iceberg table via `catalog.load_table` fails while checking whether a previously recorded snapshot id still exists in the table. This is used to decide whether an earlier epoch's commit actually landed, so a reload failure blocks the commit path entirely.

Source

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

            snapshot_num,
            data_file_count,
            "iceberg_sink_commit_succeeded",
        );

        self.notify_iceberg_compaction_scheduler(false);

        Ok(())
    }

    /// During pre-commit metadata, we record the `snapshot_id` corresponding to each batch of files.
    /// Therefore, the logic for checking whether all files in this batch are present in Iceberg
    /// has been changed to verifying if their corresponding `snapshot_id` exists in Iceberg.
    async fn is_snapshot_id_in_iceberg(&self, snapshot_id: i64) -> Result<bool> {
        let table = self
            .catalog
            .load_table(self.table.identifier())
            .await
            .map_err(|err| SinkError::Iceberg(anyhow!(err).context("reload iceberg table")))?;
        if table.metadata().snapshot_by_id(snapshot_id).is_some() {
            Ok(true)
        } else {
            Ok(false)
        }
    }

    /// Check if the specified columns already exist in the iceberg table's current schema.
    /// This is used to determine if schema change has already been applied.
    fn check_schema_change_applied(&self, schema_change: &PbSinkSchemaChange) -> Result<bool> {
        let current_schema = self.table.metadata().current_schema();
        let current_arrow_schema = schema_to_arrow_schema(current_schema.as_ref())
            .context("Failed to convert schema")
            .map_err(SinkError::Iceberg)?;

        let iceberg_arrow_convert = IcebergArrowConvert;

        let schema_matches = |expected: &[ArrowField]| {

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check the underlying catalog error in logs and verify catalog connectivity/credentials.
2. Confirm the sink's target table still exists and its identifier is unchanged; recreate the sink if the table was dropped/recreated.
3. Retry after a transient catalog outage; the sink should resume normally.
4. If using Glue/REST, raise throttling limits or add catalog-side caching.

// Health-check the catalog before relying on snapshot verification:
let table = catalog.load_table(&ident).await.context("catalog reachable?");

match is_snapshot_id_in_iceberg(snapshot_id).await {
    Ok(true) => { /* commit landed; skip re-commit */ }
    Ok(false) => { /* safe to re-commit */ }
    Err(e) => { log_catalog_error(&e); backoff_and_retry(); }
}

When it happens

Trigger: Catalog service unreachable or erroring during load_table; invalid/expired catalog credentials; the table (or namespace) was dropped or renamed by an external process so load_table returns NotFound; network partition to REST/Hive/Glue catalog.

Common situations: Transient REST catalog outage during sink recovery; table dropped/recreated while the sink was down; Glue/Hive metastore throttling; wrong catalog configuration after migration.

Understand the failure class

Background: Database query failed: Internal Server Error 500s wrapping SQL, Prisma, and connection failures — what to check first — this error's family across 16 libraries.

Related errors

• {e}
• `catalog.type` must be set
• Failed to list iceberg namespaces.
• Failed to load iceberg table.
• Failed to drop iceberg table.

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