dongjoon-hyun commented on code in PR #58881:
URL: https://github.com/apache/spark/pull/58881#discussion_r4328851424
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1322,22 +1322,17 @@ class CodegenContext extends Logging {
val args = mutable.LinkedHashMap.empty[String, VariableValue]
val inputsToEvaluate = mutable.ArrayBuffer.empty[ExprCode]
// Whether the code can read `v` from the method: false for a value no
parameter can carry
- // where the policy asks for names. `ExpandExec` hands out a
`VariableValue` naming a slot of a
- // compacted mutable state array, and a `SimpleExprValue` is an expression
rather than a name
+ // where the policy asks for names. A `SimpleExprValue` is an expression
rather than a name
// -- `posexplode_outer` gives its position the nullness `index == -1`,
and a `Byte` or `Short`
// literal's value is `(byte)1`. A field or a literal needs no parameter
and is read as it
// stands; so is a field handed out as a `VariableValue` where the policy
says so, passing one
- // being refused by the split methods' own check, a slot of a compacted
array included.
+ // being refused by the split methods' own check.
def take(v: ExprValue): Boolean = v match {
case local: VariableValue =>
val name = local.variableName
val isName = name.nonEmpty &&
Character.isJavaIdentifierStart(name.head) &&
name.forall(Character.isJavaIdentifierPart)
- // A slot of a compacted array is a field under every policy, since no
parameter can be
- // named `array[i]`, where the whole stage split is on; with it off,
every policy answers
- // as it did before that split existed. Another field is passed where
the policy says so.
- val isField = (isCompactedSlot(name) && wholeStageSplit !=
WholeStageSplit.Off) ||
- (policy.fieldsNeedNoArgument && mutableStateNames.contains(name))
+ val isField = policy.fieldsNeedNoArgument &&
mutableStateNames.contains(name)
val taken = isName || isField || !policy.onlyNames
if (taken && !isField) {
Review Comment:
Minor: With the compacted-slot rule gone, nothing catches the next producer
that hands out a slot as a `VariableValue`. Under `InputPolicy.operatorMethod`
(`onlyNames = false`), `take` adds any `VariableValue` to the arguments
whatever its name, so a `subExpr`, aggregate or switch-case method would
declare `UTF8String x_mutableStateArray_0[1]` again. The testing-only "cannot
be a global variable" checks at L1714 and L1803 cannot see it, since
`mutableStateNames` holds the array names but never a slot. Tests would stay
green unless one happens to split such a method, while production would
silently fall back. The `@param onlyNames` doc at L125 also says such a value
is "passed over", which only holds for a `SimpleExprValue`. Could you assert
here, under `Utils.isTesting` as `splitExpressions` and `splitCall` do, that
what is passed is a name? For example:
```scala
if (taken && !isField) {
if (Utils.isTesting) {
assert(isName, s"$name is handed out as a local variable, but no
parameter can " +
"carry it; hand out a field with JavaCode.global or
isNullGlobal")
}
args.getOrElseUpdate(name, local)
}
```
Every `VariableValue` and `INPUT_ROW` that reaches `operatorMethod` in the
tree today is a name, and the `mutableStateArray_0[3]` case in
`WithExpressionEvalSuite` only reaches `commonExpressionMethod`, so nothing
should trip it.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala:
##########
@@ -2767,7 +2796,7 @@ class WholeStageCodegenSuite extends SharedSparkSession
test("SPARK-33301: a split method reads a slot of a compacted mutable state
array as a field") {
withSplitAlways {
// `ExpandExec` holds an output of a type that is not primitive in a
slot of a compacted
- // mutable state array, such as `expand_mutableStateArray_0[0]`: a
field, though not a name.
+ // mutable state array, such as `expand_mutableStateArray_0[0]`, and
hands it out as a field.
// An UNPIVOT of string columns gives such an output, which the
projection above reads
// directly when the operators' consume functions are not split into
methods of their own.
// Every block reads it, the ELSE too, and each reads it as the field it
is, taking no
Review Comment:
Nit: The check behind "taking no argument for it",
`splitCaseWhenParameters(df).flatten.forall(!_.contains("mutableStateArray"))`
at L2815, cannot fail on its own. A CASE WHEN split method takes only what
`collectInputs` collects under `InputPolicy.wholeStageSplit`: a `GlobalValue`
is never an argument, a slot handed out as a `VariableValue` is refused as no
name, and an array or inlined name is a field. So if `ExpandExec` went back to
`JavaCode.variable`, every block would stay inline and
`assert(splitsCaseWhen(df))` at L2811 would fail first. That was already so
before this PR, but this test is now the CASE WHEN guard for the relabel. Could
L2815 go, or check something a regression could break on its own, e.g.
`forall(!_.contains("UTF8String"))`, as the new test's `stringParameter` does?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1322,22 +1322,17 @@ class CodegenContext extends Logging {
val args = mutable.LinkedHashMap.empty[String, VariableValue]
val inputsToEvaluate = mutable.ArrayBuffer.empty[ExprCode]
// Whether the code can read `v` from the method: false for a value no
parameter can carry
- // where the policy asks for names. `ExpandExec` hands out a
`VariableValue` naming a slot of a
- // compacted mutable state array, and a `SimpleExprValue` is an expression
rather than a name
+ // where the policy asks for names. A `SimpleExprValue` is an expression
rather than a name
// -- `posexplode_outer` gives its position the nullness `index == -1`,
and a `Byte` or `Short`
// literal's value is `(byte)1`. A field or a literal needs no parameter
and is read as it
// stands; so is a field handed out as a `VariableValue` where the policy
says so, passing one
- // being refused by the split methods' own check, a slot of a compacted
array included.
+ // being refused by the split methods' own check.
Review Comment:
Minor: Without ", a slot of a compacted array included", this sentence still
reads as if any field handed out as a `VariableValue` were read as it stands
where the policy says so. But `isField` only matches names in
`mutableStateNames`, and `addMutableState` never adds a slot there, only the
array name (L600) and inlined names (L680). So a slot handed out as a
`VariableValue`, which the new `ExpandExec` comment rightly calls a field, is
refused under `wholeStageSplit` and `commonExpressionMethod`, and becomes a
parameter that does not compile under `operatorMethod`; "the split methods' own
check" cannot see it either. The `fieldsNeedNoArgument` doc at L126-127 and the
new test name at CollectInputsSuite.scala:142-143 read the same way, though
that test's variable half needs `forceInline = true`. Could the comment say
which fields it means? For example:
```suggestion
// literal's value is `(byte)1`. A field or a literal needs no parameter
and is read as it
// stands; so is an inlined field (a name `addMutableState` registered)
handed out as a
// `VariableValue` where the policy says so, passing one being refused
by the split methods'
// own check; a slot of a compacted array is no such name, so its
producer hands it out as a
// `GlobalValue`, as `ExpandExec` does.
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/ExpandExec.scala:
##########
@@ -187,9 +187,15 @@ case class ExpandExec(
"resultValue",
v => s"$v = ${CodeGenerator.defaultValue(firstExpr.dataType)};")
+ // `addMutableState` hands back a field: a name of its own, or a
slot in a compacted array
+ // (`mutableStateArray_0[3]`) for a type it cannot inline. Either is
reachable from every
+ // method of the generated class and is not a local variable, so a
consumer collecting
+ // local variables to pass into a split function must not take it
for one -- a slot
+ // expression is not a legal parameter name. Whole-stage
subexpression elimination wraps
+ // its own mutable state the same way.
ExprCode(
- JavaCode.isNullVariable(isNull),
- JavaCode.variable(value, firstExpr.dataType))
+ JavaCode.isNullGlobal(isNull),
+ JavaCode.global(value, firstExpr.dataType))
Review Comment:
**Pre-existing and outside this diff: the switch-case split at L243-250
calls its functions by their unqualified name, so it does not compile once one
of them spills into a nested class.** The states labelled here are written by
the `switchCaseCode` functions that L243 adds with
`ctx.addNewFunction(switchCaseFunc, ...)`, but L250 calls
`$switchCaseFunc(...)` and drops the name `addNewFunction` returns. Once the
current class has passed `GENERATED_CLASS_SIZE_THRESHOLD` (1000000 chars),
`addNewFunction` puts the function in a private inner class and returns
`nestedClassInstance_N.switchCaseCode_M` (CodeGenerator.scala:885, :897). The
unqualified call then compiles only from inside that inner class: not from
`processNext` in the outer class, nor from another inner class, such as the one
the child's `doConsume_N` wrapper lands in when it is added after the
switch-case functions have filled a class.
For example, a sliding window like `GROUP BY window(ts, '1 hour', '10
seconds')` makes an Expand of 360 projections. With
`spark.sql.subexpressionElimination.enabled=false` on master, or by default on
released versions, I estimate its switch-case functions at a few MB. Janino
would then report `A method named "expand_switchCaseCode_..." is not declared
in any enclosing class ...`, and the stage falls back to running without
whole-stage codegen, or the query fails with
`spark.sql.codegen.fallback=false`. I traced this by reading the code and
estimated the sizes; I did not run it.
The other splits on this path call the name `addNewFunction` returns:
subexpression elimination (CodeGenerator.scala:2232), the split aggregate
functions (AggregateCodegenSupport.scala:358-366) and
`constructDoConsumeFunction` (WholeStageCodegenExec.scala:235). This has been
there since SPARK-35329 in 3.2.0, so it is not caused by this PR. Could it be a
separate JIRA that calls the returned name here too?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -2149,7 +2135,7 @@ class CodegenContext extends Logging {
// WHEN or a `With` definition's method. A generation that splits the
stage's expressions
// (`WholeStageSplit.Split`) removes those functions then; any other keeps
them, as the class
// did before that split existed, so that a stage kept in one piece, and
the split turned off,
- // give back the code generated before it.
+ // have the functions they had before it.
Review Comment:
Nit: After this edit, the "so that" clause only restates the "as" clause
before it. Read for the whole stage, it also no longer holds with the conf off,
since this PR changes those stages' methods: no slot or null-flag parameters,
and a `With` definition that reads a slot now gets its method. Could it name
the generations it covers instead? For example:
```suggestion
// (`WholeStageSplit.Split`) removes those functions then; any other, a
stage kept in one
// piece or the split turned off, keeps them, as the class did before
that split existed.
```
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala:
##########
@@ -1741,6 +1741,35 @@ class WholeStageCodegenSuite extends SharedSparkSession
"sinh(v) should be evaluated only once per input row without function
splitting")
}
+ test("SPARK-59589: a subexpression method reads an Expand's varying output
as a field") {
Review Comment:
Minor: This covers the `subExpr` methods, but the relabel also fixes two
other places that took the Expand's slot as a parameter, and no test runs
either. One is a nested Expand's switch-case split, whose
`getLocalInputVariableValues` (ExpandExec.scala:214) collected the child
Expand's slot; the other is the split aggregate functions
(AggregateCodegenSupport.scala:337-366). Both are reached when the consumer is
not wrapped in a consume method of its own, e.g. with
`spark.sql.codegen.splitConsumeFuncByOperator=false` or a schema wide enough to
skip the wrapper. Before this PR, with
`spark.sql.codegen.wholeStage.splitExpressions` off, these failed to compile:
```sql
-- spark.sql.codegen.splitConsumeFuncByOperator=false,
spark.sql.codegen.methodSplitThreshold=1
SELECT count(DISTINCT upper(v)), count(DISTINCT lower(v)) FROM t UNPIVOT (v
FOR c IN (a, b));
SELECT c, sum(length(upper(v))), sum(length(lower(v))) FROM t UNPIVOT (v FOR
c IN (a, b)) GROUP BY c;
```
I traced these by reading the code and did not run them. They fail the same
way on 4.1-4.3, so they would also help the backports. Could you add them here
as sibling cases that compile the code and check the answers?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1322,22 +1322,17 @@ class CodegenContext extends Logging {
val args = mutable.LinkedHashMap.empty[String, VariableValue]
val inputsToEvaluate = mutable.ArrayBuffer.empty[ExprCode]
// Whether the code can read `v` from the method: false for a value no
parameter can carry
- // where the policy asks for names. `ExpandExec` hands out a
`VariableValue` naming a slot of a
- // compacted mutable state array, and a `SimpleExprValue` is an expression
rather than a name
+ // where the policy asks for names. A `SimpleExprValue` is an expression
rather than a name
// -- `posexplode_outer` gives its position the nullness `index == -1`,
and a `Byte` or `Short`
// literal's value is `(byte)1`. A field or a literal needs no parameter
and is read as it
// stands; so is a field handed out as a `VariableValue` where the policy
says so, passing one
- // being refused by the split methods' own check, a slot of a compacted
array included.
+ // being refused by the split methods' own check.
def take(v: ExprValue): Boolean = v match {
case local: VariableValue =>
val name = local.variableName
val isName = name.nonEmpty &&
Character.isJavaIdentifierStart(name.head) &&
name.forall(Character.isJavaIdentifierPart)
- // A slot of a compacted array is a field under every policy, since no
parameter can be
- // named `array[i]`, where the whole stage split is on; with it off,
every policy answers
- // as it did before that split existed. Another field is passed where
the policy says so.
- val isField = (isCompactedSlot(name) && wholeStageSplit !=
WholeStageSplit.Off) ||
- (policy.fieldsNeedNoArgument && mutableStateNames.contains(name))
+ val isField = policy.fieldsNeedNoArgument &&
mutableStateNames.contains(name)
val taken = isName || isField || !policy.onlyNames
Review Comment:
Nit: With the slot clause gone, `isField` implies `isName`, since every name
in `mutableStateNames` is a Java identifier (fresh array names and inlined
names, never a slot). So `|| isField` no longer decides anything here:
```suggestion
val taken = isName || !policy.onlyNames
```
Relatedly, after the relabel no producer in the tree hands an inlined field
to `take` as a `VariableValue`, so `fieldsNeedNoArgument` only decides for the
value built by hand at CollectInputsSuite.scala:155-157. Dropping it would
leave a future mislabel to the testing check in `splitCall` (L1803-1804), at
the producer, which is the direction of this PR; that could be a follow-up.
Could you simplify this line at least?
--
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]
