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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
