risingwavelabs/risingwave · error · SinkError::Iceberg

Invalid snapshot id bytes

Error message

Invalid snapshot id bytes

What it means

The popped snapshot_id bytes are converted to an 8-byte array with try_into() before i64::from_le_bytes; if the byte slice length differs from 8, this error fires. It protects against truncated or malformed snapshot id payloads.

Solutions

1. Ensure snapshot_id is written exactly as i64::to_le_bytes() (8 bytes, little-endian)
2. Check writer/reader version compatibility for the metadata encoding
3. Inspect the offending payload length and regenerate the commit metadata

Example fix

// before
payload.push(snapshot_id.to_be_bytes().to_vec());
// after
payload.push(snapshot_id.to_le_bytes().to_vec());

fn is_valid_snapshot_id_bytes(b: &[u8]) -> bool {
    b.len() == 8
}

fn le_i64(b: &[u8]) -> Option<i64> {
    b.try_into().ok().map(i64::from_le_bytes)
}

let snapshot_id = i64::from_le_bytes(
    snapshot_id_bytes.try_into()
        .map_err(|_| SinkError::Iceberg(anyhow!("Invalid snapshot id bytes")))?,
);

When it happens

Trigger: The final payload element is not exactly 8 bytes — e.g. snapshot id stored as big-endian with different handling, serialized as a decimal string, or the payload slice was truncated by storage.

Common situations: Corruption or truncation in the state store; mixed writer versions using a different snapshot id encoding; hand-assembled test metadata.

Understand the failure class

Background: "Invalid ... format", "must be in format X", "does not look like a ..." — invalid argument format errors across CLI tools and libraries — this error's family across 17 libraries.

Related errors

• bounded compaction branch
• bounded compaction head sequence
• bounded compaction is not supported for copy-on-write tasks
• bounded compaction requires Iceberg format V2 or V3
• Can't create iceberg dv merger commit result from empty…

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

Appendix: source

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

            return Ok(());
        }

        // Deserialize commit metadata
        let mut payload = deserialize_metadata(commit_metadata);
        if payload.is_empty() {
            return Err(SinkError::Iceberg(anyhow!(
                "Invalid commit metadata: empty payload"
            )));
        }

        // Last element is snapshot_id
        let snapshot_id_bytes = payload.pop().ok_or_else(|| {
            SinkError::Iceberg(anyhow!("Invalid commit metadata: missing snapshot_id"))
        })?;
        let snapshot_id = i64::from_le_bytes(
            snapshot_id_bytes
                .try_into()
                .map_err(|_| SinkError::Iceberg(anyhow!("Invalid snapshot id bytes")))?,
        );

        // Remaining elements are write_results
        let write_results = payload
            .into_iter()
            .map(|p| IcebergCommitResult::try_from_serialized_bytes(&p))
            .collect::<Result<Vec<_>>>()?;

        let snapshot_committed = self.is_snapshot_id_in_iceberg(snapshot_id).await?;

        if snapshot_committed {
            tracing::info!(
                iceberg_component = "sink_committer",
                iceberg_operation = "commit",
                sink_id = %self.sink_id,
                sink_name = %self.param.sink_name,
                table = %self.table.identifier(),
                epoch,

View on GitHub (pinned to 6469eb736d)