{"record":{"id":"c344ba961d93cb80","repo":"risingwavelabs/risingwave","slug":"iceberg-sink-metadata-should-have-data-files-objec","errorCode":null,"errorMessage":"iceberg sink metadata should have data_files object","messagePattern":"iceberg sink metadata should have data_files object","errorType":"exception","errorClass":"SinkError::Iceberg","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/commit.rs","lineNumber":101,"sourceCode":"                .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,\n            partition_spec_id: partition_spec_id as i32,\n            data_files,\n        })\n    }\n}\n","sourceCodeStart":83,"sourceCodeEnd":119,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/commit.rs#L83-L119","documentation":"try_from_serialized_bytes requires the metadata JSON to include a 'data_files' key; values.remove(DATA_FILES) returning None triggers this error before the Array check. Every write result must list the data files it produced so the commit can add them to the Iceberg snapshot.","triggerScenarios":"Metadata JSON missing 'data_files', produced by older writer versions or by code that serialized a different struct; replay of stale metadata from the state store.","commonSituations":"Upgrades with format changes to sink metadata; corrupted/truncated payloads in state store; manually constructed metadata for testing.","solutions":["Regenerate metadata with the current TryFrom<&IcebergCommitResult> serializer which always writes data_files (possibly empty array)","Check version skew between components and redeploy consistently","Inspect raw JSON keys to confirm which field is actually missing"],"exampleFix":"// before\n{\"schema_id\":1,\"partition_spec_id\":0}\n// after\n{\"schema_id\":1,\"partition_spec_id\":0,\"data_files\":[]}","handlingStrategy":"validation","validationCode":"fn has_data_files(bytes: &[u8]) -> bool {\n    serde_json::from_slice::<serde_json::Value>(bytes).ok()\n        .map_or(false, |v| v.get(\"data_files\").is_some())\n}","typeGuard":"fn data_files_is_array(v: &serde_json::Value) -> bool {\n    matches!(v.get(\"data_files\"), Some(serde_json::Value::Array(_)))\n}","tryCatchPattern":"match try_from_serialized_bytes(&bytes) {\n    Ok(r) => r,\n    Err(e) => { warn!(\"dropping malformed metadata: {}\", e.as_report()); default_result() }\n}","preventionTips":["Always emit data_files even when empty (serialize empty vec, not null)","Round-trip serialization tests after any struct change","Reject non-standard metadata producers"],"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"}