Skip to content

[SPARK-59614][SQL] Compile the AS-OF join as-of/residual conditions to a Predicate - #58886

Open
david-mollitor-db wants to merge 1 commit into
apache:masterfrom
david-mollitor-db:asof-condition-predicate
Open

david-mollitor-db wants to merge 1 commit into
apache:masterfrom
david-mollitor-db:asof-condition-predicate

Conversation

@david-mollitor-db

Copy link
Copy Markdown
Contributor

What changes were proposed in this pull request?

SortMergeAsOfJoinScanner
(sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala)
evaluated its as-of and residual conditions as bound Expressions, interpreted per right-buffer
row (boundAsOfCond.eval(joinedRow) and boundResidualCond.forall(...)). This compiles them to
BasePredicate via Predicate.create(...) and evaluates them through the primitive-returning
eval(InternalRow): Boolean:

  • boundAsOfCond becomes a BasePredicate; boundResidualCond becomes Option[BasePredicate].
  • Each predicate is initialized per partition (initialize(partitionIndex), index from
    TaskContext.getPartitionId()), matching the CartesianProductExec idiom.
  • The call sites in findBestBackwardForward/findBestForwardNearest use
    if (boundAsOfCond.eval(joinedRow)) plus a small residualHolds helper that avoids the
    Option.forall closure and its boxed result.
  • The orderExpression (a distance value, not a boolean) intentionally stays a bound Expression.

Why are the changes needed?

SortMergeAsOfJoinScanner has no whole-stage codegen, so its inner scan is always interpreted.
Interpreted Expression.eval walks the expression tree and BoundReference.eval boxes every
operand it reads into non-cached java.lang.Double/java.lang.Long objects. JFR profiling of
AsOfJoinBenchmark (sort-merge cases isolated) showed this dominated the scan: ~24% of CPU
(boxToInteger 17.6% + Double.valueOf 4.1% + Long.valueOf 2.0%) and ~33% of allocation
(Double 20% + Long 13%). Compiling the conditions removes both the tree-walk and the boxing;
the profile's boxing/interpreted-eval frames are replaced by a single generated
SpecificPredicate.eval, and a no-equi-key AsOfJoinBenchmark run improved ~14% (929ms -> 800ms
best time).

Does this PR introduce any user-facing change?

No. Predicate.eval returns false for a null/false result, exactly matching the previous
x != null && x.asInstanceOf[Boolean] guard, and the binding schema is unchanged.

How was this patch tested?

Existing tests pass: SortMergeAsOfJoinSuite, DataFrameAsOfJoinSuite, AsOfJoinSQLSuite, and
AsOfJoinSortMergeSQLSuite (89 tests). Before/after JFR on AsOfJoinBenchmark confirms the
boxing and interpreted-eval frames are eliminated.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Isaac

This pull request and its description were written by Isaac.

…o a Predicate

### What changes were proposed in this pull request?

`SortMergeAsOfJoinScanner`
(`sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala`)
evaluated its as-of and residual conditions as bound `Expression`s, interpreted per right-buffer
row (`boundAsOfCond.eval(joinedRow)` and `boundResidualCond.forall(...)`). This compiles them to
`BasePredicate` via `Predicate.create(...)` and evaluates them through the primitive-returning
`eval(InternalRow): Boolean`:

- `boundAsOfCond` becomes a `BasePredicate`; `boundResidualCond` becomes `Option[BasePredicate]`.
- Each predicate is initialized per partition (`initialize(partitionIndex)`, index from
  `TaskContext.getPartitionId()`), matching the `CartesianProductExec` idiom.
- The call sites in `findBestBackwardForward`/`findBestForwardNearest` use
  `if (boundAsOfCond.eval(joinedRow))` plus a small `residualHolds` helper that avoids the
  `Option.forall` closure and its boxed result.
- The `orderExpression` (a distance value, not a boolean) intentionally stays a bound `Expression`.

### Why are the changes needed?

`SortMergeAsOfJoinScanner` has no whole-stage codegen, so its inner scan is always interpreted.
Interpreted `Expression.eval` walks the expression tree and `BoundReference.eval` boxes every
operand it reads into non-cached `java.lang.Double`/`java.lang.Long` objects. JFR profiling of
`AsOfJoinBenchmark` (sort-merge cases isolated) showed this dominated the scan: ~24% of CPU
(`boxToInteger` 17.6% + `Double.valueOf` 4.1% + `Long.valueOf` 2.0%) and ~33% of allocation
(`Double` 20% + `Long` 13%). Compiling the conditions removes both the tree-walk and the boxing;
the profile's boxing/interpreted-eval frames are replaced by a single generated
`SpecificPredicate.eval`, and a no-equi-key `AsOfJoinBenchmark` run improved ~14% (929ms -> 800ms
best time).

### Does this PR introduce _any_ user-facing change?

No. `Predicate.eval` returns `false` for a null/false result, exactly matching the previous
`x != null && x.asInstanceOf[Boolean]` guard, and the binding schema is unchanged.

### How was this patch tested?

Existing tests pass: `SortMergeAsOfJoinSuite`, `DataFrameAsOfJoinSuite`, `AsOfJoinSQLSuite`, and
`AsOfJoinSortMergeSQLSuite` (89 tests). Before/after JFR on `AsOfJoinBenchmark` confirms the
boxing and interpreted-eval frames are eliminated.

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Isaac

Co-authored-by: Isaac <no-reply@databricks.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant