MDEV-34033 Exchange partition with virtual columns fails

MDEV-28127 did is_equal() which compared vcol expressions
literally. But another table vcol expression is not equal because of
different table name.

We implement another comparison method is_identical() which respects
different table name in vcol comparison. If any field item points to
table_A and compared field item points to table_B, such items are
treated as equal in (table_A, table_B) comparison. This is done by
cloning table_B expression and renaming any table_B entries to table_A
in it.

mysql-test/main/partition_exchange.result

@@ -1321,3 +1321,25 @@ CREATE TABLE t2 (a INT, PRIMARY KEY(a)) CHECKSUM=1, ENGINE=InnoDB;
 ALTER TABLE t1 EXCHANGE PARTITION p0 WITH TABLE t2;
 ERROR HY000: Tables have different definitions
 DROP TABLE t1, t2;
+#
+# MDEV-34033 Exchange partition with virtual columns fails
+#
+create or replace table t1(
+id int primary key,
+col1 int,
+col2 boolean as (col1 is null))
+partition by list (id) ( partition p1 values in (1)
+);
+create or replace table t1_working like t1;
+alter table t1_working remove partitioning;
+alter table t1 exchange partition p1 with table t1_working;
+create or replace table t2(
+id int primary key,
+col1 int,
+col2 boolean as (true))
+partition by list (id) ( partition p1 values in (1)
+);
+create or replace table t2_working like t2;
+alter table t2_working remove partitioning;
+alter table t2 exchange partition p1 with table t2_working;
+drop tables t1, t1_working, t2, t2_working;

mysql-test/main/partition_exchange.test

@@ -554,3 +554,35 @@ ALTER TABLE t1 EXCHANGE PARTITION p0 WITH TABLE t2;
 
 # Cleanup
 DROP TABLE t1, t2;
+
+--echo #
+--echo # MDEV-34033 Exchange partition with virtual columns fails
+--echo #
+# this fails when the virtual persistent column
+# references another column
+create or replace table t1(
+  id int primary key,
+  col1 int,
+  col2 boolean as (col1 is null))
+  partition by list (id) ( partition p1 values in (1)
+);
+
+create or replace table t1_working like t1;
+alter table t1_working remove partitioning;
+alter table t1 exchange partition p1 with table t1_working;
+
+# this works when the virtual persistent column
+# does not reference another column
+create or replace table t2(
+  id int primary key,
+  col1 int,
+  col2 boolean as (true))
+  partition by list (id) ( partition p1 values in (1)
+);
+
+create or replace table t2_working like t2;
+alter table t2_working remove partitioning;
+alter table t2 exchange partition p1 with table t2_working;
+
+# Cleanup
+drop tables t1, t1_working, t2, t2_working;

sql/field.h

@@ -659,6 +659,9 @@ class Virtual_column_info: public Sql_alloc,
   bool cleanup_session_expr();
   bool fix_and_check_expr(THD *thd, TABLE *table);
   inline bool is_equal(const Virtual_column_info* vcol) const;
+  /* Same as is_equal() but for comparing with different table */
+  inline bool is_identical(THD *thd, TABLE_SHARE *share, TABLE_SHARE *vcol_share,
+                           const Virtual_column_info* vcol, bool &error) const;
   inline void print(String*);
 };
 

sql/item.cc

@@ -829,6 +829,30 @@ bool Item_field::rename_fields_processor(void *arg)
   return 0;
 }
 
+/**
+   Rename table and clean field for EXCHANGE comparison
+*/
+
+bool Item_field::rename_table_processor(void *arg)
+{
+  Item::func_processor_rename_table *p= (Item::func_processor_rename_table*) arg;
+
+  /* If (db_name, table_name) matches (p->old_db, p->old_table)
+     rename to (p->new_db, p->new_table) */
+  if (((!db_name.str && !p->old_db.str) ||
+       db_name.streq(p->old_db)) &&
+      ((!table_name.str && !p->old_table.str) ||
+       table_name.streq(p->old_table)))
+  {
+    db_name= p->new_db;
+    table_name= p->new_table;
+  }
+
+  /* Item_field equality is done by field pointer if it is set, we need to avoid that */
+  field= NULL;
+  return 0;
+}
+
 
 /**
   Check if an Item_field references some field from a list of fields.

sql/item.h

@@ -2148,6 +2148,7 @@ class Item: public Value_source,
   virtual bool check_partition_func_processor(void *arg) { return 1;}
   virtual bool post_fix_fields_part_expr_processor(void *arg) { return 0; }
   virtual bool rename_fields_processor(void *arg) { return 0; }
+  virtual bool rename_table_processor(void *arg) { return 0; }
   /*
     TRUE if the function is knowingly TRUE or FALSE.
     Not to be used for AND/OR formulas.
@@ -2176,6 +2177,13 @@ class Item: public Value_source,
     LEX_CSTRING table_name;
     List<Create_field> fields;
   };
+  struct func_processor_rename_table
+  {
+    Lex_ident_db old_db;
+    Lex_ident_table old_table;
+    Lex_ident_db new_db;
+    Lex_ident_table new_table;
+  };
   virtual bool check_vcol_func_processor(void *arg)
   {
     return mark_unsupported_function(full_name(), arg, VCOL_IMPOSSIBLE);
@@ -3665,6 +3673,7 @@ class Item_field :public Item_ident,
   bool switch_to_nullable_fields_processor(void *arg) override;
   bool update_vcol_processor(void *arg) override;
   bool rename_fields_processor(void *arg) override;
+  bool rename_table_processor(void *arg) override;
   bool check_vcol_func_processor(void *arg) override;
   bool set_fields_as_dependent_processor(void *arg) override
   {
@@ -7805,6 +7814,27 @@ inline bool Virtual_column_info::is_equal(const Virtual_column_info* vcol) const
       && expr->eq(vcol->expr, true);
 }
 
+inline bool
+Virtual_column_info::is_identical(THD *thd, TABLE_SHARE *share, TABLE_SHARE *vcol_share,
+                                  const Virtual_column_info* vcol, bool &error) const
+{
+  error= true;
+  Item *cmp_expr= vcol->expr->build_clone(thd);
+  if (!cmp_expr)
+    return false;
+  Item::func_processor_rename_table param;
+  param.old_db=    Lex_ident_db(vcol_share->db);
+  param.old_table= Lex_ident_table(vcol_share->table_name);
+  param.new_db=    Lex_ident_db(share->db);
+  param.new_table= Lex_ident_table(share->table_name);
+  cmp_expr->walk(&Item::rename_table_processor, 1, &param);
+
+  error= false;
+  return type_handler()  == vcol->type_handler()
+      && is_stored() == vcol->is_stored()
+      && expr->eq(cmp_expr, true);
+}
+
 inline void Virtual_column_info::print(String* str)
 {
   expr->print_for_table_def(str);

sql/sql_partition_admin.cc

@@ -241,6 +241,8 @@ static bool compare_table_with_partition(THD *thd, TABLE *table,
     part_create_info.row_type= table->s->row_type;
   }
 
+  part_create_info.table= part_table;
+
   /*
     NOTE: ha_blackhole does not support check_if_compatible_data,
     so this always fail for blackhole tables.

sql/sql_table.cc

@@ -7854,8 +7854,12 @@ bool mysql_compare_tables(TABLE *table, Alter_info *alter_info,
     {
       if (!tmp_new_field->field->vcol_info)
         DBUG_RETURN(false);
-      if (!field->vcol_info->is_equal(tmp_new_field->field->vcol_info))
+      bool err;
+      if (!field->vcol_info->is_identical(thd, table->s, create_info->table->s,
+                                          tmp_new_field->field->vcol_info, err))
         DBUG_RETURN(false);
+      if (err)
+        DBUG_RETURN(true);
     }
 
     /*