You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2025/02/19 12:58:21 UTC
(spark) branch branch-4.0 updated: [SPARK-51247][SQL] Move SubstituteExecuteImmediate to 'resolution' batch and prepare it for SQL Scripting local variables
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new d0a7db4dc606 [SPARK-51247][SQL] Move SubstituteExecuteImmediate to 'resolution' batch and prepare it for SQL Scripting local variables
d0a7db4dc606 is described below
commit d0a7db4dc606e168d13b4c69db34abd85c52008e
Author: Dušan Tišma <du...@databricks.com>
AuthorDate: Wed Feb 19 20:57:49 2025 +0800
[SPARK-51247][SQL] Move SubstituteExecuteImmediate to 'resolution' batch and prepare it for SQL Scripting local variables
### What changes were proposed in this pull request?
This PR changes `SubstituteExecuteImmediate` to analyze it's entire subtree within a scoped context. This will allow us to disable SQL scripting local variables in the subtree, when they are added, which is necessary in order to sandbox the generated plan.
This PR also moves `SubstituteExecuteImmediate` to `resolution` batch in the analyzer. This is necessary in order to resolve arguments of EXECUTE IMMEDIATE properly, notably if the EXECUTE IMMEDIATE is the child of a `ParameterizedQuery`. This ensured proper resolution ordering i.e. first all parameters of EXECUTE IMMEDIATE will be resolved, and only then will the generated query itself be analyzed.
Local variables PR - https://github.com/apache/spark/pull/49445
### Why are the changes needed?
They are necessaty for local variables support in SQL scripting.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing unit tests and golden files.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #49993 from dusantism-db/execute-immediate-resolution-batch.
Authored-by: Dušan Tišma <du...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit 26febf763c1bc386a719c819c0bdfbee0abd948b)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../spark/sql/catalyst/analysis/Analyzer.scala | 23 +++++++++++-
.../sql/catalyst/analysis/executeImmediate.scala | 43 +++++++++++++---------
2 files changed, 47 insertions(+), 19 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index 7f59e0f47373..e0321dbb1967 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -137,6 +137,9 @@ object FakeV2SessionCatalog extends TableCatalog with FunctionCatalog with Suppo
* if `t` was a permanent table when the current view was created, it
* should still be a permanent table when resolving the current view,
* even if a temp view `t` has been created.
+ * @param isExecuteImmediate Whether the current plan is created by EXECUTE IMMEDIATE. Used when
+ * resolving variables, as SQL Scripting local variables should not be
+ * visible from EXECUTE IMMEDIATE.
* @param outerPlan The query plan from the outer query that can be used to resolve star
* expressions in a subquery.
*/
@@ -154,6 +157,7 @@ case class AnalysisContext(
referredTempFunctionNames: mutable.Set[String] = mutable.Set.empty,
referredTempVariableNames: Seq[Seq[String]] = Seq.empty,
outerPlan: Option[LogicalPlan] = None,
+ isExecuteImmediate: Boolean = false,
/**
* This is a bridge state between this fixed-point [[Analyzer]] and a single-pass [[Resolver]].
@@ -208,7 +212,16 @@ object AnalysisContext {
originContext.relationCache,
viewDesc.viewReferredTempViewNames,
mutable.Set(viewDesc.viewReferredTempFunctionNames: _*),
- viewDesc.viewReferredTempVariableNames)
+ viewDesc.viewReferredTempVariableNames,
+ isExecuteImmediate = originContext.isExecuteImmediate)
+ set(context)
+ try f finally { set(originContext) }
+ }
+
+ def withExecuteImmediateContext[A](f: => A): A = {
+ val originContext = value.get()
+ val context = originContext.copy(isExecuteImmediate = true)
+
set(context)
try f finally { set(originContext) }
}
@@ -325,7 +338,6 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
override def batches: Seq[Batch] = Seq(
Batch("Substitution", fixedPoint,
- new SubstituteExecuteImmediate(catalogManager),
// This rule optimizes `UpdateFields` expression chains so looks more like optimization rule.
// However, when manipulating deeply nested schema, `UpdateFields` expression tree could be
// very complex and make analysis impossible. Thus we need to optimize `UpdateFields` early
@@ -401,6 +413,10 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
ResolveRowLevelCommandAssignments ::
MoveParameterizedQueriesDown ::
BindParameters ::
+ new SubstituteExecuteImmediate(
+ catalogManager,
+ resolveChild = executeSameContext,
+ checkAnalysis = checkAnalysis) ::
typeCoercionRules() ++
Seq(
ResolveWithCTE,
@@ -1670,6 +1686,9 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
case s: Sort if !s.resolved || s.missingInput.nonEmpty =>
resolveReferencesInSort(s)
+ // Pass for Execute Immediate as arguments will be resolved by [[SubstituteExecuteImmediate]].
+ case e : ExecuteImmediateQuery => e
+
case q: LogicalPlan =>
logTrace(s"Attempting to resolve ${q.simpleString(conf.maxToStringFields)}")
q.mapExpressions(resolveExpressionByPlanChildren(_, q, includeLastResort = true))
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/executeImmediate.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/executeImmediate.scala
index b9e20f59bb0e..2b7eed0dc795 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/executeImmediate.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/executeImmediate.scala
@@ -44,12 +44,14 @@ case class ExecuteImmediateQuery(
}
/**
- * This rule substitutes execute immediate query node with plan that is passed as string literal
- * or session parameter.
+ * This rule substitutes execute immediate query node with fully analyzed
+ * plan that is passed as string literal or session parameter.
*/
-class SubstituteExecuteImmediate(val catalogManager: CatalogManager)
- extends Rule[LogicalPlan]
- with ColumnResolutionHelper {
+class SubstituteExecuteImmediate(
+ val catalogManager: CatalogManager,
+ resolveChild: LogicalPlan => LogicalPlan,
+ checkAnalysis: LogicalPlan => Unit)
+ extends Rule[LogicalPlan] with ColumnResolutionHelper {
def resolveVariable(e: Expression): Expression = {
@@ -106,7 +108,12 @@ class SubstituteExecuteImmediate(val catalogManager: CatalogManager)
override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsWithPruning(_.containsPattern(EXECUTE_IMMEDIATE), ruleId) {
- case ExecuteImmediateQuery(expressions, query, targetVariables) =>
+ case e @ ExecuteImmediateQuery(expressions, _, _) if expressions.exists(!_.resolved) =>
+ e.copy(args = resolveArguments(expressions))
+
+ case ExecuteImmediateQuery(expressions, query, targetVariables)
+ if expressions.forall(_.resolved) =>
+
val queryString = extractQueryString(query)
val plan = parseStatement(queryString, targetVariables)
@@ -123,21 +130,16 @@ class SubstituteExecuteImmediate(val catalogManager: CatalogManager)
throw QueryCompilationErrors.invalidQueryMixedQueryParameters()
} else {
if (posNodes.nonEmpty) {
- PosParameterizedQuery(
- plan,
- // We need to resolve arguments before Resolution batch to make sure
- // that some rule does not accidentally resolve our parameters.
- // We do not want this as they can resolve some unsupported parameters
- resolveArguments(expressions))
+ PosParameterizedQuery(plan, expressions)
} else {
val aliases = expressions.collect {
case e: Alias => e
- case u: UnresolvedAttribute => Alias(u, u.nameParts.last)()
+ case u: VariableReference => Alias(u, u.identifier.name())()
}
if (aliases.size != expressions.size) {
val nonAliases = expressions.filter(attr =>
- !attr.isInstanceOf[Alias] && !attr.isInstanceOf[UnresolvedAttribute])
+ !attr.isInstanceOf[Alias] && !attr.isInstanceOf[VariableReference])
throw QueryCompilationErrors.invalidQueryAllParametersMustBeNamed(nonAliases)
}
@@ -148,13 +150,20 @@ class SubstituteExecuteImmediate(val catalogManager: CatalogManager)
// We need to resolve arguments before Resolution batch to make sure
// that some rule does not accidentally resolve our parameters.
// We do not want this as they can resolve some unsupported parameters.
- resolveArguments(aliases))
+ aliases)
}
}
+ // Fully analyze the generated plan. AnalysisContext.withExecuteImmediateContext makes sure
+ // that SQL scripting local variables will not be accessed from the plan.
+ val finalPlan = AnalysisContext.withExecuteImmediateContext {
+ resolveChild(queryPlan)
+ }
+ checkAnalysis(finalPlan)
+
if (targetVariables.nonEmpty) {
- SetVariable(targetVariables, queryPlan)
- } else { queryPlan }
+ SetVariable(targetVariables, finalPlan)
+ } else { finalPlan }
}
private def parseStatement(
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org