[fix](Nereids) fixed the limit offset error

fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java

@@ -1814,10 +1814,38 @@ public PlanFragment visitPhysicalNestedLoopJoin(
     public PlanFragment visitPhysicalLimit(PhysicalLimit<? extends Plan> physicalLimit, PlanTranslatorContext context) {
         PlanFragment inputFragment = physicalLimit.child(0).accept(this, context);
         PlanNode child = inputFragment.getPlanRoot();
-        child.setLimit(MergeLimits.mergeLimit(physicalLimit.getLimit(), physicalLimit.getOffset(), child.getLimit()));
-        // TODO: plan node don't support limit
-        // child.setOffset(MergeLimits.mergeOffset(physicalLimit.getOffset(), child.getOffset()));
-        updateLegacyPlanIdToPhysicalPlan(child, physicalLimit);
+
+        if (physicalLimit.getPhase().isLocal()) {
+            child.setLimit(MergeLimits.mergeLimit(physicalLimit.getLimit(), physicalLimit.getOffset(),
+                    child.getLimit()));
+        } else if (physicalLimit.getPhase().isGlobal()) {
+            if (!(child instanceof ExchangeNode)) {
+                ExchangeNode exchangeNode = new ExchangeNode(context.nextPlanNodeId(), child);
+                exchangeNode.setLimit(physicalLimit.getLimit());
+                exchangeNode.setOffset(physicalLimit.getOffset());
+                exchangeNode.setPartitionType(TPartitionType.UNPARTITIONED);
+                exchangeNode.setNumInstances(1);
+
+                PlanFragment fragment = new PlanFragment(context.nextFragmentId(), exchangeNode,
+                        DataPartition.UNPARTITIONED);
+                inputFragment.setDestination(exchangeNode);
+                inputFragment.setOutputPartition(DataPartition.UNPARTITIONED);
+
+                DataStreamSink sink = new DataStreamSink(exchangeNode.getId());
+                sink.setOutputPartition(DataPartition.UNPARTITIONED);
+                inputFragment.setSink(sink);
+
+                context.addPlanFragment(fragment);
+                inputFragment = fragment;
+            } else {
+                ExchangeNode exchangeNode = (ExchangeNode) child;
+                exchangeNode.setLimit(MergeLimits.mergeLimit(physicalLimit.getLimit(), physicalLimit.getOffset(),
+                        exchangeNode.getLimit()));
+                exchangeNode.setOffset(MergeLimits.mergeOffset(physicalLimit.getOffset(), exchangeNode.getOffset()));
+            }
+        }
+
+        updateLegacyPlanIdToPhysicalPlan(inputFragment.getPlanRoot(), physicalLimit);
         return inputFragment;
     }
 

fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/PlanPostProcessors.java

@@ -62,7 +62,6 @@ public List<PlanPostProcessor> getProcessors() {
         builder.add(new RemoveUselessProjectPostProcessor());
         builder.add(new MergeProjectPostProcessor());
         builder.add(new RecomputeLogicalPropertiesProcessor());
-        builder.add(new AddOffsetIntoDistribute());
         if (cascadesContext.getConnectContext().getSessionVariable().enableAggregateCse) {
             builder.add(new ProjectAggregateExpressionsForCse());
         }

regression-test/data/nereids_syntax_p0/test_limit.out

@@ -0,0 +1,7 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !limit1 --
+2	7844	TURNER	SALESMAN	7698	1981-09-08	1500.0	0.0	30
+
+-- !lmit2 --
+3	7934	MILLER	CLERK	7782	1982-01-23	1300.0	0.0	10
+

regression-test/suites/nereids_syntax_p0/test_limit.groovy

@@ -36,4 +36,45 @@ suite("test_limit") {
         sql "select * from test1 limit 2 offset 1"
         result([[1]])
     }
+
+    sql """
+    drop table if exists row_number_limit_tbl; 
+    """
+    sql """
+            CREATE TABLE row_number_limit_tbl (
+                k1 INT NULL,
+                k2 VARCHAR(255) NULL,
+                k3 VARCHAR(255) NULL,
+                k4 INT NULL,
+                k5 VARCHAR(255) NULL,
+                k6 FLOAT NULL,
+                k7 FLOAT NULL,
+                k8 INT NULL
+                ) ENGINE=OLAP
+                DUPLICATE KEY(k1, k2)
+                DISTRIBUTED BY HASH(k1) BUCKETS 3
+                PROPERTIES (
+                    "replication_allocation" = "tag.location.default: 1"
+                );
+        """
+
+    sql """ INSERT INTO row_number_limit_tbl VALUES (7788, 'SCOTT', 'ANALYST', 7566, '1987-04-19', 3000, 0, 20); """
+    sql """ INSERT INTO row_number_limit_tbl VALUES (7844, 'TURNER', 'SALESMAN', 7698, '1981-09-08', 1500, 0, 30); """
+    sleep(1000)
+
+    qt_limit1 """
+            select row_number() over(order by k6 desc) k6s, t.* from row_number_limit_tbl t limit 1 offset 1;
+        """
+
+    sql """ truncate table row_number_limit_tbl; """
+    sleep(1000)
+
+    sql """ INSERT INTO row_number_limit_tbl VALUES (7788, 'SCOTT', 'ANALYST', 7566, '1987-04-19', 3000, 0, 20); """
+    sql """ INSERT INTO row_number_limit_tbl VALUES (7844, 'TURNER', 'SALESMAN', 7698, '1981-09-08', 1500, 0, 30); """
+    sql """ INSERT INTO row_number_limit_tbl VALUES (7934, 'MILLER', 'CLERK', 7782, '1982-01-23', 1300, 0, 10); """
+    sleep(1000)
+
+    qt_lmit2 """
+            select row_number() over(order by k6 desc) k6s, t.* from row_number_limit_tbl t limit 1 offset 2;
+        """
 }