This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new fd194c2d7dd7 [SPARK-60614][SQL] Share scalar SQL variable assignment
helpers
fd194c2d7dd7 is described below
commit fd194c2d7dd7d2c4c010b09f2f1e30733352b6dd
Author: Vladan Vasić <[email protected]>
AuthorDate: Thu Oct 29 11:11:27 2026 +0000
[SPARK-60614][SQL] Share scalar SQL variable assignment helpers
### What changes were proposed in this pull request?
Addresses [SPARK-60614](https://issues.apache.org/jira/browse/SPARK-60614).
- Add a `TableOutputResolver.resolveVariableOutputColumns` overload
accepting resolved output attributes and an alias factory. Retain the existing
three-argument entry point.
- Extract duplicate-target validation into `VariableAssignmentUtils` and
reuse it from `ResolveSetVariable` and `ResolveFetchCursor`.
- Centralize the existing `DUPLICATE_ASSIGNMENTS` exception in
`QueryCompilationErrors.duplicateAssignmentsError`.
### Why are the changes needed?
`SET VARIABLE` and `FETCH INTO` currently duplicate target validation and
error construction. Sharing these helpers keeps their behavior consistent.
The output-resolution overload lets callers reuse assignment casts without
recursively accessing `LogicalPlan.output`, while allowing the single-pass
resolver to track generated aliases through its existing alias provider. This
PR only prepares the shared helpers; it does not add single-pass `SET VARIABLE`
support.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- Catalyst production compilation (`build/sbt -batch -no-colors
catalyst/compile`) and its dependencies passed with SBT 1.13.0, JDK 17, and
Scala 2.13. The local build used the installed SBT launcher and cached Maven
mirror.
- `git diff --check` passed.
- No new tests were added for the behavior-preserving extraction. Unit and
integration tests have not been run; the new explicit-output/custom-alias
overload does not yet have dedicated test coverage.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex CLI 0.164.0
Closes #59861 from vladanvasi-db/vladanvasi-db/variable-assignment-utils.
Authored-by: Vladan Vasić <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
---
.../sql/catalyst/analysis/ResolveFetchCursor.scala | 18 +---------
.../sql/catalyst/analysis/ResolveSetVariable.scala | 23 +++----------
.../catalyst/analysis/TableOutputResolver.scala | 39 ++++++++++++++++------
.../analysis/VariableAssignmentUtils.scala | 34 +++++++++++++++++++
.../spark/sql/errors/QueryCompilationErrors.scala | 6 ++++
5 files changed, 74 insertions(+), 46 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveFetchCursor.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveFetchCursor.scala
index 34942fcf08bc..a710a1bf3def 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveFetchCursor.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveFetchCursor.scala
@@ -18,7 +18,6 @@
package org.apache.spark.sql.catalyst.analysis
import org.apache.spark.SparkException
-import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.expressions.{Expression,
VariableReference}
import org.apache.spark.sql.catalyst.plans.logical.{FetchCursor, LogicalPlan,
SingleStatement}
import org.apache.spark.sql.catalyst.rules.Rule
@@ -37,21 +36,6 @@ class ResolveFetchCursor(val catalogManager: CatalogManager)
extends Rule[Logica
private val variableResolution =
new VariableResolution(catalogManager.tempVariableManager, catalogManager)
- /**
- * Checks for duplicate variable names and throws an exception if found.
- * Names are normalized when the variables are created.
- * No need for case insensitive comparison here.
- */
- private def checkForDuplicateVariables(variables: Seq[VariableReference]):
Unit = {
- val dups = variables.groupBy(_.identifier).filter(kv => kv._2.length > 1)
- if (dups.nonEmpty) {
- throw new AnalysisException(
- errorClass = "DUPLICATE_ASSIGNMENTS",
- messageParameters = Map("nameList" ->
- dups.keys.map(key => toSQLId(key.name())).mkString(", ")))
- }
- }
-
/**
* Resolves and validates the target variables for a FetchCursor command.
* Returns a sequence of resolved VariableReference expressions.
@@ -73,7 +57,7 @@ class ResolveFetchCursor(val catalogManager: CatalogManager)
extends Rule[Logica
}
// Check for duplicates immediately after resolution
- checkForDuplicateVariables(resolvedVars)
+ VariableAssignmentUtils.checkForDuplicateVariables(resolvedVars)
resolvedVars
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSetVariable.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSetVariable.scala
index ab80fc829cf4..22c4281d2837 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSetVariable.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSetVariable.scala
@@ -18,7 +18,6 @@
package org.apache.spark.sql.catalyst.analysis
import org.apache.spark.SparkException
-import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.logical.{Limit, LogicalPlan,
SetVariable}
import org.apache.spark.sql.catalyst.rules.Rule
@@ -36,22 +35,6 @@ class ResolveSetVariable(val catalogManager: CatalogManager)
extends Rule[Logica
private val variableResolution =
new VariableResolution(catalogManager.tempVariableManager, catalogManager)
- /**
- * Checks for duplicate variable names and throws an exception if found.
- * Names are normalized when the variables are created.
- * No need for case insensitive comparison here.
- */
- private def checkForDuplicateVariables(variables: Seq[VariableReference]):
Unit = {
- // TODO: we need to group by the qualified variable name once other
catalogs support it.
- val dups = variables.groupBy(_.identifier).filter(kv => kv._2.length > 1)
- if (dups.nonEmpty) {
- throw new AnalysisException(
- errorClass = "DUPLICATE_ASSIGNMENTS",
- messageParameters = Map("nameList" ->
- dups.keys.map(key => toSQLId(key.name())).mkString(", ")))
- }
- }
-
override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsWithPruning(
_.containsPattern(COMMAND), ruleId) {
// Resolve the left hand side of the SET VAR command
@@ -82,10 +65,12 @@ class ResolveSetVariable(val catalogManager:
CatalogManager) extends Rule[Logica
// Check for duplicate variable names - this handles both regular SET
VAR (after resolution)
// and EXECUTE IMMEDIATE ... INTO (which comes pre-resolved)
- checkForDuplicateVariables(targetVariables)
+ VariableAssignmentUtils.checkForDuplicateVariables(targetVariables)
val withCasts = TableOutputResolver.resolveVariableOutputColumns(
- targetVariables, setVariable.sourceQuery, conf)
+ expected = targetVariables,
+ query = setVariable.sourceQuery,
+ conf = conf)
val withLimit = if (withCasts.maxRows.exists(_ <= 2)) {
withCasts
} else {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala
index 719e32b02c34..62a67c51681a 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala
@@ -56,33 +56,52 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
def resolveVariableOutputColumns(
expected: Seq[VariableReference],
query: LogicalPlan,
- conf: SQLConf): LogicalPlan = {
+ conf: SQLConf): LogicalPlan =
+ resolveVariableOutputColumns(
+ expected = expected,
+ query = query,
+ conf = conf,
+ outputColumns = query.output,
+ newAlias = (child, name) => Alias(child, name)())
- if (expected.size != query.output.size) {
+ /**
+ * Resolves variable output against explicitly supplied attributes, avoiding
recursive
+ * [[LogicalPlan.output]] access.
+ *
+ * @param outputColumns The resolved output attributes of `query`, in output
order.
+ * @param newAlias Factory for aliases wrapping assignment casts.
+ */
+ def resolveVariableOutputColumns(
+ expected: Seq[VariableReference],
+ query: LogicalPlan,
+ conf: SQLConf,
+ outputColumns: Seq[Attribute],
+ newAlias: (Expression, String) => Alias): LogicalPlan = {
+ if (expected.size != outputColumns.size) {
throw new AnalysisException(
errorClass = "ASSIGNMENT_ARITY_MISMATCH",
messageParameters = Map(
"numTarget" -> expected.size.toString,
- "numExpr" -> query.output.size.toString))
+ "numExpr" -> outputColumns.size.toString))
}
val resolved: Seq[NamedExpression] = {
- query.output.zip(expected).map { case (inputCol, expected) =>
- if (DataTypeUtils.sameType(inputCol.dataType, expected.dataType)) {
- inputCol
+ outputColumns.zip(expected).map { case (inputColumn, expectedVariable) =>
+ if (DataTypeUtils.sameType(inputColumn.dataType,
expectedVariable.dataType)) {
+ inputColumn
} else {
// SET VAR always uses the ANSI store assignment policy
val cast = Cast(
- inputCol,
- expected.dataType,
+ inputColumn,
+ expectedVariable.dataType,
Option(conf.sessionLocalTimeZone),
ansiEnabled = true)
- Alias(cast, expected.identifier.name)()
+ newAlias(cast, expectedVariable.identifier.name)
}
}
}
- if (resolved == query.output) {
+ if (resolved == outputColumns) {
query
} else {
Project(resolved, query)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/VariableAssignmentUtils.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/VariableAssignmentUtils.scala
new file mode 100644
index 000000000000..ec4e4dd7af90
--- /dev/null
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/VariableAssignmentUtils.scala
@@ -0,0 +1,34 @@
+/*
+ * 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.analysis
+
+import org.apache.spark.sql.catalyst.expressions.Cast.toSQLId
+import org.apache.spark.sql.catalyst.expressions.VariableReference
+import
org.apache.spark.sql.errors.QueryCompilationErrors.duplicateAssignmentsError
+
+/** Validation for scalar SQL variable assignments. */
+object VariableAssignmentUtils {
+ /** Rejects duplicate variable identifiers, whose names are normalized at
creation. */
+ def checkForDuplicateVariables(variables: Seq[VariableReference]): Unit = {
+ val duplicates = variables.groupBy(_.identifier).filter(entry =>
entry._2.length > 1)
+ if (duplicates.nonEmpty) {
+ throw duplicateAssignmentsError(
+ duplicates.keys.map(identifier =>
toSQLId(identifier.name())).mkString(", "))
+ }
+ }
+}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index 5a51c0a646ec..e869ef819541 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -1038,6 +1038,12 @@ private[sql] object QueryCompilationErrors extends
QueryErrorsBase with Compilat
messageParameters = Map("dt" -> dt.toString))
}
+ def duplicateAssignmentsError(nameList: String): Throwable = {
+ new AnalysisException(
+ errorClass = "DUPLICATE_ASSIGNMENTS",
+ messageParameters = Map("nameList" -> nameList))
+ }
+
def unresolvedVariableError(
name: Seq[String],
pathEntries: Seq[Seq[String]],
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
