dongjoon-hyun commented on PR #59069:
URL: https://github.com/apache/spark/pull/59069#issuecomment-5848600085
I reviewed the diff at cdd72c25cb2 by tracing the generated-code paths. I
haven't run the repro queries below yet, so please double-check them.
### Correctness
**1. A `With` fill call reads `INPUT_ROW`, but the split methods don't take
it** (`CodeGenerator.scala` L1316, L1327)
`wholeStageSplitArguments` follows a `CommonExpressionRef` into its
definition but never sets `needsRow`. When `CommonExprSlots.build` turns the
fill into a method (`worthAMethod`), the reference emits
`computeCommonExpr_N(<INPUT_ROW>, ...)`, because `methodArgs` always passes a
non-null `INPUT_ROW` (L344-345). That call now lands inside a split
`caseWhen_*` method that has no `INPUT_ROW` parameter.
A `With` in a non-first branch stays in the plan. So under a keyed
`HashAggregateExec`, where `INPUT_ROW` is the local `unsafeRowBuffer`, the
following query should fail with `Unknown variable or type
"agg_unsafeRowAggBuffer_0"`. It throws under testing and falls back to non-WSCG
otherwise.
```sql
SELECT k, sum(CASE WHEN v = 0 THEN 0
WHEN v = 1 THEN nullif(nullif(v * 3, 1), 2)
WHEN v = 2 THEN v * 2 ... WHEN v = 300 THEN v * 300
ELSE 0 END)
FROM t GROUP BY k
```
- In-loop BHJ/SMJ/BNLJ join conditions, where `INPUT_ROW` is the build-row
local, have the same problem.
- This compiled before the PR. The second commit ("Pass INPUT_ROW only where
read") seems to have introduced it.
- Setting `needsRow` in the `CommonExpressionRef` arm when `INPUT_ROW !=
null` would fix it.
**2. A `computeCommonExpr_N` from the discarded non-split pass can call
removed methods** (L1197-1214, L1721-1723)
Only functions added inside the `splitExpressions` call are recorded as
discardable. `CommonExprSlots.build` adds `computeCommonExpr_N` in the same
non-split pass, but outside that window, so it is not recorded. It survives,
while the `caseWhen_*` methods it calls are removed, and Janino fails with `A
method named ... is not declared`. For example, with a `ProjectExec` and
subexpression elimination:
```scala
val cw = largeCaseWhen("id", 300)
val cw2 = (1 to 300).map(k => s"WHEN id = $k THEN id + $k").mkString("CASE
", " ", " ELSE 0 END")
spark.range(10).selectExpr(s"($cw) + 1 AS a", s"($cw) + 2 AS b",
s"IF(id < 0, 0, nullif($cw2, 0)) AS c", s"IF(id < 0, 0, nullif($cw2, 0))
AS d").collect()
```
`addNewFunctionToClass` is the only place that writes `classFunctions`.
Recording there every function added during the non-split pass would fix this.
It would also replace the snapshot and diff of all function names that is taken
on every split.
### Compatibility
**3.** Adding the default parameter `sources` changes the JVM signature of
`CodegenContext.splitExpressionsWithCurrentInputs`.
- The signature has not changed since the method was added in 2.3.0
(SPARK-22701).
- External projects call it, e.g. apache/gluten (`GenerateArrowProjection`)
and aehrc/pathling. Their jars built against an earlier 4.x would fail with
`NoSuchMethodError` on 4.4.0.
- catalyst is internal, but keeping the existing 6-arg method and adding the
new variant without default values would avoid this. Scala doesn't allow
default arguments on more than one overload.
### Tests
**4.** None of the new tests exercises the `With`/`CommonExpressionRef`
arms, the `needsRow` path, the refusals, or the removal of discarded functions,
so 1 and 2 are not caught. `max_by`/`min_by` update/merge expressions are
3-branch `CASE WHEN`s that read the buffer through `INPUT_ROW`, so a keyed
`max_by` test would cover `needsRow`.
**5.** The join test only checks answers.
- In all three queries the condition is generated inside the match loop,
where the build row is in scope. The test can't detect `INPUT_ROW` misuse, and
it can't detect splitting being disabled for joins.
- The comment "The part of the condition on the streamed side alone is
generated before the loop" doesn't hold for this query.
`spark.sql.join.splitStreamedSideJoinCondition` defaults to false, and
`ExprUtils.canEvaluateUnconditionally` does not accept `CaseWhen`.
**6.** The ROLLUP test doesn't read a compacted slot, so the slot branch in
`canPass` is never taken. `v` passes through `Expand` unchanged. Also, with
`splitConsumeFuncByOperator` (on by default), the aggregate reads `doConsume`
parameters.
**7.** `stageCodes` duplicates the existing `genCode(ds)` but lacks its
non-empty assertion. So `assert(!splitsCaseWhen(df))` in the last test passes
vacuously if the plan stops being whole-stage codegen'd.
### Minor
- Under WSCG, `buildCodeBlocks` runs two regexes per code string before
`sources.nonEmpty` is checked. This hits the 22 callers that pass no `sources`,
and it runs twice for `CaseWhen`.
- Every split block and group method takes the union of all branch inputs. A
wide `CASE WHEN` is therefore refused entirely, e.g. one with 85 or more
distinct nullable `bigint` inputs, even though that is the case most likely to
exceed 64KB.
- The 1024-char threshold also splits small `CASE WHEN`s (2-4 branches) now.
Could you add a benchmark for small and medium sizes, with the conf on and off?
- `wholeStageSplitArguments` largely duplicates
`CommonExprSlots.methodArgs`, but their rules already differ on `INPUT_ROW`,
compacted slots, fields, `LambdaVariable` and subexpression states. Item 1
comes from that gap.
- The `Cast` arm of `readsSubExprState` copies `Cast.genCode`'s pass-through
condition, so the two can drift apart.
- The `LambdaVariable` refusal says the value is "a local declared by the
loop", but `prepareLambdaVariable` makes it an inlined field.
- The `CodegenFallback` arm is unreachable under WSCG.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
