sunchao commented on code in PR #56575:
URL: https://github.com/apache/spark/pull/56575#discussion_r3730351829
##########
sql/connect/common/src/test/resources/query-tests/explain-results/function_right.explain:
##########
@@ -1,2 +1,2 @@
-Project [if (isnull(g#0)) null else if ((cast(g#0 as int) <= 0)) else
substring(g#0, -cast(g#0 as int), 2147483647) AS right(g, g)#0]
+Project [right(g#0, g#0) AS right(g, g)#0]
Review Comment:
[P2] Regenerate the Connect golden with the actual `With` projection
`ProtoToParsedPlanTestSuite.function_right` currently fails because
`RewriteWithExpression` inserts an intermediate `Project [..., cast(g#0 as int)
AS _common_expr_1#0]` between this project and `LocalRelation`, but this golden
contains only two lines. The Connect job on the current head shows that exact
mismatch. Please regenerate this golden from the actual analyzed plan, or
change the lowering so the additional projection is not produced.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/stringExpressions.scala:
##########
@@ -2630,29 +2630,40 @@ case class Substring(str: Expression, pos: Expression,
len: Expression)
since = "2.3.0",
group = "string_funcs")
// scalastyle:on line.size.limit
-case class Right(str: Expression, len: Expression) extends RuntimeReplaceable
- with ImplicitCastInputTypes with BinaryLike[Expression] {
-
- override lazy val replacement: Expression = If(
- IsNull(str),
- Literal(null, str.dataType),
- If(
- LessThanOrEqual(len, Literal(0)),
- Literal(UTF8String.EMPTY_UTF8, str.dataType),
- new Substring(str, UnaryMinus(len, failOnError = false))
- )
- )
+object Right extends DelegateFunction {
+ override val name: String = "right"
override def inputTypes: Seq[AbstractDataType] =
- Seq(
- StringTypeWithCollation(supportsTrimCollation = true),
- IntegerType
- )
- override def left: Expression = str
- override def right: Expression = len
- override protected def withNewChildrenInternal(
- newLeft: Expression, newRight: Expression): Expression = {
- copy(str = newLeft, len = newRight)
+ Seq(StringTypeWithCollation(supportsTrimCollation = true), IntegerType)
+
+ // At build time `str` is the not-yet-coerced argument (wrapped in an
`ImplicitCastInput` marker
+ // that delegates `dataType` to its child), so `str.dataType` is the *input*
type, which is not
+ // necessarily a string yet -- e.g. `right(12345, 2)` has an `IntegerType`
child the implicit cast
+ // will turn into a string. Use it for the null/empty branch literals only
when it is already a
+ // string-family type, so a CHAR(N)/VARCHAR(N) result (under
+ // `spark.sql.preserveCharVarcharTypeInfo`) or a non-default collation is
preserved through the
+ // `If` branch unification; otherwise fall back to plain `StringType`, the
type the implicit cast
+ // produces. Typing a UTF8String literal with a non-string type would be
invalid.
+ override def lower(args: Seq[Expression]): Expression = {
+ val str = args(0)
+ val len = args(1)
+ val litType = str.dataType match {
+ case _: StringType | _: CharType | _: VarcharType => str.dataType
+ case _ => StringType
+ }
+ // Keep each argument single-use while the analyzer extracts window
expressions. The
+ // definition needs to reference both arguments more than once, but
extracting those repeated
+ // trees independently can produce duplicate window aggregates.
RewriteWithExpression expands
+ // these bindings later, after window extraction.
+ With(str, len) { case Seq(strRef, lenRef) =>
Review Comment:
[P1] Preserve `right()`'s NULL short-circuit before evaluating `len`
Wrapping both arguments in an outer `With` lets `RewriteWithExpression`
precompute a non-cheap, twice-referenced `len` in a child `Project` before
`If(IsNull(str), ...)`. For example:
```sql
SELECT right(
IF(id = 0, CAST(NULL AS STRING), 'abc'),
raise_error(concat('boom-', CAST(id AS STRING)))
)
FROM range(1)
```
On master this returns `NULL`; with this change the hoisted length raises
`USER_RAISED_EXCEPTION` even though the string is null. Dynamic ANSI casts fail
the same way. Please keep the length evaluation inside the non-null branch
while preserving single-use window extraction.
##########
sql/core/src/test/scala/org/apache/spark/sql/DelegateExpressionQuerySuite.scala:
##########
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import org.apache.spark.sql.catalyst.analysis.resolver.ResolverRunner
+import org.apache.spark.sql.catalyst.expressions.{Alias, Cast,
DelegateExpression, ImplicitCastInput, Literal, MultiGetJsonObject,
TypeCheckInput}
+import org.apache.spark.sql.catalyst.plans.logical.{OneRowRelation, Project}
+import org.apache.spark.sql.execution.{LowerDelegateExpression,
WholeStageCodegenExec}
+import org.apache.spark.sql.execution.window.WindowExec
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.StringType
+
+/**
+ * End-to-end proof for the delegate-expression redesign: `right()` is built
as a
+ * [[DelegateExpression]] -- a logical-phase wrapper that stays readable in
the optimized plan and
+ * is lowered to its real definition (by `LowerDelegateExpression`) before
physical execution, so
+ * the planner, pushdown, columnar rules and codegen see the actual executed
expression.
+ */
+class DelegateExpressionQuerySuite extends QueryTest with SharedSparkSession {
+
+ test("right() is a DelegateExpression in the optimized plan, lowered before
execution") {
+ val df = spark.range(0, 3).selectExpr("right(concat('row', cast(id as
string)), 1) as r")
+ checkAnswer(df, Seq(Row("0"), Row("1"), Row("2")))
+
+ // Readable in the optimized logical plan ...
+ assert(df.queryExecution.optimizedPlan.exists(
+ _.expressions.exists(_.exists(_.isInstanceOf[DelegateExpression]))),
+ s"expected a DelegateExpression in the optimized
plan:\n${df.queryExecution.optimizedPlan}")
+ // ... but lowered away before physical execution, so engines see the real
expression.
+ val executed = df.queryExecution.executedPlan
+
assert(!executed.exists(_.expressions.exists(_.exists(_.isInstanceOf[DelegateExpression]))),
+ s"DelegateExpression should be lowered before execution:\n$executed")
+ assert(executed.exists(_.isInstanceOf[WholeStageCodegenExec]),
+ s"expected whole-stage codegen in the executed plan:\n$executed")
+ }
+
+ test("right() implicit-casts a non-string arg via the standard coercion rule
(no extra step)") {
+ // The old plain-form `right` was ImplicitCastInputTypes; the delegate
form preserves this by
+ // wrapping the arg in an ImplicitCastInput shim that the standard
TypeCoercion rule handles.
+ checkAnswer(spark.sql("SELECT right(12345, 2)"), Row("45"))
+ }
+
+ test("internal input shims are stripped at the end of analysis") {
+ val df = spark.range(0, 3).selectExpr("right(concat('row', cast(id as
string)), 2) as r")
+ val analyzed = df.queryExecution.analyzed
+ // The high-level delegate remains in the plan ...
+
assert(analyzed.exists(_.expressions.exists(_.exists(_.isInstanceOf[DelegateExpression]))))
+ // ... but the internal coercion shims are gone (they were inserted, then
stripped).
+ assert(!analyzed.exists(_.expressions.exists(_.exists(e =>
+ e.isInstanceOf[ImplicitCastInput] || e.isInstanceOf[TypeCheckInput]))),
+ s"input shims should be stripped after analysis:\n$analyzed")
+ }
+
+ test("right() produces identical results with whole-stage codegen on and
off") {
+ Seq("true", "false").foreach { flag =>
+ withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> flag) {
+ checkAnswer(
+ spark.range(0, 3).selectExpr("right(concat('row', cast(id as
string)), 2) as r"),
+ Seq(Row("w0"), Row("w1"), Row("w2")))
+ }
+ }
+ }
+
+ test("right() extracts a window input only once") {
+ val df = spark.sql(
+ """SELECT right(
+ | listagg(CAST(id AS STRING), chr(44))
+ | OVER (ORDER BY id ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT
ROW),
+ | 1)
+ |FROM range(5)""".stripMargin)
+
+ val windowExpressions = df.queryExecution.executedPlan.collect {
Review Comment:
[P1] Traverse the adaptive inner plan in this window regression test
`AdaptiveSparkPlanExec` is a `LeafExecNode`, so `.collect` on
`df.queryExecution.executedPlan` stops at the adaptive wrapper when AQE is
enabled by default. Current CI consequently fails with `List() had size 0`,
even though the rendered adaptive inner plan contains exactly one `Window`. Use
`AdaptiveSparkPlanHelper.collect`, unwrap `AdaptiveSparkPlanExec.executedPlan`,
or disable AQE for this assertion.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolverGuard.scala:
##########
@@ -600,7 +600,8 @@ class ResolverGuard(
_: TryValidateUTF8 | _: StringReplace | _: Overlay | _:
StringTranslate | _: FindInSet |
_: String2TrimExpression | _: StringTrimBoth | _: StringInstr | _:
SubstringIndex |
_: StringLocate | _: StringLPad | _: BinaryPad | _: StringRPad | _:
FormatString |
- _: InitCap | _: StringRepeat | _: StringSpace | _: Substring | _:
Right | _: Left |
+ _: InitCap | _: StringRepeat | _: StringSpace | _: Substring | _:
DelegateExpression |
Review Comment:
[P2] Allow the actual `With` subtree used by `right()`
Adding `DelegateExpression` alone is insufficient because `checkExpression`
recursively checks `expression.children`. Migrated `right` has
`DelegateExpression.definition = With(...)`, but `With`, `CommonExpressionDef`,
and `CommonExpressionRef` are not accepted here. With
`spark.sql.analyzer.singlePassResolver.enabledTentatively=true`, a chained
operation such as `spark.range(3).selectExpr("right(cast(id as string), 1) AS
r").selectExpr("r")` traverses the already-analyzed first `Project` and falls
back to fixed-point analysis. Previously `Right` and its argument children
passed the guard. Please support the full subtree or treat an already-resolved
delegate as opaque, and add chained-DataFrame coverage.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala:
##########
@@ -178,72 +178,63 @@ object GetJsonObject {
}
/**
- * Extracts multiple simple object-key and array-index paths from a JSON
string in one parse. This
- * is an internal expression used to share sibling [[GetJsonObject]]
expressions; unsupported and
- * prefix-conflicting JSON paths remain as independent GetJsonObject
expressions.
+ * Builds the internal expression that extracts multiple simple object-key and
array-index paths
+ * from a JSON string in one parse, used to share sibling [[GetJsonObject]]
expressions; unsupported
+ * and prefix-conflicting paths remain as independent `GetJsonObject`
expressions.
+ *
+ * It is inserted by `OptimizeCsvJsonExprs` (after analysis, so its inputs are
resolved), and is the
+ * optimizer-constructed showcase for [[DelegateExpression]]: instead of
hand-written
+ * eval/doGenCode, it builds a typed delegate directly -- the high-level call
+ * `multi_get_json_object(json, p1, ..., pn)` stays visible via `inputs`,
while the `definition`
+ * delegates evaluation to [[MultiGetJsonObjectEvaluator]] through an
`Invoke`. The delegate stays
+ * in logical plans and is lowered to its definition before physical planning.
*/
-case class MultiGetJsonObject(
- json: Expression,
- fallbackPaths: Seq[String])
- extends UnaryExpression
- with ExpectsInputTypes {
-
- // OptimizeCsvJsonExprs caps shared path depth to keep evaluator recursion
stack-safe.
- require(fallbackPaths.nonEmpty)
-
- override def child: Expression = json
-
- override def inputTypes: Seq[AbstractDataType] =
- Seq(StringTypeWithCollation(supportsTrimCollation = true))
-
- override lazy val dataType: DataType = StructType(fallbackPaths.indices.map
{ index =>
- StructField(s"_$index", StringType, nullable = true)
- })
-
- override def nullable: Boolean = true
-
- // This internal unary expression always returns null when its JSON child is
null.
- override def nullIntolerant: Boolean = true
-
- override def prettyName: String = "multi_get_json_object"
-
- final override val nodePatterns: Seq[TreePattern] = Seq(GET_JSON_OBJECT)
-
- @transient
- private lazy val simplePaths = fallbackPaths.map { path =>
- GetJsonObject.simplePath(UTF8String.fromString(path)).getOrElse {
- throw new IllegalArgumentException(s"Unsupported shared JSON path:
$path")
+object MultiGetJsonObject {
+ val name: String = "multi_get_json_object"
+
+ def apply(json: Expression, fallbackPaths: Seq[String]): DelegateExpression
= {
+ // OptimizeCsvJsonExprs caps shared path depth to keep evaluator recursion
stack-safe.
+ require(fallbackPaths.nonEmpty)
+ val resultType = StructType(fallbackPaths.indices.map { index =>
+ StructField(s"_$index", StringType, nullable = true)
+ })
+ val utf8Paths = fallbackPaths.map(UTF8String.fromString)
+ val simplePaths = utf8Paths.map { path =>
+ GetJsonObject.simplePath(path).getOrElse {
+ throw new IllegalArgumentException(s"Unsupported shared JSON path:
$path")
+ }
}
+ val evaluator = MultiGetJsonObjectEvaluator(utf8Paths, simplePaths)
+ // `propagateNull = true` reproduces the old null-intolerant behavior:
null json -> null result.
+ val definition = Invoke(
+ Literal.create(evaluator,
ObjectType(classOf[MultiGetJsonObjectEvaluator])),
Review Comment:
[P1] Preserve both JSON statefulness and evaluator-per-copy isolation
The previous `MultiGetJsonObject` had `stateful = true` and a per-expression
lazy evaluator. This factory now returns a `DelegateExpression` reporting
`stateful = false`, which already fails existing `JsonExpressionsSuite` on this
head: `multiGetJsonObject.stateful was false`.
There is a second problem beyond that assertion: `Invoke` itself is stateful
and gets copied, but its target is this stateless `ObjectType` literal
containing the same `MultiGetJsonObjectEvaluator`. Fresh expression copies
therefore still share the evaluator's mutable `outputBuffer` and fallback
evaluators. Simply forwarding `definition.stateful` fixes the assertion but not
evaluator isolation; please preserve both contracts.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/stringExpressions.scala:
##########
@@ -2630,29 +2630,40 @@ case class Substring(str: Expression, pos: Expression,
len: Expression)
since = "2.3.0",
group = "string_funcs")
// scalastyle:on line.size.limit
-case class Right(str: Expression, len: Expression) extends RuntimeReplaceable
- with ImplicitCastInputTypes with BinaryLike[Expression] {
-
- override lazy val replacement: Expression = If(
- IsNull(str),
- Literal(null, str.dataType),
- If(
- LessThanOrEqual(len, Literal(0)),
- Literal(UTF8String.EMPTY_UTF8, str.dataType),
- new Substring(str, UnaryMinus(len, failOnError = false))
- )
- )
+object Right extends DelegateFunction {
+ override val name: String = "right"
override def inputTypes: Seq[AbstractDataType] =
- Seq(
- StringTypeWithCollation(supportsTrimCollation = true),
- IntegerType
- )
- override def left: Expression = str
- override def right: Expression = len
- override protected def withNewChildrenInternal(
- newLeft: Expression, newRight: Expression): Expression = {
- copy(str = newLeft, len = newRight)
+ Seq(StringTypeWithCollation(supportsTrimCollation = true), IntegerType)
+
+ // At build time `str` is the not-yet-coerced argument (wrapped in an
`ImplicitCastInput` marker
+ // that delegates `dataType` to its child), so `str.dataType` is the *input*
type, which is not
+ // necessarily a string yet -- e.g. `right(12345, 2)` has an `IntegerType`
child the implicit cast
+ // will turn into a string. Use it for the null/empty branch literals only
when it is already a
+ // string-family type, so a CHAR(N)/VARCHAR(N) result (under
+ // `spark.sql.preserveCharVarcharTypeInfo`) or a non-default collation is
preserved through the
+ // `If` branch unification; otherwise fall back to plain `StringType`, the
type the implicit cast
+ // produces. Typing a UTF8String literal with a non-string type would be
invalid.
+ override def lower(args: Seq[Expression]): Expression = {
+ val str = args(0)
+ val len = args(1)
+ val litType = str.dataType match {
+ case _: StringType | _: CharType | _: VarcharType => str.dataType
+ case _ => StringType
+ }
+ // Keep each argument single-use while the analyzer extracts window
expressions. The
+ // definition needs to reference both arguments more than once, but
extracting those repeated
+ // trees independently can produce duplicate window aggregates.
RewriteWithExpression expands
+ // these bindings later, after window extraction.
+ With(str, len) { case Seq(strRef, lenRef) =>
+ If(
+ IsNull(strRef),
+ Literal(null, litType),
+ If(
+ LessThanOrEqual(lenRef, Literal(0)),
Review Comment:
[P2] Keep failed length coercions attributed to `right(...)`
`With.children` visits its main child before its definitions, so
`CheckAnalysis` reaches this `LessThanOrEqual(CommonExpressionRef, 0)` before
the invalid `ImplicitCastInput` in the length definition. `SELECT right('abc',
array(1))` now fails with `DATATYPE_MISMATCH.BINARY_OP_DIFF_TYPES` and `sqlExpr
= "(commonexpressionref() <= 0)"`, instead of the previous
`DATATYPE_MISMATCH.UNEXPECTED_INPUT_TYPE` for the second `right` argument. The
new `DelegateExpressionQuerySuite` already fails with exactly this mismatch.
Please validate the argument marker before exposing internal `With` references.
--
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]
