From d399839c254d14a2b62251a77e96bfe04e205e22 Mon Sep 17 00:00:00 2001 From: OmarGamal10 Date: Tue, 23 Jun 2026 04:33:49 +0300 Subject: [PATCH 01/16] Set the base and initial structure for streaming window functions path and criteria validation function --- sql/item_sum.h | 1 + sql/item_windowfunc.h | 6 + sql/sql_window.cc | 269 ++++++++++++++++++++++++++++++++++++++++-- sql/sql_window.h | 29 ++++- 4 files changed, 296 insertions(+), 9 deletions(-) diff --git a/sql/item_sum.h b/sql/item_sum.h index 39ed79e7c0203..f1f1ac3863090 100644 --- a/sql/item_sum.h +++ b/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()) { diff --git a/sql/item_windowfunc.h b/sql/item_windowfunc.h index bff614372e2fc..90a3f73e30eef 100644 --- a/sql/item_windowfunc.h +++ b/sql/item_windowfunc.h @@ -150,6 +150,8 @@ class Item_sum_row_number: public Item_sum_int return name; } + bool inline is_streamable() const override { return true; } + protected: Item *shallow_copy(THD *thd) const override { return get_item_copy(thd, this); } @@ -215,6 +217,8 @@ class Item_sum_rank: public Item_sum_int return name; } + bool inline 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; } + bool inline is_streamable() const override { return true; } + void setup_window_func(THD *thd, Window_spec *window_spec) override; void cleanup() override diff --git a/sql/sql_window.cc b/sql/sql_window.cc index e5b4b4de7d644..61f7c461acec4 100644 --- a/sql/sql_window.cc +++ b/sql/sql_window.cc @@ -15,14 +15,19 @@ Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA */ #include "mariadb.h" +#include "mysql/plugin.h" #include "sql_parse.h" #include "sql_select.h" #include "sql_list.h" #include "item_windowfunc.h" #include "filesort.h" #include "sql_base.h" +#include "item.h" +#include #include "sql_window.h" +static ORDER *concat_order_lists(MEM_ROOT *mem_root, ORDER *list1, + ORDER *list2); bool Window_spec::check_window_names(List_iterator_fast &it) @@ -537,6 +542,55 @@ int compare_order_lists(SQL_I_List *part_list1, return CMP_EQ; } +/* + Overloaded to take ORDER* objects instead of SQL_I_List* (the longest + wf order list, and the main query order list). + Note that we use -1 for the spec_number of the main query order list, as + window spec numbers start from 0. + Returns CMP_EQ if the lists are equal or NULL, CMP_LT_C if the first list is + NULL or a prefix of the second list, CMP_GT_C if the second list is NULL or + a prefix of the first list, and CMP_LT or CMP_GT otherwise. +*/ +static int compare_order_lists(ORDER *list1, int spec_number1, ORDER *list2, + int spec_number2) +{ + if (!list1 && !list2) + return CMP_EQ; + if (!list1) + return CMP_LT_C; + if (!list2) + return CMP_GT_C; + ORDER *elem1= list1; + ORDER *elem2= list2; + for (; elem1 && elem2; elem1= elem1->next, elem2= elem2->next) + { + int cmp; + // remove all constants as we don't need them for comparision + while (elem1 && ((*elem1->item)->real_item())->const_item()) + { + elem1= elem1->next; + continue; + } + + while (elem2 && ((*elem2->item)->real_item())->const_item()) + { + elem2= elem2->next; + continue; + } + + if (!elem1 || !elem2) + break; + + if ((cmp= + compare_order_elements(elem1, spec_number1, elem2, spec_number2))) + return cmp; + } + if (elem1) + return CMP_GT_C; + if (elem2) + return CMP_LT_C; + return CMP_EQ; +} static int compare_window_frame_bounds(Window_frame_bound *win_frame_bound1, @@ -731,12 +785,17 @@ typedef int (*Item_window_func_cmp)(Item_window_func *f1, The changes between the groups are marked by setting item_window_func->marker. */ -static -void order_window_funcs_by_window_specs(List *win_func_list) +// I think i can move this to preparation +// run this in preparation, along with other criteria and set some +// is_streamable variable on JOIN +// then test_if_need_tmp_table would check it too. +// Returns true if only one sort exists, false if none or more +static bool +order_window_funcs_by_window_specs(List *win_func_list) { if (win_func_list->elements == 0) - return; - + return false; + bool more_than_one_sort= false; bubble_sort(win_func_list, compare_window_funcs_by_window_specs, NULL); @@ -768,6 +827,7 @@ void order_window_funcs_by_window_specs(List *win_func_list) curr->marker= (MARKER_SORTORDER_CHANGE | MARKER_PARTITION_CHANGE | MARKER_FRAME_CHANGE); + more_than_one_sort= true; } else if (win_spec_prev->partition_list != win_spec_curr->partition_list) { @@ -779,8 +839,95 @@ void order_window_funcs_by_window_specs(List *win_func_list) prev= curr; } + return !more_than_one_sort; +} + +static inline bool frame_is_current_row(Window_spec *win_spec) +{ + // null is default, range between unbounded preceding and current row + return win_spec->window_frame == NULL || + (win_spec->window_frame->top_bound->precedence_type == + Window_frame_bound::CURRENT && + win_spec->window_frame->bottom_bound->precedence_type == + Window_frame_bound::CURRENT); } +static inline bool check_argument_list_aggregation(Window_spec *win_spec) +{ + for (ORDER *o= win_spec->partition_list->first; o; o= o->next) + if ((*o->item)->with_sum_func()) + return true; + + for (ORDER *o= win_spec->order_list->first; o; o= o->next) + if ((*o->item)->with_sum_func()) + return true; + + return false; +} + +// now I know number 1 is bad (side effect, but i need it for criteria, why run +// two times?) +// 1. Runs order_window_funcs_by_window_specs(), so now we sort in preparation +// 2. We check each fucntion from our subset or no (let it be rank and +// row_number for now) +// 3. frame only current row (normal), or unbounded preceding (for sum +// functions and stuff like that, can be skipped now) +// 4. Longest order is compatible with main query order (if exists) or not. +bool have_streaming_window_funcs(THD *thd, List &win_funcs, + ORDER *&longest_wf_order, + ORDER *main_query_order, + bool &streaming_wf_order_is_longer) +{ + // This checks if more than one SORTORDER_MARKER_CHANGE exists. + // Calling order early here has a problem with one of the existing tests. Not + // sure why. + if (win_funcs.elements == 0 || + !order_window_funcs_by_window_specs(&win_funcs)) + return false; + List_iterator_fast it(win_funcs); + Item_window_func *win_func; + Item_window_func *win_func_with_longest_order= NULL; + int longest_order_elements= -1; + int cmp; + + while ((win_func= it++)) + { + Window_spec *spec= win_func->window_spec; + if (check_argument_list_aggregation(spec)) + return false; + + int win_func_order_elements= + spec->partition_list->elements + spec->order_list->elements; + if (win_func_order_elements > longest_order_elements) + { + longest_order_elements= win_func_order_elements; + win_func_with_longest_order= win_func; + } + + if (!(win_func->window_func()->is_streamable() && + frame_is_current_row(win_func->window_spec))) + return false; + } + + longest_wf_order= concat_order_lists( + thd->mem_root, + win_func_with_longest_order->window_spec->partition_list->first, + win_func_with_longest_order->window_spec->order_list->first); + + // check compatibility of both + cmp= compare_order_lists( + longest_wf_order, + win_func_with_longest_order->window_spec->win_spec_number, + main_query_order, -1); + + if (!(CMP_LT_C <= cmp && cmp <= CMP_GT_C)) + return false; + if (cmp == CMP_GT_C) + streaming_wf_order_is_longer= true; + else + streaming_wf_order_is_longer= false; + return true; +} ///////////////////////////////////////////////////////////////////////////// @@ -1258,7 +1405,28 @@ class Cursor_manager List cursors; }; - +// // I think the only need for the object is to hold the group_bound_trackers, +// we +// // don't even need the functions list +// class Window_funcs_sort_streaming : public Sql_alloc +// { +// public: +// bool setup(THD *thd, List &win_funcs); +// bool process_row(); // this object is attached to the JOIN, and +// // process_row() is called for a method attached on +// // takes the window funcs and the current row by +// // end_compute_win_funcs() and calls the appropriate +// // cursors to update the aggregate functions + +// private: +// int row_num= 0; // acts like internal state for process row +// List win_funcs; +// // these correspond to the window functions in the SELECT_LEX (all +// functions +// // are streamable) +// List cursor_managers; +// List partition_trackers; +// }; ////////////////////////////////////////////////////////////////////////////// // RANGE-type frames @@ -2699,6 +2867,8 @@ static bool is_computed_with_remove(Item_sum::Sumfunctype sum_func) If the window functions share the same frame specification, those window functions will be registered to the same cursor. */ +// i can reuse this for streaming, it just creates a Cursor Manager for each +// window function bool get_window_functions_required_cursors( THD *thd, List& window_functions, @@ -2908,7 +3078,8 @@ bool compute_window_func(THD *thd, tracker->init(); partition_trackers.push_back(tracker); } - + // the frame cursor thing i think would not need much change if we assume + // current frame = current row List_iterator_fast iter_part_trackers(partition_trackers); ha_rows rownum= 0; uchar *rowid_buf= (uchar*) my_malloc(PSI_INSTRUMENT_ME, tbl->file->ref_length, MYF(0)); @@ -2926,7 +3097,8 @@ bool compute_window_func(THD *thd, iter_win_funcs.rewind(); iter_part_trackers.rewind(); iter_cursor_managers.rewind(); - + // we can use a similar appraoch for streaming, where a row is passed over + // all window functions before another is fetched (single pass) Group_bound_tracker *tracker; while ((win_func= iter_win_funcs++) && (tracker= iter_part_trackers++) && @@ -2963,6 +3135,8 @@ bool compute_window_func(THD *thd, /* We now have computed values for each window function. They can now be saved in the current row. */ + // i need to save to current row field, but might not need all that for + // streaming if (save_window_function_values(window_functions, tbl, rowid_buf)) { ret= true; @@ -3065,6 +3239,7 @@ bool Window_func_runner::exec(THD *thd, TABLE *tbl, SORT_INFO *filesort_result) Item_window_func *win_func; while ((win_func= it++)) { + // i need this so it reads live not from result_field win_func->set_phase_to_computation(); // TODO(cvicentiu) Setting the aggregator should probably be done during // setup of Window_funcs_sort. @@ -3073,6 +3248,7 @@ bool Window_func_runner::exec(THD *thd, TABLE *tbl, SORT_INFO *filesort_result) } it.rewind(); + // i would skip this now List cursor_managers; if (get_window_functions_required_cursors(thd, window_functions, &cursor_managers)) @@ -3085,6 +3261,7 @@ bool Window_func_runner::exec(THD *thd, TABLE *tbl, SORT_INFO *filesort_result) tbl, filesort_result); while ((win_func= it++)) { + // we do not want this at all in streaming win_func->set_phase_to_retrieval(); } @@ -3123,7 +3300,8 @@ bool Window_funcs_sort::setup(THD *thd, SQL_SELECT *sel, JOIN_TAB *join_tab) { Window_spec *spec; - Item_window_func *win_func= it.peek(); + Item_window_func *win_func= it.peek(); + // reuse this Item_window_func *win_func_with_longest_order= NULL; int longest_order_elements= -1; @@ -3153,6 +3331,8 @@ bool Window_funcs_sort::setup(THD *thd, SQL_SELECT *sel, in a way that the result is valid for all window functions belonging to this Window_funcs_sort. */ + // all this i should have done earlier for streaming (on base table, or + // reusing the main query order (for later)) spec= win_func_with_longest_order->window_spec; ORDER* sort_order= concat_order_lists(thd->mem_root, @@ -3204,6 +3384,11 @@ bool Window_funcs_computation::setup(THD *thd, filtering conditions when we perform sorting for window function computation. */ + // sel holds filtering conditions (where, HAVING), those happen already + // before the window function computation, when a window function is computed + // over some window, we have to respect those filters, and not operate over + // the whole window (it's needed, the temp table does have invalid rows, + // checked with gdb test main.win (does not respect having)). if (tab->filesort && tab->filesort->select) { sel= tab->filesort->select; @@ -3234,6 +3419,7 @@ bool Window_funcs_computation::exec(JOIN *join, bool keep_last_filesort_result) while ((srt = it++)) { counter++; + // hmm? bool keep_filesort_result= keep_last_filesort_result && counter == win_func_sorts.elements; if (srt->exec(join, keep_filesort_result)) @@ -3254,6 +3440,73 @@ void Window_funcs_computation::cleanup() } } +// assume now all windows are valid +// setup cursor managers for handling partitions and computations. +bool Window_funcs_sort_streaming::setup(THD *thd, + List &window_funcs) +{ + // note that this would be called before as it is criteria for streaming + // we are expected to have one valid window, that is, we do not care about + // the internal markers this function sets (MARKER_SORTORDER_CHANGE, etc) + // even the longest order would have been already applied to the table at + // this point, we should not care about this here + order_window_funcs_by_window_specs(&window_funcs); + + List_iterator_fast it(window_funcs); + Item_window_func *win_func; + get_window_functions_required_cursors(thd, window_funcs, &cursor_managers); + // we need partition trackers too, should be here in setup + while ((win_func= it++)) + { + Group_bound_tracker *tracker= + new Group_bound_tracker(thd, win_func->window_spec->partition_list); + tracker->init(); + partition_trackers.push_back(tracker); + win_func->set_phase_to_computation(); + + // sets peer tracker inside rank() + Item_sum *sum_func= win_func->window_func(); + sum_func->setup_window_func(thd, win_func->window_spec); + } + this->win_funcs= window_funcs; // internal variable points to the list + return false; +} + +// called as a callback for each row in the join loop last table, directly +// before end_send() +bool Window_funcs_sort_streaming::process_row() +{ + List_iterator_fast iter_win_funcs(win_funcs); + List_iterator_fast iter_part_trackers( + partition_trackers); + List_iterator_fast iter_cursor_managers(cursor_managers); + Item_window_func *win_func; + Cursor_manager *cursor_manager; + Group_bound_tracker *tracker; + // i copied this for now from compute_window_func + while ((win_func= iter_win_funcs++) && (tracker= iter_part_trackers++) && + (cursor_manager= iter_cursor_managers++)) + { + if (tracker->check_if_next_group() || (rownum == 0)) + { + /* TODO(cvicentiu) + Clearing window functions should happen through cursors. */ + win_func->window_func()->clear(); + cursor_manager->notify_cursors_partition_changed(rownum); + } + else + { + cursor_manager->notify_cursors_next_row(); + } + + /* Check if we found any error in the window function while adding values + through cursors. */ + if (unlikely(current_thd->is_error() || current_thd->is_killed())) + return true; + } + rownum++; + return false; +} Explain_aggr_window_funcs* Window_funcs_computation::save_explain_plan(MEM_ROOT *mem_root, diff --git a/sql/sql_window.h b/sql/sql_window.h index 7009b8895a667..5b4b64d28cb31 100644 --- a/sql/sql_window.h +++ b/sql/sql_window.h @@ -20,7 +20,10 @@ #include "filesort.h" class Item_window_func; - +class Item_sum; +class Group_bound_tracker; +class Frame_cursor; +class Cursor_manager; /* Window functions module. @@ -181,6 +184,10 @@ int setup_windows(THD *thd, Ref_ptr_array ref_pointer_array, TABLE_LIST *tables, List &fields, List &all_fields, List &win_specs, List &win_funcs); +bool have_streaming_window_funcs(THD *thd, List &win_funcs, + ORDER *&longest_wf_order, + ORDER *main_query_order, + bool &streaming_wf_order_is_longer); ////////////////////////////////////////////////////////////////////////////// // Classes that make window functions computation a part of SELECT's query plan @@ -256,5 +263,25 @@ class Window_funcs_computation : public Sql_alloc void cleanup(); }; +// I think the only need for the object is to hold the group_bound_trackers, we +// don't even need the functions list +class Window_funcs_sort_streaming : public Sql_alloc +{ +public: + bool setup(THD *thd, List &win_funcs); + bool process_row(); // this object is attached to the JOIN, and + // process_row() is called for a method attached on + // takes the window funcs and the current row by + // end_compute_win_funcs() and calls the appropriate + // cursors to update the aggregate functions + +private: + int rownum= 0; // acts like internal state for process row + List win_funcs; + // these correspond to the window functions in the SELECT_LEX (all functions + // are streamable) + List cursor_managers; + List partition_trackers; +}; #endif /* SQL_WINDOW_INCLUDED */ From 5559d964a133e39bd37f4232834663278ca12afa Mon Sep 17 00:00:00 2001 From: OmarGamal10 Date: Tue, 23 Jun 2026 04:37:04 +0300 Subject: [PATCH 02/16] Wire the streaming window functions path into the existing join loop --- sql/sql_lex.cc | 2 +- sql/sql_select.cc | 61 +++++++++++++++++++++++++++++++++++++++++++++-- sql/sql_select.h | 30 +++++++++++++++++++---- 3 files changed, 85 insertions(+), 8 deletions(-) diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc index 86555657deea8..82e06d6005098 100644 --- a/sql/sql_lex.cc +++ b/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; } diff --git a/sql/sql_select.cc b/sql/sql_select.cc index b2c6929f20311..e3e00805ee825 100644 --- a/sql/sql_select.cc +++ b/sql/sql_select.cc @@ -26,6 +26,7 @@ */ #include "mariadb.h" +#include "sql_list.h" #include "sql_priv.h" #include "unireg.h" #include "sql_select.h" @@ -71,6 +72,9 @@ #include "opt_hints.h" #include "opt_group_by_cardinality.h" +#include "sql_window.h" +#include "item_windowfunc.h" + /* A key part number that means we're using a fulltext scan. @@ -228,6 +232,8 @@ static enum_nested_loop_state end_update(JOIN *join, JOIN_TAB *join_tab, bool end_of_records); static enum_nested_loop_state end_unique_update(JOIN *join, JOIN_TAB *join_tab, bool end_of_records); +static enum_nested_loop_state +end_compute_win_func(JOIN *join, JOIN_TAB *join_tab, bool end_of_records); static int join_read_const_table(THD *thd, JOIN_TAB *tab, POSITION *pos); static int join_read_system(JOIN_TAB *tab); @@ -1600,6 +1606,7 @@ JOIN::prepare(TABLE_LIST *tables_init, COND *conds_init, uint og_num, DBUG_RETURN(-1); thd->lex->current_select->context_analysis_place= save_place; + // this sets window functions up if (setup_without_group(thd, ref_ptrs, tables_list, select_lex->leaf_tables, fields_list, all_fields, &conds, order, group_list, @@ -1608,6 +1615,14 @@ JOIN::prepare(TABLE_LIST *tables_init, COND *conds_init, uint og_num, &hidden_group_fields)) DBUG_RETURN(-1); + // this needs to decide compatibility with main query sorting if exists + // but the actual setting of which is set before test_if_need_tmp_table() + if (select_lex->n_sum_items == select_lex->window_funcs.elements && + have_streaming_window_funcs(thd, select_lex->window_funcs, + win_func_longest_order, order, + streaming_wf_order_is_longer)) + streamable_window_funcs= true; + /* Permanently remove redundant parts from the query if 1) This is a subquery @@ -3338,8 +3353,14 @@ int JOIN::optimize_stage2() 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() && !streamable_window_funcs) simple_order= FALSE; + // this means the order by should be done in a temp table (it's real purpose + // is checking if order by references only the first non-const table in JOIN) + + // i'm not very sure of this, simple_order might change later?? + if (!need_tmp && simple_order && streaming_wf_order_is_longer) + order= win_func_longest_order; /* If the hint FORCE INDEX FOR ORDER BY/GROUP BY is used for the table @@ -3576,6 +3597,27 @@ int JOIN::optimize_stage2() if (make_aggr_tables_info()) DBUG_RETURN(1); + if (streamable_window_funcs && !need_tmp) + { + JOIN_TAB *last_real_tab= &join_tab[exec_join_tab_cnt() - 1]; + // here i would attach the new streamable class (same interface ?) + if (!(last_real_tab->window_funcs_streaming_step= + new Window_funcs_sort_streaming)) + DBUG_RETURN(true); + // this sets up the list, and the partition and group tracking + // I think we would have called order_window_funcs_by_window_specs() once + // in preparation already to decide on streaming or not? + if (last_real_tab->window_funcs_streaming_step->setup( + thd, select_lex->window_funcs)) + DBUG_RETURN(true); + // i need to make SURE THAT END_SEND is not assigned to last table after + // this, this is very important + last_real_tab->next_select= + end_compute_win_func; // calls process_row and end_send + /* Count that we're using window functions. */ + status_var_increment(thd->status_var.feature_window_functions); + } + init_join_cache_and_keyread(); if (init_range_rowid_filters()) @@ -4327,7 +4369,7 @@ bool JOIN::make_aggr_tables_info() - duplicate value removal Both of these operations are done after window function computation step. */ - if (select_lex->window_funcs.elements) + if (select_lex->window_funcs.elements && need_tmp) { curr_tab= join_tab + total_join_tab_cnt(); if (!(curr_tab->window_funcs_step= new Window_funcs_computation)) @@ -24949,6 +24991,7 @@ evaluate_join_record(JOIN *join, JOIN_TAB *join_tab, { enum enum_nested_loop_state rc; /* A match from join_tab is found for the current partial join. */ + // this is the loop rc= (*join_tab->next_select)(join, join_tab+1, 0); join->thd->get_stmt_da()->inc_current_row_for_warning(); if (rc != NESTED_LOOP_OK && rc != NESTED_LOOP_NO_MORE_ROWS) @@ -26162,6 +26205,20 @@ end_send(JOIN *join, JOIN_TAB *join_tab, bool end_of_records) DBUG_RETURN(NESTED_LOOP_OK); } +enum_nested_loop_state end_compute_win_func(JOIN *join, JOIN_TAB *join_tab, + bool end_of_records) +{ + // this show call process_row with the current row, and the list of window + // functions, process row runs cursors for wfs on the current row (will + // partition trackers work?) + // Then end_send would call the window_func()->val_*() so we need phase + // computation to read the live value + // we don't even need to pass the row to the window function, because the + // add() functions read from the TABLE::record[0] directly, as we did + // NOT call split_sum_func(), so we still point to base table + (join_tab - 1)->window_funcs_streaming_step->process_row(); + return end_send(join, join_tab, end_of_records); +} /* @brief diff --git a/sql/sql_select.h b/sql/sql_select.h index 7216938ebfd00..348b69f9eef92 100644 --- a/sql/sql_select.h +++ b/sql/sql_select.h @@ -200,6 +200,8 @@ enum join_type { JT_UNKNOWN,JT_SYSTEM,JT_CONST,JT_EQ_REF,JT_REF,JT_MAYBE_REF, class JOIN; +class Window_funcs_sort_streaming; + enum enum_nested_loop_state { NESTED_LOOP_KILLED= -2, NESTED_LOOP_ERROR= -1, @@ -533,6 +535,12 @@ typedef struct st_join_table { */ Window_funcs_computation* window_funcs_step; + /* + Non-NULL value means this join_tab (last real table) must do stream window + function computation before sending + */ + Window_funcs_sort_streaming *window_funcs_streaming_step; + /** List of topmost expressions in the select list. The *next* JOIN_TAB in the plan should use it to obtain correct values. Same applicable to @@ -1753,9 +1761,20 @@ class JOIN :public Sql_alloc */ Sql_cmd_dml *sql_cmd_dml; + /* + True if the query has window functions passing the streaming criteria, + defined by have_streaming_window_funcs() + Note: this does not guarantee they will be streamed, if the query requires + a temp table for any other reason, the window functions follow the + materialization path. + */ + bool streamable_window_funcs= false; + ORDER *win_func_longest_order= NULL; + bool streaming_wf_order_is_longer= false; + JOIN(THD *thd_arg, List &fields_arg, ulonglong select_options_arg, select_result *result_arg) - :fields_list(fields_arg) + : fields_list(fields_arg) { init(thd_arg, fields_arg, select_options_arg, result_arg); } @@ -1905,11 +1924,12 @@ class JOIN :public Sql_alloc 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))) || + ((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()); + (select_lex->have_window_funcs() && + (!streamable_window_funcs || only_const_tables()))); } bool choose_subquery_plan(table_map join_tables); void get_partial_cost_and_fanout(int end_tab_idx, From cc86ea93308bc701f3a9fee195cd65edea3911d3 Mon Sep 17 00:00:00 2001 From: OmarGamal10 Date: Tue, 23 Jun 2026 17:20:58 +0300 Subject: [PATCH 03/16] Remove order_window_funcs_by_window_specs from preparation time, and just check ordering compatibility across window functions instead --- sql/sql_window.cc | 113 ++++++++++++++++++++++------------------------ 1 file changed, 54 insertions(+), 59 deletions(-) diff --git a/sql/sql_window.cc b/sql/sql_window.cc index 61f7c461acec4..1230a950a821c 100644 --- a/sql/sql_window.cc +++ b/sql/sql_window.cc @@ -785,17 +785,11 @@ typedef int (*Item_window_func_cmp)(Item_window_func *f1, The changes between the groups are marked by setting item_window_func->marker. */ -// I think i can move this to preparation -// run this in preparation, along with other criteria and set some -// is_streamable variable on JOIN -// then test_if_need_tmp_table would check it too. -// Returns true if only one sort exists, false if none or more -static bool +static void order_window_funcs_by_window_specs(List *win_func_list) { if (win_func_list->elements == 0) - return false; - bool more_than_one_sort= false; + return; bubble_sort(win_func_list, compare_window_funcs_by_window_specs, NULL); @@ -824,10 +818,8 @@ order_window_funcs_by_window_specs(List *win_func_list) cmp= compare_window_spec_joined_lists(win_spec_prev, win_spec_curr); if (!(CMP_LT_C <= cmp && cmp <= CMP_GT_C)) { - curr->marker= (MARKER_SORTORDER_CHANGE | - MARKER_PARTITION_CHANGE | + curr->marker= (MARKER_SORTORDER_CHANGE | MARKER_PARTITION_CHANGE | MARKER_FRAME_CHANGE); - more_than_one_sort= true; } else if (win_spec_prev->partition_list != win_spec_curr->partition_list) { @@ -839,7 +831,6 @@ order_window_funcs_by_window_specs(List *win_func_list) prev= curr; } - return !more_than_one_sort; } static inline bool frame_is_current_row(Window_spec *win_spec) @@ -865,9 +856,49 @@ static inline bool check_argument_list_aggregation(Window_spec *win_spec) return false; } +static Item_window_func * +find_longest_compatible_order(List_iterator_fast &it) +{ + int longest_order_elements= -1; + Item_window_func *longest, *win_func; + while ((win_func= it++)) + { + Window_spec *spec= win_func->window_spec; + int win_func_order_elements= + spec->partition_list->elements + spec->order_list->elements; + if (win_func_order_elements > longest_order_elements) + { + longest_order_elements= win_func_order_elements; + longest= win_func; + } + } + it.rewind(); + + Window_spec *longest_spec= longest->window_spec; + longest_spec->join_partition_and_order_lists(); + while ((win_func= it++)) + { + if (win_func == longest) + continue; + Window_spec *spec= win_func->window_spec; + spec->join_partition_and_order_lists(); + int cmp= compare_order_lists(longest_spec->partition_list, + longest_spec->win_spec_number, + spec->partition_list, spec->win_spec_number); + spec->disjoin_partition_and_order_lists(); + if (!(CMP_LT_C <= cmp && cmp <= CMP_GT_C)) + { + longest_spec->disjoin_partition_and_order_lists(); + return NULL; + } + } + longest_spec->disjoin_partition_and_order_lists(); + return longest; +} + // now I know number 1 is bad (side effect, but i need it for criteria, why run // two times?) -// 1. Runs order_window_funcs_by_window_specs(), so now we sort in preparation +// 1. Checks if all window function orderings are compatible. // 2. We check each fucntion from our subset or no (let it be rank and // row_number for now) // 3. frame only current row (normal), or unbounded preceding (for sum @@ -881,30 +912,24 @@ bool have_streaming_window_funcs(THD *thd, List &win_funcs, // This checks if more than one SORTORDER_MARKER_CHANGE exists. // Calling order early here has a problem with one of the existing tests. Not // sure why. - if (win_funcs.elements == 0 || - !order_window_funcs_by_window_specs(&win_funcs)) + if (win_funcs.elements == 0) return false; + List_iterator_fast it(win_funcs); + Item_window_func *win_func_with_longest_order= + find_longest_compatible_order(it); + if (!win_func_with_longest_order) + return false; + Item_window_func *win_func; - Item_window_func *win_func_with_longest_order= NULL; - int longest_order_elements= -1; int cmp; + it.rewind(); while ((win_func= it++)) { Window_spec *spec= win_func->window_spec; - if (check_argument_list_aggregation(spec)) - return false; - - int win_func_order_elements= - spec->partition_list->elements + spec->order_list->elements; - if (win_func_order_elements > longest_order_elements) - { - longest_order_elements= win_func_order_elements; - win_func_with_longest_order= win_func; - } - - if (!(win_func->window_func()->is_streamable() && + if (check_argument_list_aggregation(spec) || + !(win_func->window_func()->is_streamable() && frame_is_current_row(win_func->window_spec))) return false; } @@ -1405,29 +1430,6 @@ class Cursor_manager List cursors; }; -// // I think the only need for the object is to hold the group_bound_trackers, -// we -// // don't even need the functions list -// class Window_funcs_sort_streaming : public Sql_alloc -// { -// public: -// bool setup(THD *thd, List &win_funcs); -// bool process_row(); // this object is attached to the JOIN, and -// // process_row() is called for a method attached on -// // takes the window funcs and the current row by -// // end_compute_win_funcs() and calls the appropriate -// // cursors to update the aggregate functions - -// private: -// int row_num= 0; // acts like internal state for process row -// List win_funcs; -// // these correspond to the window functions in the SELECT_LEX (all -// functions -// // are streamable) -// List cursor_managers; -// List partition_trackers; -// }; - ////////////////////////////////////////////////////////////////////////////// // RANGE-type frames ////////////////////////////////////////////////////////////////////////////// @@ -3440,16 +3442,9 @@ void Window_funcs_computation::cleanup() } } -// assume now all windows are valid -// setup cursor managers for handling partitions and computations. bool Window_funcs_sort_streaming::setup(THD *thd, List &window_funcs) { - // note that this would be called before as it is criteria for streaming - // we are expected to have one valid window, that is, we do not care about - // the internal markers this function sets (MARKER_SORTORDER_CHANGE, etc) - // even the longest order would have been already applied to the table at - // this point, we should not care about this here order_window_funcs_by_window_specs(&window_funcs); List_iterator_fast it(window_funcs); From bf2734eb6ae358b7ce284ed43cb7bb76d0a1109f Mon Sep 17 00:00:00 2001 From: OmarGamal10 Date: Tue, 23 Jun 2026 17:27:47 +0300 Subject: [PATCH 04/16] Fallback to materialization if GROUP BY due to regressions --- sql/sql_select.cc | 1 + 1 file changed, 1 insertion(+) diff --git a/sql/sql_select.cc b/sql/sql_select.cc index e3e00805ee825..32d590e6bb829 100644 --- a/sql/sql_select.cc +++ b/sql/sql_select.cc @@ -1618,6 +1618,7 @@ JOIN::prepare(TABLE_LIST *tables_init, COND *conds_init, uint og_num, // this needs to decide compatibility with main query sorting if exists // but the actual setting of which is set before test_if_need_tmp_table() if (select_lex->n_sum_items == select_lex->window_funcs.elements && + select_lex->group_list.elements == 0 && have_streaming_window_funcs(thd, select_lex->window_funcs, win_func_longest_order, order, streaming_wf_order_is_longer)) From d3e800d05f1739b92710c942c83b757291f1616b Mon Sep 17 00:00:00 2001 From: OmarGamal10 Date: Thu, 25 Jun 2026 03:44:19 +0300 Subject: [PATCH 05/16] Add COUNT() as a streamable window function --- sql/item_sum.h | 1 + 1 file changed, 1 insertion(+) diff --git a/sql/item_sum.h b/sql/item_sum.h index f1f1ac3863090..a4f2251bd1546 100644 --- a/sql/item_sum.h +++ b/sql/item_sum.h @@ -896,6 +896,7 @@ class Item_sum_count :public Item_sum_int bool add() override; void cleanup() override; void remove() override; + inline bool is_streamable() const override { return true; } public: Item_sum_count(THD *thd, Item *item_par): From 8f87283b9e86942b8275e301db9a1f6e9e55b5a2 Mon Sep 17 00:00:00 2001 From: OmarGamal10 Date: Thu, 25 Jun 2026 03:45:45 +0300 Subject: [PATCH 06/16] Add UNBOUNDED PRECEDING as an explicit valid frame for streaming (relied on default definition before) --- sql/item_windowfunc.h | 6 +++--- sql/sql_window.cc | 20 +++++++++++--------- 2 files changed, 14 insertions(+), 12 deletions(-) diff --git a/sql/item_windowfunc.h b/sql/item_windowfunc.h index 90a3f73e30eef..994d85525a6c5 100644 --- a/sql/item_windowfunc.h +++ b/sql/item_windowfunc.h @@ -150,7 +150,7 @@ class Item_sum_row_number: public Item_sum_int return name; } - bool inline is_streamable() const override { return true; } + inline bool is_streamable() const override { return true; } protected: Item *shallow_copy(THD *thd) const override @@ -217,7 +217,7 @@ class Item_sum_rank: public Item_sum_int return name; } - bool inline is_streamable() const override { return true; } + inline bool is_streamable() const override { return true; } void setup_window_func(THD *thd, Window_spec *window_spec) override; @@ -294,7 +294,7 @@ class Item_sum_dense_rank: public Item_sum_int return name; } - bool inline is_streamable() const override { return true; } + inline bool is_streamable() const override { return true; } void setup_window_func(THD *thd, Window_spec *window_spec) override; diff --git a/sql/sql_window.cc b/sql/sql_window.cc index 1230a950a821c..09a924fb04727 100644 --- a/sql/sql_window.cc +++ b/sql/sql_window.cc @@ -835,10 +835,14 @@ order_window_funcs_by_window_specs(List *win_func_list) static inline bool frame_is_current_row(Window_spec *win_spec) { - // null is default, range between unbounded preceding and current row - return win_spec->window_frame == NULL || - (win_spec->window_frame->top_bound->precedence_type == - Window_frame_bound::CURRENT && + Window_frame *frame= win_spec->window_frame; + if (!frame) + return true; + bool unbounded_preceding_or_current= + frame->top_bound->precedence_type == Window_frame_bound::CURRENT || + (frame->top_bound->precedence_type == Window_frame_bound::PRECEDING && + frame->top_bound->is_unbounded()); + return (unbounded_preceding_or_current && win_spec->window_frame->bottom_bound->precedence_type == Window_frame_bound::CURRENT); } @@ -886,18 +890,16 @@ find_longest_compatible_order(List_iterator_fast &it) longest_spec->win_spec_number, spec->partition_list, spec->win_spec_number); spec->disjoin_partition_and_order_lists(); - if (!(CMP_LT_C <= cmp && cmp <= CMP_GT_C)) + if (cmp != CMP_GT_C) { - longest_spec->disjoin_partition_and_order_lists(); - return NULL; + longest= NULL; + break; } } longest_spec->disjoin_partition_and_order_lists(); return longest; } -// now I know number 1 is bad (side effect, but i need it for criteria, why run -// two times?) // 1. Checks if all window function orderings are compatible. // 2. We check each fucntion from our subset or no (let it be rank and // row_number for now) From 9a7ea7143277bfd3d17f67b854de249e3ce42bb8 Mon Sep 17 00:00:00 2001 From: OmarGamal10 Date: Thu, 25 Jun 2026 03:46:52 +0300 Subject: [PATCH 07/16] Add initial testing file for streaming window functions --- mysql-test/main/win_streaming.result | 521 +++++++++++++++++++++++++++ mysql-test/main/win_streaming.test | 114 ++++++ 2 files changed, 635 insertions(+) create mode 100644 mysql-test/main/win_streaming.result create mode 100644 mysql-test/main/win_streaming.test diff --git a/mysql-test/main/win_streaming.result b/mysql-test/main/win_streaming.result new file mode 100644 index 0000000000000..80e58b416e117 --- /dev/null +++ b/mysql-test/main/win_streaming.result @@ -0,0 +1,521 @@ +CREATE TABLE t1 (pk INT PRIMARY KEY, a INT, b INT); +INSERT INTO t1 VALUES (1, 1, 3); +INSERT INTO t1 VALUES (2, 1, 1); +INSERT INTO t1 VALUES (3, 2, 2); +INSERT INTO t1 VALUES (4, 2, 4); +INSERT INTO t1 VALUES (5, 3, 1); +INSERT INTO t1 VALUES (6, 3, 2); +SELECT pk, RANK() OVER (ORDER BY pk) AS rnk from t1; +pk rnk +1 1 +2 2 +3 3 +4 4 +5 5 +6 6 +EXPLAIN FORMAT=JSON SELECT pk, RANK() OVER (ORDER BY pk) AS rnk from t1 limit 2; +EXPLAIN +{ + "query_block": { + "select_id": 1, + "cost": "COST_REPLACED", + "nested_loop": [ + { + "table": { + "table_name": "t1", + "access_type": "index", + "key": "PRIMARY", + "key_length": "4", + "used_key_parts": ["pk"], + "loops": 1, + "rows": 2, + "cost": "COST_REPLACED", + "filtered": 100, + "using_index": true + } + } + ] + } +} +EXPLAIN FORMAT=JSON SELECT rank() OVER (PARTITION BY a ORDER BY b) FROM t1; +EXPLAIN +{ + "query_block": { + "select_id": 1, + "cost": "COST_REPLACED", + "nested_loop": [ + { + "read_sorted_file": { + "filesort": { + "sort_key": "t1.a, t1.b", + "table": { + "table_name": "t1", + "access_type": "ALL", + "loops": 1, + "rows": 6, + "cost": "COST_REPLACED", + "filtered": 100 + } + } + } + } + ] + } +} +SELECT row_number() OVER (PARTITION BY a ORDER BY b) as rn FROM t1; +rn +1 +2 +1 +2 +1 +2 +SELECT rank() OVER (PARTITION BY a ORDER BY b) as rnk FROM t1; +rnk +1 +2 +1 +2 +1 +2 +SELECT dense_rank() OVER (PARTITION BY a ORDER BY b) as drnk FROM t1; +drnk +1 +2 +1 +2 +1 +2 +EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a), dense_rank() OVER (ORDER BY a) FROM t1; +EXPLAIN +{ + "query_block": { + "select_id": 1, + "cost": "COST_REPLACED", + "window_functions_computation": { + "sorts": [ + { + "filesort": { + "sort_key": "t1.a" + } + } + ], + "temporary_table": { + "nested_loop": [ + { + "table": { + "table_name": "t1", + "access_type": "ALL", + "loops": 1, + "rows": 6, + "cost": "COST_REPLACED", + "filtered": 100 + } + } + ] + } + } + } +} +SELECT rank() OVER (ORDER BY a) as rnk, dense_rank() OVER (ORDER BY a) as drnk FROM t1; +rnk drnk +1 1 +1 1 +3 2 +3 2 +5 3 +5 3 +EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a, b) FROM t1 ORDER BY a; +EXPLAIN +{ + "query_block": { + "select_id": 1, + "cost": "COST_REPLACED", + "nested_loop": [ + { + "read_sorted_file": { + "filesort": { + "sort_key": "t1.a, t1.b", + "table": { + "table_name": "t1", + "access_type": "ALL", + "loops": 1, + "rows": 6, + "cost": "COST_REPLACED", + "filtered": 100 + } + } + } + } + ] + } +} +SELECT rank() OVER (ORDER BY a, b) as rnk FROM t1 ORDER BY a; +rnk +1 +2 +3 +4 +5 +6 +EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a) FROM t1 ORDER BY a, b; +EXPLAIN +{ + "query_block": { + "select_id": 1, + "cost": "COST_REPLACED", + "nested_loop": [ + { + "read_sorted_file": { + "filesort": { + "sort_key": "t1.a, t1.b", + "table": { + "table_name": "t1", + "access_type": "ALL", + "loops": 1, + "rows": 6, + "cost": "COST_REPLACED", + "filtered": 100 + } + } + } + } + ] + } +} +SELECT rank() OVER (ORDER BY a) as rnk FROM t1 ORDER BY a, b; +rnk +1 +1 +3 +3 +5 +5 +EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a) FROM t1 ORDER BY a; +EXPLAIN +{ + "query_block": { + "select_id": 1, + "cost": "COST_REPLACED", + "nested_loop": [ + { + "read_sorted_file": { + "filesort": { + "sort_key": "t1.a", + "table": { + "table_name": "t1", + "access_type": "ALL", + "loops": 1, + "rows": 6, + "cost": "COST_REPLACED", + "filtered": 100 + } + } + } + } + ] + } +} +SELECT rank() OVER (ORDER BY a) as rnk FROM t1 ORDER BY a; +rnk +1 +1 +3 +3 +5 +5 +EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a), rank() OVER (ORDER BY b) FROM t1; +EXPLAIN +{ + "query_block": { + "select_id": 1, + "cost": "COST_REPLACED", + "window_functions_computation": { + "sorts": [ + { + "filesort": { + "sort_key": "t1.b" + } + }, + { + "filesort": { + "sort_key": "t1.a" + } + } + ], + "temporary_table": { + "nested_loop": [ + { + "table": { + "table_name": "t1", + "access_type": "ALL", + "loops": 1, + "rows": 6, + "cost": "COST_REPLACED", + "filtered": 100 + } + } + ] + } + } + } +} +EXPLAIN FORMAT=JSON SELECT rank() OVER (PARTITION BY max(a) ORDER BY b) FROM t1; +EXPLAIN +{ + "query_block": { + "select_id": 1, + "cost": "COST_REPLACED", + "window_functions_computation": { + "sorts": [ + { + "filesort": { + "sort_key": "max(t1.a), t1.b" + } + } + ], + "temporary_table": { + "nested_loop": [ + { + "table": { + "table_name": "t1", + "access_type": "ALL", + "loops": 1, + "rows": 6, + "cost": "COST_REPLACED", + "filtered": 100 + } + } + ] + } + } + } +} +EXPLAIN FORMAT=JSON SELECT max(a), rank() OVER (ORDER BY b) FROM t1; +EXPLAIN +{ + "query_block": { + "select_id": 1, + "cost": "COST_REPLACED", + "window_functions_computation": { + "sorts": [ + { + "filesort": { + "sort_key": "t1.b" + } + } + ], + "temporary_table": { + "nested_loop": [ + { + "table": { + "table_name": "t1", + "access_type": "ALL", + "loops": 1, + "rows": 6, + "cost": "COST_REPLACED", + "filtered": 100 + } + } + ] + } + } + } +} +EXPLAIN FORMAT=JSON SELECT count(*) OVER (ORDER BY a RANGE BETWEEN CURRENT ROW AND 5 FOLLOWING) FROM t1; +EXPLAIN +{ + "query_block": { + "select_id": 1, + "cost": "COST_REPLACED", + "window_functions_computation": { + "sorts": [ + { + "filesort": { + "sort_key": "t1.a" + } + } + ], + "temporary_table": { + "nested_loop": [ + { + "table": { + "table_name": "t1", + "access_type": "ALL", + "loops": 1, + "rows": 6, + "cost": "COST_REPLACED", + "filtered": 100 + } + } + ] + } + } + } +} +EXPLAIN FORMAT=JSON SELECT count(*) OVER (ORDER BY a RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) FROM t1; +EXPLAIN +{ + "query_block": { + "select_id": 1, + "cost": "COST_REPLACED", + "nested_loop": [ + { + "read_sorted_file": { + "filesort": { + "sort_key": "t1.a", + "table": { + "table_name": "t1", + "access_type": "ALL", + "loops": 1, + "rows": 6, + "cost": "COST_REPLACED", + "filtered": 100 + } + } + } + } + ] + } +} +EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a) FROM t1 GROUP BY a; +EXPLAIN +{ + "query_block": { + "select_id": 1, + "cost": "COST_REPLACED", + "filesort": { + "sort_key": "t1.a", + "window_functions_computation": { + "sorts": [ + { + "filesort": { + "sort_key": "t1.a" + } + } + ], + "temporary_table": { + "nested_loop": [ + { + "table": { + "table_name": "t1", + "access_type": "ALL", + "loops": 1, + "rows": 6, + "cost": "COST_REPLACED", + "filtered": 100 + } + } + ] + } + } + } + } +} +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); +EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY t1.b) FROM t1 JOIN t2 ON t1.pk = t2.pk; +EXPLAIN +{ + "query_block": { + "select_id": 1, + "cost": "COST_REPLACED", + "nested_loop": [ + { + "read_sorted_file": { + "filesort": { + "sort_key": "t1.b", + "table": { + "table_name": "t1", + "access_type": "ALL", + "possible_keys": ["PRIMARY"], + "loops": 1, + "rows": 6, + "cost": "COST_REPLACED", + "filtered": 100 + } + } + } + }, + { + "table": { + "table_name": "t2", + "access_type": "eq_ref", + "possible_keys": ["PRIMARY"], + "key": "PRIMARY", + "key_length": "4", + "used_key_parts": ["pk"], + "ref": ["test.t1.pk"], + "loops": 6, + "rows": 1, + "cost": "COST_REPLACED", + "filtered": 100, + "using_index": true + } + } + ] + } +} +SELECT rank() OVER (ORDER BY t1.b) as rnk FROM t1 JOIN t2 ON t1.pk = t2.pk; +rnk +1 +1 +3 +3 +5 +6 +SELECT rank() OVER (PARTITION BY t1.a ORDER BY t1.b) as rnk FROM t1 JOIN t2 ON t1.pk = t2.pk; +rnk +1 +2 +1 +2 +1 +2 +SELECT rank() OVER (ORDER BY t1.b) as rnk FROM t1 LEFT JOIN t2 ON t1.pk = t2.pk; +rnk +1 +1 +3 +3 +5 +6 +DROP TABLE t2; +EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a) as rnk FROM (SELECT a FROM t1 WHERE a > 1) derived; +EXPLAIN +{ + "query_block": { + "select_id": 1, + "cost": "COST_REPLACED", + "nested_loop": [ + { + "read_sorted_file": { + "filesort": { + "sort_key": "t1.a", + "table": { + "table_name": "t1", + "access_type": "ALL", + "loops": 1, + "rows": 6, + "cost": "COST_REPLACED", + "filtered": 100, + "attached_condition": "t1.a > 1" + } + } + } + } + ] + } +} +SELECT rank() OVER (ORDER BY a) as rnk FROM (SELECT a FROM t1 WHERE a > 1) derived; +rnk +1 +1 +3 +3 +DROP TABLE t1; diff --git a/mysql-test/main/win_streaming.test b/mysql-test/main/win_streaming.test new file mode 100644 index 0000000000000..3f0123534f82b --- /dev/null +++ b/mysql-test/main/win_streaming.test @@ -0,0 +1,114 @@ +# +# 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 + +# For now, row_number(), rank(), dense_rank() are streamable. +# should I only test on rank? would it make sense to add EXPLAIN for all? + +# Basic streaming (one rank with explain, others only correctness) + +# 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 + +CREATE TABLE t1 (pk INT PRIMARY KEY, a INT, b INT); +INSERT INTO t1 VALUES (1, 1, 3); +INSERT INTO t1 VALUES (2, 1, 1); +INSERT INTO t1 VALUES (3, 2, 2); +INSERT INTO t1 VALUES (4, 2, 4); +INSERT INTO t1 VALUES (5, 3, 1); +INSERT INTO t1 VALUES (6, 3, 2); + +# Basic streaming: one rank with explain, others only correctness +SELECT pk, RANK() OVER (ORDER BY pk) AS rnk from t1; +--source include/explain-no-costs.inc +EXPLAIN FORMAT=JSON SELECT pk, RANK() OVER (ORDER BY pk) AS rnk from t1 limit 2; + +# question: should i add test for other streamable functions? + +# Order by and partition: explain with rank to show streamable +--source include/explain-no-costs.inc +EXPLAIN FORMAT=JSON SELECT rank() OVER (PARTITION BY a ORDER BY b) FROM t1; +SELECT row_number() OVER (PARTITION BY a ORDER BY b) as rn FROM t1; +SELECT rank() OVER (PARTITION BY a ORDER BY b) as rnk FROM t1; +SELECT dense_rank() OVER (PARTITION BY a ORDER BY b) as drnk FROM t1; + +# Show compatible functions also stream +--source include/explain-no-costs.inc +EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a), dense_rank() OVER (ORDER BY a) FROM t1; +SELECT rank() OVER (ORDER BY a) as rnk, dense_rank() OVER (ORDER BY a) as drnk FROM t1; + +# Windows reusing the main query order (whichever is longer) +--source include/explain-no-costs.inc +EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a, b) FROM t1 ORDER BY a; +SELECT rank() OVER (ORDER BY a, b) as rnk FROM t1 ORDER BY a; + +--source include/explain-no-costs.inc +EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a) FROM t1 ORDER BY a, b; +SELECT rank() OVER (ORDER BY a) as rnk FROM t1 ORDER BY a, b; + +--source include/explain-no-costs.inc +EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a) FROM t1 ORDER BY a; +SELECT rank() OVER (ORDER BY a) as rnk FROM t1 ORDER BY a; + +# Incompatible orders materialize +--source include/explain-no-costs.inc +EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a), rank() OVER (ORDER BY b) FROM t1; + +# Aggregate functions inside partition lists materialize +--source include/explain-no-costs.inc +EXPLAIN FORMAT=JSON SELECT rank() OVER (PARTITION BY max(a) ORDER BY b) FROM t1; + +# Aggregate functions anywhere in the select list materialize +--source include/explain-no-costs.inc +EXPLAIN FORMAT=JSON SELECT max(a), rank() OVER (ORDER BY b) FROM t1; + +# Incompatible RANGE frame materializes +--source include/explain-no-costs.inc +EXPLAIN FORMAT=JSON SELECT count(*) OVER (ORDER BY a RANGE BETWEEN CURRENT ROW AND 5 FOLLOWING) FROM t1; + +--source include/explain-no-costs.inc +EXPLAIN FORMAT=JSON SELECT count(*) OVER (ORDER BY a RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) FROM t1; + +# GROUP BY materializes (for now?) +--source include/explain-no-costs.inc +EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a) FROM t1 GROUP BY a; + +# Multi-table join +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); + +--source include/explain-no-costs.inc +EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY t1.b) FROM t1 JOIN t2 ON t1.pk = t2.pk; +SELECT rank() OVER (ORDER BY t1.b) as rnk FROM t1 JOIN t2 ON t1.pk = t2.pk; +SELECT rank() OVER (PARTITION BY t1.a ORDER BY t1.b) as rnk FROM t1 JOIN t2 ON t1.pk = t2.pk; +SELECT rank() OVER (ORDER BY t1.b) as rnk FROM t1 LEFT JOIN t2 ON t1.pk = t2.pk; + +DROP TABLE t2; + +# Subquery in FROM +--source include/explain-no-costs.inc +EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a) as rnk FROM (SELECT a FROM t1 WHERE a > 1) derived; + +# Cases to look harder later (subqueries, expressions) + +DROP TABLE t1; From 227812db3167821c9249baa60d1258c0c274cf4d Mon Sep 17 00:00:00 2001 From: OmarGamal10 Date: Thu, 25 Jun 2026 04:13:10 +0300 Subject: [PATCH 08/16] Cache THD in Window_funcs_sort_streaming to avoid calling current_thd in process_row() --- mysql-test/main/win_streaming.result | 6 ------ sql/sql_select.cc | 6 ++---- sql/sql_window.cc | 3 +-- sql/sql_window.h | 6 ++++-- 4 files changed, 7 insertions(+), 14 deletions(-) diff --git a/mysql-test/main/win_streaming.result b/mysql-test/main/win_streaming.result index 80e58b416e117..146158ce433f3 100644 --- a/mysql-test/main/win_streaming.result +++ b/mysql-test/main/win_streaming.result @@ -512,10 +512,4 @@ EXPLAIN ] } } -SELECT rank() OVER (ORDER BY a) as rnk FROM (SELECT a FROM t1 WHERE a > 1) derived; -rnk -1 -1 -3 -3 DROP TABLE t1; diff --git a/sql/sql_select.cc b/sql/sql_select.cc index 32d590e6bb829..6218472e9301b 100644 --- a/sql/sql_select.cc +++ b/sql/sql_select.cc @@ -3603,13 +3603,11 @@ int JOIN::optimize_stage2() JOIN_TAB *last_real_tab= &join_tab[exec_join_tab_cnt() - 1]; // here i would attach the new streamable class (same interface ?) if (!(last_real_tab->window_funcs_streaming_step= - new Window_funcs_sort_streaming)) + new Window_funcs_sort_streaming(thd))) DBUG_RETURN(true); // this sets up the list, and the partition and group tracking - // I think we would have called order_window_funcs_by_window_specs() once - // in preparation already to decide on streaming or not? if (last_real_tab->window_funcs_streaming_step->setup( - thd, select_lex->window_funcs)) + select_lex->window_funcs)) DBUG_RETURN(true); // i need to make SURE THAT END_SEND is not assigned to last table after // this, this is very important diff --git a/sql/sql_window.cc b/sql/sql_window.cc index 09a924fb04727..235d5a7caa2f1 100644 --- a/sql/sql_window.cc +++ b/sql/sql_window.cc @@ -3444,8 +3444,7 @@ void Window_funcs_computation::cleanup() } } -bool Window_funcs_sort_streaming::setup(THD *thd, - List &window_funcs) +bool Window_funcs_sort_streaming::setup(List &window_funcs) { order_window_funcs_by_window_specs(&window_funcs); diff --git a/sql/sql_window.h b/sql/sql_window.h index 5b4b64d28cb31..f243632a36628 100644 --- a/sql/sql_window.h +++ b/sql/sql_window.h @@ -268,7 +268,8 @@ class Window_funcs_computation : public Sql_alloc class Window_funcs_sort_streaming : public Sql_alloc { public: - bool setup(THD *thd, List &win_funcs); + Window_funcs_sort_streaming(THD *thd) : thd(thd) {} + bool setup(List &win_funcs); bool process_row(); // this object is attached to the JOIN, and // process_row() is called for a method attached on // takes the window funcs and the current row by @@ -276,7 +277,8 @@ class Window_funcs_sort_streaming : public Sql_alloc // cursors to update the aggregate functions private: - int rownum= 0; // acts like internal state for process row + int rownum= 0; // internal state for process row + THD *thd= NULL; List win_funcs; // these correspond to the window functions in the SELECT_LEX (all functions // are streamable) From d3445679d4db043160892c6dcf8075b5cda7e434 Mon Sep 17 00:00:00 2001 From: OmarGamal10 Date: Thu, 25 Jun 2026 04:49:33 +0300 Subject: [PATCH 09/16] Add missing check in Window_funcs_sort_streaming::setup --- sql/sql_window.cc | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/sql/sql_window.cc b/sql/sql_window.cc index 235d5a7caa2f1..09732820fb542 100644 --- a/sql/sql_window.cc +++ b/sql/sql_window.cc @@ -911,9 +911,6 @@ bool have_streaming_window_funcs(THD *thd, List &win_funcs, ORDER *main_query_order, bool &streaming_wf_order_is_longer) { - // This checks if more than one SORTORDER_MARKER_CHANGE exists. - // Calling order early here has a problem with one of the existing tests. Not - // sure why. if (win_funcs.elements == 0) return false; @@ -3450,7 +3447,9 @@ bool Window_funcs_sort_streaming::setup(List &window_funcs) List_iterator_fast it(window_funcs); Item_window_func *win_func; - get_window_functions_required_cursors(thd, window_funcs, &cursor_managers); + if (get_window_functions_required_cursors(thd, window_funcs, + &cursor_managers)) + return true; // we need partition trackers too, should be here in setup while ((win_func= it++)) { From d1db37c3889f5658294e567dc184b8ea6a27ca4b Mon Sep 17 00:00:00 2001 From: OmarGamal10 Date: Tue, 30 Jun 2026 21:48:07 +0300 Subject: [PATCH 10/16] Add a cleanup method for Window_funcs_sort_streaming --- sql/sql_select.cc | 4 ++++ sql/sql_window.cc | 19 +++++++++++++++---- sql/sql_window.h | 7 ++----- 3 files changed, 21 insertions(+), 9 deletions(-) diff --git a/sql/sql_select.cc b/sql/sql_select.cc index 6218472e9301b..239bee8ecd299 100644 --- a/sql/sql_select.cc +++ b/sql/sql_select.cc @@ -16824,6 +16824,10 @@ void JOIN_TAB::cleanup() cache->free(); cache= 0; } + if (window_funcs_streaming_step) + { + window_funcs_streaming_step->cleanup(); + } limit= 0; // Free select that was created for filesort outside of create_sort_index if (filesort && filesort->select && !filesort->own_select) diff --git a/sql/sql_window.cc b/sql/sql_window.cc index 09732820fb542..2da868a158672 100644 --- a/sql/sql_window.cc +++ b/sql/sql_window.cc @@ -785,11 +785,12 @@ typedef int (*Item_window_func_cmp)(Item_window_func *f1, The changes between the groups are marked by setting item_window_func->marker. */ -static void -order_window_funcs_by_window_specs(List *win_func_list) +static +void order_window_funcs_by_window_specs(List *win_func_list) { if (win_func_list->elements == 0) return; + bubble_sort(win_func_list, compare_window_funcs_by_window_specs, NULL); @@ -818,7 +819,8 @@ order_window_funcs_by_window_specs(List *win_func_list) cmp= compare_window_spec_joined_lists(win_spec_prev, win_spec_curr); if (!(CMP_LT_C <= cmp && cmp <= CMP_GT_C)) { - curr->marker= (MARKER_SORTORDER_CHANGE | MARKER_PARTITION_CHANGE | + curr->marker= (MARKER_SORTORDER_CHANGE | + MARKER_PARTITION_CHANGE | MARKER_FRAME_CHANGE); } else if (win_spec_prev->partition_list != win_spec_curr->partition_list) @@ -3462,6 +3464,9 @@ bool Window_funcs_sort_streaming::setup(List &window_funcs) // sets peer tracker inside rank() Item_sum *sum_func= win_func->window_func(); sum_func->setup_window_func(thd, win_func->window_spec); + + win_func->window_func()->set_aggregator(thd, + Aggregator::SIMPLE_AGGREGATOR); } this->win_funcs= window_funcs; // internal variable points to the list return false; @@ -3496,13 +3501,19 @@ bool Window_funcs_sort_streaming::process_row() /* Check if we found any error in the window function while adding values through cursors. */ - if (unlikely(current_thd->is_error() || current_thd->is_killed())) + if (unlikely(thd->is_error() || thd->is_killed())) return true; } rownum++; return false; } +void Window_funcs_sort_streaming::cleanup() +{ + cursor_managers.delete_elements(); + partition_trackers.delete_elements(); +} + Explain_aggr_window_funcs* Window_funcs_computation::save_explain_plan(MEM_ROOT *mem_root, bool is_analyze) diff --git a/sql/sql_window.h b/sql/sql_window.h index f243632a36628..229c8c810b724 100644 --- a/sql/sql_window.h +++ b/sql/sql_window.h @@ -263,25 +263,22 @@ class Window_funcs_computation : public Sql_alloc void cleanup(); }; -// I think the only need for the object is to hold the group_bound_trackers, we -// don't even need the functions list class Window_funcs_sort_streaming : public Sql_alloc { public: Window_funcs_sort_streaming(THD *thd) : thd(thd) {} bool setup(List &win_funcs); - bool process_row(); // this object is attached to the JOIN, and + bool process_row(); // this object is attached to the JOIN_TAB, and // process_row() is called for a method attached on // takes the window funcs and the current row by // end_compute_win_funcs() and calls the appropriate // cursors to update the aggregate functions + void cleanup(); private: int rownum= 0; // internal state for process row THD *thd= NULL; List win_funcs; - // these correspond to the window functions in the SELECT_LEX (all functions - // are streamable) List cursor_managers; List partition_trackers; }; From 739015cfa218651ce07ab053a0a5037a4e76d8d7 Mon Sep 17 00:00:00 2001 From: OmarGamal10 Date: Tue, 30 Jun 2026 22:38:43 +0300 Subject: [PATCH 11/16] Refactor find_longest_compatible_order to take a const reference of the list instead of iterator --- sql/sql_window.cc | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/sql/sql_window.cc b/sql/sql_window.cc index 2da868a158672..cffdc8bac0b12 100644 --- a/sql/sql_window.cc +++ b/sql/sql_window.cc @@ -863,10 +863,14 @@ static inline bool check_argument_list_aggregation(Window_spec *win_spec) } static Item_window_func * -find_longest_compatible_order(List_iterator_fast &it) +find_longest_compatible_order(const List &win_funcs) { + if (win_funcs.elements == 0) + return nullptr; int longest_order_elements= -1; Item_window_func *longest, *win_func; + List tmp_win_funcs= win_funcs; + List_iterator_fast it(tmp_win_funcs); while ((win_func= it++)) { Window_spec *spec= win_func->window_spec; @@ -894,7 +898,7 @@ find_longest_compatible_order(List_iterator_fast &it) spec->disjoin_partition_and_order_lists(); if (cmp != CMP_GT_C) { - longest= NULL; + longest= nullptr; break; } } @@ -916,16 +920,15 @@ bool have_streaming_window_funcs(THD *thd, List &win_funcs, if (win_funcs.elements == 0) return false; - List_iterator_fast it(win_funcs); Item_window_func *win_func_with_longest_order= - find_longest_compatible_order(it); + find_longest_compatible_order(win_funcs); if (!win_func_with_longest_order) return false; + List_iterator_fast it(win_funcs); Item_window_func *win_func; int cmp; - it.rewind(); while ((win_func= it++)) { Window_spec *spec= win_func->window_spec; @@ -3100,8 +3103,6 @@ bool compute_window_func(THD *thd, iter_win_funcs.rewind(); iter_part_trackers.rewind(); iter_cursor_managers.rewind(); - // we can use a similar appraoch for streaming, where a row is passed over - // all window functions before another is fetched (single pass) Group_bound_tracker *tracker; while ((win_func= iter_win_funcs++) && (tracker= iter_part_trackers++) && @@ -3452,11 +3453,13 @@ bool Window_funcs_sort_streaming::setup(List &window_funcs) if (get_window_functions_required_cursors(thd, window_funcs, &cursor_managers)) return true; - // we need partition trackers too, should be here in setup + + Group_bound_tracker *tracker; while ((win_func= it++)) { - Group_bound_tracker *tracker= - new Group_bound_tracker(thd, win_func->window_spec->partition_list); + if (!(tracker= new Group_bound_tracker( + thd, win_func->window_spec->partition_list))) + return true; tracker->init(); partition_trackers.push_back(tracker); win_func->set_phase_to_computation(); From 8ebea41393e4f4ef69eb04518805bb713b9fdb3c Mon Sep 17 00:00:00 2001 From: OmarGamal10 Date: Tue, 30 Jun 2026 23:26:53 +0300 Subject: [PATCH 12/16] Check that the frame looks at rows and not ranges --- sql/item_sum.h | 1 - sql/sql_window.cc | 9 +++++++-- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/sql/item_sum.h b/sql/item_sum.h index a4f2251bd1546..f1f1ac3863090 100644 --- a/sql/item_sum.h +++ b/sql/item_sum.h @@ -896,7 +896,6 @@ class Item_sum_count :public Item_sum_int bool add() override; void cleanup() override; void remove() override; - inline bool is_streamable() const override { return true; } public: Item_sum_count(THD *thd, Item *item_par): diff --git a/sql/sql_window.cc b/sql/sql_window.cc index cffdc8bac0b12..a2829651e4304 100644 --- a/sql/sql_window.cc +++ b/sql/sql_window.cc @@ -835,11 +835,16 @@ void order_window_funcs_by_window_specs(List *win_func_list) } } -static inline bool frame_is_current_row(Window_spec *win_spec) +/* + Returns true if the window frame is unbounded preceding or current row. +*/ +static inline bool frame_is_streaming_compatible(Window_spec *win_spec) { Window_frame *frame= win_spec->window_frame; if (!frame) return true; + if (frame->units != Window_frame::Frame_units::UNITS_ROWS) + return false; bool unbounded_preceding_or_current= frame->top_bound->precedence_type == Window_frame_bound::CURRENT || (frame->top_bound->precedence_type == Window_frame_bound::PRECEDING && @@ -934,7 +939,7 @@ bool have_streaming_window_funcs(THD *thd, List &win_funcs, Window_spec *spec= win_func->window_spec; if (check_argument_list_aggregation(spec) || !(win_func->window_func()->is_streamable() && - frame_is_current_row(win_func->window_spec))) + frame_is_streaming_compatible(win_func->window_spec))) return false; } From 48d7f9f0652f800a5db2d14badda5f0610ea11a8 Mon Sep 17 00:00:00 2001 From: OmarGamal10 Date: Sat, 1 Aug 2026 22:58:15 +0300 Subject: [PATCH 13/16] Allow streaming for group by when an index satsifies the group list (when QUICK_SELECT can be used) --- sql/sql_select.cc | 6 ++-- sql/sql_select.h | 15 +++++++-- sql/sql_window.cc | 81 ++++++++++++++++++----------------------------- sql/sql_window.h | 1 + 4 files changed, 47 insertions(+), 56 deletions(-) diff --git a/sql/sql_select.cc b/sql/sql_select.cc index 239bee8ecd299..32a64d56d425e 100644 --- a/sql/sql_select.cc +++ b/sql/sql_select.cc @@ -1618,9 +1618,8 @@ JOIN::prepare(TABLE_LIST *tables_init, COND *conds_init, uint og_num, // this needs to decide compatibility with main query sorting if exists // but the actual setting of which is set before test_if_need_tmp_table() if (select_lex->n_sum_items == select_lex->window_funcs.elements && - select_lex->group_list.elements == 0 && have_streaming_window_funcs(thd, select_lex->window_funcs, - win_func_longest_order, order, + win_func_longest_order, order, group_list, streaming_wf_order_is_longer)) streamable_window_funcs= true; @@ -3360,6 +3359,7 @@ int JOIN::optimize_stage2() // is checking if order by references only the first non-const table in JOIN) // i'm not very sure of this, simple_order might change later?? + // Should we skip this if the window function is longer than the main query? if (!need_tmp && simple_order && streaming_wf_order_is_longer) order= win_func_longest_order; @@ -16827,6 +16827,7 @@ void JOIN_TAB::cleanup() if (window_funcs_streaming_step) { window_funcs_streaming_step->cleanup(); + window_funcs_streaming_step= nullptr; } limit= 0; // Free select that was created for filesort outside of create_sort_index @@ -24777,7 +24778,6 @@ sub_select(JOIN *join,JOIN_TAB *join_tab,bool end_of_records) join_tab->loosescan_key_len); skip_over= TRUE; } - error= info->read_record(); if (skip_over && likely(!error)) diff --git a/sql/sql_select.h b/sql/sql_select.h index 348b69f9eef92..15430f9198012 100644 --- a/sql/sql_select.h +++ b/sql/sql_select.h @@ -1769,8 +1769,15 @@ class JOIN :public Sql_alloc materialization path. */ bool streamable_window_funcs= false; - ORDER *win_func_longest_order= NULL; + + /* + These are set in have_streaming_window_funcs(). + streaming_wf_order_is_longer is True if the partition + order list of the + longest window function is longer than AND compatible with the ORDER BY + clause of the main query. + */ bool streaming_wf_order_is_longer= false; + ORDER *win_func_longest_order= NULL; JOIN(THD *thd_arg, List &fields_arg, ulonglong select_options_arg, select_result *result_arg) @@ -1917,7 +1924,8 @@ class JOIN :public Sql_alloc - 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. + - We are using WINDOW functions that do not align with the streaming + criteria (see have_streaming_window_funcs()). When the WITH ROLLUP modifier is present, we cannot skip temporary table creation for the DISTINCT clause just because there are only const tables. */ @@ -1929,7 +1937,8 @@ class JOIN :public Sql_alloc MY_TEST(select_options & OPTION_BUFFER_RESULT))) || (rollup.state != ROLLUP::STATE_NONE && select_distinct) || (select_lex->have_window_funcs() && - (!streamable_window_funcs || only_const_tables()))); + (!streamable_window_funcs || only_const_tables() || + group_optimized_away))); } bool choose_subquery_plan(table_map join_tables); void get_partial_cost_and_fanout(int end_tab_idx, diff --git a/sql/sql_window.cc b/sql/sql_window.cc index a2829651e4304..eb8f8fa689f22 100644 --- a/sql/sql_window.cc +++ b/sql/sql_window.cc @@ -502,46 +502,6 @@ int compare_order_elements(ORDER *ord1, int weight1, return cmp > 0 ? CMP_GT : CMP_LT; } -static -int compare_order_lists(SQL_I_List *part_list1, - int spec_number1, - SQL_I_List *part_list2, - int spec_number2) -{ - if (part_list1 == part_list2) - return CMP_EQ; - ORDER *elem1= part_list1->first; - ORDER *elem2= part_list2->first; - for ( ; elem1 && elem2; elem1= elem1->next, elem2= elem2->next) - { - int cmp; - // remove all constants as we don't need them for comparision - while(elem1 && ((*elem1->item)->real_item())->const_item()) - { - elem1= elem1->next; - continue; - } - - while(elem2 && ((*elem2->item)->real_item())->const_item()) - { - elem2= elem2->next; - continue; - } - - if (!elem1 || !elem2) - break; - - if ((cmp= compare_order_elements(elem1, spec_number1, - elem2, spec_number2))) - return cmp; - } - if (elem1) - return CMP_GT_C; - if (elem2) - return CMP_LT_C; - return CMP_EQ; -} - /* Overloaded to take ORDER* objects instead of SQL_I_List* (the longest wf order list, and the main query order list). @@ -592,6 +552,15 @@ static int compare_order_lists(ORDER *list1, int spec_number1, ORDER *list2, return CMP_EQ; } +static int compare_order_lists(SQL_I_List *part_list1, int spec_number1, + SQL_I_List *part_list2, int spec_number2) +{ + if (part_list1 == part_list2) + return CMP_EQ; + return compare_order_lists(part_list1->first, spec_number1, + part_list2->first, spec_number2); +} + static int compare_window_frame_bounds(Window_frame_bound *win_frame_bound1, Window_frame_bound *win_frame_bound2, @@ -891,6 +860,8 @@ find_longest_compatible_order(const List &win_funcs) Window_spec *longest_spec= longest->window_spec; longest_spec->join_partition_and_order_lists(); + + // Check compatibility with other window function frames while ((win_func= it++)) { if (win_func == longest) @@ -901,7 +872,7 @@ find_longest_compatible_order(const List &win_funcs) longest_spec->win_spec_number, spec->partition_list, spec->win_spec_number); spec->disjoin_partition_and_order_lists(); - if (cmp != CMP_GT_C) + if (!(cmp == CMP_EQ || cmp == CMP_GT_C)) { longest= nullptr; break; @@ -920,6 +891,7 @@ find_longest_compatible_order(const List &win_funcs) bool have_streaming_window_funcs(THD *thd, List &win_funcs, ORDER *&longest_wf_order, ORDER *main_query_order, + ORDER *main_query_group_list, bool &streaming_wf_order_is_longer) { if (win_funcs.elements == 0) @@ -948,7 +920,6 @@ bool have_streaming_window_funcs(THD *thd, List &win_funcs, win_func_with_longest_order->window_spec->partition_list->first, win_func_with_longest_order->window_spec->order_list->first); - // check compatibility of both cmp= compare_order_lists( longest_wf_order, win_func_with_longest_order->window_spec->win_spec_number, @@ -960,6 +931,22 @@ bool have_streaming_window_funcs(THD *thd, List &win_funcs, streaming_wf_order_is_longer= true; else streaming_wf_order_is_longer= false; + + // Ordering keys after the complete GROUP BY key does not affect the ordering + // of the grouped result: there is exactly one row per group key, so a + // trailing key is never reached as a tie-breaker. Hence it's safe even for + // non-grouped columns, whose values are plan-dependent but never participate + // in tie breaking. (Assumes the whole group key is matched as a prefix, and + // no WITH ROLLUP.) + if (main_query_group_list) + { + cmp= compare_order_lists( + longest_wf_order, + win_func_with_longest_order->window_spec->win_spec_number, + main_query_group_list, -1); + if (!(CMP_LT_C <= cmp && cmp <= CMP_GT_C)) + return false; + } return true; } @@ -3144,8 +3131,6 @@ bool compute_window_func(THD *thd, /* We now have computed values for each window function. They can now be saved in the current row. */ - // i need to save to current row field, but might not need all that for - // streaming if (save_window_function_values(window_functions, tbl, rowid_buf)) { ret= true; @@ -3393,11 +3378,6 @@ bool Window_funcs_computation::setup(THD *thd, filtering conditions when we perform sorting for window function computation. */ - // sel holds filtering conditions (where, HAVING), those happen already - // before the window function computation, when a window function is computed - // over some window, we have to respect those filters, and not operate over - // the whole window (it's needed, the temp table does have invalid rows, - // checked with gdb test main.win (does not respect having)). if (tab->filesort && tab->filesort->select) { sel= tab->filesort->select; @@ -3428,7 +3408,6 @@ bool Window_funcs_computation::exec(JOIN *join, bool keep_last_filesort_result) while ((srt = it++)) { counter++; - // hmm? bool keep_filesort_result= keep_last_filesort_result && counter == win_func_sorts.elements; if (srt->exec(join, keep_filesort_result)) @@ -3473,6 +3452,8 @@ bool Window_funcs_sort_streaming::setup(List &window_funcs) Item_sum *sum_func= win_func->window_func(); sum_func->setup_window_func(thd, win_func->window_spec); + // for handling aggregate functions (not done yet, still need to define + // frame for those). win_func->window_func()->set_aggregator(thd, Aggregator::SIMPLE_AGGREGATOR); } diff --git a/sql/sql_window.h b/sql/sql_window.h index 229c8c810b724..5b0f349695ec9 100644 --- a/sql/sql_window.h +++ b/sql/sql_window.h @@ -187,6 +187,7 @@ int setup_windows(THD *thd, Ref_ptr_array ref_pointer_array, TABLE_LIST *tables, bool have_streaming_window_funcs(THD *thd, List &win_funcs, ORDER *&longest_wf_order, ORDER *main_query_order, + ORDER *main_query_group_list, bool &streaming_wf_order_is_longer); ////////////////////////////////////////////////////////////////////////////// From b61b45bf1cd52d2fd4dc1e59eeb6f18d70a01f41 Mon Sep 17 00:00:00 2001 From: OmarGamal10 Date: Sat, 1 Aug 2026 23:25:30 +0300 Subject: [PATCH 14/16] Add and refine tests for streaming --- mysql-test/main/win_streaming.result | 790 +++++++++++++-------------- mysql-test/main/win_streaming.test | 270 ++++++--- sql/sql_select.cc | 7 +- 3 files changed, 569 insertions(+), 498 deletions(-) diff --git a/mysql-test/main/win_streaming.result b/mysql-test/main/win_streaming.result index 146158ce433f3..4291e34257547 100644 --- a/mysql-test/main/win_streaming.result +++ b/mysql-test/main/win_streaming.result @@ -1,131 +1,92 @@ CREATE TABLE t1 (pk INT PRIMARY KEY, a INT, b INT); -INSERT INTO t1 VALUES (1, 1, 3); -INSERT INTO t1 VALUES (2, 1, 1); -INSERT INTO t1 VALUES (3, 2, 2); -INSERT INTO t1 VALUES (4, 2, 4); -INSERT INTO t1 VALUES (5, 3, 1); -INSERT INTO t1 VALUES (6, 3, 2); -SELECT pk, RANK() OVER (ORDER BY pk) AS rnk from t1; -pk rnk -1 1 -2 2 -3 3 -4 4 -5 5 -6 6 -EXPLAIN FORMAT=JSON SELECT pk, RANK() OVER (ORDER BY pk) AS rnk from t1 limit 2; -EXPLAIN -{ - "query_block": { - "select_id": 1, - "cost": "COST_REPLACED", - "nested_loop": [ - { - "table": { - "table_name": "t1", - "access_type": "index", - "key": "PRIMARY", - "key_length": "4", - "used_key_parts": ["pk"], - "loops": 1, - "rows": 2, - "cost": "COST_REPLACED", - "filtered": 100, - "using_index": true - } - } - ] - } -} -EXPLAIN FORMAT=JSON SELECT rank() OVER (PARTITION BY a ORDER BY b) FROM t1; -EXPLAIN -{ - "query_block": { - "select_id": 1, - "cost": "COST_REPLACED", - "nested_loop": [ - { - "read_sorted_file": { - "filesort": { - "sort_key": "t1.a, t1.b", - "table": { - "table_name": "t1", - "access_type": "ALL", - "loops": 1, - "rows": 6, - "cost": "COST_REPLACED", - "filtered": 100 - } - } - } - } - ] - } -} -SELECT row_number() OVER (PARTITION BY a ORDER BY b) as rn FROM t1; -rn -1 -2 -1 -2 -1 -2 -SELECT rank() OVER (PARTITION BY a ORDER BY b) as rnk FROM t1; -rnk -1 -2 -1 -2 -1 -2 -SELECT dense_rank() OVER (PARTITION BY a ORDER BY b) as drnk FROM t1; -drnk -1 -2 -1 -2 -1 -2 -EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a), dense_rank() OVER (ORDER BY a) FROM t1; -EXPLAIN -{ - "query_block": { - "select_id": 1, - "cost": "COST_REPLACED", - "window_functions_computation": { - "sorts": [ - { - "filesort": { - "sort_key": "t1.a" - } - } - ], - "temporary_table": { - "nested_loop": [ - { - "table": { - "table_name": "t1", - "access_type": "ALL", - "loops": 1, - "rows": 6, - "cost": "COST_REPLACED", - "filtered": 100 - } - } - ] - } - } - } -} -SELECT rank() OVER (ORDER BY a) as rnk, dense_rank() OVER (ORDER BY a) as drnk FROM t1; -rnk drnk -1 1 -1 1 -3 2 -3 2 -5 3 -5 3 -EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a, b) FROM t1 ORDER BY a; +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); +EXPLAIN EXTENDED SELECT 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); +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using filesort +EXPLAIN EXTENDED SELECT SQL_BUFFER_RESULT 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); +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary +SELECT 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); +pk a b rn rnk drnk +1 1 10 1 1 1 +2 1 10 2 2 2 +3 1 20 3 3 3 +4 2 20 1 1 1 +5 2 20 2 2 2 +6 2 30 3 3 3 +7 3 10 1 1 1 +8 3 30 2 2 2 +9 3 30 3 3 3 +SELECT SQL_BUFFER_RESULT 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); +pk a b rn rnk drnk +1 1 10 1 1 1 +2 1 10 2 2 2 +3 1 20 3 3 3 +4 2 20 1 1 1 +5 2 20 2 2 2 +6 2 30 3 3 3 +7 3 10 1 1 1 +8 3 30 2 2 2 +9 3 30 3 3 3 +EXPLAIN EXTENDED SELECT pk, a, b, rank() OVER w AS rnk, dense_rank() OVER w AS drnk FROM t1 WINDOW w AS (ORDER BY a); +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using filesort +SELECT pk, a, b, rank() OVER w AS rnk, dense_rank() OVER w AS drnk FROM t1 WINDOW w AS (ORDER BY a); +pk a b rnk drnk +1 1 10 1 1 +2 1 10 1 1 +3 1 20 1 1 +4 2 20 4 2 +5 2 20 4 2 +6 2 30 4 2 +7 3 10 7 3 +8 3 30 7 3 +9 3 30 7 3 +SELECT SQL_BUFFER_RESULT pk, a, b, rank() OVER w AS rnk, dense_rank() OVER w AS drnk FROM t1 WINDOW w AS (ORDER BY a); +pk a b rnk drnk +1 1 10 1 1 +2 1 10 1 1 +3 1 20 1 1 +4 2 20 4 2 +5 2 20 4 2 +6 2 30 4 2 +7 3 10 7 3 +8 3 30 7 3 +9 3 30 7 3 +EXPLAIN EXTENDED SELECT 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); +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using filesort +SELECT 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); +pk a rn rnk drnk +1 1 1 1 1 +2 1 2 2 2 +3 1 3 3 3 +4 2 4 4 4 +5 2 5 5 5 +6 2 6 6 6 +7 3 7 7 7 +8 3 8 8 8 +9 3 9 9 9 +SELECT SQL_BUFFER_RESULT 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); +pk a rn rnk drnk +1 1 1 1 1 +2 1 2 2 2 +3 1 3 3 3 +4 2 4 4 4 +5 2 5 5 5 +6 2 6 6 6 +7 3 7 7 7 +8 3 8 8 8 +9 3 9 9 9 +EXPLAIN FORMAT=JSON SELECT pk, a, b, rank() OVER (ORDER BY a, b) AS rnk FROM t1 ORDER BY a; EXPLAIN { "query_block": { @@ -140,7 +101,7 @@ EXPLAIN "table_name": "t1", "access_type": "ALL", "loops": 1, - "rows": 6, + "rows": 9, "cost": "COST_REPLACED", "filtered": 100 } @@ -150,15 +111,29 @@ EXPLAIN ] } } -SELECT rank() OVER (ORDER BY a, b) as rnk FROM t1 ORDER BY a; -rnk -1 -2 -3 -4 -5 -6 -EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a) FROM t1 ORDER BY a, b; +SELECT pk, a, b, rank() OVER (ORDER BY a, b) AS rnk FROM t1 ORDER BY a; +pk a b rnk +1 1 10 1 +2 1 10 1 +3 1 20 3 +4 2 20 4 +5 2 20 4 +6 2 30 6 +7 3 10 7 +8 3 30 8 +9 3 30 8 +SELECT SQL_BUFFER_RESULT pk, a, b, rank() OVER (ORDER BY a, b) AS rnk FROM t1 ORDER BY a; +pk a b rnk +1 1 10 1 +2 1 10 1 +3 1 20 3 +4 2 20 4 +5 2 20 4 +6 2 30 6 +7 3 10 7 +8 3 30 8 +9 3 30 8 +EXPLAIN FORMAT=JSON SELECT pk, a, b, rank() OVER (ORDER BY a) AS rnk FROM t1 ORDER BY a, b; EXPLAIN { "query_block": { @@ -173,7 +148,7 @@ EXPLAIN "table_name": "t1", "access_type": "ALL", "loops": 1, - "rows": 6, + "rows": 9, "cost": "COST_REPLACED", "filtered": 100 } @@ -183,15 +158,29 @@ EXPLAIN ] } } -SELECT rank() OVER (ORDER BY a) as rnk FROM t1 ORDER BY a, b; -rnk -1 -1 -3 -3 -5 -5 -EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a) FROM t1 ORDER BY a; +SELECT pk, a, b, rank() OVER (ORDER BY a) AS rnk FROM t1 ORDER BY a, b; +pk a b rnk +1 1 10 1 +2 1 10 1 +3 1 20 1 +4 2 20 4 +5 2 20 4 +6 2 30 4 +7 3 10 7 +8 3 30 7 +9 3 30 7 +SELECT SQL_BUFFER_RESULT pk, a, b, rank() OVER (ORDER BY a) AS rnk FROM t1 ORDER BY a, b; +pk a b rnk +1 1 10 1 +2 1 10 1 +3 1 20 1 +4 2 20 4 +5 2 20 4 +6 2 30 4 +7 3 10 7 +8 3 30 7 +9 3 30 7 +EXPLAIN FORMAT=JSON SELECT pk, a, rank() OVER (ORDER BY a) AS rnk FROM t1 ORDER BY a; EXPLAIN { "query_block": { @@ -206,7 +195,7 @@ EXPLAIN "table_name": "t1", "access_type": "ALL", "loops": 1, - "rows": 6, + "rows": 9, "cost": "COST_REPLACED", "filtered": 100 } @@ -216,202 +205,53 @@ EXPLAIN ] } } -SELECT rank() OVER (ORDER BY a) as rnk FROM t1 ORDER BY a; -rnk -1 -1 -3 -3 -5 -5 -EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a), rank() OVER (ORDER BY b) FROM t1; -EXPLAIN -{ - "query_block": { - "select_id": 1, - "cost": "COST_REPLACED", - "window_functions_computation": { - "sorts": [ - { - "filesort": { - "sort_key": "t1.b" - } - }, - { - "filesort": { - "sort_key": "t1.a" - } - } - ], - "temporary_table": { - "nested_loop": [ - { - "table": { - "table_name": "t1", - "access_type": "ALL", - "loops": 1, - "rows": 6, - "cost": "COST_REPLACED", - "filtered": 100 - } - } - ] - } - } - } -} -EXPLAIN FORMAT=JSON SELECT rank() OVER (PARTITION BY max(a) ORDER BY b) FROM t1; -EXPLAIN -{ - "query_block": { - "select_id": 1, - "cost": "COST_REPLACED", - "window_functions_computation": { - "sorts": [ - { - "filesort": { - "sort_key": "max(t1.a), t1.b" - } - } - ], - "temporary_table": { - "nested_loop": [ - { - "table": { - "table_name": "t1", - "access_type": "ALL", - "loops": 1, - "rows": 6, - "cost": "COST_REPLACED", - "filtered": 100 - } - } - ] - } - } - } -} -EXPLAIN FORMAT=JSON SELECT max(a), rank() OVER (ORDER BY b) FROM t1; -EXPLAIN -{ - "query_block": { - "select_id": 1, - "cost": "COST_REPLACED", - "window_functions_computation": { - "sorts": [ - { - "filesort": { - "sort_key": "t1.b" - } - } - ], - "temporary_table": { - "nested_loop": [ - { - "table": { - "table_name": "t1", - "access_type": "ALL", - "loops": 1, - "rows": 6, - "cost": "COST_REPLACED", - "filtered": 100 - } - } - ] - } - } - } -} -EXPLAIN FORMAT=JSON SELECT count(*) OVER (ORDER BY a RANGE BETWEEN CURRENT ROW AND 5 FOLLOWING) FROM t1; -EXPLAIN -{ - "query_block": { - "select_id": 1, - "cost": "COST_REPLACED", - "window_functions_computation": { - "sorts": [ - { - "filesort": { - "sort_key": "t1.a" - } - } - ], - "temporary_table": { - "nested_loop": [ - { - "table": { - "table_name": "t1", - "access_type": "ALL", - "loops": 1, - "rows": 6, - "cost": "COST_REPLACED", - "filtered": 100 - } - } - ] - } - } - } -} -EXPLAIN FORMAT=JSON SELECT count(*) OVER (ORDER BY a RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) FROM t1; -EXPLAIN -{ - "query_block": { - "select_id": 1, - "cost": "COST_REPLACED", - "nested_loop": [ - { - "read_sorted_file": { - "filesort": { - "sort_key": "t1.a", - "table": { - "table_name": "t1", - "access_type": "ALL", - "loops": 1, - "rows": 6, - "cost": "COST_REPLACED", - "filtered": 100 - } - } - } - } - ] - } -} -EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a) FROM t1 GROUP BY a; -EXPLAIN -{ - "query_block": { - "select_id": 1, - "cost": "COST_REPLACED", - "filesort": { - "sort_key": "t1.a", - "window_functions_computation": { - "sorts": [ - { - "filesort": { - "sort_key": "t1.a" - } - } - ], - "temporary_table": { - "nested_loop": [ - { - "table": { - "table_name": "t1", - "access_type": "ALL", - "loops": 1, - "rows": 6, - "cost": "COST_REPLACED", - "filtered": 100 - } - } - ] - } - } - } - } -} +SELECT pk, a, rank() OVER (ORDER BY a) AS rnk FROM t1 ORDER BY a; +pk a rnk +1 1 1 +2 1 1 +3 1 1 +4 2 4 +5 2 4 +6 2 4 +7 3 7 +8 3 7 +9 3 7 +SELECT SQL_BUFFER_RESULT pk, a, rank() OVER (ORDER BY a) AS rnk FROM t1 ORDER BY a; +pk a rnk +1 1 1 +2 1 1 +3 1 1 +4 2 4 +5 2 4 +6 2 4 +7 3 7 +8 3 7 +9 3 7 +EXPLAIN EXTENDED SELECT 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); +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using filesort +SELECT 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); +pk a r r_plus diff +1 1 1 2 0 +2 1 1 2 0 +3 1 1 2 0 +4 2 4 5 2 +5 2 4 5 2 +6 2 4 5 2 +7 3 7 8 4 +8 3 7 8 4 +9 3 7 8 4 +SELECT SQL_BUFFER_RESULT 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); +pk a r r_plus diff +1 1 1 2 0 +2 1 1 2 0 +3 1 1 2 0 +4 2 4 5 2 +5 2 4 5 2 +6 2 4 5 2 +7 3 7 8 4 +8 3 7 8 4 +9 3 7 8 4 CREATE TABLE t2 (pk INT PRIMARY KEY, c INT); INSERT INTO t2 VALUES (1, 100); INSERT INTO t2 VALUES (2, 200); @@ -419,97 +259,205 @@ INSERT INTO t2 VALUES (3, 300); INSERT INTO t2 VALUES (4, 400); INSERT INTO t2 VALUES (5, 500); INSERT INTO t2 VALUES (6, 600); -EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY t1.b) FROM t1 JOIN t2 ON t1.pk = t2.pk; -EXPLAIN +INSERT INTO t2 VALUES (7, 700); +INSERT INTO t2 VALUES (8, 800); +INSERT INTO t2 VALUES (9, 900); +EXPLAIN EXTENDED SELECT 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; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL PRIMARY NULL NULL NULL 9 100.00 Using filesort +1 SIMPLE t2 eq_ref PRIMARY PRIMARY 4 test.t1.pk 1 100.00 Using index +SELECT 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; +pk a b rnk +1 1 10 1 +2 1 10 2 +3 1 20 4 +4 2 20 5 +5 2 20 6 +6 2 30 7 +7 3 10 3 +8 3 30 8 +9 3 30 9 +SELECT SQL_BUFFER_RESULT 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; +pk a b rnk +1 1 10 1 +2 1 10 2 +3 1 20 4 +4 2 20 5 +5 2 20 6 +6 2 30 7 +7 3 10 3 +8 3 30 8 +9 3 30 9 +EXPLAIN EXTENDED SELECT 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; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using filesort +SELECT 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; +pk a b rnk +1 1 10 1 +2 1 10 2 +3 1 20 4 +4 2 20 5 +5 2 20 6 +6 2 30 7 +7 3 10 3 +8 3 30 8 +9 3 30 9 +SELECT SQL_BUFFER_RESULT 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; +pk a b rnk +1 1 10 1 +2 1 10 2 +3 1 20 4 +4 2 20 5 +5 2 20 6 +6 2 30 7 +7 3 10 3 +8 3 30 8 +9 3 30 9 +DROP TABLE t2; +EXPLAIN EXTENDED SELECT 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; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using where; Using filesort +SELECT 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; +pk a b rnk drnk +4 2 20 1 1 +5 2 20 1 1 +6 2 30 3 2 +7 3 10 1 1 +8 3 30 2 2 +9 3 30 2 2 +SELECT SQL_BUFFER_RESULT 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; +pk a b rnk drnk +4 2 20 1 1 +5 2 20 1 1 +6 2 30 3 2 +7 3 10 1 1 +8 3 30 2 2 +9 3 30 2 2 +EXPLAIN EXTENDED SELECT d.a, d.rnk FROM (SELECT a, rank() OVER (ORDER BY a) AS rnk FROM t1) AS d; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 PRIMARY ALL NULL NULL NULL NULL 9 100.00 +2 DERIVED t1 ALL NULL NULL NULL NULL 9 100.00 Using filesort +SELECT d.a, d.rnk FROM (SELECT a, rank() OVER (ORDER BY a) AS rnk FROM t1) AS d; +a rnk +1 1 +1 1 +1 1 +2 4 +2 4 +2 4 +3 7 +3 7 +3 7 +ANALYZE FORMAT=JSON SELECT pk, rank() OVER (ORDER BY pk) AS rnk FROM t1 LIMIT 2; +ANALYZE { + "query_optimization": { + "r_total_time_ms": "REPLACED" + }, "query_block": { "select_id": 1, - "cost": "COST_REPLACED", + "cost": "REPLACED", + "r_loops": 1, + "r_total_time_ms": "REPLACED", "nested_loop": [ - { - "read_sorted_file": { - "filesort": { - "sort_key": "t1.b", - "table": { - "table_name": "t1", - "access_type": "ALL", - "possible_keys": ["PRIMARY"], - "loops": 1, - "rows": 6, - "cost": "COST_REPLACED", - "filtered": 100 - } - } - } - }, { "table": { - "table_name": "t2", - "access_type": "eq_ref", - "possible_keys": ["PRIMARY"], + "table_name": "t1", + "access_type": "index", "key": "PRIMARY", "key_length": "4", "used_key_parts": ["pk"], - "ref": ["test.t1.pk"], - "loops": 6, - "rows": 1, - "cost": "COST_REPLACED", + "loops": 1, + "r_loops": 1, + "rows": 9, + "r_rows": 2, + "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, "using_index": true } } ] } } -SELECT rank() OVER (ORDER BY t1.b) as rnk FROM t1 JOIN t2 ON t1.pk = t2.pk; -rnk -1 -1 -3 -3 -5 -6 -SELECT rank() OVER (PARTITION BY t1.a ORDER BY t1.b) as rnk FROM t1 JOIN t2 ON t1.pk = t2.pk; -rnk -1 -2 -1 -2 -1 -2 -SELECT rank() OVER (ORDER BY t1.b) as rnk FROM t1 LEFT JOIN t2 ON t1.pk = t2.pk; -rnk -1 -1 -3 -3 -5 -6 -DROP TABLE t2; -EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a) as rnk FROM (SELECT a FROM t1 WHERE a > 1) derived; -EXPLAIN -{ - "query_block": { - "select_id": 1, - "cost": "COST_REPLACED", - "nested_loop": [ - { - "read_sorted_file": { - "filesort": { - "sort_key": "t1.a", - "table": { - "table_name": "t1", - "access_type": "ALL", - "loops": 1, - "rows": 6, - "cost": "COST_REPLACED", - "filtered": 100, - "attached_condition": "t1.a > 1" - } - } - } - } - ] - } -} +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); +EXPLAIN EXTENDED SELECT a, b, rank() OVER (ORDER BY a) AS rnk, dense_rank() OVER (ORDER BY a) AS drnk FROM tg GROUP BY a, b; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE tg range NULL a 10 NULL 6 100.00 Using index for group-by +SELECT a, b, rank() OVER (ORDER BY a) AS rnk, dense_rank() OVER (ORDER BY a) AS drnk FROM tg GROUP BY a, b; +a b rnk drnk +1 1 1 1 +1 2 1 1 +2 1 3 2 +2 2 3 2 +2 3 3 2 +3 1 6 3 +SELECT SQL_BUFFER_RESULT a, b, rank() OVER (ORDER BY a) AS rnk, dense_rank() OVER (ORDER BY a) AS drnk FROM tg GROUP BY a, b; +a b rnk drnk +1 1 1 1 +1 2 1 1 +2 1 3 2 +2 2 3 2 +2 3 3 2 +3 1 6 3 +EXPLAIN EXTENDED SELECT a, rank() OVER (ORDER BY a, b) AS rnk, dense_rank() OVER (ORDER BY a, b) AS drnk FROM tg GROUP BY a; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE tg range NULL a 5 NULL 6 100.00 Using index for group-by +SELECT a, rank() OVER (ORDER BY a, b) AS rnk, dense_rank() OVER (ORDER BY a, b) AS drnk FROM tg GROUP BY a; +a rnk drnk +1 1 1 +2 2 2 +3 3 3 +SELECT SQL_BUFFER_RESULT a, rank() OVER (ORDER BY a, b) AS rnk, dense_rank() OVER (ORDER BY a, b) AS drnk FROM tg GROUP BY a; +a rnk drnk +1 1 1 +2 2 2 +3 3 3 +EXPLAIN EXTENDED SELECT rank() OVER (ORDER BY a), rank() OVER (ORDER BY b) FROM t1; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary +EXPLAIN EXTENDED SELECT rank() OVER (PARTITION BY max(a) ORDER BY b) FROM t1; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary +EXPLAIN EXTENDED SELECT max(a), rank() OVER (ORDER BY b) FROM t1; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary +EXPLAIN EXTENDED SELECT rank() OVER (ORDER BY a) FROM t1 GROUP BY a; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary; Using filesort +EXPLAIN EXTENDED SELECT rank() OVER (ORDER BY a) FROM tg GROUP BY 1+2; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE tg index NULL a 10 NULL 6 100.00 Using index; Using temporary +EXPLAIN EXTENDED SELECT pk, rank() OVER (ORDER BY a) AS x FROM t1 ORDER BY x; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary; Using filesort +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); +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary +EXPLAIN EXTENDED SELECT sum(b) OVER (ORDER BY b) FROM t1; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary +EXPLAIN EXTENDED SELECT count(*) OVER (ORDER BY a RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) FROM t1; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary +EXPLAIN EXTENDED SELECT count(*) OVER (ORDER BY a RANGE BETWEEN CURRENT ROW AND 5 FOLLOWING) FROM t1; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary +EXPLAIN EXTENDED SELECT sum(b) OVER () FROM t1; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary +EXPLAIN EXTENDED SELECT sum(b) OVER (PARTITION BY a) FROM t1; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary +DROP TABLE tg; DROP TABLE t1; diff --git a/mysql-test/main/win_streaming.test b/mysql-test/main/win_streaming.test index 3f0123534f82b..c75790a34ba35 100644 --- a/mysql-test/main/win_streaming.test +++ b/mysql-test/main/win_streaming.test @@ -1,5 +1,5 @@ # -# Streaming Window Functions Tests +# 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. @@ -7,13 +7,8 @@ # Explain per scenario to lock up streaming path # Explains for non streamable cases, no need to check output -# For now, row_number(), rank(), dense_rank() are streamable. -# should I only test on rank? would it make sense to add EXPLAIN for all? - -# Basic streaming (one rank with explain, others only correctness) - # Order by and partition -# Explain with one function to show streamable and index or filesort is +# 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 @@ -23,72 +18,91 @@ # 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 - -CREATE TABLE t1 (pk INT PRIMARY KEY, a INT, b INT); -INSERT INTO t1 VALUES (1, 1, 3); -INSERT INTO t1 VALUES (2, 1, 1); -INSERT INTO t1 VALUES (3, 2, 2); -INSERT INTO t1 VALUES (4, 2, 4); -INSERT INTO t1 VALUES (5, 3, 1); -INSERT INTO t1 VALUES (6, 3, 2); - -# Basic streaming: one rank with explain, others only correctness -SELECT pk, RANK() OVER (ORDER BY pk) AS rnk from t1; ---source include/explain-no-costs.inc -EXPLAIN FORMAT=JSON SELECT pk, RANK() OVER (ORDER BY pk) AS rnk from t1 limit 2; - -# question: should i add test for other streamable functions? - -# Order by and partition: explain with rank to show streamable ---source include/explain-no-costs.inc -EXPLAIN FORMAT=JSON SELECT rank() OVER (PARTITION BY a ORDER BY b) FROM t1; -SELECT row_number() OVER (PARTITION BY a ORDER BY b) as rn FROM t1; -SELECT rank() OVER (PARTITION BY a ORDER BY b) as rnk FROM t1; -SELECT dense_rank() OVER (PARTITION BY a ORDER BY b) as drnk FROM t1; +# Test with limit and analyze to show we read only rows needed -# Show compatible functions also stream ---source include/explain-no-costs.inc -EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a), dense_rank() OVER (ORDER BY a) FROM t1; -SELECT rank() OVER (ORDER BY a) as rnk, dense_rank() OVER (ORDER BY a) as drnk FROM t1; +# 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. -# Windows reusing the main query order (whichever is longer) ---source include/explain-no-costs.inc -EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a, b) FROM t1 ORDER BY a; -SELECT rank() OVER (ORDER BY a, b) as rnk FROM t1 ORDER BY a; - ---source include/explain-no-costs.inc -EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a) FROM t1 ORDER BY a, b; -SELECT rank() OVER (ORDER BY a) as rnk FROM t1 ORDER BY a, b; - ---source include/explain-no-costs.inc -EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a) FROM t1 ORDER BY a; -SELECT rank() OVER (ORDER BY a) as rnk FROM t1 ORDER BY a; - -# Incompatible orders materialize ---source include/explain-no-costs.inc -EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a), rank() OVER (ORDER BY b) FROM t1; - -# Aggregate functions inside partition lists materialize ---source include/explain-no-costs.inc -EXPLAIN FORMAT=JSON SELECT rank() OVER (PARTITION BY max(a) ORDER BY b) FROM t1; +# 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 -# Aggregate functions anywhere in the select list materialize ---source include/explain-no-costs.inc -EXPLAIN FORMAT=JSON SELECT max(a), rank() OVER (ORDER BY b) FROM t1; - -# Incompatible RANGE frame materializes +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 -EXPLAIN FORMAT=JSON SELECT count(*) OVER (ORDER BY a RANGE BETWEEN CURRENT ROW AND 5 FOLLOWING) FROM t1; +--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 -EXPLAIN FORMAT=JSON SELECT count(*) OVER (ORDER BY a RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) FROM t1; +--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; -# GROUP BY materializes (for now?) +# window order equals main ORDER BY --source include/explain-no-costs.inc -EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a) FROM t1 GROUP BY a; - -# Multi-table join +--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); @@ -96,19 +110,125 @@ INSERT INTO t2 VALUES (3, 300); INSERT INTO t2 VALUES (4, 400); INSERT INTO t2 VALUES (5, 500); INSERT INTO t2 VALUES (6, 600); - ---source include/explain-no-costs.inc -EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY t1.b) FROM t1 JOIN t2 ON t1.pk = t2.pk; -SELECT rank() OVER (ORDER BY t1.b) as rnk FROM t1 JOIN t2 ON t1.pk = t2.pk; -SELECT rank() OVER (PARTITION BY t1.a ORDER BY t1.b) as rnk FROM t1 JOIN t2 ON t1.pk = t2.pk; -SELECT rank() OVER (ORDER BY t1.b) as rnk FROM t1 LEFT JOIN t2 ON t1.pk = t2.pk; +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; -# Subquery in FROM ---source include/explain-no-costs.inc -EXPLAIN FORMAT=JSON SELECT rank() OVER (ORDER BY a) as rnk FROM (SELECT a FROM t1 WHERE a > 1) derived; +# 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 an index supplies the group list in order there is +# no temp table, 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; + +# +# 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; -# Cases to look harder later (subqueries, expressions) +# 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; +# 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; + +# +# 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 tg; DROP TABLE t1; + +--enable_warnings diff --git a/sql/sql_select.cc b/sql/sql_select.cc index 32a64d56d425e..e86b3942cc93a 100644 --- a/sql/sql_select.cc +++ b/sql/sql_select.cc @@ -26,6 +26,7 @@ */ #include "mariadb.h" +#include "my_dbug.h" #include "sql_list.h" #include "sql_priv.h" #include "unireg.h" @@ -26219,8 +26220,10 @@ enum_nested_loop_state end_compute_win_func(JOIN *join, JOIN_TAB *join_tab, // we don't even need to pass the row to the window function, because the // add() functions read from the TABLE::record[0] directly, as we did // NOT call split_sum_func(), so we still point to base table - (join_tab - 1)->window_funcs_streaming_step->process_row(); - return end_send(join, join_tab, end_of_records); + DBUG_ENTER("end_compute_win_func"); + if ((join_tab - 1)->window_funcs_streaming_step->process_row()) + DBUG_RETURN(NESTED_LOOP_ERROR); + DBUG_RETURN(end_send(join, join_tab, end_of_records)); } /* From cc9bfec6555eb4448dfa3670b5cbc0ce131d65e9 Mon Sep 17 00:00:00 2001 From: OmarGamal10 Date: Sat, 8 Aug 2026 19:56:58 +0300 Subject: [PATCH 15/16] Add a check for end_of_records before calling process_row() While this does not hurt correctness (for innoDB at least), it runs a useless pass of process_row() over the last read row again. --- sql/sql_select.cc | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/sql/sql_select.cc b/sql/sql_select.cc index e86b3942cc93a..676f7b107faa5 100644 --- a/sql/sql_select.cc +++ b/sql/sql_select.cc @@ -26221,7 +26221,8 @@ enum_nested_loop_state end_compute_win_func(JOIN *join, JOIN_TAB *join_tab, // add() functions read from the TABLE::record[0] directly, as we did // NOT call split_sum_func(), so we still point to base table DBUG_ENTER("end_compute_win_func"); - if ((join_tab - 1)->window_funcs_streaming_step->process_row()) + if (!end_of_records && + (join_tab - 1)->window_funcs_streaming_step->process_row()) DBUG_RETURN(NESTED_LOOP_ERROR); DBUG_RETURN(end_send(join, join_tab, end_of_records)); } From 9cd7a448869189ae7891ce7cd5b683ce4b7f9554 Mon Sep 17 00:00:00 2001 From: OmarGamal10 Date: Sun, 9 Aug 2026 07:40:34 +0300 Subject: [PATCH 16/16] Allow GROUP BY to stream only when a loose index scan is used This is the only case for which the rows arrive in group order and one row per group, hence no further grouping / aggregration is needed. The other case is when the GROUP BY is rewrittein into ORDER when the keys are covered by a unique non-null index, which is not covered in this commit. --- mysql-test/main/win_streaming.result | 42 +++++++++++++++++++++++++++- mysql-test/main/win_streaming.test | 42 ++++++++++++++++++++++++---- sql/sql_select.cc | 24 ++++++---------- sql/sql_select.h | 18 ++++++++---- sql/sql_window.cc | 21 +++++++------- sql/sql_window.h | 17 ++++++----- 6 files changed, 120 insertions(+), 44 deletions(-) diff --git a/mysql-test/main/win_streaming.result b/mysql-test/main/win_streaming.result index 4291e34257547..cc9e197d4c639 100644 --- a/mysql-test/main/win_streaming.result +++ b/mysql-test/main/win_streaming.result @@ -423,6 +423,25 @@ a rnk drnk 1 1 1 2 2 2 3 3 3 +EXPLAIN EXTENDED SELECT 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); +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE tg range NULL a 10 NULL 6 100.00 Using index for group-by +SELECT 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); +a b rn rnk drnk +1 1 1 1 1 +1 2 2 2 2 +2 1 1 1 1 +2 2 2 2 2 +2 3 3 3 3 +3 1 1 1 1 +SELECT SQL_BUFFER_RESULT 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); +a b rn rnk drnk +1 1 1 1 1 +1 2 2 2 2 +2 1 1 1 1 +2 2 2 2 2 +2 3 3 3 3 +3 1 1 1 1 EXPLAIN EXTENDED SELECT rank() OVER (ORDER BY a), rank() OVER (ORDER BY b) FROM t1; id select_type table type possible_keys key key_len ref rows filtered Extra 1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary @@ -435,12 +454,34 @@ id select_type table type possible_keys key key_len ref rows filtered Extra EXPLAIN EXTENDED SELECT rank() OVER (ORDER BY a) FROM t1 GROUP BY a; id select_type table type possible_keys key key_len ref rows filtered Extra 1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary; Using filesort +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); +EXPLAIN SELECT tg.a, rank() OVER (ORDER BY tg.a) FROM tg JOIN t2 ON tg.a=t2.a GROUP BY tg.a; +id select_type table type possible_keys key key_len ref rows Extra +1 SIMPLE t2 index a a 5 NULL 4 Using where; Using index; Using temporary +1 SIMPLE tg ref a a 5 test.t2.a 1 Using index +EXPLAIN SELECT a, x FROM t2 FORCE INDEX FOR GROUP BY (a) GROUP BY a; +id select_type table type possible_keys key key_len ref rows Extra +1 SIMPLE t2 index NULL a 5 NULL 4 +EXPLAIN SELECT a, x, rank() OVER (ORDER BY a) FROM t2 FORCE INDEX FOR GROUP BY (a) GROUP BY a; +id select_type table type possible_keys key key_len ref rows Extra +1 SIMPLE t2 index NULL a 5 NULL 4 Using temporary EXPLAIN EXTENDED SELECT rank() OVER (ORDER BY a) FROM tg GROUP BY 1+2; id select_type table type possible_keys key key_len ref rows filtered Extra 1 SIMPLE tg index NULL a 10 NULL 6 100.00 Using index; Using temporary +EXPLAIN EXTENDED SELECT pk, a, b FROM t1 GROUP BY pk; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using filesort +EXPLAIN EXTENDED SELECT pk, a, rank() OVER (ORDER BY pk) AS rnk FROM t1 GROUP BY pk; +id select_type table type possible_keys key key_len ref rows filtered Extra +1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary; Using filesort EXPLAIN EXTENDED SELECT pk, rank() OVER (ORDER BY a) AS x FROM t1 ORDER BY x; id select_type table type possible_keys key key_len ref rows filtered Extra 1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary; Using filesort +DROP TABLE tg, t2; 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); id select_type table type possible_keys key key_len ref rows filtered Extra 1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary @@ -459,5 +500,4 @@ id select_type table type possible_keys key key_len ref rows filtered Extra EXPLAIN EXTENDED SELECT sum(b) OVER (PARTITION BY a) FROM t1; id select_type table type possible_keys key key_len ref rows filtered Extra 1 SIMPLE t1 ALL NULL NULL NULL NULL 9 100.00 Using temporary -DROP TABLE tg; DROP TABLE t1; diff --git a/mysql-test/main/win_streaming.test b/mysql-test/main/win_streaming.test index c75790a34ba35..b1aa065ead1ee 100644 --- a/mysql-test/main/win_streaming.test +++ b/mysql-test/main/win_streaming.test @@ -146,9 +146,10 @@ SELECT d.a, d.rnk FROM (SELECT a, rank() OVER (ORDER BY a) AS rnk FROM t1) AS d; --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 an index supplies the group list in order there is -# no temp table, and if the window order is compatible with the group list the -# window functions are computed on the streamed grouped rows. +# 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)); @@ -175,6 +176,14 @@ 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 # @@ -188,16 +197,40 @@ 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 +# 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. # @@ -228,7 +261,6 @@ EXPLAIN EXTENDED SELECT count(*) OVER (ORDER BY a RANGE BETWEEN CURRENT ROW AND EXPLAIN EXTENDED SELECT sum(b) OVER () FROM t1; EXPLAIN EXTENDED SELECT sum(b) OVER (PARTITION BY a) FROM t1; -DROP TABLE tg; DROP TABLE t1; --enable_warnings diff --git a/sql/sql_select.cc b/sql/sql_select.cc index 676f7b107faa5..09b3bd2c94d45 100644 --- a/sql/sql_select.cc +++ b/sql/sql_select.cc @@ -3349,18 +3349,14 @@ 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 - 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 window functions are present and not streamable, then we can't have + simple_order set to TRUE as the window function needs a temp table for + computation. In this case, 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() && !streamable_window_funcs) simple_order= FALSE; - // this means the order by should be done in a temp table (it's real purpose - // is checking if order by references only the first non-const table in JOIN) - // i'm not very sure of this, simple_order might change later?? - // Should we skip this if the window function is longer than the main query? if (!need_tmp && simple_order && streaming_wf_order_is_longer) order= win_func_longest_order; @@ -3601,19 +3597,17 @@ int JOIN::optimize_stage2() if (streamable_window_funcs && !need_tmp) { - JOIN_TAB *last_real_tab= &join_tab[exec_join_tab_cnt() - 1]; - // here i would attach the new streamable class (same interface ?) + JOIN_TAB *last_real_tab= join_tab + exec_join_tab_cnt() - 1; + DBUG_ASSERT(last_real_tab->next_select == end_send); + if (!(last_real_tab->window_funcs_streaming_step= new Window_funcs_sort_streaming(thd))) DBUG_RETURN(true); - // this sets up the list, and the partition and group tracking if (last_real_tab->window_funcs_streaming_step->setup( select_lex->window_funcs)) DBUG_RETURN(true); - // i need to make SURE THAT END_SEND is not assigned to last table after - // this, this is very important - last_real_tab->next_select= - end_compute_win_func; // calls process_row and end_send + + last_real_tab->next_select= end_compute_win_func; /* Count that we're using window functions. */ status_var_increment(thd->status_var.feature_window_functions); } diff --git a/sql/sql_select.h b/sql/sql_select.h index 15430f9198012..0a61f430065bc 100644 --- a/sql/sql_select.h +++ b/sql/sql_select.h @@ -1924,10 +1924,16 @@ class JOIN :public Sql_alloc - 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 that do not align with the streaming - criteria (see have_streaming_window_funcs()). - When the WITH ROLLUP modifier is present, we cannot skip temporary table - creation for the DISTINCT clause just because there are only const tables. + - We are using WINDOW functions that cannot be computed by streaming. + The streaming step attaches to end_send, so it is only viable when the + last next_select is end_send. We must fall back to a temp table when: + * the window functions fail the streaming criteria + (see have_streaming_window_funcs()), or + * there are no real tables to stream from (only_const_tables()), or + * the plan would run an executor-side grouping step (end_send_group) + rather than end_send: i.e. grouping was optimized away to a single + implicit group (group_optimized_away), or there is a GROUP BY not + satisfied by a loose index scan. */ bool test_if_need_tmp_table() { @@ -1938,7 +1944,9 @@ class JOIN :public Sql_alloc (rollup.state != ROLLUP::STATE_NONE && select_distinct) || (select_lex->have_window_funcs() && (!streamable_window_funcs || only_const_tables() || - group_optimized_away))); + group_optimized_away || + (group_list && + !join_tab[const_tables].is_using_loose_index_scan())))); } bool choose_subquery_plan(table_map join_tables); void get_partial_cost_and_fanout(int end_tab_idx, diff --git a/sql/sql_window.cc b/sql/sql_window.cc index eb8f8fa689f22..037e4f76f705d 100644 --- a/sql/sql_window.cc +++ b/sql/sql_window.cc @@ -934,10 +934,11 @@ bool have_streaming_window_funcs(THD *thd, List &win_funcs, // Ordering keys after the complete GROUP BY key does not affect the ordering // of the grouped result: there is exactly one row per group key, so a - // trailing key is never reached as a tie-breaker. Hence it's safe even for - // non-grouped columns, whose values are plan-dependent but never participate - // in tie breaking. (Assumes the whole group key is matched as a prefix, and - // no WITH ROLLUP.) + // trailing key can never be reached as a tie-breaker. Hence it is safe to + // drop the trailing keys even if the window function references non-grouped + // columns, whose values are plan-dependent but cannot affect the ordering + // between grouped rows. (Assumes the whole GROUP BY key is matched as a + // prefix, and no WITH ROLLUP.) if (main_query_group_list) { cmp= compare_order_lists( @@ -2865,8 +2866,6 @@ static bool is_computed_with_remove(Item_sum::Sumfunctype sum_func) If the window functions share the same frame specification, those window functions will be registered to the same cursor. */ -// i can reuse this for streaming, it just creates a Cursor Manager for each -// window function bool get_window_functions_required_cursors( THD *thd, List& window_functions, @@ -3441,11 +3440,13 @@ bool Window_funcs_sort_streaming::setup(List &window_funcs) Group_bound_tracker *tracker; while ((win_func= it++)) { - if (!(tracker= new Group_bound_tracker( - thd, win_func->window_spec->partition_list))) - return true; + tracker= + new Group_bound_tracker(thd, win_func->window_spec->partition_list); tracker->init(); partition_trackers.push_back(tracker); + + // So that end_send gets the live value of the window function on calling + // val_*(), and not the value from result_field. win_func->set_phase_to_computation(); // sets peer tracker inside rank() @@ -3461,8 +3462,6 @@ bool Window_funcs_sort_streaming::setup(List &window_funcs) return false; } -// called as a callback for each row in the join loop last table, directly -// before end_send() bool Window_funcs_sort_streaming::process_row() { List_iterator_fast iter_win_funcs(win_funcs); diff --git a/sql/sql_window.h b/sql/sql_window.h index 5b0f349695ec9..f24daf24d9eea 100644 --- a/sql/sql_window.h +++ b/sql/sql_window.h @@ -269,16 +269,19 @@ class Window_funcs_sort_streaming : public Sql_alloc public: Window_funcs_sort_streaming(THD *thd) : thd(thd) {} bool setup(List &win_funcs); - bool process_row(); // this object is attached to the JOIN_TAB, and - // process_row() is called for a method attached on - // takes the window funcs and the current row by - // end_compute_win_funcs() and calls the appropriate - // cursors to update the aggregate functions + /* + The object is attached to the last real JOIN_TAB in the query. This + function is called by end_compute_win_func() to run the window functions + computation over the current row in the JOIN output, assuming the row sits + in TABLE::record[0]. Then end_send calls val_*() methods of the window + functions to retrieve the live computed values and sends the row to output. + */ + bool process_row(); void cleanup(); private: - int rownum= 0; // internal state for process row - THD *thd= NULL; + int rownum= 0; // Internal state for process row + THD *thd= nullptr; List win_funcs; List cursor_managers; List partition_trackers;