risingwavelabs/risingwave · error · SinkError::Iceberg

iceberg sink metadata should be an object

Error message

iceberg sink metadata should be an object

What it means

`IcebergCommitResult::try_from_serialized_bytes` parses the sink metadata bytes as JSON and requires a top-level JSON object whose keys carry schema_id, partition spec, and data files. If the parsed JSON is an array, string, number, etc. instead of an object, it bails with this error.

Source

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

}

impl IcebergCommitResult {
    pub fn try_from(value: &SinkMetadata) -> Result<Self> {
        let Some(Serialized(value)) = &value.metadata else {
            bail!("Can't create iceberg sink write result from empty data!");
        };

        Self::try_from_serialized_bytes(&value.metadata)
    }

    pub fn try_from_serialized_bytes(value: &[u8]) -> Result<Self> {
        let mut values = if let serde_json::Value::Object(value) =
            serde_json::from_slice::<serde_json::Value>(value)
                .context("Can't parse iceberg sink metadata")?
        {
            value
        } else {
            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");

View on GitHub (pinned to 6469eb736d)

Solutions

1. Verify the metadata bytes belong to an Iceberg sink commit (object-shaped JSON with schema_id)
2. Check for RisingWave version mismatch between writer and reader of the metadata
3. Re-run the commit from intact metadata; recreate the sink if metadata is corrupted
4. Log the raw bytes (or their prefix) to confirm what is actually stored

let v: serde_json::Value = serde_json::from_slice(&bytes).context("Can't parse iceberg sink metadata")?;
if !v.is_object() {
    return Err("iceberg sink metadata must be a JSON object".into());
}

fn is_metadata_object(bytes: &[u8]) -> bool {
    serde_json::from_slice::<serde_json::Value>(bytes)
        .map(|v| v.is_object())
        .unwrap_or(false)
}

match IcebergCommitResult::try_from_serialized_bytes(&bytes) {
    Err(e) if e.to_string().contains("should be an object") => {
        // inspect/corroborate stored bytes; likely wrong sink type or version mismatch
    }
    other => other?,
}

When it happens

Trigger: Committing an Iceberg sink whose serialized metadata bytes deserialize to a non-object JSON value — e.g. corrupted metadata bytes, wrong bytes passed (a different sink's metadata format), or a version where the metadata layout changed.

Common situations: Pointing the commit path at metadata produced by a different sink type or older RisingWave version; byte corruption in storage; manual edits to serialized metadata.

Related errors

• schema_id should be a u64
• Can't create iceberg sink write result from empty data!
• partition spec {} not found
• Fail to convert version_hint from utf8 to string: {}
• partition_spec_id should be a u64

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