@@ -44,8 +44,8 @@ use datafusion_common::encryption::FileDecryptionProperties;
 use datafusion_common::stats::Precision;
 use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion};
 use datafusion_common::{
-    ColumnStatistics, DataFusionError, Result, ScalarValue, Statistics, exec_err,
-    not_impl_err,
+    ColumnStatistics, DataFusionError, HashSet, Result, ScalarValue, Statistics,
+    exec_err, not_impl_err,
 };
 use datafusion_datasource::{PartitionedFile, TableSchema};
 use datafusion_physical_expr::expressions::Column;
@@ -1411,7 +1411,7 @@ fn append_fields(base: &SchemaRef, extra: &[FieldRef]) -> SchemaRef {
 /// stringly-typed allowlist here.
 fn validate_supported_virtual_columns(virtual_columns: &[FieldRef]) -> Result<()> {
     for field in virtual_columns {
-        ParquetVirtualColumn::try_from(Arc::clone(field))?;
+        ParquetVirtualColumn::try_from(field)?;
     }
     Ok(())
 }
@@ -1425,27 +1425,24 @@ fn validate_supported_virtual_columns(virtual_columns: &[FieldRef]) -> Result<()
 /// inside a row filter. Silently dropping such a predicate would produce wrong
 /// results, so we fail loudly here.
 ///
-/// Callers constructing a `ParquetSource` should rely on
-/// `FileSource::try_pushdown_filters` (invoked by the `FilterPushdown` physical
-/// optimizer rule) to classify filters correctly: filters referencing virtual
-/// columns are reported as `PushedDown::No` and stay in the enclosing
-/// `FilterExec`, while the scan emits the virtual columns for the filter to
-/// consume. Callers building plans manually must keep the `FilterExec` above
-/// the `DataSourceExec` themselves rather than setting the predicate on
-/// `ParquetSource` with pushdown enabled.
+/// `ParquetSource::try_pushdown_filters` already classifies virtual-column
+/// filters as `PushedDown::No` so the `FilterPushdown` optimizer leaves them
+/// above the scan; this check is defense-in-depth for callers that build plans
+/// manually and set `with_pushdown_filters(true)` alongside a predicate
+/// referencing virtual columns.
 fn validate_predicate_does_not_reference_virtual_columns(
     predicate: &Arc<dyn PhysicalExpr>,
     virtual_columns: &[FieldRef],
 ) -> Result<()> {
     if virtual_columns.is_empty() {
         return Ok(());
     }
+    let virtual_names: HashSet<&str> =
+        virtual_columns.iter().map(|f| f.name().as_str()).collect();
     let mut offender: Option<String> = None;
     predicate.apply(|node: &Arc<dyn PhysicalExpr>| {
         if let Some(column) = node.downcast_ref::<Column>()
-            && virtual_columns
-                .iter()
-                .any(|f| f.name().as_str() == column.name())
+            && virtual_names.contains(column.name())
         {
             offender = Some(column.name().to_string());
             return Ok(TreeNodeRecursion::Stop);
@@ -1455,14 +1452,9 @@ fn validate_predicate_does_not_reference_virtual_columns(
     if let Some(name) = offender {
         return not_impl_err!(
             "Predicate references virtual column '{name}' while \
-             pushdown_filters=true. DataFusion cannot push predicates on \
-             virtual columns into the Parquet reader: arrow-rs's RowFilter \
-             operates on a ProjectionMask over file leaves, and virtual \
-             columns (e.g. row_number) are synthesized by the reader after \
-             filter evaluation. Either leave the filter in a FilterExec above \
-             the scan (this is what `ParquetSource::try_pushdown_filters` / \
-             the FilterPushdown optimizer rule will do automatically), or \
-             disable pushdown via `with_pushdown_filters(false)`."
+             pushdown_filters=true; predicates on virtual columns must be \
+             evaluated above the scan. Disable filter pushdown or leave the \
+             filter above the scan."
         );
     }
     Ok(())
@@ -3218,119 +3210,90 @@ mod test {
             );
         }
 
-        #[tokio::test]
-        async fn test_row_index_predicate_pushdown_mixed_or_errors() {
-            // Mixed `row_number = 2 OR value = 4` with pushdown_filters=true.
-            // Silently dropping this in the scan would return all 5 rows; we
-            // instead surface a guidance-rich error so callers realize the
-            // predicate needs to stay above the scan.
-            let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
-            let (file_schema, data_size) =
-                write_grouped_file(&store, "pushdown_mixed.parquet", 1, 5).await;
-
+        /// Build a morselizer + file for a 5-row single-row-group parquet at
+        /// `path`, with a single `row_number` virtual column and the given
+        /// physical predicate applied to
+        /// `table_schema = [value(0), row_number(1)]`.
+        async fn build_pushdown_morselizer(
+            store: &Arc<dyn ObjectStore>,
+            path: &str,
+            predicate_expr: datafusion_expr::Expr,
+            pushdown_filters: bool,
+        ) -> (ParquetMorselizer, PartitionedFile) {
+            let (file_schema, data_size) = write_grouped_file(store, path, 1, 5).await;
             let rn_field = row_number_field("row_number", false);
             let table_schema = TableSchema::from_file_schema(Arc::clone(&file_schema))
                 .with_virtual_columns(vec![Arc::clone(&rn_field)]);
             let projection =
                 ProjectionExprs::from_indices(&[0, 1], table_schema.table_schema());
-
-            let expr = col("row_number")
-                .eq(lit(2i64))
-                .or(col("value").eq(lit(4i64)));
-            let predicate = logical2physical(&expr, table_schema.table_schema());
+            let predicate =
+                logical2physical(&predicate_expr, table_schema.table_schema());
 
             let morselizer = ParquetMorselizerBuilder::new()
-                .with_store(Arc::clone(&store))
+                .with_store(Arc::clone(store))
                 .with_table_schema(table_schema)
                 .with_projection(projection)
                 .with_predicate(predicate)
-                .with_pushdown_filters(true)
+                .with_pushdown_filters(pushdown_filters)
                 .build();
 
-            let file = PartitionedFile::new(
-                "pushdown_mixed.parquet".to_string(),
-                u64::try_from(data_size).unwrap(),
-            );
+            let file =
+                PartitionedFile::new(path.to_string(), u64::try_from(data_size).unwrap());
+            (morselizer, file)
+        }
+
+        #[tokio::test]
+        async fn test_row_index_predicate_pushdown_mixed_or_errors() {
+            // Silent drop in the scan would return all 5 rows; we want a loud
+            // error instead.
+            let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+            let expr = col("row_number")
+                .eq(lit(2i64))
+                .or(col("value").eq(lit(4i64)));
+            let (morselizer, file) =
+                build_pushdown_morselizer(&store, "pushdown_mixed.parquet", expr, true)
+                    .await;
+
             let err = morselizer.plan_file(file).unwrap_err();
             let msg = err.to_string();
             assert!(
                 msg.contains("row_number"),
                 "error should name the offending virtual column, got: {msg}"
             );
             assert!(
-                msg.contains("arrow-rs") && msg.contains("RowFilter"),
-                "error should explain the arrow-rs limitation, got: {msg}"
-            );
-            assert!(
-                msg.contains("FilterExec")
-                    || msg.contains("FilterPushdown")
-                    || msg.contains("try_pushdown_filters"),
-                "error should point at the correct API path, got: {msg}"
+                msg.contains("virtual column") && msg.contains("pushdown"),
+                "error should explain the virtual-column + pushdown context, \
+                 got: {msg}"
             );
         }
 
         #[tokio::test]
         async fn test_row_index_predicate_pushdown_virtual_only_errors() {
-            // Predicate referencing only a virtual column with pushdown_filters=true.
             let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
-            let (file_schema, data_size) =
-                write_grouped_file(&store, "pushdown_virtual_only.parquet", 1, 5).await;
-
-            let rn_field = row_number_field("row_number", false);
-            let table_schema = TableSchema::from_file_schema(Arc::clone(&file_schema))
-                .with_virtual_columns(vec![Arc::clone(&rn_field)]);
-            let projection =
-                ProjectionExprs::from_indices(&[0, 1], table_schema.table_schema());
-
             let expr = col("row_number").eq(lit(2i64));
-            let predicate = logical2physical(&expr, table_schema.table_schema());
-
-            let morselizer = ParquetMorselizerBuilder::new()
-                .with_store(Arc::clone(&store))
-                .with_table_schema(table_schema)
-                .with_projection(projection)
-                .with_predicate(predicate)
-                .with_pushdown_filters(true)
-                .build();
+            let (morselizer, file) = build_pushdown_morselizer(
+                &store,
+                "pushdown_virtual_only.parquet",
+                expr,
+                true,
+            )
+            .await;
 
-            let file = PartitionedFile::new(
-                "pushdown_virtual_only.parquet".to_string(),
-                u64::try_from(data_size).unwrap(),
-            );
             let err = morselizer.plan_file(file).unwrap_err();
             assert!(err.to_string().contains("row_number"));
         }
 
         #[tokio::test]
         async fn test_row_index_predicate_allowed_when_pushdown_disabled() {
-            // Same predicate, but pushdown_filters=false. The predicate is only
-            // used for stats pruning (which is a no-op for row_number) and must
-            // NOT error.
+            // Guards the `pushdown_filters=false` path: the predicate is only
+            // used for stats pruning (a no-op for row_number) and must not
+            // trip the pushdown-guard error.
             let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
-            let (file_schema, data_size) =
-                write_grouped_file(&store, "pushdown_off.parquet", 1, 5).await;
-
-            let rn_field = row_number_field("row_number", false);
-            let table_schema = TableSchema::from_file_schema(Arc::clone(&file_schema))
-                .with_virtual_columns(vec![Arc::clone(&rn_field)]);
-            let projection =
-                ProjectionExprs::from_indices(&[0, 1], table_schema.table_schema());
-
             let expr = col("row_number").eq(lit(2i64));
-            let predicate = logical2physical(&expr, table_schema.table_schema());
-
-            let morselizer = ParquetMorselizerBuilder::new()
-                .with_store(Arc::clone(&store))
-                .with_table_schema(table_schema)
-                .with_projection(projection)
-                .with_predicate(predicate)
-                .build(); // pushdown_filters defaults to false
+            let (morselizer, file) =
+                build_pushdown_morselizer(&store, "pushdown_off.parquet", expr, false)
+                    .await;
 
-            let file = PartitionedFile::new(
-                "pushdown_off.parquet".to_string(),
-                u64::try_from(data_size).unwrap(),
-            );
-            // Should open cleanly and return all rows (no row filter applied).
             let stream = open_file(&morselizer, file).await.unwrap();
             let (_batches, rows) = count_batches_and_rows(stream).await;
             assert_eq!(rows, 5);
 
@@ -981,8 +981,6 @@ mod tests {
 
         let source = ParquetSource::new(table_schema).with_pushdown_filters(true);
 
-        // Three filters: pure file-col (pushable), pure virtual (not pushable),
-        // and a mixed OR conjunct (not pushable).
         let full_schema = source.table_schema.table_schema();
 
         let pushable = logical2physical(&col("value").eq(logical_lit(1i64)), full_schema);
 
@@ -17,29 +17,21 @@
 
 //! Typed wrapper for parquet virtual columns.
 //!
-//! arrow-rs identifies virtual columns (produced by the reader, not read from
-//! the file) via arrow extension types carried on the `FieldRef`. This module
-//! lifts that contract into the type system: [`ParquetVirtualColumn`] is the
-//! set of virtual columns DataFusion currently supports, and its
-//! [`TryFrom<FieldRef>`] impl validates the extension type at the boundary so
-//! downstream code can pattern-match on variants rather than string-compare.
+//! arrow-rs identifies virtual columns via arrow extension types carried on
+//! the `FieldRef`. [`ParquetVirtualColumn`] lifts that contract into the type
+//! system so callers validate at the boundary (via `TryFrom<&FieldRef>`)
+//! rather than string-comparing extension-type names deep inside the reader.
 
 use arrow::datatypes::FieldRef;
 use arrow_schema::extension::ExtensionType;
 use datafusion_common::{DataFusionError, Result, not_impl_err};
 use parquet::arrow::RowNumber;
+use std::sync::Arc;
 
 /// A parquet virtual column validated to have a supported arrow extension
 /// type.
 ///
-/// Virtual columns are synthesized by the parquet reader (e.g.
-/// [`RowNumber`] produces the absolute row number within the file) rather
-/// than being read from column data. Because arrow-rs identifies them via
-/// extension types on `FieldRef`, it would be easy to pass an unsupported or
-/// misspelled extension type and have the error surface deep in the reader
-/// pipeline; this enum forces the check to the construction boundary.
-///
-/// Construct via [`TryFrom<FieldRef>`]; add a new variant (and update the
+/// Construct via [`TryFrom<&FieldRef>`]; add a new variant (and update the
 /// `TryFrom` impl) when DataFusion gains support for another arrow-rs virtual
 /// extension type.
 #[derive(Debug, Clone)]
@@ -50,9 +42,6 @@ pub enum ParquetVirtualColumn {
 }
 
 impl ParquetVirtualColumn {
-    /// The underlying arrow field, with its extension type and metadata
-    /// preserved. Suitable for passing to
-    /// `ArrowReaderOptions::with_virtual_columns`.
     pub fn field(&self) -> &FieldRef {
         match self {
             Self::RowNumber(field) => field,
@@ -68,10 +57,10 @@ impl From<ParquetVirtualColumn> for FieldRef {
     }
 }
 
-impl TryFrom<FieldRef> for ParquetVirtualColumn {
+impl TryFrom<&FieldRef> for ParquetVirtualColumn {
     type Error = DataFusionError;
 
-    fn try_from(field: FieldRef) -> Result<Self> {
+    fn try_from(field: &FieldRef) -> Result<Self> {
         let Some(name) = field.extension_type_name() else {
             return not_impl_err!(
                 "Virtual column '{}' is missing an Arrow extension type; \
@@ -81,7 +70,7 @@ impl TryFrom<FieldRef> for ParquetVirtualColumn {
             );
         };
         match name {
-            n if n == RowNumber::NAME => Ok(Self::RowNumber(field)),
+            n if n == RowNumber::NAME => Ok(Self::RowNumber(Arc::clone(field))),
             other => not_impl_err!(
                 "Virtual column '{}' uses unsupported Arrow extension type '{}'; \
                  supported types: [{}]. Add a ParquetVirtualColumn variant and \
@@ -98,28 +87,25 @@ impl TryFrom<FieldRef> for ParquetVirtualColumn {
 mod tests {
     use super::*;
     use arrow::datatypes::{DataType, Field};
-    use std::sync::Arc;
 
     #[test]
     fn row_number_field_converts() {
         let field: FieldRef = Arc::new(
             Field::new("row_number", DataType::Int64, false)
                 .with_extension_type(RowNumber),
         );
-        let col =
-            ParquetVirtualColumn::try_from(Arc::clone(&field)).expect("valid row_number");
+        let col = ParquetVirtualColumn::try_from(&field).expect("valid row_number");
         assert!(matches!(col, ParquetVirtualColumn::RowNumber(_)));
         assert_eq!(col.field().name(), "row_number");
     }
 
     #[test]
     fn missing_extension_type_rejected() {
         let field: FieldRef = Arc::new(Field::new("plain", DataType::Int64, false));
-        let err = ParquetVirtualColumn::try_from(field).unwrap_err();
-        let msg = err.to_string();
+        let err = ParquetVirtualColumn::try_from(&field).unwrap_err();
         assert!(
-            msg.contains("missing an Arrow extension type"),
-            "got: {msg}"
+            err.to_string().contains("missing an Arrow extension type"),
+            "got: {err}"
         );
     }
 
@@ -130,11 +116,10 @@ mod tests {
             Field::new("row_group_index", DataType::Int64, false)
                 .with_extension_type(parquet::arrow::RowGroupIndex),
         );
-        let err = ParquetVirtualColumn::try_from(field).unwrap_err();
-        let msg = err.to_string();
+        let err = ParquetVirtualColumn::try_from(&field).unwrap_err();
         assert!(
-            msg.contains("parquet.virtual.row_group_index"),
-            "error should name the offending extension type, got: {msg}"
+            err.to_string().contains("parquet.virtual.row_group_index"),
+            "error should name the offending extension type, got: {err}"
         );
     }
 }
@@ -259,9 +259,7 @@ impl TableSchema {
         if self.virtual_columns.is_empty() {
             return Arc::clone(&self.table_schema);
         }
-        let mut builder = SchemaBuilder::from(self.file_schema.as_ref());
-        builder.extend(self.table_partition_cols.iter().cloned());
-        Arc::new(builder.finish())
+        build_table_schema(&self.file_schema, &self.table_partition_cols, &[])
     }
 }
Original file line number	Diff line number	Diff line change
`@@ -259,9 +259,7 @@ impl TableSchema {`
`259`	`259`	`if self.virtual_columns.is_empty() {`
`260`	`260`	`return Arc::clone(&self.table_schema);`
`261`	`261`	`}`
`262`		`- let mut builder = SchemaBuilder::from(self.file_schema.as_ref());`
`263`		`- builder.extend(self.table_partition_cols.iter().cloned());`
`264`		`- Arc::new(builder.finish())`
	`262`	`+ build_table_schema(&self.file_schema, &self.table_partition_cols, &[])`
`265`	`263`	`}`
`266`	`264`	`}`
`267`	`265`