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.

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
• Can't find schema by id
• `commit-checkpoint-interval` must be greater than 0

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

Appendix: 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)