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

Code before:
            "column count must match schema column count".to_owned(),
        ));
    }

    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 new `calculate_statistics` function lacks documentation. According to the project guidelines, rustdoc comments should be added to explain the purpose of this function, its parameters, and return value. This is especially important since this function is critical for cost-based optimization in DataFusion.

Code after:
            "column count must match schema column count".to_owned(),
        ));
    }

    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),
    }
}