dongjoon-hyun commented on code in PR #58881:
URL: https://github.com/apache/spark/pull/58881#discussion_r4345271716
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1322,24 +1324,25 @@ 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.
+ // stands; so is a `VariableValue` naming a field `addMutableState`
registered, where the
+ // policy says so, passing one being refused by the split methods' own
check in tests. A slot
+ // of a compacted array is not a name at all, so it has to be handed out
as a `GlobalValue`;
+ // where the policy does not ask for names, testing checks for one handed
out as a variable.
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 taken = isName || isField || !policy.onlyNames
+ val isField = policy.fieldsNeedNoArgument &&
mutableStateNames.contains(name)
+ val taken = isName || !policy.onlyNames
if (taken && !isField) {
+ if (Utils.isTesting) {
+ assert(isName, s"$name is handed out as a local variable, but no
parameter can " +
Review Comment:
**The removed rule also covered `INPUT_ROW`.** L1352-1353 pass
`JavaCode.variable(INPUT_ROW, ...)` through `take` under `rowWherever`, so the
compacted-slot rule also treated an `INPUT_ROW` that is a slot as a field. Now,
if an operator sets `ctx.INPUT_ROW` to a non-inlined
`addMutableState("InternalRow", ...)`, an `operatorMethod` split declares
`InternalRow x_mutableStateArray_0[0]` again. This message's advice, "hand out
a field with JavaCode.global or isNullGlobal", does not apply there, since
`INPUT_ROW` is a `String`.
No operator does this today: `SortAggregateExec` uses `forceInline = true`
for exactly this reason (SortAggregateExec.scala:240-243). Could the message
also mention `INPUT_ROW`, e.g. "or set INPUT_ROW to an inlined field"?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1322,24 +1324,25 @@ 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.
+ // stands; so is a `VariableValue` naming a field `addMutableState`
registered, where the
+ // policy says so, passing one being refused by the split methods' own
check in tests. A slot
Review Comment:
**Nit:** "passing one being refused by the split methods' own check in
tests" only holds for `splitExpressions` and `splitCall` (L1718-1724 and
L1807-1813, `!mutableStateNames.contains(...)`). Under `operatorMethod`
(`fieldsNeedNoArgument = false`), the `subExpr`, `doAggregate_*` and
`switchCaseCode` methods build their own parameter lists from
`getLocalInputVariableValues`, and nothing checks them. So an inlined field
handed out as a `VariableValue` would be passed as a parameter there without
any check. Could the comment name the check it means, e.g. "by the whole stage
split's own check (`splitCall`)"?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1322,24 +1324,25 @@ 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.
+ // stands; so is a `VariableValue` naming a field `addMutableState`
registered, where the
+ // policy says so, passing one being refused by the split methods' own
check in tests. A slot
+ // of a compacted array is not a name at all, so it has to be handed out
as a `GlobalValue`;
+ // where the policy does not ask for names, testing checks for one handed
out as a variable.
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 taken = isName || isField || !policy.onlyNames
+ val isField = policy.fieldsNeedNoArgument &&
mutableStateNames.contains(name)
+ val taken = isName || !policy.onlyNames
if (taken && !isField) {
Review Comment:
**Nit:** When `policy.onlyNames` is true, `taken == isName`, so this
assertion can only fail under `operatorMethod`. Writing the condition as `if
(Utils.isTesting && !policy.onlyNames)`, or saying so in the comment above,
would make the scope of the check visible (see the comment on L1340).
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala:
##########
@@ -1771,6 +1801,41 @@ class WholeStageCodegenSuite extends SharedSparkSession
}
}
+ test("SPARK-59589: split switch cases and aggregate functions read an
Expand's output as a " +
Review Comment:
**Nit:** This test and the one at L1747 build the same UNPIVOT view and run
the same split on/off loop (`genCode`, `methodsHolding(c, "].lpad(")`, compile,
`checkAnswer`). Only the first one also checks for a `UTF8String` parameter. A
small private helper next to `methodsHolding`, taking the method-name fragment
and the query, would keep the two checks in step.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1322,24 +1324,25 @@ 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.
+ // stands; so is a `VariableValue` naming a field `addMutableState`
registered, where the
+ // policy says so, passing one being refused by the split methods' own
check in tests. A slot
+ // of a compacted array is not a name at all, so it has to be handed out
as a `GlobalValue`;
+ // where the policy does not ask for names, testing checks for one handed
out as a variable.
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 taken = isName || isField || !policy.onlyNames
+ val isField = policy.fieldsNeedNoArgument &&
mutableStateNames.contains(name)
+ val taken = isName || !policy.onlyNames
Review Comment:
**This guard only covers `operatorMethod`.** Under `wholeStageSplit` and
`commonExpressionMethod` (`onlyNames = true`), `taken == isName`, so a slot
handed out as a `VariableValue` is still refused silently: the CASE WHEN block
stays inline, which brings back the 64KB failure SPARK-33301 fixed for that
stage, or the `With` definition stays inline. No test notices either. In
production, the `operatorMethod` path still fails to compile and falls back.
A check at the producer would cover all three policies, e.g. a testing-only
check in `JavaCode.variable` / `isNullVariable` that the name is a Java
identifier, or an `addMutableState` variant that hands back a `GlobalValue`.
The hand-built values in `WithExpressionEvalSuite` and `CollectInputsSuite`
would need to change for the former. This could be a separate JIRA; it does not
need to block this PR.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1322,24 +1324,25 @@ 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.
+ // stands; so is a `VariableValue` naming a field `addMutableState`
registered, where the
+ // policy says so, passing one being refused by the split methods' own
check in tests. A slot
+ // of a compacted array is not a name at all, so it has to be handed out
as a `GlobalValue`;
+ // where the policy does not ask for names, testing checks for one handed
out as a variable.
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 taken = isName || isField || !policy.onlyNames
+ val isField = policy.fieldsNeedNoArgument &&
mutableStateNames.contains(name)
+ val taken = isName || !policy.onlyNames
if (taken && !isField) {
+ if (Utils.isTesting) {
Review Comment:
**The assertion also fires where no method is generated.** `take` runs while
the inputs are collected, not when a method is emitted.
`subexpressionEliminationForWholeStageCodegen` calls
`getLocalInputVariableValues` for every common subexpression (L2175-2178)
before it decides `needSplit`, so a stage whose subexpressions stay inline
still goes through this check. The same holds when `splitAggregateExpressions`
gives up on the parameter length, and when an `ExpandExec` switch case stays
inline for the same reason.
So if a producer hands out a `VariableValue` that is not a name under a
`Project` with a small common subexpression, production compiles and runs the
stage, while a test fails with this `AssertionError`. Nothing in the tree does
that today, and CI is green, but the check is stricter than the failure it
guards. Could it move to where a method is actually built, e.g. an assertion on
the parameter names at the three call sites right before `addNewFunction`?
--
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]
