MDEV-31466 Add optional correlation column list for derived tables

Extend derived table syntax to support column name assignment.
(subquery expression) [as|=] ident [comma separated column name list].
Prior to this patch, the optional comma separated column name list is
not supported.

Processing within the unit of the subquery expression will use
original column names, outside the unit will use the new names.

For example, in the query

select a1, a2 from
  (select c1, c2, c3 from t1 where c2 > 0) as dt (a1, a2, a3)
where a2 > 10;

we see the second column of the derived table dt being used both within,
(where c2 > 0), and outside, (where a2 > 10), the specification.
Both conditions apply to t1.c2.

When multiple unit preparations are required, such as when being used within
a prepared statement or procedure, original column names are needed for
correct resolution. Original names are reset within mysql_derived_reinit().

Item_holder items, used for result tables in both TVC and union preparations
are renamed before use within st_select_lex_unit::prepare().

During wildcard expansion, if column names are present, items names are
set directly after creation.

Reviewed by Igor Babaev (igor@mariadb.com)

Author: mariadb-RexJohnston

mysql-test/main/derived.result

@@ -12280,3 +12280,6 @@ ER_SEQUENCE_TABLE_ORDER_BY
         eng "ORDER BY"
 ER_VARIABLE_IGNORED
         eng "The variable '%s' is ignored. It only exists for compatibility with old installations and will be removed in a future release"
+ER_INCORRECT_COLUMN_NAME_COUNT
+        eng "Incorrect column name count for derived table"
+        chi "派生表的列名计数不正确"

mysql-test/main/derived.test

@@ -8635,6 +8635,10 @@ insert_fields(THD *thd, Name_resolution_context *context,
     */
     field_iterator.set(tables);
 
+    List_iterator_fast<Lex_ident_sys> ni;
+    if (tables->column_names)
+      ni.init(*tables->column_names);
+
     for (; !field_iterator.end_of_fields(); field_iterator.next())
     {
       /*
@@ -8650,6 +8654,9 @@ insert_fields(THD *thd, Name_resolution_context *context,
       if (!(item= field_iterator.create_item(thd)))
         DBUG_RETURN(TRUE);
 
+      if (tables->column_names)
+        lex_string_set(&item->name, (ni++)->str);
+
       /* cache the table for the Item_fields inserted by expanding stars */
       if (item->type() == Item::FIELD_ITEM && tables->cacheable_table)
         ((Item_field *)item)->cached_table= tables;

sql/share/errmsg-utf8.txt

@@ -1652,10 +1652,14 @@ void With_clause::print(THD *thd, String *str, enum_query_type query_type)
 }
 
 
-static void list_strlex_print(THD *thd, String *str, List<Lex_ident_sys> *list)
+void list_strlex_print(THD *thd, String *str, List<Lex_ident_sys> *list,
+                              bool bracketed)
 {
   List_iterator_fast<Lex_ident_sys> li(*list);
   bool first= TRUE;
+
+  if (bracketed)
+    str->append('(');
   while(Lex_ident_sys *col_name= li++)
   {
     if (first)
@@ -1664,6 +1668,8 @@ static void list_strlex_print(THD *thd, String *str, List<Lex_ident_sys> *list)
       str->append(',');
     append_identifier(thd, str, col_name);
   }
+  if (bracketed)
+    str->append(')');
 }
 
 
@@ -1684,12 +1690,7 @@ void With_element::print(THD *thd, String *str, enum_query_type query_type)
 {
   str->append(get_name());
   if (column_list.elements)
-  {
-    List_iterator_fast<Lex_ident_sys> li(column_list);
-    str->append('(');
-    list_strlex_print(thd, str, &column_list);
-    str->append(')');
-  }
+    list_strlex_print(thd, str, &column_list, true);
   str->append(STRING_WITH_LEN(" as ("));
   spec->print(str, query_type);
   str->append(')');

sql/sql_base.cc

@@ -550,4 +550,7 @@ void st_select_lex::set_with_clause(With_clause *with_clause)
     with_clause->set_owner(master_unit());
 }
 
+void list_strlex_print(THD *thd, String *str, List<Lex_ident_sys> *list,
+                              bool bracketed= false);
+
 #endif /* SQL_CTE_INCLUDED */

sql/sql_cte.cc

@@ -1354,6 +1354,9 @@ bool mysql_derived_reinit(THD *thd, LEX *lex, TABLE_LIST *derived)
                        derived->get_unit()));
   st_select_lex_unit *unit= derived->get_unit();
 
+  if (derived->original_names_are_saved)
+    unit->first_select()->set_item_list_names(derived->original_names);
+
   // reset item names to that saved after wildcard expansion in JOIN::prepare
   for(st_select_lex *sl= unit->first_select(); sl; sl= sl->next_select())
     sl->restore_item_list_names();

sql/sql_cte.h

@@ -3606,6 +3606,42 @@ List<Item>* st_select_lex::get_item_list()
 }
 
 
+/**
+  @brief
+    Replace the name of each item in the item_list with a new name.
+
+  @param
+    new_names     pointer to a List of Lex_ident_sys from which replacement
+    names are taken.
+
+  @details
+    This is used in derived tables to optionally set the names in the item_list.
+    Usually called in unit::prepare().
+
+  @retval
+    true:  an error occurred
+    false: success
+*/
+
+bool st_select_lex::set_item_list_names(List<Lex_ident_sys> *new_names)
+{
+  if (item_list.elements != new_names->elements)
+  {
+    my_error(ER_INCORRECT_COLUMN_NAME_COUNT, MYF(0));
+    return true;
+  }
+
+  List_iterator<Lex_ident_sys> it(*new_names);
+  List_iterator_fast<Item> li(item_list);
+  Item *item;
+
+  while ((item= li++))
+    lex_string_set( &item->name, (it++)->str);
+
+  return false;
+}
+
+
 uint st_select_lex::get_cardinality_of_ref_ptrs_slice(uint order_group_num_arg)
 {
   if (!((options & SELECT_DISTINCT) && !group_list.elements))
@@ -10705,7 +10741,8 @@ SELECT_LEX *LEX::parsed_TVC_end()
 
 TABLE_LIST *LEX::parsed_derived_table(SELECT_LEX_UNIT *unit,
                                      int for_system_time,
-                                     LEX_CSTRING *alias)
+                                     LEX_CSTRING *alias,
+                                     List<Lex_ident_sys> *column_names)
 {
   TABLE_LIST *res;
   derived_tables|= DERIVED_SUBQUERY;
@@ -10726,6 +10763,15 @@ TABLE_LIST *LEX::parsed_derived_table(SELECT_LEX_UNIT *unit,
   {
     res->vers_conditions= vers_conditions;
   }
+
+  if (column_names && column_names->elements > 0)
+  {
+    res->column_names= column_names;
+    // pre-allocate space to save item_list names
+    res->original_names= new (thd->mem_root) List<Lex_ident_sys>;
+    for (uint i= 0; i < column_names->elements; i++)
+      res->original_names->push_back( new Lex_ident_sys );
+  }
   return res;
 }
 

sql/sql_derived.cc

@@ -803,6 +803,8 @@ class st_select_lex_unit: public st_select_lex_node {
   bool is_view:1;
   bool describe:1; /* union exec() called for EXPLAIN */
   bool columns_are_renamed:1;
+  inline bool rename_item_list(TABLE_LIST *derived_arg);
+  inline void rename_types_list(List<Lex_ident_sys> *new_names);
 
 protected:
   /* This is bool, not bit, as it's used and set in many places */
@@ -1322,7 +1324,7 @@ class st_select_lex: public st_select_lex_node
   List<Item>* get_item_list();
   bool save_item_list_names(THD *thd);
   void restore_item_list_names();
-
+  bool set_item_list_names( List<Lex_ident_sys> *overwrite );
   ulong get_table_join_options();
   void set_lock_for_tables(thr_lock_type lock_type, bool for_update,
                            bool skip_locks);
@@ -4728,7 +4730,8 @@ struct LEX: public Query_tables_list
   SELECT_LEX *parsed_TVC_end();
   TABLE_LIST *parsed_derived_table(SELECT_LEX_UNIT *unit,
                                    int for_system_time,
-                                   LEX_CSTRING *alias);
+                                   LEX_CSTRING *alias,
+                                   List<Lex_ident_sys> *column_names= nullptr);
   bool parsed_create_view(SELECT_LEX_UNIT *unit, int check);
   bool select_finalize(st_select_lex_unit *expr);
   bool select_finalize(st_select_lex_unit *expr, Lex_select_lock l);

sql/sql_lex.cc

@@ -31613,6 +31613,9 @@ void TABLE_LIST::print(THD *thd, table_map eliminated_tables, String *str,
       str->append(' ');
       append_identifier_opt_casedn(thd, str, alias,
                                    lower_case_table_names == 1);
+
+      if (column_names && (column_names->elements > 0))
+        list_strlex_print(thd, str, column_names, true);
     }
 
     if (index_hints)