{"record":{"id":"0c3652d6ce5edd23","repo":"risingwavelabs/risingwave","slug":"reload-iceberg-table","errorCode":null,"errorMessage":"reload iceberg table","messagePattern":"reload iceberg table","errorType":"exception","errorClass":"SinkError::Iceberg","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/commit.rs","lineNumber":828,"sourceCode":"            snapshot_num,\n            data_file_count,\n            \"iceberg_sink_commit_succeeded\",\n        );\n\n        self.notify_iceberg_compaction_scheduler(false);\n\n        Ok(())\n    }\n\n    /// During pre-commit metadata, we record the `snapshot_id` corresponding to each batch of files.\n    /// Therefore, the logic for checking whether all files in this batch are present in Iceberg\n    /// has been changed to verifying if their corresponding `snapshot_id` exists in Iceberg.\n    async fn is_snapshot_id_in_iceberg(&self, snapshot_id: i64) -> Result<bool> {\n        let table = self\n            .catalog\n            .load_table(self.table.identifier())\n            .await\n            .map_err(|err| SinkError::Iceberg(anyhow!(err).context(\"reload iceberg table\")))?;\n        if table.metadata().snapshot_by_id(snapshot_id).is_some() {\n            Ok(true)\n        } else {\n            Ok(false)\n        }\n    }\n\n    /// Check if the specified columns already exist in the iceberg table's current schema.\n    /// This is used to determine if schema change has already been applied.\n    fn check_schema_change_applied(&self, schema_change: &PbSinkSchemaChange) -> Result<bool> {\n        let current_schema = self.table.metadata().current_schema();\n        let current_arrow_schema = schema_to_arrow_schema(current_schema.as_ref())\n            .context(\"Failed to convert schema\")\n            .map_err(SinkError::Iceberg)?;\n\n        let iceberg_arrow_convert = IcebergArrowConvert;\n\n        let schema_matches = |expected: &[ArrowField]| {","sourceCodeStart":810,"sourceCodeEnd":846,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/commit.rs#L810-L846","documentation":"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.","triggerScenarios":"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.","commonSituations":"Transient REST catalog outage during sink recovery; table dropped/recreated while the sink was down; Glue/Hive metastore throttling; wrong catalog configuration after migration.","solutions":["Check the underlying catalog error in logs and verify catalog connectivity/credentials.","Confirm the sink's target table still exists and its identifier is unchanged; recreate the sink if the table was dropped/recreated.","Retry after a transient catalog outage; the sink should resume normally.","If using Glue/REST, raise throttling limits or add catalog-side caching."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// Health-check the catalog before relying on snapshot verification:\nlet table = catalog.load_table(&ident).await.context(\"catalog reachable?\");","typeGuard":null,"tryCatchPattern":"match is_snapshot_id_in_iceberg(snapshot_id).await {\n    Ok(true) => { /* commit landed; skip re-commit */ }\n    Ok(false) => { /* safe to re-commit */ }\n    Err(e) => { log_catalog_error(&e); backoff_and_retry(); }\n}","preventionTips":["Monitor catalog service availability and credentials expiry.","Don't drop/rename tables backing active sinks.","Use a catalog with stable read semantics (avoid eventual-consistency surprises)."],"tags":["iceberg","catalog","table-reload","network"],"backgroundTag":"database-query-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"}