[SPARK-39057][SQL] Offset could work without Limit

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

@@ -433,6 +433,13 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog {
 
           case Offset(offsetExpr, _) => checkLimitLikeClause("offset", offsetExpr)
 
+          case o if !o.isInstanceOf[GlobalLimit] && !o.isInstanceOf[LocalLimit]
+            && o.children.exists(_.isInstanceOf[Offset]) =>
+            failAnalysis(
+              s"""
+                 |The OFFSET clause is allowed in the LIMIT clause or be the outermost node,
+                 |but the OFFSET clause found in: ${o.nodeName}.""".stripMargin.replace("\n", " "))
+
           case Tail(limitExpr, _) => checkLimitLikeClause("tail", limitExpr)
 
           case _: Union | _: SetOperation if operator.children.length > 1 =>
@@ -608,7 +615,6 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog {
         }
     }
     checkCollectedMetrics(plan)
-    checkOffsetOperator(plan)
     extendedCheckRules.foreach(_(plan))
     plan.foreachUp {
       case o if !o.resolved =>
@@ -851,30 +857,6 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog {
     check(plan)
   }
 
-  /**
-   * Validate whether the [[Offset]] is valid.
-   */
-  private def checkOffsetOperator(plan: LogicalPlan): Unit = {
-    plan.foreachUp {
-      case o if !o.isInstanceOf[GlobalLimit] && !o.isInstanceOf[LocalLimit]
-        && o.children.exists(_.isInstanceOf[Offset]) =>
-        failAnalysis(
-          s"""
-             |The OFFSET clause is only allowed in the LIMIT clause, but the OFFSET
-             |clause found in: ${o.nodeName}.""".stripMargin.replace("\n", " "))
-      case _ =>
-    }
-    plan match {
-      case Offset(offsetExpr, _) =>
-        checkLimitLikeClause("offset", offsetExpr)
-        failAnalysis(
-          s"""
-             |The OFFSET clause is only allowed in the LIMIT clause, but the OFFSET
-             |clause is found to be the outermost node.""".stripMargin.replace("\n", " "))
-      case _ =>
-    }
-  }
-
   /**
    * Validates to make sure the outer references appearing inside the subquery
    * are allowed.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -96,7 +96,6 @@ abstract class Optimizer(catalogManager: CatalogManager)
         CollapseWindow,
         CombineFilters,
         EliminateLimits,
-        RewriteOffsets,
         CombineUnions,
         // Constant folding and strength reduction
         OptimizeRepartition,
@@ -673,7 +672,7 @@ object RemoveNoopUnion extends Rule[LogicalPlan] {
 }
 
 /**
- * Pushes down [[LocalLimit]] beneath UNION ALL and joins.
+ * Pushes down [[LocalLimit]] beneath UNION ALL, OFFSET and joins.
  */
 object LimitPushDown extends Rule[LogicalPlan] {
 
@@ -750,6 +749,11 @@ object LimitPushDown extends Rule[LogicalPlan] {
       Limit(le, Project(a.aggregateExpressions, LocalLimit(le, a.child)))
     case Limit(le @ IntegerLiteral(1), p @ Project(_, a: Aggregate)) if a.groupOnly =>
       Limit(le, p.copy(child = Project(a.aggregateExpressions, LocalLimit(le, a.child))))
+    // Merge offset value and limit value into LocalLimit and pushes down LocalLimit through Offset.
+    case LocalLimit(le, Offset(oe @ IntegerLiteral(offset), grandChild)) if offset > 0 =>
+      Offset(oe, LocalLimit(Add(le, oe), grandChild))
+    // Eliminate Offset if offset value equals 0.
+    case Offset(IntegerLiteral(0), child) => child
   }
 }
 
@@ -1870,25 +1874,6 @@ object EliminateLimits extends Rule[LogicalPlan] {
   }
 }
 
-/**
- * Rewrite [[Offset]] as [[GlobalLimitAndOffset]] or [[LocalLimit]],
- * merging the expressions into one single expression. See [[Limit]] for more information
- * about the difference between [[LocalLimit]] and [[GlobalLimit]].
- */
-object RewriteOffsets extends Rule[LogicalPlan] {
-  def apply(plan: LogicalPlan): LogicalPlan = plan transform {
-    case GlobalLimit(le, Offset(oe, grandChild)) =>
-      GlobalLimitAndOffset(le, oe, grandChild)
-    case localLimit @ LocalLimit(le, Offset(oe, grandChild)) =>
-      val offset = oe.eval().asInstanceOf[Int]
-      if (offset == 0) {
-        localLimit.withNewChildren(Seq(grandChild))
-      } else {
-        Offset(oe, LocalLimit(Add(le, oe), grandChild))
-      }
-  }
-}
-
 /**
  * Check if there any cartesian products between joins of any type in the optimized plan tree.
  * Throw an error if a cartesian product is found without an explicit cross join specified.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

@@ -1304,36 +1304,16 @@ case class LocalLimit(limitExpr: Expression, child: LogicalPlan) extends OrderPr
 }
 
 object LimitAndOffset {
-  def unapply(p: GlobalLimitAndOffset): Option[(Expression, Expression, LogicalPlan)] = {
+  def unapply(p: GlobalLimit): Option[(Expression, Expression, LogicalPlan)] = {
     p match {
-      case GlobalLimitAndOffset(le1, le2, LocalLimit(le3, child)) if le1.eval().asInstanceOf[Int]
+      case GlobalLimit(le1, Offset(le2, LocalLimit(le3, child))) if le1.eval().asInstanceOf[Int]
         + le2.eval().asInstanceOf[Int] == le3.eval().asInstanceOf[Int] =>
         Some((le1, le2, child))
       case _ => None
     }
   }
 }
 
-/**
- * A global (coordinated) limit with offset. This operator can skip at most `offsetExpr` number and
- * emit at most `limitExpr` number in total. For example, if we have LIMIT 10 OFFSET 5, we impose a
- * total limit of 10 + 5 = 15 rows and then discard the first 5, leaving 10 rows remaining.
- */
-case class GlobalLimitAndOffset(
-    limitExpr: Expression,
-    offsetExpr: Expression,
-    child: LogicalPlan) extends OrderPreservingUnaryNode {
-  override def output: Seq[Attribute] = child.output
-  override def maxRows: Option[Long] = {
-    limitExpr match {
-      case IntegerLiteral(limit) => Some(limit)
-      case _ => None
-    }
-  }
-  override protected def withNewChildInternal(newChild: LogicalPlan): GlobalLimitAndOffset =
-    copy(child = newChild)
-}
-
 /**
  * This is similar with [[Limit]] except:
  *

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisErrorSuite.scala

@@ -556,18 +556,11 @@ class AnalysisErrorSuite extends AnalysisTest {
     "The offset expression must be equal to or greater than 0, but got -1" :: Nil
   )
 
-  errorTest(
-    "OFFSET clause is outermost node",
-    testRelation.offset(Literal(10, IntegerType)),
-    "The OFFSET clause is only allowed in the LIMIT clause, but the OFFSET" +
-      " clause is found to be the outermost node." :: Nil
-  )
-
   errorTest(
     "OFFSET clause in other node",
     testRelation2.offset(Literal(10, IntegerType)).where('b > 1),
-    "The OFFSET clause is only allowed in the LIMIT clause, but the OFFSET" +
-      " clause found in: Filter." :: Nil
+    "The OFFSET clause is allowed in the LIMIT clause or be the outermost node," +
+      " but the OFFSET clause found in: Filter." :: Nil
   )
 
   errorTest(

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

@@ -93,14 +93,16 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
         case LimitAndOffset(IntegerLiteral(limit), IntegerLiteral(offset),
           Sort(order, true, child)) if limit + offset < conf.topKSortFallbackThreshold =>
           TakeOrderedAndProjectExec(
-            limit, order, child.output, planLater(child), Some(offset)) :: Nil
+            limit, order, child.output, planLater(child), offset) :: Nil
         case LimitAndOffset(IntegerLiteral(limit), IntegerLiteral(offset),
           Project(projectList, Sort(order, true, child)))
             if limit + offset < conf.topKSortFallbackThreshold =>
           TakeOrderedAndProjectExec(
-            limit, order, projectList, planLater(child), Some(offset)) :: Nil
+            limit, order, projectList, planLater(child), offset) :: Nil
         case LimitAndOffset(IntegerLiteral(limit), IntegerLiteral(offset), child) =>
-          CollectLimitExec(limit, planLater(child), Some(offset)) :: Nil
+          CollectLimitExec(limit, planLater(child), offset) :: Nil
+        case logical.Offset(IntegerLiteral(offset), child) =>
+          CollectLimitExec(child = planLater(child), offset = offset) :: Nil
         case Tail(IntegerLiteral(limit), child) =>
           CollectTailExec(limit, planLater(child)) :: Nil
         case other => planLater(other) :: Nil
@@ -116,20 +118,18 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
       // In this case we generate a physical top-K sorting operator, passing down
       // the limit and offset values to be evaluated inline during the physical
       // sorting operation for greater efficiency.
-      case LimitAndOffset(
-          IntegerLiteral(limit),
-          IntegerLiteral(offset),
-          Sort(order, true, child))
-          if limit + offset < conf.topKSortFallbackThreshold =>
-        TakeOrderedAndProjectExec(
-          limit, order, child.output, planLater(child), Some(offset)) :: Nil
-      case LimitAndOffset(
-          IntegerLiteral(limit),
-          IntegerLiteral(offset),
-          Project(projectList, Sort(order, true, child)))
+      case LimitAndOffset(IntegerLiteral(limit), IntegerLiteral(offset),
+        Sort(order, true, child)) if limit + offset < conf.topKSortFallbackThreshold =>
+          TakeOrderedAndProjectExec(
+            limit, order, child.output, planLater(child), offset) :: Nil
+      case LimitAndOffset(IntegerLiteral(limit), IntegerLiteral(offset),
+        Project(projectList, Sort(order, true, child)))
           if limit + offset < conf.topKSortFallbackThreshold =>
-        TakeOrderedAndProjectExec(limit, order, projectList, planLater(child), Some(offset)) :: Nil
-      case _ => Nil
+        TakeOrderedAndProjectExec(limit, order, projectList, planLater(child), offset) :: Nil
+      case LimitAndOffset(IntegerLiteral(limit), IntegerLiteral(offset), child) =>
+        GlobalLimitAndOffsetExec(limit, offset, planLater(child)) :: Nil
+      case _ =>
+        Nil
     }
   }
 
@@ -818,8 +818,8 @@ 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.GlobalLimitAndOffset(IntegerLiteral(limit), IntegerLiteral(offset), child) =>
-        execution.GlobalLimitAndOffsetExec(limit, offset, planLater(child)) :: Nil
+      case logical.Offset(IntegerLiteral(offset), child) =>
+        GlobalLimitAndOffsetExec(offset = offset, child = planLater(child)) :: Nil
       case union: logical.Union =>
         execution.UnionExec(union.children.map(planLater)) :: Nil
       case g @ logical.Generate(generator, _, outer, _, _, child) =>