sunchao commented on code in PR #56575:
URL: https://github.com/apache/spark/pull/56575#discussion_r3816848826
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/stringExpressions.scala:
##########
@@ -2690,29 +2690,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 both arguments single-use while the analyzer extracts window
expressions. The length
+ // is bound inside the non-null branch so right's null short-circuit is
preserved.
+ With(str) { case Seq(strRef) =>
Review Comment:
[P1] Preserve enclosing evaluation boundaries around right()
The outer `With(str)` lets `RewriteWithExpression` hoist a non-cheap,
repeated string argument into a child `Project`, even when an enclosing
expression would skip evaluating the `right()` call. For example:
```sql
SELECT transform(
IF(id = 0, array(), array('x')),
x -> right(
IF(id = 0, CAST(raise_error('boom') AS STRING), 'b'),
1))
FROM range(2)
```
The equivalent pre-PR `If`/`Substring` definition returns `[]` and `['b']`.
The delegate version instead throws `USER_RAISED_EXCEPTION` on the empty-array
row, before the lambda should run. The same problem occurs when `right()` is
the second argument of `instr` and its first argument is null on that row. I
reproduced both cases with codegen enabled and disabled; all four
legacy-expression controls passed. Could we keep the computation inside its
enclosing evaluation boundary while preserving single-use window extraction?
##########
sql/core/src/test/scala/org/apache/spark/sql/DelegateExpressionQuerySuite.scala:
##########
@@ -0,0 +1,285 @@
+/*
+ * 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.adaptive.AdaptiveSparkPlanHelper
+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 with AdaptiveSparkPlanHelper {
+
+ 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() does not evaluate the length when the string is null") {
+ checkAnswer(
+ spark.sql(
+ """SELECT right(
+ | IF(id = 0, CAST(NULL AS STRING), 'abc'),
+ | raise_error(concat('boom-', CAST(id AS STRING))))
+ |FROM range(1)""".stripMargin),
+ Row(null))
+ }
+
+ test("right() supports constant inline-table values") {
+ checkAnswer(spark.sql("VALUES (right('abc', 1))"), Row("c"))
+ }
+
+ 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 = collect(df.queryExecution.executedPlan) {
+ case window: WindowExec => window.windowExpression
+ }.flatten
+ assert(windowExpressions.size == 1,
+ s"right() should produce one window aggregate,
got:\n${df.queryExecution.executedPlan}")
+ checkAnswer(df, (0 until 5).map(i => Row(i.toString)))
+ }
+
+ test("right() extracts a window length only once") {
+ val df = spark.sql(
+ """SELECT right(
+ | 'abcdef',
+ | CAST(sum(id) OVER (ORDER BY id) AS INT))
+ |FROM range(5)""".stripMargin)
+
+ val windowExpressions = collect(df.queryExecution.executedPlan) {
+ case window: WindowExec => window.windowExpression
+ }.flatten
+ assert(windowExpressions.size == 1,
+ s"right() should produce one window aggregate,
got:\n${df.queryExecution.executedPlan}")
+ checkAnswer(df, Seq("", "f", "def", "abcdef", "abcdef").map(Row(_)))
+ }
+
+ test("optimizer-inserted MultiGetJsonObject is a delegate in the optimized
plan, lowered " +
+ "before execution") {
+ import testImplicits._
+ withSQLConf(SQLConf.GET_JSON_OBJECT_SHARED_PARSING_ENABLED.key -> "true") {
+ val df = Seq("""{"a":1,"b":2}""").toDF("j")
+ .selectExpr("get_json_object(j, '$.a') as a", "get_json_object(j,
'$.b') as b")
+ checkAnswer(df, Row("1", "2"))
+
+ // The two sibling get_json_object calls were shared into one delegate,
readable in the
+ // optimized plan ...
+ assert(df.queryExecution.optimizedPlan.exists(
+ _.expressions.exists(_.exists(MultiGetJsonObject.isInstance))),
+ s"expected a multi_get_json_object delegate in the optimized plan")
+ // ... and lowered to its Invoke definition before execution.
+ val executed = df.queryExecution.executedPlan
+
assert(!executed.exists(_.expressions.exists(_.exists(MultiGetJsonObject.isInstance))),
+ s"delegate should be lowered before execution:\n$executed")
+ assert(executed.exists(_.isInstanceOf[WholeStageCodegenExec]))
+ }
+ }
+
+ test("right() resolves cleanly under the single-pass resolver (input-type
markers stripped)") {
+ // The single-pass resolver builds DelegateFunctions through the same
registry path (inserting
+ // the input-type markers) but has no fixed-point batch to strip them;
FunctionResolver must
+ // remove them after coercion, else the Unevaluable markers would reach
execution. We assert at
+ // the analyzed-plan level (where the fix lives): single-pass does not yet
support the
+ // DeserializeToObject operator a typed `collect`/`checkAnswer`
introduces, so the right()
+ // execution results stay covered by the fixed-point tests above.
+ withSQLConf(SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED.key -> "true") {
+ // 12345 (int) exercises the ImplicitCastInput path: it must be cast to
string.
+ val analyzed = spark.sql("SELECT right(12345, 2) AS
r").queryExecution.analyzed
+ // single-pass actually ran (there is no fallback when the conf is on)
...
+
assert(analyzed.getTagValue(ResolverRunner.SINGLE_PASS_ANALYSIS_MARKER).contains(true),
+ s"expected single-pass analysis to run:\n$analyzed")
+
assert(analyzed.exists(_.expressions.exists(_.exists(_.isInstanceOf[DelegateExpression]))),
+ s"expected the right() delegate in the analyzed plan:\n$analyzed")
+ // ... the DelegateFunction's input-type markers were stripped ...
+ assert(!analyzed.exists(_.expressions.exists(_.exists(e =>
+ e.isInstanceOf[ImplicitCastInput] || e.isInstanceOf[TypeCheckInput]))),
+ s"input shims should be stripped under single-pass:\n$analyzed")
+ // ... and the implicit cast the marker drove still applies (the marker
was removed, not the
+ // Cast).
+
assert(analyzed.exists(_.expressions.exists(_.exists(_.isInstanceOf[Cast]))),
+ s"expected the implicit Cast to survive marker removal:\n$analyzed")
+ }
+ }
+
+ test("ResolverGuard accepts right() in an already-resolved subtree") {
+ withSQLConf(SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED_TENTATIVELY.key
-> "true") {
+ val analyzed = spark.range(3)
+ .selectExpr("right(cast(id as string), 1) AS r")
+ .selectExpr("r")
+ .queryExecution.analyzed
+
assert(analyzed.getTagValue(ResolverRunner.SINGLE_PASS_ANALYSIS_MARKER).contains(true),
+ s"expected single-pass analysis to run:\n$analyzed")
+ }
+ }
+
+ test("hybrid analyzer ignores display-only state in delegate inputs") {
+ withSQLConf(
+ SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key -> "true",
+ SQLConf.ANALYZER_DUAL_RUN_SAMPLE_RATE.key -> "1.0",
+ SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED_TENTATIVELY.key -> "false",
+ SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_EXPOSE_RESOLVER_GUARD_FAILURE.key
-> "true") {
+ val analyzed = spark.sql(
+ "SELECT right(assert_true(CAST(1 AS BOOLEAN)),
1)").queryExecution.analyzed
+ assert(analyzed.resolved)
+ }
+ }
+
+ test("LowerDelegateExpression fully unwraps a directly-nested
delegate-of-delegate") {
+ // A delegate whose `definition` is itself a delegate (e.g. one delegate
function composing
+ // another). transformDown does not re-apply the rule to the replacement
it produces, so the
+ // rule must unwrap the chain itself -- otherwise the inner wrapper would
reach the planner.
+ val inner = DelegateExpression("inner", Seq(Literal(1)), Literal(1))
+ val outer = DelegateExpression("outer", Seq(Literal(1)), inner)
+ val lowered = LowerDelegateExpression(Project(Seq(Alias(outer, "c")()),
OneRowRelation()))
+
assert(!lowered.exists(_.expressions.exists(_.exists(_.isInstanceOf[DelegateExpression]))),
+ s"nested delegates should be fully lowered:\n$lowered")
+ }
+
+ test("right() preserves the input column's collation in its output type") {
+ // `Right.lower` builds the null/empty `If` branches as plain StringType
literals (it cannot
+ // read the not-yet-coerced arg's dataType); type coercion then re-unifies
the branches to the
+ // column's collation, since string literals carry the weakest collation
strength.
+ val df = spark.sql("SELECT right('Hello' COLLATE UTF8_LCASE, 3) AS r")
+ assert(df.schema("r").dataType === StringType("UTF8_LCASE"),
+ s"right() should preserve the UTF8_LCASE collation, got
${df.schema("r").dataType}")
+ checkAnswer(df, Row("llo"))
+ }
+
+ test("right() preserves the input CHAR/VARCHAR type with
preserveCharVarcharTypeInfo") {
+ // `Right.lower` types its null/empty `If` branch literals with
`str.dataType` (the resolved
+ // input type the marker delegates), so the result keeps CHAR(N) instead
of being widened to
+ // plain string when type coercion unifies the branches.
+ withSQLConf(SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true") {
+ checkAnswer(spark.sql("SELECT typeof(right(CAST('abc' AS CHAR(5)), 2))
AS t"), Row("char(5)"))
+ }
+ }
+
+ test("right() rejects a wrong number of arguments with WRONG_NUM_ARGS") {
+ // `DelegateFunction.build` validates arity before lowering, so too
few/too many arguments fail
+ // with the structured error rather than an IndexOutOfBoundsException or a
silently ignored
+ // extra arg.
+ Seq("SELECT right('abcd')", "SELECT right('abcd', 1, 99)").foreach { q =>
+ val e = intercept[AnalysisException](spark.sql(q))
+ assert(e.getCondition == "WRONG_NUM_ARGS.WITHOUT_SUGGESTION",
+ s"unexpected error condition for `$q`: ${e.getCondition}")
+ }
+ }
+
+ test("a delegate over a HAVING aggregate gets a clean generated name
(TempResolvedColumn " +
+ "trimmed)") {
+ // SPARK-52385-style: in an aggregate/HAVING, `v` is wrapped in a
`TempResolvedColumn` while it
+ // resolves against the grouping input. That wrapper rides in the
delegate's display-only
+ // `inputs`, which the pretty-printer's `transform` never rewrites;
without an explicit trim the
+ // generated column name would leak `right(tempresolvedcolumn(v), 1)`
instead of `right(v, 1)`.
+ import testImplicits._
+ withTempView("hav") {
+ Seq((1, 3), (1, 5)).toDF("k", "v").createOrReplaceTempView("hav")
+ val df = spark.sql("SELECT max(right(v, 1)) FROM hav HAVING max(right(v,
1)) IS NOT NULL")
+ val name = df.schema.fields.head.name
+ assert(!name.contains("tempresolvedcolumn"),
+ s"generated name should not leak the temp-resolution marker, got:
$name")
+ assert(name == "max(right(v, 1))", s"unexpected generated name: $name")
+ }
+ }
+
+ test("an un-castable argument is reported against the delegate call, not the
internal marker") {
+ // A failed implicit cast leaves the `ImplicitCastInput` marker in the
tree so `CheckAnalysis`
+ // (walking bottom-up) can reject it. The marker is an analysis-only
detail, so it reports as if
+ // the check ran on the high-level `right(...)` call: the `sqlExpr` stays
`right('abc', ...)`
+ // (not `implicitcastinput(...)`) and `paramIndex` is the real argument
position (`second`),
+ // matching the pre-delegate `right` -- no user-facing change.
+ checkError(
+ exception = intercept[AnalysisException](spark.sql("SELECT right('abc',
array(1))")),
+ condition = "DATATYPE_MISMATCH.UNEXPECTED_INPUT_TYPE",
+ parameters = Map(
+ "sqlExpr" -> "\"right(abc, array(1))\"",
+ "paramIndex" -> "second",
+ "requiredType" -> "\"INT\"",
+ "inputSql" -> "\"array(1)\"",
+ "inputType" -> "\"ARRAY<INT>\""),
+ queryContext = Array(ExpectedContext(
+ fragment = "right('abc', array(1))", start = 7, stop = 28)))
+
+ checkError(
+ exception = intercept[AnalysisException](spark.sql("SELECT
right(array(1), 1)")),
+ condition = "DATATYPE_MISMATCH.UNEXPECTED_INPUT_TYPE",
+ parameters = Map(
+ "sqlExpr" -> "\"right(array(1), 1)\"",
+ "paramIndex" -> "first",
+ "requiredType" -> "\"STRING\"",
+ "inputSql" -> "\"array(1)\"",
+ "inputType" -> "\"ARRAY<INT>\""),
+ queryContext = Array(ExpectedContext(
+ fragment = "right(array(1), 1)", start = 7, stop = 23)))
Review Comment:
[P2] Correct the expected error-context end position
`right(array(1), 1)` is 18 characters long and starts at index 7, so its
inclusive ending position is 24, not 23. The current
`DelegateExpressionQuerySuite` fails with `queryContext[0].stopIndex: expected
23 but got 24`. The actual error class, parameters, fragment, and starting
position all match; this is a test-only expectation error. Please change `stop`
to 24.
--
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]
