adriangb · Sep 25, 2026 · Sep 24, 2026 · Sep 24, 2026 · Sep 24, 2026 · Sep 24, 2026
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -406,3 +406,25 @@ debug = false
 debug-assertions = false
 strip = "debuginfo"
 incremental = false
+
+# Pin arrow-rs to pydantic/arrow-rs claude/push-decoder-batch-granular-scan-plan-60.0.0:
+# 60.0.0 plus apache/arrow-rs#6946 (batch-granular decoding) and
+# apache/arrow-rs#10555 (scan_plan). Patch every arrow crate together.
+[patch.crates-io]
+arrow-arith = { git = "https://github.com/pydantic/arrow-rs.git", branch = "claude/push-decoder-batch-granular-scan-plan-60.0.0" }
+arrow-array = { git = "https://github.com/pydantic/arrow-rs.git", branch = "claude/push-decoder-batch-granular-scan-plan-60.0.0" }
+arrow-avro = { git = "https://github.com/pydantic/arrow-rs.git", branch = "claude/push-decoder-batch-granular-scan-plan-60.0.0" }
+arrow-buffer = { git = "https://github.com/pydantic/arrow-rs.git", branch = "claude/push-decoder-batch-granular-scan-plan-60.0.0" }
+arrow-cast = { git = "https://github.com/pydantic/arrow-rs.git", branch = "claude/push-decoder-batch-granular-scan-plan-60.0.0" }
+arrow-csv = { git = "https://github.com/pydantic/arrow-rs.git", branch = "claude/push-decoder-batch-granular-scan-plan-60.0.0" }
+arrow-data = { git = "https://github.com/pydantic/arrow-rs.git", branch = "claude/push-decoder-batch-granular-scan-plan-60.0.0" }
+arrow-flight = { git = "https://github.com/pydantic/arrow-rs.git", branch = "claude/push-decoder-batch-granular-scan-plan-60.0.0" }
+arrow-ipc = { git = "https://github.com/pydantic/arrow-rs.git", branch = "claude/push-decoder-batch-granular-scan-plan-60.0.0" }
+arrow-json = { git = "https://github.com/pydantic/arrow-rs.git", branch = "claude/push-decoder-batch-granular-scan-plan-60.0.0" }
+arrow-ord = { git = "https://github.com/pydantic/arrow-rs.git", branch = "claude/push-decoder-batch-granular-scan-plan-60.0.0" }
+arrow-row = { git = "https://github.com/pydantic/arrow-rs.git", branch = "claude/push-decoder-batch-granular-scan-plan-60.0.0" }
+arrow-schema = { git = "https://github.com/pydantic/arrow-rs.git", branch = "claude/push-decoder-batch-granular-scan-plan-60.0.0" }
+arrow-select = { git = "https://github.com/pydantic/arrow-rs.git", branch = "claude/push-decoder-batch-granular-scan-plan-60.0.0" }
+arrow-string = { git = "https://github.com/pydantic/arrow-rs.git", branch = "claude/push-decoder-batch-granular-scan-plan-60.0.0" }
+arrow = { git = "https://github.com/pydantic/arrow-rs.git", branch = "claude/push-decoder-batch-granular-scan-plan-60.0.0" }
+parquet = { git = "https://github.com/pydantic/arrow-rs.git", branch = "claude/push-decoder-batch-granular-scan-plan-60.0.0" }
diff --git a/benchmarks/README.md b/benchmarks/README.md
@@ -966,6 +966,28 @@ Several queries are included to test hash joins under various workloads.
 ./bench.sh run hj
 ```
 
+## Hash Join Dynamic Filter on an Ordered Subset
+
+This benchmark (`hj_ordered_subset`) measures the dynamic filter that a hash join pushes to its probe-side Parquet scan when the probe table is sorted by the join key and the build side matches an ordered subset of the key range. This is common for time-ordered fact tables: the events of one day, or of a range of days.
+
+The load SQL writes an `events` table of 100 days x 200,000 rows (20M rows, 100,000-row row groups) sorted by `event_id`, and two build tables with one row for each event. The queries join `events` to 1 day (Q01, Q04), to 10 days (Q02, Q05), or to 1% of the events scattered over the whole range (Q03, Q06, the control). The bounds of the dynamic filter (`event_id >= min AND event_id <= max`) can prune the `events` row groups outside the matched range, and the membership check passes almost every row that remains. In the control, the bounds cannot prune anything.
+
+The `partitioned` subgroup (Q01-Q03) forces `HashJoinExec` mode `Partitioned`, and the `collect_left` subgroup (Q04-Q06) forces mode `CollectLeft`.
+
+### Example Run
+
+```bash
+# No need to generate data: the suite's load SQL writes the Parquet files
+
+./bench.sh run hj_ordered_subset
+
+# With parquet filter pushdown
+DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS=true ./bench.sh run hj_ordered_subset
+
+# Smaller data: 50 days x 100,000 rows
+HJOS_DAYS=50 HJOS_ROWS_PER_DAY=100000 ./bench.sh run hj_ordered_subset
+```
+
 ## Null-Aware Join
 
 This benchmark focuses on `NOT IN` subqueries, which plan as null-aware joins: an

diff --git a/benchmarks/bench.sh b/benchmarks/bench.sh
@@ -119,6 +119,10 @@ spill_views:            Sort and GROUP BY queries that spill StringView/BinaryVi
 null_aware_join:        Null-aware (NOT IN) hash join micro-benchmarks: uncorrelated, non-equality-correlated and equality-correlated
                           NOT IN across NULL fractions, to measure the per-pair join-filter work the correlated cases do
                           (data generated inline by the suite's load SQL from range(); knobs: NAJ_ROWS, NAJ_LARGE_ROWS)
+hj_ordered_subset:      Hash join dynamic filter on a probe table sorted by the join key, where the build side matches an ordered subset
+                          (1 day, 10 days) of the key range, so the filter bounds can prune probe row groups; plus a scattered 1% control
+                          (subgroups via BENCH_SUBGROUP: partitioned, collect_left)
+                          (data generated inline by the suite's load SQL; knobs: HJOS_DAYS, HJOS_ROWS_PER_DAY, HJOS_RG_SIZE)
 projection_subquery:    IN / NOT IN / EXISTS subqueries in the SELECT list (see https://github.com/apache/datafusion/issues/25341); each query projects the
                           boolean subquery result and aggregates it, so the cost is the decorrelation plan and not the output size
                           (q07 correlates on '<' instead of '=', so it keeps the nested-loop plan and acts as the control)
@@ -295,6 +299,10 @@ main() {
                     # Data is generated inline by the suite's load SQL.
                     echo "projection_subquery: no external data to generate"
                     ;;
+                hj_ordered_subset)
+                    # Data is generated inline by the suite's load SQL (COPY).
+                    echo "hj_ordered_subset: no external data to generate"
+                    ;;
                 asof_join)
                     data_asof_join
                     ;;
@@ -552,6 +560,9 @@ main() {
                 projection_subquery)
                     run_projection_subquery
                     ;;
+                hj_ordered_subset)
+                    run_hj_ordered_subset
+                    ;;
                 asof_join)
                     run_asof_join
                     ;;
@@ -1009,6 +1020,29 @@ run_null_aware_join() {
       bash -c "$SQL_CARGO_COMMAND"
 }
 
+# Runs the hj_ordered_subset suite: a hash join whose probe table is sorted by
+# the join key and whose build side matches an ordered subset of the key range,
+# so the bounds of the join's dynamic filter can prune probe row groups. Data is
+# generated inline by the load SQL (COPY), so there is no data step. Set
+# DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS to vary how the scan uses the
+# filter.
+# Knobs (string-substituted into the load SQL, not engine config):
+#   BENCH_SUBGROUP     run one subgroup (partitioned, collect_left)
+#   HJOS_DAYS          days in the events table      (default 100, at least 50)
+#   HJOS_ROWS_PER_DAY  events per day                (default 200_000)
+#   HJOS_RG_SIZE       parquet row-group size        (default 100_000)
+run_hj_ordered_subset() {
+    echo "Running hj_ordered_subset benchmark (subgroup=${BENCH_SUBGROUP:-all}, days=${HJOS_DAYS:-100}, rows_per_day=${HJOS_ROWS_PER_DAY:-200000}, rg_size=${HJOS_RG_SIZE:-100000})..."
+    debug_run env BENCH_NAME=hj_ordered_subset \
+      BENCH_RESULTS_FILE="$(sql_results_file hj_ordered_subset)" \
+      ${BENCH_SUBGROUP:+BENCH_SUBGROUP="${BENCH_SUBGROUP}"} \
+      HJOS_DAYS="${HJOS_DAYS:-100}" \
+      HJOS_ROWS_PER_DAY="${HJOS_ROWS_PER_DAY:-200000}" \
+      HJOS_RG_SIZE="${HJOS_RG_SIZE:-100000}" \
+      ${QUERY:+BENCH_QUERY="${QUERY}"}  \
+      bash -c "$SQL_CARGO_COMMAND"
+}
+
 # Runs the projection_subquery suite: IN / NOT IN / EXISTS subqueries that sit
 # in the SELECT list instead of a filter (see
 # https://github.com/apache/datafusion/issues/25341). The load SQL builds the

diff --git a/benchmarks/sql_benchmarks/README.md b/benchmarks/sql_benchmarks/README.md
@@ -35,6 +35,7 @@ in the community:
 | `clickbench_sorted`   | ClickBench benchmark using a pre-sorted hits file.                 |
 | `h2o`                 | The `h2o` benchmark                                                |
 | `hj`                  | Hash join benchmark                                                |
+| `hj_ordered_subset`   | Hash join dynamic filter on a probe table sorted by the join key, where the build side matches an ordered subset of the key range (1 day, 10 days) or a scattered 1% (control). Subgroups (`--subgroup`): `partitioned`, `collect_left`. Size the data with `HJOS_DAYS`, `HJOS_ROWS_PER_DAY` and `HJOS_RG_SIZE`. |
 | `imdb`                | IMDb benchmark                                                     |
 | `nlj`                 | Nested‑loop join benchmark                                         |
 | `null_aware_join`     | Null-aware (`NOT IN`) hash join micro-benchmarks. Q01-Q03 are uncorrelated `NOT IN` across NULL fractions and are linear in the table size (`NAJ_LARGE_ROWS`, default `1000000`). Q04-Q08 are correlated, so the correlation predicate stays behind as a join filter that the join applies per candidate (build row × probe row) pair while deciding which outer rows are UNKNOWN; without an equality correlation there are no scope keys to narrow those pairs, so their cost grows with the square of `NAJ_ROWS` (default `10000`). Q08 adds an equality correlation, which turns those pairs into a hash lookup. All tables are built inline from `range()`, so there is no data step. |

diff --git a/benchmarks/sql_benchmarks/hj_ordered_subset/benchmarks/q01.benchmark b/benchmarks/sql_benchmarks/hj_ordered_subset/benchmarks/q01.benchmark
@@ -0,0 +1,10 @@
+# Partitioned join, build side = 1 day (1% of the event_id range, ordered).
+# The bounds of the dynamic filter can prune ~99% of the events row groups.
+subgroup partitioned
+
+template sql_benchmarks/hj_ordered_subset/hj_ordered_subset.benchmark.template
+NAME=Q01_partitioned_ordered_1_day
+JOIN_MODE=partitioned
+PLAN_MODE=Partitioned
+BUILD_TABLE=labels_by_day
+BUILD_FILTER=l.day = 42
diff --git a/benchmarks/sql_benchmarks/hj_ordered_subset/benchmarks/q02.benchmark b/benchmarks/sql_benchmarks/hj_ordered_subset/benchmarks/q02.benchmark
@@ -0,0 +1,10 @@
+# Partitioned join, build side = 10 days (10% of the event_id range, ordered).
+# The bounds of the dynamic filter can prune ~90% of the events row groups.
+subgroup partitioned
+
+template sql_benchmarks/hj_ordered_subset/hj_ordered_subset.benchmark.template
+NAME=Q02_partitioned_ordered_10_days
+JOIN_MODE=partitioned
+PLAN_MODE=Partitioned
+BUILD_TABLE=labels_by_day
+BUILD_FILTER=l.day BETWEEN 40 AND 49
diff --git a/benchmarks/sql_benchmarks/hj_ordered_subset/benchmarks/q03.benchmark b/benchmarks/sql_benchmarks/hj_ordered_subset/benchmarks/q03.benchmark
@@ -0,0 +1,11 @@
+# Control: partitioned join, build side = 1% of the events, scattered over the
+# whole event_id range. The bounds cannot prune anything; only the membership
+# check of the dynamic filter rejects rows.
+subgroup partitioned
+
+template sql_benchmarks/hj_ordered_subset/hj_ordered_subset.benchmark.template
+NAME=Q03_partitioned_scattered_1_pct
+JOIN_MODE=partitioned
+PLAN_MODE=Partitioned
+BUILD_TABLE=labels_by_bucket
+BUILD_FILTER=l.bucket = 42
diff --git a/benchmarks/sql_benchmarks/hj_ordered_subset/benchmarks/q04.benchmark b/benchmarks/sql_benchmarks/hj_ordered_subset/benchmarks/q04.benchmark
@@ -0,0 +1,9 @@
+# Same as Q01 with a CollectLeft join.
+subgroup collect_left
+
+template sql_benchmarks/hj_ordered_subset/hj_ordered_subset.benchmark.template
+NAME=Q04_collect_left_ordered_1_day
+JOIN_MODE=collect_left
+PLAN_MODE=CollectLeft
+BUILD_TABLE=labels_by_day
+BUILD_FILTER=l.day = 42
diff --git a/benchmarks/sql_benchmarks/hj_ordered_subset/benchmarks/q05.benchmark b/benchmarks/sql_benchmarks/hj_ordered_subset/benchmarks/q05.benchmark
@@ -0,0 +1,9 @@
+# Same as Q02 with a CollectLeft join.
+subgroup collect_left
+
+template sql_benchmarks/hj_ordered_subset/hj_ordered_subset.benchmark.template
+NAME=Q05_collect_left_ordered_10_days
+JOIN_MODE=collect_left
+PLAN_MODE=CollectLeft
+BUILD_TABLE=labels_by_day
+BUILD_FILTER=l.day BETWEEN 40 AND 49