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/26 00:42:57 UTC
(datafusion) branch gh-readonly-queue/main/pr-25574-e8e41ae958a5e6856ab9b0e7d8fb5deece0a9e91 created (now d6cb2c571b)
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-25574-e8e41ae958a5e6856ab9b0e7d8fb5deece0a9e91
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at d6cb2c571b perf: mirror ASOF equality-key filters to right (#25574)
This branch includes the following new commits:
new d6cb2c571b perf: mirror ASOF equality-key filters to right (#25574)
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: perf: mirror ASOF equality-key filters to right (#25574)
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-25574-e8e41ae958a5e6856ab9b0e7d8fb5deece0a9e91
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit d6cb2c571b010b20444b9e0bbed8e715a2d62d0c
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Sat Sep 26 00:42:32 2026 +0000
perf: mirror ASOF equality-key filters to right (#25574)
## Which issue does this PR close?
- Part of #318.
- Follow-up to #24801.
## Rationale for this change
When a filter selects one ASOF equality-key group on the left, rows in
other right-side groups cannot match any surviving left row. The
broadcast join still sorts and retains those right rows unless the same
key filter is applied to the right input.
## Performance
Q09 has 50M rows on each side across 500K equality groups. `WHERE l.key
= 42` retains one group and produces 100 rows with both revisions. On an
Apple M4 Max, using independent release builds and two 7-iteration runs
per revision, the median elapsed time across all 14 iterations was:
| Revision | Median elapsed |
| --- | ---: |
| `main` without this optimization (`92559afc0`) | 972.1 ms |
| This change (`c7522fa37`) | 78.5 ms |
This selective keyed workload is about **12.4x faster** (91.9% lower
elapsed time). The benchmark file and command were the same for both
binaries:
```shell
target/release/benchmark_runner asof_join --query 9 --iterations 7
```
## What changes are included in this PR?
- Mirror a pushed `left_key = literal` predicate to the corresponding
right key when both join keys are direct columns of the same type.
- Keep the original left filter and ASOF null-padding behavior
unchanged. Other predicate shapes and coerced or expression-based keys
are not inferred.
- Add result and plan coverage in `asof_join.slt`, plus a keyed
selective workload as ASOF benchmark Q09.
## What is the testing strategy for this PR?
The SLT cases check the mirrored plan, matching and unmatched results, a
right-side `IS NULL` filter, and a left disjunction that must not prune
the right input. Q09 runs the user-visible query with 50M rows per side
across 500K groups, selecting one group.
## Are there any user-facing changes?
No API or result changes. Eligible ASOF joins can sort and retain fewer
right-side rows.
---
.../sql_benchmarks/asof_join/asof_join.suite | 5 ++
.../asof_join/benchmarks/q09.benchmark | 25 ++++++++
datafusion/optimizer/src/push_down_filter.rs | 59 +++++++++++++++++-
datafusion/sqllogictest/test_files/asof_join.slt | 71 ++++++++++++++++++++++
4 files changed, 159 insertions(+), 1 deletion(-)
diff --git a/benchmarks/sql_benchmarks/asof_join/asof_join.suite b/benchmarks/sql_benchmarks/asof_join/asof_join.suite
index 12372f5677..9dff295976 100644
--- a/benchmarks/sql_benchmarks/asof_join/asof_join.suite
+++ b/benchmarks/sql_benchmarks/asof_join/asof_join.suite
@@ -5,6 +5,7 @@
# - latest right row at or before, or earliest right row at or after, each left row
# - narrow integer or wide UTF-8 payloads
# - selective left-side predicates that can run before matching
+# - equality-key predicates that can also prune the right input
description = "ASOF join SQL benchmarks"
@@ -24,3 +25,7 @@ description = "Run the pre-sorted keyed ASOF join query."
[[examples]]
command = "cargo run --release --bin benchmark_runner -- asof_join --query 8"
description = "Run the selective left-filter ASOF join query."
+
+[[examples]]
+command = "cargo run --release --bin benchmark_runner -- asof_join --query 9"
+description = "Run the equality-key filter ASOF join query."
diff --git a/benchmarks/sql_benchmarks/asof_join/benchmarks/q09.benchmark b/benchmarks/sql_benchmarks/asof_join/benchmarks/q09.benchmark
new file mode 100644
index 0000000000..95f85044a3
--- /dev/null
+++ b/benchmarks/sql_benchmarks/asof_join/benchmarks/q09.benchmark
@@ -0,0 +1,25 @@
+name Q09
+group asof_join
+
+expect_plan AsOfJoinExec
+
+run
+-- Both sides: 50M rows in 500K equality groups. The left predicate retains
+-- one group; mirroring it prunes the right input before sorting and broadcast.
+WITH left_input AS (
+ SELECT value % 500000 AS key,
+ value / 500000 + 1 AS ts,
+ value AS payload
+ FROM range(50000000)
+),
+right_input AS (
+ SELECT value % 500000 AS key,
+ value / 500000 AS ts,
+ value AS payload
+ FROM range(50000000)
+)
+SELECT l.key, l.ts, l.payload, r.payload AS right_payload
+FROM left_input l
+ASOF JOIN right_input r MATCH_CONDITION (l.ts >= r.ts)
+ON l.key = r.key
+WHERE l.key = 42;
diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs
index 8b090d4c9d..be0d4c57bb 100644
--- a/datafusion/optimizer/src/push_down_filter.rs
+++ b/datafusion/optimizer/src/push_down_filter.rs
@@ -40,7 +40,7 @@ use datafusion_expr::utils::{
conjunction, expr_to_columns, split_conjunction, split_conjunction_owned,
};
use datafusion_expr::{
- BinaryExpr, Distinct, Expr, Filter, Operator, Projection,
+ BinaryExpr, Distinct, Expr, ExprSchemable, Filter, Operator, Projection,
TableProviderFilterPushDown, and, or,
};
@@ -1141,6 +1141,21 @@ impl OptimizerRule for PushDownFilter {
result.map_data(|plan| Ok(with_filters(keep_predicates, plan)))
}
LogicalPlan::Join(join) => push_down_join(join, Some(filter.predicate)),
+ // Pushes deterministic left-only predicates below the ASOF join and
+ // mirrors eligible equality-key predicates to the right input.
+ // Example:
+ // Before:
+ // Filter: l.key = 42
+ // AsOfJoin: on=[l.key = r.key]
+ // Left
+ // Right
+ // ---
+ // After:
+ // AsOfJoin: on=[l.key = r.key]
+ // Filter: l.key = 42
+ // Left
+ // Filter: r.key = 42
+ // Right
LogicalPlan::AsOfJoin(mut join) => {
// ASOF emits exactly one output row per left row without
// changing left values, so deterministic left-only predicates
@@ -1156,6 +1171,48 @@ impl OptimizerRule for PushDownFilter {
})
});
+ // A literal comparison on an equal, same-typed key has the
+ // same value for every matching pair. Mirroring it to the
+ // right can prune groups without changing the ASOF candidate.
+ let mut right_predicates = Vec::new();
+ for predicate in &push_predicates {
+ let Expr::BinaryExpr(BinaryExpr {
+ left,
+ op: Operator::Eq,
+ right,
+ }) = predicate
+ else {
+ continue;
+ };
+ let ((Expr::Column(left_column), Expr::Literal(_, _))
+ | (Expr::Literal(_, _), Expr::Column(left_column))) =
+ (left.as_ref(), right.as_ref())
+ else {
+ continue;
+ };
+ for (left_key, right_key) in &join.on {
+ let (Some(left_key_column), Some(right_key_column)) =
+ (left_key.try_as_col(), right_key.try_as_col())
+ else {
+ continue;
+ };
+ if left_column == left_key_column
+ && left_key.get_type(join.left.schema())?
+ == right_key.get_type(join.right.schema())?
+ {
+ let replacements =
+ HashMap::from([(left_key_column, right_key_column)]);
+ right_predicates
+ .push(replace_col(predicate.clone(), &replacements)?);
+ break;
+ }
+ }
+ }
+ if let Some(predicate) = conjunction(right_predicates) {
+ join.right =
+ Arc::new(LogicalPlan::Filter(Filter::new(predicate, join.right)));
+ }
+
let result = if let Some(predicate) = conjunction(push_predicates) {
filter.predicate = predicate;
filter.input = join.left;
diff --git a/datafusion/sqllogictest/test_files/asof_join.slt b/datafusion/sqllogictest/test_files/asof_join.slt
index 7abb500d23..e745a95b0e 100644
--- a/datafusion/sqllogictest/test_files/asof_join.slt
+++ b/datafusion/sqllogictest/test_files/asof_join.slt
@@ -342,6 +342,77 @@ physical_plan
07)----SortExec: expr=[grp@0 ASC, ts@1 ASC], preserve_partitioning=[false]
08)------DataSourceExec: partitions=1, partition_sizes=[1]
+# An equality-key predicate on the left can also remove right-side groups
+# without changing matches or the unmatched left rows.
+query IT
+SELECT l.id, r.val
+FROM asof_left l
+ASOF JOIN asof_right r
+MATCH_CONDITION (l.ts >= r.ts)
+ON l.grp = r.grp
+WHERE l.grp = 'A'
+ORDER BY l.id;
+----
+1 NULL
+2 a4
+3 a6
+7 NULL
+
+query IT
+SELECT l.id, r.val
+FROM asof_left l
+ASOF JOIN asof_right r
+MATCH_CONDITION (l.ts >= r.ts)
+ON l.grp = r.grp
+WHERE l.grp = 'A' AND r.val IS NULL
+ORDER BY l.id;
+----
+1 NULL
+7 NULL
+
+# The disjunction also retains group B, so it cannot prune the right to A.
+query IT
+SELECT l.id, r.val
+FROM asof_left l
+ASOF JOIN asof_right r
+MATCH_CONDITION (l.ts >= r.ts)
+ON l.grp = r.grp
+WHERE l.grp = 'A' OR l.id = 4
+ORDER BY l.id;
+----
+1 NULL
+2 a4
+3 a6
+4 b1
+7 NULL
+
+query TT
+EXPLAIN SELECT l.id, r.val
+FROM asof_left l
+ASOF JOIN asof_right r
+MATCH_CONDITION (l.ts >= r.ts)
+ON l.grp = r.grp
+WHERE l.grp = 'A';
+----
+logical_plan
+01)Projection: l.id, r.val
+02)--AsOf Join: match=[l.ts >= r.ts], constraint=On, on=[l.grp = r.grp]
+03)----SubqueryAlias: l
+04)------Filter: asof_left.grp = Utf8View("A")
+05)--------TableScan: asof_left projection=[id, grp, ts]
+06)----SubqueryAlias: r
+07)------Filter: asof_right.grp = Utf8View("A")
+08)--------TableScan: asof_right projection=[grp, ts, val]
+physical_plan
+01)AsOfJoinExec: on=[(grp = grp)], match=[ts >= ts], projection=[id@0, val@5]
+02)--SortExec: expr=[ts@2 ASC], preserve_partitioning=[true]
+03)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+04)------FilterExec: grp@1 = A
+05)--------DataSourceExec: partitions=1, partition_sizes=[1]
+06)--SortExec: expr=[ts@1 ASC], preserve_partitioning=[false]
+07)----FilterExec: grp@0 = A
+08)------DataSourceExec: partitions=1, partition_sizes=[1]
+
# An ASOF join emits each left row exactly once, so the primary key from
# asof_left remains unique. Because the outer LEFT JOIN does not use columns
# from the ASOF side, it can be removed entirely.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org