Skip to content

[SPARK-59689][SQL] Run EXECUTE IMMEDIATE commands at execution time #58944

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
matvei-zamiatin-db wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[SPARK-59689][SQL] Run EXECUTE IMMEDIATE commands at execution time #58944

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 13 additions & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{DataTypeMismatch,
import org.apache.spark.sql.catalyst.catalog.{FunctionResource, RoutineLanguage}
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.DescribeCommandSchema
import org.apache.spark.sql.catalyst.plans.{DescribeCommandSchema, QueryPlan}
import org.apache.spark.sql.catalyst.trees.BinaryLike
import org.apache.spark.sql.catalyst.trees.TreePattern.{DELETE_FROM_TABLE, MERGE_INTO_TABLE, REPLACE_DATA, TreePattern, UPDATE_TABLE, WRITE_DELTA}
import org.apache.spark.sql.catalyst.types.DataTypeUtils
Expand Down Expand Up @@ -2247,6 +2247,18 @@ case class SetVariable(
copy(sourceQuery = newChild)
}

/**
* The logical plan of an EXECUTE IMMEDIATE command payload. It supervises the already-analyzed
* inner command in a non-child slot; it does not execute it. Keeping the payload out of the
* children keeps it off the eager-command path and gives EXPLAIN a stable node. Execution happens
* only when this node is planned to `ExecuteImmediateExec`, the sole executor of the payload. The
* payload is surfaced via [[innerChildren]] so EXPLAIN still shows it.
*/
case class ExecuteImmediateCommand(sourceStatement: LogicalPlan) extends LeafCommand {
override def output: Seq[Attribute] = sourceStatement.output
override def innerChildren: Seq[QueryPlan[_]] = Seq(sourceStatement)
}

/**
* The logical plan of the DECLARE CURSOR statement.
*
Expand Down
35 changes: 35 additions & 0 deletions ...server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectServiceSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -881,6 +881,41 @@ class SparkConnectServiceSuite
}
}

test("SPARK-59689: AnalyzePlanRequest does not execute an EXECUTE IMMEDIATE command payload") {
withTable("ei_analyze") {
spark.sql("CREATE TABLE ei_analyze (col1 INT, col2 STRING)")
// The command payload is deferred to the execution level, so analyzing the EXECUTE IMMEDIATE
// via the Connect analyze path (CommandExecutionMode.SKIP) must not run the DROP.
val sqlString = "EXECUTE IMMEDIATE 'DROP TABLE ei_analyze'"
val plan = proto.Plan
.newBuilder()
.setRoot(
proto.Relation
.newBuilder()
.setCommon(proto.RelationCommon.newBuilder().setPlanId(1))
.setSql(proto.SQL.newBuilder().setQuery(sqlString).build())
.build())
.build()

val handler = new SparkConnectAnalyzeHandler(null)

val request = proto.AnalyzePlanRequest
.newBuilder()
.setExplain(
proto.AnalyzePlanRequest.Explain
.newBuilder()
.setPlan(plan)
.setExplainMode(proto.AnalyzePlanRequest.Explain.ExplainMode.EXPLAIN_MODE_EXTENDED)
.build())
.build()

handler.process(request, sparkSessionHolder)

assert(spark.catalog.tableExists("ei_analyze"),
"EXECUTE IMMEDIATE command payload must not run during a Connect analyze request")
}
}

test("Test explain mode in analyze response") {
withTable("test") {
spark.sql("""
Expand Down
126 changes: 76 additions & 50 deletions sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveExecuteImmediate.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,23 +18,33 @@
package org.apache.spark.sql.catalyst.analysis

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.QueryPlanningTracker
import org.apache.spark.sql.catalyst.SqlScriptingContextManager
import org.apache.spark.sql.catalyst.catalog.{SqlScriptingContextManager => SqlScriptingContextManagerTrait}
import org.apache.spark.sql.catalyst.expressions.{Alias, Expression, VariableReference}
import org.apache.spark.sql.catalyst.plans.logical.{Command, CompoundBody, LogicalPlan, SetVariable}
import org.apache.spark.sql.catalyst.plans.logical.{CompoundBody, ExecuteImmediateCommand, LogicalPlan, SetVariable}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin}
import org.apache.spark.sql.catalyst.trees.TreePattern.EXECUTE_IMMEDIATE
import org.apache.spark.sql.classic.{SparkSession => ClassicSparkSession}
import org.apache.spark.sql.connector.catalog.CatalogManager
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.execution.QueryExecution
import org.apache.spark.sql.execution.command.v2.ParameterBindingUtils
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.
* Analysis rule that resolves EXECUTE IMMEDIATE statements during analysis, parsing and analyzing
* the dynamic SQL and replacing the node with the resolved inner plan: a query is spliced, a
* command payload is wrapped in [[ExecuteImmediateCommand]] to run at the execution level, and an
* INTO clause becomes a [[SetVariable]]. Command payloads are not executed during analysis; nodes
* that execute during analysis, such as CALL, are an exception (see `resolveInnerStatement`).
*
* {{{
* EXECUTE IMMEDIATE 'INSERT INTO t VALUES (?)' USING 1 => ExecuteImmediateCommand(...)
* EXECUTE IMMEDIATE 'SELECT ?' USING 1 => analyzed query, spliced and run lazily
* EXECUTE IMMEDIATE 'SELECT 1' INTO v => SetVariable over the analyzed query
* }}}
*
* 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.
Expand All @@ -61,15 +71,16 @@ case class ResolveExecuteImmediate(sparkSession: SparkSession, catalogManager: C
object ResolveExecuteImmediate {

/**
* Resolves an [[UnresolvedExecuteImmediate]] node into an executable plan.
* Resolves an [[UnresolvedExecuteImmediate]] node into the plan for its dynamic SQL.
*
* 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).
* When an `INTO` clause is present, the dynamic SQL is parsed and analyzed, and the analyzed
* plan is wrapped in a [[SetVariable]] plan that assigns output columns to the target variables.
* Without `INTO`, a command payload is wrapped in [[ExecuteImmediateCommand]] to run at the
* execution level and a query is returned analyzed for lazy execution. Command payloads are not
* executed during analysis.
*/
def resolveExecuteImmediate(
sparkSession: SparkSession,
Expand All @@ -78,11 +89,11 @@ object ResolveExecuteImmediate {
targetVariables: Seq[Expression]): LogicalPlan = {
if (targetVariables.nonEmpty) {
val finalTargetVars = extractTargetVariables(targetVariables)
val executedSource = executeImmediateQuery(
val analyzedSource = resolveInnerStatement(
sparkSession, sqlStmtStr, args, hasIntoClause = true)
SetVariable(finalTargetVars, executedSource)
SetVariable(finalTargetVars, analyzedSource)
} else {
executeImmediateQuery(sparkSession, sqlStmtStr, args, hasIntoClause = false)
resolveInnerStatement(sparkSession, sqlStmtStr, args, hasIntoClause = false)
}
}

Expand All @@ -105,7 +116,15 @@ object ResolveExecuteImmediate {
}
}

private def executeImmediateQuery(
/**
* Parses and analyzes the dynamic SQL and returns the plan that replaces the EXECUTE IMMEDIATE
* node. No command payload is executed here: with an INTO clause the analyzed query is returned
* for the caller to wrap in [[SetVariable]] (a command is rejected); otherwise a command payload
* is wrapped in [[ExecuteImmediateCommand]] to run at the execution level and a query is returned
* analyzed for lazy execution. Nodes that execute during analysis, such as CALL, are still run by
* the analyzer here (see the branch below).
*/
private def resolveInnerStatement(
sparkSession: SparkSession,
sqlStmtStr: Expression,
args: Seq[Expression],
Expand All @@ -123,50 +142,57 @@ object ResolveExecuteImmediate {
stopIndex = Some(sqlString.length - 1)
)

// Execute the query with local variables hidden and EXECUTE IMMEDIATE origin set.
// Both must cover parsing, analysis, and execution phases.
// Parse and analyze the inner statement with local variables hidden and EXECUTE IMMEDIATE
// origin set, but without executing it (command payloads run later at the execution level).
// Both must cover parsing and analysis.
// CurrentOrigin.withOrigin ensures expressions created during parsing get the proper context.
val result = withHiddenLocalVariables {
val analyzed = withHiddenLocalVariables {
CurrentOrigin.withOrigin(executeImmediateOrigin) {
// Use shared parameterized query execution logic (same as OPEN CURSOR)
val df = if (args.isEmpty) {
// No parameters - execute directly
sparkSession.sql(sqlString)
} else {
// For parameterized queries, use shared parameter binding utility
val (paramValues, paramNames) = ParameterBindingUtils.buildUnifiedParameters(args)

sparkSession.asInstanceOf[ClassicSparkSession]
.sql(sqlString, paramValues, paramNames)
}

// SQL scripts (BEGIN/END blocks) are explicitly disallowed in EXECUTE IMMEDIATE.
// This is a design constraint: EXECUTE IMMEDIATE is for executing single SQL statements,
// and SQL scripts have their own variable scoping and control flow that would conflict
// with EXECUTE IMMEDIATE's parameter passing semantics.
if (df.queryExecution.logical.isInstanceOf[CompoundBody]) {
throw QueryCompilationErrors.sqlScriptInExecuteImmediate(sqlString)
}

// Force analysis to happen while local variables are hidden. This is critical because
// DataFrames are lazy and analysis would otherwise happen after withHiddenLocalVariables
// has restored the original context.
df.queryExecution.analyzed
df
parseAndAnalyzeInnerStatement(
sparkSession.asInstanceOf[ClassicSparkSession], sqlString, args)
}
}

// If this EXECUTE IMMEDIATE has an INTO clause, commands are not allowed
if (hasIntoClause && result.queryExecution.analyzed.isInstanceOf[Command]) {
throw QueryCompilationErrors.invalidStatementForExecuteInto(sqlString)
if (hasIntoClause) {
// If this EXECUTE IMMEDIATE has an INTO clause, commands are not allowed.
// The caller wraps the analyzed query in SetVariable.
if (QueryExecution.isEagerlyExecutedCommand(analyzed)) {
throw QueryCompilationErrors.invalidStatementForExecuteInto(sqlString)
}
analyzed
} else if (QueryExecution.isEagerlyExecutedCommand(analyzed)) {
// Defer eager-command payloads to the execution level. This matches the shapes
// QueryExecution.eagerlyExecuteCommands runs (not just Command). CALL is not among them and
// already ran during the analysis above.
ExecuteImmediateCommand(analyzed)
} else {
// Splice the query; it executes lazily like any other query.
analyzed
}
}

// For commands, use commandExecuted to avoid double execution
// For queries, use analyzed to avoid eager evaluation
if (result.queryExecution.analyzed.isInstanceOf[Command]) {
result.queryExecution.commandExecuted
} else {
result.queryExecution.analyzed
/**
* Parses and analyzes the dynamic SQL, returning the analyzed plan without executing it. Parsing
* and parameter binding are shared with `SparkSession.sql` via
* [[org.apache.spark.sql.classic.SparkSession.parseParameterizedPlan]]. Unlike `sql`, the plan is
* analyzed here but never wrapped in a `Dataset`, so command payloads are not run during
* analysis; the caller defers them to the execution level.
*/
private def parseAndAnalyzeInnerStatement(
session: ClassicSparkSession,
sqlString: String,
args: Seq[Expression]): LogicalPlan = {
// Extract the USING arguments the same way OPEN CURSOR does.
val (values, paramNames) = ParameterBindingUtils.buildUnifiedParameters(args)
// parseParameterizedPlan self-activates the session; withActive here also covers analysis.
// The caller has already hidden local variables and set the EXECUTE IMMEDIATE origin.
session.withActive {
val parsedPlan = session.parseParameterizedPlan(sqlString, values, paramNames)
// EXECUTE IMMEDIATE does not support SQL scripts.
if (parsedPlan.isInstanceOf[CompoundBody]) {
throw QueryCompilationErrors.sqlScriptInExecuteImmediate(sqlString)
}
session.sessionState.analyzer.executeAndCheck(parsedPlan, new QueryPlanningTracker)
}
}

Expand Down
116 changes: 61 additions & 55 deletions sql/core/src/main/scala/org/apache/spark/sql/classic/SparkSession.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{CompoundBody, LocalRelation,
import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
import org.apache.spark.sql.catalyst.util.CharVarcharUtils
import org.apache.spark.sql.classic.SparkSession.applyAndLoadExtensions
import org.apache.spark.sql.errors.{QueryCompilationErrors, SqlScriptingErrors}
import org.apache.spark.sql.errors.SqlScriptingErrors
import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.command.ExternalCommandExecutor
import org.apache.spark.sql.execution.datasources.LogicalRelation
Expand Down Expand Up @@ -644,68 +644,74 @@ class SparkSession private(
tracker: QueryPlanningTracker): DataFrame =
withActive {
val plan = tracker.measurePhase(QueryPlanningTracker.PARSING) {
// Always parse with parameter context to detect unbound parameter markers.
// Even if args is empty, we need to detect and reject parameter markers in the SQL.
val parsedPlan = if (args.nonEmpty) {
// Resolve and validate parameter arguments
val paramMap = args.zipWithIndex.map { case (arg, idx) =>
val name = if (idx < paramNames.length && paramNames(idx).nonEmpty) {
paramNames(idx)
} else {
s"_pos_$idx"
}
val expr = arg match {
case literal: Literal =>
// Already a Literal expression from ResolveExecuteImmediate - use it directly
literal
case _ =>
// Raw value or Column - convert to expression using lit()
lit(arg).expr
}
name -> expr
}.toMap

val resolvedParams = resolveAndValidateParameters(paramMap)
val paramExpressions = args.indices.map { idx =>
val name = if (idx < paramNames.length && paramNames(idx).nonEmpty) {
paramNames(idx)
} else {
s"_pos_$idx"
}
resolvedParams(name)
}.toSeq

val paramContext = HybridParameterContext(paramExpressions, paramNames.toSeq)

val parsed = sessionState.sqlParser.parsePlanWithParameters(sqlText, paramContext)
parseParameterizedPlan(sqlText, args, paramNames)
}
Dataset.ofRows(self, plan, tracker)
}

// In legacy mode, wrap with GeneralParameterizedQuery for analyzer binding
if (sessionState.conf.legacyParameterSubstitutionConstantsOnly) {
GeneralParameterizedQuery(
parsed,
args.map(lit(_).expr).toImmutableArraySeq,
paramNames.toImmutableArraySeq
)
} else {
parsed
}
/**
* Parses `sqlText` with the given parameters into an unresolved plan. `paramNames(i)` names
* `args(i)`, or is empty for a positional argument. In legacy mode the plan is wrapped in
* [[GeneralParameterizedQuery]] for analyzer binding. Self-activates the session, so callers
* need not wrap it in [[withActive]].
*/
private[sql] def parseParameterizedPlan(
sqlText: String,
args: Array[_],
paramNames: Array[String]): LogicalPlan = withActive {
// Always parse with parameter context to detect unbound parameter markers.
// Even if args is empty, we need to detect and reject parameter markers in the SQL.
val parsedPlan = if (args.nonEmpty) {
// Resolve and validate parameter arguments
val paramMap = args.zipWithIndex.map { case (arg, idx) =>
val name = if (idx < paramNames.length && paramNames(idx).nonEmpty) {
paramNames(idx)
} else {
// No arguments provided, but still need to detect parameter markers
val paramContext = HybridParameterContext(Seq.empty, Seq.empty)
sessionState.sqlParser.parsePlanWithParameters(sqlText, paramContext)
s"_pos_$idx"
}

// Check for SQL scripts in EXECUTE IMMEDIATE (applies to both empty and non-empty args)
if (parsedPlan.isInstanceOf[CompoundBody]) {
throw QueryCompilationErrors.sqlScriptInExecuteImmediate(sqlText)
val expr = arg match {
case literal: Literal =>
// Already a Literal expression from ResolveExecuteImmediate - use it directly
literal
case _ =>
// Raw value or Column - convert to expression using lit()
lit(arg).expr
}
name -> expr
}.toMap

parsedPlan
val resolvedParams = resolveAndValidateParameters(paramMap)
val paramExpressions = args.indices.map { idx =>
val name = if (idx < paramNames.length && paramNames(idx).nonEmpty) {
paramNames(idx)
} else {
s"_pos_$idx"
}
resolvedParams(name)
}.toSeq

val paramContext = HybridParameterContext(paramExpressions, paramNames.toSeq)
val parsed = sessionState.sqlParser.parsePlanWithParameters(sqlText, paramContext)

// In legacy mode, wrap with GeneralParameterizedQuery for analyzer binding
if (sessionState.conf.legacyParameterSubstitutionConstantsOnly) {
GeneralParameterizedQuery(
parsed,
args.map(lit(_).expr).toImmutableArraySeq,
paramNames.toImmutableArraySeq
)
} else {
parsed
}

Dataset.ofRows(self, plan, tracker)
} else {
// No arguments provided, but still need to detect parameter markers
val paramContext = HybridParameterContext(Seq.empty, Seq.empty)
sessionState.sqlParser.parsePlanWithParameters(sqlText, paramContext)
}

parsedPlan
}

/** @inheritdoc */
override def sql(sqlText: String): DataFrame = sql(sqlText, Map.empty[String, Any])

Expand Down
Loading