{"record":{"id":"83ba728adc17849f","repo":"risingwavelabs/risingwave","slug":"iceberg-sink-metadata-should-have-partition-spec-i","errorCode":null,"errorMessage":"iceberg sink metadata should have partition_spec_id","messagePattern":"iceberg sink metadata should have partition_spec_id","errorType":"exception","errorClass":"SinkError::Iceberg","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/commit.rs","lineNumber":95,"sourceCode":"            bail!(\"iceberg sink metadata should be an object\");\n        };\n\n        let schema_id;\n        if let Some(serde_json::Value::Number(value)) = values.remove(SCHEMA_ID) {\n            schema_id = value\n                .as_u64()\n                .ok_or_else(|| anyhow!(\"schema_id should be a u64\"))?;\n        } else {\n            bail!(\"iceberg sink metadata should have schema_id\");\n        }\n\n        let partition_spec_id;\n        if let Some(serde_json::Value::Number(value)) = values.remove(PARTITION_SPEC_ID) {\n            partition_spec_id = value\n                .as_u64()\n                .ok_or_else(|| anyhow!(\"partition_spec_id should be a u64\"))?;\n        } else {\n            bail!(\"iceberg sink metadata should have partition_spec_id\");\n        }\n\n        let data_files: Vec<SerializedDataFile>;\n        if let serde_json::Value::Array(values) = values\n            .remove(DATA_FILES)\n            .ok_or_else(|| anyhow!(\"iceberg sink metadata should have data_files object\"))?\n        {\n            data_files = values\n                .into_iter()\n                .map(from_value::<SerializedDataFile>)\n                .collect::<std::result::Result<_, _>>()\n                .unwrap();\n        } else {\n            bail!(\"iceberg sink metadata should have data_files object\");\n        }\n\n        Ok(Self {\n            schema_id: schema_id as i32,","sourceCodeStart":77,"sourceCodeEnd":113,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/commit.rs#L77-L113","documentation":"try_from_serialized_bytes requires the metadata JSON object to contain a 'partition_spec_id' key. When the key is missing, values.remove(PARTITION_SPEC_ID) returns None and the code bails with this message. Partition spec id is needed later to validate that all data files in a snapshot belong to the same partition spec.","triggerScenarios":"Deserializing commit metadata whose JSON lacks 'partition_spec_id' — typically metadata written by an older format/version, or bytes that are not an IcebergCommitResult at all.","commonSituations":"Version skew between writer and reader of sink metadata; mixing metadata types (e.g. feeding position-delete commit results into IcebergCommitResult parsing); corrupted state-store payloads.","solutions":["Verify the payload is actually an IcebergCommitResult (has schema_id/partition_spec_id/data_files), not another metadata type like IcebergPositionDeleteCommitResult","Align writer/reader versions of RisingWave so the serialization format matches","Dump and inspect the metadata bytes to confirm the key set"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"fn has_partition_spec_id(bytes: &[u8]) -> bool {\n    serde_json::from_slice::<serde_json::Value>(bytes).ok()\n        .map_or(false, |v| v.get(\"partition_spec_id\").is_some())\n}","typeGuard":"fn is_commit_result_json(v: &serde_json::Value) -> bool {\n    [\"schema_id\", \"partition_spec_id\", \"data_files\"].iter().all(|k| v.get(k).is_some())\n}","tryCatchPattern":"if !has_partition_spec_id(&bytes) {\n    return Err(SinkError::Iceberg(anyhow!(\"metadata missing partition_spec_id\")));\n}","preventionTips":["Confirm the payload type before parsing (commit result vs position-delete result)","Round-trip test metadata serialization after format changes","Gate format changes behind version markers in the serialized blob"],"tags":["rust","iceberg","serialization","missing-field"],"backgroundTag":"missing-required-argument","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"}