[SPARK-59044][SQL] Handle OFFSET 0 in physical planning when EliminateOffsets is excluded

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala

@@ -115,6 +115,11 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
           CollectLimitExec(limit = offset + limit, child = planLater(child), offset = offset)
         case Limit(IntegerLiteral(limit), child) =>
           CollectLimitExec(limit = limit, child = planLater(child))
+        case logical.Offset(IntegerLiteral(0), child) =>
+          // OFFSET 0 is a no-op. It is normally removed by the EliminateOffsets optimizer rule,
+          // but that rule is excludable, so handle it defensively here to avoid constructing a
+          // CollectLimitExec with no limit and a zero offset (which fails its assertion).
+          planLater(child)
         case logical.Offset(IntegerLiteral(offset), child) =>
           CollectLimitExec(child = planLater(child), offset = offset)
         case Tail(IntegerLiteral(limit), child) =>
@@ -1240,6 +1245,11 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
         execution.LocalLimitExec(limit, planLater(child)) :: Nil
       case logical.GlobalLimit(IntegerLiteral(limit), child) =>
         execution.GlobalLimitExec(limit, planLater(child)) :: Nil
+      case logical.Offset(IntegerLiteral(0), child) =>
+        // OFFSET 0 is a no-op; see the note in SpecialLimits. Excluding EliminateOffsets leaves
+        // the Offset node in place, so avoid building a GlobalLimitExec with no limit and a zero
+        // offset (which fails its assertion).
+        planLater(child) :: Nil
       case logical.Offset(IntegerLiteral(offset), child) =>
         GlobalLimitExec(child = planLater(child), offset = offset) :: Nil
       case union: logical.Union =>

sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

@@ -35,7 +35,7 @@ import org.apache.spark.sql.catalyst.ExtendedAnalysisException
 import org.apache.spark.sql.catalyst.expressions.{CodegenObjectFactoryMode, GenericRow, Hex}
 import org.apache.spark.sql.catalyst.expressions.Cast._
 import org.apache.spark.sql.catalyst.expressions.aggregate.{Complete, Partial}
-import org.apache.spark.sql.catalyst.optimizer.{ConvertToLocalRelation, NestedColumnAliasingSuite, RewriteWithExpression}
+import org.apache.spark.sql.catalyst.optimizer.{ConvertToLocalRelation, EliminateOffsets, NestedColumnAliasingSuite, RewriteWithExpression}
 import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.catalyst.plans.logical.{LocalLimit, Project, RepartitionByExpression, Sort}
 import org.apache.spark.sql.connector.catalog.CatalogManager
@@ -5329,6 +5329,20 @@ class SQLQuerySuite extends SharedSparkSession with AdaptiveSparkPlanHelper
         Row(false))
     }
   }
+
+  test("SPARK-59044: OFFSET 0 succeeds when EliminateOffsets is in excludedRules") {
+    // EliminateOffsets normally removes an OFFSET 0 (a no-op) before physical planning. When the
+    // rule is excluded the Offset node survives; before the fix this failed the assertion in
+    // CollectLimitExec/GlobalLimitExec during physical planning.
+    withSQLConf(SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> EliminateOffsets.ruleName) {
+      // Collected to the driver -> SpecialLimits -> CollectLimitExec path.
+      checkAnswer(sql("SELECT 1 AS x OFFSET 0"), Row(1))
+      // Non-terminal OFFSET 0 -> BasicOperators -> GlobalLimitExec path.
+      checkAnswer(
+        sql("SELECT * FROM (SELECT id FROM range(3) OFFSET 0) ORDER BY id"),
+        Seq(Row(0), Row(1), Row(2)))
+    }
+  }
 }
 
 case class Foo(bar: Option[String])