[SPARK-53444] Rework execute immediate

common/utils/src/main/resources/error/error-conditions.json

@@ -2737,6 +2737,12 @@
     ],
     "sqlState" : "42001"
   },
+  "INVALID_EXPR_TYPE_FOR_QUERY_EXECUTE_IMMEDIATE" : {
+    "message" : [
+      "Expression type must be string type but got <exprType>."
+    ],
+    "sqlState" : "42K09"
+  },
   "INVALID_EXTERNAL_TYPE" : {
     "message" : [
       "The external type <externalType> is not valid for the type <type> at the expression <expr>."
@@ -3914,12 +3920,6 @@
     },
     "sqlState" : "42K0M"
   },
-  "INVALID_VARIABLE_TYPE_FOR_QUERY_EXECUTE_IMMEDIATE" : {
-    "message" : [
-      "Variable type must be string type but got <varType>."
-    ],
-    "sqlState" : "42K09"
-  },
   "INVALID_VARIANT_CAST" : {
     "message" : [
       "The variant value `<value>` cannot be cast into `<dataType>`. Please use `try_variant_get` instead."

sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4

@@ -397,27 +397,14 @@ setResetStatement
     ;
 
 executeImmediate
-    : EXECUTE IMMEDIATE queryParam=executeImmediateQueryParam (INTO targetVariable=multipartIdentifierList)? executeImmediateUsing?
+    : EXECUTE IMMEDIATE queryParam=expression (INTO targetVariable=multipartIdentifierList)? executeImmediateUsing?
     ;
 
 executeImmediateUsing
     : USING LEFT_PAREN params=namedExpressionSeq RIGHT_PAREN
     | USING params=namedExpressionSeq
     ;
 
-executeImmediateQueryParam
-    : stringLit
-    | multipartIdentifier
-    ;
-
-executeImmediateArgument
-    : (constant|multipartIdentifier) (AS name=errorCapturingIdentifier)?
-    ;
-
-executeImmediateArgumentSeq
-    : executeImmediateArgument (COMMA executeImmediateArgument)*
-    ;
-
 timezone
     : stringLit
     | LOCAL

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -135,9 +135,6 @@ 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.
  */
@@ -155,7 +152,6 @@ case class AnalysisContext(
     referredTempFunctionNames: mutable.Set[String] = mutable.Set.empty,
     referredTempVariableNames: Seq[Seq[String]] = Seq.empty,
     outerPlan: Option[LogicalPlan] = None,
-    isExecuteImmediate: Boolean = false,
     collation: Option[String] = None,
 
     /**
@@ -212,19 +208,11 @@ object AnalysisContext {
       viewDesc.viewReferredTempViewNames,
       mutable.Set(viewDesc.viewReferredTempFunctionNames: _*),
       viewDesc.viewReferredTempVariableNames,
-      isExecuteImmediate = originContext.isExecuteImmediate,
       collation = viewDesc.collation)
     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) }
-  }
 
   def withNewAnalysisContext[A](f: => A): A = {
     val originContext = value.get()
@@ -444,6 +432,7 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
       AddMetadataColumns ::
       DeduplicateRelations ::
       ResolveCollationName ::
+      new ResolveExecuteImmediate(catalogManager)::
       ResolveMergeIntoSchemaEvolution ::
       new ResolveReferences(catalogManager) ::
       // Please do not insert any other rules in between. See the TODO comments in rule
@@ -495,10 +484,6 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
       RewriteMergeIntoTable ::
       MoveParameterizedQueriesDown ::
       BindParameters ::
-      new SubstituteExecuteImmediate(
-        catalogManager,
-        resolveChild = executeSameContext,
-        checkAnalysis = checkAnalysis) ::
       typeCoercionRules() ++
       Seq(
         ResolveWithCTE,
@@ -1802,8 +1787,6 @@ 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 d: DataFrameDropColumns if !d.resolved =>
         resolveDataFrameDropColumns(d)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala

@@ -241,7 +241,6 @@ trait ColumnResolutionHelper extends Logging with DataTypeErrorsBase {
       variableResolution.resolveMultipartName(
         nameParts = nameParts,
         resolvingView = AnalysisContext.get.catalogAndNamespace.nonEmpty,
-        resolvingExecuteImmediate = AnalysisContext.get.isExecuteImmediate,
         referredTempVariableNames = AnalysisContext.get.referredTempVariableNames
       ).map(e => Alias(e, nameParts.last)())
     }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala

@@ -142,8 +142,7 @@ class ResolveCatalogs(val catalogManager: CatalogManager)
   }
 
   private def withinSqlScript: Boolean =
-    SqlScriptingContextManager.get().map(_.getVariableManager).isDefined &&
-      !AnalysisContext.get.isExecuteImmediate
+    SqlScriptingContextManager.get().map(_.getVariableManager).isDefined
 
   private def assertValidSessionVariableNameParts(
       nameParts: Seq[String],

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSetVariable.scala

@@ -42,8 +42,7 @@ class ResolveSetVariable(val catalogManager: CatalogManager) extends Rule[Logica
       val resolvedVars = setVariable.targetVariables.map {
         case u: UnresolvedAttribute =>
           variableResolution.lookupVariable(
-            nameParts = u.nameParts,
-            resolvingExecuteImmediate = AnalysisContext.get.isExecuteImmediate
+            nameParts = u.nameParts
           ) match {
             case Some(variable) => variable.copy(canFold = false)
             case _ => throw unresolvedVariableError(u.nameParts, Seq("SYSTEM", "SESSION"))

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/VariableResolution.scala

@@ -52,9 +52,6 @@ class VariableResolution(tempVariableManager: TempVariableManager) extends SQLCo
    *   (e.g., ["catalog", "schema", "variable", "field1", "field2"])
    * @param resolvingView Whether this resolution is happening within a view context.
    *   When true, only variables explicitly referred to in the view definition are accessible.
-   * @param resolvingExecuteImmediate Whether this resolution is happening within an
-   *   EXECUTE IMMEDIATE context. When true, local variables are not accessible, only session
-   *   variables.
    * @param referredTempVariableNames When resolving within a view, this contains the list of
    *   variable names that the view explicitly references and should have access to.
    *
@@ -65,7 +62,6 @@ class VariableResolution(tempVariableManager: TempVariableManager) extends SQLCo
   def resolveMultipartName(
       nameParts: Seq[String],
       resolvingView: Boolean,
-      resolvingExecuteImmediate: Boolean,
       referredTempVariableNames: Seq[Seq[String]]): Option[Expression] = {
     var resolvedVariable: Option[Expression] = None
     // We only support temp variables for now, so the variable name can at most have 3 parts.
@@ -76,7 +72,6 @@ class VariableResolution(tempVariableManager: TempVariableManager) extends SQLCo
       resolvedVariable = resolveVariable(
         nameParts = nameParts.dropRight(numInnerFields),
         resolvingView = resolvingView,
-        resolvingExecuteImmediate = resolvingExecuteImmediate,
         referredTempVariableNames = referredTempVariableNames
       )
 
@@ -99,16 +94,12 @@ class VariableResolution(tempVariableManager: TempVariableManager) extends SQLCo
 
   /**
    * Look up variable by nameParts.
-   * If in SQL Script, first check local variables, unless in EXECUTE IMMEDIATE
-   * (EXECUTE IMMEDIATE generated query cannot access local variables).
-   * if not found fall back to session variables.
+   * If in SQL Script, first check local variables.
+   * If not found fall back to session variables.
    * @param nameParts NameParts of the variable.
-   * @param resolvingExecuteImmediate Whether the current context is in EXECUTE IMMEDIATE.
    * @return Reference to the variable.
    */
-  def lookupVariable(
-      nameParts: Seq[String],
-      resolvingExecuteImmediate: Boolean): Option[VariableReference] = {
+  def lookupVariable(nameParts: Seq[String]): Option[VariableReference] = {
     val namePartsCaseAdjusted = if (conf.caseSensitiveAnalysis) {
       nameParts
     } else {
@@ -118,8 +109,6 @@ class VariableResolution(tempVariableManager: TempVariableManager) extends SQLCo
     SqlScriptingContextManager
       .get()
       .map(_.getVariableManager)
-      // If we are in EXECUTE IMMEDIATE lookup only session variables.
-      .filterNot(_ => resolvingExecuteImmediate)
       // If variable name is qualified with session.<varName> treat it as a session variable.
       .filterNot(
         _ =>
@@ -156,16 +145,15 @@ class VariableResolution(tempVariableManager: TempVariableManager) extends SQLCo
   private def resolveVariable(
       nameParts: Seq[String],
       resolvingView: Boolean,
-      resolvingExecuteImmediate: Boolean,
       referredTempVariableNames: Seq[Seq[String]]): Option[Expression] = {
     if (resolvingView) {
       if (referredTempVariableNames.contains(nameParts)) {
-        lookupVariable(nameParts = nameParts, resolvingExecuteImmediate = resolvingExecuteImmediate)
+        lookupVariable(nameParts = nameParts)
       } else {
         None
       }
     } else {
-      lookupVariable(nameParts = nameParts, resolvingExecuteImmediate = resolvingExecuteImmediate)
+      lookupVariable(nameParts = nameParts)
     }
   }