LuciferYang commented on code in PR #58881:
URL: https://github.com/apache/spark/pull/58881#discussion_r4333860539
##########
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
Review Comment:
Applied in 6435821cb1c, thanks.
##########
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") {
+ // `ExpandExec` holds an UNPIVOT's string output in a slot of a compacted
mutable state array,
+ // such as `expand_mutableStateArray_0[1]`. A large common subexpression
reading it is computed
+ // in a method of its own, which takes the local variables it reads as
parameters. Taken for a
+ // local variable, the slot became a parameter `UTF8String
expand_mutableStateArray_0[1]` and
+ // the stage did not compile. Handed out as a field, it is read as one,
split on or off.
+ withTempView("t") {
+ spark.range(10).selectExpr("concat('a', CAST(id AS STRING)) AS a",
+ "concat('b', CAST(id AS STRING)) AS b").createOrReplaceTempView("t")
+ val f = "concat(upper(v), lower(v), reverse(v), trim(v), ltrim(v),
rtrim(v), initcap(v), " +
+ "repeat(v, 2), lpad(v, 10, 'x'), rpad(v, 10, 'y'))"
+ def df: DataFrame =
+ sql(s"SELECT c, $f AS x, concat($f, 'z') AS y FROM t UNPIVOT (v FOR c
IN (a, b))")
+ val expected = withoutWholeStage(df)
+ val stringParameter = "private void \\w*subExpr_\\d+\\([^)]*UTF8String
".r
+ Seq("true", "false").foreach { split =>
+ withSQLConf(SQLConf.WHOLESTAGE_SPLIT_EXPRESSIONS.key -> split) {
+ val code = genCode(df)
+ assert(code.exists { c =>
+ methodsHolding(c,
"expand_mutableStateArray").exists(_.contains("subExpr"))
Review Comment:
Thanks, changed in 6435821cb1c. The anchor now requires `].lpad(`, a method
call on a compacted slot, which in this query is only `lpad(v, 10, 'x')` on the
slot holding `v`; a method's own write into the array never matches it. To
check it, I had `ExpandExec` inline `v` as a named field so it no longer goes
through the slot: the old anchor still passed under both split values, and the
new one fails.
##########
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:
Thanks, this is fixed by your SPARK-60398 (#59650), which this branch now
has after merging master, so there is nothing left to do here.
##########
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:
Added in 6435821cb1c, with your message. With `ExpandExec` handing out
`JavaCode.variable` / `isNullVariable` again, it trips on the reproducer and on
both of your queries, naming the slot, e.g. `expand_mutableStateArray_0[1] is
handed out as a local variable ...`. `CollectInputsSuite` also checks it
directly: a slot handed out as a variable is refused under the two policies
that ask for names, and trips the assertion under `operatorMethod`. The `@param
onlyNames` doc now says a variable is passed and testing checks it is a name.
--
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]
