diff --git a/mysql-test/main/streaming_wf.result b/mysql-test/main/streaming_wf.result new file mode 100644 index 0000000000000..5154eccfc20f9 --- /dev/null +++ b/mysql-test/main/streaming_wf.result @@ -0,0 +1,276 @@ +DROP TABLE IF EXISTS t1, t2, t_multi; +CREATE TABLE t1 (id INT PRIMARY KEY AUTO_INCREMENT, val INT); +INSERT INTO t1 (val) VALUES (10),(20),(30),(40),(50),(60),(70),(80),(90),(100); +CREATE TABLE t2 (id INT PRIMARY KEY AUTO_INCREMENT, grp INT, INDEX idx_grp(grp)); +INSERT INTO t2 (grp) VALUES (1),(1),(2),(2),(3),(3); +CREATE TABLE t_multi ( +id INT PRIMARY KEY AUTO_INCREMENT, +a INT, +b INT, +c INT, +INDEX idx_a_b_c (a, b, c), +INDEX idx_b (b) +); +INSERT INTO t_multi (a, b, c) VALUES +(1,1,1),(1,1,2),(1,2,1),(1,2,2), +(2,1,1),(2,1,2),(2,2,1),(2,2,2); +SELECT row_number() OVER () AS rn, val FROM t1 LIMIT 5; +rn val +1 10 +2 20 +3 30 +4 40 +5 50 +FLUSH STATUS; +SELECT row_number() OVER () AS rn FROM t1 LIMIT 3; +rn +1 +2 +3 +SHOW STATUS LIKE 'Created_tmp_tables'; +Variable_name Value +Created_tmp_tables 0 +SELECT row_number() OVER (ORDER BY id) AS rn, id FROM t1 LIMIT 5; +rn id +1 1 +2 2 +3 3 +4 4 +5 5 +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY id) AS rn FROM t1 LIMIT 3; +rn +1 +2 +3 +SHOW STATUS LIKE 'Created_tmp_tables'; +Variable_name Value +Created_tmp_tables 0 +EXPLAIN SELECT row_number() OVER () AS rn FROM t1 LIMIT 5; +id select_type table type possible_keys key key_len ref rows Extra +1 SIMPLE t1 index NULL PRIMARY 4 NULL 10 Using index +EXPLAIN SELECT row_number() OVER (ORDER BY id) AS rn FROM t1 LIMIT 5; +id select_type table type possible_keys key key_len ref rows Extra +1 SIMPLE t1 index NULL PRIMARY 4 NULL 10 Using index +SELECT rank() OVER (ORDER BY id) AS rnk, id FROM t1 LIMIT 5; +rnk id +1 1 +2 2 +3 3 +4 4 +5 5 +FLUSH STATUS; +SELECT rank() OVER (ORDER BY id) AS rnk FROM t1 LIMIT 3; +rnk +1 +2 +3 +SHOW STATUS LIKE 'Created_tmp_tables'; +Variable_name Value +Created_tmp_tables 0 +SELECT dense_rank() OVER (ORDER BY id) AS dr, id FROM t1 LIMIT 5; +dr id +1 1 +2 2 +3 3 +4 4 +5 5 +FLUSH STATUS; +SELECT dense_rank() OVER (ORDER BY id) AS dr FROM t1 LIMIT 3; +dr +1 +2 +3 +SHOW STATUS LIKE 'Created_tmp_tables'; +Variable_name Value +Created_tmp_tables 0 +SELECT 1 + row_number() OVER () AS rn FROM t1 LIMIT 3; +rn +2 +3 +4 +SELECT +row_number() OVER () AS rn1, +row_number() OVER () AS rn2 +FROM t1 LIMIT 3; +rn1 rn2 +1 1 +2 2 +3 3 +SELECT * FROM ( +SELECT row_number() OVER () AS rn, val FROM t1 +) sub WHERE rn <= 3; +rn val +1 10 +2 20 +3 30 +SELECT rn, val FROM ( +SELECT row_number() OVER (ORDER BY id) AS rn, val, id FROM t1 +) sub WHERE rn BETWEEN 2 AND 4; +rn val +2 20 +3 30 +4 40 +FLUSH STATUS; +SELECT row_number() OVER () AS rn, t1.val, t2.grp +FROM t1 JOIN t2 ON t1.id = t2.id LIMIT 3; +rn val grp +1 10 1 +2 20 1 +3 30 2 +SHOW STATUS LIKE 'Created_tmp_tables'; +Variable_name Value +Created_tmp_tables 0 +SELECT row_number() OVER () AS rn, t1.id, t2.grp +FROM t1 LEFT JOIN t2 ON t1.id = t2.id LIMIT 3; +rn id grp +1 1 1 +2 2 1 +3 3 2 +SELECT row_number() OVER () AS rn, val FROM t1 WHERE val > 50 LIMIT 3; +rn val +1 60 +2 70 +3 80 +SELECT row_number() OVER (ORDER BY id) AS rn, id, val FROM t1 WHERE id > 5 LIMIT 3; +rn id val +1 6 60 +2 7 70 +3 8 80 +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY a, b, c) AS rn, +rank() OVER (ORDER BY a, b, c) AS rnk +FROM t_multi LIMIT 5; +rn rnk +1 1 +2 2 +3 3 +4 4 +5 5 +SHOW STATUS LIKE 'Created_tmp_tables'; +Variable_name Value +Created_tmp_tables 0 +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY a, b, c) AS rn, +rank() OVER (ORDER BY a, b) AS rnk +FROM t_multi LIMIT 5; +rn rnk +1 1 +2 1 +3 3 +4 3 +5 5 +SHOW STATUS LIKE 'Created_tmp_tables'; +Variable_name Value +Created_tmp_tables 0 +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY a, b, c) AS rn, +rank() OVER (ORDER BY a, b) AS rnk, +dense_rank() OVER (ORDER BY a) AS dr +FROM t_multi LIMIT 5; +rn rnk dr +1 1 1 +2 1 1 +3 3 1 +4 3 1 +5 5 2 +SHOW STATUS LIKE 'Created_tmp_tables'; +Variable_name Value +Created_tmp_tables 0 +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY a, b) AS rn, +row_number() OVER () AS rnk +FROM t_multi LIMIT 5; +rn rnk +1 1 +2 2 +3 3 +4 4 +5 5 +SHOW STATUS LIKE 'Created_tmp_tables'; +Variable_name Value +Created_tmp_tables 0 +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY a, b, c) AS rn, +rank() OVER (ORDER BY b) AS rnk +FROM t_multi LIMIT 5; +rn rnk +1 1 +2 1 +3 5 +4 5 +5 1 +SHOW STATUS LIKE 'Created_tmp_tables'; +Variable_name Value +Created_tmp_tables 1 +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY a, b, c) AS rn, +rank() OVER (ORDER BY a, c) AS rnk +FROM t_multi LIMIT 5; +rn rnk +1 1 +2 3 +3 1 +4 3 +5 5 +SHOW STATUS LIKE 'Created_tmp_tables'; +Variable_name Value +Created_tmp_tables 1 +FLUSH STATUS; +SELECT row_number() OVER (), sum(val) FROM t1; +row_number() OVER () sum(val) +1 550 +SHOW STATUS LIKE 'Created_tmp_tables'; +Variable_name Value +Created_tmp_tables 1 +FLUSH STATUS; +SELECT rank() OVER (PARTITION BY val ORDER BY id) AS rnk, val, id FROM t1 LIMIT 5; +rnk val id +1 10 1 +1 20 2 +1 30 3 +1 40 4 +1 50 5 +SHOW STATUS LIKE 'Created_tmp_tables'; +Variable_name Value +Created_tmp_tables 1 +FLUSH STATUS; +SELECT rank() OVER (ORDER BY val) AS rnk, val FROM t1 LIMIT 5; +rnk val +1 10 +2 20 +3 30 +4 40 +5 50 +SHOW STATUS LIKE 'Created_tmp_tables'; +Variable_name Value +Created_tmp_tables 1 +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY val) AS rn, val FROM t1 ORDER BY val LIMIT 5; +rn val +1 10 +2 20 +3 30 +4 40 +5 50 +SHOW STATUS LIKE 'Created_tmp_tables'; +Variable_name Value +Created_tmp_tables 1 +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY id), +rank() OVER (ORDER BY grp) +FROM t2 LIMIT 3; +row_number() OVER (ORDER BY id) rank() OVER (ORDER BY grp) +1 1 +2 1 +3 3 +SHOW STATUS LIKE 'Created_tmp_tables'; +Variable_name Value +Created_tmp_tables 1 +EXPLAIN SELECT rank() OVER (PARTITION BY val ORDER BY id) AS rnk FROM t1; +id select_type table type possible_keys key key_len ref rows Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 10 Using temporary +EXPLAIN SELECT rank() OVER (ORDER BY val) AS rnk FROM t1; +id select_type table type possible_keys key key_len ref rows Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 10 Using temporary +DROP TABLE t1, t2, t_multi; diff --git a/mysql-test/main/streaming_wf.test b/mysql-test/main/streaming_wf.test new file mode 100644 index 0000000000000..06c2780e5630d --- /dev/null +++ b/mysql-test/main/streaming_wf.test @@ -0,0 +1,166 @@ +# MDEV-38970: Streaming window functions step 1 +# Tests streaming execution for ROW_NUMBER, RANK, DENSE_RANK +# Verifies: correctness, no temp table, LIMIT pushdown + +-- source include/not_embedded.inc +-- source include/no_protocol.inc + +--disable_warnings +DROP TABLE IF EXISTS t1, t2, t_multi; +--enable_warnings + +CREATE TABLE t1 (id INT PRIMARY KEY AUTO_INCREMENT, val INT); +INSERT INTO t1 (val) VALUES (10),(20),(30),(40),(50),(60),(70),(80),(90),(100); + +CREATE TABLE t2 (id INT PRIMARY KEY AUTO_INCREMENT, grp INT, INDEX idx_grp(grp)); +INSERT INTO t2 (grp) VALUES (1),(1),(2),(2),(3),(3); + +# t_multi: composite index to test prefix ORDER BY coverage +CREATE TABLE t_multi ( + id INT PRIMARY KEY AUTO_INCREMENT, + a INT, + b INT, + c INT, + INDEX idx_a_b_c (a, b, c), + INDEX idx_b (b) +); +INSERT INTO t_multi (a, b, c) VALUES + (1,1,1),(1,1,2),(1,2,1),(1,2,2), + (2,1,1),(2,1,2),(2,2,1),(2,2,2); + +# ROW_NUMBER() OVER () - basic streaming +SELECT row_number() OVER () AS rn, val FROM t1 LIMIT 5; +FLUSH STATUS; +SELECT row_number() OVER () AS rn FROM t1 LIMIT 3; +SHOW STATUS LIKE 'Created_tmp_tables'; + +# ROW_NUMBER() OVER (ORDER BY id) - streaming via primary key index +SELECT row_number() OVER (ORDER BY id) AS rn, id FROM t1 LIMIT 5; +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY id) AS rn FROM t1 LIMIT 3; +SHOW STATUS LIKE 'Created_tmp_tables'; + +# EXPLAIN must not show Using temporary for streaming cases +EXPLAIN SELECT row_number() OVER () AS rn FROM t1 LIMIT 5; +EXPLAIN SELECT row_number() OVER (ORDER BY id) AS rn FROM t1 LIMIT 5; + +# RANK() - streaming via primary key index +SELECT rank() OVER (ORDER BY id) AS rnk, id FROM t1 LIMIT 5; +FLUSH STATUS; +SELECT rank() OVER (ORDER BY id) AS rnk FROM t1 LIMIT 3; +SHOW STATUS LIKE 'Created_tmp_tables'; + +# DENSE_RANK() - streaming via primary key index +SELECT dense_rank() OVER (ORDER BY id) AS dr, id FROM t1 LIMIT 5; +FLUSH STATUS; +SELECT dense_rank() OVER (ORDER BY id) AS dr FROM t1 LIMIT 3; +SHOW STATUS LIKE 'Created_tmp_tables'; + +# Arithmetic expression containing streaming window function +SELECT 1 + row_number() OVER () AS rn FROM t1 LIMIT 3; + +# Two window functions same SELECT - both stream +SELECT + row_number() OVER () AS rn1, + row_number() OVER () AS rn2 +FROM t1 LIMIT 3; + +# Window function in derived table subquery +SELECT * FROM ( + SELECT row_number() OVER () AS rn, val FROM t1 +) sub WHERE rn <= 3; + +# Window function in derived table with ORDER BY index +SELECT rn, val FROM ( + SELECT row_number() OVER (ORDER BY id) AS rn, val, id FROM t1 +) sub WHERE rn BETWEEN 2 AND 4; + +# INNER JOIN - no temp table created +FLUSH STATUS; +SELECT row_number() OVER () AS rn, t1.val, t2.grp +FROM t1 JOIN t2 ON t1.id = t2.id LIMIT 3; +SHOW STATUS LIKE 'Created_tmp_tables'; + +# LEFT JOIN +SELECT row_number() OVER () AS rn, t1.id, t2.grp +FROM t1 LEFT JOIN t2 ON t1.id = t2.id LIMIT 3; + +# WHERE clause - filters applied before window function +SELECT row_number() OVER () AS rn, val FROM t1 WHERE val > 50 LIMIT 3; +SELECT row_number() OVER (ORDER BY id) AS rn, id, val FROM t1 WHERE id > 5 LIMIT 3; + +# Same ORDER BY - should stream +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY a, b, c) AS rn, + rank() OVER (ORDER BY a, b, c) AS rnk +FROM t_multi LIMIT 5; +SHOW STATUS LIKE 'Created_tmp_tables'; + +# Widest (a,b,c) covers prefix (a,b) - should stream +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY a, b, c) AS rn, + rank() OVER (ORDER BY a, b) AS rnk +FROM t_multi LIMIT 5; +SHOW STATUS LIKE 'Created_tmp_tables'; + +# Widest (a,b,c) covers multiple prefixes - should stream +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY a, b, c) AS rn, + rank() OVER (ORDER BY a, b) AS rnk, + dense_rank() OVER (ORDER BY a) AS dr +FROM t_multi LIMIT 5; +SHOW STATUS LIKE 'Created_tmp_tables'; + +# OVER() mixed with ORDER BY - should stream +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY a, b) AS rn, + row_number() OVER () AS rnk +FROM t_multi LIMIT 5; +SHOW STATUS LIKE 'Created_tmp_tables'; + +# (ORDER BY b) is NOT a prefix of (ORDER BY a,b,c) - must NOT stream +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY a, b, c) AS rn, + rank() OVER (ORDER BY b) AS rnk +FROM t_multi LIMIT 5; +SHOW STATUS LIKE 'Created_tmp_tables'; + +# (ORDER BY a,c) is NOT a prefix of (ORDER BY a,b,c) - must NOT stream +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY a, b, c) AS rn, + rank() OVER (ORDER BY a, c) AS rnk +FROM t_multi LIMIT 5; +SHOW STATUS LIKE 'Created_tmp_tables'; + +# Aggregate + window function - must NOT stream +FLUSH STATUS; +SELECT row_number() OVER (), sum(val) FROM t1; +SHOW STATUS LIKE 'Created_tmp_tables'; + +# PARTITION BY forces temp table +FLUSH STATUS; +SELECT rank() OVER (PARTITION BY val ORDER BY id) AS rnk, val, id FROM t1 LIMIT 5; +SHOW STATUS LIKE 'Created_tmp_tables'; + +# ORDER BY on non-indexed column forces temp table +FLUSH STATUS; +SELECT rank() OVER (ORDER BY val) AS rnk, val FROM t1 LIMIT 5; +SHOW STATUS LIKE 'Created_tmp_tables'; + +# Global ORDER BY forces temp table +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY val) AS rn, val FROM t1 ORDER BY val LIMIT 5; +SHOW STATUS LIKE 'Created_tmp_tables'; + +# Different ORDER BY columns - must NOT stream +FLUSH STATUS; +SELECT row_number() OVER (ORDER BY id), + rank() OVER (ORDER BY grp) +FROM t2 LIMIT 3; +SHOW STATUS LIKE 'Created_tmp_tables'; + +# EXPLAIN shows Using temporary for non-streaming cases +EXPLAIN SELECT rank() OVER (PARTITION BY val ORDER BY id) AS rnk FROM t1; +EXPLAIN SELECT rank() OVER (ORDER BY val) AS rnk FROM t1; + +DROP TABLE t1, t2, t_multi; \ No newline at end of file diff --git a/mysql-test/main/win.result b/mysql-test/main/win.result index ec9ea7469eb83..4618be64afd79 100644 --- a/mysql-test/main/win.result +++ b/mysql-test/main/win.result @@ -3911,42 +3911,25 @@ ANALYZE "cost": "REPLACED", "r_loops": 1, "r_total_time_ms": "REPLACED", - "window_functions_computation": { - "sorts": [ - { - "filesort": { - "sort_key": "`row_number() OVER()`", - "r_loops": 1, - "r_total_time_ms": "REPLACED", - "r_used_priority_queue": false, - "r_output_rows": 3, - "r_buffer_size": "REPLACED", - "r_sort_mode": "sort_key,rowid" - } + "nested_loop": [ + { + "table": { + "table_name": "t1", + "access_type": "ALL", + "loops": 1, + "r_loops": 1, + "rows": 3, + "r_rows": 3, + "cost": "REPLACED", + "r_table_time_ms": "REPLACED", + "r_other_time_ms": "REPLACED", + "r_engine_stats": REPLACED, + "filtered": 100, + "r_total_filtered": 100, + "r_filtered": 100 } - ], - "temporary_table": { - "nested_loop": [ - { - "table": { - "table_name": "t1", - "access_type": "ALL", - "loops": 1, - "r_loops": 1, - "rows": 3, - "r_rows": 3, - "cost": "REPLACED", - "r_table_time_ms": "REPLACED", - "r_other_time_ms": "REPLACED", - "r_engine_stats": REPLACED, - "filtered": 100, - "r_total_filtered": 100, - "r_filtered": 100 - } - } - ] } - } + ] } } SELECT row_number() OVER() FROM t1; diff --git a/mysql-test/main/win_empty_over.result b/mysql-test/main/win_empty_over.result index 2f75f80c6ad01..2e479b0588f56 100644 --- a/mysql-test/main/win_empty_over.result +++ b/mysql-test/main/win_empty_over.result @@ -37,33 +37,22 @@ EXPLAIN "query_block": { "select_id": 1, "cost": "COST_REPLACED", - "window_functions_computation": { - "sorts": [ - { - "filesort": { - "sort_key": "pk" - } + "nested_loop": [ + { + "table": { + "table_name": "t1", + "access_type": "index", + "key": "PRIMARY", + "key_length": "4", + "used_key_parts": ["pk"], + "loops": 1, + "rows": 11, + "cost": "COST_REPLACED", + "filtered": 100, + "using_index": true } - ], - "temporary_table": { - "nested_loop": [ - { - "table": { - "table_name": "t1", - "access_type": "index", - "key": "PRIMARY", - "key_length": "4", - "used_key_parts": ["pk"], - "loops": 1, - "rows": 11, - "cost": "COST_REPLACED", - "filtered": 100, - "using_index": true - } - } - ] } - } + ] } } explain FORMAT=JSON select row_number() over (), pk from t1; @@ -72,33 +61,22 @@ EXPLAIN "query_block": { "select_id": 1, "cost": "COST_REPLACED", - "window_functions_computation": { - "sorts": [ - { - "filesort": { - "sort_key": "`row_number() over ()`" - } + "nested_loop": [ + { + "table": { + "table_name": "t1", + "access_type": "index", + "key": "PRIMARY", + "key_length": "4", + "used_key_parts": ["pk"], + "loops": 1, + "rows": 11, + "cost": "COST_REPLACED", + "filtered": 100, + "using_index": true } - ], - "temporary_table": { - "nested_loop": [ - { - "table": { - "table_name": "t1", - "access_type": "index", - "key": "PRIMARY", - "key_length": "4", - "used_key_parts": ["pk"], - "loops": 1, - "rows": 11, - "cost": "COST_REPLACED", - "filtered": 100, - "using_index": true - } - } - ] } - } + ] } } select row_number() over () from (select 4) as t; diff --git a/mysql-test/suite/encryption/r/tempfiles_encrypted.result b/mysql-test/suite/encryption/r/tempfiles_encrypted.result index 16ab0b376da18..0c2e6ce27bdee 100644 --- a/mysql-test/suite/encryption/r/tempfiles_encrypted.result +++ b/mysql-test/suite/encryption/r/tempfiles_encrypted.result @@ -3917,42 +3917,25 @@ ANALYZE "cost": "REPLACED", "r_loops": 1, "r_total_time_ms": "REPLACED", - "window_functions_computation": { - "sorts": [ - { - "filesort": { - "sort_key": "`row_number() OVER()`", - "r_loops": 1, - "r_total_time_ms": "REPLACED", - "r_used_priority_queue": false, - "r_output_rows": 3, - "r_buffer_size": "REPLACED", - "r_sort_mode": "sort_key,rowid" - } + "nested_loop": [ + { + "table": { + "table_name": "t1", + "access_type": "ALL", + "loops": 1, + "r_loops": 1, + "rows": 3, + "r_rows": 3, + "cost": "REPLACED", + "r_table_time_ms": "REPLACED", + "r_other_time_ms": "REPLACED", + "r_engine_stats": REPLACED, + "filtered": 100, + "r_total_filtered": 100, + "r_filtered": 100 } - ], - "temporary_table": { - "nested_loop": [ - { - "table": { - "table_name": "t1", - "access_type": "ALL", - "loops": 1, - "r_loops": 1, - "rows": 3, - "r_rows": 3, - "cost": "REPLACED", - "r_table_time_ms": "REPLACED", - "r_other_time_ms": "REPLACED", - "r_engine_stats": REPLACED, - "filtered": 100, - "r_total_filtered": 100, - "r_filtered": 100 - } - } - ] } - } + ] } } SELECT row_number() OVER() FROM t1; diff --git a/sql/item_windowfunc.h b/sql/item_windowfunc.h index 260619c3e75d8..450af00cf283e 100644 --- a/sql/item_windowfunc.h +++ b/sql/item_windowfunc.h @@ -1086,15 +1086,32 @@ class Item_window_func : public Item_func_or_sum : Item_func_or_sum(thd, (Item *) win_func), window_name(win_name), window_spec(NULL), force_return_blank(true), - read_value_from_result_field(false) {} + read_value_from_result_field(false), + m_is_streaming(false) {} Item_window_func(THD *thd, Item_sum *win_func, Window_spec *win_spec) : Item_func_or_sum(thd, (Item *) win_func), window_name(NULL), window_spec(win_spec), force_return_blank(true), - read_value_from_result_field(false) {} + read_value_from_result_field(false), + m_is_streaming(false) {} Item_sum *window_func() const { return (Item_sum *) args[0]; } + bool is_streamable() const; + + void streaming_wf_init(THD *thd) + { + m_is_streaming= true; + Item_sum *wf= window_func(); + if (window_spec->order_list->elements > 0 && + (dynamic_cast(wf) || + dynamic_cast(wf))) + wf->setup_window_func(thd, window_spec); + else + wf->clear(); + } + + void streaming_wf_add() { window_func()->add(); } void update_used_tables() override; @@ -1238,6 +1255,22 @@ class Item_window_func : public Item_func_or_sum */ bool force_return_blank; bool read_value_from_result_field; + + /* + When true, val_int()/val_real()/val_str()/val_decimal() delegate directly + to window_func()->val_*() instead of reading from result_field. + + In the normal (non-streaming) execution path, window function values are + computed by compute_window_func() and stored in a temp table column + (result_field). val_int() then reads the pre-computed value from there + during the retrieval phase (read_value_from_result_field=true). + + In streaming mode no temp table is created, so result_field is NULL. + Instead, the window function state (cur_rank, count, etc.) is updated + per-row by streaming_wf_add() just before val_int() is called, so + val_int() can read the current value directly from the Item_sum. + */ + bool m_is_streaming; void print_for_percentile_functions(String *str, enum_query_type query_type); public: @@ -1271,6 +1304,12 @@ class Item_window_func : public Item_func_or_sum double val_real() override { double res; + if (m_is_streaming) + { + res= window_func()->val_real(); + null_value= window_func()->null_value; + return res; + } if (force_return_blank) { res= 0.0; @@ -1292,6 +1331,12 @@ class Item_window_func : public Item_func_or_sum longlong val_int() override { longlong res; + if (m_is_streaming) + { + res= window_func()->val_int(); + null_value= window_func()->null_value; + return res; + } if (force_return_blank) { res= 0; @@ -1313,6 +1358,12 @@ class Item_window_func : public Item_func_or_sum String* val_str(String* str) override { String *res; + if (m_is_streaming) + { + res= window_func()->val_str(str); + null_value= window_func()->null_value; + return res; + } if (force_return_blank) { null_value= true; @@ -1335,6 +1386,10 @@ class Item_window_func : public Item_func_or_sum bool val_native(THD *thd, Native *to) override { + if (m_is_streaming) + { + return val_native_from_item(thd, window_func(), to); + } if (force_return_blank) return null_value= true; if (read_value_from_result_field) @@ -1345,6 +1400,12 @@ class Item_window_func : public Item_func_or_sum my_decimal* val_decimal(my_decimal* dec) override { my_decimal *res; + if (m_is_streaming) + { + res= window_func()->val_decimal(dec); + null_value= window_func()->null_value; + return res; + } if (force_return_blank) { null_value= true; diff --git a/sql/sql_lex.h b/sql/sql_lex.h index 438f8824ab99f..37ef535b538f2 100644 --- a/sql/sql_lex.h +++ b/sql/sql_lex.h @@ -1567,6 +1567,7 @@ class st_select_lex: public st_select_lex_node bool add_window_func(Item_window_func *win_func); bool have_window_funcs() const { return (window_funcs.elements !=0); } + bool have_streaming_window_funcs_candidates() const; ORDER *find_common_window_func_partition_fields(THD *thd); bool cond_pushdown_is_allowed() const diff --git a/sql/sql_select.cc b/sql/sql_select.cc index b8fbbe2f62f58..25ef2af6bc8f2 100644 --- a/sql/sql_select.cc +++ b/sql/sql_select.cc @@ -68,6 +68,7 @@ #include "create_tmp_table.h" #include "optimizer_defaults.h" #include "derived_handler.h" +#include "item_windowfunc.h" #include "opt_hints.h" #include "opt_group_by_cardinality.h" @@ -217,6 +218,9 @@ static COND *optimize_cond(JOIN *join, COND *conds, bool const_expression_in_where(COND *conds,Item *item, Item **comp_item); static int do_select(JOIN *join, Procedure *procedure); +static enum_nested_loop_state +end_send_streaming_wf(JOIN *join, JOIN_TAB *join_tab, bool end_of_records); + static enum_nested_loop_state evaluate_join_record(JOIN *, JOIN_TAB *, int); static enum_nested_loop_state evaluate_null_complemented_join_record(JOIN *join, JOIN_TAB *join_tab); @@ -3333,12 +3337,20 @@ int JOIN::optimize_stage2() need_tmp= test_if_need_tmp_table(); /* - If window functions are present then we can't have simple_order set to + For streaming window functions with ORDER BY, assign the + covering index to the driving JOIN_TAB + */ + if (!need_tmp && select_lex->have_window_funcs()) + setup_streaming_wf_index(); + + /* + If non-streaming window functions are present then we can't have simple_order set to TRUE as the window function needs a temp table for computation. ORDER BY is computed after the window function computation is done, so the sort will be done on the temp table. */ - if (select_lex->have_window_funcs()) + if (select_lex->have_window_funcs() && + !window_funcs_can_stream()) simple_order= FALSE; /* @@ -3605,7 +3617,8 @@ int JOIN::optimize_stage2() simple_order= TRUE; select_distinct= FALSE; - if (select_lex->have_window_funcs()) + if (select_lex->have_window_funcs() && + !window_funcs_can_stream()) { if (!(join_tab= thd->alloc(1))) DBUG_RETURN(1); @@ -4329,14 +4342,23 @@ bool JOIN::make_aggr_tables_info() */ if (select_lex->window_funcs.elements) { - curr_tab= join_tab + total_join_tab_cnt(); - if (!(curr_tab->window_funcs_step= new Window_funcs_computation)) - DBUG_RETURN(true); - if (curr_tab->window_funcs_step->setup(thd, &select_lex->window_funcs, - curr_tab)) - DBUG_RETURN(true); - /* Count that we're using window functions. */ - status_var_increment(thd->status_var.feature_window_functions); + if (window_funcs_can_stream()) + { + List_iterator_fast it(select_lex->window_funcs); + Item_window_func *wf; + while ((wf= it++)) + wf->streaming_wf_init(thd); + } + else + { + curr_tab= join_tab + total_join_tab_cnt(); + if (!(curr_tab->window_funcs_step= new Window_funcs_computation)) + DBUG_RETURN(true); + if (curr_tab->window_funcs_step->setup(thd, &select_lex->window_funcs, + curr_tab)) + DBUG_RETURN(true); + status_var_increment(thd->status_var.feature_window_functions); + } } if (select_lex->custom_agg_func_used()) status_var_increment(thd->status_var.feature_custom_aggregate_functions); @@ -4363,11 +4385,19 @@ bool JOIN::make_aggr_tables_info() fields= curr_fields_list; // Reset before execution set_items_ref_array(items0); - if (join_tab) - join_tab[exec_join_tab_cnt() + aggr_tables - 1].next_select= - setup_end_select_func(this); group= has_group_by; + if (join_tab) + { + JOIN_TAB *last_aggr_tab= + &join_tab[exec_join_tab_cnt() + aggr_tables - 1]; + + if (window_funcs_can_stream()) + join_tab[top_join_tab_count - 1].next_select= end_send_streaming_wf; + else + last_aggr_tab->next_select= setup_end_select_func(this); + } + DBUG_RETURN(false); } @@ -23940,6 +23970,22 @@ Next_select_func setup_end_select_func(JOIN *join) return end_send; } +static enum_nested_loop_state +end_send_streaming_wf(JOIN *join, JOIN_TAB *join_tab, bool end_of_records) +{ + DBUG_ENTER("end_send_streaming_wf"); + + if (end_of_records) + DBUG_RETURN(end_send(join, join_tab, end_of_records)); + + List_iterator_fast it(join->select_lex->window_funcs); + Item_window_func *wf; + while ((wf= it++)) + wf->streaming_wf_add(); + + DBUG_RETURN(end_send(join, join_tab, end_of_records)); +} + /** Make a join of all tables and write it on socket or to table. @@ -34648,6 +34694,216 @@ bool JOIN::transform_all_conds_and_on_exprs_in_join_list( return false; } +/* + Check if we need to create a temporary table. + This has to be done if all tables are not already read (const tables) + and one of the following conditions holds: + - We are using DISTINCT (simple distinct's are already optimized away) + - We are using an ORDER BY or GROUP BY on fields not in the first table + - We are using different ORDER BY and GROUP BY orders + - The user wants us to buffer the result. + - We are using non-streaming WINDOW functions. + When the WITH ROLLUP modifier is present, we cannot skip temporary table + creation for the DISTINCT clause just because there are only const tables. +*/ + +bool JOIN::test_if_need_tmp_table() +{ + return ((const_tables != table_count && + ((select_distinct || !simple_order || !simple_group) || + (group_list && order) || + MY_TEST(select_options & OPTION_BUFFER_RESULT))) || + (rollup.state != ROLLUP::STATE_NONE && select_distinct) || + (select_lex->have_window_funcs() && + !window_funcs_can_stream())); +} + +/* + Find the index on the driving table that covers the ORDER BY clause + of the given window spec. + + @param spec Window spec to check ORDER BY for + @param keyno OUT: index number that covers the ORDER BY, if found + + @return true if a matching index was found, false otherwise +*/ +bool JOIN::find_wf_order_index(const Window_spec *spec, uint *keyno) const +{ + DBUG_ASSERT(spec->order_list->elements > 0); + + /* Must not be called when all tables are const */ + if (!join_tab || const_tables >= table_count) + return false; + + JOIN_TAB *tab= join_tab + const_tables; + if (!tab || !tab->table) + return false; + + TABLE *table= tab->table; + uint n_order= spec->order_list->elements; + + for (uint k= 0; k < table->s->keys; k++) + { + KEY *key_info= table->key_info + k; + + if (key_info->usable_key_parts < n_order) + continue; + + bool matches= true; + uint i= 0; + for (ORDER *ord= spec->order_list->first; ord; ord= ord->next, i++) + { + Item *item= *ord->item; + if (item->type() != Item::FIELD_ITEM) + { matches= false; break; } + + Item_field *ifield= (Item_field *) item; + if (ifield->field != key_info->key_part[i].field) + { matches= false; break; } + + bool key_asc= !(key_info->key_part[i].key_part_flag & HA_REVERSE_SORT); + bool ord_asc= (ord->direction != ORDER::ORDER_DESC); + if (key_asc != ord_asc) + { matches= false; break; } + } + + if (matches) + { + *keyno= k; + return true; + } + } + + return false; +} + + +/* + Check whether all streaming window functions have their ORDER BY + covered by an existing index. +*/ +bool JOIN::window_funcs_can_stream() const +{ + if (!select_lex->have_streaming_window_funcs_candidates()) + return false; + + /* Streaming requires at least one real scan tab. */ + if (top_join_tab_count == 0 || + !join_tab || + const_tables >= table_count) + return false; + + /* + Streaming is only supported for SELECT queries. + can we support other DML commands? + */ + if (thd->lex->sql_command != SQLCOM_SELECT) + return false; + + /* + Step 1: Find the window function with the most ORDER BY columns. + */ + const Window_spec *widest_spec= NULL; + uint widest_n= 0; + + List_iterator_fast it(select_lex->window_funcs); + Item_window_func *wf; + while ((wf= it++)) + { + uint n= wf->window_spec->order_list->elements; + if (n > widest_n) + { + widest_n= n; + widest_spec= wf->window_spec; + } + } + + /* All window functions have OVER() — no ORDER BY anywhere */ + if (widest_n == 0) + return true; + + /* + Step 2: Find an index that fully covers the widest ORDER BY. + */ + uint keyno= 0; + if (!find_wf_order_index(widest_spec, &keyno)) + return false; + + /* + Step 3: Verify every other window function's ORDER BY is a prefix + of the widest ORDER BY. + */ + it.rewind(); + while ((wf= it++)) + { + const Window_spec *spec= wf->window_spec; + uint n= spec->order_list->elements; + + if (n == 0) + continue; + + if (n == widest_n && spec == widest_spec) + continue; + + /* Compare column by column against the widest spec */ + ORDER *ord_wide= widest_spec->order_list->first; + ORDER *ord_this= spec->order_list->first; + + for (; ord_this; ord_wide= ord_wide->next, ord_this= ord_this->next) + { + Item *item_wide= *ord_wide->item; + Item *item_this= *ord_this->item; + + if (item_wide->type() != Item::FIELD_ITEM || + item_this->type() != Item::FIELD_ITEM) + return false; + + if (((Item_field*)item_wide)->field != + ((Item_field*)item_this)->field) + return false; + + if (ord_wide->direction != ord_this->direction) + return false; + } + } + + return true; +} + + +/* + Assign the index that covers each streaming window function's ORDER BY + to the driving JOIN_TAB. +*/ +void JOIN::setup_streaming_wf_index() +{ + const Window_spec *widest_spec= NULL; + uint widest_n= 0; + + List_iterator_fast it(select_lex->window_funcs); + Item_window_func *wf; + while ((wf= it++)) + { + uint n= wf->window_spec->order_list->elements; + if (n > widest_n) + { + widest_n= n; + widest_spec= wf->window_spec; + } + } + + if (widest_n == 0) + return; + + uint keyno= 0; + if (!find_wf_order_index(widest_spec, &keyno)) + return; + + JOIN_TAB *tab= join_tab + const_tables; + tab->index= keyno; + tab->type= JT_NEXT; +} + static void init_join_plan_search_state(JOIN *join) { join->cur_sj_inner_tables= 0; diff --git a/sql/sql_select.h b/sql/sql_select.h index 1b001494e2fa2..7247db9f6bbc3 100644 --- a/sql/sql_select.h +++ b/sql/sql_select.h @@ -1890,27 +1890,10 @@ class JOIN :public Sql_alloc ulonglong needed_space); void set_allowed_join_cache_types(); bool is_allowed_hash_join_access(const TABLE *table); - /* - Check if we need to create a temporary table. - This has to be done if all tables are not already read (const tables) - and one of the following conditions holds: - - We are using DISTINCT (simple distinct's are already optimized away) - - We are using an ORDER BY or GROUP BY on fields not in the first table - - We are using different ORDER BY and GROUP BY orders - - The user wants us to buffer the result. - - We are using WINDOW functions. - When the WITH ROLLUP modifier is present, we cannot skip temporary table - creation for the DISTINCT clause just because there are only const tables. - */ - bool test_if_need_tmp_table() - { - return ((const_tables != table_count && - ((select_distinct || !simple_order || !simple_group) || - (group_list && order) || - MY_TEST(select_options & OPTION_BUFFER_RESULT))) || - (rollup.state != ROLLUP::STATE_NONE && select_distinct) || - select_lex->have_window_funcs()); - } + bool test_if_need_tmp_table(); + bool find_wf_order_index(const Window_spec *spec, uint *keyno) const; + bool window_funcs_can_stream() const; + void setup_streaming_wf_index(); bool choose_subquery_plan(table_map join_tables); void get_partial_cost_and_fanout(int end_tab_idx, table_map filter_map, diff --git a/sql/sql_window.cc b/sql/sql_window.cc index 6f5fb99e82866..3007b2b38bfba 100644 --- a/sql/sql_window.cc +++ b/sql/sql_window.cc @@ -3281,6 +3281,67 @@ bool st_select_lex::add_window_func(Item_window_func *win_func) return window_funcs.push_back(win_func); } +bool Item_window_func::is_streamable() const +{ + const Window_spec *spec= window_spec; + + // No PARTITION BY. + if (spec->partition_list->elements > 0) + return false; + + // Frame must not reference future rows (no FOLLOWING) + if (spec->window_frame) + { + const Window_frame_bound *end_bound = spec->window_frame->top_bound; + if (end_bound && + end_bound->precedence_type == Window_frame_bound::FOLLOWING) + return false; + } + + // Supported function types only (ROW_NUMBER(), RANK(), DENSE_RANK()) + const Item_sum *wf= window_func(); + if (!dynamic_cast(wf) && + !dynamic_cast(wf) && + !dynamic_cast(wf)) + return false; + + return true; +} + +bool SELECT_LEX::have_streaming_window_funcs_candidates() const +{ + if (!have_window_funcs()) + return false; + + // No global ORDER BY or GROUP BY. + if (order_list.elements > 0 || group_list.elements > 0) + return false; + + /* + Cannot stream when non-window aggregate functions are present. + TODO: Running aggregates supported later. + */ + List_iterator_fast it1(const_cast(this)->item_list); + Item *item; + while ((item= it1++)) + { + if (item->with_sum_func() && + item->type() != Item::WINDOW_FUNC_ITEM) + return false; + } + + // Every window function in the SELECT list must be individually streamable. + List_iterator_fast it2(const_cast(this)->window_funcs); + Item_window_func *wf; + while ((wf = it2++)) + { + if (!wf->is_streamable()) + return false; + } + + return true; +} + ///////////////////////////////////////////////////////////////////////////// // Unneeded comments (will be removed when we develop a replacement for // the feature that was attempted here