MDEV-36290: Fix conv_table fields when slave is missing a column

There was an issue with the conversion table, where its columns
weren't in sync with the table_def generated by the
Table_map_log_event. I.e., when the slave didn't have the column,
the conv_table generation logic would simply skip the rest of the
loop and go on to the next column (without adding a field for the
missing column). The compatible_with and unpack_row logic, however,
assumed the conv_table was in alignment with the columns from the
Rows_log_event. I thought of two ways to fix this:
  1. Add a dummy field to the conv_table (and later reset it to NULL)
  2. Have a separate index variable which tracks the conv_table fields

This patch implements the second, as it is more efficient.

Also, it extends testing to cover basic ADD COLUMN and DROP COLUMN
options.

mysql-test/suite/rpl/t/rpl_master_slave_mismatched_columns.inc

@@ -19,6 +19,10 @@
 #                      shall generate an error
 #    $engine_type      The storage engine to be used for storing table
 #                      on both master and slave
+#    $add_col          Boolean.
+#    $add_col_loc      One of "first", "middle", "end"
+#    $drop_col         One of the existing columns, i.e. pk, a, b, or c
+#
 #
 
 if (!$engine_type)
@@ -53,51 +57,87 @@ eval create table t1(
 --echo #
 eval alter table t1 modify b $target_type;
 
-alter table t1 add column new_col_first int default 99 first;
+if ($add_col)
+{
+  if (`SELECT strcmp("$add_col_loc","first")=0`)
+  {
+    --let $loc_pos= first
+  }
+  if (`SELECT strcmp("$add_col_loc","middle")=0`)
+  {
+    --let $loc_pos= after a
+  }
+  if (`SELECT strcmp("$add_col_loc","end")=0`)
+  {
+    --let $loc_pos= after c
+  }
+
+  --eval alter table t1 add column new_col int default 99 $loc_pos
+}
+
+if ($drop_col)
+{
+  --eval alter table t1 drop column $drop_col
+}
 --source include/save_master_gtid.inc
 
 --connection server_3
 --source include/sync_with_master_gtid.inc
 
 connection server_1;
+--let $master_pk= 1
 --let $master_a= 1000
 --let $master_b= $source_value
 --let $master_c= 2000
-eval INSERT INTO t1 (pk, a, b, c) VALUES(1, $master_a, $master_b, $master_c);
+eval INSERT INTO t1 (pk, a, b, c) VALUES($master_pk, $master_a, $master_b, $master_c);
 --source include/save_master_gtid.inc
 
 --connection server_3
 --source include/sync_with_master_gtid.inc
+
+--echo # Ensure server_2 and server_3 are consistent with one another..
+--let $diff_tables= server_2:test.t1, server_3:test.t1
+--source include/diff_tables.inc
 --connection server_1
 select * from t1;
+
+--connection server_2
+select * from t1;
+
 if ($can_convert) {
+  --echo # Ensure server_2 state is correct (server_3 is proven to be
+  --echo #   consistent, so we don't validate its data)..
 
-  --let $conn=2
-  while ($conn)
+  # Index-to-table column map for value checking:
+  #  4=pk, 3=a, 2=b, 1=c
+  --let $cols_to_test= 4
+  while ($cols_to_test)
   {
-    --let $conn_no= `SELECT ($conn+1)`
-    --let $rpl_connection_name= server_$conn_no
-    --source include/rpl_connection.inc
-
-    # Map so we can test columns:
-    # 3=a, 2=b, 1=c
-    --let $cols_to_test= 3
-    while ($cols_to_test)
+    --let $col= pk
+    --let $master_val= $master_pk
+    if (`SELECT $cols_to_test=3`)
     {
       --let $col= a
       --let $master_val= $master_a
-      if (`SELECT $cols_to_test=2`)
-      {
-        --let $col= b
-        --let $master_val= $master_b
-      }
-      if (`SELECT $cols_to_test=1`)
-      {
-        --let $col= c
-        --let $master_val= $master_c
-      }
+    }
+    if (`SELECT $cols_to_test=2`)
+    {
+      --let $col= b
+      --let $master_val= $master_b
+    }
+    if (`SELECT $cols_to_test=1`)
+    {
+      --let $col= c
+      --let $master_val= $master_c
+    }
 
-      eval select $col from t1;
+    if (`SELECT strcmp("$col","$drop_col")=0`)
+    {
+      --echo # Column $col was dropped, skipping its validation
+    }
+
+    if (`SELECT strcmp("$col","$drop_col")!=0`)
+    {
       if (`select $col != $master_val from t1`)
       {
         --let $real_val= `SELECT $col from t1`
@@ -106,12 +146,11 @@ if ($can_convert) {
         --echo # Real value (server_$conn_no): $real_val
         --die Server_2: Field name lookup with type conversion failed
       }
-
-      --dec $cols_to_test
     }
 
-    --dec $conn
+    --dec $cols_to_test
   }
+  --echo # ..done
 }
 if (!$can_convert) {
   --echo #

mysql-test/suite/rpl/t/rpl_master_slave_mismatched_columns.test

@@ -34,10 +34,14 @@
 #   10. ALTER TABLE adds a column with the same name as an existing column,
 #      but with a prefix, and before the existing column. Replication should
 #      not update the new column. *TODO
+#   11. Multi-table transaction with each target table on the slave having
+#       different column structure than the master. *TODO
 #
 # TODO:
 #   * BINLOG_ROW_IMAGE=MINIMAL and BINLOG_ROW_METADATA=FULL
 #   * ERROR Cases
+#     E1. Master and slave have matching table names but no matching column
+#         names
 #
 # References:
 #   MDEV-36290: ALTER TABLE with multi-master can cause data loss
@@ -63,20 +67,55 @@ set global slave_type_conversions='ALL_NON_LOSSY';
 set @saved_slave_type_conversions = @@global.slave_type_conversions;
 set global slave_type_conversions='ALL_NON_LOSSY';
 
-let $source_type  = TINYBLOB;
-let $target_type  = TINYBLOB;
-let $source_value = 'aaa';
-let $target_value = 'aaa';
-let $can_convert   = 1;
+--echo #
+--echo # Test Cases 1-3: Add columns
+--let $source_type= TINYBLOB
+--let $target_type= BLOB
+--let $source_value= 'aaa'
+--let $target_value= 'aaa'
+--let $can_convert= 1
+--let $add_col= 1
+--let $add_col_loc= first
 --source rpl_master_slave_mismatched_columns.inc
 
-let $source_type= TINYBLOB;
-let $target_type= BLOB;
-let $source_value= 'aaa';
-let $target_value= 'aaa';
-let $can_convert= 1;
+--let $add_col_loc= middle
 --source rpl_master_slave_mismatched_columns.inc
 
+--let $add_col_loc= end
+--source rpl_master_slave_mismatched_columns.inc
+
+--let $source_type=
+--let $target_type=
+--let $source_value=
+--let $target_value=
+--let $can_convert=
+--let $add_col=
+--let $add_col_loc=
+
+--echo #
+--echo # Test Case 4-6: Drop columns
+--let $source_type= TINYBLOB
+--let $target_type= BLOB
+--let $source_value= 'aaa'
+--let $target_value= 'aaa'
+--let $can_convert= 1
+
+--let $drop_col= a
+--source rpl_master_slave_mismatched_columns.inc
+--let $drop_col= b
+--source rpl_master_slave_mismatched_columns.inc
+--let $drop_col= c
+--source rpl_master_slave_mismatched_columns.inc
+--let $drop_col= pk
+--source rpl_master_slave_mismatched_columns.inc
+
+--let $source_type=
+--let $target_type=
+--let $source_value=
+--let $target_value=
+--let $can_convert=
+--let $drop_col=
+
 --echo #
 --echo # Cleanup
 --connection server_1

sql/rpl_record.cc

@@ -252,6 +252,7 @@ unpack_row(rpl_group_info *rgi,
   uchar const *pack_ptr= row_data + master_null_byte_count;
 
   Field **field_ptr= NULL;
+  uint conv_table_idx= 0;
 
   if (bitmap_is_clear_all(cols))
   {
@@ -351,7 +352,7 @@ unpack_row(rpl_group_info *rgi,
      */
     fprintf(stderr, "\n\tpack_row Unpacking master col %u to slave col %u\n", master_idx, slave_field_idx);
     Field *conv_field=
-      conv_table ? conv_table->field[master_idx] : NULL;
+      conv_table ? conv_table->field[conv_table_idx++] : NULL;
     Field **field_ptr= &(table->field[slave_field_idx]);
     Field *const f=
       conv_field ? conv_field : *field_ptr;

sql/rpl_utility.h

@@ -226,6 +226,16 @@ public:
    field is going to be pushed into, so the target table that the data
    eventually need to be pushed into need to be supplied.
 
+   Note that the fields generated in the conversion table are not guaranteed to
+   align with the fields from this table_def. If the slave doesn't have the
+   target field, we don't generate a field in the conversion_table, as it would
+   serve no purpose. If the conversion table is referenced while iterating
+   through this table_def, one needs a separate index to keep track of the
+   conv_table fields, which are only incremented when the slave has that
+   column. This can be checked by using RPL_TABLE_LIST::lookup_slave_column().
+   See other member function of table_def compatible_with() for an example of
+   this.
+
    @param thd Thread to allocate memory from.
    @param rli Relay log info structure, for error reporting.
    @param target_table Target table for fields.

sql/rpl_utility_server.cc

@@ -929,6 +929,7 @@ bool table_def::compatible_with(THD *thd, rpl_group_info *rgi,
   ulong master_table_n_cols= size();
   Relay_log_info *rli= rgi->rli;
   TABLE *tmp_table= NULL;
+  uint conv_table_idx= 0;
 
   for (uint col= 0 ; col < master_table_n_cols ; ++col)
   {
@@ -976,13 +977,18 @@ bool table_def::compatible_with(THD *thd, rpl_group_info *rgi,
             return false;
         /*
           Clear all fields up to, but not including, this column.
+
+          Note we use conv_table_idx to stay consistent with the columns
+          in the conv_table, as the slave may be missing columns that were
+          present on the master, and these columns are not added to the
+          conv_table.
         */
-        for (unsigned int i= 0; i < col; ++i)
+        for (unsigned int i= 0; i < conv_table_idx; ++i)
           tmp_table->field[i]= NULL;
       }
 
       if (convtype == CONV_TYPE_PRECISE && tmp_table != NULL)
-        tmp_table->field[col]= NULL;
+        tmp_table->field[conv_table_idx]= NULL;
     }
     else
     {
@@ -1008,6 +1014,7 @@ bool table_def::compatible_with(THD *thd, rpl_group_info *rgi,
                   source_type.c_ptr_safe(), target_type.c_ptr_safe());
       return false;
     }
+    conv_table_idx++;
   }
 
 #ifndef DBUG_OFF
@@ -1100,7 +1107,13 @@ TABLE *table_def::create_conversion_table(THD *thd, rpl_group_info *rgi,
       conv_table->init(cols_to_create))
     goto err;
 
-  for (uint col= 0 ; col < cols_to_create; ++col)
+  /*
+    Iterate through the number of columns logged on the master, and if any are
+    missing on the slave, that column will just be skipped. Skipped columns
+    cannot be added to the conv_table, as there is no column on the slave to
+    use as the reference for the target_field.
+  */
+  for (uint col= 0 ; col < size(); ++col)
   {
     uint slave_col_idx;
     if(target_table_list_el->lookup_slave_column(col, &slave_col_idx))