You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "LuciferYang (via GitHub)" <gi...@apache.org> on 2026/09/17 10:56:09 UTC
[PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
LuciferYang opened a new pull request, #58881:
URL: https://github.com/apache/spark/pull/58881
### What changes were proposed in this pull request?
`ExpandExec.doConsume` allocates a mutable state for each output column that varies across the Expand's branches, and hands it to the consumer as `JavaCode.variable(value, dataType)`. This changes it to `JavaCode.global` / `JavaCode.isNullGlobal`.
`CodegenContext.addMutableState` returns a field of the generated class: a name of its own when it can inline the type, otherwise a slot in a compacted array, e.g. `mutableStateArray_0[3]`. Either form is reachable from every method of that class, and neither is a local variable. Whole-stage subexpression elimination already wraps its own `addMutableState` results with `JavaCode.global` / `JavaCode.isNullGlobal`, so this brings `ExpandExec` in line with it.
`GlobalValue` and `VariableValue` print the same text, so the generated code is unchanged wherever the classification was not asked about.
### Why are the changes needed?
`CodeGenerator.getLocalInputVariableValues` collects the `VariableValue`s among an operator's input so that a caller can pass them into a split function; `AggregateCodegenSupport.splitAggregateExpressions` and `CommonExprSlots.fill` both do that. Taking a field for a local variable adds a parameter that shadows the field, and for the compacted form the parameter name is a slot expression, which does not compile. SPARK-59295 added an "is this a Java identifier" check over the candidate parameters partly for this source; that check stays, since it also covers whatever else a consumer may hold, but it no longer has to catch this one.
No query is known to reach the bad parameter today, so this closes a latent trap rather than a failure anyone can hit. With a consume function per operator -- the default -- the value is passed as an argument to a fresh parameter, and with that turned off, neither a CUBE over string columns nor a regular aggregate over a varying string column under a distinct rewrite puts the slot in a parameter list. If reviewers would rather see a reachable failure before taking this, that is a fair objection to raise.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
A new case in `WholeStageCodegenSuite` asserts the invariant on the shape that comes closest to reaching the trap: the child of a regular aggregate under a distinct rewrite is a varying `string` column, the buffers stay fixed width so the plan is a `HashAggregateExec`, aggregate function splitting is forced with a threshold of 1, and the consume function per operator is turned off so that the aggregate sees the Expand's own expressions rather than fresh parameters. It checks that the compacted slot is in the generated code and that no `private void ...(...)` parameter list holds one. Being an invariant on a shape that does not currently break, it passes before this change as well.
`WholeStageCodegenSuite` (63 passed, 3 ignored) and `DataFrameAggregateSuite` (175 passed); `sql/scalastyle` and `sql/Test/scalastyle` clean.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #58881:
URL: https://github.com/apache/spark/pull/58881#issuecomment-6098653590
Thanks @MaxGekk, the repro settles it. Done in c841872:
- Merged master and replaced the old test with your UNPIVOT query, run with `spark.sql.codegen.wholeStage.splitExpressions` on and off. It fails without the `ExpandExec` change.
- Removed `CompactedSlot`, `isCompactedSlot` and the slot clause in `collectInputs`, along with the mentions of the rule in the conf doc and in `WholeStageCodegenExec`.
- Removed the SPARK-33301 test you pointed at. The other slot test still passes and covers a split method reading the Expand's output, so I kept it. The two `CollectInputsSuite` slot cases are replaced by one that pins how `collectInputs` treats a field handed out as a global and as a variable.
The description is updated and the PR is out of draft.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "LuciferYang (via GitHub) via reviews" <re...@spark.apache.org>.
LuciferYang commented on code in PR #58881:
URL: https://github.com/apache/spark/pull/58881#discussion_r4454462014
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggregateCodegenSupport.scala:
##########
@@ -351,6 +351,7 @@ trait AggregateCodegenSupport
if (inputVars.forall(_.isDefined)) {
val splitCodes = inputVars.flatten.zipWithIndex.map { case (args, i) =>
val doAggFunc = ctx.freshName(s"doAggregate_${aggNames(i)}")
+ ctx.assertParametersAreNames(args.map(_.variableName), doAggFunc)
Review Comment:
Done in d6cf7c05e87: `assertMethodParameters` runs both checks, and the three sites call it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1411,13 +1417,16 @@ class CodegenContext extends Logging {
}
}
- /**
- * Whether `name` is a slot of a compacted mutable state array, such as `mutableStateArray_0[3]`,
- * which `addMutableState` hands out for a state it does not inline: a field of the class.
- */
- private def isCompactedSlot(name: String): Boolean = name match {
- case CodeGenerator.CompactedSlot(array) => mutableStateNames.contains(array)
- case _ => false
+ /** Whether `name` can name a parameter: a slot of a compacted array, `array[i]`, cannot. */
+ private def isJavaName(name: String): Boolean =
+ name.nonEmpty && Character.isJavaIdentifierStart(name.head) &&
+ name.forall(Character.isJavaIdentifierPart)
Review Comment:
Applied in d6cf7c05e87.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/MergeRowsExec.scala:
##########
@@ -156,7 +156,7 @@ case class MergeRowsExec(
|}
|$rowIdBitmap.add(${currentRowId.value});
""".stripMargin
- ExprCode(code, FalseLiteral, JavaCode.variable(rowIdBitmap, bitmapClass))
+ ExprCode(code, FalseLiteral, JavaCode.global(rowIdBitmap, bitmapClass))
Review Comment:
Reverted in d6cf7c05e87, and the description no longer mentions it.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "LuciferYang (via GitHub) via reviews" <re...@spark.apache.org>.
LuciferYang commented on code in PR #58881:
URL: https://github.com/apache/spark/pull/58881#discussion_r4466842384
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1322,23 +1330,26 @@ 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 whole stage split's own check (`splitCall`)
+ // in tests. A slot of a compacted array is not a name at all, so it has to be handed out as a
+ // `GlobalValue`, and `INPUT_ROW` cannot be one. Testing checks for one handed out as a
+ // variable: here, where the policy asks for names, for what the walk reaches before its first
+ // refusal; elsewhere, in the method taking it (`assertMethodParameters`). The row a whole stage
+ // split appends is not checked (SPARK-60510).
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 isName = isJavaName(name)
+ if (Utils.isTesting && policy.onlyNames) {
+ assert(isName, s"$name cannot name a parameter of a method this code moves into; " +
+ nameFix(name))
+ }
+ val isField = policy.fieldsNeedNoArgument && mutableStateNames.contains(name)
+ val taken = isName || !policy.onlyNames
Review Comment:
Done in 97c803e1beb: `CodegenContext.checkParameterNames` defaults to `Utils.isTesting`, and `collectInputs` and `assertParametersAreNames` use it. With it off, `CollectInputsSuite` checks that the whole-stage split and common expression policies refuse the method for a slot and for an `INPUT_ROW` that is a slot, and `WithExpressionEvalSuite` checks the inline body again with `markersIn(...) == 4`. With `val taken = true`, those three tests fail.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/ExpandExec.scala:
##########
@@ -189,9 +189,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 state not inlined. 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.
Review Comment:
Done in 97c803e1beb: the `@return` of `addMutableState` now says to hand the result out with `JavaCode.global` or `isNullGlobal`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1411,13 +1422,15 @@ class CodegenContext extends Logging {
}
}
- /**
- * Whether `name` is a slot of a compacted mutable state array, such as `mutableStateArray_0[3]`,
- * which `addMutableState` hands out for a state it does not inline: a field of the class.
- */
- private def isCompactedSlot(name: String): Boolean = name match {
- case CodeGenerator.CompactedSlot(array) => mutableStateNames.contains(array)
- case _ => false
+ /** Whether `name` can name a parameter: a slot of a compacted array, `array[i]`, cannot. */
+ private def isJavaName(name: String): Boolean =
+ SourceVersion.isIdentifier(name) && !SourceVersion.isKeyword(name)
+
+ /** How the producer of `name`, a value that cannot name a parameter, can hand it out instead. */
+ private def nameFix(name: String): String = if (name == INPUT_ROW) {
+ "set INPUT_ROW to a local or a field added with forceInline = true"
+ } else {
+ "hand out a field with JavaCode.global or isNullGlobal"
}
Review Comment:
Applied in 97c803e1beb, with an assertion on the hint for `false` in `CollectInputsSuite`.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala:
##########
@@ -1932,6 +1962,55 @@ class WholeStageCodegenSuite extends SharedSparkSession
query.collect().toSeq
}
+ /**
+ * Checks the query `query` builds from a FROM clause that UNPIVOTs two string columns into
+ * `v`, with the whole stage split on and off: a method named like `method` calls `lpad` on the
+ * slot of the compacted mutable state array that `ExpandExec` holds `v` in, `check` holds of
+ * the code, the code compiles, and the rows match those computed without whole stage codegen.
+ * With the slot handed out as a local variable instead, building that method fails the testing
+ * check on its parameters.
+ */
+ private def checkExpandSlotRead(
+ method: String,
+ query: String => String,
+ check: Seq[CodeAndComment] => Unit = _ => ()): Unit = {
+ withTempView("t") {
+ spark.range(10).selectExpr("concat('a', CAST(id AS STRING)) AS a",
+ "concat('b', CAST(id AS STRING)) AS b").createOrReplaceTempView("t")
+ def df: DataFrame = sql(query("FROM t UNPIVOT (v FOR c IN (a, b))"))
+ val expected = withoutWholeStage(df)
+ Seq("true", "false").foreach { split =>
+ withSQLConf(SQLConf.WHOLESTAGE_SPLIT_EXPRESSIONS.key -> split) {
+ val code = genCode(df)
+ // A subexpression method or a switch case also writes into a slot of the same array,
+ // so the array's name alone does not show a read.
+ assert(code.exists(c => methodsHolding(c, "].lpad(").exists(_.contains(method))),
+ s"$method $split")
+ withClue(s"$method $split: ")(check(code))
+ code.foreach(CodeGenerator.compile)
+ checkAnswer(df, expected)
+ }
+ }
+ withSQLConf(SQLConf.WHOLESTAGE_SPLIT_EXPRESSIONS.key -> "false") {
+ val e = intercept[AssertionError](genCodeWithLocalExpandOutput(df))
+ assert(s"\\w*$method\\w* takes \\w+\\[\\d+\\]".r.findFirstIn(e.getMessage).isDefined,
+ e.getMessage)
+ }
+ }
+ }
+
+ /**
+ * `genCode`, with the bottom `ExpandExec` of the plan handing out its output as local variables,
+ * as it did before SPARK-59589.
+ */
+ private def genCodeWithLocalExpandOutput(df: DataFrame): Seq[CodeAndComment] =
Review Comment:
Done in 97c803e1beb: `genCode` takes the change of the plan, and the helper is gone.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "LuciferYang (via GitHub) via reviews" <re...@spark.apache.org>.
LuciferYang commented on PR #58881:
URL: https://github.com/apache/spark/pull/58881#issuecomment-6650162573
Thanks @dongjoon-hyun @cloud-fan . I will manually create pr branch-4.3/branch-4.2/branch-4.1
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58881:
URL: https://github.com/apache/spark/pull/58881#discussion_r4323282482
##########
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:
**Nit (P3):** Compaction isn't decided by type alone. `addMutableState` inlines a primitive only while the class holds fewer than `OUTER_CLASS_VARIABLES_THRESHOLD` (10000) inlined states. Past that, the boolean `resultIsNull` here also comes back as a slot of a compacted `boolean[]`, since it is allocated without `forceInline`. That is the case that makes `isNullGlobal` necessary for the null flag too, but "for a type it cannot inline" reads as if only the non-primitive value can be a slot. Maybe:
```suggestion
// (`mutableStateArray_0[3]`) for a state not inlined. Either is reachable from every
```
**Verification:**
- **Inspection:** The comment's description of when addMutableState returns a compacted-array slot matches its documented and implemented inline conditions (forced inlining, a primitive while under the outer-class threshold, a multi-dimensional array), so both the value and the null flag are covered by the explanation for labelling them as globals.
##########
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:
**Nit (P3):** This holds even if no `subExpr` method reads the Expand's slot. In the split path, whole-stage subexpression elimination stores each string result with `addMutableState("UTF8String", "subExprValue")`. That takes a slot of the same per-type array `ExpandExec` created first, so it is also `expand_mutableStateArray_0[k]`, and the `subExpr` body writes `expand_mutableStateArray_0[k] = ...;`. So every split string `subExpr` method contains the text, whatever it reads.
The negative regex and the compile do catch the bug in today's plan. But if a later change stopped routing `v` through the Expand's slot, everything here would still pass and the test would quietly stop covering the bug. The removed SPARK-33301 test guarded against that with its conf-off `nonEmpty` check, which this fix makes impossible. Could the anchor require a read of the slot instead? For example, a method call on it, like `expand_mutableStateArray_0[1].lpad(` from `lpad(v, 10, 'x')`, which the method's own `... = ...` write never matches.
**Verification:**
- **Regression:** Under both split conf values, the positive assertion passes for the reproducer and does not accept a subExpr method that only writes its own compacted result slot without reading an Expand varying-output slot, so the test cannot pass once it stops exercising the fixed shape.
- **Behavior:** Under both split conf values, the reproducer's generated code still has no subExpr method taking a UTF8String parameter, compiles, and returns the rows computed without whole-stage codegen.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "LuciferYang (via GitHub) via reviews" <re...@spark.apache.org>.
LuciferYang commented on code in PR #58881:
URL: https://github.com/apache/spark/pull/58881#discussion_r4454460112
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1926,6 +1935,19 @@ class CodegenContext extends Logging {
}
}
+ /**
+ * Under testing, asserts that each of `parameters`, the parameters of the method `method` an
+ * operator moves its code into, is a Java name. The parameters are the variables
+ * `InputPolicy.operatorMethod` collects, which passes a `VariableValue` whatever its name.
+ */
+ private[sql] def assertParametersAreNames(parameters: Seq[String], method: String): Unit = {
Review Comment:
Done in d6cf7c05e87: `checkExpandSlotRead` now also builds each query with a test operator, `LocalOutputExec`, that hands out the Expand's fields as local variables, as `ExpandExec` did before this PR, and expects the check on the parameters of the subExpr, switch case or doAggregate method to fail. Dropping any of the three calls fails its test. This covers the subExpr call as well, so I didn't add the catalyst test.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1322,23 +1326,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 whole stage split's own check (`splitCall`)
+ // in tests. A slot of a compacted array is not a name at all, so it has to be handed out as a
+ // `GlobalValue`, and `INPUT_ROW` cannot be one. Testing checks for one handed out as a
+ // variable: here, where the policy asks for names; elsewhere, in the method taking it
+ // (`assertParametersAreNames`).
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 isName = isJavaName(name)
+ if (Utils.isTesting && policy.onlyNames) {
Review Comment:
Updated the comment in d6cf7c05e87: the check covers what the walk reaches before its first refusal.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1322,23 +1326,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 whole stage split's own check (`splitCall`)
+ // in tests. A slot of a compacted array is not a name at all, so it has to be handed out as a
+ // `GlobalValue`, and `INPUT_ROW` cannot be one. Testing checks for one handed out as a
+ // variable: here, where the policy asks for names; elsewhere, in the method taking it
+ // (`assertParametersAreNames`).
Review Comment:
Applied in d6cf7c05e87, and `@param onlyNames` says the same.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/WithExpressionEvalSuite.scala:
##########
@@ -305,18 +305,18 @@ class WithExpressionEvalSuite extends SparkFunSuite with SQLHelper {
s"the innermost body was emitted $count times at depth $depth")
assert(!methods.contains("computeCommonExpr"), methods)
}
- // A value whose name is not one a parameter can take: `ExpandExec` hands out a compacted
- // mutable state array slot as if it were a local. The marker count says the fallback pasted
- // the body at both references; that a method was attempted at all is the case above, which
- // gets one from the same shape with a passable name.
+ // A value whose name is not one a parameter can take, such as a slot of a compacted mutable
+ // state array handed out as a local. Its producer has to hand it out as a field, and tests
+ // fail where a method would be built; the previous test gets a method from the same shape
+ // with a passable name.
val slot = ExprCode(EmptyBlock, FalseLiteral,
JavaCode.variable("mutableStateArray_0[3]", IntegerType))
- val (slotBody, slotMethods) = generateWithLocalInput(nested(2), slot)
- assert(!slotMethods.contains("computeCommonExpr"), slotMethods)
- assert(markersIn(slotBody + slotMethods) == 4, slotBody + slotMethods)
+ val e = intercept[AssertionError](generateWithLocalInput(nested(2), slot))
Review Comment:
Moved into a test of its own in d6cf7c05e87, with the title you suggested.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/codegen/CollectInputsSuite.scala:
##########
@@ -139,29 +139,58 @@ class CollectInputsSuite extends SparkFunSuite with SQLHelper {
assert(!ctx.collectInputs(Seq(input(0)), operator, Map.empty).get.readsRow)
}
- test("a slot of a compacted mutable state array is a field under every policy with the whole " +
- "stage split on, never a parameter, which no method could declare") {
+ test("a field handed out as a global is never an argument; an inlined one handed out as a " +
+ "variable is left out only where the policy says so") {
val ctx = context()
- ctx.wholeStageSplit = WholeStageSplit.Record
- // A state of a type that is not primitive is compacted into an array, and its name is a slot.
+ val expr = Length(BoundReference(0, StringType, nullable = false))
+ // A state of a type that is not primitive is a slot of a compacted array.
val slot = ctx.addMutableState("UTF8String", "value")
assert(slot.matches("\\w+\\[\\d+\\]"), slot)
- ctx.currentVars = Seq(ExprCode(EmptyBlock, FalseLiteral, JavaCode.variable(slot, StringType)))
- val expr = Length(BoundReference(0, StringType, nullable = false))
+ ctx.currentVars = Seq(ExprCode(EmptyBlock, FalseLiteral, JavaCode.global(slot, StringType)))
for (policy <- Seq(InputPolicy.wholeStageSplit, commonExpr, operator)) {
assert(names(ctx.collectInputs(Seq(expr), policy, Map.empty).get).isEmpty, policy)
}
+ // An inlined state handed out as a variable is read as the field only by a whole stage split
+ // method.
+ val field = ctx.addMutableState("UTF8String", "field", forceInline = true)
+ ctx.currentVars = Seq(ExprCode(EmptyBlock, FalseLiteral, JavaCode.variable(field, StringType)))
+ assert(names(ctx.collectInputs(Seq(expr), InputPolicy.wholeStageSplit, Map.empty).get).isEmpty)
+ for (policy <- Seq(commonExpr, operator)) {
+ assert(names(ctx.collectInputs(Seq(expr), policy, Map.empty).get) == Seq(field), policy)
+ }
}
- test("with the whole stage split off, a slot is answered as before that split existed") {
- // A common expression's method refuses it, so its definition stays inline, and an operator's
- // method lists it among its inputs.
+ test("a slot handed out as a variable fails the testing check: while collecting where the " +
+ "policy asks for names, and in the method that takes it where it does not") {
val ctx = context()
val slot = ctx.addMutableState("UTF8String", "value")
ctx.currentVars = Seq(ExprCode(EmptyBlock, FalseLiteral, JavaCode.variable(slot, StringType)))
val expr = Length(BoundReference(0, StringType, nullable = false))
- assert(ctx.collectInputs(Seq(expr), commonExpr, Map.empty).isEmpty)
- assert(names(ctx.collectInputs(Seq(expr), operator, Map.empty).get) == Seq(slot))
+ for (policy <- Seq(InputPolicy.wholeStageSplit, commonExpr)) {
+ val e = intercept[AssertionError](ctx.collectInputs(Seq(expr), policy, Map.empty))
+ assert(e.getMessage.contains(s"$slot cannot name a parameter") &&
+ e.getMessage.contains("JavaCode.global"), e.getMessage)
+ }
+ // An operator may keep its code inline, so collecting does not check.
+ val parameters = names(ctx.collectInputs(Seq(expr), operator, Map.empty).get)
+ assert(parameters == Seq(slot))
+ val e = intercept[AssertionError](ctx.assertParametersAreNames(parameters, "m"))
+ assert(e.getMessage.contains(s"m takes $slot") && e.getMessage.contains("JavaCode.global"),
+ e.getMessage)
+ ctx.assertParametersAreNames(Seq("value_0", "isNull_0"), "m")
+ }
+
+ test("an INPUT_ROW that is a slot fails the testing check: while collecting for a common " +
+ "expression's method, and in an operator's method that takes it") {
+ val ctx = context()
+ ctx.INPUT_ROW = ctx.addMutableState("InternalRow", "row")
Review Comment:
Applied in d6cf7c05e87.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #58881:
URL: https://github.com/apache/spark/pull/58881#discussion_r4360362094
##########
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:
Reworded in fadfa0208c1: "by the whole stage split's own check (`splitCall`) in tests". The last sentences now say where testing checks a slot handed out as a variable: in `collectInputs` where the policy asks for names, and in `assertParametersAreNames` otherwise.
##########
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:
Done in fadfa0208c1: both tests go through a new `checkExpandSlotRead(method, query, check)` next to `methodsHolding`, and the first one passes its `UTF8String` parameter check as `check`, which runs with the method and the split setting as a clue.
##########
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:
This went away with the move (see L1342): `assertParametersAreNames` runs only where the `operatorMethod` arguments become a method's parameters, and the check left in `collectInputs` runs only under `onlyNames` (see L1340).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "dongjoon-hyun (via GitHub) via reviews" <re...@spark.apache.org>.
dongjoon-hyun commented on code in PR #58881:
URL: https://github.com/apache/spark/pull/58881#discussion_r4449324648
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1926,6 +1935,19 @@ class CodegenContext extends Logging {
}
}
+ /**
+ * Under testing, asserts that each of `parameters`, the parameters of the method `method` an
+ * operator moves its code into, is a Java name. The parameters are the variables
+ * `InputPolicy.operatorMethod` collects, which passes a `VariableValue` whatever its name.
+ */
+ private[sql] def assertParametersAreNames(parameters: Seq[String], method: String): Unit = {
Review Comment:
**No test fails if one of the three calls to this check is removed.** `CollectInputsSuite` calls `assertParametersAreNames` directly (L177-180, L192), but no test reaches the calls at L2245, ExpandExec.scala:241 or AggregateCodegenSupport.scala:354 with a value that is not a name. Every producer in the tree hands out names, so if a later change drops one of these calls, or moves it out of the branch that builds the method, CI stays green. The next producer that hands out a slot as a `VariableValue` then gets a method like `doAggregate_sum(UTF8String x_mutableStateArray_0[1])` in production again. The check you ran, with `ExpandExec` handing out `JavaCode.variable` again, is not kept in any test.
The `subExpr` call can be pinned in catalyst without an operator: `subexpressionEliminationForWholeStageCodegen` with `currentVars` holding a slot `VariableValue` and a split threshold of 1, expecting an `AssertionError` that names `subExpr`. For example, in `CollectInputsSuite` (I traced this by reading the code and did not run it):
```scala
test("a subexpression method that takes a slot fails the testing check") {
withSQLConf(SQLConf.CODEGEN_METHOD_SPLIT_THRESHOLD.key -> "1") {
val ctx = context()
val slot = ctx.addMutableState("UTF8String", "value")
val input = ExprCode(EmptyBlock, FalseLiteral, JavaCode.variable(slot, StringType))
ctx.currentVars = Seq(input)
val common = Upper(BoundReference(0, StringType, nullable = false))
val e = intercept[AssertionError] {
ctx.subexpressionEliminationForWholeStageCodegen(Seq(Concat(Seq(common, common))))
}
assert(e.getMessage.contains("subExpr") && e.getMessage.contains(slot), e.getMessage)
}
}
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1322,23 +1326,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 whole stage split's own check (`splitCall`)
+ // in tests. A slot of a compacted array is not a name at all, so it has to be handed out as a
+ // `GlobalValue`, and `INPUT_ROW` cannot be one. Testing checks for one handed out as a
+ // variable: here, where the policy asks for names; elsewhere, in the method taking it
+ // (`assertParametersAreNames`).
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 isName = isJavaName(name)
+ if (Utils.isTesting && policy.onlyNames) {
Review Comment:
**Whether a mislabeled slot fails a test depends on the order of the inputs.** `take` runs only while `possible` holds: the loop at L1363 stops at the first refusal, and `take(input.value) && take(input.isNull)` at L1390 skips the null flag once the value is refused. So under `commonExpressionMethod`, a definition `Add(slot, deferred)`, where `deferred` is an input the operator has not evaluated yet, pops `deferred` first, sets `possible = false`, and never reaches the slot, so there is no `AssertionError`. `Add(deferred, slot)` does fail the test. In the same way, `wholeStageSplitPossible` (L1680) stops at the first piece that can be split, so a slot in a later piece is only reported when the Split pass runs.
Production refuses the method either way, so nothing breaks, but the check can miss a mislabeled producer depending on the operand order. Making it complete would mean walking on after a refusal, which may not be worth it. Could the comment above at least say that the check covers what the walk reaches before its first refusal?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1322,23 +1326,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 whole stage split's own check (`splitCall`)
+ // in tests. A slot of a compacted array is not a name at all, so it has to be handed out as a
+ // `GlobalValue`, and `INPUT_ROW` cannot be one. Testing checks for one handed out as a
+ // variable: here, where the policy asks for names; elsewhere, in the method taking it
+ // (`assertParametersAreNames`).
Review Comment:
**Nit:** "`INPUT_ROW` cannot be one. Testing checks for one handed out as a variable: here, ...; elsewhere, in the method taking it" reads as if every such value were checked. But `wholeStageSplitArguments` (L1700) appends `INPUT_ROW` without going through `take` or any other check, which is the gap you filed as SPARK-60510. So a slot `INPUT_ROW` read by a CASE WHEN block becomes a parameter in `splitCall`, and no testing check sees it. `@param onlyNames` (L126-128) says the same ("fails testing either way"), and its "a variable the walk takes" is a value this policy refuses rather than takes. Could both mention the exception until SPARK-60510 lands? For example:
```suggestion
// `GlobalValue`, and `INPUT_ROW` cannot be one. Testing checks for one handed out as a
// variable: here, where the policy asks for names; elsewhere, in the method taking it
// (`assertParametersAreNames`), except the row a whole stage split appends (SPARK-60510).
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1411,13 +1417,16 @@ class CodegenContext extends Logging {
}
}
- /**
- * Whether `name` is a slot of a compacted mutable state array, such as `mutableStateArray_0[3]`,
- * which `addMutableState` hands out for a state it does not inline: a field of the class.
- */
- private def isCompactedSlot(name: String): Boolean = name match {
- case CodeGenerator.CompactedSlot(array) => mutableStateNames.contains(array)
- case _ => false
+ /** Whether `name` can name a parameter: a slot of a compacted array, `array[i]`, cannot. */
+ private def isJavaName(name: String): Boolean =
+ name.nonEmpty && Character.isJavaIdentifierStart(name.head) &&
+ name.forall(Character.isJavaIdentifierPart)
Review Comment:
**Nit:** The JDK already has this check: `javax.lang.model.SourceVersion.isIdentifier`, which `ScalaReflection` uses for the same purpose at ScalaReflection.scala:405-406, together with `SourceVersion.isKeyword`. The check here accepts keywords and literals such as `this`, `int` or `null`, which cannot name a parameter either, and it looks at UTF-16 chars rather than code points. For example:
```suggestion
private def isJavaName(name: String): Boolean =
SourceVersion.isIdentifier(name) && !SourceVersion.isKeyword(name)
```
This needs `import javax.lang.model.SourceVersion`. No producer hands out such a name today, so this is only a cleanup.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggregateCodegenSupport.scala:
##########
@@ -351,6 +351,7 @@ trait AggregateCodegenSupport
if (inputVars.forall(_.isDefined)) {
val splitCodes = inputVars.flatten.zipWithIndex.map { case (args, i) =>
val doAggFunc = ctx.freshName(s"doAggregate_${aggNames(i)}")
+ ctx.assertParametersAreNames(args.map(_.variableName), doAggFunc)
Review Comment:
**Nit:** This line and the next are now the same pair of testing checks, with the same parameter names and method, at all three places an operator builds a method from `getLocalInputVariableValues`: here, CodeGenerator.scala:2245-2246 and ExpandExec.scala:241-242. A fourth such place has to remember both, and missing the new one loses the name check without any test noticing (see the comment on CodeGenerator.scala L1943). Could they be one testing helper, e.g. `ctx.assertMethodParameters(body, parameters, method)` running both, or could `assertSplitCallsWithin` check the names as well? The `computeCommonExpr` call at CodeGenerator.scala:428 would then check names that `collectInputs` already checked, which is harmless.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/MergeRowsExec.scala:
##########
@@ -156,7 +156,7 @@ case class MergeRowsExec(
|}
|$rowIdBitmap.add(${currentRowId.value});
""".stripMargin
- ExprCode(code, FalseLiteral, JavaCode.variable(rowIdBitmap, bitmapClass))
+ ExprCode(code, FalseLiteral, JavaCode.global(rowIdBitmap, bitmapClass))
Review Comment:
**Nit:** The value of this `ExprCode` is never read: the only caller, L184, takes `.code`. So this relabel changes no generated code and no input collection, and no test can observe it. The PR description says `MergeRowsExec` "hands out its row-id bitmap, also a slot, with `JavaCode.global` as well", which reads as a second producer fix that has to travel with the Expand one, e.g. in a backport. Could this line go, to keep the change to the fix, or could the description say it is for consistency only?
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/codegen/CollectInputsSuite.scala:
##########
@@ -139,29 +139,58 @@ class CollectInputsSuite extends SparkFunSuite with SQLHelper {
assert(!ctx.collectInputs(Seq(input(0)), operator, Map.empty).get.readsRow)
}
- test("a slot of a compacted mutable state array is a field under every policy with the whole " +
- "stage split on, never a parameter, which no method could declare") {
+ test("a field handed out as a global is never an argument; an inlined one handed out as a " +
+ "variable is left out only where the policy says so") {
val ctx = context()
- ctx.wholeStageSplit = WholeStageSplit.Record
- // A state of a type that is not primitive is compacted into an array, and its name is a slot.
+ val expr = Length(BoundReference(0, StringType, nullable = false))
+ // A state of a type that is not primitive is a slot of a compacted array.
val slot = ctx.addMutableState("UTF8String", "value")
assert(slot.matches("\\w+\\[\\d+\\]"), slot)
- ctx.currentVars = Seq(ExprCode(EmptyBlock, FalseLiteral, JavaCode.variable(slot, StringType)))
- val expr = Length(BoundReference(0, StringType, nullable = false))
+ ctx.currentVars = Seq(ExprCode(EmptyBlock, FalseLiteral, JavaCode.global(slot, StringType)))
for (policy <- Seq(InputPolicy.wholeStageSplit, commonExpr, operator)) {
assert(names(ctx.collectInputs(Seq(expr), policy, Map.empty).get).isEmpty, policy)
}
+ // An inlined state handed out as a variable is read as the field only by a whole stage split
+ // method.
+ val field = ctx.addMutableState("UTF8String", "field", forceInline = true)
+ ctx.currentVars = Seq(ExprCode(EmptyBlock, FalseLiteral, JavaCode.variable(field, StringType)))
+ assert(names(ctx.collectInputs(Seq(expr), InputPolicy.wholeStageSplit, Map.empty).get).isEmpty)
+ for (policy <- Seq(commonExpr, operator)) {
+ assert(names(ctx.collectInputs(Seq(expr), policy, Map.empty).get) == Seq(field), policy)
+ }
}
- test("with the whole stage split off, a slot is answered as before that split existed") {
- // A common expression's method refuses it, so its definition stays inline, and an operator's
- // method lists it among its inputs.
+ test("a slot handed out as a variable fails the testing check: while collecting where the " +
+ "policy asks for names, and in the method that takes it where it does not") {
val ctx = context()
val slot = ctx.addMutableState("UTF8String", "value")
ctx.currentVars = Seq(ExprCode(EmptyBlock, FalseLiteral, JavaCode.variable(slot, StringType)))
val expr = Length(BoundReference(0, StringType, nullable = false))
- assert(ctx.collectInputs(Seq(expr), commonExpr, Map.empty).isEmpty)
- assert(names(ctx.collectInputs(Seq(expr), operator, Map.empty).get) == Seq(slot))
+ for (policy <- Seq(InputPolicy.wholeStageSplit, commonExpr)) {
+ val e = intercept[AssertionError](ctx.collectInputs(Seq(expr), policy, Map.empty))
+ assert(e.getMessage.contains(s"$slot cannot name a parameter") &&
+ e.getMessage.contains("JavaCode.global"), e.getMessage)
+ }
+ // An operator may keep its code inline, so collecting does not check.
+ val parameters = names(ctx.collectInputs(Seq(expr), operator, Map.empty).get)
+ assert(parameters == Seq(slot))
+ val e = intercept[AssertionError](ctx.assertParametersAreNames(parameters, "m"))
+ assert(e.getMessage.contains(s"m takes $slot") && e.getMessage.contains("JavaCode.global"),
+ e.getMessage)
+ ctx.assertParametersAreNames(Seq("value_0", "isNull_0"), "m")
+ }
+
+ test("an INPUT_ROW that is a slot fails the testing check: while collecting for a common " +
+ "expression's method, and in an operator's method that takes it") {
+ val ctx = context()
+ ctx.INPUT_ROW = ctx.addMutableState("InternalRow", "row")
Review Comment:
**Nit:** This test relies on `addMutableState("InternalRow", "row")` returning a slot, but unlike the test above (L147-148) it does not assert that. If that state were ever inlined, both `intercept`s would fail with "Expected exception java.lang.AssertionError to be thrown, but no exception was thrown", which points at the check rather than at the setup. For example:
```suggestion
ctx.INPUT_ROW = ctx.addMutableState("InternalRow", "row")
assert(ctx.INPUT_ROW.matches("\\w+\\[\\d+\\]"), ctx.INPUT_ROW)
```
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/WithExpressionEvalSuite.scala:
##########
@@ -305,18 +305,18 @@ class WithExpressionEvalSuite extends SparkFunSuite with SQLHelper {
s"the innermost body was emitted $count times at depth $depth")
assert(!methods.contains("computeCommonExpr"), methods)
}
- // A value whose name is not one a parameter can take: `ExpandExec` hands out a compacted
- // mutable state array slot as if it were a local. The marker count says the fallback pasted
- // the body at both references; that a method was attempted at all is the case above, which
- // gets one from the same shape with a passable name.
+ // A value whose name is not one a parameter can take, such as a slot of a compacted mutable
+ // state array handed out as a local. Its producer has to hand it out as a field, and tests
+ // fail where a method would be built; the previous test gets a method from the same shape
+ // with a passable name.
val slot = ExprCode(EmptyBlock, FalseLiteral,
JavaCode.variable("mutableStateArray_0[3]", IntegerType))
- val (slotBody, slotMethods) = generateWithLocalInput(nested(2), slot)
- assert(!slotMethods.contains("computeCommonExpr"), slotMethods)
- assert(markersIn(slotBody + slotMethods) == 4, slotBody + slotMethods)
+ val e = intercept[AssertionError](generateWithLocalInput(nested(2), slot))
Review Comment:
**Nit:** This test is titled "the shapes a method is not possible for keep the inline body", but this case now expects an `AssertionError` instead, and its `markersIn(...) == 4` check is gone. The inline fallback for a value no parameter can name is now covered only by the `position` case below. Could the slot case move into a test of its own, e.g. "a slot handed out as a local fails the testing check where a definition's method is considered", so that this title stays true?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "MaxGekk (via GitHub)" <gi...@apache.org>.
MaxGekk commented on PR #58881:
URL: https://github.com/apache/spark/pull/58881#issuecomment-6082783706
Hi @LuciferYang, this draft fixes a failure that is reachable, so it may be worth finishing and taking as a bug fix rather than a latent trap.
**A reachable failure.** An UNPIVOT of string columns whose large common subexpression is computed in a `subExpr` method:
```sql
CREATE OR REPLACE TEMP VIEW t AS
SELECT concat('a', CAST(id AS STRING)) AS a, concat('b', CAST(id AS STRING)) AS b FROM range(10);
SELECT c,
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')) AS x,
concat(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')), 'z') AS y
FROM t UNPIVOT (v FOR c IN (a, b));
```
The `subExpr` method takes `UTF8String expand_mutableStateArray_0[1]` as a parameter, so the stage fails to compile (Janino: `')' expected instead of '['`) and falls back to the non-codegen path, or fails with the fallback off. @dongjoon-hyun reproduced it on released 4.1.3, 4.2.0 and 4.3.0 while reviewing #59225, and checked that with the two labels changed as in this PR the query compiles under Janino and javac and matches the non-codegen results, for aggregate, consume-function, broadcast-join-key and ROLLUP-over-UNPIVOT shapes too. I filed SPARK-60021 for it before finding this PR, and will link it as a duplicate of SPARK-59589. So SPARK-59589 could be a Bug affecting 4.1.3, 4.2.0 and 4.3.0 (SPARK-35329 put the outputs in mutable state in 3.2.0), and a backport candidate.
**What SPARK-33301 (#59225, merged yesterday) changed around this.** It works around the mislabel on master and branch-4.x: `CodegenContext.collectInputs` reads a compacted slot as the field it is when `spark.sql.codegen.wholeStage.splitExpressions` is on, so the query above now compiles with that conf on, and still fails with it off. With the relabel the workaround is redundant and could go in the same change: `CompactedSlot`, `isCompactedSlot` and the slot clause in `collectInputs` (`CodeGenerator.scala`), the slot case in `CollectInputsSuite`, and the two slot tests in `WholeStageCodegenSuite` ("a split method reads a slot of a compacted mutable state array as a field" and "with the split on, a subexpression method reads a compacted slot as a field"). I applied this PR on today's master (it applies cleanly) and ran these: the second of those fails, at its conf-off half, which asserts that a `subExpr` method takes a `UTF8String` parameter (`WholeStageCodegenSuite.scala:2847`), w
hich the relabel makes impossible; the first, your new test and `CollectInputsSuite` pass, so the `CollectInputsSuite` case goes only because it is redundant. The query above makes a good replacement test, since it fails before the change and passes after.
Happy to help with any of this if useful.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #58881:
URL: https://github.com/apache/spark/pull/58881#issuecomment-6283951799
cc @cloud-fan @dongjoon-hyun @MaxGekk
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #58881:
URL: https://github.com/apache/spark/pull/58881#discussion_r4360359881
##########
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:
Moved in fadfa0208c1. `collectInputs` no longer checks. `CodegenContext.assertParametersAreNames` runs at the three places that build a method from these arguments, right before `addNewFunction` and inside the branch that generates the method: the `subExpr` method, `doAggregate_*` and `switchCaseCode`. With `ExpandExec` handing out `JavaCode.variable` again, it names `project_subExpr`, `expand_switchCaseCode` and `hashAgg_doAggregate_sum` in the two SPARK-59589 tests. `CollectInputsSuite` now asserts that collecting under the operator policy does not throw and that the check does.
##########
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:
Since this PR removes the rule that read such a slot as a field, I covered these two policies here in fadfa0208c1: under testing, `collectInputs` now fails on a `VariableValue` that is not a name where the policy asks for names, instead of refusing the method silently. It fires wherever such a method is considered, including the whole stage split's recording pass, which is where the refusal used to happen silently. `WithExpressionEvalSuite` and `CollectInputsSuite` now expect the assertion for a slot, and `MergeRowsExec`, which wrapped its bitmap slot as a variable, now hands it out with `JavaCode.global`.
A check in `JavaCode.variable` itself would fire on every value created, whether or not any code moves into a method, and `collectInputs` wraps `INPUT_ROW` with it too, so it needs its own design. I filed SPARK-60510 for it, with the remaining gap: `wholeStageSplitArguments` appends `INPUT_ROW` without going through `collectInputs`.
##########
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:
Good catch. The check now says `set INPUT_ROW to a local or a field added with forceInline = true` when the parameter is `INPUT_ROW`, and keeps the other advice otherwise. The comment above `take` mentions `INPUT_ROW` too. A new `CollectInputsSuite` test sets `INPUT_ROW` to a slot and checks that both the collection for a common expression's method and an operator's method that takes it fail with that message.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #58881:
URL: https://github.com/apache/spark/pull/58881#discussion_r4333863432
##########
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:
Applied in 6435821cb1c, thanks.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "dongjoon-hyun (via GitHub) via reviews" <re...@spark.apache.org>.
dongjoon-hyun commented on code in PR #58881:
URL: https://github.com/apache/spark/pull/58881#discussion_r4464485155
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1322,23 +1330,26 @@ 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 whole stage split's own check (`splitCall`)
+ // in tests. A slot of a compacted array is not a name at all, so it has to be handed out as a
+ // `GlobalValue`, and `INPUT_ROW` cannot be one. Testing checks for one handed out as a
+ // variable: here, where the policy asks for names, for what the walk reaches before its first
+ // refusal; elsewhere, in the method taking it (`assertMethodParameters`). The row a whole stage
+ // split appends is not checked (SPARK-60510).
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 isName = isJavaName(name)
+ if (Utils.isTesting && policy.onlyNames) {
+ assert(isName, s"$name cannot name a parameter of a method this code moves into; " +
+ nameFix(name))
+ }
+ val isField = policy.fieldsNeedNoArgument && mutableStateNames.contains(name)
+ val taken = isName || !policy.onlyNames
Review Comment:
**Under testing, no test reaches the refusal of a value that is not a name.**
Every test runs with `Utils.isTesting`. So where the policy asks for names, a `VariableValue` that is not a name fails the assertion at L1347-1350 before `taken` is used. As a result, `taken` is `true` in every test run, and no test runs the branch that production relies on: refusing the method and keeping the code inline. Before this PR, `CollectInputsSuite` checked that branch with `collectInputs(Seq(expr), commonExpr, Map.empty).isEmpty`, and `WithExpressionEvalSuite` checked the inline body with `markersIn(...) == 4`. Both are now `intercept[AssertionError]`. If L1352 regressed, for example to `val taken = true`, CI would stay green. Then a producer outside Spark that hands out a slot as a `VariableValue` would get a parameter such as `UTF8String x_mutableStateArray_0[1]` in production, and the stage would fail to compile. This continues item 4 of my previous review: the slot case has its own test now, but the inline fallback for a `VariableValue` is still not covered.
Could the check sit behind a flag of the context that defaults to `Utils.isTesting`? Then one test per policy could turn it off and assert the refusal again. For example (not compiled):
```scala
/** Whether to assert, as under testing, that each variable a method takes is a Java name. */
private[sql] var checkParameterNames: Boolean = Utils.isTesting
```
with `if (checkParameterNames && policy.onlyNames)` at L1347. `assertParametersAreNames` could use the same flag.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala:
##########
@@ -1411,13 +1422,15 @@ class CodegenContext extends Logging {
}
}
- /**
- * Whether `name` is a slot of a compacted mutable state array, such as `mutableStateArray_0[3]`,
- * which `addMutableState` hands out for a state it does not inline: a field of the class.
- */
- private def isCompactedSlot(name: String): Boolean = name match {
- case CodeGenerator.CompactedSlot(array) => mutableStateNames.contains(array)
- case _ => false
+ /** Whether `name` can name a parameter: a slot of a compacted array, `array[i]`, cannot. */
+ private def isJavaName(name: String): Boolean =
+ SourceVersion.isIdentifier(name) && !SourceVersion.isKeyword(name)
+
+ /** How the producer of `name`, a value that cannot name a parameter, can hand it out instead. */
+ private def nameFix(name: String): String = if (name == INPUT_ROW) {
+ "set INPUT_ROW to a local or a field added with forceInline = true"
+ } else {
+ "hand out a field with JavaCode.global or isNullGlobal"
}
Review Comment:
**Nit:** Since `isJavaName` uses `SourceVersion.isKeyword`, it also rejects keywords and the literals `true`, `false` and `null`. But the scaladoc at L1425 still names only a slot, and `nameFix` gives the hint for a slot to every rejected name. For example, a producer that hands out `JavaCode.isNullVariable("false")` instead of `FalseLiteral` now fails with "hand out a field with JavaCode.global or isNullGlobal", which points away from the fix. No producer does this today, so this only affects the message. Could the scaladoc mention keywords and literals, and could `nameFix` give them a hint of their own? For example (not compiled):
```suggestion
/**
* Whether `name` can name a parameter: a slot of a compacted array, `array[i]`, cannot, nor can
* a keyword or a literal such as `false`.
*/
private def isJavaName(name: String): Boolean =
SourceVersion.isIdentifier(name) && !SourceVersion.isKeyword(name)
/** How the producer of `name`, a value that cannot name a parameter, can hand it out instead. */
private def nameFix(name: String): String = if (name == INPUT_ROW) {
"set INPUT_ROW to a local or a field added with forceInline = true"
} else if (SourceVersion.isKeyword(name)) {
"hand out a literal as a LiteralValue, or give the variable another name"
} else {
"hand out a field with JavaCode.global or isNullGlobal"
}
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/ExpandExec.scala:
##########
@@ -189,9 +189,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 state not inlined. 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.
Review Comment:
**Nit:** This comment explains why the value has to be a `GlobalValue`. But the next producer will read the `@return` of `addMutableState` ([CodeGenerator.scala L661-L663](https://github.com/apache/spark/blob/d6cf7c05e8797031093092070dfe7499f55544eb/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala#L661-L663)), which only says that the result may be "an array access". Could that `@return` also say how to hand the result out? For example:
```scala
* To hand it out as an `ExprValue`, use `JavaCode.global` or `isNullGlobal`, never
* `JavaCode.variable`: an array access cannot name a method parameter
* (`CodegenContext.collectInputs`).
```
That puts the advice where the mistake is made. The testing checks only catch a mislabeled value that a test reaches, and the row a whole stage split appends is not checked yet (SPARK-60510).
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala:
##########
@@ -1932,6 +1962,55 @@ class WholeStageCodegenSuite extends SharedSparkSession
query.collect().toSeq
}
+ /**
+ * Checks the query `query` builds from a FROM clause that UNPIVOTs two string columns into
+ * `v`, with the whole stage split on and off: a method named like `method` calls `lpad` on the
+ * slot of the compacted mutable state array that `ExpandExec` holds `v` in, `check` holds of
+ * the code, the code compiles, and the rows match those computed without whole stage codegen.
+ * With the slot handed out as a local variable instead, building that method fails the testing
+ * check on its parameters.
+ */
+ private def checkExpandSlotRead(
+ method: String,
+ query: String => String,
+ check: Seq[CodeAndComment] => Unit = _ => ()): Unit = {
+ withTempView("t") {
+ spark.range(10).selectExpr("concat('a', CAST(id AS STRING)) AS a",
+ "concat('b', CAST(id AS STRING)) AS b").createOrReplaceTempView("t")
+ def df: DataFrame = sql(query("FROM t UNPIVOT (v FOR c IN (a, b))"))
+ val expected = withoutWholeStage(df)
+ Seq("true", "false").foreach { split =>
+ withSQLConf(SQLConf.WHOLESTAGE_SPLIT_EXPRESSIONS.key -> split) {
+ val code = genCode(df)
+ // A subexpression method or a switch case also writes into a slot of the same array,
+ // so the array's name alone does not show a read.
+ assert(code.exists(c => methodsHolding(c, "].lpad(").exists(_.contains(method))),
+ s"$method $split")
+ withClue(s"$method $split: ")(check(code))
+ code.foreach(CodeGenerator.compile)
+ checkAnswer(df, expected)
+ }
+ }
+ withSQLConf(SQLConf.WHOLESTAGE_SPLIT_EXPRESSIONS.key -> "false") {
+ val e = intercept[AssertionError](genCodeWithLocalExpandOutput(df))
+ assert(s"\\w*$method\\w* takes \\w+\\[\\d+\\]".r.findFirstIn(e.getMessage).isDefined,
+ e.getMessage)
+ }
+ }
+ }
+
+ /**
+ * `genCode`, with the bottom `ExpandExec` of the plan handing out its output as local variables,
+ * as it did before SPARK-59589.
+ */
+ private def genCodeWithLocalExpandOutput(df: DataFrame): Seq[CodeAndComment] =
Review Comment:
**Nit:** This repeats `genCode` (L893-L898) without its `assert(wholeStageCodeGenExecs.nonEmpty, ...)`. If the plan ever has no whole stage, for example without `DisableAdaptiveExecutionSuite`, this returns an empty `Seq`. The `intercept` in `checkExpandSlotRead` then fails with "Expected exception java.lang.AssertionError to be thrown, but no exception was thrown", which reads as if the check were gone. Could `genCode` take the change of the plan instead? For example (not compiled):
```scala
def genCode(ds: Dataset[_], transform: SparkPlan => SparkPlan = identity)
: Seq[CodeAndComment] = {
val plan = transform(ds.queryExecution.executedPlan)
val wholeStageCodeGenExecs = plan.collect { case p: WholeStageCodegenExec => p }
assert(wholeStageCodeGenExecs.nonEmpty, "WholeStageCodegenExec is expected")
wholeStageCodeGenExecs.map(_.doCodeGen()._2)
}
```
and in `checkExpandSlotRead`:
```scala
val e = intercept[AssertionError](genCode(df, _.transformUp {
case x: ExpandExec if !x.child.exists(_.isInstanceOf[ExpandExec]) => LocalOutputExec(x)
}))
```
Adding the same assertion here would work too.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "LuciferYang (via GitHub) via reviews" <re...@spark.apache.org>.
LuciferYang commented on PR #58881:
URL: https://github.com/apache/spark/pull/58881#issuecomment-6650128627
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/fc227c422e81e88d947f55ddb72663c8f1e729c4
- merged into branch-4.x https://github.com/apache/spark/commit/862472cac1d21050f145ca279416aeb182610bb2
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "LuciferYang (via GitHub) via reviews" <re...@spark.apache.org>.
LuciferYang closed pull request #58881: [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals
URL: https://github.com/apache/spark/pull/58881
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59589][SQL] Hand Expand's varying output columns to the consumer as globals [spark]
Posted by "LuciferYang (via GitHub) via reviews" <re...@spark.apache.org>.
LuciferYang commented on PR #58881:
URL: https://github.com/apache/spark/pull/58881#issuecomment-6649799302
<img width="2686" height="1544" alt="image" src="https://github.com/user-attachments/assets/0f6e91ac-f395-471f-9046-a898655e2d8b" />
CI passed
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org