MDEV-38970: Streaming window functions step 1

mysql-test/main/win_streaming.test

@@ -0,0 +1,328 @@
+#
+# 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 (or query's, if it's longer) 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.
+
+# Since adding --sorted_result prohibits us from testing the the actual query sorting, EXPLAIN FORMAT=JSON is used
+# to show what sort key is used for the streaming path. This is used in cases where the window order is
+# longer than the main query order, and when partition + order by are used in a window, to show that the sort key
+# uses that of the window function and that one sort is done for the whole query.
+
+# 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)
+# FORMAT=JSON to lock the sort-key shape: the partition columns are prepended to
+# the window's ORDER BY, so the single streaming sort is (a, b, pk).
+--source include/explain-no-costs.inc
+eval EXPLAIN FORMAT=JSON SELECT $q;
+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;
+--sorted_result
+eval SELECT $q;
+--sorted_result
+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;
+--sorted_result
+eval SELECT $q;
+--sorted_result
+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;
+--sorted_result
+eval SELECT $q;
+--sorted_result
+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;
+
+# User defined functions / stored procedures in the order list force materialization.
+# The outer sort key should be only 'a' and not 'a, nd(b), pk'
+DELIMITER |;
+CREATE FUNCTION nd(x INT) RETURNS INT NOT DETERMINISTIC
+BEGIN
+  RETURN x;
+END|
+DELIMITER ;|
+
+EXPLAIN FORMAT=JSON SELECT pk, a, b, rank() OVER (ORDER BY a, nd(b), pk) AS r FROM t1 ORDER BY a;
+
+DROP FUNCTION nd;
+
+DROP TABLE tg, t2;
+
+CREATE TABLE th (a INT, b INT, KEY(a, b));
+INSERT INTO th VALUES (1,1);
+INSERT INTO th VALUES (1,2);
+INSERT INTO th VALUES (2,2);
+INSERT INTO th VALUES (2,3);
+INSERT INTO th VALUES (3,3);
+INSERT INTO th VALUES (3,4);
+
+# Loose index scan is used and HAVING is pushed down to WHERE
+--let $q= SELECT a, b, rank() OVER (ORDER BY a) AS rnk FROM th GROUP BY a, b HAVING b > 1
+eval EXPLAIN FORMAT=JSON $q;
+eval $q;
+
+# Loose index scan is used and HAVING is not pushed down to WHERE
+--let $q= SELECT a, b, rank() OVER (ORDER BY a) AS rnk FROM th GROUP BY a HAVING b > 2
+eval EXPLAIN FORMAT=JSON $q;
+eval $q;
+
+DROP TABLE th;
+
+#
+# Streamable aggregate window functions: SUM / COUNT / AVG / MIN / MAX.
+#
+# Aggregate window functions are streamable for ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW (which does not require removal)
+
+--let $q= pk, a, b, count(*) OVER w AS cnt, sum(a + b) OVER w AS s, avg(b) OVER w AS av, min(b) OVER w AS mn, max(b) OVER w AS mx FROM t1 WINDOW w AS (ORDER BY b, pk ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)
+eval EXPLAIN EXTENDED SELECT $q;
+--sorted_result
+eval SELECT $q;
+--sorted_result
+eval SELECT SQL_BUFFER_RESULT $q;
+
+--let $q= pk, a, b, count(*) OVER w AS cnt, sum(b) OVER w AS s, avg(b) OVER w AS av, min(b) OVER w AS mn, max(b) OVER w AS mx FROM t1 WINDOW w AS (PARTITION BY a ORDER BY b, pk ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)
+eval EXPLAIN EXTENDED SELECT $q;
+--sorted_result
+eval SELECT $q;
+--sorted_result
+eval SELECT SQL_BUFFER_RESULT $q;
+
+#
+# Aggregate window functions that do NOT stream (still "Using temporary").
+#
+
+# 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.
+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;
+# ROWS but bottom bound past CURRENT ROW needs lookahead.
+EXPLAIN EXTENDED SELECT count(*) OVER (ORDER BY a ROWS BETWEEN UNBOUNDED PRECEDING AND 5 FOLLOWING) FROM t1;
+# ROWS but top bound is CURRENT ROW (not UNBOUNDED PRECEDING): the frame start
+# moves and removes rows, which needs random access to the departing row.
+EXPLAIN EXTENDED SELECT sum(b) OVER (ORDER BY b, pk ROWS BETWEEN CURRENT ROW AND CURRENT ROW) 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;
+
+# DISTINCT aggregates as window functions are not supported on either path (streaming / materialization).
+--error ER_NOT_SUPPORTED_YET
+SELECT sum(DISTINCT b) OVER (ORDER BY b ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) 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()) {
@@ -872,6 +873,8 @@ class Item_sum_sum :public Item_sum_num,
     return true;
   }
 
+  bool is_streamable() const override { return true; }
+
 private:
   void add_helper(bool perform_removal);
   ulonglong count;
@@ -948,6 +951,8 @@ class Item_sum_count :public Item_sum_int
     return true;
   }
 
+  bool is_streamable() const override { return true; }
+
 protected:
   Item *shallow_copy(THD *thd) const override
   { return get_item_copy<Item_sum_count>(thd, this); }
@@ -1007,6 +1012,8 @@ class Item_sum_avg :public Item_sum_sum
     return true;
   }
 
+  bool is_streamable() const override { return true; }
+
 protected:
   Item *shallow_copy(THD *thd) const override
   { return get_item_copy<Item_sum_avg>(thd, this); }
@@ -1210,6 +1217,13 @@ class Item_sum_min_max :public Item_sum_hybrid
   Field *create_tmp_field(MEM_ROOT *root, bool group, TABLE *table) override;
   void setup_caches(THD *thd) override
   { setup_hybrid(thd, arguments()[0], NULL); }
+
+  /*
+    MIN and MAX skip the creation of a Frame_scan_cursor in the case of ROWS
+    BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW, which does not require any
+    removal of rows and thus is the streaming case.
+  */
+  bool is_streamable() const override { return true; }
 };