risingwavelabs/risingwave · error

load iceberg table for pk-index sink

Error message

load iceberg table for pk-index sink

What it means

After creating the catalog, `load_catalog_and_table` loads the target Iceberg table using the configured table identifier. Failures (table not found in the catalog, permission denied, catalog request error) are wrapped with this context by the pk-index sink coordinator's `init`.

Source

Thrown at src/meta/src/manager/iceberg_pk_index_sink/coordinator.rs:378

                    return Ok(());
                }
            }
        }
    }
    Ok(())
}

async fn load_catalog_and_table(
    iceberg_config: &IcebergConfig,
) -> Result<(Arc<dyn Catalog>, Table)> {
    let catalog = iceberg_config
        .create_catalog()
        .await
        .map_err(|e| anyhow!(e).context("create iceberg catalog for pk-index sink"))?;
    let table = iceberg_config
        .load_table()
        .await
        .map_err(|e| anyhow!(e).context("load iceberg table for pk-index sink"))?;
    Ok((catalog, table))
}

/// Read every persisted row for this sink, recovering `prev_committed_epoch` and pending commits.
async fn recovery(
    db: &DatabaseConnection,
    sink_id: SinkId,
) -> Result<(Option<u64>, Vec<EpochCommit>)> {
    fail::fail_point!("iceberg_v3_recovery_fail", |_| Err(anyhow::anyhow!(
        "injected: iceberg_v3_recovery_fail"
    )));
    let rows = list_sink_states_ordered_by_epoch(db, sink_id)
        .await
        .context("list pending sink states for pk-index sink recovery")?;

    let mut prev_committed_epoch = None;
    let mut pending = Vec::new();
    let mut aborted_epochs = Vec::new();

View on GitHub (pinned to 6469eb736d)

Solutions

  1. Confirm the table identifier in the sink config matches an existing table in the catalog (list tables via the catalog REST API or spark).
  2. Check whether the table was dropped/renamed externally; recreate it or recreate the sink pointing at the right table.
  3. Verify permissions/credentials for the principal used by the meta node; retry after transient catalog errors.

Example fix

// before: table missing after external drop
CREATE SINK s FROM mv WITH (...) ('table.name'='db.my_table');
// after: recreate table then sink, or correct identifier
WITH ('table.name'='db.index_tbl_v2')
Defensive patterns

Strategy: try-catch

Validate before calling

// verify table exists in catalog before sink init
// curl -s $CATALOG_URI/v1/{prefix}/namespaces/{ns}/tables/{table}

Try / catch

match load_catalog_and_table(&cfg).await {
    Ok((catalog, table)) => { /* proceed */ }
    Err(e) if e.to_string().contains("load iceberg table") => {
        tracing::error!(error = ?e, "target table missing or inaccessible");
        // recreate table or fix identifier, then retry init
    }
    Err(e) => return Err(e),
}

Prevention

When it happens

Trigger: `iceberg_config.load_table()` fails during coordinator init: table identifier doesn't exist in the catalog, table was dropped/renamed, catalog rejects the request, transient network error.

Common situations: Downstream Iceberg table dropped or renamed by an external job while the RisingWave sink is active; wrong database/table name in sink options; IAM/ACL changes revoke read/write on the table.

Understand the failure class

Background: "Not found" and "does not exist" errors: why "Task not found", "No such folder", and "Can't find" fire when a lookup comes back empty — this error's family across 14 libraries.

Related errors

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