perf: push left filters through ASOF joins

benchmarks/sql_benchmarks/asof_join/asof_join.suite

@@ -4,6 +4,7 @@
 # - no equality key, many balanced groups, or heavily skewed groups
 # - 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
 
 description = "ASOF join SQL benchmarks"
 
@@ -19,3 +20,7 @@ description = "Run all ASOF join queries."
 [[examples]]
 command = "cargo run --release --bin benchmark_runner -- asof_join --query 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."

benchmarks/sql_benchmarks/asof_join/benchmarks/q08.benchmark

@@ -0,0 +1,19 @@
+name Q08
+group asof_join
+
+expect_plan AsOfJoinExec
+
+run
+-- Left: 10M rows before a 1%-selective predicate; right: 100K rows; no equality key.
+-- Both inputs are ordered by timestamp. The left predicate can run before matching.
+-- Matches the latest right row at or before each retained left timestamp (l.ts >= r.ts).
+WITH left_input AS (
+    SELECT value AS ts, value AS payload FROM range(10000000)
+),
+right_input AS (
+    SELECT value AS ts, value AS payload FROM range(100000)
+)
+SELECT l.ts, l.payload, r.payload AS right_payload
+FROM left_input l
+ASOF JOIN right_input r MATCH_CONDITION (l.ts >= r.ts)
+WHERE l.ts % 100 = 0;

datafusion/optimizer/src/push_down_filter.rs

@@ -1141,6 +1141,32 @@ impl OptimizerRule for PushDownFilter {
                 result.map_data(|plan| Ok(with_filters(keep_predicates, plan)))
             }
             LogicalPlan::Join(join) => push_down_join(join, Some(filter.predicate)),
+            LogicalPlan::AsOfJoin(mut join) => {
+                // ASOF emits exactly one output row per left row without
+                // changing left values, so deterministic left-only predicates
+                // can run before matching. Right or mixed predicates must stay
+                // above the join because unmatched right fields are NULL-padded.
+                let (push_predicates, keep_predicates): (Vec<_>, Vec<_>) =
+                    split_conjunction_owned(filter.predicate)
+                        .into_iter()
+                        .partition(|predicate| {
+                            !predicate.is_volatile()
+                                && predicate.column_refs().iter().all(|column| {
+                                    join.left.schema().is_column_from_schema(column)
+                                })
+                        });
+
+                let result = if let Some(predicate) = conjunction(push_predicates) {
+                    filter.predicate = predicate;
+                    filter.input = join.left;
+                    join.left = Arc::new(LogicalPlan::Filter(filter));
+                    Transformed::yes(LogicalPlan::AsOfJoin(join))
+                } else {
+                    Transformed::no(LogicalPlan::AsOfJoin(join))
+                };
+
+                result.map_data(|plan| Ok(with_filters(keep_predicates, plan)))
+            }
             LogicalPlan::TableScan(mut scan) => {
                 let filter_predicates = split_conjunction(&filter.predicate);
                 // Filters containing scalar subqueries cannot be pushed to

datafusion/sqllogictest/test_files/asof_join.slt

@@ -256,6 +256,92 @@ physical_plan
 04)--SortExec: expr=[grp@0 ASC, ts@1 ASC], preserve_partitioning=[false]
 05)----DataSourceExec: partitions=1, partition_sizes=[1]
 
+# Deterministic predicates that reference only the left input can run before
+# matching. Right-only and mixed predicates must remain above the ASOF join.
+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.id > 1 AND r.val <> 'a2' AND l.id <> length(r.val);
+----
+logical_plan
+01)Filter: r.val != Utf8View("a2") AND l.id != character_length(r.val)
+02)--Projection: l.id, r.val
+03)----AsOf Join: match=[l.ts >= r.ts], constraint=On, on=[l.grp = r.grp]
+04)------SubqueryAlias: l
+05)--------Filter: asof_left.id > Int32(1)
+06)----------TableScan: asof_left projection=[id, grp, ts]
+07)------SubqueryAlias: r
+08)--------TableScan: asof_right projection=[grp, ts, val]
+physical_plan
+01)FilterExec: val@1 != a2 AND id@0 != character_length(val@1)
+02)--AsOfJoinExec: on=[(grp = grp)], match=[ts >= ts], projection=[id@0, val@5]
+03)----SortExec: expr=[grp@1 ASC, ts@2 ASC], preserve_partitioning=[true]
+04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+05)--------FilterExec: id@0 > 1
+06)----------DataSourceExec: partitions=1, partition_sizes=[1]
+07)----SortExec: expr=[grp@0 ASC, ts@1 ASC], preserve_partitioning=[false]
+08)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Filtering the right input before matching would select a4 for id 3 instead
+# of filtering out its actual match, a6.
+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.id > 1 AND r.val <> 'a6' AND l.id <> length(r.val)
+ORDER BY l.id;
+----
+4 b1
+5 b6
+
+# Left-only disjunctions retain unmatched rows with NULL equality or match keys.
+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 IS NULL OR l.ts IS NULL OR l.id = 1
+ORDER BY l.id;
+----
+1 NULL
+6 NULL
+7 NULL
+
+# Volatile predicates retain their evaluation point even when they reference
+# only the left input, while independent deterministic conjuncts still move.
+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.id > 1 AND l.id + random() > 0;
+----
+logical_plan
+01)Filter: CAST(l.id AS Float64) + random() > Float64(0)
+02)--Projection: l.id, r.val
+03)----AsOf Join: match=[l.ts >= r.ts], constraint=On, on=[l.grp = r.grp]
+04)------SubqueryAlias: l
+05)--------Filter: asof_left.id > Int32(1)
+06)----------TableScan: asof_left projection=[id, grp, ts]
+07)------SubqueryAlias: r
+08)--------TableScan: asof_right projection=[grp, ts, val]
+physical_plan
+01)FilterExec: CAST(id@0 AS Float64) + random() > 0
+02)--AsOfJoinExec: on=[(grp = grp)], match=[ts >= ts], projection=[id@0, val@5]
+03)----SortExec: expr=[grp@1 ASC, ts@2 ASC], preserve_partitioning=[true]
+04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+05)--------FilterExec: id@0 > 1
+06)----------DataSourceExec: partitions=1, partition_sizes=[1]
+07)----SortExec: expr=[grp@0 ASC, ts@1 ASC], preserve_partitioning=[false]
+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.