You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/28 03:14:11 UTC
(datafusion) branch gh-readonly-queue/main/pr-25780-59d74cabfdcb358e93133b498d33bbc94ca9c934 created (now b131db538a)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-25780-59d74cabfdcb358e93133b498d33bbc94ca9c934
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at b131db538a fix: derive scan equivalences only from filters the source applies exactly (#25780)
This branch includes the following new commits:
new b131db538a fix: derive scan equivalences only from filters the source applies exactly (#25780)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: fix: derive scan equivalences only from filters the source applies exactly (#25780)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25780-59d74cabfdcb358e93133b498d33bbc94ca9c934
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit b131db538a36a24d098f3c3de617b2d4d148839a
Author: Adrian Garcia Badaracco <17...@users.noreply.github.com>
AuthorDate: Mon Sep 28 03:13:47 2026 +0000
fix: derive scan equivalences only from filters the source applies exactly (#25780)
## Which issue does this PR close?
- Closes https://github.com/apache/datafusion/issues/25779.
## Rationale for this change
With `datafusion.execution.parquet.pushdown_filters = false` (the
default), a Parquet scan claimed equivalence properties from a predicate
that it uses only for pruning. An order-preserving repartition directly
on the scan then merged on the wrong keys, and `ORDER BY` and `ORDER BY
... LIMIT` returned wrong results. The issue has the reproducer.
## What changes are included in this PR?
| Change | File |
| --- | --- |
| New `FileSource::exact_filter()`: the part of `filter()` that every
output row satisfies. Default `None`. | `datasource/src/file.rs` |
| `FileScanConfig::eq_properties` uses `exact_filter()` instead of
`filter()`. `filter()` is unchanged, because `statistics()` still needs
it to mark row counts inexact after pruning. |
`datasource/src/file_scan_config/mod.rs` |
| `ParquetSource::exact_filter()` returns `None` when `pushdown_filters`
is off. When it is on, it returns only the conjuncts that pass
`can_expr_be_pushed_down_with_schemas`, the same test
`try_pushdown_filters` uses before it replies `PushedDown::Yes`. |
`datasource-parquet/src/source.rs` |
| Upgrade guide entry for custom `FileSource` implementations. |
`upgrading/56.0.0.md` |
The behavior from https://github.com/apache/datafusion/pull/16686 /
https://github.com/apache/datafusion/issues/16563 is kept: with
`pushdown_filters = true` the scan still reports the equivalences, and
the existing plans do not change.
## What is the testing strategy for this PR?
- New `sqllogictest` cases in `push_down_filter_parquet.slt`: one
`EXPLAIN` (the repartition now merges on `a, b`) and four result checks
(`ORDER BY`, `ORDER BY ... LIMIT`, `GROUP BY ... ORDER BY`, and the `a =
b` form). Without the fix, all 5 fail.
- `cte.slt`: one expected plan changes. The scan no longer claims that
`k` is constant from its pruning-only `k = 2` predicate, so it reports
its real `[k]` ordering, and the single-input `RepartitionExec` now
shows `maintains_sort_order=true`.
- Differential check against DuckDB: 14 query shapes (order by, top-k,
group by, window, hash join with dynamic filter, sort-merge join,
`DISTINCT ON`, `first_value`, `a = b`) × 18 configurations
(`pushdown_filters`, `prefer_existing_sort`, `repartition_file_scans`,
`target_partitions`, `prefer_hash_join`). 15 failures before this
change, 0 after.
## Are there any user-facing changes?
Queries return correct results. A custom `FileSource` that applies its
filter to every row must implement `exact_filter` to keep the
equivalence properties of its scan (see the upgrade guide). This is a
new trait method with a default, so it is not a breaking change at
compile time.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <no...@anthropic.com>
---
datafusion/datasource-parquet/src/source.rs | 18 ++++
datafusion/datasource/src/file.rs | 15 +++
datafusion/datasource/src/file_scan_config/mod.rs | 4 +-
datafusion/sqllogictest/test_files/cte.slt | 2 +-
.../test_files/push_down_filter_parquet.slt | 108 +++++++++++++++++++++
docs/source/library-user-guide/upgrading/56.0.0.md | 28 ++++++
6 files changed, 173 insertions(+), 2 deletions(-)
diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs
index 32fc8c9f3d..42e7116dff 100644
--- a/datafusion/datasource-parquet/src/source.rs
+++ b/datafusion/datasource-parquet/src/source.rs
@@ -50,6 +50,7 @@ use datafusion_datasource::file_scan_config::FileScanConfig;
use datafusion_functions::core::file_row_index::FileRowIndexFunc;
use datafusion_physical_expr::expressions::{Column, DynamicFilterTracking};
use datafusion_physical_expr::projection::ProjectionExprs;
+use datafusion_physical_expr::utils::split_conjunction;
use datafusion_physical_expr::{EquivalenceProperties, conjunction};
use datafusion_physical_expr_adapter::DefaultPhysicalExprAdapterFactory;
use datafusion_physical_expr_adapter::rewrite::{
@@ -702,6 +703,23 @@ impl FileSource for ParquetSource {
self.predicate.clone()
}
+ /// The predicate is applied to every row only when filter pushdown is
+ /// enabled, and then only the conjuncts that can become a `RowFilter`.
+ /// Otherwise the predicate is used only for pruning.
+ fn exact_filter(&self) -> Option<Arc<dyn PhysicalExpr>> {
+ if !self.pushdown_filters() {
+ return None;
+ }
+ let predicate = self.predicate.as_ref()?;
+ let pushable_schema = self.table_schema.schema_without_virtual_columns();
+ let exact = split_conjunction(predicate)
+ .into_iter()
+ .filter(|expr| can_expr_be_pushed_down_with_schemas(expr, pushable_schema))
+ .cloned()
+ .collect::<Vec<_>>();
+ (!exact.is_empty()).then(|| conjunction(exact))
+ }
+
fn with_batch_size(&self, batch_size: usize) -> Arc<dyn FileSource> {
let mut conf = self.clone();
conf.batch_size = Some(batch_size);
diff --git a/datafusion/datasource/src/file.rs b/datafusion/datasource/src/file.rs
index f1a94f2e12..d0a1595692 100644
--- a/datafusion/datasource/src/file.rs
+++ b/datafusion/datasource/src/file.rs
@@ -114,6 +114,21 @@ pub trait FileSource: Any + Send + Sync {
None
}
+ /// Returns the part of [`Self::filter`] that every output row is
+ /// guaranteed to satisfy.
+ ///
+ /// [`FileScanConfig`] derives equivalence properties (constants and equal
+ /// columns) from this filter. A filter that the source uses only to skip
+ /// data (for example, row group or page pruning with statistics) does not
+ /// remove every non-matching row, so it must not be returned here.
+ ///
+ /// Returns `None` by default.
+ ///
+ /// [`FileScanConfig`]: crate::file_scan_config::FileScanConfig
+ fn exact_filter(&self) -> Option<Arc<dyn PhysicalExpr>> {
+ None
+ }
+
/// Return the projection that will be applied to the output stream on top
/// of [`Self::table_schema`].
///
diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs
index 61b41fd1e9..18f952172a 100644
--- a/datafusion/datasource/src/file_scan_config/mod.rs
+++ b/datafusion/datasource/src/file_scan_config/mod.rs
@@ -910,7 +910,9 @@ impl DataSource for FileScanConfig {
)
.with_constraints(self.constraints.clone());
- if let Some(filter) = self.file_source.filter() {
+ // Only a filter that the source applies exactly gives valid equivalences.
+ // A filter that is used only for pruning lets non-matching rows through.
+ if let Some(filter) = self.file_source.exact_filter() {
// We need to remap column indexes to match the projected schema since that's what the equivalence properties deal with.
// Note that this will *ignore* any non-projected columns: these don't factor into ordering / equivalence.
match Self::add_filter_equivalence_info(&filter, &mut eq_properties, schema) {
diff --git a/datafusion/sqllogictest/test_files/cte.slt b/datafusion/sqllogictest/test_files/cte.slt
index 289134d9f1..376fb82a27 100644
--- a/datafusion/sqllogictest/test_files/cte.slt
+++ b/datafusion/sqllogictest/test_files/cte.slt
@@ -1700,7 +1700,7 @@ physical_plan
08)------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
09)--------------WorkTableExec: name=r
10)--FilterExec: k@0 = 2, projection=[v2@1]
-11)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+11)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, maintains_sort_order=true
12)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/cte/test.parquet]]}, projection=[k, v2], output_ordering=[k@0 ASC NULLS LAST], file_type=parquet, predicate=k@0 = 2, pruning_predicate=k_null_count@2 != row_count@3 AND k_min@0 <= 2 AND 2 <= k_max@1, required_guarantees=[k in (2)]
query II
diff --git a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt
index 8dd763205f..674f866fc6 100644
--- a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt
+++ b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt
@@ -1336,3 +1336,111 @@ RESET datafusion.execution.parquet.pushdown_filters;
statement ok
drop table t;
+
+# A predicate that the Parquet scan keeps only for pruning
+# (pushdown_filters = false) must not make the scan claim equivalences.
+# Each file is sorted by (a, b). If the scan claims that `a` is constant,
+# the order-preserving repartition below the FilterExec merges on `b` only
+# and the rows with a = 5 come out in the wrong order.
+
+statement ok
+COPY (SELECT * FROM (VALUES (3, 9), (5, 1)) v(a, b))
+TO 'test_files/scratch/push_down_filter_parquet/pruning_only_eq/f1.parquet'
+STORED AS PARQUET;
+
+statement ok
+COPY (SELECT * FROM (VALUES (5, 2)) v(a, b))
+TO 'test_files/scratch/push_down_filter_parquet/pruning_only_eq/f2.parquet'
+STORED AS PARQUET;
+
+statement ok
+CREATE EXTERNAL TABLE pruning_only_eq (a INT, b INT)
+STORED AS PARQUET
+LOCATION 'test_files/scratch/push_down_filter_parquet/pruning_only_eq/'
+WITH ORDER (a ASC, b ASC);
+
+statement ok
+set datafusion.optimizer.repartition_file_scans = false;
+
+statement ok
+set datafusion.optimizer.prefer_existing_sort = true;
+
+statement ok
+set datafusion.execution.parquet.pushdown_filters = false;
+
+statement ok
+set datafusion.explain.physical_plan_only = true;
+
+# The repartition merges on (a, b), the real file ordering
+query TT
+EXPLAIN SELECT b FROM pruning_only_eq WHERE a = 5 ORDER BY b LIMIT 1;
+----
+physical_plan
+01)SortPreservingMergeExec: [b@0 ASC NULLS LAST], fetch=1
+02)--FilterExec: a@0 = 5, projection=[b@1], fetch=1
+03)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=2, preserve_order=true, sort_exprs=a@0 ASC NULLS LAST, b@1 ASC NULLS LAST
+04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/pruning_only_eq/f1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/pruning_only_eq/f2.parquet]]}, projection=[a, b], output_ordering=[a@0 ASC NULLS LAST, b@1 ASC NULLS LAST], file_type=parquet, predicate=a@0 = 5, pruning_predicate=a_null_count@2 != row_count@3 AND a_min@0 <= 5 AND 5 <= a_max@1, required_guarante [...]
+
+query I
+SELECT b FROM pruning_only_eq WHERE a = 5 ORDER BY b LIMIT 1;
+----
+1
+
+query II
+SELECT a, b FROM pruning_only_eq WHERE a = 5 ORDER BY b;
+----
+5 1
+5 2
+
+statement ok
+set datafusion.optimizer.prefer_existing_sort = false;
+
+query III
+SELECT a, b, count(*) FROM pruning_only_eq WHERE a = 5 GROUP BY a, b ORDER BY b;
+----
+5 1 1
+5 2 1
+
+# The same filter, with columns `a` and `b` in the opposite roles
+statement ok
+set datafusion.optimizer.prefer_existing_sort = true;
+
+statement ok
+COPY (SELECT * FROM (VALUES (1, 9), (3, 3)) v(a, b))
+TO 'test_files/scratch/push_down_filter_parquet/pruning_only_col_eq/f1.parquet'
+STORED AS PARQUET;
+
+statement ok
+COPY (SELECT * FROM (VALUES (5, 5)) v(a, b))
+TO 'test_files/scratch/push_down_filter_parquet/pruning_only_col_eq/f2.parquet'
+STORED AS PARQUET;
+
+statement ok
+CREATE EXTERNAL TABLE pruning_only_col_eq (a INT, b INT)
+STORED AS PARQUET
+LOCATION 'test_files/scratch/push_down_filter_parquet/pruning_only_col_eq/'
+WITH ORDER (a ASC);
+
+query II
+SELECT a, b FROM pruning_only_col_eq WHERE a = b ORDER BY a;
+----
+3 3
+5 5
+
+statement ok
+drop table pruning_only_eq;
+
+statement ok
+drop table pruning_only_col_eq;
+
+statement ok
+RESET datafusion.optimizer.repartition_file_scans;
+
+statement ok
+RESET datafusion.optimizer.prefer_existing_sort;
+
+statement ok
+RESET datafusion.execution.parquet.pushdown_filters;
+
+statement ok
+RESET datafusion.explain.physical_plan_only;
diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md
index cad4bc71bc..79067402a7 100644
--- a/docs/source/library-user-guide/upgrading/56.0.0.md
+++ b/docs/source/library-user-guide/upgrading/56.0.0.md
@@ -632,3 +632,31 @@ Wire compatibility is directional:
- A 55.0 reader must not consume an alias-preserving 56.0 MERGE payload. It
ignores the unknown field but cannot preserve the qualifier required by the
expressions, which can cause resolution failure or incorrect rebinding.
+
+### `FileSource::exact_filter` controls scan equivalence properties
+
+`FileScanConfig` no longer derives equivalence properties (constant columns
+and equal columns) from `FileSource::filter`. It uses the new
+`FileSource::exact_filter` method instead. This method returns the part of the
+filter that every output row satisfies. A filter that the source uses only for
+pruning (for example, Parquet row group statistics when
+`datafusion.execution.parquet.pushdown_filters` is `false`) lets
+non-matching rows through, and equivalences derived from it gave wrong query
+results.
+
+`exact_filter` returns `None` by default. `ParquetSource` returns the pushed-down
+conjuncts when `pushdown_filters` is enabled.
+
+**Migration guide:**
+
+If your custom `FileSource` applies its filter to every row, implement
+`exact_filter` to keep the equivalence properties of the scan:
+
+```rust,ignore
+impl FileSource for MySource {
+ fn exact_filter(&self) -> Option<Arc<dyn PhysicalExpr>> {
+ self.filter()
+ }
+ // ...
+}
+```
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org