Repository: risingwavelabs/risingwave
File: src/frontend/src/datafusion/iceberg_executor.rs (Rust)

Code before:
    let mut target_columns = Vec::with_capacity(batch.num_columns());
    for (column, target_field) in batch.columns().iter().zip_eq_fast(target_schema.fields()) {
        if column.data_type() == target_field.data_type() {
            target_columns.push(column.clone());
        } else {
            let casted_column = datafusion::arrow::compute::cast(column, target_field.data_type())?;
            target_columns.push(casted_column);
        }
    }

    let res = RecordBatch::try_new_with_options(
        target_schema.clone(),
        target_columns,
        &RecordBatchOptions::new().with_row_count(Some(batch.num_rows())),
    )?;
    Ok(res)
}

fn calculate_statistics<'a>(
    tasks: impl Iterator<Item = &'a FileScanTask>,
    schema: &Schema,
) -> Statistics {
    let mut total_rows: usize = 0;
    let mut total_bytes: usize = 0;
    for task in tasks {
        total_rows += task.record_count.unwrap() as usize;
        total_bytes += task.file_size_in_bytes as usize;
    }
    Statistics {
        num_rows: Precision::Exact(total_rows),
        total_byte_size: Precision::Exact(total_bytes),
        column_statistics: Statistics::unknown_column(schema),
    }
}


Review comment (suggestion):
The `unwrap()` call on `task.record_count` can panic if the record count is None. Looking at other code in the codebase (e.g., `src/connector/src/source/iceberg/mod.rs` line 500), the `record_count` field is an `Option` and uses `.expect("must have")` to handle this case. This should be changed to properly handle the None case, either by using `unwrap_or(0)`, `expect()` with a descriptive message, or returning an error if None is unexpected for statistics calculation.
```suggestion
        total_rows += task
            .record_count
            .expect("FileScanTask must have record_count for statistics calculation")
            as usize;
```

Code after:
    let mut target_columns = Vec::with_capacity(batch.num_columns());
    for (column, target_field) in batch.columns().iter().zip_eq_fast(target_schema.fields()) {
        if column.data_type() == target_field.data_type() {
            target_columns.push(column.clone());
        } else {
            let casted_column = datafusion::arrow::compute::cast(column, target_field.data_type())?;
            target_columns.push(casted_column);
        }
    }

    let res = RecordBatch::try_new_with_options(
        target_schema.clone(),
        target_columns,
        &RecordBatchOptions::new().with_row_count(Some(batch.num_rows())),
    )?;
    Ok(res)
}

fn calculate_statistics<'a>(
    tasks: impl Iterator<Item = &'a FileScanTask>,
    schema: &Schema,
) -> Statistics {
    let mut total_rows: Option<usize> = Some(0);
    let mut total_bytes: usize = 0;
    for task in tasks {
        match (task.record_count, total_rows) {
            (Some(count), Some(ref mut total_rows)) => *total_rows += count as usize,
            (None, _) => total_rows = None,
            _ => {}
        };
        total_bytes += task.file_size_in_bytes as usize;
    }

    let num_rows = match total_rows {
        Some(rows) => Precision::Exact(rows),
        None => Precision::Absent,
    };
    Statistics {
        num_rows,
        total_byte_size: Precision::Exact(total_bytes),
        column_statistics: Statistics::unknown_column(schema),
    }
}