MDEV-38734: Virtual columns wrongly included in binlog_row_image=MINIMAL

The original symptom of this was an assertion 'marked_for_read()' failing in
RBR with unique blobs and binlog_row_image=MINIMAL. The problem was that the
hidden DB_ROW_HASH_1 virtual column was included in the after-image of the
update, but the underlying blob column was not being updated, so it was not
in the read_set/write_set.

It seems clearly wrong to include the DB_ROW_HASH_1 in the after-image when
the underlying blob isn't even being updated. The cause of this is the
following commit:

Author: Monty <[email protected]>
Date:   Wed May 23 22:42:29 2018 +0300

    MDEV-15243 Crash with virtual fields and row based binary logging

That patch removed a check for if the underlying fields of a virtual column
were being updated, and just added them unconditionally. This seems wrong.

So revert that part of the commit, restoring the logic to only add a virtual
column if any underlying field is actually in the write_set.

Also fix a typo in that commit where a code reformat accidentally reversed a
condition.

Also fix an assertion when InnoDB goes to update secondary indexes: If any
part of the primary key is being updated, then add all virtual columns that
are part of secondary indexes to the read_set.

Signed-off-by: Kristian Nielsen <[email protected]>

Author: knielsen

mysql-test/suite/rpl/r/rpl_mdev38734.result

@@ -0,0 +1,17 @@
+include/master-slave.inc
+[connection master]
+connection slave;
+SET @row_image_save = @@global.binlog_row_image;
+SET GLOBAL binlog_row_image=MINIMAL;
+include/stop_slave.inc
+include/start_slave.inc
+connection master;
+SET binlog_row_image=MINIMAL;
+CREATE TABLE t (pk int primary key, a text, b text, unique(b));
+INSERT INTO t VALUES (1,'foo', 'bar');
+UPDATE t SET a = 'baz';
+connection slave;
+SET GLOBAL binlog_row_image = @row_image_save;
+connection master;
+DROP TABLE t;
+include/rpl_end.inc

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

@@ -0,0 +1,21 @@
+--source include/have_binlog_format_row.inc
+--source include/master-slave.inc
+ 
+--connection slave
+SET @row_image_save = @@global.binlog_row_image;
+SET GLOBAL binlog_row_image=MINIMAL;
+--source include/stop_slave.inc
+--source include/start_slave.inc
+ 
+--connection master
+SET binlog_row_image=MINIMAL;
+CREATE TABLE t (pk int primary key, a text, b text, unique(b));
+INSERT INTO t VALUES (1,'foo', 'bar');
+UPDATE t SET a = 'baz';
+ 
+--sync_slave_with_master
+SET GLOBAL binlog_row_image = @row_image_save;
+ 
+--connection master
+DROP TABLE t;
+--source include/rpl_end.inc

sql/table.cc

@@ -8186,8 +8186,7 @@ void TABLE::mark_columns_per_binlog_row_image()
     be added to read_set either.
 */
 
-bool TABLE::mark_virtual_columns_for_write(bool insert_fl
-                                           __attribute__((unused)))
+bool TABLE::mark_virtual_columns_for_write(bool insert_fl)
 {
   Field **vfield_ptr, *tmp_vfield;
   bool bitmap_updated= false;
@@ -8202,9 +8201,29 @@ bool TABLE::mark_virtual_columns_for_write(bool insert_fl
              (tmp_vfield->flags & (PART_KEY_FLAG | FIELD_IN_PART_FUNC_FLAG |
                                    PART_INDIRECT_KEY_FLAG)))
     {
-      bitmap_set_bit(write_set, tmp_vfield->field_index);
-      mark_virtual_column_with_deps(tmp_vfield);
-      bitmap_updated= true;
+      if (insert_fl)
+      {
+        bitmap_set_bit(write_set, tmp_vfield->field_index);
+        mark_virtual_column_with_deps(tmp_vfield);
+        bitmap_updated= true;
+      }
+      else
+      {
+        MY_BITMAP *save_read_set= read_set;
+        Item *vcol_item= tmp_vfield->vcol_info->expr;
+        DBUG_ASSERT(vcol_item);
+        bitmap_clear_all(&tmp_set);
+        read_set= &tmp_set;
+        vcol_item->walk(&Item::register_field_in_read_map, 1, 0);
+        read_set= save_read_set;
+        if (bitmap_is_overlapping(&tmp_set, write_set))
+        {
+          bitmap_set_bit(write_set, tmp_vfield->field_index);
+          bitmap_set_bit(read_set, tmp_vfield->field_index);
+          bitmap_union(read_set, &tmp_set);
+          bitmap_updated= true;
+        }
+      }
     }
   }
   if (bitmap_updated)
@@ -9267,7 +9286,7 @@ int TABLE::update_virtual_fields(handler *h, enum_vcol_update_mode update_mode)
       /* Read indexed fields that was not updated in VCOL_UPDATE_FOR_READ */
       update= (!vcol_info->is_stored() &&
                (vf->flags & (PART_KEY_FLAG | PART_INDIRECT_KEY_FLAG)) &&
-               !bitmap_is_set(read_set, vf->field_index));
+               bitmap_is_set(read_set, vf->field_index));
       swap_values= 1;
       break;
     }

storage/innobase/handler/ha_innodb.cc

@@ -5046,14 +5046,33 @@ Ensure that indexed virtual columns will be computed.
 Needs to be done for indexes that are being added with inplace ALTER
 in a different thread, because from the server point of view these
 columns are not yet indexed.
+Also needed if the primary key is being updated.
 */
 void ha_innobase::column_bitmaps_signal()
 {
   if (!table->vfield || table->current_lock != F_WRLCK)
     return;
 
   dict_index_t* clust_index= dict_table_get_first_index(m_prebuilt->table);
-  if (!clust_index->online_log)
+  bool is_online_log= clust_index->online_log;
+
+  /*
+    Check if the clustered index is being updated.
+    If so, it is necessary to compute virtual columns that are part of
+    secondary indexes, to be able to re-compute those indexes.
+  */
+  bool upd_pk= false;
+  for (uint j= 0; j < clust_index->n_user_defined_cols; ++j)
+  {
+    dict_col_t *col= clust_index->fields[j].col;
+    if (bitmap_is_set(table->write_set, static_cast<uint>(col->ind)))
+    {
+      upd_pk= 1;
+      break;
+    }
+  }
+
+  if (!is_online_log && !upd_pk)
     return;
 
   uint num_v= 0;
@@ -5064,7 +5083,7 @@ void ha_innobase::column_bitmaps_signal()
 
     dict_col_t *col= &m_prebuilt->table->v_cols[num_v].m_col;
     if (col->ord_part ||
-        (dict_index_is_online_ddl(clust_index) &&
+        (is_online_log && dict_index_is_online_ddl(clust_index) &&
          row_log_col_is_indexed(clust_index, num_v)))
       table->mark_virtual_column_with_deps(table->vfield[j]);
     num_v++;