[fix](Nereids) offset do more than once when have shuffle after limit (…

…#42576)

pick from master #42576

intro by #39316. it want to fix a problem intro by #36699.
but forgot to remove all wrong code in #36699.

after #39316, we should not set offset on exchange, when the exchange is
on the top of a limit with offset.

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

@@ -293,9 +293,6 @@ public PlanFragment visitPhysicalDistribute(PhysicalDistribute<? extends Plan> d
                     .collect(Collectors.toList());
             keys.addAll(validOutputIds);
             validOutputIds = keys;
-        } else if (child instanceof PhysicalLimit && ((PhysicalLimit<?>) child).getPhase().isGlobal()) {
-            // because sort already contains Offset, we don't need to handle PhysicalTopN
-            exchangeNode.setOffset(((PhysicalLimit<?>) child).getOffset());
         }
         if (inputFragment instanceof MultiCastPlanFragment) {
             // TODO: remove this logic when we split to multi-window in logical window to physical window conversion

regression-test/data/nereids_syntax_p0/sub_query_correlated.out

@@ -187,6 +187,15 @@
 -- !exist_corr_limit0 --
 
 -- !exist_unCorrelated_limit1_offset1 --
+1	2
+1	3
+2	4
+2	5
+3	3
+3	4
+20	2
+22	3
+24	4
 
 -- !exist_unCorrelated_limit0_offset1 --
 

regression-test/suites/load_p0/stream_load/test_csv_with_double_quotes.groovy

@@ -44,7 +44,7 @@ suite("test_csv_with_double_quotes", "p0") {
     }
 
     sql "sync"
-    qt_sql "select * from ${tableName} order by k1, k2"
+    qt_sql "select * from ${tableName} order by k1, k2, v1, v2"
 
     sql """truncate table ${tableName}"""
     streamLoad {

regression-test/suites/nereids_syntax_p0/test_limit.groovy

@@ -33,6 +33,13 @@ suite("test_limit") {
         result([[1]])
     }
 
+    test {
+        sql """
+            select * from test1 t1 join (select * from test1 limit 1 offset 1) t2
+        """
+        result([[1,1],[1,1]])
+    }
+
     sql """
     drop table if exists row_number_limit_tbl; 
     """
@@ -55,16 +62,24 @@ suite("test_limit") {
         """
     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); """
-    qt_limit1 """
-            select row_number() over(order by k6 desc) k6s, t.* from row_number_limit_tbl t order by k6s limit 1 offset 1;
+
+    test {
+        sql """
+            select row_number() over(order by k6 desc) k6s, t.* from row_number_limit_tbl t order by k6s limit 1 offset 1
         """
+        rowNum 1
+    }
 
     sql """ truncate table row_number_limit_tbl; """
     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); """
 
-    qt_lmit2 """
-            select row_number() over(order by k6 desc) k6s, t.* from row_number_limit_tbl t limit 1 offset 2;
+    test {
+        sql """
+            select row_number() over(order by k6 desc) k6s, t.* from row_number_limit_tbl t limit 1 offset 2
         """
+        rowNum 1
+    }
 }
+