sunchao commented on code in PR #56575:
URL: https://github.com/apache/spark/pull/56575#discussion_r3502269821
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala:
##########
@@ -817,7 +817,10 @@ object QueryExecution {
plan: LogicalPlan): SparkPlan = {
// TODO: We use next(), i.e. take the first plan returned by the planner,
here for now,
// but we will implement to choose the best plan.
- planner.plan(ReturnAnswer(plan)).next()
+ // Strip DelegateExpression to its definition right before planning, so
the planner and every
+ // physical consumer (pushdown, columnar rules, codegen) sees the real
executed expression. The
+ // wrapper is purely informational and stays in the optimized logical plan
for EXPLAIN.
+ planner.plan(ReturnAnswer(LowerDelegateExpression(plan))).next()
Review Comment:
[P2] Expose delegated equalities to logical join consumers
This lowering is late enough for physical join selection, but several
logical consumers run before `createSparkPlan`. `ExtractEquiJoinKeys` splits
the join condition and matches only a top-level `EqualTo`/`EqualNullSafe`;
making `definition` a child does not help that pattern match. A supported
boolean delegate whose definition is `EqualTo(leftKey, rightKey)` therefore
reaches the physical planner as a valid equi-join, but earlier consumers miss
the keys: `PartitionPruning`/`InjectRuntimeFilter` lose DPP and runtime
filters, `JoinEstimation`/`DistinctKeyVisitor` lose equi-key statistics, and
`StreamingJoinHelper` can reject a watermarked stream-stream outer join. The
new V2 tests already treat predicate delegates as a supported shape.
Could the logical join consumers see the exposed definition as well
(including `PartitionPruning`'s direct conjunct match), with focused
DPP/runtime-filter and watermarked streaming-join regressions? This is distinct
from the older physical join-key thread: that planner path is fixed by this
line; the remaining gap is before planning.
_[ :robot: posted by Codex on behalf of sunchao using the code-review-for-me
skill :robot: ]_
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/NormalizePlan.scala:
##########
@@ -86,7 +86,15 @@ object NormalizePlan extends PredicateHelper {
* we must normalize them to check if two different queries are identical.
*/
def normalizeExprIds(plan: LogicalPlan): LogicalPlan = {
- plan.transformAllExpressions {
+ // Defined as a named rule (rather than inline) so it can also be applied
to a
+ // `DelegateExpression`'s `inputs`, which are display-only metadata -- not
children -- and so
+ // are never reached by `transformAllExpressions`. Normalizing them
explicitly keeps the
+ // informational call deterministic across runs (e.g. `right(g#0, g#0)` in
EXPLAIN), since expr
+ // ids come from a process-global counter; the `definition` child is
reached by the normal
+ // traversal.
+ lazy val rule: PartialFunction[Expression, Expression] = {
+ case d: DelegateExpression =>
+ d.copy(inputs = d.inputs.map(_.transform(rule)))
Review Comment:
[P2] Apply the full expression normalization to metadata inputs
This special case applies only the expr-id rule to `inputs`. The earlier
`normalizeExpressions` pass also uses `transformAllExpressions`, so it skips
these non-child inputs entirely. Random seeds, `CommonExpressionId`s, and
`TempResolvedColumn` wrappers in the display metadata therefore remain
unnormalized. In Hybrid Analyzer mode, for example, fixed-point and single-pass
resolution can assign different seeds to `right(rand(), 1)`; the `Rand` inside
`definition` is normalized to seed 0, but the `Rand` in `inputs` is not, so the
structurally compared plans can still fail with a false
`HYBRID_ANALYZER_LOGICAL_PLAN_COMPARISON_MISMATCH`.
Could delegate inputs go through the same full expression-normalization
pipeline rather than only this expr-id partial function? A dual-run
`right(rand(), 1)` case plus an aggregate/HAVING case that carries
`TempResolvedColumn` metadata would protect both dimensions.
_[ :robot: posted by Codex on behalf of sunchao using the code-review-for-me
skill :robot: ]_
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/package.scala:
##########
@@ -127,6 +127,16 @@ package object util extends Logging {
),
dataType = r.dataType
)
+ case d: DelegateExpression =>
+ // `inputs` are display-only metadata, not children, so `transform`
never descends into them.
+ // Render the high-level call with each input prettified (qualifiers
stripped, string literals
+ // unquoted, ...) so generated column names match the pre-delegate
function, e.g. the column
+ // name for `right(c7, 2)` stays `right(c7, 2)` rather than
`right(spark_catalog....c7, 2)`.
+ PrettyAttribute(
+ name = s"${d.name}(${d.inputs
+ .map(i => toPrettySQL(i,
shouldTrimTempResolvedColumn)).mkString(", ")})",
Review Comment:
[P2] Trim temporary resolution markers before rendering delegate inputs
Passing `shouldTrimTempResolvedColumn = true` into `toPrettySQL` does not
actually unwrap a `TempResolvedColumn`: `usePrettyExpression` has no marker
case, so it only prettifies the marker's child and leaves
`tempresolvedcolumn(...)` around it. The `InheritAnalysisRules` branch avoids
this by explicitly applying `trimTempResolvedColumn` to its non-child
parameters first. Delegate inputs are also non-children, so aggregate/HAVING
alias generation can now produce names such as
`count(right(tempresolvedcolumn(v), 1))` instead of the prior `count(right(v,
1))`.
Could this mirror the `proposedParameters` handling above and pre-trim
`d.inputs` when the flag is set? A SPARK-52385-style aggregate/HAVING
regression using `right` would catch the generated-name change.
_[ :robot: posted by Codex on behalf of sunchao using the code-review-for-me
skill :robot: ]_
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/DelegateExpression.scala:
##########
@@ -0,0 +1,183 @@
+/*
+ * 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.catalyst.expressions
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.ExpressionBuilder
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext,
ExprCode}
+import org.apache.spark.sql.catalyst.trees.TreePattern.{DELEGATE_EXPRESSION,
INPUT_TYPE_MARKER, TreePattern}
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.types.{AbstractDataType, AnyDataType, DataType}
+
+/**
+ * A transparent, named delegate over a `definition` expression -- a
LOGICAL-phase construct.
+ *
+ * `DelegateExpression` lets a high-level function (e.g. `right(a, b)`) stay
readable in the
+ * analyzed and optimized logical plan, and lets optimizer rules introduce
such nodes (e.g.
+ * `multi_get_json_object`), without hand-written `eval`/`doGenCode`. Every
behavior delegates to
+ * `definition`, a real child fully visible to the analyzer and optimizer.
+ *
+ * `name`/`inputs` are purely informational (EXPLAIN/SQL): nothing enforces
that `definition`
+ * matches what they claim, so the wrapper is never exposed to physical
planning or external
+ * systems.
+ * `LowerDelegateExpression` strips it to `definition` in
`QueryExecution.createSparkPlan` -- the
+ * single entry point to the planner, used by both the main query and AQE
re-planning -- so the
+ * planner and every physical consumer (join-key extraction, data source
pushdown, columnar rules,
+ * codegen) sees the real executed expression. (Data source V2 pushdown runs
earlier, in the logical
+ * optimizer, so it unfolds the wrapper directly in `V2ExpressionBuilder`.)
The wrapper survives the
+ * logical optimizer, so the optimized plan stays readable and optimizer rules
can introduce these
+ * nodes; `eval`/`doGenCode` still delegate, as a safety net if a delegate
ever reaches execution.
+ *
+ * Note: because the strip runs before planning, a `DelegateExpression`
created by a *physical* rule
+ * (after `createSparkPlan`) is not stripped and may reach an external system
un-lowered. That is
+ * acceptable -- like any other expression the system does not recognize, it
simply falls back, and
+ * `eval`/`doGenCode` keep it correct within Spark. Analysis- and
optimizer-inserted nodes (the
+ * common case) are always stripped, so physical-rule insertion is the only
uncovered path.
+ */
+case class DelegateExpression(
+ name: String,
+ inputs: Seq[Expression],
+ definition: Expression)
+ extends Expression with UnaryLike[Expression] {
+
+ override def child: Expression = definition
+ override def dataType: DataType = definition.dataType
+ override def nullable: Boolean = definition.nullable
+ override def foldable: Boolean = definition.foldable
+ // Delegate `nullIntolerant` too (it is not derived from children, unlike
`throwable`), so that
+ // null-intolerance optimizations -- `IsNotNull`-constraint inference in
+ // `QueryPlanConstraints.scanNullIntolerantAttribute` and
`NullPropagation`'s `IsNull`/`IsNotNull`
+ // simplifications -- still fire while the wrapper is in the logical plan
(e.g. for the
+ // `multi_get_json_object` delegate, whose `Invoke` definition is
null-intolerant).
+ override def nullIntolerant: Boolean = definition.nullIntolerant
+ override lazy val deterministic: Boolean = definition.deterministic
+ override lazy val canonicalized: Expression = definition.canonicalized
+
+ override def eval(input: InternalRow): Any = definition.eval(input)
+ override protected def doGenCode(ctx: CodegenContext, ev: ExprCode):
ExprCode =
+ definition.genCode(ctx)
+
+ final override val nodePatterns: Seq[TreePattern] = Seq(DELEGATE_EXPRESSION)
+ override protected def withNewChildInternal(newChild: Expression):
DelegateExpression =
+ copy(definition = newChild)
+
+ override def prettyName: String = name
+ override def sql: String = s"$name(${inputs.map(_.sql).mkString(", ")})"
+ override def toString: String = s"$name(${inputs.mkString(", ")})"
+}
+
+/**
+ * Analysis-only marker that requests an implicit cast of `child` to
`expectedType`: it declares the
+ * expected type so the standard `TypeCoercion` rule casts the child, then is
removed at the end of
+ * analysis by
[[org.apache.spark.sql.catalyst.analysis.RemoveInputTypeMarkers]]. It never
reaches
+ * execution, hence [[Unevaluable]]. Modeled on
+ * [[org.apache.spark.sql.catalyst.analysis.TempResolvedColumn]].
+ */
+case class ImplicitCastInput(child: Expression, expectedType: AbstractDataType)
+ extends UnaryExpression with Unevaluable with ImplicitCastInputTypes {
+ override def inputTypes: Seq[AbstractDataType] = Seq(expectedType)
+ override def dataType: DataType = child.dataType
+ override def nullable: Boolean = child.nullable
+ override lazy val canonicalized: Expression = child.canonicalized
+ final override val nodePatterns: Seq[TreePattern] = Seq(INPUT_TYPE_MARKER)
+ override protected def withNewChildInternal(newChild: Expression):
ImplicitCastInput =
+ copy(child = newChild)
+}
+
+/**
+ * Analysis-only marker that requires `child` to already match `expectedType`
(no cast is inserted),
+ * failing analysis otherwise. Removed at the end of analysis like
[[ImplicitCastInput]].
+ */
+case class TypeCheckInput(child: Expression, expectedType: AbstractDataType)
+ extends UnaryExpression with Unevaluable with ExpectsInputTypes {
+ override def inputTypes: Seq[AbstractDataType] = Seq(expectedType)
+ override def dataType: DataType = child.dataType
+ override def nullable: Boolean = child.nullable
+ override lazy val canonicalized: Expression = child.canonicalized
+ final override val nodePatterns: Seq[TreePattern] = Seq(INPUT_TYPE_MARKER)
+ override protected def withNewChildInternal(newChild: Expression):
TypeCheckInput =
+ copy(child = newChild)
+}
+
+/**
+ * The per-function object each built-in function defines (e.g. `object Right
extends
+ * DelegateFunction`). It is just an [[ExpressionBuilder]] -- registered with
the ordinary
+ * `expressionBuilder(...)`, with its `@ExpressionDescription` annotation read
off the object as
+ * usual -- specialized for the delegate pattern: replace the
`InheritAnalysisRules` ceremony with
+ * one `lower` method plus a couple of flags. `apply` is the
direct-construction entry point.
+ *
+ * Input-type contract, covering all three cases (applied per argument):
+ * - `inputTypes` empty (or `AnyDataType` for a position): accept any type
(no check, no cast).
+ * - `inputTypes` set, `implicitCast = true` (default): implicit-cast each
arg to its type.
+ * - `inputTypes` set, `implicitCast = false` : type-check each
arg, no cast.
+ */
+trait DelegateFunction extends ExpressionBuilder {
+ def name: String
+ def inputTypes: Seq[AbstractDataType] = Nil
+ def implicitCast: Boolean = true
+
+ /** Lower the function into the expression it delegates to. */
+ def lower(args: Seq[Expression]): Expression
+
+ /**
+ * ExpressionBuilder contract: invoked by the registry during function
resolution. ONLY this
+ * (analysis-time) path inserts the input-type markers, because the
analyzer's `TypeCoercion`
+ * casts them and `RemoveInputTypeMarkers` strips them afterwards.
+ */
+ override def build(funcName: String, expressions: Seq[Expression]):
Expression = {
+ // `inputTypes` carries one entry per argument position (`AnyDataType` for
an accept-any-type
+ // position), so when it is set its length is the function's arity.
Validate it here so a
+ // wrong-arity call fails with the structured WRONG_NUM_ARGS error rather
than an
+ // IndexOutOfBounds from `lower` (too few args) or a silently-ignored
extra argument (too many).
+ // An empty `inputTypes` marks a variadic function whose `lower` accepts
any argument count.
+ if (inputTypes.nonEmpty && expressions.length != inputTypes.length) {
+ throw QueryCompilationErrors.wrongNumArgsError(
+ funcName, Seq(inputTypes.length), expressions.length)
+ }
+ val args = expressions.zipWithIndex.map { case (e, i) =>
+ val expected = if (i < inputTypes.length) inputTypes(i) else AnyDataType
+ expected match {
+ case AnyDataType => e
+ case t if implicitCast => ImplicitCastInput(e, t)
Review Comment:
[P2] Keep failed coercions attributed to the delegate call
A failed cast leaves this `ImplicitCastInput` in the tree (correctly, so
`CheckAnalysis` can reject it), but `CheckAnalysis` walks bottom-up and calls
`failOnTypeCheckResult` on the marker before it reaches the outer delegate. For
example, `SELECT right('abc', array(1))` is now reported against
`implicitcastinput(array(1))`, and because the marker has one child it
identifies the *first* parameter. On master the failing expression is
`right('abc', array(1))` and the bad input is the second parameter. This
exposes an analysis-only implementation detail and changes the public error
parameters despite the stated no-user-facing-change goal.
Could the marker retain enough delegate/argument context (or have validation
routed through `DelegateExpression`) so the error remains attributed to `right`
and argument 2? A regression asserting the
`DATATYPE_MISMATCH.UNEXPECTED_INPUT_TYPE` message parameters for an uncastable
`right` argument would cover this.
_[ :robot: posted by Codex on behalf of sunchao using the code-review-for-me
skill :robot: ]_
--
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]
