From d13479fc4c2881921eefc4e37c53b633f0971b7c Mon Sep 17 00:00:00 2001 From: Hemanth Boyina Date: Fri, 4 Sep 2026 14:07:43 +0530 Subject: [PATCH] [SPARK-59044][SQL] Handle OFFSET 0 in physical planning when EliminateOffsets is excluded --- .../spark/sql/execution/SparkStrategies.scala | 10 ++++++++++ .../org/apache/spark/sql/SQLQuerySuite.scala | 16 +++++++++++++++- 2 files changed, 25 insertions(+), 1 deletion(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala index d395b3f986e83..bb2c1fc553818 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala +++ b/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 => diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala index 713fa39a9243c..9b8e795116189 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala +++ b/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])