From 9ba5643f4bb7697ee918f1de83390db58dd81e29 Mon Sep 17 00:00:00 2001 From: Matvei Zamiatin Date: Mon, 21 Sep 2026 12:45:43 +0000 Subject: [PATCH] [SPARK-59689][SQL] Run EXECUTE IMMEDIATE commands at execution time instead of during analysis --- .../catalyst/plans/logical/v2Commands.scala | 14 +- .../planner/SparkConnectServiceSuite.scala | 36 +++ .../analysis/ResolveExecuteImmediate.scala | 126 ++++++---- .../spark/sql/classic/SparkSession.scala | 116 +++++----- .../spark/sql/execution/QueryExecution.scala | 32 ++- .../command/v2/ExecuteImmediateExec.scala | 48 ++++ .../command/v2/V2CommandStrategy.scala | 3 + .../execute-immediate.sql.out | 10 +- .../identifier-clause.sql.out | 6 +- .../ExecuteImmediateEndToEndSuite.scala | 215 ++++++++++++++++++ .../sql/execution/QueryExecutionSuite.scala | 25 +- 11 files changed, 506 insertions(+), 125 deletions(-) create mode 100644 sql/core/src/main/scala/org/apache/spark/sql/execution/command/v2/ExecuteImmediateExec.scala diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala index b816016a3ec84..845d9fa75c41a 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala @@ -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 @@ -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. * diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectServiceSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectServiceSuite.scala index ae3b167c3da4a..bc2cf4a9dba9f 100644 --- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectServiceSuite.scala +++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectServiceSuite.scala @@ -881,6 +881,42 @@ 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(""" 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 c09498b41c974..e0314e8371086 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 @@ -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. @@ -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, @@ -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) } } @@ -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], @@ -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) } } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/classic/SparkSession.scala b/sql/core/src/main/scala/org/apache/spark/sql/classic/SparkSession.scala index ffbfed21bb29d..a35c0aa01530a 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/classic/SparkSession.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/classic/SparkSession.scala @@ -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 @@ -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]) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala index a95cc12db8319..b583f02a2fd45 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala @@ -279,16 +279,14 @@ class QueryExecution( result.toImmutableArraySeq) } p transformDown { - case u @ Union(children, _, _) if children.forall(_.isInstanceOf[Command]) => - eagerlyExecute(u, "multi-commands", CommandExecutionMode.SKIP) - case w @ WithCTE(u @ Union(children, _, _), _) if children.forall(_.isInstanceOf[Command]) => - eagerlyExecute(w, "multi-commands", CommandExecutionMode.SKIP) - case c: Command => - val name = commandExecutionName(c) - eagerlyExecute(c, name, CommandExecutionMode.NON_ROOT) - case w @ WithCTE(c: Command, _) => - val name = commandExecutionName(c) - eagerlyExecute(w, name, CommandExecutionMode.SKIP) + // isEagerlyExecutedCommand decides the shapes; this only maps each to a name and mode. + case node if QueryExecution.isEagerlyExecutedCommand(node) => + val (name, mode) = node match { + case c: Command => (commandExecutionName(c), CommandExecutionMode.NON_ROOT) + case WithCTE(c: Command, _) => (commandExecutionName(c), CommandExecutionMode.SKIP) + case _ => ("multi-commands", CommandExecutionMode.SKIP) // Union / WithCTE(Union) + } + eagerlyExecute(node, name, mode) } } @@ -796,6 +794,20 @@ object QueryExecution { private def nextExecutionId: Long = _nextExecutionId.getAndIncrement + /** + * Whether [[QueryExecution.eagerlyExecuteCommands]] would eagerly execute `plan` as a command: + * a `Command`, a `Union` of commands, or either wrapped in a `WithCTE`. The single source of + * truth for those shapes, gated on by `eagerlyExecuteCommands`. EXECUTE IMMEDIATE uses it to + * defer matching inner payloads to the execution level. + */ + private[sql] def isEagerlyExecutedCommand(plan: LogicalPlan): Boolean = plan match { + case Union(children, _, _) => children.forall(_.isInstanceOf[Command]) + case WithCTE(Union(children, _, _), _) => children.forall(_.isInstanceOf[Command]) + case _: Command => true + case WithCTE(_: Command, _) => true + case _ => false + } + private[execution] def create( sparkSession: SparkSession, logical: LogicalPlan, diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/v2/ExecuteImmediateExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/v2/ExecuteImmediateExec.scala new file mode 100644 index 0000000000000..385b66ccd2857 --- /dev/null +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/v2/ExecuteImmediateExec.scala @@ -0,0 +1,48 @@ +/* + * 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.execution.command.v2 + +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.Attribute +import org.apache.spark.sql.catalyst.plans.QueryPlan +import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan +import org.apache.spark.sql.execution.QueryExecution +import org.apache.spark.sql.execution.datasources.v2.LeafV2CommandExec +import org.apache.spark.util.ArrayImplicits._ + +/** + * Physical plan for an EXECUTE IMMEDIATE command payload. Runs the already-analyzed inner command + * once, here at the execution level, and returns its rows. [[ExecuteImmediateCommand]] keeps the + * payload out of the logical plan's children, so the eager-command path does not run it first and + * this node is its sole executor. The inner plan is exposed via [[innerChildren]] so EXPLAIN shows + * the payload. + */ +case class ExecuteImmediateExec( + output: Seq[Attribute], + sourceStatement: LogicalPlan) extends LeafV2CommandExec { + + override protected def run(): Seq[InternalRow] = { + // sourceStatement is already analyzed, so runCommand does not re-bind names (local variables + // stay hidden as resolved by ResolveExecuteImmediate). + val (_, result) = QueryExecution.runCommand(session, sourceStatement, "execute-immediate") + result.toImmutableArraySeq + } + + // Expose the inner plan so EXPLAIN shows what EXECUTE IMMEDIATE runs. + override def innerChildren: Seq[QueryPlan[_]] = Seq(sourceStatement) +} diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/v2/V2CommandStrategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/v2/V2CommandStrategy.scala index 3cba5679fc755..cb427f049b0b3 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/v2/V2CommandStrategy.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/v2/V2CommandStrategy.scala @@ -40,6 +40,9 @@ object V2CommandStrategy extends Strategy { case SetVariable(variables, query) => SetVariableExec(variables.map(_.asInstanceOf[VariableReference]), planLater(query)) :: Nil + case ExecuteImmediateCommand(sourceStatement) => + ExecuteImmediateExec(sourceStatement.output, sourceStatement) :: Nil + case DeclareCursor(cursorName, queryText, asensitive) => DeclareCursorExec(cursorName, queryText, asensitive) :: Nil diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/execute-immediate.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/execute-immediate.sql.out index 2dffe7a92ca70..4a2c581e8a9ea 100644 --- a/sql/core/src/test/resources/sql-tests/analyzer-results/execute-immediate.sql.out +++ b/sql/core/src/test/resources/sql-tests/analyzer-results/execute-immediate.sql.out @@ -48,14 +48,14 @@ SetVariable [variablereference(system.session.sql_string=CAST(NULL AS STRING))] -- !query EXECUTE IMMEDIATE 'SET spark.sql.ansi.enabled=true' -- !query analysis -CommandResult [key#x, value#x], Execute SetCommand, [[spark.sql.ansi.enabled,true]] +ExecuteImmediateCommand +- SetCommand (spark.sql.ansi.enabled,Some(true)) -- !query EXECUTE IMMEDIATE 'CREATE TEMPORARY VIEW IDENTIFIER(:tblName) AS SELECT id, name FROM tbl_view' USING 'tbl_view_tmp' as tblName -- !query analysis -CommandResult Execute CreateViewCommand +ExecuteImmediateCommand +- CreateViewCommand `tbl_view_tmp`, SELECT id, name FROM tbl_view, false, false, LocalTempView, UNSUPPORTED, true +- Project [id#x, name#x] +- SubqueryAlias tbl_view @@ -85,7 +85,7 @@ Project [id#x, name#x] -- !query EXECUTE IMMEDIATE 'REFRESH TABLE IDENTIFIER(:tblName)' USING 'x' as tblName -- !query analysis -CommandResult Execute RefreshTableCommand +ExecuteImmediateCommand +- RefreshTableCommand `spark_catalog`.`default`.`x` @@ -206,7 +206,7 @@ Project [id#x, name#x, data#x] -- !query EXECUTE IMMEDIATE 'INSERT INTO x VALUES(?)' USING 1 -- !query analysis -CommandResult Execute InsertIntoHadoopFsRelationCommand file:[not included in comparison]/{warehouse_dir}/x, false, CSV, [path=file:[not included in comparison]/{warehouse_dir}/x], Append, `spark_catalog`.`default`.`x`, org.apache.spark.sql.execution.datasources.InMemoryFileIndex(file:[not included in comparison]/{warehouse_dir}/x), [id] +ExecuteImmediateCommand +- InsertIntoHadoopFsRelationCommand file:[not included in comparison]/{warehouse_dir}/x, false, CSV, [path=file:[not included in comparison]/{warehouse_dir}/x], Append, `spark_catalog`.`default`.`x`, org.apache.spark.sql.execution.datasources.InMemoryFileIndex(file:[not included in comparison]/{warehouse_dir}/x), [id] +- Project [col1#x AS id#x] +- LocalRelation [col1#x] @@ -311,7 +311,7 @@ Project [id#x, name#x, data#x, name7 AS p#x] -- !query EXECUTE IMMEDIATE 'SET VAR sql_string = ?' USING 'SELECT id from tbl_view where name = :first' -- !query analysis -CommandResult SetVariable [variablereference(system.session.sql_string='SELECT * from tbl_view where name = :first or id = :second')] +ExecuteImmediateCommand +- SetVariable [variablereference(system.session.sql_string='SELECT * from tbl_view where name = :first or id = :second')] +- Project [SELECT id from tbl_view where name = :first AS sql_string#x] +- OneRowRelation diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/identifier-clause.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/identifier-clause.sql.out index 5e760c01b59fa..2b1db879cb58e 100644 --- a/sql/core/src/test/resources/sql-tests/analyzer-results/identifier-clause.sql.out +++ b/sql/core/src/test/resources/sql-tests/analyzer-results/identifier-clause.sql.out @@ -2496,7 +2496,7 @@ Sort [c1#x DESC NULLS LAST, c2#x ASC NULLS FIRST], true EXECUTE IMMEDIATE 'INSERT INTO integration_test(IDENTIFIER(:col1), IDENTIFIER(:col2)) VALUES (:val1, :val2)' USING 'c1' AS col1, 'c2' AS col2, 3 AS val1, 'c' AS val2 -- !query analysis -CommandResult Execute InsertIntoHadoopFsRelationCommand file:[not included in comparison]/{warehouse_dir}/identifier_clause_test_schema.db/integration_test, false, CSV, [path=file:[not included in comparison]/{warehouse_dir}/identifier_clause_test_schema.db/integration_test], Append, `spark_catalog`.`identifier_clause_test_schema`.`integration_test`, org.apache.spark.sql.execution.datasources.InMemoryFileIndex(file:[not included in comparison]/{warehouse_dir}/identifier_clause_test_schema.db/integration_test), [c1, c2] +ExecuteImmediateCommand +- InsertIntoHadoopFsRelationCommand file:[not included in comparison]/{warehouse_dir}/identifier_clause_test_schema.db/integration_test, false, CSV, [path=file:[not included in comparison]/{warehouse_dir}/identifier_clause_test_schema.db/integration_test], Append, `spark_catalog`.`identifier_clause_test_schema`.`integration_test`, org.apache.spark.sql.execution.datasources.InMemoryFileIndex(file:[not included in comparison]/{warehouse_dir}/identifier_clause_test_schema.db/integration_test), [c1, c2] +- Project [c1#x AS c1#x, c2#x AS c2#x] +- Project [col1#x AS c1#x, col2#x AS c2#x] @@ -2543,7 +2543,7 @@ WithCTE EXECUTE IMMEDIATE 'CREATE OR REPLACE TEMPORARY VIEW IDENTIFIER(:view_name)(IDENTIFIER(:col_name)) AS VALUES(1)' USING 'test_view' AS view_name, 'test_col' AS col_name -- !query analysis -CommandResult Execute CreateViewCommand +ExecuteImmediateCommand +- CreateViewCommand `test_view`, [(test_col,None)], VALUES(1), false, true, LocalTempView, UNSUPPORTED, true +- LocalRelation [col1#x] @@ -2569,7 +2569,7 @@ DropTempViewCommand test_view, false EXECUTE IMMEDIATE 'ALTER TABLE IDENTIFIER(:tab) ADD COLUMN IDENTIFIER(:new_col) INT' USING 'integration_test' AS tab, 'c4' AS new_col -- !query analysis -CommandResult Execute AlterTableAddColumnsCommand +ExecuteImmediateCommand +- AlterTableAddColumnsCommand `spark_catalog`.`identifier_clause_test_schema`.`integration_test`, [StructField(c4,IntegerType,true)] diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/ExecuteImmediateEndToEndSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/ExecuteImmediateEndToEndSuite.scala index eb642f9b6bd24..6fd0f95abcd3c 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/ExecuteImmediateEndToEndSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/ExecuteImmediateEndToEndSuite.scala @@ -16,7 +16,9 @@ */ package org.apache.spark.sql.execution +import org.apache.spark.SparkThrowable import org.apache.spark.sql.{AnalysisException, Row} +import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.SharedSparkSession class ExecuteImmediateEndToEndSuite extends SharedSparkSession { @@ -93,6 +95,35 @@ class ExecuteImmediateEndToEndSuite extends SharedSparkSession { fragment = "v1")) } + test("EXECUTE IMMEDIATE does not resolve local variables in a command payload") { + withTable("ei_local_cmd") { + spark.sql("CREATE TABLE ei_local_cmd (id INT) USING parquet") + // A deferred command payload is analyzed with local variables hidden, just like a query + // payload, and that analysis is not re-run at the execution level. Referencing a local + // variable must therefore fail at analysis. + val result = intercept[AnalysisException] { + spark.sql( + """ + |BEGIN + | DECLARE v1 = 5; + | EXECUTE IMMEDIATE 'INSERT INTO ei_local_cmd SELECT v1'; + |END + |""".stripMargin) + } + checkError( + exception = result, + condition = "UNRESOLVED_COLUMN.WITHOUT_SUGGESTION", + sqlState = "42703", + parameters = Map("objectName" -> "`v1`"), + context = ExpectedContext( + objectType = "EXECUTE IMMEDIATE", + objectName = "", + startIndex = 32, + stopIndex = 33, + fragment = "v1")) + } + } + test("EXECUTE IMMEDIATE resolves local variable in USING clause") { val result = spark.sql( """ @@ -143,4 +174,188 @@ class ExecuteImmediateEndToEndSuite extends SharedSparkSession { fragment = "v2")) } } + + test("EXPLAIN EXECUTE IMMEDIATE does not execute the command payload") { + withTable("execute_immediate_explain") { + spark.sql("CREATE TABLE execute_immediate_explain (id INT) USING parquet") + // EXPLAIN analyzes the payload but must not run it: command execution is deferred to the + // execution level, so the DROP should have no effect here. + spark.sql("EXPLAIN EXECUTE IMMEDIATE 'DROP TABLE execute_immediate_explain'").collect() + assert(spark.catalog.tableExists("execute_immediate_explain"), + "EXPLAIN must not execute the EXECUTE IMMEDIATE command payload") + } + } + + test("EXECUTE IMMEDIATE executes the command payload when run") { + withTable("execute_immediate_run") { + spark.sql("CREATE TABLE execute_immediate_run (id INT) USING parquet") + spark.sql("EXECUTE IMMEDIATE 'DROP TABLE execute_immediate_run'") + assert(!spark.catalog.tableExists("execute_immediate_run"), + "EXECUTE IMMEDIATE must execute the command payload") + } + } + + test("EXECUTE IMMEDIATE runs a command payload exactly once") { + withTable("execute_immediate_once") { + spark.sql("CREATE TABLE execute_immediate_once (id INT) USING parquet") + // ExecuteImmediateExec is the sole executor of the payload; a double execution would insert + // the row twice. Asserting exactly one row guards the single-execution invariant. + spark.sql("EXECUTE IMMEDIATE 'INSERT INTO execute_immediate_once VALUES (?)' USING 1") + checkAnswer(spark.table("execute_immediate_once"), Row(1)) + } + } + + test("EXPLAIN shows the EXECUTE IMMEDIATE command payload node") { + withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") { + val plan = spark.sql("EXPLAIN EXECUTE IMMEDIATE 'SET spark.sql.ansi.enabled=true'") + .collect().map(_.getString(0)).mkString("\n") + // The physical node renders as "ExecuteImmediate" (TreeNode.nodeName strips the "Exec" + // suffix); assert it appears together with its supervised payload, which EXPLAIN surfaces via + // innerChildren. + assert(plan.contains("ExecuteImmediate") && plan.contains("SetCommand"), + s"EXPLAIN should show the ExecuteImmediate node wrapping its payload, but was:\n$plan") + // EXPLAIN must analyze but not run the SET, so the conf stays at its pre-EXPLAIN value; + // otherwise it would pollute later tests in this suite. + assert(spark.conf.get(SQLConf.ANSI_ENABLED.key) == "false", + "EXPLAIN must not execute the EXECUTE IMMEDIATE SET payload") + } + } + + test("EXECUTE IMMEDIATE runs a nested command payload exactly once") { + withTable("ei_nested") { + spark.sql("CREATE TABLE ei_nested (id INT) USING parquet") + // The inner statement is itself an EXECUTE IMMEDIATE command, so the payload is a nested + // ExecuteImmediateCommand: ExecuteImmediateExec.run runs it via QueryExecution.runCommand, + // which plans it to another ExecuteImmediateExec. Asserting exactly one row guards the + // single-execution invariant through the recursive deferral. + spark.sql( + """EXECUTE IMMEDIATE 'EXECUTE IMMEDIATE \'INSERT INTO ei_nested VALUES (1)\''""") + checkAnswer(spark.table("ei_nested"), Row(1)) + } + } + + test("EXECUTE IMMEDIATE runs a deferred command payload inside a SQL script") { + withTable("ei_script") { + spark.sql("CREATE TABLE ei_script (id INT) USING parquet") + // A command payload deferred to the execution level must still run inside a BEGIN...END + // script frame, where variable hiding and the scripting context are set up during analysis. + spark.sql( + """ + |BEGIN + | EXECUTE IMMEDIATE 'INSERT INTO ei_script VALUES (1)'; + |END + |""".stripMargin).collect() + checkAnswer(spark.table("ei_script"), Row(1)) + } + } + + test("EXPLAIN EXECUTE IMMEDIATE does not run a parameterized command payload") { + withTable("ei_explain_param") { + spark.sql("CREATE TABLE ei_explain_param (id INT) USING parquet") + val plan = spark.sql( + "EXPLAIN EXECUTE IMMEDIATE 'INSERT INTO ei_explain_param VALUES (?)' USING 1") + .collect().map(_.getString(0)).mkString("\n") + assert(plan.contains("ExecuteImmediate"), + s"EXPLAIN should show the ExecuteImmediate node, but was:\n$plan") + // EXPLAIN analyzes and binds the parameter but must not run the command payload. + checkAnswer(spark.table("ei_explain_param"), Seq.empty[Row]) + } + } + + test("EXPLAIN EXECUTE IMMEDIATE splices a query payload instead of wrapping it") { + val plan = spark.sql("EXPLAIN EXECUTE IMMEDIATE 'SELECT 1'") + .collect().map(_.getString(0)).mkString("\n") + // A query payload is spliced directly, so no ExecuteImmediate node wraps it (unlike a + // command payload); EXPLAIN renders the query plan itself. + assert(plan.nonEmpty && !plan.contains("ExecuteImmediate"), + s"query payloads should be spliced, not wrapped, but was:\n$plan") + } + + test("EXECUTE IMMEDIATE runtime error in a deferred command references the dynamic SQL") { + withTable("ei_ctx_src", "ei_ctx_dst") { + withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") { + spark.sql("CREATE TABLE ei_ctx_src (v INT) USING parquet") + spark.sql("INSERT INTO ei_ctx_src VALUES (0)") + spark.sql("CREATE TABLE ei_ctx_dst (r BIGINT) USING parquet") + // The command payload runs at the execution level, not during analysis. A runtime failure + // inside it must still carry the dynamic SQL's query context (origin objectType + // "EXECUTE IMMEDIATE"); `v` is a column so the division is not constant-folded and fails + // during execution. + val e = intercept[Exception] { + spark.sql("EXECUTE IMMEDIATE 'INSERT INTO ei_ctx_dst SELECT 1 div v FROM ei_ctx_src'") + } + // The SparkThrowable may be wrapped in an execution/task exception, so scan the cause chain + // for the first one carrying a query context. + val contexts = Iterator.iterate(e: Throwable)(_.getCause).takeWhile(_ != null) + .collect { case st: SparkThrowable if st.getQueryContext.nonEmpty => st.getQueryContext } + .toSeq + assert(contexts.nonEmpty, s"runtime error should carry a query context, but was: $e") + val ctx = contexts.head.head + assert(ctx.objectType() == "EXECUTE IMMEDIATE", + s"context should reference the EXECUTE IMMEDIATE origin, but was '${ctx.objectType()}'") + assert(ctx.fragment().contains("div"), + s"context should point at the failing dynamic-SQL fragment, but was '${ctx.fragment()}'") + } + } + } + + test("EXPLAIN of EXECUTE IMMEDIATE INTO does not assign, and INTO rejects command payloads") { + withSessionVariable("ei_into_v") { + spark.sql("DECLARE ei_into_v INT") + // The INTO clause becomes a deferred SetVariable; EXPLAIN must analyze but not assign it. + spark.sql("EXPLAIN EXECUTE IMMEDIATE 'SELECT 1' INTO ei_into_v").collect() + checkAnswer(spark.sql("SELECT ei_into_v"), Row(null)) + // Executing it assigns the variable. + spark.sql("EXECUTE IMMEDIATE 'SELECT 1' INTO ei_into_v") + checkAnswer(spark.sql("SELECT ei_into_v"), Row(1)) + // A command payload with INTO is rejected during analysis. + withTable("ei_into_cmd") { + spark.sql("CREATE TABLE ei_into_cmd (id INT) USING parquet") + checkError( + exception = intercept[AnalysisException] { + spark.sql("EXECUTE IMMEDIATE 'INSERT INTO ei_into_cmd VALUES (1)' INTO ei_into_v") + }, + condition = "INVALID_STATEMENT_FOR_EXECUTE_INTO", + parameters = Map("sqlString" -> "INSERT INTO EI_INTO_CMD VALUES (1)")) + } + } + } + + test("EXECUTE IMMEDIATE preserves a multi-column command's output schema and rows") { + withTable("ei_show_tbl") { + spark.sql("CREATE TABLE ei_show_tbl (id INT) USING parquet") + // SHOW TABLES is a command with multi-column output. ExecuteImmediateExec.output is + // sourceStatement.output, so the deferred command must expose the same schema and rows as a + // direct SHOW TABLES (guards output stability across the execution-level re-plan). + val direct = spark.sql("SHOW TABLES") + val viaEI = spark.sql("EXECUTE IMMEDIATE 'SHOW TABLES'") + assert(viaEI.schema == direct.schema, + s"schema mismatch: EI=${viaEI.schema} direct=${direct.schema}") + checkAnswer(viaEI, direct.collect().toIndexedSeq) + } + } + + test("EXECUTE IMMEDIATE defers a multi-INSERT (Union of commands) to the execution level") { + withTable("ei_multi_src", "ei_multi_a", "ei_multi_b") { + spark.sql("CREATE TABLE ei_multi_src (id INT) USING parquet") + spark.sql("INSERT INTO ei_multi_src VALUES (1), (2)") + spark.sql("CREATE TABLE ei_multi_a (id INT) USING parquet") + spark.sql("CREATE TABLE ei_multi_b (id INT) USING parquet") + // A multi-INSERT analyzes to a Union of INSERT commands, which is not itself a Command. + // ExecuteImmediate must still wrap it (isEagerlyExecutedCommand covers a Union of commands), + // so EXPLAIN shows the ExecuteImmediate node and does not run the inserts. + val multiInsert = "FROM ei_multi_src " + + "INSERT INTO ei_multi_a SELECT id INSERT INTO ei_multi_b SELECT id" + val plan = spark.sql(s"EXPLAIN EXECUTE IMMEDIATE '$multiInsert'") + .collect().map(_.getString(0)).mkString("\n") + assert(plan.contains("ExecuteImmediate"), + s"multi-INSERT payload should be wrapped, not spliced, but was:\n$plan") + assert(spark.table("ei_multi_a").isEmpty && spark.table("ei_multi_b").isEmpty, + "EXPLAIN must not run the multi-INSERT command payload") + // Running it performs both inserts. + spark.sql(s"EXECUTE IMMEDIATE '$multiInsert'") + checkAnswer(spark.table("ei_multi_a"), Seq(Row(1), Row(2))) + checkAnswer(spark.table("ei_multi_b"), Seq(Row(1), Row(2))) + } + } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala index ab3b4c3df78d6..70a74a3459d5b 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala @@ -27,7 +27,7 @@ import org.apache.spark.sql.catalyst.{QueryPlanningTracker, QueryPlanningTracker import org.apache.spark.sql.catalyst.analysis.{CurrentNamespace, UnresolvedFunction, UnresolvedRelation} import org.apache.spark.sql.catalyst.expressions.{Alias, NamedLambdaVariable, RegExpReplace, UnsafeRow} import org.apache.spark.sql.catalyst.plans.QueryPlan -import org.apache.spark.sql.catalyst.plans.logical.{CommandResult, LogicalPlan, OneRowRelation, Project, ShowTables, SubqueryAlias} +import org.apache.spark.sql.catalyst.plans.logical.{Command, CommandResult, LogicalPlan, OneRowRelation, Project, ShowTables, SubqueryAlias, Union, WithCTE} import org.apache.spark.sql.catalyst.trees.TreeNodeTag import org.apache.spark.sql.catalyst.util.StringUtils.PlanStringConcat import org.apache.spark.sql.classic.Dataset @@ -754,6 +754,29 @@ class QueryExecutionSuite extends SharedSparkSession { assert(trackerReadyForExecution != null) } } + + test("SPARK-59689: isEagerlyExecutedCommand classifies eager-command plan shapes") { + val parser = spark.sessionState.sqlParser + // A bare command and a non-command query, obtained without executing them. + val command = parser.parsePlan("SET spark.sql.ansi.enabled=true") + assert(command.isInstanceOf[Command]) + val query = parser.parsePlan("SELECT 1") + assert(!query.isInstanceOf[Command]) + + // The WithCTE-wrapped shapes cannot arise from SQL parsing (a CTE on a DML command is pushed + // into the command's query child), so build them directly. This is the classifier EXECUTE + // IMMEDIATE's deferral and INTO rejection both gate on. + assert(QueryExecution.isEagerlyExecutedCommand(command)) + assert(QueryExecution.isEagerlyExecutedCommand(Union(Seq(command, command)))) + assert(QueryExecution.isEagerlyExecutedCommand(WithCTE(command, Nil))) + assert(QueryExecution.isEagerlyExecutedCommand(WithCTE(Union(Seq(command, command)), Nil))) + + // Queries -- including a Union or WithCTE that wraps them -- are not eager commands. + assert(!QueryExecution.isEagerlyExecutedCommand(query)) + assert(!QueryExecution.isEagerlyExecutedCommand(Union(Seq(query, query)))) + assert(!QueryExecution.isEagerlyExecutedCommand(WithCTE(query, Nil))) + assert(!QueryExecution.isEagerlyExecutedCommand(Union(Seq(command, query)))) + } } class ExtendedInfo extends ExtendedExplainGenerator {