What causes "iceberg sink metadata should have data_files object" in risingwavelabs/risingwave?

Regenerate metadata with the current TryFrom serializer which always writes data_files (possibly empty array)

risingwavelabs/risingwave · error · SinkError::Iceberg

iceberg sink metadata should have data_files object

Error message

iceberg sink metadata should have data_files object

What it means

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.

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

Example fix

// before
{"schema_id":1,"partition_spec_id":0}
// after
{"schema_id":1,"partition_spec_id":0,"data_files":[]}

Defensive patterns

Strategy: validation

Validate before calling

fn has_data_files(bytes: &[u8]) -> bool {
    serde_json::from_slice::<serde_json::Value>(bytes).ok()
        .map_or(false, |v| v.get("data_files").is_some())
}

Type guard

fn data_files_is_array(v: &serde_json::Value) -> bool {
    matches!(v.get("data_files"), Some(serde_json::Value::Array(_)))
}

Try / catch

match try_from_serialized_bytes(&bytes) {
    Ok(r) => r,
    Err(e) => { warn!("dropping malformed metadata: {}", e.as_report()); default_result() }
}

Prevention

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

When it happens

Trigger: 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.

Common situations: Upgrades with format changes to sink metadata; corrupted/truncated payloads in state store; manually constructed metadata for testing.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/c344ba961d93cb80. Report an issue: GitHub.

Appendix: source

Thrown at src/connector/src/sink/iceberg/commit.rs:101

                .as_u64()
                .ok_or_else(|| anyhow!("schema_id should be a u64"))?;
        } else {
            bail!("iceberg sink metadata should have schema_id");
        }

        let partition_spec_id;
        if let Some(serde_json::Value::Number(value)) = values.remove(PARTITION_SPEC_ID) {
            partition_spec_id = value
                .as_u64()
                .ok_or_else(|| anyhow!("partition_spec_id should be a u64"))?;
        } else {
            bail!("iceberg sink metadata should have partition_spec_id");
        }

        let data_files: Vec<SerializedDataFile>;
        if let serde_json::Value::Array(values) = values
            .remove(DATA_FILES)
            .ok_or_else(|| anyhow!("iceberg sink metadata should have data_files object"))?
        {
            data_files = values
                .into_iter()
                .map(from_value::<SerializedDataFile>)
                .collect::<std::result::Result<_, _>>()
                .unwrap();
        } else {
            bail!("iceberg sink metadata should have data_files object");
        }

        Ok(Self {
            schema_id: schema_id as i32,
            partition_spec_id: partition_spec_id as i32,
            data_files,
        })
    }
}

View on GitHub (pinned to 6469eb736d)