Skip to content

Fix parquet filter_pushdown: respect parquet filter pushdown config in scan #16646

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alamb merged 2 commits into from
Jul 2, 2025
Merged

Fix parquet filter_pushdown: respect parquet filter pushdown config in scan #16646

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6b49ce7
respect parquet filter pushdown config in scan
adriangb Jul 1, 2025
2b464d1
Add test
adriangb Jul 1, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 9 additions & 4 deletions datafusion/datasource-parquet/src/source.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -343,9 +343,7 @@ impl ParquetSource {
}

/// If true, the predicate will be used during the parquet scan.
/// Defaults to false
///
/// [`Expr`]: datafusion_expr::Expr
/// Defaults to false.
Copy link
Contributor

@alamb alamb Jul 1, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How did this result in wrong results?

Is it because the planner assumed filters would be pushed down but they weren't?

Copy link
Contributor Author

@adriangb adriangb Jul 1, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Well this bit is just a comment 😛

But this produced wrong results because we'd return Supported(filter) up to the parents -> FilterExec removed from the plan but ParquetOpener.pushdown_filters = false -> pruning did not actually happen during the scan.

pub fn with_pushdown_filters(mut self, pushdown_filters: bool) -> Self {
self.table_parquet_options.global.pushdown_filters = pushdown_filters;
self
Expand Down Expand Up @@ -625,7 +623,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.
Copy link
Contributor

@ianthetechie ianthetechie Jul 2, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I had wondered about this; thanks for updating this comment!

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;
Expand All @@ -647,6 +651,7 @@ impl FileSource for ParquetSource {
None => conjunction(allowed_filters.iter().cloned()),
};
source.predicate = Some(predicate);
source = source.with_pushdown_filters(pushdown_filters);
Copy link
Contributor Author

@adriangb adriangb Jul 1, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@alamb this is the relevant LOC that fixes things

let source = Arc::new(source);
// If pushdown_filters is false we tell our parents that they still have to handle the filters,
// even if we updated the predicate to include the filters (they will only be used for stats pruning).
Expand Down
123 changes: 123 additions & 0 deletions datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
----
Expand All @@ -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
Expand Down
Loading