{"record":{"id":"eb108e002e7d7a55","repo":"risingwavelabs/risingwave","slug":"schema-id-and-partition-spec-id-should-be-the-same","errorCode":null,"errorMessage":"schema_id and partition_spec_id should be the same in all write results","messagePattern":"schema_id and partition_spec_id should be the same in all write results","errorType":"exception","errorClass":"SinkError::Iceberg","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/commit.rs","lineNumber":625,"sourceCode":"        );\n\n        // Skip if no data to commit\n        if write_results.is_empty() || write_results.iter().all(|r| r.data_files.is_empty()) {\n            return Ok(None);\n        }\n\n        let expect_schema_id = write_results[0].schema_id;\n        let expect_partition_spec_id = write_results[0].partition_spec_id;\n\n        // guarantee that all write results has same schema_id and partition_spec_id\n        if write_results\n            .iter()\n            .any(|r| r.schema_id != expect_schema_id)\n            || write_results\n                .iter()\n                .any(|r| r.partition_spec_id != expect_partition_spec_id)\n        {\n            return Err(SinkError::Iceberg(anyhow!(\n                \"schema_id and partition_spec_id should be the same in all write results\"\n            )));\n        }\n\n        let snapshot_id = FastAppendAction::generate_snapshot_id(&self.table);\n        tracing::debug!(\n            iceberg_component = \"sink_committer\",\n            iceberg_operation = \"pre_commit\",\n            sink_id = %self.sink_id,\n            table = %self.table.identifier(),\n            epoch,\n            snapshot_id,\n            schema_id = expect_schema_id,\n            partition_spec_id = expect_partition_spec_id,\n            data_file_count,\n            \"iceberg_sink_pre_commit_snapshot_assigned\",\n        );\n","sourceCodeStart":607,"sourceCodeEnd":643,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/commit.rs#L607-L643","documentation":"pre_commit_inner validates that every write result in the batch used the same schema_id and partition_spec_id expected for the table (expect_schema_id / expect_partition_spec_id). If any result diverges, the snapshot would mix files from incompatible schemas/specs, so it aborts with this error before committing to Iceberg.","triggerScenarios":"A sink epoch contains write results produced under a different table schema or partition spec than the current one — e.g. the table was evolved (ALTER / schema evolution / partition spec rewrite) while in-flight write results from the old schema were still being committed.","commonSituations":"Concurrent Iceberg table schema/partition evolution by another engine or a RisingWave schema change mid-epoch; sinks recovered across a table definition change; stale actors continuing to write after schema evolution.","solutions":["Ensure no schema/partition evolution happens while a sink epoch is in flight (pause evolution or drain the sink first)","Restart/rebuild the sink so all actors use the current schema_id and partition_spec_id","Verify the table's current schema/spec matches what the sink was created with and update the sink definition if it drifted","Check for concurrent writers (other engines) evolving the table and coordinate external evolution"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"fn write_results_consistent(results: &[IcebergCommitResult], schema_id: i32, spec_id: i32) -> bool {\n    results.iter().all(|r| r.schema_id == schema_id && r.partition_spec_id == spec_id)\n}","typeGuard":null,"tryCatchPattern":"if !write_results_consistent(&write_results, expect_schema_id, expect_partition_spec_id) {\n    return Err(SinkError::Iceberg(anyhow!(\"inconsistent schema_id/partition_spec_id in write results\")));\n}","preventionTips":["Avoid evolving table schema/partition spec while sink epochs are in flight","Restart sinks after table definition changes so all writers share the same ids","Monitor for external engines evolving the table concurrently"],"tags":["rust","iceberg","schema-evolution","consistency"],"backgroundTag":"schema-validation-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"}