[SPARK-51131][SQL] Throw exception when SQL Script is found inside EXECUTE IMMEDIATE command

### What changes were proposed in this pull request?
An exception is thrown when user tries to use SQL Script as query string in EXECUTE IMMEDIATE.

### Why are the changes needed?
This changes are needed to forbid scripts inside EXECUTE IMMEDIATE to not introduce additional complexity in script execution or leave system in undefined state.

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
New test in `ExecuteImmediateEndToEndSuite`.

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

Closes #49850 from miland-db/milan-dankovic_data/scripting-execute-immediate.

Authored-by: Milan Dankovic <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
(cherry picked from commit 4305bb1)
Signed-off-by: Max Gekk <[email protected]>

Author: miland-db

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

@@ -4622,6 +4622,12 @@
     ],
     "sqlState" : "42K0I"
   },
+  "SQL_SCRIPT_IN_EXECUTE_IMMEDIATE" : {
+    "message" : [
+      "SQL Scripts in EXECUTE IMMEDIATE commands are not allowed. Please ensure that the SQL query provided (<sqlString>) is not SQL Script. Make sure the sql_string is a well-formed SQL statement and does not contain BEGIN and END."
+    ],
+    "sqlState" : "07501"
+  },
   "STAR_GROUP_BY_POS" : {
     "message" : [
       "Star (*) is not allowed in a select list when GROUP BY an ordinal position is used."

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

@@ -21,7 +21,7 @@ import scala.util.{Either, Left, Right}
 
 import org.apache.spark.sql.catalyst.expressions.{Alias, Expression, VariableReference}
 import org.apache.spark.sql.catalyst.parser.ParseException
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SetVariable}
+import org.apache.spark.sql.catalyst.plans.logical.{CompoundBody, LogicalPlan, SetVariable}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern.{EXECUTE_IMMEDIATE, TreePattern}
 import org.apache.spark.sql.connector.catalog.CatalogManager
@@ -179,6 +179,10 @@ class SubstituteExecuteImmediate(val catalogManager: CatalogManager)
       catalogManager.v1SessionCatalog.parser.parsePlan(queryString)
     }
 
+    if (plan.isInstanceOf[CompoundBody]) {
+      throw QueryCompilationErrors.sqlScriptInExecuteImmediate(queryString)
+    }
+
     // do not allow nested execute immediate
     if (plan.containsPattern(EXECUTE_IMMEDIATE)) {
       throw QueryCompilationErrors.nestedExecuteImmediate(queryString)

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala

@@ -4103,6 +4103,12 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
       messageParameters = Map("sqlString" -> toSQLStmt(queryString)))
   }
 
+  def sqlScriptInExecuteImmediate(sqlScriptString: String): Throwable = {
+    throw new AnalysisException(
+      errorClass = "SQL_SCRIPT_IN_EXECUTE_IMMEDIATE",
+      messageParameters = Map("sqlString" -> toSQLStmt(sqlScriptString)))
+  }
+
   def dataSourceTableSchemaMismatchError(
       dsSchema: StructType, expectedSchema: StructType): Throwable = {
     new AnalysisException(

sql/core/src/test/scala/org/apache/spark/sql/execution/ExecuteImmediateEndToEndSuite.scala

@@ -16,7 +16,8 @@
  */
 package org.apache.spark.sql.execution
 
-import org.apache.spark.sql.{QueryTest}
+import org.apache.spark.sql.{AnalysisException, QueryTest}
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.SharedSparkSession
 
 class ExecuteImmediateEndToEndSuite extends QueryTest with SharedSparkSession {
@@ -36,4 +37,16 @@ class ExecuteImmediateEndToEndSuite extends QueryTest with SharedSparkSession {
       spark.sql("DROP TEMPORARY VARIABLE IF EXISTS parm;")
     }
   }
+
+  test("SQL Scripting not supported inside EXECUTE IMMEDIATE") {
+    withSQLConf(SQLConf.SQL_SCRIPTING_ENABLED.key -> "true") {
+      val executeImmediateText = "EXECUTE IMMEDIATE 'BEGIN SELECT 1; END'"
+      checkError(
+        exception = intercept[AnalysisException ] {
+          spark.sql(executeImmediateText)
+        },
+        condition = "SQL_SCRIPT_IN_EXECUTE_IMMEDIATE",
+        parameters = Map("sqlString" -> "BEGIN SELECT 1; END"))
+    }
+  }
 }