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

Code before:
        inner.plan_properties.partitioning = Partitioning::UnknownPartitioning(inner.tasks.len());

        Ok(Self {
            inner: Arc::new(inner),
        })
    }
}

impl IcebergScanInner {
    #[try_stream(ok = FileScanTask, error = DataFusionError)]
    async fn list_iceberg_scan_task(&self) {
        let mut scan_builder = self.table.scan().select(&self.column_names);
        if let Some(snapshot_id) = self.snapshot_id {
            scan_builder = scan_builder.snapshot_id(snapshot_id);
        }
        if matches!(
            self.iceberg_scan_type,
            IcebergScanType::EqualityDeleteScan | IcebergScanType::PositionDeleteScan
        ) {
            scan_builder = scan_builder.with_delete_file_processing_enabled(true);
        }
        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

Review comment (performance):
[nitpick] The HashSets for deduplicating delete files are declared inside the generator function but outside the scan type match. For `DataScan` type, these sets are allocated but never used, which is wasteful. Consider moving the HashSet declarations inside their respective match arms to avoid unnecessary allocations for DataScan operations.

Code after:
        inner.tasks = IcebergSplitEnumerator::split_n_vecs(scan_tasks, batch_parallelism);
        inner.plan_properties.partitioning = Partitioning::UnknownPartitioning(inner.tasks.len());

        Ok(Self {
            inner: Arc::new(inner),
        })
    }
}

impl IcebergScanInner {
    #[try_stream(ok = FileScanTask, error = DataFusionError)]
    async fn list_iceberg_scan_task(&self) {
        let mut scan_builder = self.table.scan().select(&self.column_names);
        if let Some(snapshot_id) = self.snapshot_id {
            scan_builder = scan_builder.snapshot_id(snapshot_id);
        }
        if matches!(
            self.iceberg_scan_type,
            IcebergScanType::EqualityDeleteScan | IcebergScanType::PositionDeleteScan
        ) {
            scan_builder = scan_builder.with_delete_file_processing_enabled(true);
        }
        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 {