From b0ee37051fac6e10b139d8de7a2cbaa31820949e Mon Sep 17 00:00:00 2001 From: Mihai Budiu Date: Thu, 9 Jul 2026 11:27:29 -0700 Subject: [PATCH] [SQL] Compile SQL window ROWS queries to RANGE (ROW_NUMBER) Signed-off-by: Mihai Budiu --- docs.feldera.com/docs/sql/aggregates.md | 54 +-- .../docs/sql/unsupported-operations.md | 14 +- .../lateness_tests/test_lateness_check.py | 1 - .../optimizer/CalciteOptimizer.java | 2 + .../optimizer/RowsToRangeRule.java | 356 ++++++++++++++++++ .../sqlCompiler/compiler/sql/WindowTests.java | 137 +++++++ .../compiler/sql/quidem/FoodmartTests.java | 39 +- .../compiler/sql/quidem/WinAggPostTests.java | 8 +- .../compiler/sql/quidem/WinAggTests.java | 181 ++++++++- .../compiler/sql/simple/Regression2Tests.java | 4 +- .../compiler/sql/suites/TpcDsTest.java | 1 - .../sql/suites/nexmark/NexmarkTest.java | 2 +- .../SQL-compiler/src/test/resources/tpcds.sql | 3 +- 13 files changed, 746 insertions(+), 56 deletions(-) create mode 100644 sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/RowsToRangeRule.java diff --git a/docs.feldera.com/docs/sql/aggregates.md b/docs.feldera.com/docs/sql/aggregates.md index f57d73fbdcb..9570b1e8a5c 100644 --- a/docs.feldera.com/docs/sql/aggregates.md +++ b/docs.feldera.com/docs/sql/aggregates.md @@ -221,33 +221,6 @@ The following window aggregate functions are supported: -:::warning Potential inefficiency - -The window aggregate functions `RANK`, `DENSE_RANK`, and `ROW_NUMBER` -may be very expensive to evaluate incrementally, because it's possible -for a very small input change to produce a very large output change: -inserting or deleting a single row can change the numbering of all -subsequent rows in the same group. These functions can have a -reasonable cost in three circumstances: - -- each modified group (created by `PARTITION BY`) is relatively small in size - -- new insertions and deletions feature rows that appear towards the - end of the order produced by the `ORDER BY` clause - -- they are used in a TopK pattern with a small limit. - The topK is expressed in SQL with the following structure: - -```sql -SELECT * FROM ( - SELECT empno, - row_number() OVER (ORDER BY empno) rn - FROM empsalary) emp -WHERE rn < 3 -``` - -::: - ## Pivots The SQL `PIVOT` operation can be used to turn rows into columns. It @@ -375,6 +348,33 @@ Window aggregation functions need to store the entire collection that is being aggregated -- the space overhead is thus O(N). The work performed is expected to be O(D log N). +The window aggregate functions `RANK`, `DENSE_RANK`, and `ROW_NUMBER` +may be very expensive to evaluate incrementally, because it's possible +for a very small input change to produce a very large output change: +inserting or deleting a single row can change the numbering of all +subsequent rows in the same group. These functions can have a +reasonable cost in three circumstances: + +- each modified group (created by `PARTITION BY`) is relatively small in size + +- new insertions and deletions feature rows that appear towards the + end of the order produced by the `ORDER BY` clause + +- they are used in a TopK pattern with a small limit. + The topK is expressed in SQL with the following structure: + +```sql +SELECT * FROM ( + SELECT empno, + row_number() OVER (ORDER BY empno) rn + FROM empsalary) emp +WHERE rn < 3 +``` + +Window aggregation functions involving `ROWS BETWEEN` are implemented +behind the scenes by computing `ROW_NUMBER` for each `PARTITION BY` +group, so they inherit the cost of `ROW_NUMBER` as described above. + ### `DISTINCT` The `DISTINCT` operation can be used with an aggregation or in a diff --git a/docs.feldera.com/docs/sql/unsupported-operations.md b/docs.feldera.com/docs/sql/unsupported-operations.md index c391aca0ada..f68f8f15e82 100644 --- a/docs.feldera.com/docs/sql/unsupported-operations.md +++ b/docs.feldera.com/docs/sql/unsupported-operations.md @@ -22,9 +22,11 @@ The following aggregate functions are not supported: ### `FIRST_VALUE` and `LAST_VALUE` limited to unbounded range -`FIRST_VALUE()` and `LAST_VALUE()` are only supported for windows with -an unbounded range (e.g., `RANGE BETWEEN UNBOUNDED PRECEDING AND -CURRENT ROW`). Custom `RANGE` bounds or `ROWS` frames are not yet +`FIRST_VALUE()` and `LAST_VALUE()` are only supported for frames whose +bounds are `UNBOUNDED PRECEDING`, `CURRENT ROW`, or `UNBOUNDED +FOLLOWING` (with `RANGE` or `ROWS`): `FIRST_VALUE` requires the frame +to start at `UNBOUNDED PRECEDING`, and `LAST_VALUE` requires the frame +to end at `UNBOUNDED FOLLOWING`. Numeric bounds are not yet supported. See [#3918](https://github.com/feldera/feldera/issues/3918). @@ -34,12 +36,6 @@ Window functions using `ORDER BY` on `VARCHAR`/`STRING`, `DOUBLE`/`FLOAT` or `VARBINARY` columns are not yet supported. See [#457](https://github.com/feldera/feldera/issues/457). -### `ROWS` frame type not supported - -The `ROWS` frame specification in window functions is not yet -supported. Only `RANGE` frames are currently accepted. See -[#457](https://github.com/feldera/feldera/issues/457). This -limitation affects TPC-DS query q51. ### `EXCLUDE` clause not supported diff --git a/python/tests/runtime_aggtest/lateness_tests/test_lateness_check.py b/python/tests/runtime_aggtest/lateness_tests/test_lateness_check.py index a99ce511abb..9bab6636554 100644 --- a/python/tests/runtime_aggtest/lateness_tests/test_lateness_check.py +++ b/python/tests/runtime_aggtest/lateness_tests/test_lateness_check.py @@ -65,7 +65,6 @@ def __init__(self): ts, SUM(value) OVER (ORDER BY ts ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS rolling_sum FROM purchase""" - self.expected_error = "Not yet implemented" class lateness_interval_add(TstView): diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/CalciteOptimizer.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/CalciteOptimizer.java index 29042df2295..021d61b322e 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/CalciteOptimizer.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/CalciteOptimizer.java @@ -364,6 +364,8 @@ public String toString() { // See discussion in https://issues.apache.org/jira/browse/CALCITE-6020 CoreRules.PROJECT_TO_LOGICAL_PROJECT_AND_WINDOW )); + this.addStep(new SimpleOptimizerStep("Window ROWS to RANGE", 0, + new RowsToRangeRule())); this.addStep(new SimpleOptimizerStep("Isolate DISTINCT aggregates", 0, CoreRules.AGGREGATE_EXPAND_DISTINCT_AGGREGATES_TO_JOIN, CoreRules.AGGREGATE_EXPAND_DISTINCT_AGGREGATES diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/RowsToRangeRule.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/RowsToRangeRule.java new file mode 100644 index 00000000000..7b1637bcdf4 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/RowsToRangeRule.java @@ -0,0 +1,356 @@ +package org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.optimizer; + +import com.google.common.collect.ImmutableList; +import org.apache.calcite.plan.RelOptRuleCall; +import org.apache.calcite.plan.RelRule; +import org.apache.calcite.rel.RelCollation; +import org.apache.calcite.rel.RelCollations; +import org.apache.calcite.rel.RelFieldCollation; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.core.Window; +import org.apache.calcite.rel.logical.LogicalWindow; +import org.apache.calcite.rel.rules.TransformationRule; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.rel.type.RelDataTypeField; +import org.apache.calcite.rex.RexBuilder; +import org.apache.calcite.rex.RexInputRef; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.rex.RexShuttle; +import org.apache.calcite.rex.RexWindowBound; +import org.apache.calcite.rex.RexWindowBounds; +import org.apache.calcite.rex.RexWindowExclusion; +import org.apache.calcite.sql.SqlAggFunction; +import org.apache.calcite.sql.SqlKind; +import org.apache.calcite.sql.fun.SqlStdOperatorTable; +import org.apache.calcite.sql.parser.SqlParserPos; +import org.apache.calcite.sql.validate.SqlValidatorUtil; +import org.apache.calcite.util.ImmutableBitSet; +import org.dbsp.util.Utilities; + +import java.util.ArrayList; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * Rule that rewrites window aggregates using ROWS frames into aggregates + * using RANGE frames over an intermediate ROW_NUMBER() value. + * + *

Within a partition, ROW_NUMBER() assigns consecutive integers following + * the ORDER BY of the frame. A physical frame such as + * {@code ROWS BETWEEN 2 PRECEDING AND CURRENT ROW} therefore selects exactly + * the rows whose row number lies within {@code [rn - 2, rn]}, which is the + * logical frame {@code RANGE BETWEEN 2 PRECEDING AND CURRENT ROW} evaluated + * over the row number. (Row numbers have no duplicates, so the peer rows + * that make RANGE differ from ROWS cannot occur.) + * + *

Rewrite of a window over an input with fields $0..$n-1. SUM is + * frame-sensitive and is rewritten; ROW_NUMBER and RANK ignore the frame + * and keep the original partitioning and ordering. The quoted names are + * the variables in {@link #onMatch} that hold each plan node: + * 

+ * LogicalWindow(window#0=[window(partition {p} order by [o]    -- "window"
+ *                          rows between 2 PRECEDING and 3 FOLLOWING
+ *                          aggs [SUM($1), ROW_NUMBER(), RANK()])])
+ *                                                 -- fields $n, $n+1, $n+2
+ *   Input(fields $0..$n-1)                                     -- "input"
+ *
+ * becomes + * 
+ * LogicalProject($0, ..., $n-1, $n+2, $n, $n+1)                -- "result"
+ *   LogicalWindow(                                             -- "outer"
+ *       window#0=[window(partition {p} order by [o]
+ *                  rows between 2 PRECEDING and 3 FOLLOWING
+ *                  aggs [RANK()])],                            -- field $n+1
+ *       window#1=[window(partition {p} order by [$n]
+ *                  range between 2 PRECEDING and 3 FOLLOWING
+ *                  aggs [SUM($1)])])                           -- field $n+2
+ *     LogicalWindow(window#0=[window(partition {p} order by [o]  -- "outerInput"
+ *                              aggs [ROW_NUMBER()])])          -- field $n
+ *       Input(fields $0..$n-1)                                 -- "input"
+ *
+ * The projection restores the original aggregate order. + * The bottom window synthesizes the row number that orders SUM's RANGE + * frame; it carries the default frame, which ROW_NUMBER ignores. The + * user's ROW_NUMBER has the same partitioning and ordering, so it computes + * the same value as field $n, and the projection reuses that field instead + * of keeping the call. RANK is tie-sensitive, so it keeps the original + * ORDER BY in its own group; that group remains a ROWS group, which is + * harmless because RANK ignores the frame. + * + *

Window functions that ignore the frame (their result depends only on + * the partition and the ORDER BY) keep the original group: the ROWS flag + * has no effect on their result, and ordering by the row number instead of + * the original columns would corrupt the tie-sensitive ones (RANK and + * friends). If such calls share a group with rewritten calls, the group + * is split in two and a final projection restores the original column + * order. + */ +public class RowsToRangeRule + extends RelRule> + implements TransformationRule { + protected RowsToRangeRule() { + super(CONFIG); + } + + /** List of window functions whose result depends only on the partition and the + * ORDER BY, never on the frame. SQL semantics alone determines this + * set. Groups holding only such calls need no rewrite, because the + * ROWS flag has no effect on their result. Moreover, RANK, DENSE_RANK, + * PERCENT_RANK, and CUME_DIST are tie-sensitive: ordering by the row + * number would break ties and corrupt their result. */ + private static final EnumSet FRAME_INSENSITIVE = EnumSet.of( + SqlKind.RANK, SqlKind.DENSE_RANK, SqlKind.PERCENT_RANK, SqlKind.CUME_DIST, + SqlKind.ROW_NUMBER, SqlKind.NTILE, + SqlKind.LAG, SqlKind.LEAD); + + /** What the rule does to one window group. */ + private enum Action { + /** Leave the group unchanged. */ + NONE, + /** The frame covers the whole partition: RANGE selects the same rows + * as ROWS, so implement always using RANGE. */ + USE_RANGE, + /** Order by an intermediate ROW_NUMBER() and use a RANGE frame. */ + REWRITE + } + + /** Partitioning and ordering of a ROW_NUMBER() computed by the inner window. */ + private record OrderSpec(ImmutableBitSet keys, RelCollation orderKeys) { + static OrderSpec of(Window.Group group) { + return new OrderSpec(group.keys, group.orderKeys); + } + } + + /** True when the call's result does not depend on the window frame. */ + static boolean ignoresFrame(Window.RexWinAggCall aggCall) { + return FRAME_INSENSITIVE.contains(aggCall.getOperator().getKind()); + } + + /** Decide what the rule does to one window group. + * + *

A group is left unchanged when it does not use a ROWS frame, or + * when every call in it ignores the frame (the ROWS flag then has no + * effect on any result), or when a peer-dependent exclusion (EXCLUDE + * GROUP, EXCLUDE TIES) makes the rewrite unsound: peers are defined by + * the ORDER BY columns, and the row number ordering that the rewrite + * introduces has no peers. + * + *

A ROWS frame that spans the whole partition selects the same rows + * as the corresponding RANGE frame, so such a group only needs its ROWS + * flag cleared; its ORDER BY, and therefore any EXCLUDE clause, keeps + * its meaning. + * + *

Every other group is rewritten to a RANGE frame ordered by an + * intermediate ROW_NUMBER(). EXCLUDE CURRENT ROW is peer-independent, + * so it survives this rewrite. + * + * @param group A group of aggregate calls sharing one window frame. + * @return The action to apply to the group. + */ + static Action classify(Window.Group group) { + if (!group.isRows) + return Action.NONE; + if (group.aggCalls.stream().allMatch(RowsToRangeRule::ignoresFrame)) + // The ROWS flag has no effect on any result. + return Action.NONE; + if (group.lowerBound.isUnboundedPreceding() + && group.upperBound.isUnboundedFollowing()) + // The frame is the whole partition whether it is interpreted + // physically or logically. The ORDER BY is unchanged, so any + // EXCLUDE clause keeps its meaning. + return Action.USE_RANGE; + if (group.exclude == RexWindowExclusion.EXCLUDE_GROUP + || group.exclude == RexWindowExclusion.EXCLUDE_TIES) + // These exclusions remove peers of the current row; the row + // number ordering has no peers, which would change the excluded + // set; a rewrite would be unsound. + // (EXCLUDE CURRENT ROW is peer-independent and survives the rewrite.) + return Action.NONE; + return Action.REWRITE; + } + + @Override + public void onMatch(RelOptRuleCall call) { + final LogicalWindow window = call.rel(0); + // How to rewrite each group + final List actions = new ArrayList<>(window.groups.size()); + final List rowNumberSpecs = new ArrayList<>(); + for (Window.Group group : window.groups) { + Action action = classify(group); + actions.add(action); + if (action == Action.REWRITE && !rowNumberSpecs.contains(OrderSpec.of(group))) + rowNumberSpecs.add(OrderSpec.of(group)); + } + if (actions.stream().allMatch(a -> a == Action.NONE)) + return; + + final RelNode input = window.getInput(); + final int inputFieldCount = input.getRowType().getFieldCount(); + final RelDataTypeFactory typeFactory = window.getCluster().getTypeFactory(); + final RexBuilder rexBuilder = window.getCluster().getRexBuilder(); + final int rowNumberQueriesInserted = rowNumberSpecs.size(); + + // Inner window: one ROW_NUMBER() per distinct (partition, order), + // appended after the input fields. + final RelNode outerInput; + if (rowNumberQueriesInserted == 0) { + outerInput = input; + } else { + // The type ROW_NUMBER has under this type system. + RelDataType rowNumberType = + typeFactory.getTypeSystem().deriveRankType(typeFactory); + List rowNumberGroups = new ArrayList<>(rowNumberQueriesInserted); + RelDataTypeFactory.Builder innerType = typeFactory.builder(); + innerType.addAll(input.getRowType().getFieldList()); + Set usedNames = new HashSet<>(input.getRowType().getFieldNames()); + usedNames.addAll(window.getRowType().getFieldNames()); + for (int i = 0; i < rowNumberQueriesInserted; i++) { + Window.RexWinAggCall rowNumber = new Window.RexWinAggCall( + SqlParserPos.ZERO, SqlStdOperatorTable.ROW_NUMBER, rowNumberType, + ImmutableList.of(), i, false, false); + rowNumberGroups.add(new Window.Group( + rowNumberSpecs.get(i).keys(), false, + // ROW_NUMBER ignores the window frame, so use the default RANGE frame + RexWindowBounds.UNBOUNDED_PRECEDING, RexWindowBounds.CURRENT_ROW, + RexWindowExclusion.EXCLUDE_NO_OTHER, rowNumberSpecs.get(i).orderKeys(), + ImmutableList.of(rowNumber))); + String name = SqlValidatorUtil.uniquify( + "$row_number" + i, usedNames, SqlValidatorUtil.ATTEMPT_SUGGESTER); + innerType.add(name, rowNumberType); + } + outerInput = LogicalWindow.create(window.getTraitSet(), ImmutableList.of(), + input, ImmutableList.of(), innerType.build(), rowNumberGroups); + } + + // In a Window, a RexInputRef with index >= inputFieldCount denotes a + // constant; the row number columns push the constants up by rowNumberQueriesInserted. + final RexShuttle shiftConstants = new RexShuttle() { + @Override public RexNode visitInputRef(RexInputRef ref) { + if (ref.getIndex() >= inputFieldCount) + return new RexInputRef(ref.getIndex() + rowNumberQueriesInserted, ref.getType()); + return ref; + } + }; + + // Rebuild the groups + final List newGroups = new ArrayList<>(); + // Maps the position of each aggregate in the new window to its position in the original one. + final List origPositions = new ArrayList<>(); + final Map reusedRowNumbers = new HashMap<>(); + int origPosition = 0; + for (int g = 0; g < window.groups.size(); g++) { + Window.Group group = window.groups.get(g); + List keep = new ArrayList<>(); + List move = new ArrayList<>(); + List keepPositions = new ArrayList<>(); + List movePositions = new ArrayList<>(); + Action action = actions.get(g); + int specIndex = rowNumberSpecs.indexOf(OrderSpec.of(group)); + for (Window.RexWinAggCall aggCall : group.aggCalls) { + if (specIndex >= 0 + && aggCall.getOperator().getKind() == SqlKind.ROW_NUMBER + && !aggCall.distinct) { + // A ROW_NUMBER over the same (partition, order) as a synthesized row + // number computes the same value, so it is dropped here and the + // final projection reads the inner window's column instead. + reusedRowNumbers.put(origPosition, specIndex); + } else { + boolean moves = action == Action.REWRITE && !ignoresFrame(aggCall); + (moves ? move : keep).add(aggCall); + (moves ? movePositions : keepPositions).add(origPosition); + } + origPosition++; + } + RexWindowBound lowerBound = group.lowerBound.accept(shiftConstants); + RexWindowBound upperBound = group.upperBound.accept(shiftConstants); + if (!keep.isEmpty()) { + boolean isRows = action != Action.USE_RANGE && group.isRows; + newGroups.add(new Window.Group(group.keys, isRows, + lowerBound, upperBound, + group.exclude, group.orderKeys, + rebuildCalls(keep, keepPositions, shiftConstants, origPositions))); + } + if (!move.isEmpty()) { + RelCollation rowNumberOrder = RelCollations.of( + new RelFieldCollation(inputFieldCount + specIndex)); + newGroups.add(new Window.Group(group.keys, false, + lowerBound, upperBound, + group.exclude, rowNumberOrder, + rebuildCalls(move, movePositions, shiftConstants, origPositions))); + } + } + + final RelDataTypeFactory.Builder outerType = typeFactory.builder(); + outerType.addAll(outerInput.getRowType().getFieldList()); + for (int position : origPositions) { + RelDataTypeField field = + window.getRowType().getFieldList().get(inputFieldCount + position); + outerType.add(field.getName(), field.getType()); + } + final LogicalWindow outer = LogicalWindow.create(window.getTraitSet(), + window.getHints(), outerInput, window.constants, outerType.build(), newGroups); + if (rowNumberQueriesInserted == 0) { + // No columns were added and no group was split; the aggregates + // kept their positions. + call.transformTo(outer); + return; + } + + // Final projection: restore the original aggregate order. Reused + // ROW_NUMBERs read the inner window's row number column; the other + // row number columns are dropped. + final int aggregateCount = window.getRowType().getFieldCount() - inputFieldCount; + Utilities.enforce(origPositions.size() + reusedRowNumbers.size() == aggregateCount); + final int[] newPositions = new int[aggregateCount]; + for (int newPosition = 0; newPosition < origPositions.size(); newPosition++) + newPositions[origPositions.get(newPosition)] = newPosition; + final List projects = new ArrayList<>(); + for (int i = 0; i < inputFieldCount; i++) + projects.add(rexBuilder.makeInputRef(outer, i)); + for (int position = 0; position < aggregateCount; position++) { + Integer specIndex = reusedRowNumbers.get(position); + projects.add(rexBuilder.makeInputRef(outer, + specIndex != null + ? inputFieldCount + specIndex + : inputFieldCount + rowNumberQueriesInserted + newPositions[position])); + } + + RelNode result = call.builder() + .push(outer) + .project(projects, window.getRowType().getFieldNames()) + .build(); + call.transformTo(result); + } + + /** Copy the calls with constant references shifted and with ordinals that + * match their position in the rebuilt window; extends origPositions with + * the calls' original positions. */ + private static List rebuildCalls( + List calls, List positions, + RexShuttle shiftConstants, List origPositions) { + List result = new ArrayList<>(calls.size()); + for (int i = 0; i < calls.size(); i++) { + Window.RexWinAggCall aggCall = calls.get(i); + result.add(new Window.RexWinAggCall( + aggCall.getParserPosition(), + (SqlAggFunction) aggCall.getOperator(), + aggCall.getType(), + shiftConstants.visitList(aggCall.getOperands()), + origPositions.size(), + aggCall.distinct, + aggCall.ignoreNulls)); + origPositions.add(positions.get(i)); + } + return result; + } + + public static final DefaultOptRuleConfig CONFIG = + DefaultOptRuleConfig.create() + .withOperandSupplier( + b -> b.operand(LogicalWindow.class).anyInputs()); +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/WindowTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/WindowTests.java index 5e32b59472d..cd6d2019e3a 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/WindowTests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/WindowTests.java @@ -658,4 +658,141 @@ CREATE VIEW V AS WITH ranked AS ( e| 70 | 50 | 1 | 1 f| 60 | 60 | 1 | 1"""); } + + @Test + public void rowsFrameTests() { + // Validated on Postgres. + this.qst(""" + SELECT empno, SUM(sal) OVER ( + PARTITION BY deptno ORDER BY empno + ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS s + FROM emp; + empno | s + ------------------ + 7369 | 800.00 + 7499 | 1600.00 + 7521 | 2850.00 + 7566 | 3775.00 + 7654 | 4100.00 + 7698 | 5350.00 + 7782 | 2450.00 + 7788 | 6775.00 + 7839 | 7450.00 + 7844 | 5600.00 + 7876 | 7075.00 + 7900 | 5300.00 + 7902 | 7100.00 + 7934 | 8750.00 + (14 rows) + + SELECT empno, MIN(sal) OVER ( + PARTITION BY deptno ORDER BY sal DESC, empno + ROWS BETWEEN 1 FOLLOWING AND 2 FOLLOWING) AS m + FROM emp; + empno | m + ------------------ + 7369 | + 7499 | 1250.00 + 7521 | 950.00 + 7566 | 800.00 + 7654 | 950.00 + 7698 | 1500.00 + 7782 | 1300.00 + 7788 | 2975.00 + 7839 | 1300.00 + 7844 | 1250.00 + 7876 | 800.00 + 7900 | + 7902 | 1100.00 + 7934 | + (14 rows) + + SELECT empno, COUNT(*) OVER ( + PARTITION BY deptno ORDER BY empno + ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS c + FROM emp; + empno | c + ------------------ + 7369 | 5 + 7499 | 6 + 7521 | 6 + 7566 | 5 + 7654 | 6 + 7698 | 6 + 7782 | 3 + 7788 | 5 + 7839 | 3 + 7844 | 6 + 7876 | 5 + 7900 | 6 + 7902 | 5 + 7934 | 3 + (14 rows) + + SELECT empno, + SUM(sal) OVER (PARTITION BY deptno ORDER BY empno ROWS UNBOUNDED PRECEDING) AS s, + FIRST_VALUE(sal) OVER (PARTITION BY deptno ORDER BY empno ROWS UNBOUNDED PRECEDING) AS f + FROM emp; + empno | s | f + -------------------------- + 7369 | 800.00 | 800.00 + 7499 | 1600.00 | 1600.00 + 7521 | 2850.00 | 1600.00 + 7566 | 3775.00 | 800.00 + 7654 | 4100.00 | 1600.00 + 7698 | 6950.00 | 1600.00 + 7782 | 2450.00 | 2450.00 + 7788 | 6775.00 | 800.00 + 7839 | 7450.00 | 2450.00 + 7844 | 8450.00 | 1600.00 + 7876 | 7875.00 | 800.00 + 7900 | 9400.00 | 1600.00 + 7902 | 10875.00 | 800.00 + 7934 | 8750.00 | 2450.00 + (14 rows) + + SELECT empno, + ROW_NUMBER() OVER (PARTITION BY deptno ORDER BY empno) AS r, + SUM(sal) OVER (PARTITION BY deptno ORDER BY empno ROWS UNBOUNDED PRECEDING) AS s + FROM emp; + empno | r | s + ------------------ + 7369 | 1 | 800.00 + 7499 | 1 | 1600.00 + 7521 | 2 | 2850.00 + 7566 | 2 | 3775.00 + 7654 | 3 | 4100.00 + 7698 | 4 | 6950.00 + 7782 | 1 | 2450.00 + 7788 | 3 | 6775.00 + 7839 | 2 | 7450.00 + 7844 | 5 | 8450.00 + 7876 | 4 | 7875.00 + 7900 | 6 | 9400.00 + 7902 | 5 | 10875.00 + 7934 | 3 | 8750.00 + (14 rows) + + SELECT COUNT(*) OVER ( + PARTITION BY deptno + ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS c + FROM emp; + c + ---- + 1 + 1 + 1 + 2 + 2 + 2 + 3 + 3 + 3 + 3 + 3 + 3 + 3 + 3 + (14 rows)"""); + } } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/quidem/FoodmartTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/quidem/FoodmartTests.java index cf82c454802..523f9f6e858 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/quidem/FoodmartTests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/quidem/FoodmartTests.java @@ -4,14 +4,43 @@ import org.junit.Test; public class FoodmartTests extends FoodmartBaseTests { - @Test@Ignore("Calcite desugars qualify into WINDOW aggregate with ROWS; not yet implemented") + @Test public void testQualify() { - this.q(""" - SELECT empno, ename, deptno, job, mgr + this.qst(""" + SELECT empno, ename, deptno + FROM emp + QUALIFY ROW_NUMBER() over (partition by ename order by deptno) = 1; + EMPNO | ENAME | DEPTNO + ------------------------- + 7369 | SMITH | 20 + 7499 | ALLEN | 30 + 7521 | WARD | 30 + 7566 | JONES | 20 + 7654 | MARTIN | 30 + 7698 | BLAKE | 30 + 7782 | CLARK | 10 + 7788 | SCOTT | 20 + 7839 | KING | 10 + 7844 | TURNER | 30 + 7876 | ADAMS | 20 + 7900 | JAMES | 30 + 7902 | FORD | 20 + 7934 | MILLER | 10 + (14 rows) + + SELECT empno, ename, deptno FROM emp + WHERE deptno > 20 QUALIFY ROW_NUMBER() over (partition by ename order by deptno) = 1; - empno | ename | deptno | job | mgr - ------------------------------------"""); + EMPNO | ENAME | DEPTNO + ------------------------- + 7499 | ALLEN | 30 + 7521 | WARD | 30 + 7654 | MARTIN | 30 + 7698 | BLAKE | 30 + 7844 | TURNER | 30 + 7900 | JAMES | 30 + (6 rows)"""); } @Test diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/quidem/WinAggPostTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/quidem/WinAggPostTests.java index 946255675ea..8d2d2899d9c 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/quidem/WinAggPostTests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/quidem/WinAggPostTests.java @@ -179,8 +179,8 @@ public void test0() { | M | 20 | 1 | | M | 50 | 1 | +--------+--------+--------+ - (9 rows)"""); - this.qst(""" + (9 rows) + -- [CALCITE-1540] Support multiple columns in PARTITION BY clause of window function select gender,deptno, count(*) over (partition by gender,deptno) as count1 @@ -240,8 +240,8 @@ public void test2() { Eve | 50 | F | 50 Grace | 60 | F | 50 Wilma | | F | 50 - (9 rows)"""); - this.qst(""" + (9 rows) + select *, first_value(ename) over () from emp; ename | deptno | gender | first_value -------+--------+--------+------------- diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/quidem/WinAggTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/quidem/WinAggTests.java index d6dd7bd0dbd..64169460e34 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/quidem/WinAggTests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/quidem/WinAggTests.java @@ -35,9 +35,182 @@ public void testWindows0() { (14 rows)"""); } - @Test @Ignore("ROWS not yet implemented in WINDOW https://github.com/feldera/feldera/issues/457") - public void testWindows() { + @Test @Ignore("We don't support windows without ORDER BY spec") + public void testWindowIllegal() { this.qst(""" + -- [CALCITE-6538] OVER (ROWS CURRENT ROW) should return a window with one row, not all rows + -- (RANGE CURRENT ROW returns all rows, and is already correct.) + select ename, + sal, + sum(sal) over (rows current row) as row_sum_sal, + sum(sal) over (range current row) as range_sum_sal + from emp + where job = 'MANAGER'; + +-------+---------+-------------+---------------+ + | ENAME | SAL | ROW_SUM_SAL | RANGE_SUM_SAL | + +-------+---------+-------------+---------------+ + | BLAKE | 2850.00 | 2850.00 | 8275.00 | + | CLARK | 2450.00 | 2450.00 | 8275.00 | + | JONES | 2975.00 | 2975.00 | 8275.00 | + +-------+---------+-------------+---------------+ + (3 rows) + + select count(*) over (rows between 3 preceding and 2 preceding) as c3p2p, + sum(two) over (rows between 3 preceding and 2 preceding) as s3p2p, + sum(two) over (rows 1 preceding) as s1p, + sum(two) over (rows between 1 preceding and 1 following) as s1p1f, + sum(two) over (rows between current row and 2 following) as s2f + from (select *, 2 as two from emp) + order by 1; + +-------+-------+-----+-------+-----+ + | C3P2P | S3P2P | S1P | S1P1F | S2F | + +-------+-------+-----+-------+-----+ + | 0 | | 2 | 4 | 6 | + | 0 | | 4 | 6 | 6 | + | 1 | 2 | 4 | 6 | 6 | + | 2 | 4 | 4 | 6 | 6 | + | 2 | 4 | 4 | 6 | 6 | + | 2 | 4 | 4 | 6 | 6 | + | 2 | 4 | 4 | 6 | 6 | + | 2 | 4 | 4 | 6 | 6 | + | 2 | 4 | 4 | 6 | 6 | + | 2 | 4 | 4 | 6 | 6 | + | 2 | 4 | 4 | 6 | 6 | + | 2 | 4 | 4 | 6 | 6 | + | 2 | 4 | 4 | 6 | 4 | + | 2 | 4 | 4 | 4 | 2 | + +-------+-------+-----+-------+-----+ + (14 rows) + + -- Various combinations of ROWS, RANGE, UNBOUNDED PRECEDING, + -- UNBOUNDED FOLLOWING, and CURRENT ROW. + -- + -- 'OVER (ROWS CURRENT ROW)' includes one row; + -- + -- 'OVER (ROWS UNBOUNDED PRECEDING)' includes the previous rows; + -- + -- 'OVER ()' includes all rows and is equivalent to + -- 'OVER (ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)' and + -- 'OVER (RANGE BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)' and + -- 'OVER (RANGE UNBOUNDED PRECEDING)'. + -- Checked on Postgres. + select count(*) over (rows current row) as "rows", + count(*) over (rows unbounded preceding) as "rowsUp", + count(*) over (range unbounded preceding) as "rangeUp", + count(*) over () as "empty", + count(*) over (rows between unbounded preceding and unbounded following) as "rowsUpUf", + count(*) over (range between unbounded preceding and unbounded following) as "rangeUpUf", + count(*) over (range unbounded preceding) as "rangeUp" + from emp + order by 3; + +------+--------+---------+-------+----------+-----------+---------+ + | rows | rowsUp | rangeUp | empty | rowsUpUf | rangeUpUf | rangeUp | + +------+--------+---------+-------+----------+-----------+---------+ + | 1 | 1 | 14 | 14 | 14 | 14 | 14 | + | 1 | 2 | 14 | 14 | 14 | 14 | 14 | + | 1 | 3 | 14 | 14 | 14 | 14 | 14 | + | 1 | 4 | 14 | 14 | 14 | 14 | 14 | + | 1 | 5 | 14 | 14 | 14 | 14 | 14 | + | 1 | 6 | 14 | 14 | 14 | 14 | 14 | + | 1 | 7 | 14 | 14 | 14 | 14 | 14 | + | 1 | 8 | 14 | 14 | 14 | 14 | 14 | + | 1 | 9 | 14 | 14 | 14 | 14 | 14 | + | 1 | 10 | 14 | 14 | 14 | 14 | 14 | + | 1 | 11 | 14 | 14 | 14 | 14 | 14 | + | 1 | 12 | 14 | 14 | 14 | 14 | 14 | + | 1 | 13 | 14 | 14 | 14 | 14 | 14 | + | 1 | 14 | 14 | 14 | 14 | 14 | 14 | + +------+--------+---------+-------+----------+-----------+---------+ + (14 rows) + + -- Various combinations of ROWS, RANGE, UNBOUNDED PRECEDING, + -- UNBOUNDED FOLLOWING, and CURRENT ROW, with PARTITION BY. + -- The equivalence sets are the same the previous query (without PARTITION BY). + -- Checked on Postgres. + select deptno, + count(*) over (partition by deptno rows current row) as "rows", + count(*) over (partition by deptno rows unbounded preceding) as "rowsUp", + count(*) over (partition by deptno range unbounded preceding) as "rangeUp", + count(*) over (partition by deptno) as "empty", + count(*) over (partition by deptno rows between unbounded preceding and unbounded following) as "rowsUpUf", + count(*) over (partition by deptno range between unbounded preceding and unbounded following) as "rangeUpUf", + count(*) over (partition by deptno range unbounded preceding) as "rangeUp" + from emp + order by 1, 4; + +--------+------+--------+---------+-------+----------+-----------+---------+ + | DEPTNO | rows | rowsUp | rangeUp | empty | rowsUpUf | rangeUpUf | rangeUp | + +--------+------+--------+---------+-------+----------+-----------+---------+ + | 10 | 1 | 1 | 3 | 3 | 3 | 3 | 3 | + | 10 | 1 | 2 | 3 | 3 | 3 | 3 | 3 | + | 10 | 1 | 3 | 3 | 3 | 3 | 3 | 3 | + | 20 | 1 | 1 | 5 | 5 | 5 | 5 | 5 | + | 20 | 1 | 2 | 5 | 5 | 5 | 5 | 5 | + | 20 | 1 | 3 | 5 | 5 | 5 | 5 | 5 | + | 20 | 1 | 4 | 5 | 5 | 5 | 5 | 5 | + | 20 | 1 | 5 | 5 | 5 | 5 | 5 | 5 | + | 30 | 1 | 1 | 6 | 6 | 6 | 6 | 6 | + | 30 | 1 | 2 | 6 | 6 | 6 | 6 | 6 | + | 30 | 1 | 3 | 6 | 6 | 6 | 6 | 6 | + | 30 | 1 | 4 | 6 | 6 | 6 | 6 | 6 | + | 30 | 1 | 5 | 6 | 6 | 6 | 6 | 6 | + | 30 | 1 | 6 | 6 | 6 | 6 | 6 | 6 | + +--------+------+--------+---------+-------+----------+-----------+---------+ + (14 rows)"""); + } + + @Test + public void testWindowRows() { + this.qst(""" + -- As previous, with PARTITION BY and ORDER BY. + -- + -- 'OVER (... ROWS CURRENT ROW)' (rows) includes one row; + -- + -- 'OVER (...)' (empty), + -- 'OVER (... RANGE UNBOUNDED PRECEDING)' (rangeUp) and + -- 'OVER (... RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)' (rangeUpC) + -- are equivalent to each other, + -- + -- 'OVER (... ROWS UNBOUNDED PRECEDING)' (rowsUp) and + -- 'OVER (... ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)' (rowsUpC) + -- are equivalent to each other and include N rows, + -- and are similar to empty/rangeUp/rangeUpC except when there are ties (WARD, SCOTT); + -- + -- 'OVER (... ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)' (rowsUpUf) and + -- 'OVER (... RANGE BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)' (rangeUpUf) + -- are equivalent and return all rows. + -- + -- Checked on Postgres. + select deptno, sal, ename, + count(*) over (partition by deptno order by sal rows current row) as "rows", + count(*) over (partition by deptno order by sal) as "empty", + count(*) over (partition by deptno order by sal range unbounded preceding) as "rangeUp", + count(*) over (partition by deptno order by sal range between unbounded preceding and current row) as "rangeUpC", + count(*) over (partition by deptno order by sal rows unbounded preceding) as "rowsUp", + count(*) over (partition by deptno order by sal rows between unbounded preceding and current row) as "rowsUpC", + count(*) over (partition by deptno order by sal range between unbounded preceding and unbounded following) as "rangeUpUf", + count(*) over (partition by deptno order by sal rows between unbounded preceding and unbounded following) as "rowsUpUf" + from emp + order by 1, 2; + +--------+---------+--------+------+-------+---------+----------+--------+---------+-----------+----------+ + | DEPTNO | SAL | ENAME | rows | empty | rangeUp | rangeUpC | rowsUp | rowsUpC | rangeUpUf | rowsUpUf | + +--------+---------+--------+------+-------+---------+----------+--------+---------+-----------+----------+ + | 10 | 1300.00 | MILLER | 1 | 1 | 1 | 1 | 1 | 1 | 3 | 3 | + | 10 | 2450.00 | CLARK | 1 | 2 | 2 | 2 | 2 | 2 | 3 | 3 | + | 10 | 5000.00 | KING | 1 | 3 | 3 | 3 | 3 | 3 | 3 | 3 | + | 20 | 800.00 | SMITH | 1 | 1 | 1 | 1 | 1 | 1 | 5 | 5 | + | 20 | 1100.00 | ADAMS | 1 | 2 | 2 | 2 | 2 | 2 | 5 | 5 | + | 20 | 2975.00 | JONES | 1 | 3 | 3 | 3 | 3 | 3 | 5 | 5 | + | 20 | 3000.00 | SCOTT | 1 | 5 | 5 | 5 | 4 | 4 | 5 | 5 | + | 20 | 3000.00 | FORD | 1 | 5 | 5 | 5 | 5 | 5 | 5 | 5 | + | 30 | 950.00 | JAMES | 1 | 1 | 1 | 1 | 1 | 1 | 6 | 6 | + | 30 | 1250.00 | WARD | 1 | 3 | 3 | 3 | 2 | 2 | 6 | 6 | + | 30 | 1250.00 | MARTIN | 1 | 3 | 3 | 3 | 3 | 3 | 6 | 6 | + | 30 | 1500.00 | TURNER | 1 | 4 | 4 | 4 | 4 | 4 | 6 | 6 | + | 30 | 1600.00 | ALLEN | 1 | 5 | 5 | 5 | 5 | 5 | 6 | 6 | + | 30 | 2850.00 | BLAKE | 1 | 6 | 6 | 6 | 6 | 6 | 6 | 6 | + +--------+---------+--------+------+-------+---------+----------+--------+---------+-----------+----------+ + (14 rows) + -- Check default brackets. Note that: -- c2 and c3 are equivalent to c1; -- c5 is equivalent to c4; @@ -99,8 +272,8 @@ public void testWindows() { +-------+--------+ | 7499 | | | 7521 | 141.42 | - | 7654 | 585.95 | - | 7698 | 585.95 | + | 7654 | 585.94 | + | 7698 | 585.94 | | 7844 | 602.77 | | 7900 | 602.77 | +-------+--------+ diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression2Tests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression2Tests.java index 1b143e31cfb..39d45c06f32 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression2Tests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression2Tests.java @@ -777,7 +777,7 @@ amount DECIMAL(12, 2), @Test public void rowsTest() { - this.statementsFailingInCompilation(""" + this.getCCS(""" CREATE TABLE purchase ( ts TIMESTAMP NOT NULL, amount BIGINT, @@ -787,7 +787,7 @@ CREATE TABLE purchase ( CREATE MATERIALIZED VIEW rolling_sum AS SELECT ts, SUM(value) OVER (ORDER BY ts ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS rolling_sum - FROM purchase;""", "Not yet implemented: Window aggregates with ROWS"); + FROM purchase;"""); } @Test diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/suites/TpcDsTest.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/suites/TpcDsTest.java index 81cd4c957c2..c87384a45ae 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/suites/TpcDsTest.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/suites/TpcDsTest.java @@ -21,7 +21,6 @@ public CompilerOptions testOptions() { return options; } - // View q51 is disabled in tpcds.sql due to OVER with ROWS aggregate @Test public void compileIndividually() throws IOException { String tpcds = TestUtil.readStringFromResourceFile("tpcds.sql"); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/suites/nexmark/NexmarkTest.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/suites/nexmark/NexmarkTest.java index 30e8c3cc505..0d11b49d588 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/suites/nexmark/NexmarkTest.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/suites/nexmark/NexmarkTest.java @@ -742,7 +742,7 @@ public void q5Test() { ---------------"""); } - @Test @Ignore("OVER with ROWS") + @Test public void q6test() { this.createTest(6, """ diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/tpcds.sql b/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/tpcds.sql index c78ade34381..dffcd4da760 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/tpcds.sql +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/tpcds.sql @@ -2966,7 +2966,7 @@ order by s_store_name limit 100 ; -/* sql_51.sql +/* sql_51.sql */ CREATE VIEW sql_51 AS WITH web_v1 as ( select @@ -3011,7 +3011,6 @@ order by item_sk ,d_date limit 100 ; -*/ /* sql_52.sql */ CREATE VIEW sql_52 AS