MDEV-29155 CREATE OR REPLACE with self-referencing CHECK hangs

	   forever, cannot be killed

Virtual_column_info::fix_and_check_expr() first does fix_expr() which
finds all fields in query tables and then check_vcol_func_processor()
which prohibits SELECT expression. So before we get the error
prohibiting SELECT in vcol expression we must satisfy fix_expr() with
all the opened tables. The patch achieves this by iterating
query_tables from parsed vcol expression and by assigning opened
tables from the main parser context (as they are already preopened if
the parser sees some SELECT expressions).

But the problem is, we cannot use these TABLE objects fully for vcol
expression, at least the comment about MERGE tables states so:

  /* MERGE tables need to access parent and child TABLE_LISTs. */
  DBUG_ASSERT(tables->table->pos_in_table_list == tables);

Therefore after we have done vcol check we should revert back
TABLE_LISTs to the original TABLE-less state.

As CREATE OR REPLACE first drops the original table (at least prior to
MDEV-25292) we can use the above hack from the currently opening
table. And that is possible only after bitmaps initialized, so we move
their execution to a little earlier stage, before vcol parsing. But
partitioning depends on uninitialized bitmaps, so we temporarily
revert some to make partitioning initialization happy.

Note that plain CREATE TABLE just fails this case in parser with
NO_SUCH_TABLE, CREATE OR REPLACE doesn't fail in parser as the old
table still exists.

Now to the hang, mysql_rm_table_no_locks() does TDC_RT_REMOVE_ALL
which waits while share is closed. The table normally is open only as
OPEN_STUB, this is what parser does for CREATE TABLE. But for SELECT
the table is opened not as a stub. If it is the same table name we
anyway have two TABLE_LIST objects: stub and not stub. So for "not
stub" TDC_RT_REMOVE_ALL sees open count and decides to wait until it
is closed. And of course it hangs, because that was opened in the same
thread. Now we force close such TABLE objects before
mysql_rm_table_no_locks().

And the third, condition for sequences was wrong: we have to check
TABLE_LIST::sequence property to make sure we processing sequence.

mysql-test/suite/vcol/r/not_supported.result

@@ -55,3 +55,28 @@ ERROR HY000: Function or expression 'b' cannot be used in the GENERATED ALWAYS A
 #
 # End of 10.3 tests
 #
+#
+# MDEV-29155 CREATE OR REPLACE with self-referencing CHECK hangs forever, cannot be killed
+#
+create table t1 (a int);
+create table t2 (b int)
+# create or replace table t (b int);
+create table t3 (c int, check(exists(select a from t1) or exists(select b from t2)));
+ERROR HY000: Function or expression 'select ...' cannot be used in the CHECK clause of `CONSTRAINT_1`
+create table t3 (c int, check(exists(select c from t3)));
+ERROR 42S02: Table 'test.t3' doesn't exist
+create table t3 (d int);
+create or replace table t3 (c int, check(exists(select a from t1) or exists(select b from t2)));
+ERROR HY000: Function or expression 'select ...' cannot be used in the CHECK clause of `CONSTRAINT_1`
+create table t3 (d int);
+create or replace table t3 (c int, check(exists(select c from t3)));
+ERROR HY000: Function or expression 'select ...' cannot be used in the CHECK clause of `CONSTRAINT_1`
+create table t3 (c int);
+alter table t3 add check(exists(select a from t1) or exists(select b from t2));
+ERROR HY000: Function or expression 'select ...' cannot be used in the CHECK clause of `CONSTRAINT_1`
+alter table t3 add check(exists(select c from t3));
+ERROR HY000: Function or expression 'select ...' cannot be used in the CHECK clause of `CONSTRAINT_1`
+drop tables t1, t2, t3;
+#
+# End of 10.4 tests
+#

mysql-test/suite/vcol/t/not_supported.test

@@ -64,3 +64,30 @@ create table t1 (a int auto_increment primary key,
 --echo #
 --echo # End of 10.3 tests
 --echo #
+
+--echo #
+--echo # MDEV-29155 CREATE OR REPLACE with self-referencing CHECK hangs forever, cannot be killed
+--echo #
+create table t1 (a int);
+create table t2 (b int)
+# create or replace table t (b int);
+--error ER_GENERATED_COLUMN_FUNCTION_IS_NOT_ALLOWED
+create table t3 (c int, check(exists(select a from t1) or exists(select b from t2)));
+--error ER_NO_SUCH_TABLE
+create table t3 (c int, check(exists(select c from t3)));
+create table t3 (d int);
+--error ER_GENERATED_COLUMN_FUNCTION_IS_NOT_ALLOWED
+create or replace table t3 (c int, check(exists(select a from t1) or exists(select b from t2)));
+create table t3 (d int);
+--error ER_GENERATED_COLUMN_FUNCTION_IS_NOT_ALLOWED
+create or replace table t3 (c int, check(exists(select c from t3)));
+create table t3 (c int);
+--error ER_GENERATED_COLUMN_FUNCTION_IS_NOT_ALLOWED
+alter table t3 add check(exists(select a from t1) or exists(select b from t2));
+--error ER_GENERATED_COLUMN_FUNCTION_IS_NOT_ALLOWED
+alter table t3 add check(exists(select c from t3));
+drop tables t1, t2, t3;
+
+--echo #
+--echo # End of 10.4 tests
+--echo #

sql/lex_string.h

@@ -28,6 +28,20 @@ static inline bool lex_string_cmp(CHARSET_INFO *charset, const LEX_CSTRING *a,
   return my_strcasecmp(charset, a->str, b->str);
 }
 
+/*
+  This should be replaced by C++20 "spaceship" operator (<=>)
+*/
+template <typename T>
+inline int
+cmp_any(T a, T b)
+{
+  if (a < b)
+    return -1;
+  if (b < a)
+    return 1;
+  return 0;
+}
+
 /*
   Compare to LEX_CSTRING's and return 0 if equal
 */
@@ -41,6 +55,19 @@ static inline bool cmp(const LEX_CSTRING a, const LEX_CSTRING b)
 {
   return a.length != b.length || (a.length && memcmp(a.str, b.str, a.length));
 }
+static inline int cmp_table(const LEX_CSTRING a, const LEX_CSTRING b)
+{
+  /*
+    NB: no my_strncasecmp() and therefore the below assertions must pass.
+  */
+  DBUG_ASSERT(strlen(a.str) == a.length);
+  DBUG_ASSERT(strlen(b.str) == b.length);
+  int ret= cmp_any(a.length, b.length);
+  if (ret)
+    return ret;
+  return my_strcasecmp(table_alias_charset, a.str, b.str);
+}
+
 
 /*
   Compare if two LEX_CSTRING are equal. Assumption is that

sql/sql_table.cc

@@ -5108,6 +5108,20 @@ int create_table_impl(THD *thd, const LEX_CSTRING &orig_db,
           thd->transaction.stmt.m_unsafe_rollback_flags=
             save_unsafe_rollback_flags;
         }
+        bool closed= false;
+        if (thd->lex->query_tables)
+          for (TABLE_LIST *tl= thd->lex->query_tables->next_global; tl; tl= tl->next_global)
+          {
+            if (!tl->table || tl->cmp_name(&table_list))
+              continue;
+            if (!closed)
+            {
+              close_all_tables_for_name(thd, tl->table->s,
+                                        HA_EXTRA_PREPARE_FOR_DROP, NULL);
+              closed= true;
+            }
+            tl->table= NULL;
+          }
         /* Remove normal table without logging. Keep tables locked */
         if (mysql_rm_table_no_locks(thd, &table_list, 0, 0, 0, 0, 1, 1))
           goto err;

sql/table.cc

@@ -3551,6 +3551,7 @@ unpack_vcol_info_from_frm(THD *thd, TABLE *table,
   LEX *old_lex= thd->lex;
   LEX lex;
   bool error;
+  TABLE_LIST *sequence;
   DBUG_ENTER("unpack_vcol_info_from_frm");
 
   DBUG_ASSERT(vcol->expr == NULL);
@@ -3568,10 +3569,33 @@ unpack_vcol_info_from_frm(THD *thd, TABLE *table,
   if (unlikely(error))
     goto end;
 
-  if (lex.current_select->table_list.first[0].next_global)
+  /*
+    Assign opened TABLE objects to lex.query_tables to make fix_and_check_expr()
+    happy. Probably can be useful for implementing SELECT inside vcol expressions.
+  */
+  for (TABLE_LIST *vcol_tab= lex.query_tables; vcol_tab; vcol_tab= vcol_tab->next_global)
+  {
+    if (!vcol_tab->cmp_name(table))
+    {
+      /*
+        This one is important for CREATE OR REPLACE as the original table was
+        already deleted, so cannot get TABLE from old_lex.
+      */
+      vcol_tab->table= table;
+      continue;
+    }
+    for (TABLE_LIST *tab= old_lex->query_tables; tab; tab= tab->next_global)
+    {
+      if (!tab->table || vcol_tab->cmp_name(tab))
+        continue;
+      vcol_tab->table= tab->table;
+    }
+  }
+
+  sequence= lex.current_select->table_list.first[0].next_global;
+  if (sequence && sequence->sequence)
   {
     /* We are using NEXT VALUE FOR sequence. Remember table name for open */
-    TABLE_LIST *sequence= lex.current_select->table_list.first[0].next_global;
     sequence->next_global= table->internal_tables;
     table->internal_tables= sequence;
   }
@@ -3584,6 +3608,14 @@ unpack_vcol_info_from_frm(THD *thd, TABLE *table,
   {
     *vcol_ptr= vcol_info= vcol_storage.vcol_info;   // Expression ok
     DBUG_ASSERT(vcol_info->expr);
+    /*
+      Revert back TABLE objects assignment.
+      open_and_process_table() will be unhappy at:
+
+        DBUG_ASSERT(tables->table->pos_in_table_list == tables);
+    */
+    for (TABLE_LIST *vcol_tab= lex.query_tables; vcol_tab; vcol_tab= vcol_tab->next_global)
+      vcol_tab->table= NULL;
     goto end;
   }
   *error_reported= TRUE;
@@ -3940,6 +3972,40 @@ enum open_frm_error open_table_from_share(THD *thd, TABLE_SHARE *share,
     }
   }
 
+  /* Allocate bitmaps */
+
+  bitmap_size= share->column_bitmap_size;
+  bitmap_count= 7;
+  if (share->virtual_fields)
+    bitmap_count++;
+
+  if (!(bitmaps= (uchar*) alloc_root(&outparam->mem_root,
+                                     bitmap_size * bitmap_count)))
+    goto err;
+
+  my_bitmap_init(&outparam->def_read_set,
+                 (my_bitmap_map*) bitmaps, share->fields, FALSE);
+  bitmaps+= bitmap_size;
+  my_bitmap_init(&outparam->def_write_set,
+                 (my_bitmap_map*) bitmaps, share->fields, FALSE);
+  bitmaps+= bitmap_size;
+
+  my_bitmap_init(&outparam->has_value_set,
+                 (my_bitmap_map*) bitmaps, share->fields, FALSE);
+  bitmaps+= bitmap_size;
+  my_bitmap_init(&outparam->tmp_set,
+                 (my_bitmap_map*) bitmaps, share->fields, FALSE);
+  bitmaps+= bitmap_size;
+  my_bitmap_init(&outparam->eq_join_set,
+                 (my_bitmap_map*) bitmaps, share->fields, FALSE);
+  bitmaps+= bitmap_size;
+  my_bitmap_init(&outparam->cond_set,
+                 (my_bitmap_map*) bitmaps, share->fields, FALSE);
+  bitmaps+= bitmap_size;
+  my_bitmap_init(&outparam->def_rpl_write_set,
+                 (my_bitmap_map*) bitmaps, share->fields, FALSE);
+  outparam->default_column_bitmaps();
+
   /*
     Process virtual and default columns, if any.
   */
@@ -4016,6 +4082,7 @@ enum open_frm_error open_table_from_share(THD *thd, TABLE_SHARE *share,
   bool work_part_info_used;
   if (share->partition_info_str_len && outparam->file)
   {
+    MY_BITMAP *write_set;
   /*
     In this execution we must avoid calling thd->change_item_tree since
     we might release memory before statement is completed. We do this
@@ -4052,6 +4119,8 @@ enum open_frm_error open_table_from_share(THD *thd, TABLE_SHARE *share,
       We should perform the fix_partition_func in either local or
       caller's arena depending on work_part_info_used value.
     */
+    write_set= outparam->write_set;
+    outparam->write_set= NULL;
     if (!work_part_info_used)
       tmp= fix_partition_func(thd, outparam, is_create_table);
     thd->stmt_arena= backup_stmt_arena_ptr;
@@ -4061,6 +4130,7 @@ enum open_frm_error open_table_from_share(THD *thd, TABLE_SHARE *share,
       if (work_part_info_used)
         tmp= fix_partition_func(thd, outparam, is_create_table);
     }
+    outparam->write_set= write_set;
     outparam->part_info->item_free_list= part_func_arena.free_list;
 partititon_err:
     if (tmp)
@@ -4090,40 +4160,6 @@ enum open_frm_error open_table_from_share(THD *thd, TABLE_SHARE *share,
     goto err;
   }
 
-  /* Allocate bitmaps */
-
-  bitmap_size= share->column_bitmap_size;
-  bitmap_count= 7;
-  if (share->virtual_fields)
-    bitmap_count++;
-
-  if (!(bitmaps= (uchar*) alloc_root(&outparam->mem_root,
-                                     bitmap_size * bitmap_count)))
-    goto err;
-
-  my_bitmap_init(&outparam->def_read_set,
-                 (my_bitmap_map*) bitmaps, share->fields, FALSE);
-  bitmaps+= bitmap_size;
-  my_bitmap_init(&outparam->def_write_set,
-                 (my_bitmap_map*) bitmaps, share->fields, FALSE);
-  bitmaps+= bitmap_size;
-
-  my_bitmap_init(&outparam->has_value_set,
-                 (my_bitmap_map*) bitmaps, share->fields, FALSE);
-  bitmaps+= bitmap_size;
-  my_bitmap_init(&outparam->tmp_set,
-                 (my_bitmap_map*) bitmaps, share->fields, FALSE);
-  bitmaps+= bitmap_size;
-  my_bitmap_init(&outparam->eq_join_set,
-                 (my_bitmap_map*) bitmaps, share->fields, FALSE);
-  bitmaps+= bitmap_size;
-  my_bitmap_init(&outparam->cond_set,
-                 (my_bitmap_map*) bitmaps, share->fields, FALSE);
-  bitmaps+= bitmap_size;
-  my_bitmap_init(&outparam->def_rpl_write_set,
-                 (my_bitmap_map*) bitmaps, share->fields, FALSE);
-  outparam->default_column_bitmaps();
-
   outparam->cond_selectivity= 1.0;
 
   /* The table struct is now initialized;  Open the table */

sql/table.h

@@ -2958,6 +2958,23 @@ struct TABLE_LIST
     tabledef_version.str= (const uchar *) version->str;
     tabledef_version.length= version->length;
   }
+
+  int cmp_name(TABLE_LIST *tl)
+  {
+    int res= cmp_table(db, tl->db);
+    if (res)
+      return res;
+    return cmp_table(table_name, tl->table_name);
+  }
+
+  int cmp_name(TABLE *tab)
+  {
+    int res= cmp_table(db, tab->s->db);
+    if (res)
+      return res;
+    return cmp_table(table_name, tab->s->table_name);
+  }
+
 private:
   bool prep_check_option(THD *thd, uint8 check_opt_type);
   bool prep_where(THD *thd, Item **conds, bool no_where_clause);