{"record":{"id":"2ba95ef5327084ae","repo":"risingwavelabs/risingwave","slug":"load-iceberg-table-for-pk-index-sink","errorCode":null,"errorMessage":"load iceberg table for pk-index sink","messagePattern":"load iceberg table for pk-index sink","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/iceberg_pk_index_sink/coordinator.rs","lineNumber":378,"sourceCode":"                    return Ok(());\n                }\n            }\n        }\n    }\n    Ok(())\n}\n\nasync fn load_catalog_and_table(\n    iceberg_config: &IcebergConfig,\n) -> Result<(Arc<dyn Catalog>, Table)> {\n    let catalog = iceberg_config\n        .create_catalog()\n        .await\n        .map_err(|e| anyhow!(e).context(\"create iceberg catalog for pk-index sink\"))?;\n    let table = iceberg_config\n        .load_table()\n        .await\n        .map_err(|e| anyhow!(e).context(\"load iceberg table for pk-index sink\"))?;\n    Ok((catalog, table))\n}\n\n/// Read every persisted row for this sink, recovering `prev_committed_epoch` and pending commits.\nasync fn recovery(\n    db: &DatabaseConnection,\n    sink_id: SinkId,\n) -> Result<(Option<u64>, Vec<EpochCommit>)> {\n    fail::fail_point!(\"iceberg_v3_recovery_fail\", |_| Err(anyhow::anyhow!(\n        \"injected: iceberg_v3_recovery_fail\"\n    )));\n    let rows = list_sink_states_ordered_by_epoch(db, sink_id)\n        .await\n        .context(\"list pending sink states for pk-index sink recovery\")?;\n\n    let mut prev_committed_epoch = None;\n    let mut pending = Vec::new();\n    let mut aborted_epochs = Vec::new();","sourceCodeStart":360,"sourceCodeEnd":396,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/iceberg_pk_index_sink/coordinator.rs#L360-L396","documentation":"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`.","triggerScenarios":"`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.","commonSituations":"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.","solutions":["Confirm the table identifier in the sink config matches an existing table in the catalog (list tables via the catalog REST API or spark).","Check whether the table was dropped/renamed externally; recreate it or recreate the sink pointing at the right table.","Verify permissions/credentials for the principal used by the meta node; retry after transient catalog errors."],"exampleFix":"// before: table missing after external drop\nCREATE SINK s FROM mv WITH (...) ('table.name'='db.my_table');\n// after: recreate table then sink, or correct identifier\nWITH ('table.name'='db.index_tbl_v2')","handlingStrategy":"try-catch","validationCode":"// verify table exists in catalog before sink init\n// curl -s $CATALOG_URI/v1/{prefix}/namespaces/{ns}/tables/{table}","typeGuard":null,"tryCatchPattern":"match load_catalog_and_table(&cfg).await {\n    Ok((catalog, table)) => { /* proceed */ }\n    Err(e) if e.to_string().contains(\"load iceberg table\") => {\n        tracing::error!(error = ?e, \"target table missing or inaccessible\");\n        // recreate table or fix identifier, then retry init\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Confirm the sink's table.name matches an existing catalog table at creation time.","Don't drop/rename the downstream table while a sink depends on it.","Audit IAM/ACL policies on the table before rotating credentials."],"tags":["iceberg","resource-not-found","catalog","initialization"],"backgroundTag":"entity-not-found","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"}