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

The cause of this was several different bugs:

- When using binary logging with binlog_row_image=FULL
  the all bits in read_set was set, which caused a
  different (wrong) pattern for marking vcol_set.
- TABLE::mark_virtual_columns_for_write() didn't in all
  cases mark vcol_set with the vcol_field.
- TABLE::update_virtual_fields() has to update all
  vcol fields on REPLACE if binary logging with FULL
  is used.
- VCOL_UPDATE_INDEXED should update all vcol fields part
  of an index that was not updated by VCOL_UPDATE_FOR_READ
- max_row_length() calculated length of NULL and not
  used fields. This didn't cause any crash, but used
  more memory than needed.

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

+include/master-slave.inc
+[connection master]
+CREATE TABLE t1 (
+pk SERIAL,
+vcol_date DATE AS (col_date) PERSISTENT,
+vcol_int INT AS (col_int) VIRTUAL,
+vcol_year YEAR AS (col_year) PERSISTENT,
+vcol_blob BLOB AS (col_blob) VIRTUAL,
+col_date DATE,
+col_int INT NULL,
+col_blob BLOB NULL,
+col_year YEAR,
+PRIMARY KEY(pk)
+) ENGINE=InnoDB;
+INSERT INTO t1 (col_date,col_int,col_blob,col_year) VALUES ('2010-04-24',5,'foo',1981);
+SET SQL_MODE='';
+set binlog_row_image="FULL";
+CREATE VIEW v1 AS SELECT * FROM t1;
+REPLACE INTO v1 SELECT pk, vcol_date, vcol_int, vcol_year, vcol_blob, col_date, col_int, col_blob, 1982 FROM t1;
+Warnings:
+Warning	1906	The value specified for generated column 'vcol_date' in table 't1' ignored
+Warning	1906	The value specified for generated column 'vcol_int' in table 't1' ignored
+Warning	1906	The value specified for generated column 'vcol_year' in table 't1' ignored
+Warning	1906	The value specified for generated column 'vcol_blob' in table 't1' ignored
+select col_date,col_int,col_blob,col_year from v1;
+col_date	col_int	col_blob	col_year
+2010-04-24	5	foo	1982
+connection slave;
+select col_date,col_int,col_blob,col_year from v1;
+col_date	col_int	col_blob	col_year
+2010-04-24	5	foo	1982
+connection master;
+DROP VIEW v1;
+set binlog_row_image="MINIMAL";
+CREATE VIEW v1 AS SELECT * FROM t1;
+REPLACE INTO v1 SELECT pk, vcol_date, vcol_int, vcol_year, vcol_blob, col_date, col_int, col_blob, 1983 FROM t1;
+Warnings:
+Warning	1906	The value specified for generated column 'vcol_date' in table 't1' ignored
+Warning	1906	The value specified for generated column 'vcol_int' in table 't1' ignored
+Warning	1906	The value specified for generated column 'vcol_year' in table 't1' ignored
+Warning	1906	The value specified for generated column 'vcol_blob' in table 't1' ignored
+select col_date,col_int,col_blob,col_year from v1;
+col_date	col_int	col_blob	col_year
+2010-04-24	5	foo	1983
+connection slave;
+select col_date,col_int,col_blob,col_year from v1;
+col_date	col_int	col_blob	col_year
+2010-04-24	5	foo	1983
+connection master;
+DROP VIEW v1;
+set @@binlog_row_image="NOBLOB";
+CREATE VIEW v1 AS SELECT * FROM t1;
+REPLACE INTO v1 SELECT pk, vcol_date, vcol_int, vcol_year, vcol_blob, col_date, col_int, col_blob, 1984 FROM t1;
+Warnings:
+Warning	1906	The value specified for generated column 'vcol_date' in table 't1' ignored
+Warning	1906	The value specified for generated column 'vcol_int' in table 't1' ignored
+Warning	1906	The value specified for generated column 'vcol_year' in table 't1' ignored
+Warning	1906	The value specified for generated column 'vcol_blob' in table 't1' ignored
+select col_date,col_int,col_blob,col_year from v1;
+col_date	col_int	col_blob	col_year
+2010-04-24	5	foo	1984
+connection slave;
+select col_date,col_int,col_blob,col_year from v1;
+col_date	col_int	col_blob	col_year
+2010-04-24	5	foo	1984
+connection master;
+DROP VIEW v1;
+set @@binlog_row_image=default;
+DROP TABLE t1;
+include/rpl_end.inc

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

@@ -9,8 +9,20 @@ a	b	c
 2	3	4
 drop table t1;
 create table t1 (a int, c int as(a), p varchar(20) as(y), y char(20), index (p,c));
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL,
+  `c` int(11) GENERATED ALWAYS AS (`a`) VIRTUAL,
+  `p` varchar(20) GENERATED ALWAYS AS (`y`) VIRTUAL,
+  `y` char(20) DEFAULT NULL,
+  KEY `p` (`p`,`c`)
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
 insert into t1 (a,y) values(1, "yyy");
 update t1 set a = 100 where a = 1;
+check table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
 drop table t1;
 create table t1 (
 a varchar(10000),

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

+--source include/have_innodb.inc
+--source include/have_binlog_format_row.inc
+--source include/master-slave.inc
+
+#
+# MDEV-15243
+# Server crashes in in Field_blob::pack upon REPLACE into view with virtual
+# columns with binlog enabled
+#
+
+CREATE TABLE t1 (
+ pk SERIAL,
+ vcol_date DATE AS (col_date) PERSISTENT,
+ vcol_int INT AS (col_int) VIRTUAL,
+ vcol_year YEAR AS (col_year) PERSISTENT,
+ vcol_blob BLOB AS (col_blob) VIRTUAL,
+ col_date DATE,
+ col_int INT NULL,
+ col_blob BLOB NULL,
+ col_year YEAR,
+ PRIMARY KEY(pk)
+) ENGINE=InnoDB;
+
+INSERT INTO t1 (col_date,col_int,col_blob,col_year) VALUES ('2010-04-24',5,'foo',1981);
+SET SQL_MODE='';
+
+set binlog_row_image="FULL";
+CREATE VIEW v1 AS SELECT * FROM t1;
+REPLACE INTO v1 SELECT pk, vcol_date, vcol_int, vcol_year, vcol_blob, col_date, col_int, col_blob, 1982 FROM t1;
+select col_date,col_int,col_blob,col_year from v1;
+sync_slave_with_master;
+select col_date,col_int,col_blob,col_year from v1;
+connection master;
+DROP VIEW v1;
+set binlog_row_image="MINIMAL";
+CREATE VIEW v1 AS SELECT * FROM t1;
+REPLACE INTO v1 SELECT pk, vcol_date, vcol_int, vcol_year, vcol_blob, col_date, col_int, col_blob, 1983 FROM t1;
+select col_date,col_int,col_blob,col_year from v1;
+sync_slave_with_master;
+select col_date,col_int,col_blob,col_year from v1;
+connection master;
+DROP VIEW v1;
+set @@binlog_row_image="NOBLOB";
+CREATE VIEW v1 AS SELECT * FROM t1;
+REPLACE INTO v1 SELECT pk, vcol_date, vcol_int, vcol_year, vcol_blob, col_date, col_int, col_blob, 1984 FROM t1;
+select col_date,col_int,col_blob,col_year from v1;
+sync_slave_with_master;
+select col_date,col_int,col_blob,col_year from v1;
+connection master;
+DROP VIEW v1;
+set @@binlog_row_image=default;
+
+DROP TABLE t1;
+
+--source include/rpl_end.inc

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

@@ -15,8 +15,10 @@ drop table t1;
 # this tests TABLE::mark_columns_needed_for_update()
 #
 create table t1 (a int, c int as(a), p varchar(20) as(y), y char(20), index (p,c));
+show create table t1;
 insert into t1 (a,y) values(1, "yyy");
 update t1 set a = 100 where a = 1;
+check table t1;
 drop table t1;
 
 #

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

+#
+# Check that vcol update works with binlog enabled
+#
+
+--source include/have_binlog_format_row.inc
+
+set binlog_row_image="FULL";
+set @@default_storage_engine="myisam";
+
+--source update.test
+
+set binlog_row_image="MINIMAL";
+
+--source update.test

sql/sql_class.cc

@@ -6567,15 +6567,17 @@ int THD::binlog_delete_row(TABLE* table, bool is_trans,
 }
 
 
+/**
+   Remove from read_set spurious columns. The write_set has been
+   handled before in table->mark_columns_needed_for_update.
+*/
+
 void THD::binlog_prepare_row_images(TABLE *table)
 {
   DBUG_ENTER("THD::binlog_prepare_row_images");
-  /**
-    Remove from read_set spurious columns. The write_set has been
-    handled before in table->mark_columns_needed_for_update.
-   */
 
-  DBUG_PRINT_BITSET("debug", "table->read_set (before preparing): %s", table->read_set);
+  DBUG_PRINT_BITSET("debug", "table->read_set (before preparing): %s",
+                    table->read_set);
   THD *thd= table->in_use;
 
   /**
@@ -6593,7 +6595,7 @@ void THD::binlog_prepare_row_images(TABLE *table)
     */
     DBUG_ASSERT(table->read_set != &table->tmp_set);
 
-    switch(thd->variables.binlog_row_image)
+    switch (thd->variables.binlog_row_image)
     {
       case BINLOG_ROW_IMAGE_MINIMAL:
         /* MINIMAL: Mark only PK */
@@ -6623,7 +6625,8 @@ void THD::binlog_prepare_row_images(TABLE *table)
                                         table->write_set);
   }
 
-  DBUG_PRINT_BITSET("debug", "table->read_set (after preparing): %s", table->read_set);
+  DBUG_PRINT_BITSET("debug", "table->read_set (after preparing): %s",
+                    table->read_set);
   DBUG_VOID_RETURN;
 }
 

sql/table.cc

@@ -6566,6 +6566,12 @@ void TABLE::mark_columns_per_binlog_row_image()
         DBUG_ASSERT(FALSE);
       }
     }
+    /*
+      We have to ensure that all virtual columns that are part of read set
+      are calculated.
+    */
+    if (vcol_set)
+      bitmap_union(vcol_set, read_set);
     file->column_bitmaps_signal();
   }
 
@@ -6607,7 +6613,8 @@ bool TABLE::mark_virtual_col(Field *field)
 /* 
   @brief Mark virtual columns for update/insert commands
 
-  @param insert_fl    <-> virtual columns are marked for insert command 
+  @param insert_fl    true if virtual columns are marked for insert command
+                      For the moment this is not used, may be used in future.
 
   @details
     The function marks virtual columns used in a update/insert commands
@@ -6632,7 +6639,8 @@ bool TABLE::mark_virtual_col(Field *field)
     be added to read_set either.
 */
 
-bool TABLE::mark_virtual_columns_for_write(bool insert_fl)
+bool TABLE::mark_virtual_columns_for_write(bool insert_fl
+                                           __attribute__((unused)))
 {
   Field **vfield_ptr, *tmp_vfield;
   bool bitmap_updated= false;
@@ -6642,35 +6650,13 @@ bool TABLE::mark_virtual_columns_for_write(bool insert_fl)
   {
     tmp_vfield= *vfield_ptr;
     if (bitmap_is_set(write_set, tmp_vfield->field_index))
-      bitmap_updated= mark_virtual_col(tmp_vfield);
+      bitmap_updated|= mark_virtual_col(tmp_vfield);
     else if (tmp_vfield->vcol_info->stored_in_db ||
              (tmp_vfield->flags & (PART_KEY_FLAG | FIELD_IN_PART_FUNC_FLAG)))
     {
-      if (insert_fl)
-      {
-        bitmap_set_bit(write_set, tmp_vfield->field_index);
-        mark_virtual_col(tmp_vfield);
-        bitmap_updated= true;
-      }
-      else
-      {
-        MY_BITMAP *save_read_set= read_set, *save_vcol_set= vcol_set;
-        Item *vcol_item= tmp_vfield->vcol_info->expr;
-        DBUG_ASSERT(vcol_item);
-        bitmap_clear_all(&tmp_set);
-        read_set= vcol_set= &tmp_set;
-        vcol_item->walk(&Item::register_field_in_read_map, 1, 0);
-        read_set= save_read_set;
-        vcol_set= save_vcol_set;
-        if (bitmap_is_overlapping(&tmp_set, write_set))
-        {
-          bitmap_set_bit(write_set, tmp_vfield->field_index);
-          bitmap_set_bit(vcol_set, tmp_vfield->field_index);
-          bitmap_union(read_set, &tmp_set);
-          bitmap_union(vcol_set, &tmp_set);
-          bitmap_updated= true;
-        }
-      }
+      bitmap_set_bit(write_set, tmp_vfield->field_index);
+      mark_virtual_col(tmp_vfield);
+      bitmap_updated= true;
     }
   }
   if (bitmap_updated)
@@ -7262,8 +7248,8 @@ bool TABLE_LIST::process_index_hints(TABLE *tbl)
     }
 
     /*
-      TODO: get rid of tbl->force_index (on if any FORCE INDEX is specified) and
-      create tbl->force_index_join instead.
+      TODO: get rid of tbl->force_index (on if any FORCE INDEX is specified)
+      and create tbl->force_index_join instead.
       Then use the correct force_index_XX instead of the global one.
     */
     if (!index_join[INDEX_HINT_FORCE].is_clear_all() ||
@@ -7299,15 +7285,21 @@ size_t max_row_length(TABLE *table, const uchar *data)
   size_t length= table_s->reclength + 2 * table_s->fields;
   uint *const beg= table_s->blob_field;
   uint *const end= beg + table_s->blob_fields;
+  my_ptrdiff_t const rec_offset= (my_ptrdiff_t) (data - table->record[0]);
+  DBUG_ENTER("max_row_length");
 
   for (uint *ptr= beg ; ptr != end ; ++ptr)
   {
-    Field_blob* const blob= (Field_blob*) table->field[*ptr];
-    length+= blob->get_length((const uchar*)
-                              (data + blob->offset(table->record[0]))) +
-      HA_KEY_BLOB_LENGTH;
+    Field * const field= table->field[*ptr];
+    if (bitmap_is_set(table->read_set, field->field_index) &&
+        !field->is_null(rec_offset))
+    {
+      Field_blob * const blob= (Field_blob*) field;
+      length+= blob->get_length(rec_offset) + HA_KEY_BLOB_LENGTH;
+    }
   }
-  return length;
+  DBUG_PRINT("exit", ("length: %lld", (longlong) length));
+  DBUG_RETURN(length);
 }
 
 
@@ -7422,7 +7414,7 @@ int TABLE::update_virtual_fields(handler *h, enum_vcol_update_mode update_mode)
   Query_arena backup_arena;
   Turn_errors_to_warnings_handler Suppress_errors;
   int error;
-  bool handler_pushed= 0;
+  bool handler_pushed= 0, update_all_columns= 1;
   DBUG_ASSERT(vfield);
 
   if (h->keyread_enabled())
@@ -7439,6 +7431,16 @@ int TABLE::update_virtual_fields(handler *h, enum_vcol_update_mode update_mode)
     in_use->push_internal_handler(&Suppress_errors);
     handler_pushed= 1;
   }
+  else if (update_mode == VCOL_UPDATE_FOR_REPLACE &&
+           in_use->is_current_stmt_binlog_format_row() &&
+           in_use->variables.binlog_row_image != BINLOG_ROW_IMAGE_MINIMAL)
+  {
+    /*
+      If we are doing a replace with not minimal binary logging, we have to
+      calculate all virtual columns.
+    */
+    update_all_columns= 1;
+  }
 
   /* Iterate over virtual fields in the table */
   for (vfield_ptr= vfield; *vfield_ptr; vfield_ptr++)
@@ -7451,8 +7453,8 @@ int TABLE::update_virtual_fields(handler *h, enum_vcol_update_mode update_mode)
     bool update= 0, swap_values= 0;
     switch (update_mode) {
     case VCOL_UPDATE_FOR_READ:
-      update= !vcol_info->stored_in_db
-           && bitmap_is_set(vcol_set, vf->field_index);
+      update= (!vcol_info->stored_in_db &&
+               bitmap_is_set(vcol_set, vf->field_index));
       swap_values= 1;
       break;
     case VCOL_UPDATE_FOR_DELETE:
@@ -7460,8 +7462,9 @@ int TABLE::update_virtual_fields(handler *h, enum_vcol_update_mode update_mode)
       update= bitmap_is_set(vcol_set, vf->field_index);
       break;
     case VCOL_UPDATE_FOR_REPLACE:
-      update= !vcol_info->stored_in_db && (vf->flags & PART_KEY_FLAG)
-           && bitmap_is_set(vcol_set, vf->field_index);
+      update= ((!vcol_info->stored_in_db && (vf->flags & PART_KEY_FLAG) &&
+                bitmap_is_set(vcol_set, vf->field_index)) ||
+               update_all_columns);
       if (update && (vf->flags & BLOB_FLAG))
       {
         /*
@@ -7478,8 +7481,8 @@ int TABLE::update_virtual_fields(handler *h, enum_vcol_update_mode update_mode)
     case VCOL_UPDATE_INDEXED:
     case VCOL_UPDATE_INDEXED_FOR_UPDATE:
       /* Read indexed fields that was not updated in VCOL_UPDATE_FOR_READ */
-      update= !vcol_info->stored_in_db && (vf->flags & PART_KEY_FLAG) &&
-               bitmap_is_set(vcol_set, vf->field_index);
+      update= (!vcol_info->stored_in_db && (vf->flags & PART_KEY_FLAG) &&
+               !bitmap_is_set(vcol_set, vf->field_index));
       swap_values= 1;
       break;
     }