Fix

Author: iwanttobepowerful

core/src/main/java/org/apache/calcite/rex/RexWindow.java

@@ -146,52 +146,16 @@ private StringBuilder appendDigest_(StringBuilder sb, boolean allowFraming) {
         sb.append(orderKeys.get(i));
       }
     }
-    // There are 3 reasons to skip the ROWS/RANGE clause.
-    // 1. If this window is being used with a RANK-style function that does not
-    //    allow framing, or
-    // 2. If it is RANGE without ORDER BY (in which case all frames yield all
-    //    rows),
-    // 3. If it is an unbounded range
-    //    ("RANGE BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING"
-    //    or "ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING")
-    //    with no ORDER BY.
-    if (!allowFraming // 1
-        || (!isRows && orderKeys.isEmpty()) // 2
-        || (orderKeys.isEmpty()
-            && lowerBound.isUnboundedPreceding() // 3
-            && upperBound.isUnboundedFollowing())) {
-      // Don't print a ROWS or RANGE clause
-    } else if (upperBound.isCurrentRow()) {
-      // Per MSSQL: If ROWS/RANGE is specified and <window frame preceding>
-      // is used for <window frame extent> (short syntax) then this
-      // specification is used for the window frame boundary starting point and
-      // CURRENT ROW is used for the boundary ending point. For example
-      // "ROWS 5 PRECEDING" is equal to "ROWS BETWEEN 5 PRECEDING AND CURRENT
-      // ROW".
-      //
-      // We print the shorter option if it is
-      // the default. If the RexWindow is, say, "ROWS BETWEEN 5 PRECEDING AND
-      // CURRENT ROW", we output "ROWS 5 PRECEDING" because it is equivalent and
-      // is shorter.
-      if (!isRows && lowerBound.isUnboundedPreceding()) {
-        // OVER (ORDER BY x)
-        // OVER (ORDER BY x RANGE UNBOUNDED PRECEDING)
-        // OVER (ORDER BY x RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)
-        // are equivalent, so print the first (i.e. nothing).
-      } else {
-        sb.append(sb.length() > initialLength ? " " : "")
-            .append(isRows ? "ROWS" : "RANGE")
-            .append(' ')
-            .append(lowerBound);
-      }
-    } else {
+
+    if (allowFraming) {
       sb.append(sb.length() > initialLength ? " " : "")
           .append(isRows ? "ROWS" : "RANGE")
           .append(" BETWEEN ")
           .append(lowerBound)
           .append(" AND ")
           .append(upperBound);
     }
+
     if (exclude != RexWindowExclusion.EXCLUDE_NO_OTHER) {
       sb.append(" ").append(exclude).append(" ");
     }

core/src/main/java/org/apache/calcite/sql/SqlRankFunction.java

@@ -42,7 +42,7 @@ public SqlRankFunction(SqlKind kind, SqlReturnTypeInference returnTypes,
   //~ Methods ----------------------------------------------------------------
 
   @Override public boolean allowsFraming() {
-    return false;
+    return true;
   }
 
 }

core/src/main/java/org/apache/calcite/sql2rel/RelDecorrelator.java

@@ -73,6 +73,7 @@
 import org.apache.calcite.rex.RexSubQuery;
 import org.apache.calcite.rex.RexUtil;
 import org.apache.calcite.rex.RexVisitorImpl;
+import org.apache.calcite.rex.RexWindowBounds;
 import org.apache.calcite.runtime.PairList;
 import org.apache.calcite.sql.SqlAggFunction;
 import org.apache.calcite.sql.SqlExplainFormat;
@@ -1108,6 +1109,7 @@ private static void shiftMapping(Map<Integer, Integer> mapping, int startIndex,
         .over()
         .partitionBy(corVarProjects.leftList())
         .orderBy(sortExprs)
+        .let(c -> c.rowsBetween(RexWindowBounds.UNBOUNDED_PRECEDING, RexWindowBounds.CURRENT_ROW))
         .toRex();
     newProjExprs.add(rowNumberCall, "rn"); // Add the row number column
     relBuilder.project(newProjExprs.leftList(), newProjExprs.rightList());

core/src/test/java/org/apache/calcite/sql2rel/RelDecorrelatorTest.java

@@ -584,4 +584,67 @@ public static Frameworks.ConfigBuilder config() {
         + "        LogicalTableScan(table=[[scott, DEPT]])\n";
     assertThat(decorrelatedNoRules, hasTree(planDecorrelatedNoRules));
   }
+
+  @Test void testDecorrelateCorrelatedOrderByLimitToRowNumberRows() {
+    final FrameworkConfig frameworkConfig = config().build();
+    final RelBuilder builder = RelBuilder.create(frameworkConfig);
+    final RelOptCluster cluster = builder.getCluster();
+    final Planner planner = Frameworks.getPlanner(frameworkConfig);
+    final String sql = ""
+        + "SELECT dname FROM  dept WHERE 2000 > (\n"
+        + "SELECT emp.sal FROM  emp where dept.deptno = emp.deptno\n"
+        + "ORDER BY year(hiredate), emp.sal limit 1)";
+    final RelNode originalRel;
+    try {
+      final SqlNode parse = planner.parse(sql);
+      final SqlNode validate = planner.validate(parse);
+      originalRel = planner.rel(validate).rel;
+    } catch (Exception e) {
+      throw TestUtil.rethrow(e);
+    }
+
+    final HepProgram hepProgram = HepProgram.builder()
+        .addRuleCollection(
+            ImmutableList.of(
+                // SubQuery program rules
+                CoreRules.FILTER_SUB_QUERY_TO_CORRELATE,
+                CoreRules.PROJECT_SUB_QUERY_TO_CORRELATE,
+                CoreRules.JOIN_SUB_QUERY_TO_CORRELATE))
+        .build();
+    final Program program =
+        Programs.of(hepProgram, true,
+            requireNonNull(cluster.getMetadataProvider()));
+    final RelNode before =
+        program.run(cluster.getPlanner(), originalRel, cluster.traitSet(),
+            Collections.emptyList(), Collections.emptyList());
+    final String planBefore = ""
+        + "LogicalProject(DNAME=[$1])\n"
+        + "  LogicalProject(DEPTNO=[$0], DNAME=[$1], LOC=[$2])\n"
+        + "    LogicalFilter(condition=[>(2000.00, CAST($3):DECIMAL(12, 2))])\n"
+        + "      LogicalCorrelate(correlation=[$cor0], joinType=[left], requiredColumns=[{0}])\n"
+        + "        LogicalTableScan(table=[[scott, DEPT]])\n"
+        + "        LogicalProject(SAL=[$0])\n"
+        + "          LogicalSort(sort0=[$1], sort1=[$0], dir0=[ASC], dir1=[ASC], fetch=[1])\n"
+        + "            LogicalProject(SAL=[$5], EXPR$1=[EXTRACT(FLAG(YEAR), $4)])\n"
+        + "              LogicalFilter(condition=[=($cor0.DEPTNO, $7)])\n"
+        + "                LogicalTableScan(table=[[scott, EMP]])\n";
+    assertThat(before, hasTree(planBefore));
+
+    // Decorrelate without any rules, just "purely" decorrelation algorithm on RelDecorrelator
+    final RelNode after =
+        RelDecorrelator.decorrelateQuery(before, builder, RuleSets.ofList(Collections.emptyList()),
+            RuleSets.ofList(Collections.emptyList()));
+    // Verify plan
+    final String planAfter = ""
+        + "LogicalProject(DNAME=[$1])\n"
+        + "  LogicalJoin(condition=[=($0, $4)], joinType=[inner])\n"
+        + "    LogicalTableScan(table=[[scott, DEPT]])\n"
+        + "    LogicalFilter(condition=[>(2000.00, CAST($0):DECIMAL(12, 2))])\n"
+        + "      LogicalProject(SAL=[$0], DEPTNO=[$2])\n"
+        + "        LogicalFilter(condition=[<=($3, 1)])\n"
+        + "          LogicalProject(SAL=[$5], EXPR$1=[EXTRACT(FLAG(YEAR), $4)], DEPTNO=[$7], rn=[ROW_NUMBER() OVER (PARTITION BY $7 ORDER BY EXTRACT(FLAG(YEAR), $4) NULLS LAST, $5 NULLS LAST ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)])\n"
+        + "            LogicalFilter(condition=[IS NOT NULL($7)])\n"
+        + "              LogicalTableScan(table=[[scott, EMP]])\n";
+    assertThat(after, hasTree(planAfter));
+  }
 }