[branch-48] Fix parquet filter_pushdown: respect parquet filter pushdown config i…

datafusion/datasource-parquet/src/source.rs

@@ -344,9 +344,7 @@ impl ParquetSource {
     }
 
     /// If true, the predicate will be used during the parquet scan.
-    /// Defaults to false
-    ///
-    /// [`Expr`]: datafusion_expr::Expr
+    /// Defaults to false.
     pub fn with_pushdown_filters(mut self, pushdown_filters: bool) -> Self {
         self.table_parquet_options.global.pushdown_filters = pushdown_filters;
         self
@@ -617,7 +615,13 @@ impl FileSource for ParquetSource {
         let Some(file_schema) = self.file_schema.clone() else {
             return Ok(FilterPushdownPropagation::unsupported(filters));
         };
-        // Can we push down the filters themselves into the scan or only use stats pruning?
+        // Determine if based on configs we should push filters down.
+        // If either the table / scan itself or the config has pushdown enabled,
+        // we will push down the filters.
+        // If both are disabled, we will not push down the filters.
+        // By default they are both disabled.
+        // Regardless of pushdown, we will update the predicate to include the filters
+        // because even if scan pushdown is disabled we can still use the filters for stats pruning.
         let config_pushdown_enabled = config.execution.parquet.pushdown_filters;
         let table_pushdown_enabled = self.pushdown_filters();
         let pushdown_filters = table_pushdown_enabled || config_pushdown_enabled;
@@ -646,6 +650,7 @@ impl FileSource for ParquetSource {
             None => conjunction(allowed_filters.iter().cloned()),
         };
         source.predicate = Some(predicate);
+        source = source.with_pushdown_filters(pushdown_filters);
         let source = Arc::new(source);
         let filters = PredicateSupports::new(
             allowed_filters

datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt

@@ -75,6 +75,31 @@ NULL
 NULL
 NULL
 
+query T
+select a from t_pushdown where b > 2 ORDER BY a;
+----
+baz
+foo
+NULL
+NULL
+NULL
+
+query TT
+EXPLAIN select a from t where b > 2 ORDER BY a;
+----
+logical_plan
+01)Sort: t.a ASC NULLS LAST
+02)--Projection: t.a
+03)----Filter: t.b > Int32(2)
+04)------TableScan: t projection=[a, b], partial_filters=[t.b > Int32(2)]
+physical_plan
+01)SortPreservingMergeExec: [a@0 ASC NULLS LAST]
+02)--SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]
+03)----CoalesceBatchesExec: target_batch_size=8192
+04)------FilterExec: b@1 > 2, projection=[a@0]
+05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=2
+06)----------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a, b], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
+
 query TT
 EXPLAIN select a from t_pushdown where b > 2 ORDER BY a;
 ----
@@ -88,6 +113,104 @@ physical_plan
 02)--SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]
 03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
 
+# If we set the setting to `true` it override's the table's setting
+statement ok
+set datafusion.execution.parquet.pushdown_filters = true;
+
+query T
+select a from t where b > 2 ORDER BY a;
+----
+baz
+foo
+NULL
+NULL
+NULL
+
+query T
+select a from t_pushdown where b > 2 ORDER BY a;
+----
+baz
+foo
+NULL
+NULL
+NULL
+
+query TT
+EXPLAIN select a from t where b > 2 ORDER BY a;
+----
+logical_plan
+01)Sort: t.a ASC NULLS LAST
+02)--Projection: t.a
+03)----Filter: t.b > Int32(2)
+04)------TableScan: t projection=[a, b], partial_filters=[t.b > Int32(2)]
+physical_plan
+01)SortPreservingMergeExec: [a@0 ASC NULLS LAST]
+02)--SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]
+03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
+
+query TT
+EXPLAIN select a from t_pushdown where b > 2 ORDER BY a;
+----
+logical_plan
+01)Sort: t_pushdown.a ASC NULLS LAST
+02)--Projection: t_pushdown.a
+03)----Filter: t_pushdown.b > Int32(2)
+04)------TableScan: t_pushdown projection=[a, b], partial_filters=[t_pushdown.b > Int32(2)]
+physical_plan
+01)SortPreservingMergeExec: [a@0 ASC NULLS LAST]
+02)--SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]
+03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
+
+# If we reset the default the table created without pushdown goes back to disabling it
+statement ok
+set datafusion.execution.parquet.pushdown_filters = false;
+
+query T
+select a from t where b > 2 ORDER BY a;
+----
+baz
+foo
+NULL
+NULL
+NULL
+
+query T
+select a from t_pushdown where b > 2 ORDER BY a;
+----
+baz
+foo
+NULL
+NULL
+NULL
+
+query TT
+EXPLAIN select a from t where b > 2 ORDER BY a;
+----
+logical_plan
+01)Sort: t.a ASC NULLS LAST
+02)--Projection: t.a
+03)----Filter: t.b > Int32(2)
+04)------TableScan: t projection=[a, b], partial_filters=[t.b > Int32(2)]
+physical_plan
+01)SortPreservingMergeExec: [a@0 ASC NULLS LAST]
+02)--SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]
+03)----CoalesceBatchesExec: target_batch_size=8192
+04)------FilterExec: b@1 > 2, projection=[a@0]
+05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=2
+06)----------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a, b], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
+
+query TT
+EXPLAIN select a from t_pushdown where b > 2 ORDER BY a;
+----
+logical_plan
+01)Sort: t_pushdown.a ASC NULLS LAST
+02)--Projection: t_pushdown.a
+03)----Filter: t_pushdown.b > Int32(2)
+04)------TableScan: t_pushdown projection=[a, b], partial_filters=[t_pushdown.b > Int32(2)]
+physical_plan
+01)SortPreservingMergeExec: [a@0 ASC NULLS LAST]
+02)--SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true]
+03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
 
 # When filter pushdown *is* enabled, ParquetExec can filter exactly,
 # not just metadata, so we expect to see no FilterExec