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

[matvei-zamiatin-db]

What changes were proposed in this pull request?

EXECUTE IMMEDIATE command payloads now run at execution time instead of during analysis.

Previously the ResolveExecuteImmediate analyzer rule parsed, analyzed, and executed the inner statement during analysis. Now the rule only parses and analyzes:

• a command payload (e.g. INSERT, SET, CREATE VIEW, REFRESH) is wrapped in a new logical node ExecuteImmediateCommand and planned to a new physical node ExecuteImmediateExec, which runs it once at the execution level
• a query is spliced in analyzed and runs lazily as usual
• an INTO clause still becomes SetVariable

The parameterized-parse logic in SparkSession.sql was extracted into a shared parseParameterizedPlan.

Why are the changes needed?

Execution during analysis is wrong and produces double-execution.

Does this PR introduce any user-facing change?

Yes. EXPLAIN EXECUTE IMMEDIATE '<command>' no longer executes the command payload, it only shows the plan. Previously the command ran during analysis (e.g. EXPLAIN EXECUTE IMMEDIATE 'DROP TABLE t' dropped the table). Running EXECUTE IMMEDIATE '<command>' directly is unchanged.

How was this patch tested?

New tests in ExecuteImmediateEndToEndSuite. The execute-immediate.sql analyzer-results golden file was regenerated to reflect the deferred (unexecuted) plan.

Was this patch authored or co-authored using generative AI tooling?

Yes, generated-by: Opus 4.8.

[uros-b]

Thank you @matvei-zamiatin-db for working on this! Please note a few comments below.

Issues

1. ExecuteImmediateExec.run() is a fragile way to run a command
   Dataset.ofRows(session, sourceStatement)
   .queryExecution.executedPlan.executeCollect()
   That depends on an implicit pipeline:

Dataset construction hits commandExecuted and runs the inner command (CommandExecutionMode.ALL).
The logical plan becomes CommandResult.
executedPlan.executeCollect() only reads cached rows.
It works today, but it is easy to break, creates an extra QueryExecution, and does not pass analyzerOpt. QueryExecution is explicit that nested QEs must propagate the outer analyzer, or a transactional write can load tables outside the transaction catalog.

Prefer QueryExecution.runCommand, which already returns the rows:

val (_, result) = QueryExecution.runCommand(
session, sourceStatement, "execute-immediate")
result.toImmutableArraySeq
Default mode is SKIP, which is correct for a single inner Command.

2. The wrapper is justified — make that contract explicit
   Splicing the analyzed Command (the same as query payloads) would already stop EXPLAIN from running it, because ExplainCommand is a SupervisingCommand leaf and SKIP does not execute the supervised plan.

The wrapper is still useful so a parent’s mapChildren(eagerlyExecuteCommands) cannot run the payload, and so EXPLAIN has a stable ExecuteImmediate node. That is worth one sentence on ExecuteImmediateCommand: this is a supervisor, not a second execution engine. Right now ExecuteImmediateExec both re-plans and re-executes via Dataset.ofRows, which blurs that.

3. CALL still runs during nested analysis
   Call is ExecutableDuringAnalysis, not Command. Inner analyzer.executeAndCheck still runs InvokeProcedures. So:

EXPLAIN EXECUTE IMMEDIATE 'CALL some_proc()'
still executes the procedure. That was true before, but the PR’s user-facing claim is broader than isInstanceOf[Command]. Either wrap/defer Call the same way, or document that only Command payloads are deferred. A test would lock the intended behavior.

4. isInstanceOf[Command] misses some executable shapes
   eagerlyExecuteCommands also special-cases WithCTE(command) and Union of commands. CTE substitution can leave DML as WithCTE(Insert..., ctes) implementing CTEInChildren, which is not a Command. Those stay spliced:

Run path: still executed later by the outer commandExecuted (OK).
EXPLAIN path: not wrapped, so you lose the ExecuteImmediate node and the “payload is not a child” guarantee.
If the wrapper exists to police eager execution, the predicate should match what QueryExecution treats as an eager command, not only isInstanceOf[Command].

5. Hidden locals / origin are analysis-only
   withHiddenLocalVariables and CurrentOrigin.withOrigin now cover parse+analyze only. Execution relies on Analyzer.executeAndCheck returning immediately when plan.analyzed is true, so Dataset.ofRows does not re-bind names.

That is probably fine, but it is implicit. A short comment on ExecuteImmediateExec.run() would help, plus a command-payload version of the existing local-variable test (INSERT ... SELECT v_local inside a script should still fail at analysis).

Tests

The new suite is the right place and covers the important invariants. A few gaps:

• The EXPLAIN ... SET spark.sql.ansi.enabled=true test only checks plan text. If EXPLAIN actually ran the SET, later tests in the same suite would be polluted. Assert the conf is unchanged, same as the DROP TABLE existence check.
• No coverage for EXPLAIN EXECUTE IMMEDIATE 'CALL ...'.
• No Connect analyze coverage (CommandExecutionMode.SKIP is a main beneficiary).
• PR description says only execute-immediate.sql.out was regenerated; identifier-clause.sql.out changed too.

Nits

JIRA component is Spark Core; this is SQL.

parseParameterizedPlan still throws SQL_SCRIPT_IN_EXECUTE_IMMEDIATE for any caller of the unified sql(text, args, paramNames) API (including OPEN CURSOR). Pre-existing, but the new scaladoc makes it look like an EXECUTE IMMEDIATE-only helper.

LeafV2CommandExec is documented as “no Spark job”. An inner INSERT will start jobs from run(). Same pattern as other driver-side commands; maybe not the ideal base, but acceptable.