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

Code before:
    ) -> DFResult<Self> {
        let plan_properties = PlanProperties::new(
            EquivalenceProperties::new(provider.schema()),
            // TODO: determine partitioning
            Partitioning::UnknownPartitioning(1),
            EmissionType::Incremental,
            Boundedness::Bounded,
        );
        let mut column_names: Vec<String> = provider
            .arrow_schema
            .fields()
            .iter()
            .map(|f| f.name().clone())
            .collect();
        let table = provider
            .iceberg_properties
            .load_table()
            .await
            .map_err(to_datafusion_error)?;
        let need_seq_num = column_names
            .iter()
            .any(|name| name == ICEBERG_SEQUENCE_NUM_COLUMN_NAME);
        let need_file_path_and_pos = column_names
            .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,
        };
        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),

Review comment (suggestion):
The condition `matches!(provider.iceberg_scan_type, IcebergScanType::DataScan)` seems incomplete. The logic checks if file path column is requested AND the scan type is DataScan. However, in the `append_metadata` function (line 357-375), file path and position are added whenever `need_file_path_and_pos` is true, regardless of scan type. This could lead to inconsistency - position delete scans might also need file path information. Consider whether this restriction to only DataScan is correct, or if position delete scans should also support these metadata columns.
```suggestion
            .any(|name| name == ICEBERG_FILE_PATH_COLUMN_NAME);
```

Code after:
    ) -> DFResult<Self> {
        let plan_properties = PlanProperties::new(
            EquivalenceProperties::new(provider.schema()),
            // TODO: determine partitioning
            Partitioning::UnknownPartitioning(1),
            EmissionType::Incremental,
            Boundedness::Bounded,
        );
        let mut column_names: Vec<String> = provider
            .arrow_schema
            .fields()
            .iter()
            .map(|f| f.name().clone())
            .collect();
        let table = provider
            .iceberg_properties
            .load_table()
            .await
            .map_err(to_datafusion_error)?;
        let need_seq_num = column_names
            .iter()
            .any(|name| name == ICEBERG_SEQUENCE_NUM_COLUMN_NAME);
        let need_file_path_and_pos = column_names
            .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 {