{"record":{"id":"564a90e6007fb545","repo":"risingwavelabs/risingwave","slug":"can-t-find-schema-by-id","errorCode":null,"errorMessage":"Can't find schema by id {}","messagePattern":"Can't find schema by id (.+?)","errorType":"exception","errorClass":"SinkError::Iceberg","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/commit.rs","lineNumber":692,"sourceCode":"            schema_id = expect_schema_id,\n            partition_spec_id = expect_partition_spec_id,\n            data_file_count,\n            retry_num = self.commit_retry_num,\n            \"iceberg_sink_commit_applying\",\n        );\n\n        // Load the latest table to avoid concurrent modification with the best effort.\n        self.table = commit_retry::reload_table(\n            self.catalog.as_ref(),\n            self.table.identifier(),\n            expect_schema_id,\n            expect_partition_spec_id,\n        )\n        .await\n        .map_err(SinkError::Iceberg)?;\n\n        let Some(schema) = self.table.metadata().schema_by_id(expect_schema_id) else {\n            return Err(SinkError::Iceberg(anyhow!(\n                \"Can't find schema by id {}\",\n                expect_schema_id\n            )));\n        };\n        let partition_type = resolve_partition_type(&self.table, expect_partition_spec_id, schema)?;\n\n        let data_files = write_results\n            .into_iter()\n            .flat_map(|r| {\n                r.data_files.into_iter().map(|f| {\n                    f.try_into(expect_partition_spec_id, &partition_type, schema)\n                        .map_err(|err| SinkError::Iceberg(anyhow!(err)))\n                })\n            })\n            .collect::<Result<Vec<DataFile>>>()?;\n\n        // # TODO:\n        // This retry behavior should be revert and do in iceberg-rust when it supports retry(Track in: https://github.com/apache/iceberg-rust/issues/964)","sourceCodeStart":674,"sourceCodeEnd":710,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/commit.rs#L674-L710","documentation":"During an Iceberg sink commit, RisingWave looks up the table schema by the schema_id recorded in the data files' write results. The reloaded table metadata does not contain a schema with that id, meaning the writer and the current table metadata have diverged. This is thrown from commit_data_impl in commit.rs when schema_by_id(expect_schema_id) returns None.","triggerScenarios":"The table's metadata was replaced (e.g., schema dropped/rewritten by an external process or catalog swap) so the schema_id captured at write time no longer exists; or stale/buggy write results carry an invalid schema_id; or the catalog loaded a different table version concurrently.","commonSituations":"Another engine (Spark/Trino/Flink) replaced or expired table metadata under RisingWave; a `CREATE OR REPLACE` / drop-and-recreate of the target table; pointing the sink at a table that was recreated with different schema ids; iceberg catalog backends with eventual consistency returning older metadata.","solutions":["Reload the table and verify the schema id with `SELECT ... metadata.current_schema_id` or inspect the catalog; ensure the sink's expected schema id still exists.","If the table was recreated or replaced, recreate the RisingWave sink so it re-reads the current table metadata.","Check for concurrent external writers performing incompatible schema operations; pause them or enable schema evolution handling.","Upgrade RisingWave / iceberg-rust if a known metadata-versioning bug is involved."],"exampleFix":"// before\nlet Some(schema) = self.table.metadata().schema_by_id(expect_schema_id) else {\n    return Err(SinkError::Iceberg(anyhow!(\"Can't find schema by id {}\", expect_schema_id)));\n};\n// after: reload once more before failing, then surface a clearer message\nlet table = commit_retry::reload_table(self.catalog.as_ref(), self.table.identifier(), expect_schema_id, expect_partition_spec_id).await?;\nlet Some(schema) = table.metadata().schema_by_id(expect_schema_id) else {\n    return Err(SinkError::Iceberg(anyhow!(\"schema id {} not found in table {} metadata; table may have been replaced\", expect_schema_id, self.table.identifier())));\n};","handlingStrategy":"validation","validationCode":"// Before relying on the sink, check the table schema id still exists:\nlet table = catalog.load_table(&table_ident).await?;\nlet schema = table.metadata().schema_by_id(expected_schema_id);\nif schema.is_none() {\n    // table was replaced or schema id is stale: recreate the sink\n}","typeGuard":"fn schema_exists(table: &Table, schema_id: i32) -> bool {\n    table.metadata().schema_by_id(schema_id).is_some()\n}","tryCatchPattern":null,"preventionTips":["Never drop-and-recreate the target Iceberg table while its sink exists.","Keep external writers from mutating table metadata managed by RisingWave.","Monitor catalog metadata versions for unexpected changes."],"tags":["iceberg","schema","metadata","sink-commit"],"backgroundTag":"resource-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"}