[Enhancement] Support rewrite window function sort partition by in lowcardinality optimization

fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rewrite/AddDecodeNodeForDictStringRule.java

@@ -125,6 +125,14 @@ public DecodeContext(Map<Long, List<Integer>> tableIdToStringColumnIds, ColumnRe
             }
         }
 
+        // if column ref is an applied optimized string column, return the dictionary column.
+        // else return column ref itself
+        ColumnRefOperator getMappedOperator(ColumnRefOperator columnRef) {
+            int id = columnRef.getId();
+            Integer mapped = stringColumnIdToDictColumnIds.getOrDefault(id, id);
+            return columnRefFactory.getColumnRef(mapped);
+        }
+
         public void clear() {
             stringColumnIdToDictColumnIds.clear();
             stringFunctions.clear();
@@ -473,6 +481,14 @@ private PhysicalTopNOperator rewriteTopNOperator(PhysicalTopNOperator operator,
                     orderingList.add(orderDesc);
                 }
             }
+
+            List<ColumnRefOperator> partitionByColumns = null;
+            if (operator.getPartitionByColumns() != null) {
+                partitionByColumns =
+                        operator.getPartitionByColumns().stream().map(context::getMappedOperator)
+                                .collect(Collectors.toList());
+            }
+
             OrderSpec newOrderSpec = new OrderSpec(orderingList);
 
             ScalarOperator predicate = operator.getPredicate();
@@ -487,7 +503,7 @@ private PhysicalTopNOperator rewriteTopNOperator(PhysicalTopNOperator operator,
 
             return new PhysicalTopNOperator(newOrderSpec, operator.getLimit(),
                     operator.getOffset(),
-                    null,
+                    partitionByColumns,
                     Operator.DEFAULT_LIMIT,
                     operator.getSortPhase(),
                     operator.getTopNType(),

fe/fe-core/src/test/java/com/starrocks/sql/plan/LowCardinalityTest.java

@@ -827,7 +827,7 @@ public void testJoin() throws Exception {
         plan = getVerboseExplain(sql);
         Assert.assertFalse(plan.contains("Decode"));
 
-        sql =  "select count(*) from supplier l join [broadcast] (select max(id_int) as id_int from table_int) r on l.S_ADDRESS = r.id_int where l.S_ADDRESS not like '%key%'";
+        sql = "select count(*) from supplier l join [broadcast] (select max(id_int) as id_int from table_int) r on l.S_ADDRESS = r.id_int where l.S_ADDRESS not like '%key%'";
         plan = getVerboseExplain(sql);
         Assert.assertFalse(plan.contains("Decode"));
 
@@ -944,6 +944,24 @@ public void testGroupByWithOrderBy() throws Exception {
         connectContext.getSessionVariable().setNewPlanerAggStage(0);
     }
 
+    @Test
+    public void testAnalytic() throws Exception {
+        // Test partition by string column
+        String sql;
+        String plan;
+        // Analytic with where
+        sql = "select sum(rm) from (" +
+                "select row_number() over( partition by L_COMMENT order by L_PARTKEY) as rm from lineitem" +
+                ") t where rm < 10";
+        plan = getCostExplain(sql);
+
+        Assert.assertTrue(plan.contains("  3:SORT\n" +
+                "  |  order by: [20, INT, false] ASC, [2, INT, false] ASC"));
+        Assert.assertTrue(plan.contains("  1:PARTITION-TOP-N\n" +
+                "  |  partition by: [20: L_COMMENT, INT, false] "));
+        Assert.assertTrue(plan.contains("  |  order by: [20, INT, false] ASC, [2, INT, false] ASC"));
+    }
+
     @Test
     public void testProjectionPredicate() throws Exception {
         String sql = "select count(t.a) from(select S_ADDRESS in ('kks', 'kks2') as a from supplier) as t";