{"record":{"id":"db6404ef24c5dda3","repo":"risingwavelabs/risingwave","slug":"iceberg-sink-metadata-should-have-schema-id","errorCode":null,"errorMessage":"iceberg sink metadata should have schema_id","messagePattern":"iceberg sink metadata should have schema_id","errorType":"exception","errorClass":"SinkError::Iceberg","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/commit.rs","lineNumber":86,"sourceCode":"    }\n\n    pub fn try_from_serialized_bytes(value: &[u8]) -> Result<Self> {\n        let mut values = if let serde_json::Value::Object(value) =\n            serde_json::from_slice::<serde_json::Value>(value)\n                .context(\"Can't parse iceberg sink metadata\")?\n        {\n            value\n        } else {\n            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()","sourceCodeStart":68,"sourceCodeEnd":104,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/commit.rs#L68-L104","documentation":"try_from_serialized_bytes parses the JSON metadata blob attached to an Iceberg sink write result. It requires the top-level object to carry a numeric 'schema_id' field; if the key is absent (removed earlier or never written by the producer), the parser bails with this error. It guards the invariant that every serialized IcebergCommitResult records which table schema its data files were written against.","triggerScenarios":"Calling IcebergCommitResult::try_from or try_from_serialized_bytes on metadata bytes that deserialize to a JSON object without a 'schema_id' key — e.g. metadata produced by an older RisingWave version, hand-edited metadata, or a producer that serialized a different struct shape.","commonSituations":"Rolling upgrade where old commit metadata on disk/state store lacks schema_id and is replayed by newer code; bugs in custom serialization paths that drop the field; corrupt or truncated metadata in the meta/state store.","solutions":["Ensure the metadata was serialized by the matching version's TryFrom<&IcebergCommitResult> for Vec<u8> implementation which always writes schema_id","Check for version skew between the writer (stream actor) and reader (meta node) and upgrade/downgrade consistently","Inspect the raw metadata JSON bytes to confirm schema_id is present and the payload is not corrupt","If metadata is from a stale/failed epoch, discard it and let the sink rewrite the files"],"exampleFix":"// before (malformed metadata)\n{\"partition_spec_id\":0,\"data_files\":[]}\n// after (valid metadata)\n{\"schema_id\":1,\"partition_spec_id\":0,\"data_files\":[]}","handlingStrategy":"validation","validationCode":"fn has_schema_id(bytes: &[u8]) -> bool {\n    serde_json::from_slice::<serde_json::Value>(bytes)\n        .ok()\n        .and_then(|v| v.get(\"schema_id\").cloned())\n        .map_or(false, |v| v.as_u64().is_some())\n}","typeGuard":"fn as_u64_field(v: &serde_json::Value, key: &str) -> Option<u64> {\n    v.get(key).and_then(|x| x.as_u64())\n}","tryCatchPattern":"match IcebergCommitResult::try_from_serialized_bytes(&bytes) {\n    Ok(r) => r,\n    Err(e) => { warn!(\"bad iceberg metadata: {}\", e.as_report()); return; }\n}","preventionTips":["Always serialize metadata via TryFrom<&IcebergCommitResult> for Vec<u8>","Keep writer and reader RisingWave versions aligned during upgrades","Log raw metadata bytes when parsing fails to ease diagnosis"],"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"}