diff --git a/sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingExecutionNode.scala b/sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingExecutionNode.scala
index e3559e8f18ae2..99719ce19e3de 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingExecutionNode.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingExecutionNode.scala
@@ -163,6 +163,14 @@ class SingleStatementExec(
   override def reset(): Unit = isExecuted = false
 }
 
+/**
+ * NO-OP leaf node, which does nothing when returned to the iterator.
+ * It is emitted by empty BEGIN END blocks.
+ */
+class NoOpStatementExec extends LeafStatementExec {
+  override def reset(): Unit = ()
+}
+
 /**
  * Executable node for CompoundBody.
  * @param statements
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreter.scala b/sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreter.scala
index a3dc3d4599314..5d3edeefc532b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreter.scala
@@ -86,9 +86,14 @@ case class SqlScriptingInterpreter(session: SparkSession) {
           .map(varName => DropVariable(varName, ifExists = true))
           .map(new SingleStatementExec(_, Origin(), args, isInternal = true))
           .reverse
-        new CompoundBodyExec(
-          collection.map(st => transformTreeIntoExecutable(st, args)) ++ dropVariables,
-          label)
+
+        val statements =
+          collection.map(st => transformTreeIntoExecutable(st, args)) ++ dropVariables match {
+            case Nil => Seq(new NoOpStatementExec)
+            case s => s
+          }
+
+        new CompoundBodyExec(statements, label)
 
       case IfElseStatement(conditions, conditionalBodies, elseBody) =>
         val conditionsExec = conditions.map(condition =>
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreterSuite.scala
index 71556c5502225..2ec42c4554e09 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/scripting/SqlScriptingInterpreterSuite.scala
@@ -110,6 +110,61 @@ class SqlScriptingInterpreterSuite extends QueryTest with SharedSparkSession {
     }
   }
 
+  test("empty begin end block") {
+    val sqlScript =
+      """
+        |BEGIN
+        |END
+        |""".stripMargin
+    val expected = Seq.empty[Seq[Row]]
+    verifySqlScriptResult(sqlScript, expected)
+  }
+
+  test("empty begin end blocks") {
+    val sqlScript =
+      """
+        |BEGIN
+        | BEGIN
+        | END;
+        | BEGIN
+        | END;
+        |END
+        |""".stripMargin
+    val expected = Seq.empty[Seq[Row]]
+    verifySqlScriptResult(sqlScript, expected)
+  }
+
+  test("empty begin end blocks with single statement") {
+    val sqlScript =
+      """
+        |BEGIN
+        | BEGIN
+        | END;
+        | SELECT 1;
+        | BEGIN
+        | END;
+        |END
+        |""".stripMargin
+    val expected = Seq(Seq(Row(1)))
+    verifySqlScriptResult(sqlScript, expected)
+  }
+
+  test("empty begin end blocks - nested") {
+    val sqlScript =
+      """
+        |BEGIN
+        | BEGIN
+        |   BEGIN
+        |   END;
+        |   BEGIN
+        |   END;
+        | END;
+        |END
+        |""".stripMargin
+    val expected = Seq.empty[Seq[Row]]
+    verifySqlScriptResult(sqlScript, expected)
+  }
+
   test("session vars - set and read (SET VAR)") {
     val sqlScript =
       """
@@ -240,6 +295,40 @@ class SqlScriptingInterpreterSuite extends QueryTest with SharedSparkSession {
     verifySqlScriptResult(commands, expected)
   }
 
+  test("if - empty body") {
+    val commands =
+      """
+        |BEGIN
+        | IF 1=1 THEN
+        |   BEGIN
+        |   END;
+        | END IF;
+        |END
+        |""".stripMargin
+    val expected = Seq.empty[Seq[Row]]
+    verifySqlScriptResult(commands, expected)
+  }
+
+  test("if - nested empty body") {
+    val commands =
+      """
+        |BEGIN
+        | IF 1=1 THEN
+        |   BEGIN
+        |     BEGIN
+        |     END;
+        |   END;
+        |   BEGIN
+        |     BEGIN
+        |     END;
+        |   END;
+        | END IF;
+        |END
+        |""".stripMargin
+    val expected = Seq.empty[Seq[Row]]
+    verifySqlScriptResult(commands, expected)
+  }
+
   test("if nested") {
     val commands =
       """
@@ -389,6 +478,42 @@ class SqlScriptingInterpreterSuite extends QueryTest with SharedSparkSession {
     verifySqlScriptResult(commands, expected)
   }
 
+  test("searched case - empty body") {
+    val commands =
+      """
+        |BEGIN
+        | CASE
+        |   WHEN 1 = 1 THEN
+        |     BEGIN
+        |     END;
+        | END CASE;
+        |END
+        |""".stripMargin
+    val expected = Seq.empty[Seq[Row]]
+    verifySqlScriptResult(commands, expected)
+  }
+
+  test("searched case - nested empty body") {
+    val commands =
+      """
+        |BEGIN
+        | CASE
+        |   WHEN 1 = 1 THEN
+        |     BEGIN
+        |       BEGIN
+        |       END;
+        |     END;
+        |     BEGIN
+        |       BEGIN
+        |       END;
+        |     END;
+        | END CASE;
+        |END
+        |""".stripMargin
+    val expected = Seq.empty[Seq[Row]]
+    verifySqlScriptResult(commands, expected)
+  }
+
   test("searched case nested") {
     val commands =
       """
@@ -589,6 +714,42 @@ class SqlScriptingInterpreterSuite extends QueryTest with SharedSparkSession {
     verifySqlScriptResult(commands, expected)
   }
 
+  test("simple case - empty body") {
+    val commands =
+      """
+        |BEGIN
+        | CASE 1
+        |   WHEN 1 THEN
+        |     BEGIN
+        |     END;
+        | END CASE;
+        |END
+        |""".stripMargin
+    val expected = Seq.empty[Seq[Row]]
+    verifySqlScriptResult(commands, expected)
+  }
+
+  test("simple case - nested empty body") {
+    val commands =
+      """
+        |BEGIN
+        | CASE 1
+        |   WHEN 1 THEN
+        |     BEGIN
+        |       BEGIN
+        |       END;
+        |     END;
+        |     BEGIN
+        |       BEGIN
+        |       END;
+        |     END;
+        | END CASE;
+        |END
+        |""".stripMargin
+    val expected = Seq.empty[Seq[Row]]
+    verifySqlScriptResult(commands, expected)
+  }
+
   test("simple case nested") {
     val commands =
       """
@@ -985,6 +1146,42 @@ class SqlScriptingInterpreterSuite extends QueryTest with SharedSparkSession {
     verifySqlScriptResult(commands, expected)
   }
 
+  test("repeat - empty body") {
+    val commands =
+      """
+        |BEGIN
+        | REPEAT
+        |   BEGIN
+        |   END;
+        | UNTIL 1 = 1
+        | END REPEAT;
+        |END
+        |""".stripMargin
+
+    val expected = Seq.empty[Seq[Row]]
+    verifySqlScriptResult(commands, expected)
+  }
+
+  test("repeat - nested empty body") {
+    val commands =
+      """
+        |BEGIN
+        | REPEAT
+        |   BEGIN
+        |     BEGIN
+        |     END;
+        |   END;
+        |   BEGIN
+        |   END;
+        | UNTIL 1 = 1
+        | END REPEAT;
+        |END
+        |""".stripMargin
+
+    val expected = Seq.empty[Seq[Row]]
+    verifySqlScriptResult(commands, expected)
+  }
+
   test("nested repeat") {
     val commands =
       """
@@ -1795,7 +1992,7 @@ class SqlScriptingInterpreterSuite extends QueryTest with SharedSparkSession {
     }
   }
 
-  test("for statement empty result") {
+  test("for statement - empty result") {
     withTable("t") {
       val sqlScript =
         """
@@ -1814,6 +2011,64 @@ class SqlScriptingInterpreterSuite extends QueryTest with SharedSparkSession {
     }
   }
 
+  test("for statement - empty body") {
+    withTable("t") {
+      val sqlScript =
+        """
+          |BEGIN
+          | CREATE TABLE t (intCol INT, stringCol STRING, doubleCol DOUBLE) using parquet;
+          | INSERT INTO t VALUES (1, 'first', 1.0);
+          | FOR row AS SELECT * FROM t DO
+          |   BEGIN
+          |   END;
+          | END FOR;
+          |END
+          |""".stripMargin
+
+      val expected = Seq(
+        Seq.empty[Row], // create table
+        Seq.empty[Row], // insert
+        Seq.empty[Row], // drop local var
+        Seq.empty[Row], // drop local var
+        Seq.empty[Row], // drop local var
+        Seq.empty[Row] // drop local var
+      )
+      verifySqlScriptResult(sqlScript, expected)
+    }
+  }
+
+  test("for statement - nested empty body") {
+    withTable("t") {
+      val sqlScript =
+        """
+          |BEGIN
+          | CREATE TABLE t (intCol INT, stringCol STRING, doubleCol DOUBLE) using parquet;
+          | INSERT INTO t VALUES (1, 'first', 1.0);
+          | FOR row AS SELECT * FROM t DO
+          |   BEGIN
+          |     BEGIN
+          |     END;
+          |   END;
+          |   BEGIN
+          |     BEGIN
+          |     END;
+          |   END;
+          | END FOR;
+          |END
+          |""".stripMargin
+
+      val expected = Seq(
+        Seq.empty[Row], // create table
+        Seq.empty[Row], // insert
+        Seq.empty[Row], // drop local var
+        Seq.empty[Row], // drop local var
+        Seq.empty[Row], // drop local var
+        Seq.empty[Row] // drop local var
+      )
+      verifySqlScriptResult(sqlScript, expected)
+    }
+  }
+
   test("for statement iterate") {
     withTable("t") {
       val sqlScript =