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

Code before:
        filters: &[&Expr],
    ) -> Result<Vec<TableProviderFilterPushDown>, datafusion::error::DataFusionError> {
        Ok(vec![TableProviderFilterPushDown::Inexact; filters.len()])
    }
}

impl IcebergTableProvider {
    pub fn from_logical_plan(plan: &LogicalIcebergScan) -> RwResult<Self> {
        let source = plan.source_catalog().ok_or_else(|| {
            RwError::from(ErrorCode::InternalError(
                "Iceberg source catalog is missing".into(),
            ))
        })?;
        let config = ConnectorProperties::extract(source.with_properties.clone(), false)?;
        let ConnectorProperties::Iceberg(iceberg_properties) = config else {
            return Err(RwError::from(ErrorCode::ConnectorError(
                "Expected iceberg connector properties".into(),
            )));
        };
        let iceberg_properties = Arc::from(iceberg_properties);

        let arrow_fields: ArrowFields = plan
            .core
            .column_catalog
            .iter()
            .map(|column| {
                let column_desc = &column.column_desc;
                let field = IcebergArrowConvert
                    .to_arrow_field(&column_desc.name, &column_desc.data_type)?;
                Ok(field.with_nullable(column_desc.nullable))
            })
            .collect::<Result<_, ArrayError>>()?;
        let arrow_schema = Arc::new(ArrowSchema::new(arrow_fields));

        Ok(Self {
            iceberg_properties,
            arrow_schema,
            snapshot_id: plan.snapshot_id,
            iceberg_scan_type: plan.iceberg_scan_type,
        })
    }
}


Review comment (suggestion):
Removing the filter for hidden columns means that Iceberg metadata columns (sequence number, file path, file position) will now be included in the arrow schema. This seems intentional given the new logic in `IcebergScan::new()` that checks for and handles these columns (lines 149-163). However, please verify this is the desired behavior and that these hidden columns should be exposed in the DataFusion schema.
```suggestion
            .iter()
            .filter(|column| !column.is_hidden())
```

Code after:
        &self,
        filters: &[&Expr],
    ) -> Result<Vec<TableProviderFilterPushDown>, datafusion::error::DataFusionError> {
        Ok(vec![TableProviderFilterPushDown::Inexact; filters.len()])
    }
}

impl IcebergTableProvider {
    pub fn from_logical_plan(plan: &LogicalIcebergScan) -> RwResult<Self> {
        let source = plan.source_catalog().ok_or_else(|| {
            RwError::from(ErrorCode::InternalError(
                "Iceberg source catalog is missing".into(),
            ))
        })?;
        let config = ConnectorProperties::extract(source.with_properties.clone(), false)?;
        let ConnectorProperties::Iceberg(iceberg_properties) = config else {
            return Err(RwError::from(ErrorCode::ConnectorError(
                "Expected iceberg connector properties".into(),
            )));
        };
        let iceberg_properties = Arc::from(iceberg_properties);

        let arrow_fields: ArrowFields = plan
            .core
            .column_catalog
            .iter()
            .map(|column| {
                let column_desc = &column.column_desc;
                let field = IcebergArrowConvert
                    .to_arrow_field(&column_desc.name, &column_desc.data_type)?;
                Ok(field.with_nullable(column_desc.nullable))
            })
            .collect::<Result<_, ArrayError>>()?;
        let arrow_schema = Arc::new(ArrowSchema::new(arrow_fields));

        Ok(Self {
            iceberg_properties,
            arrow_schema,
            snapshot_id: plan.snapshot_id,
            iceberg_scan_type: plan.iceberg_scan_type,
        })
    }
}