Skip to content

perf: evaluate a hash join filter conjunct by conjunct, per input row where it can - #27

Open
msafonov wants to merge 3 commits into
branch-1.1from
joom/1.1-join-filter-lazy
Open

msafonov wants to merge 3 commits into
branch-1.1from
joom/1.1-join-filter-lazy

Conversation

@msafonov

Copy link
Copy Markdown
Collaborator

Which issue does this PR close?

Closes #. (Joom fork, gcs_cost_insights_v2 on Comet: 4.5x slower than vanilla, mostly in one broadcast hash join)

Rationale for this change

gcs_cost_insights_v2 joins object paths to table locations with a broadcast hash join on (bucket, parts[0]) and a join filter of 20 conjuncts isnull(loc.parts[i]) OR loc.parts[i] = obj.parts[i] (40 GetArrayItem). One key holds 11.7k of the 19k locations, so a night meets 25.7e9 candidate pairs (inventory) and 10.6e9 (accesses). For every pair apply_join_filter_to_indices took both whole parts lists into an intermediate batch and evaluated the whole AND tree: 3.6 us per pair against <= 0.13 us in Spark; stage 209 26.1 task-h vs 0.9 vanilla (joom-1.1-12, ds 2026-10-17). Upstream apache/datafusion main has the same code path.

What changes are included in this PR?

HashJoinStream evaluates its filter with the new hash_join/lazy_filter.rs LazyJoinFilter (vendored physical-plan):

  • the top-level AND is split and each conjunct runs only over the pairs the earlier ones kept (Spark's short-circuit AND);
  • the largest non-volatile subexpressions over one input alone (loc.parts[3], isnull(loc.parts[3]), obj.parts[3]) are lifted out; a pair copies only their results and the bare columns its conjunct reads, never the lists;
  • lifted build-side expressions: per pair until the stream evaluated as many pairs as there are build rows, then once over the build batch, cached for the stream; if that full evaluation fails on some row, they stay per pair, so only rows reached by a pair can raise errors;
  • without the cache: once per run of pairs sharing a row (a probe row with many matches), otherwise per pair over the side's columns taken once per lookup and filtered with the survivors.

Pair order is preserved, so outer/semi/anti/mark handling after the filter is unchanged. A filter that reads no column keeps the old path; the symmetric hash join is untouched.

How are these changes tested?

  • lazy_filter.rs tests: the kept pairs equal apply_join_filter_to_indices (old path) exactly, in order, for 9 filters (the 5-level parts filter, mixed conjuncts, OR at the top, CASE, nullable conjuncts, NULL literal conjunct, probe-only, build-only + probe-only, strict cast) over sorted and shuffled pairs, before and after the build cache; errors only for pairs the filter reaches (cast error of an unpaired build row is not raised, of a paired one it is); lifted expressions evaluated per row, not per pair; HashJoinExec equals NestedLoopJoinExec over l.k = r.k AND filter for all 10 join types x batch sizes 7/64/8192 x 9 filters.
  • New bench spark-expr/benches/hash_join_array_filter.rs (Comet ListExtract, the 20-conjunct filter, batch 8192), base -> this PR: skewed 12k locations x 200 objects of one key (2.4M pairs) 131 -> 44 ms (55 -> 18 ns/pair); unique keys (200k pairs passing every conjunct) 19.2 -> 17.2 ms.
  • New JVM test CometJoinSuite "Broadcast HashJoin with a join filter over array elements of both sides" (INNER, LEFT, LEFT SEMI, LEFT ANTI, RIGHT vs Spark).
  • Local (aarch64 container): vendored physical-plan 1768 passed / 0 failed; workspace green (core 661, spark-expr 973, shuffle 193, jni-bridge 28, common 70); fmt and clippy -D warnings clean (workspace; vendored crate except the pre-existing needless_pass_by_value in sorts/sort/wide_payload.rs).
  • Cluster validation on the datasci-ab playground follows in a comment.

🤖 Generated with Claude Code

… where it can

gcs_cost_insights_v2 joins object paths to table locations with a broadcast
hash join on (bucket, parts[0]) and a filter of 20 conjuncts
`isnull(loc.parts[i]) OR loc.parts[i] = obj.parts[i]`. One key holds
11.7k locations, so the join meets 25.7e9 candidate pairs a night. For every
pair apply_join_filter_to_indices took both whole `parts` lists into an
intermediate batch and evaluated all 40 GetArrayItem and the whole AND tree:
3.6 us per pair on Comet against <= 0.13 us in Spark, 26 task-h instead of
0.9 in one stage.

HashJoinStream now evaluates the filter with LazyJoinFilter:
- the top-level AND is split and each conjunct is evaluated only over the
  pairs the earlier ones kept, like Spark's short-circuiting AND;
- the largest non-volatile subexpressions that read one input alone
  (`loc.parts[3]`, `isnull(loc.parts[3])`, `obj.parts[3]`) are lifted out and
  a pair only copies their results and the bare columns its conjunct reads;
- lifted build-side expressions are evaluated per pair until the stream has
  evaluated as many pairs as there are build rows, then once over the build
  batch and cached for the rest of the stream; if that fails on some row,
  they stay per pair, so only rows a pair reaches can raise errors;
- without the cache, lifted expressions are evaluated once per run of pairs
  sharing a row (a probe row with many matches), otherwise per pair over the
  columns they read, taken once per lookup and filtered with the survivors.
Pairs keep their order, so outer/semi/anti/mark handling is unchanged.
Filters reading no column, and the symmetric hash join, keep the old path.

New bench spark-expr/benches/hash_join_array_filter.rs (Comet ListExtract,
the 20-conjunct location filter, batch 8192), base -> this commit:
  skewed, 12k locations x 200 objects of one key (2.4M pairs): 131 -> 44 ms
  unique keys (200k pairs, all pass every conjunct): 19.2 -> 17.2 ms

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
@github-actions github-actions Bot added enhancement New feature or request performance labels Oct 19, 2026
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
@msafonov

Copy link
Copy Markdown
Collaborator Author

Cluster validation (datasci-ab playground, em64 12x8, pyspark-comet-env 1.1-13 + this PR, gcs_cost_insights_v2 ds 2026-10-17, count + sum(xxhash64(*)) of attach_best_table_location):

case total task-h join stage task-h rows checksum
inventory, vanilla 1.35 1.08 124474305 -66800546382105602140532
inventory, Comet + PR 0.96 0.76 (native BHJ, Total time for joining 0.60 h = 0.084 us/pair over 25.7e9 pairs) 124474305 equal
inventory, Comet 1.1-12 (prod stage 209, same input) - 26.1 - -
accesses join from a vanilla pre-aggregate, vanilla 0.92 0.66 89525562 60241607469135094626419
accesses join, Comet + PR 0.72 0.49 (native BHJ, 0.35 h = 0.12 us/pair over 10.6e9 pairs) 89525562 equal

Apps spark-e0396c99bd464f719e5ea6e777988b1c, spark-13eb452dab24448c8994ac055104ffb7, spark-8f2ebaf0ced4486aaf70155db69f4208 (deleted). Local CometJoinSuite: 62 / 0.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

enhancement New feature or request performance

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant