MDEV-38970: Streaming window functions step 1

mysql-test/main/win_streaming.test

@@ -0,0 +1,266 @@
+#
+# Streaming Window Functions Tests
+#
+
+# I will remove these comments when I'm done testing, will just write the # cases I'll consider for testing here.
+
+# Explain per scenario to lock up streaming path
+# Explains for non streamable cases, no need to check output
+
+# Order by and partition
+# Explain with one function to show streamable and index or filesort is
+# used
+# Show compatible functions also stream
+# show results only to prove partition tracking correctness
+# windows reusing the main query order (whichever is longer) under its mdev
+# incompatible orders materialize
+# aggregate functions inside partition lists materialize
+# aggregate functions anywhere in the select list materialize
+# cases to look harder later (subqueries, expressions)
+
+# Test with limit and analyze to show we read only rows needed
+
+# For each streamable case we run the query twice: once as-is (streaming path)
+# and once with SQL_BUFFER_RESULT, which forces a temp table and so the old
+# materialized path. The two must agree, which is what proves the streamed
+# values are correct.
+# We wrap both in --sorted_result because the streaming path emits rows in the
+# window's sort order while the buffered path emits them from the temp table,
+# so the row order can differ even when every value matches. Sorting both and
+# selecting the key columns (pk,a,b) lets us compare them as multisets.
+
+# I add this because EXPLAIN EXTENDED emits a Note 1003 with the reconstructed query for every
+# statement, which I think is not necessary and clutters result.
+--disable_warnings
+
+CREATE TABLE t1 (pk INT PRIMARY KEY, a INT, b INT);
+INSERT INTO t1 VALUES (1, 1, 10);
+INSERT INTO t1 VALUES (2, 1, 10);
+INSERT INTO t1 VALUES (3, 1, 20);
+INSERT INTO t1 VALUES (4, 2, 20);
+INSERT INTO t1 VALUES (5, 2, 20);
+INSERT INTO t1 VALUES (6, 2, 30);
+INSERT INTO t1 VALUES (7, 3, 10);
+INSERT INTO t1 VALUES (8, 3, 30);
+INSERT INTO t1 VALUES (9, 3, 30);
+
+--let $q= pk, a, b, row_number() OVER w AS rn, rank() OVER w AS rnk, dense_rank() OVER w AS drnk FROM t1 WINDOW w AS (PARTITION BY a ORDER BY b, pk)
+eval EXPLAIN EXTENDED SELECT $q;
+eval EXPLAIN EXTENDED SELECT SQL_BUFFER_RESULT $q;
+--sorted_result
+eval SELECT $q;
+--sorted_result
+eval SELECT SQL_BUFFER_RESULT $q;
+
+# Tests peer handling as duplicates exist when pk is not in the order list.
+# row_number() is dropped here because the number assigned to a given row can differ between streaming and materialization.
+# Filesort is not stable and there is no tie breaking.
+--let $q= pk, a, b, rank() OVER w AS rnk, dense_rank() OVER w AS drnk FROM t1 WINDOW w AS (ORDER BY a)
+eval EXPLAIN EXTENDED SELECT $q;
+--sorted_result
+eval SELECT $q;
+--sorted_result
+eval SELECT SQL_BUFFER_RESULT $q;
+
+# Order-only (no partition): all three functions, total order.
+--let $q= pk, a, row_number() OVER w AS rn, rank() OVER w AS rnk, dense_rank() OVER w AS drnk FROM t1 WINDOW w AS (ORDER BY a, pk)
+eval EXPLAIN EXTENDED SELECT $q;
+--sorted_result
+eval SELECT $q;
+--sorted_result
+eval SELECT SQL_BUFFER_RESULT $q;
+
+# Windows reusing the main query order (whichever order is longer is used for a
+# single sort). We use EXPLAIN FORMAT=JSON here to show which sort key is used.
+# window order longer than main ORDER BY
+--source include/explain-no-costs.inc
+--let $q= pk, a, b, rank() OVER (ORDER BY a, b) AS rnk FROM t1 ORDER BY a
+eval EXPLAIN FORMAT=JSON SELECT $q;
+eval SELECT $q;
+eval SELECT SQL_BUFFER_RESULT $q;
+
+# main ORDER BY longer than window order
+--source include/explain-no-costs.inc
+--let $q= pk, a, b, rank() OVER (ORDER BY a) AS rnk FROM t1 ORDER BY a, b
+eval EXPLAIN FORMAT=JSON SELECT $q;
+eval SELECT $q;
+eval SELECT SQL_BUFFER_RESULT $q;
+
+# window order equals main ORDER BY
+--source include/explain-no-costs.inc
+--let $q= pk, a, rank() OVER (ORDER BY a) AS rnk FROM t1 ORDER BY a
+eval EXPLAIN FORMAT=JSON SELECT $q;
+eval SELECT $q;
+eval SELECT SQL_BUFFER_RESULT $q;
+
+# Window functions inside expressions still stream as long as they're not
+# aggregate functions that already require materialization.
+--let $q= pk, a, rank() OVER w AS r, rank() OVER w + 1 AS r_plus, rank() OVER w - dense_rank() OVER w AS diff FROM t1 WINDOW w AS (ORDER BY a)
+eval EXPLAIN EXTENDED SELECT $q;
+--sorted_result
+eval SELECT $q;
+--sorted_result
+eval SELECT SQL_BUFFER_RESULT $q;
+
+# Multi-table joins
+CREATE TABLE t2 (pk INT PRIMARY KEY, c INT);
+INSERT INTO t2 VALUES (1, 100);
+INSERT INTO t2 VALUES (2, 200);
+INSERT INTO t2 VALUES (3, 300);
+INSERT INTO t2 VALUES (4, 400);
+INSERT INTO t2 VALUES (5, 500);
+INSERT INTO t2 VALUES (6, 600);
+INSERT INTO t2 VALUES (7, 700);
+INSERT INTO t2 VALUES (8, 800);
+INSERT INTO t2 VALUES (9, 900);
+
+--let $q= t1.pk, t1.a, t1.b, rank() OVER (ORDER BY t1.b, t1.pk) AS rnk FROM t1 JOIN t2 ON t1.pk = t2.pk
+eval EXPLAIN EXTENDED SELECT $q;
+--sorted_result
+eval SELECT $q;
+--sorted_result
+eval SELECT SQL_BUFFER_RESULT $q;
+
+--let $q= t1.pk, t1.a, t1.b, rank() OVER (ORDER BY t1.b, t1.pk) AS rnk FROM t1 LEFT JOIN t2 ON t1.pk = t2.pk
+eval EXPLAIN EXTENDED SELECT $q;
+--sorted_result
+eval SELECT $q;
+--sorted_result
+eval SELECT SQL_BUFFER_RESULT $q;
+
+DROP TABLE t2;
+
+# Derived table in FROM: window functions in the outer query over a subquery.
+--let $q= d.pk, d.a, d.b, rank() OVER (PARTITION BY d.a ORDER BY d.b) AS rnk, dense_rank() OVER (PARTITION BY d.a ORDER BY d.b) AS drnk FROM (SELECT pk, a, b FROM t1 WHERE a > 1) AS d
+eval EXPLAIN EXTENDED SELECT $q;
+--sorted_result
+eval SELECT $q;
+--sorted_result
+eval SELECT SQL_BUFFER_RESULT $q;
+
+# Window function INSIDE a subquery
+EXPLAIN EXTENDED SELECT d.a, d.rnk FROM (SELECT a, rank() OVER (ORDER BY a) AS rnk FROM t1) AS d;
+SELECT d.a, d.rnk FROM (SELECT a, rank() OVER (ORDER BY a) AS rnk FROM t1) AS d;
+
+# r_rows should be equal to the limit.
+--source include/analyze-format.inc
+ANALYZE FORMAT=JSON SELECT pk, rank() OVER (ORDER BY pk) AS rnk FROM t1 LIMIT 2;
+
+# GROUP BY can stream: when the rows come out of the join already 
+# grouped (in the case of a single table loose index scan), and 
+# if the window order is compatible with the group list the window
+# functions are computed on the streamed grouped rows.
+# Note that this applies even if the window function order list is longer than the group list.
+# As long as the group list is a prefix of the longest window order list.
+CREATE TABLE tg (a INT, b INT, KEY(a, b));
+INSERT INTO tg VALUES (1, 1);
+INSERT INTO tg VALUES (1, 2);
+INSERT INTO tg VALUES (2, 1);
+INSERT INTO tg VALUES (2, 2);
+INSERT INTO tg VALUES (2, 3);
+INSERT INTO tg VALUES (3, 1);
+
+# GROUP BY longer than the window order
+--let $q= a, b, rank() OVER (ORDER BY a) AS rnk, dense_rank() OVER (ORDER BY a) AS drnk FROM tg GROUP BY a, b
+eval EXPLAIN EXTENDED SELECT $q;
+--sorted_result
+eval SELECT $q;
+--sorted_result
+eval SELECT SQL_BUFFER_RESULT $q;
+
+# Group by shorter than the window order
+--let $q= a, rank() OVER (ORDER BY a, b) AS rnk, dense_rank() OVER (ORDER BY a, b) AS drnk FROM tg GROUP BY a
+eval EXPLAIN EXTENDED SELECT $q;
+--sorted_result
+eval SELECT $q;
+--sorted_result
+eval SELECT SQL_BUFFER_RESULT $q;
+
+# Loose index scan with partition
+--let $q= a, b, row_number() OVER w AS rn, rank() OVER w AS rnk, dense_rank() OVER w AS drnk FROM tg GROUP BY a, b WINDOW w AS (PARTITION BY a ORDER BY b)
+eval EXPLAIN EXTENDED SELECT $q;
+--sorted_result
+eval SELECT $q;
+--sorted_result
+eval SELECT SQL_BUFFER_RESULT $q;
+
+#
+# Cases that fall back to materialization
+#
+
+# Incompatible orders between the two functions
+EXPLAIN EXTENDED SELECT rank() OVER (ORDER BY a), rank() OVER (ORDER BY b) FROM t1;
+
+# Aggregate inside the PARTITION BY list
+EXPLAIN EXTENDED SELECT rank() OVER (PARTITION BY max(a) ORDER BY b) FROM t1;
+
+# A non-window aggregate anywhere in the select list
+EXPLAIN EXTENDED SELECT max(a), rank() OVER (ORDER BY b) FROM t1;
+
+# GROUP BY with no usable index needs a temp table for the grouping, materialize
+EXPLAIN EXTENDED SELECT rank() OVER (ORDER BY a) FROM t1 GROUP BY a;
+
+CREATE TABLE t2 (a INT, x INT, KEY(a));
+INSERT INTO t2 VALUES (1,10);
+INSERT INTO t2 VALUES (1,20);
+INSERT INTO t2 VALUES (2,20);
+INSERT INTO t2 VALUES (2,30);
+
+# GROUP BY across a multi-table join
+EXPLAIN SELECT tg.a, rank() OVER (ORDER BY tg.a) FROM tg JOIN t2 ON tg.a=t2.a GROUP BY tg.a;
+
+# GROUP BY that uses a tight index scan (select list is not satisfied by the index)
+EXPLAIN SELECT a, x  FROM t2 FORCE INDEX FOR GROUP BY (a) GROUP BY a;
+EXPLAIN SELECT a, x, rank() OVER (ORDER BY a) FROM t2 FORCE INDEX FOR GROUP BY (a) GROUP BY a;
+
+# Implicit/constant grouping (GROUP BY a constant expression) collapses to a
+# single group (grouping optimized away) and does not stream
+EXPLAIN EXTENDED SELECT rank() OVER (ORDER BY a) FROM tg GROUP BY 1+2;
+
+# GROUP BY on a unique NOT NULL index (here the PRIMARY KEY) is optimized away:
+# every group is exactly one row, so no grouping operation is performed and the
+# rows go straight through end_send() with no temp table (the GROUP BY is just
+# rewritten to an ORDER BY). Shown here without a window function:
+EXPLAIN EXTENDED SELECT pk, a, b FROM t1 GROUP BY pk;
+# Adding a window function disables that unique-index optimization for not 
+# so the GROUP BY is kept. (This is not yet fixed for streaming)
+EXPLAIN EXTENDED SELECT pk, a, rank() OVER (ORDER BY pk) AS rnk FROM t1 GROUP BY pk;
+
+# Ordering the outer query by the window function value, it needs to save the value first.
+EXPLAIN EXTENDED SELECT pk, rank() OVER (ORDER BY a) AS x FROM t1 ORDER BY x;
+
+DROP TABLE tg, t2;
+
+#
+# Aggregate window functions: not streamable yet.
+#
+# Aggregates return is_streamable() == false today, so every case below falls
+# back and shows "Using temporary". They split into two groups: the cumulative
+# running frame is only blocked by that is_streamable() gate and is intended to
+# stream once aggregates are enabled; the rest fall back for frame reasons that
+# will remain even then.
+
+# Will stream once aggregates are enabled: an explicit ROWS ... CURRENT ROW
+# frame makes "current row" a physical position, so row N depends only on rows
+# <= N and an O(1) running accumulator is enough. The EXPLAIN EXTENDED will then
+# show "Using filesort" and NOT "Using temporary".
+EXPLAIN EXTENDED SELECT count(*) OVER w, sum(b) OVER w, avg(b) OVER w, min(b) OVER w, max(b) OVER w FROM t1 WINDOW w AS (ORDER BY b, pk ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW);
+
+# Default-frame: ORDER BY with no explicit frame defaults to RANGE, not ROWS.
+# Under RANGE "current row" is the end of the peer group, so a row's value is
+# unknown until the group is scanned ahead and buffered. Looks like the
+# streamable ROWS form above but will not stream.
+EXPLAIN EXTENDED SELECT sum(b) OVER (ORDER BY b) FROM t1;
+# Explicit RANGE, same peer-group reason.
+EXPLAIN EXTENDED SELECT count(*) OVER (ORDER BY a RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) FROM t1;
+# Bottom bound past CURRENT ROW needs lookahead regardless of units.
+EXPLAIN EXTENDED SELECT count(*) OVER (ORDER BY a RANGE BETWEEN CURRENT ROW AND 5 FOLLOWING) FROM t1;
+
+# Those span whole partitions, the whole partition must be buffered before the row is emitted,
+# hence no streaming.
+EXPLAIN EXTENDED SELECT sum(b) OVER () FROM t1;
+EXPLAIN EXTENDED SELECT sum(b) OVER (PARTITION BY a) FROM t1;
+
+DROP TABLE t1;
+
+--enable_warnings

sql/item_sum.h

@@ -418,6 +418,7 @@ class Item_sum :public Item_func_or_sum
   Item_sum(THD *thd, Item_sum *item);
   enum Type type() const override { return SUM_FUNC_ITEM; }
   virtual enum Sumfunctype sum_func () const=0;
+  virtual inline bool is_streamable() const { return false; }
   bool is_aggr_sum_func()
   {
     switch (sum_func()) {

sql/item_windowfunc.h

@@ -150,6 +150,8 @@ class Item_sum_row_number: public Item_sum_int
     return name;
   }
 
+  inline bool is_streamable() const override { return true; }
+
 protected:
   Item *shallow_copy(THD *thd) const override
   { return get_item_copy<Item_sum_row_number>(thd, this); }
@@ -215,6 +217,8 @@ class Item_sum_rank: public Item_sum_int
     return name;
   }
 
+  inline bool is_streamable() const override { return true; }
+
   void setup_window_func(THD *thd, Window_spec *window_spec) override;
 
   void cleanup() override
@@ -290,6 +294,8 @@ class Item_sum_dense_rank: public Item_sum_int
     return name;
   }
 
+  inline bool is_streamable() const override { return true; }
+
   void setup_window_func(THD *thd, Window_spec *window_spec) override;
 
   void cleanup() override

sql/sql_lex.cc

@@ -3793,7 +3793,7 @@ uint st_select_lex::get_cardinality_of_ref_ptrs_slice(uint order_group_num_arg)
           select_n_where_fields * winfunc_factor +
           order_group_num * 2 * winfunc_factor +
           hidden_bit_fields +
-          fields_in_window_functions + 1;
+          fields_in_window_functions + 1; // consider this case for streaming
   return n;
 }