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

Code before:
            .any(|name| name == ICEBERG_FILE_PATH_COLUMN_NAME)
            && matches!(provider.iceberg_scan_type, IcebergScanType::DataScan);
        column_names.retain(|name| {
            ![
                ICEBERG_FILE_PATH_COLUMN_NAME,
                ICEBERG_SEQUENCE_NUM_COLUMN_NAME,
                ICEBERG_FILE_POS_COLUMN_NAME,
            ]
            .contains(&name.as_str())
        });

        let mut inner = IcebergScanInner {
            table,
            snapshot_id: provider.snapshot_id,
            tasks: vec![],
            iceberg_scan_type: provider.iceberg_scan_type,
            arrow_schema: provider.arrow_schema.clone(),
            column_names,
            need_seq_num,
            need_file_path_and_pos,
            plan_properties,
        };
        inner.tasks = inner.list_iceberg_scan_task().try_collect().await?;
        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)? {

Review comment (performance):
We have a session variable called `batch_parallelism`, I think we can use this value to control the parallelism instead of using the task number directly, because the task number could be very large.

Code after:
            .iter()
            .any(|name| name == ICEBERG_FILE_PATH_COLUMN_NAME)
            && matches!(provider.iceberg_scan_type, IcebergScanType::DataScan);
        column_names.retain(|name| {
            ![
                ICEBERG_FILE_PATH_COLUMN_NAME,
                ICEBERG_SEQUENCE_NUM_COLUMN_NAME,
                ICEBERG_FILE_POS_COLUMN_NAME,
            ]
            .contains(&name.as_str())
        });

        let mut inner = IcebergScanInner {
            table,
            snapshot_id: provider.snapshot_id,
            tasks: vec![],
            iceberg_scan_type: provider.iceberg_scan_type,
            arrow_schema: provider.arrow_schema.clone(),
            column_names,
            need_seq_num,
            need_file_path_and_pos,
            plan_properties,
        };
        let scan_tasks = inner.list_iceberg_scan_task().try_collect().await?;
        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();