Revert use file schema in parquet pruning (#16086)

* wip

* comment

* Update datafusion/core/src/datasource/physical_plan/parquet.rs

* remove prints

* better test

* fmt

Author: adriangb

datafusion/core/src/datasource/physical_plan/parquet.rs

@@ -39,7 +39,7 @@ mod tests {
     use crate::test::object_store::local_unpartitioned_file;
     use arrow::array::{
         ArrayRef, AsArray, Date64Array, Int32Array, Int64Array, Int8Array, StringArray,
-        StructArray,
+        StringViewArray, StructArray,
     };
     use arrow::datatypes::{DataType, Field, Fields, Schema, SchemaBuilder};
     use arrow::record_batch::RecordBatch;
@@ -100,6 +100,7 @@ mod tests {
         predicate: Option<Expr>,
         pushdown_predicate: bool,
         page_index_predicate: bool,
+        bloom_filters: bool,
     }
 
     impl RoundTrip {
@@ -132,6 +133,11 @@ mod tests {
             self
         }
 
+        fn with_bloom_filters(mut self) -> Self {
+            self.bloom_filters = true;
+            self
+        }
+
         /// run the test, returning only the resulting RecordBatches
         async fn round_trip_to_batches(
             self,
@@ -156,10 +162,20 @@ mod tests {
                 source = source
                     .with_pushdown_filters(true)
                     .with_reorder_filters(true);
+            } else {
+                source = source.with_pushdown_filters(false);
             }
 
             if self.page_index_predicate {
                 source = source.with_enable_page_index(true);
+            } else {
+                source = source.with_enable_page_index(false);
+            }
+
+            if self.bloom_filters {
+                source = source.with_bloom_filter_on_read(true);
+            } else {
+                source = source.with_bloom_filter_on_read(false);
             }
 
             source.with_schema(Arc::clone(&file_schema))
@@ -817,7 +833,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn evolved_schema_filter() {
+    async fn evolved_schema_column_order_filter() {
         let c1: ArrayRef =
             Arc::new(StringArray::from(vec![Some("Foo"), None, Some("bar")]));
 
@@ -848,6 +864,88 @@ mod tests {
         assert_eq!(read.len(), 0);
     }
 
+    #[tokio::test]
+    async fn evolved_schema_column_type_filter_strings() {
+        // The table and filter have a common data type, but the file schema differs
+        let c1: ArrayRef =
+            Arc::new(StringViewArray::from(vec![Some("foo"), Some("bar")]));
+        let batch = create_batch(vec![("c1", c1.clone())]);
+
+        let schema = Arc::new(Schema::new(vec![Field::new("c1", DataType::Utf8, false)]));
+
+        // Predicate should prune all row groups
+        let filter = col("c1").eq(lit(ScalarValue::Utf8(Some("aaa".to_string()))));
+        let rt = RoundTrip::new()
+            .with_predicate(filter)
+            .with_schema(schema.clone())
+            .round_trip(vec![batch.clone()])
+            .await;
+        // There should be no predicate evaluation errors
+        let metrics = rt.parquet_exec.metrics().unwrap();
+        assert_eq!(get_value(&metrics, "predicate_evaluation_errors"), 0);
+        assert_eq!(get_value(&metrics, "pushdown_rows_matched"), 0);
+        assert_eq!(rt.batches.unwrap().len(), 0);
+
+        // Predicate should prune no row groups
+        let filter = col("c1").eq(lit(ScalarValue::Utf8(Some("foo".to_string()))));
+        let rt = RoundTrip::new()
+            .with_predicate(filter)
+            .with_schema(schema)
+            .round_trip(vec![batch])
+            .await;
+        // There should be no predicate evaluation errors
+        let metrics = rt.parquet_exec.metrics().unwrap();
+        assert_eq!(get_value(&metrics, "predicate_evaluation_errors"), 0);
+        assert_eq!(get_value(&metrics, "pushdown_rows_matched"), 0);
+        let read = rt
+            .batches
+            .unwrap()
+            .iter()
+            .map(|b| b.num_rows())
+            .sum::<usize>();
+        assert_eq!(read, 2, "Expected 2 rows to match the predicate");
+    }
+
+    #[tokio::test]
+    async fn evolved_schema_column_type_filter_ints() {
+        // The table and filter have a common data type, but the file schema differs
+        let c1: ArrayRef = Arc::new(Int8Array::from(vec![Some(1), Some(2)]));
+        let batch = create_batch(vec![("c1", c1.clone())]);
+
+        let schema =
+            Arc::new(Schema::new(vec![Field::new("c1", DataType::UInt64, false)]));
+
+        // Predicate should prune all row groups
+        let filter = col("c1").eq(lit(ScalarValue::UInt64(Some(5))));
+        let rt = RoundTrip::new()
+            .with_predicate(filter)
+            .with_schema(schema.clone())
+            .round_trip(vec![batch.clone()])
+            .await;
+        // There should be no predicate evaluation errors
+        let metrics = rt.parquet_exec.metrics().unwrap();
+        assert_eq!(get_value(&metrics, "predicate_evaluation_errors"), 0);
+        assert_eq!(rt.batches.unwrap().len(), 0);
+
+        // Predicate should prune no row groups
+        let filter = col("c1").eq(lit(ScalarValue::UInt64(Some(1))));
+        let rt = RoundTrip::new()
+            .with_predicate(filter)
+            .with_schema(schema)
+            .round_trip(vec![batch])
+            .await;
+        // There should be no predicate evaluation errors
+        let metrics = rt.parquet_exec.metrics().unwrap();
+        assert_eq!(get_value(&metrics, "predicate_evaluation_errors"), 0);
+        let read = rt
+            .batches
+            .unwrap()
+            .iter()
+            .map(|b| b.num_rows())
+            .sum::<usize>();
+        assert_eq!(read, 2, "Expected 2 rows to match the predicate");
+    }
+
     #[tokio::test]
     async fn evolved_schema_disjoint_schema_filter() {
         let c1: ArrayRef =
@@ -1748,6 +1846,7 @@ mod tests {
         let rt = RoundTrip::new()
             .with_predicate(filter.clone())
             .with_pushdown_predicate()
+            .with_bloom_filters()
             .round_trip(vec![batch1])
             .await;
 

datafusion/datasource-parquet/src/opener.rs

@@ -55,8 +55,9 @@ pub(super) struct ParquetOpener {
     pub limit: Option<usize>,
     /// Optional predicate to apply during the scan
     pub predicate: Option<Arc<dyn PhysicalExpr>>,
-    /// Schema of the output table
-    pub table_schema: SchemaRef,
+    /// Schema of the output table without partition columns.
+    /// This is the schema we coerce the physical file schema into.
+    pub logical_file_schema: SchemaRef,
     /// Optional hint for how large the initial request to read parquet metadata
     /// should be
     pub metadata_size_hint: Option<usize>,
@@ -104,13 +105,13 @@ impl FileOpener for ParquetOpener {
         let batch_size = self.batch_size;
 
         let projected_schema =
-            SchemaRef::from(self.table_schema.project(&self.projection)?);
+            SchemaRef::from(self.logical_file_schema.project(&self.projection)?);
         let schema_adapter_factory = Arc::clone(&self.schema_adapter_factory);
         let schema_adapter = self
             .schema_adapter_factory
-            .create(projected_schema, Arc::clone(&self.table_schema));
+            .create(projected_schema, Arc::clone(&self.logical_file_schema));
         let predicate = self.predicate.clone();
-        let table_schema = Arc::clone(&self.table_schema);
+        let logical_file_schema = Arc::clone(&self.logical_file_schema);
         let reorder_predicates = self.reorder_filters;
         let pushdown_filters = self.pushdown_filters;
         let coerce_int96 = self.coerce_int96;
@@ -141,17 +142,20 @@ impl FileOpener for ParquetOpener {
                     .await?;
 
             // Note about schemas: we are actually dealing with **3 different schemas** here:
-            // - The table schema as defined by the TableProvider. This is what the user sees, what they get when they `SELECT * FROM table`, etc.
-            // - The "virtual" file schema: this is the table schema minus any hive partition columns and projections. This is what the file schema is coerced to.
+            // - The table schema as defined by the TableProvider.
+            //   This is what the user sees, what they get when they `SELECT * FROM table`, etc.
+            // - The logical file schema: this is the table schema minus any hive partition columns and projections.
+            //   This is what the physicalfile schema is coerced to.
             // - The physical file schema: this is the schema as defined by the parquet file. This is what the parquet file actually contains.
             let mut physical_file_schema = Arc::clone(reader_metadata.schema());
 
             // The schema loaded from the file may not be the same as the
             // desired schema (for example if we want to instruct the parquet
             // reader to read strings using Utf8View instead). Update if necessary
-            if let Some(merged) =
-                apply_file_schema_type_coercions(&table_schema, &physical_file_schema)
-            {
+            if let Some(merged) = apply_file_schema_type_coercions(
+                &logical_file_schema,
+                &physical_file_schema,
+            ) {
                 physical_file_schema = Arc::new(merged);
                 options = options.with_schema(Arc::clone(&physical_file_schema));
                 reader_metadata = ArrowReaderMetadata::try_new(
@@ -178,7 +182,7 @@ impl FileOpener for ParquetOpener {
             // Build predicates for this specific file
             let (pruning_predicate, page_pruning_predicate) = build_pruning_predicates(
                 predicate.as_ref(),
-                &physical_file_schema,
+                &logical_file_schema,
                 &predicate_creation_errors,
             );
 
@@ -215,7 +219,7 @@ impl FileOpener for ParquetOpener {
                 let row_filter = row_filter::build_row_filter(
                     &predicate,
                     &physical_file_schema,
-                    &table_schema,
+                    &logical_file_schema,
                     builder.metadata(),
                     reorder_predicates,
                     &file_metrics,

datafusion/datasource-parquet/src/row_filter.rs

@@ -426,7 +426,7 @@ fn columns_sorted(_columns: &[usize], _metadata: &ParquetMetaData) -> Result<boo
 pub fn build_row_filter(
     expr: &Arc<dyn PhysicalExpr>,
     physical_file_schema: &SchemaRef,
-    table_schema: &SchemaRef,
+    logical_file_schema: &SchemaRef,
     metadata: &ParquetMetaData,
     reorder_predicates: bool,
     file_metrics: &ParquetFileMetrics,
@@ -447,7 +447,7 @@ pub fn build_row_filter(
             FilterCandidateBuilder::new(
                 Arc::clone(expr),
                 Arc::clone(physical_file_schema),
-                Arc::clone(table_schema),
+                Arc::clone(logical_file_schema),
                 Arc::clone(schema_adapter_factory),
             )
             .build(metadata)

datafusion/datasource-parquet/src/source.rs

@@ -481,7 +481,7 @@ impl FileSource for ParquetSource {
                 .expect("Batch size must set before creating ParquetOpener"),
             limit: base_config.limit,
             predicate: self.predicate.clone(),
-            table_schema: Arc::clone(&base_config.file_schema),
+            logical_file_schema: Arc::clone(&base_config.file_schema),
             metadata_size_hint: self.metadata_size_hint,
             metrics: self.metrics().clone(),
             parquet_file_reader_factory,