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

Code before:
                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);
        let reader = self
            .table
            .reader_builder()
            .with_batch_size(chunk_size)
            .build();
        let task = self.tasks[partition].clone();
        let stream = reader
            .read(tokio_stream::once(Ok(task)).boxed())
            .await

Review comment (suggestion):
For position delete scans, the `project_field_ids` is being cleared (set to empty Vec). This means no fields will be projected. However, position delete files typically need at least the file path and position columns to identify which rows to delete. Consider whether clearing `project_field_ids` is correct, or if specific fields should be projected for position delete scans.
```suggestion
                            // For position delete files, project file_path (field id 0) and position (field id 1)
                            task.project_field_ids = vec![0, 1];
```

Code after:
            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 reader = self
            .table
            .reader_builder()
            .with_batch_size(chunk_size)
            .build();

        for task in &self.tasks[partition] {
            let stream = reader
                .clone()