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

Code before:
        let scan = scan_builder.build().map_err(to_datafusion_error)?;

        let mut position_delete_files_set = HashSet::new();
        let mut equality_delete_files_set = HashSet::new();

        #[for_await]
        for scan_task in scan.plan_files().await.map_err(to_datafusion_error)? {
            let scan_task = scan_task.map_err(to_datafusion_error)?;
            match self.iceberg_scan_type {
                IcebergScanType::DataScan => {
                    if scan_task.data_file_content != DataContentType::Data {
                        return internal_err!(
                            "Files of type {:?} should not be in the data files",
                            scan_task.data_file_content
                        );
                    }
                    yield scan_task;
                }
                IcebergScanType::EqualityDeleteScan => {
                    for delete_file in scan_task.deletes {
                        if delete_file.data_file_content == DataContentType::EqualityDeletes
                            && equality_delete_files_set.insert(delete_file.data_file_path.clone())
                        {
                            yield delete_file.as_ref().clone()
                        }
                    }
                }
                IcebergScanType::PositionDeleteScan => {
                    for delete_file in scan_task.deletes {
                        if delete_file.data_file_content == DataContentType::PositionDeletes
                            && position_delete_files_set.insert(delete_file.data_file_path.clone())
                        {
                            let mut task = delete_file.as_ref().clone();
                            task.project_field_ids = Vec::new();
                            yield task;
                        }
                    }
                }
                _ => {
                    return not_impl_err!(
                        "Iceberg scan type {:?} is not supported",
                        self.iceberg_scan_type
                    );
                }
            }
        }
    }

    #[try_stream(ok = RecordBatch, error = DataFusionError)]
    pub async fn execute_inner(self: Arc<Self>, chunk_size: usize, partition: usize) {
        let mut buffer = RecordBatchBuffer::new(chunk_size);

Review comment (bug):
[nitpick] The code doesn't validate that equality delete files are actually of type `EqualityDeletes` before checking the content type. If `delete_file.data_file_content` is some other delete type, it would be silently skipped. Consider adding validation or logging for unexpected delete file types to aid debugging.

Code after:
        }
        let scan = scan_builder.build().map_err(to_datafusion_error)?;

        let mut position_delete_files_set = HashSet::new();
        let mut equality_delete_files_set = HashSet::new();

        #[for_await]
        for scan_task in scan.plan_files().await.map_err(to_datafusion_error)? {
            let scan_task = scan_task.map_err(to_datafusion_error)?;
            match self.iceberg_scan_type {
                IcebergScanType::DataScan => {
                    if scan_task.data_file_content != DataContentType::Data {
                        return internal_err!(
                            "Files of type {:?} should not be in the data files",
                            scan_task.data_file_content
                        );
                    }
                    yield scan_task;
                }
                IcebergScanType::EqualityDeleteScan => {
                    for delete_file in scan_task.deletes {
                        if delete_file.data_file_content == DataContentType::EqualityDeletes
                            && equality_delete_files_set.insert(delete_file.data_file_path.clone())
                        {
                            yield delete_file.as_ref().clone()
                        }
                    }
                }
                IcebergScanType::PositionDeleteScan => {
                    for delete_file in scan_task.deletes {
                        if delete_file.data_file_content == DataContentType::PositionDeletes
                            && position_delete_files_set.insert(delete_file.data_file_path.clone())
                        {
                            let mut task = delete_file.as_ref().clone();
                            task.project_field_ids = Vec::new();
                            yield task;
                        }
                    }
                }
                _ => {
                    return not_impl_err!(
                        "Iceberg scan type {:?} is not supported",
                        self.iceberg_scan_type
                    );
                }
            }
        }
    }

    #[try_stream(ok = RecordBatch, error = DataFusionError)]
    pub async fn execute_inner(self: Arc<Self>, chunk_size: usize, partition: usize) {