Simplify OFFSET implementation

presto-main/src/main/java/io/prestosql/sql/planner/PlanOptimizers.java

@@ -46,11 +46,7 @@
 import io.prestosql.sql.planner.iterative.rule.GatherAndMergeWindows;
 import io.prestosql.sql.planner.iterative.rule.ImplementBernoulliSampleAsFilter;
 import io.prestosql.sql.planner.iterative.rule.ImplementFilteredAggregations;
-import io.prestosql.sql.planner.iterative.rule.ImplementOffsetOverOther;
-import io.prestosql.sql.planner.iterative.rule.ImplementOffsetOverProjectSort;
-import io.prestosql.sql.planner.iterative.rule.ImplementOffsetOverProjectTopN;
-import io.prestosql.sql.planner.iterative.rule.ImplementOffsetOverSort;
-import io.prestosql.sql.planner.iterative.rule.ImplementOffsetOverTopN;
+import io.prestosql.sql.planner.iterative.rule.ImplementOffset;
 import io.prestosql.sql.planner.iterative.rule.InlineProjections;
 import io.prestosql.sql.planner.iterative.rule.MergeFilters;
 import io.prestosql.sql.planner.iterative.rule.MergeLimitOverProjectWithSort;
@@ -343,23 +339,9 @@ public PlanOptimizers(
                         estimatedExchangesCostCalculator,
                         // Temporary hack: separate optimizer step to avoid the sample node being replaced by filter before pushing
                         // it to table scan node
-                        ImmutableSet.of(new ImplementBernoulliSampleAsFilter())),
-                new IterativeOptimizer(
-                        ruleStats,
-                        statsCalculator,
-                        estimatedExchangesCostCalculator,
                         ImmutableSet.of(
-                                new ImplementOffsetOverTopN(),
-                                new ImplementOffsetOverProjectTopN(),
-                                new ImplementOffsetOverSort(),
-                                new ImplementOffsetOverProjectSort())),
-                new IterativeOptimizer(
-                        ruleStats,
-                        statsCalculator,
-                        estimatedExchangesCostCalculator,
-                        // Temporary hack: separate optimizer step to avoid a plan that's missing a SortNode over a RowNumber node
-                        // if the rules fires too early
-                        ImmutableSet.of(new ImplementOffsetOverOther())),
+                                new ImplementBernoulliSampleAsFilter(),
+                                new ImplementOffset())),
                 simplifyOptimizer,
                 new UnaliasSymbolReferences(),
                 new IterativeOptimizer(

presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/ImplementOffsetOverOther.java → presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/ImplementOffset.java

@@ -23,22 +23,19 @@
 import io.prestosql.sql.planner.plan.OffsetNode;
 import io.prestosql.sql.planner.plan.ProjectNode;
 import io.prestosql.sql.planner.plan.RowNumberNode;
-import io.prestosql.sql.planner.plan.SortNode;
-import io.prestosql.sql.planner.plan.TopNNode;
 import io.prestosql.sql.tree.ComparisonExpression;
 import io.prestosql.sql.tree.GenericLiteral;
 
 import java.util.Optional;
 
 import static io.prestosql.spi.type.BigintType.BIGINT;
 import static io.prestosql.sql.planner.plan.Patterns.offset;
-import static io.prestosql.sql.planner.plan.Patterns.source;
 
 /**
  * Transforms:
  * <pre>
  * - Offset (row count = x)
- *    - Source (other than Sort, TopN)
+ *    - Source
  * </pre>
  * Into:
  * <pre>
@@ -47,12 +44,14 @@
  *       - RowNumber
  *          - Source
  * </pre>
+ * Relies on RowNumberNode's property of keeping order of its input.
+ * If the query contains an ORDER BY clause, the sorted order
+ * will be respected when leading rows are removed.
  */
-public class ImplementOffsetOverOther
+public class ImplementOffset
         implements Rule<OffsetNode>
 {
-    private static final Pattern<OffsetNode> PATTERN = offset()
-            .with(source().matching(node -> !(node instanceof TopNNode || node instanceof SortNode)));
+    private static final Pattern<OffsetNode> PATTERN = offset();
 
     @Override
     public Pattern<OffsetNode> getPattern()