risingwavelabs/risingwave · error · SinkError::Iceberg

iceberg sink metadata should have partition_spec_id

Error message

iceberg sink metadata should have partition_spec_id

What it means

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.

Source

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

            bail!("iceberg sink metadata should be an object");
        };

        let schema_id;
        if let Some(serde_json::Value::Number(value)) = values.remove(SCHEMA_ID) {
            schema_id = value
                .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,

View on GitHub (pinned to 6469eb736d)

Solutions

  1. Verify the payload is actually an IcebergCommitResult (has schema_id/partition_spec_id/data_files), not another metadata type like IcebergPositionDeleteCommitResult
  2. Align writer/reader versions of RisingWave so the serialization format matches
  3. Dump and inspect the metadata bytes to confirm the key set
Defensive patterns

Strategy: validation

Validate before calling

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

Type guard

fn is_commit_result_json(v: &serde_json::Value) -> bool {
    ["schema_id", "partition_spec_id", "data_files"].iter().all(|k| v.get(k).is_some())
}

Try / catch

if !has_partition_spec_id(&bytes) {
    return Err(SinkError::Iceberg(anyhow!("metadata missing partition_spec_id")));
}

Prevention

When it happens

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

Common situations: 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.

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/83ba728adc17849f. Report an issue: GitHub.