risingwavelabs/risingwave · error · SinkError::Iceberg

Can't create iceberg dv merger commit result from empty…

Error message

Can't create iceberg dv merger commit result from empty data!

What it means

IcebergPositionDeleteCommitResult::try_from(&SinkMetadata) requires metadata to be Some(Serialized(..)); if the SinkMetadata carries no serialized payload, the conversion bails. This indicates the position-delete (DV merger) write result was never attached, so there is nothing to deserialize.

Solutions

1. Skip conversion for SinkMetadata with metadata: None instead of attempting TryFrom
2. Ensure the DV merger actor always attaches Serialized metadata before handing results to the commit coordinator
3. Check for state-store write failures or epoch mismatches that dropped the metadata

Example fix

// before
let result = IcebergPositionDeleteCommitResult::try_from(&meta)?;
// after
if meta.metadata.is_none() {
    return Ok(IcebergPositionDeleteCommitResult::default());
}
let result = IcebergPositionDeleteCommitResult::try_from(&meta)?;

fn position_delete_meta_ready(meta: &SinkMetadata) -> bool {
    matches!(meta.metadata, Some(Serialized(_)))
}

fn get_serialized<'a>(m: &'a SinkMetadata) -> Option<&'a SerializedMetadata> {
    match &m.metadata { Some(Serialized(s)) => Some(s), None => None }
}

let Some(Serialized(value)) = &meta.metadata else {
    tracing::warn!("no dv merger metadata; skipping");
    return Ok(IcebergPositionDeleteCommitResult::default());
};

When it happens

Trigger: Calling TryFrom<&SinkMetadata> for IcebergPositionDeleteCommitResult on a SinkMetadata whose metadata field is None — e.g. an empty write result produced when the actor wrote no delete files, or a lost/failed metadata write.

Common situations: Sinks with zero rows in an epoch producing empty metadata; failures persisting metadata to the state store before commit coordination reads it back.

Understand the failure class

Background: "must not be empty", "cannot be empty" — required-field validation errors across open-source libraries — this error's family across 41 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
• Cannot find the snapshot id in the iceberg table.

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

Appendix: source

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

        );
        Ok(serde_json::to_vec(&json_value).context("Can't serialize iceberg sink metadata")?)
    }
}

#[derive(Default, Clone, Serialize, Deserialize)]
pub struct IcebergPositionDeleteCommitResult {
    pub schema_id: i32,
    pub partition_spec_id: i32,
    pub delete_files: Vec<SerializedDataFile>,
    pub overwrite_files: Vec<SerializedDataFile>,
}

impl<'a> TryFrom<&'a SinkMetadata> for IcebergPositionDeleteCommitResult {
    type Error = SinkError;

    fn try_from(value: &'a SinkMetadata) -> Result<Self> {
        let Some(Serialized(value)) = &value.metadata else {
            bail!("Can't create iceberg dv merger commit result from empty data!");
        };
        let value = serde_json::from_slice(&value.metadata)
            .context("Can't deserialize iceberg dv merger commit result from metadata")?;
        Ok(value)
    }
}

impl<'a> TryFrom<&'a IcebergPositionDeleteCommitResult> for SinkMetadata {
    type Error = SinkError;

    fn try_from(value: &'a IcebergPositionDeleteCommitResult) -> Result<SinkMetadata> {
        let bytes = serde_json::to_vec(value)
            .context("Can't serialize iceberg dv merger commit result to metadata")?;
        Ok(SinkMetadata {
            metadata: Some(Serialized(SerializedMetadata { metadata: bytes })),
        })
    }
}

View on GitHub (pinned to 6469eb736d)