[OPTIQ-311] Wrong results when filtering the results of windowed aggregation

core/src/main/java/org/eigenbase/rel/rules/CalcRelSplitter.java

@@ -86,6 +86,9 @@ public abstract class CalcRelSplitter {
     this.typeFactory = calc.getCluster().getTypeFactory();
     this.child = calc.getChild();
     this.relTypes = relTypes;
+    assert "CalcRelType".equals(relTypes[0].name)
+        : "The first RelType should be CalcRelType for proper RexLiteral"
+        + " implementation at the last level, got " + relTypes[0].name;
   }
 
   //~ Methods ----------------------------------------------------------------
@@ -101,11 +104,11 @@ RelNode execute() {
     assert !RexUtil.containComplexExprs(exprList);
 
     // Figure out what level each expression belongs to.
-    int[] exprLevels = new int[exprs.length];
+    int[] exprLevels = new int[exprs.length + 1];
 
     // The reltype of a level is given by
     // relTypes[levelTypeOrdinals[level]].
-    int[] levelTypeOrdinals = new int[exprs.length];
+    int[] levelTypeOrdinals = new int[exprs.length + 1];
 
     int levelCount = chooseLevels(exprs, -1, exprLevels, levelTypeOrdinals);
 
@@ -249,7 +252,9 @@ private int chooseLevels(
       int[] levelTypeOrdinals) {
     final int inputFieldCount = program.getInputRowType().getFieldCount();
 
-    int levelCount = 0;
+    // Ensure the first level is CalcRelType so the input projection
+    // trims unnecessary fields.
+    int levelCount = 1;
     final MaxInputFinder maxInputFinder = new MaxInputFinder(exprLevels);
     boolean[] relTypesPossibleForTopLevel = new boolean[relTypes.length];
     Arrays.fill(relTypesPossibleForTopLevel, true);
@@ -371,9 +376,6 @@ private int chooseLevels(
     if (levelCount > 0) {
       // The latest level should be CalcRelType otherwise literals cannot be
       // implemented.
-      assert "CalcRelType".equals(relTypes[0].name)
-          : "The first RelType should be CalcRelType for proper RexLiteral"
-          + " implementation at the last level, got " + relTypes[0].name;
       if (levelTypeOrdinals[levelCount - 1] != 0) {
         levelCount++;
       }

core/src/main/java/org/eigenbase/rel/rules/PushFilterPastProjectRule.java

@@ -36,9 +36,17 @@ public class PushFilterPastProjectRule extends RelOptRule {
    */
   private PushFilterPastProjectRule() {
     super(
-        operand(
-            FilterRel.class,
-            operand(ProjectRel.class, any())));
+        operand(FilterRel.class,
+            some(new RelOptRuleOperand(ProjectRel.class, null, any()) {
+              @Override
+              public boolean matches(RelNode rel) {
+                // Avoid pushing filters through windowed aggregates
+                // TODO: Implement real filter push that considers partition by
+                return super.matches(rel)
+                    && !RexOver.containsOver(((ProjectRel) rel).getProjects(),
+                        null);
+              }
+            })));
   }
 
   //~ Methods ----------------------------------------------------------------

core/src/test/java/net/hydromatic/optiq/test/JdbcTest.java

@@ -3028,6 +3028,72 @@ private static ImmutableMultimap<Class, Integer> x() {
             "M=10002.0");
   }
 
+  /**
+   * Tests if optiq can push filters through WindowRel when filter matches
+   * partition by.
+   */
+  @Test
+  public void testWinAggWithFilter() {
+    OptiqAssert.that()
+        .with(OptiqAssert.Config.REGULAR)
+        .query(
+            "select * from (select \"empid\", \"deptno\",\n"
+            + "count(*) over () c\n"
+            + "from \"hr\".\"emps\"\n"
+            + ") where \"deptno\"=10 and \"empid\"=100")
+        .explainContains(
+            "EnumerableCalcRel(expr#0..2=[{inputs}], expr#3=[CAST($t1):INTEGER NOT NULL], expr#4=[10], expr#5=[=($t3, $t4)], expr#6=[CAST($t0):INTEGER NOT NULL], expr#7=[100], expr#8=[=($t6, $t7)], expr#9=[AND($t5, $t8)], proj#0..2=[{exprs}], $condition=[$t9])\n"
+            + "  EnumerableWindowRel(window#0=[window(partition {} order by [] range between UNBOUNDED PRECEDING and UNBOUNDED FOLLOWING aggs [COUNT()])])\n"
+            + "    EnumerableCalcRel(expr#0..4=[{inputs}], proj#0..1=[{exprs}])\n"
+            + "      EnumerableTableAccessRel(table=[[hr, emps]])")
+        .returns("empid=100; deptno=10; C=4\n");
+    // There are 4 employees in total
+  }
+
+  /**
+   * Tests that optiq does not push filter through WindowRel when it is
+   * not allowed.
+   */
+  @Test
+  public void testWinAggWithFilterPush() {
+    OptiqAssert.that()
+        .with(OptiqAssert.Config.REGULAR)
+        .query(
+            "select * from (select \"empid\", \"deptno\",\n"
+            + " count(*) over (partition by \"deptno\") c\n"
+            + "from \"hr\".\"emps\"\n"
+            + ") where \"deptno\"=10 and \"empid\"=100")
+        .explainContains(
+            "EnumerableCalcRel(expr#0..2=[{inputs}], expr#3=[CAST($t1):INTEGER NOT NULL], expr#4=[10], expr#5=[=($t3, $t4)], expr#6=[CAST($t0):INTEGER NOT NULL], expr#7=[100], expr#8=[=($t6, $t7)], expr#9=[AND($t5, $t8)], proj#0..2=[{exprs}], $condition=[$t9])\n"
+            + "  EnumerableWindowRel(window#0=[window(partition {1} order by [] range between UNBOUNDED PRECEDING and UNBOUNDED FOLLOWING aggs [COUNT()])])\n"
+            + "    EnumerableCalcRel(expr#0..4=[{inputs}], proj#0..1=[{exprs}])\n"
+            + "      EnumerableTableAccessRel(table=[[hr, emps]])")
+        .returns("empid=100; deptno=10; C=3\n");
+    // There are 3 employees in department 10
+  }
+
+  /**
+   * Tests if optiq does not evaluate filters early as a part of input
+   * calculation for the window aggregate.
+   */
+  @Test
+  public void testWinAggWithFilterCalcFirst() {
+    OptiqAssert.that()
+        .with(OptiqAssert.Config.REGULAR)
+        .query(
+            "select * from (select \"empid\", \"deptno\",\n"
+                + "count(*) over (partition by \"deptno\"*0) c\n"
+                + "from \"hr\".\"emps\"\n"
+                + ") where \"deptno\"=10 and \"empid\"=100")
+        .explainContains(
+            "EnumerableCalcRel(expr#0..3=[{inputs}], expr#4=[CAST($t1):INTEGER NOT NULL], expr#5=[10], expr#6=[=($t4, $t5)], expr#7=[CAST($t0):INTEGER NOT NULL], expr#8=[100], expr#9=[=($t7, $t8)], expr#10=[AND($t6, $t9)], proj#0..1=[{exprs}], $2=[$t3], $condition=[$t10])\n"
+            + "  EnumerableWindowRel(window#0=[window(partition {2} order by [] range between UNBOUNDED PRECEDING and UNBOUNDED FOLLOWING aggs [COUNT()])])\n"
+            + "    EnumerableCalcRel(expr#0..4=[{inputs}], expr#5=[0], expr#6=[*($t1, $t5)], proj#0..1=[{exprs}], $2=[$t6])\n"
+            + "      EnumerableTableAccessRel(table=[[hr, emps]])")
+        .returns("empid=100; deptno=10; C=4\n");
+    // There are 4 employees in total
+  }
+
   /** Tests for RANK and ORDER BY ... DESCENDING, NULLS FIRST, NULLS LAST. */
   @Test public void testWinAggRank() {
     OptiqAssert.that()

core/src/test/java/org/eigenbase/test/RelOptRulesTest.java

@@ -95,6 +95,14 @@ protected DiffRepository getDiffRepos() {
         + " where d.name = 'Charlie'");
   }
 
+  @Test public void testPushFilterThroughWindow() {
+    // TODO: implement true push filter through window
+    checkPlanning(
+        PushFilterPastProjectRule.INSTANCE,
+        "select * from (select e.ename, e.deptno+e.empno qq, row_number() over (partition by e.deptno+e.empno order by e.empno) r from sales.emp e) where qq=10"
+    );
+  }
+
   @Test public void testReduceAverage() {
     checkPlanning(
         ReduceAggregatesRule.INSTANCE,

core/src/test/resources/org/eigenbase/test/RelOptRulesTest.xml

@@ -89,6 +89,27 @@ ProjectRel(EXPR$0=[1])
     FilterRel(condition=[=($1, 'Charlie')])
       TableAccessRel(table=[[CATALOG, SALES, DEPT]])
     TableAccessRel(table=[[CATALOG, SALES, EMP]])
+]]>
+        </Resource>
+    </TestCase>
+    <TestCase name="testPushFilterThroughWindow">
+        <Resource name="sql">
+            <![CDATA[select * from (select e.ename, e.deptno+e.empno qq, row_number() over (partition by e.deptno+e.empno order by e.empno) from sales.emp e) where qq=10]]>
+        </Resource>
+        <Resource name="planBefore">
+            <![CDATA[
+ProjectRel(ENAME=[$0], QQ=[$1], R=[$2])
+  FilterRel(condition=[=($1, 10)])
+    ProjectRel(ENAME=[$1], QQ=[+($7, $0)], R=[ROW_NUMBER() OVER (PARTITION BY +($7, $0) ORDER BY $0 RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)])
+      TableAccessRel(table=[[CATALOG, SALES, EMP]])
+]]>
+        </Resource>
+        <Resource name="planAfter">
+            <![CDATA[
+ProjectRel(ENAME=[$0], QQ=[$1], R=[$2])
+  FilterRel(condition=[=($1, 10)])
+    ProjectRel(ENAME=[$1], QQ=[+($7, $0)], R=[ROW_NUMBER() OVER (PARTITION BY +($7, $0) ORDER BY $0 RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)])
+      TableAccessRel(table=[[CATALOG, SALES, EMP]])
 ]]>
         </Resource>
     </TestCase>