Repository: risingwavelabs/risingwave
File: src/connector/src/sink/iceberg/mod.rs (Rust)

Code before:
    }
}

const SCHEMA_ID: &str = "schema_id";
const PARTITION_SPEC_ID: &str = "partition_spec_id";
const DATA_FILES: &str = "data_files";

/// Maximum size for column statistics (min/max values) in bytes.
/// Column statistics larger than this will be truncated to avoid metadata bloat.
/// This is especially important for large fields like JSONB, TEXT, BINARY, etc.
///
/// Temporary workaround for large column statistics in `DataFile` metadata.
/// TODO: Remove this once iceberg-rust supports configurable statistics size limits.
const MAX_COLUMN_STAT_SIZE: usize = 1024; // 1KB

/// Truncate large column statistics from serialized data file metadata.
///
/// This function removes `lower_bounds` and `upper_bounds` that exceed
/// `MAX_COLUMN_STAT_SIZE` to prevent metadata bloat and OOM errors.
///
/// For large fields (JSONB, TEXT, BINARY), storing full min/max values provides
/// little query optimization benefit but can cause metadata to balloon to gigabytes.
///
/// # Arguments
/// * `json_value` - Mutable reference to the JSON representation of a `DataFile`
fn truncate_large_column_stats(json_value: &mut serde_json::Value) {
    if let serde_json::Value::Object(obj) = json_value {
        // Process lower_bounds
        if let Some(serde_json::Value::Object(lower_bounds)) = obj.get_mut("lower_bounds") {
            lower_bounds.retain(|field_id, value| {
                if let serde_json::Value::Array(bytes) = value {
                    let size = bytes.len();
                    if size > MAX_COLUMN_STAT_SIZE {
                        tracing::debug!(
                            field_id = field_id,
                            size = size,
                            "Truncating large lower_bound statistic"
                        );
                        return false;
                    }
                }
                true
            });
        }

        // Process upper_bounds
        if let Some(serde_json::Value::Object(upper_bounds)) = obj.get_mut("upper_bounds") {
            upper_bounds.retain(|field_id, value| {
                if let serde_json::Value::Array(bytes) = value {
                    let size = bytes.len();
                    if size > MAX_COLUMN_STAT_SIZE {

Review comment (bug):
I suspect this method fail to truncate the large column, because iceberg datafile doesn't serialize statistics in this struct.

Code after:

            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()
                    .map(from_value::<SerializedDataFile>)
                    .collect::<std::result::Result<_, _>>()
                    .unwrap();
            } else {
                bail!("iceberg sink metadata should have data_files object");
            }

            Ok(Self {
                schema_id: schema_id as i32,
                partition_spec_id: partition_spec_id as i32,
                data_files,
            })
        } else {
            bail!("Can't create iceberg sink write result from empty data!")
        }
    }

    fn try_from_serialized_bytes(value: Vec<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");