(spark) branch master updated: [SPARK-57369][SQL] Move main EXECUTE IMMEDIATE resolution logic to common code

[we...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

cloud-fan 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 a1922b5e2b15 [SPARK-57369][SQL] Move main EXECUTE IMMEDIATE resolution logic to common code
a1922b5e2b15 is described below

commit a1922b5e2b1507206a4ecbcf65bb62870a3934b3
Author: Mihailo Aleksic <mi...@databricks.com>
AuthorDate: Thu Jun 11 13:51:00 2026 -0700

    [SPARK-57369][SQL] Move main EXECUTE IMMEDIATE resolution logic to common code
    
    ### What changes were proposed in this pull request?
    In this PR I propose to move main EXECUTE IMMEDIATE resolution logic to common code in order to ease the single-pass implementation.
    
    ### Why are the changes needed?
    In order to ease the single-pass implementation.
    
    ### Does this PR introduce _any_ user-facing change?
    No, it's just a refactor.
    
    ### How was this patch tested?
    Existing tests.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    Yes.
    
    Closes #56429 from mihailoale-db/execimmrefactor.
    
    Authored-by: Mihailo Aleksic <mi...@databricks.com>
    Signed-off-by: Wenchen Fan <we...@databricks.com>
---
 .../analysis/ResolveExecuteImmediate.scala         | 50 ++++++++++++++++------
 1 file changed, 38 insertions(+), 12 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveExecuteImmediate.scala b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveExecuteImmediate.scala
index bf57d8df5229..c09498b41c97 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveExecuteImmediate.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveExecuteImmediate.scala
@@ -35,6 +35,9 @@ import org.apache.spark.sql.types.StringType
  * Analysis rule that resolves and executes EXECUTE IMMEDIATE statements during analysis,
  * replacing them with the results, similar to how CALL statements work.
  * This rule combines resolution and execution in a single pass.
+ *
+ * When sub-expressions are not yet resolved, the node is returned unchanged so that the
+ * fixed-point analyzer re-applies this rule on the next iteration.
  */
 case class ResolveExecuteImmediate(sparkSession: SparkSession, catalogManager: CatalogManager)
   extends Rule[LogicalPlan] {
@@ -43,23 +46,45 @@ case class ResolveExecuteImmediate(sparkSession: SparkSession, catalogManager: C
     plan.resolveOperatorsWithPruning(_.containsPattern(EXECUTE_IMMEDIATE), ruleId) {
       case node @ UnresolvedExecuteImmediate(sqlStmtStr, args, targetVariables) =>
         if (sqlStmtStr.resolved && targetVariables.forall(_.resolved) && args.forall(_.resolved)) {
-          // All resolved - execute immediately and handle INTO clause if present
-          if (targetVariables.nonEmpty) {
-            // EXECUTE IMMEDIATE ... INTO should generate SetVariable plan with eagerly executed
-            // source
-            val finalTargetVars = extractTargetVariables(targetVariables)
-            val executedSource = executeImmediateQuery(sqlStmtStr, args, hasIntoClause = true)
-            SetVariable(finalTargetVars, executedSource)
-          } else {
-            // Regular EXECUTE IMMEDIATE without INTO - execute and return result directly
-            executeImmediateQuery(sqlStmtStr, args, hasIntoClause = false)
-          }
+          ResolveExecuteImmediate.resolveExecuteImmediate(
+            sparkSession, sqlStmtStr, args, targetVariables)
         } else {
-          // Not all resolved yet - wait for next iteration
           node
         }
     }
   }
+}
+
+/**
+ * Companion object containing shared resolution logic for `EXECUTE IMMEDIATE` statements.
+ */
+object ResolveExecuteImmediate {
+
+  /**
+   * Resolves an [[UnresolvedExecuteImmediate]] node into an executable plan.
+   *
+   * All expressions (`sqlStmtStr`, `args`, `targetVariables`) must already be resolved
+   * before calling this method.
+   *
+   * When an `INTO` clause is present, the dynamic SQL is eagerly parsed and analyzed, and
+   * the analyzed plan is wrapped in a [[SetVariable]] plan that assigns output columns to
+   * the target variables. Without `INTO`, the resulting plan is returned directly
+   * (commands are executed eagerly; queries are returned analyzed but unexecuted).
+   */
+  def resolveExecuteImmediate(
+      sparkSession: SparkSession,
+      sqlStmtStr: Expression,
+      args: Seq[Expression],
+      targetVariables: Seq[Expression]): LogicalPlan = {
+    if (targetVariables.nonEmpty) {
+      val finalTargetVars = extractTargetVariables(targetVariables)
+      val executedSource = executeImmediateQuery(
+        sparkSession, sqlStmtStr, args, hasIntoClause = true)
+      SetVariable(finalTargetVars, executedSource)
+    } else {
+      executeImmediateQuery(sparkSession, sqlStmtStr, args, hasIntoClause = false)
+    }
+  }
 
   private def extractTargetVariables(targetVariables: Seq[Expression]): Seq[VariableReference] = {
     targetVariables.map {
@@ -81,6 +106,7 @@ case class ResolveExecuteImmediate(sparkSession: SparkSession, catalogManager: C
   }
 
   private def executeImmediateQuery(
+      sparkSession: SparkSession,
       sqlStmtStr: Expression,
       args: Seq[Expression],
       hasIntoClause: Boolean): LogicalPlan = {


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org