risingwavelabs/risingwave · error · SinkError::Iceberg

iceberg sink metadata should have schema_id

Error message

iceberg sink metadata should have schema_id

What it means

try_from_serialized_bytes parses the JSON metadata blob attached to an Iceberg sink write result. It requires the top-level object to carry a numeric 'schema_id' field; if the key is absent (removed earlier or never written by the producer), the parser bails with this error. It guards the invariant that every serialized IcebergCommitResult records which table schema its data files were written against.

Source

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

    }

    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");
        }

        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()

View on GitHub (pinned to 6469eb736d)

Solutions

1. Ensure the metadata was serialized by the matching version's TryFrom<&IcebergCommitResult> for Vec<u8> implementation which always writes schema_id
2. Check for version skew between the writer (stream actor) and reader (meta node) and upgrade/downgrade consistently
3. Inspect the raw metadata JSON bytes to confirm schema_id is present and the payload is not corrupt
4. If metadata is from a stale/failed epoch, discard it and let the sink rewrite the files

Example fix

// before (malformed metadata)
{"partition_spec_id":0,"data_files":[]}
// after (valid metadata)
{"schema_id":1,"partition_spec_id":0,"data_files":[]}

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

fn as_u64_field(v: &serde_json::Value, key: &str) -> Option<u64> {
    v.get(key).and_then(|x| x.as_u64())
}

match IcebergCommitResult::try_from_serialized_bytes(&bytes) {
    Ok(r) => r,
    Err(e) => { warn!("bad iceberg metadata: {}", e.as_report()); return; }
}

When it happens

Trigger: Calling IcebergCommitResult::try_from or try_from_serialized_bytes on metadata bytes that deserialize to a JSON object without a 'schema_id' key — e.g. metadata produced by an older RisingWave version, hand-edited metadata, or a producer that serialized a different struct shape.

Common situations: Rolling upgrade where old commit metadata on disk/state store lacks schema_id and is replayed by newer code; bugs in custom serialization paths that drop the field; corrupt or truncated metadata in the meta/state store.

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

• iceberg sink metadata should have partition_spec_id
• iceberg sink metadata should have data_files object
• Invalid commit metadata: missing snapshot_id
• query_epoch not set in distributed lookup join
• table {} not found

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