table->write_set was changed if binary logging was used, which caused the

changes in query execution plans.
Fixed by introducing table->rpl_write_set which holds which columns should
be stored in the binary log.

Other things:
- Removed some not needed references to read_set and write_set to make
  code really changing read_set and write_set easier to read
  (in opt_range.cc)
- Added error handling of failed unpack_current_row()
- Added missing call to mark_columns_needed_for_insert() for DELAYED INSERT
- Removed not used functions in_read_set() and in_write_set()
- In rpl_record.cc, removed not used variable error

sql/field.h

@@ -3093,8 +3093,6 @@ public:
   uint32 max_display_length();
   uint32 char_length();
   uint is_equal(Create_field *new_field);
-  inline bool in_read_set() { return bitmap_is_set(table->read_set, field_index); }
-  inline bool in_write_set() { return bitmap_is_set(table->write_set, field_index); }
 private:
   int do_save_field_metadata(uchar *first_byte);
 };

sql/log_event.cc

@@ -9851,6 +9851,7 @@ int Rows_log_event::do_apply_event(rpl_group_info *rgi)
       bitmap_intersect(table->read_set,&m_cols);
 
     bitmap_set_all(table->write_set);
+    table->rpl_write_set= table->write_set;
 
     /* WRITE ROWS EVENTS store the bitmap in m_cols instead of m_cols_ai */
     MY_BITMAP *after_image= ((get_general_type_code() == UPDATE_ROWS_EVENT) ?
@@ -11109,8 +11110,8 @@ void Table_map_log_event::print(FILE *, PRINT_EVENT_INFO *print_event_info)
 Write_rows_log_event::Write_rows_log_event(THD *thd_arg, TABLE *tbl_arg,
                                            ulong tid_arg,
                                            bool is_transactional)
-  : Rows_log_event(thd_arg, tbl_arg, tid_arg, tbl_arg->write_set,
+  :Rows_log_event(thd_arg, tbl_arg, tid_arg, tbl_arg->rpl_write_set,
-                   is_transactional, WRITE_ROWS_EVENT_V1)
+                  is_transactional, WRITE_ROWS_EVENT_V1)
 {
 }
 #endif
@@ -11218,8 +11219,10 @@ Write_rows_log_event::do_after_row_operations(const Slave_reporting_capability *
    */
   if (is_auto_inc_in_extra_columns())
   {
-    bitmap_clear_bit(m_table->write_set, m_table->next_number_field->field_index);
+    bitmap_clear_bit(m_table->rpl_write_set,
-    bitmap_clear_bit( m_table->read_set, m_table->next_number_field->field_index);
+                     m_table->next_number_field->field_index);
+    bitmap_clear_bit(m_table->read_set,
+                     m_table->next_number_field->field_index);
 
     if (get_flags(STMT_END_F))
       m_table->file->ha_release_auto_increment();
@@ -11388,8 +11391,8 @@ Rows_log_event::write_row(rpl_group_info *rgi,
     m_table->next_number_field->set_null();
   
   DBUG_DUMP("record[0]", table->record[0], table->s->reclength);
-  DBUG_PRINT_BITSET("debug", "write_set = %s", table->write_set);
+  DBUG_PRINT_BITSET("debug", "rpl_write_set: %s", table->rpl_write_set);
-  DBUG_PRINT_BITSET("debug", "read_set = %s", table->read_set);
+  DBUG_PRINT_BITSET("debug", "read_set:      %s", table->read_set);
 
   if (invoke_triggers &&
       process_triggers(TRG_EVENT_INSERT, TRG_ACTION_BEFORE, TRUE))
@@ -12259,7 +12262,7 @@ Update_rows_log_event::Update_rows_log_event(THD *thd_arg, TABLE *tbl_arg,
 : Rows_log_event(thd_arg, tbl_arg, tid, tbl_arg->read_set, is_transactional,
                  UPDATE_ROWS_EVENT_V1)
 {
-  init(tbl_arg->write_set);
+  init(tbl_arg->rpl_write_set);
 }
 
 void Update_rows_log_event::init(MY_BITMAP const *cols)

sql/log_event.h

@@ -4304,7 +4304,7 @@ public:
                  bitmap_cmp return logic).
 
    */
-  virtual bool read_write_bitmaps_cmp(TABLE *table)
+  bool read_write_bitmaps_cmp(TABLE *table)
   {
     bool res= FALSE;
 
@@ -4315,10 +4315,10 @@ public:
         break;
       case UPDATE_ROWS_EVENT:
         res= (bitmap_cmp(get_cols(), table->read_set) &&
-              bitmap_cmp(get_cols_ai(), table->write_set));
+              bitmap_cmp(get_cols_ai(), table->rpl_write_set));
         break;
       case WRITE_ROWS_EVENT:
-        res= bitmap_cmp(get_cols(), table->write_set);
+        res= bitmap_cmp(get_cols(), table->rpl_write_set);
         break;
       default:
         /*

sql/log_event_old.cc

@@ -1522,6 +1522,7 @@ int Old_rows_log_event::do_apply_event(rpl_group_info *rgi)
     bitmap_set_all(table->write_set);
     if (!get_flags(COMPLETE_ROWS_F))
       bitmap_intersect(table->write_set,&m_cols);
+    table->rpl_write_set= table->write_set;
 
     // Do event specific preparations 
     
@@ -1896,8 +1897,9 @@ Old_rows_log_event::write_row(rpl_group_info *rgi, const bool overwrite)
     DBUG_RETURN(error);
   
   /* unpack row into table->record[0] */
-  error= unpack_current_row(rgi); // TODO: how to handle errors?
+  if ((error= unpack_current_row(rgi)))
-
+    DBUG_RETURN(error);
+  
 #ifndef DBUG_OFF
   DBUG_DUMP("record[0]", table->record[0], table->s->reclength);
   DBUG_PRINT_BITSET("debug", "write_set = %s", table->write_set);

sql/opt_range.cc

@@ -1519,15 +1519,14 @@ end:
   DBUG_ASSERT(head->read_set == &column_bitmap);
   /*
     We are only going to read key fields and call position() on 'file'
-    The following sets head->tmp_set to only use this key and then updates
+    The following sets head->read_set (== column_bitmap) to only use this
-    head->read_set and head->write_set to use this bitmap.
+    key. The 'column_bitmap' is used in ::get_next()
-    The now bitmap is stored in 'column_bitmap' which is used in ::get_next()
   */
   org_file= head->file;
   org_key_read= head->key_read;
   head->file= file;
   head->key_read= 0;
-  head->mark_columns_used_by_index_no_reset(index, head->read_set);
+  head->mark_columns_used_by_index_no_reset(index, &column_bitmap);
 
   if (!head->no_keyread)
   {
@@ -1551,8 +1550,7 @@ end:
       file->ha_close();
       goto failure;
     }
-    else
+    DBUG_RETURN(1);
-      DBUG_RETURN(1);
   }
   DBUG_RETURN(0);
 
@@ -11141,26 +11139,21 @@ err:
 int QUICK_RANGE_SELECT::get_next()
 {
   range_id_t dummy;
+  int result;
+  DBUG_ENTER("QUICK_RANGE_SELECT::get_next");
+
+  if (!in_ror_merged_scan)
+    DBUG_RETURN(file->multi_range_read_next(&dummy));
+
   MY_BITMAP * const save_read_set= head->read_set;
   MY_BITMAP * const save_write_set= head->write_set;
-
+  /*
-  DBUG_ENTER("QUICK_RANGE_SELECT::get_next");
+    We don't need to signal the bitmap change as the bitmap is always the
-  if (in_ror_merged_scan)
+    same for this head->file
-  {
+  */
-    /*
+  head->column_bitmaps_set_no_signal(&column_bitmap, &column_bitmap);
-      We don't need to signal the bitmap change as the bitmap is always the
+  result= file->multi_range_read_next(&dummy);
-      same for this head->file
+  head->column_bitmaps_set_no_signal(save_read_set, save_write_set);
-    */
-    head->column_bitmaps_set_no_signal(&column_bitmap, &column_bitmap);
-  }
-
-  int result= file->multi_range_read_next(&dummy);
-
-  if (in_ror_merged_scan)
-  {
-    /* Restore bitmaps set on entry */
-    head->column_bitmaps_set_no_signal(save_read_set, save_write_set);
-  }
   DBUG_RETURN(result);
 }
 

sql/rpl_record.cc

@@ -200,7 +200,6 @@ unpack_row(rpl_group_info *rgi,
   DBUG_ASSERT(row_data);
   DBUG_ASSERT(table);
   size_t const master_null_byte_count= (bitmap_bits_set(cols) + 7) / 8;
-  int error= 0;
 
   uchar const *null_ptr= row_data;
   uchar const *pack_ptr= row_data + master_null_byte_count;
@@ -217,7 +216,7 @@ unpack_row(rpl_group_info *rgi,
      */
     *current_row_end= pack_ptr;
     *master_reclength= 0;
-    DBUG_RETURN(error);
+    DBUG_RETURN(0);
   }
   DBUG_ASSERT(null_ptr < row_data + master_null_byte_count);
 
@@ -434,7 +433,7 @@ unpack_row(rpl_group_info *rgi,
       *master_reclength = table->s->reclength;
   }
   
-  DBUG_RETURN(error);
+  DBUG_RETURN(0);
 }
 
 /**

sql/sql_class.cc

@@ -6173,7 +6173,7 @@ int THD::binlog_write_row(TABLE* table, bool is_trans,
 
   uchar *row_data= memory.slot(0);
 
-  size_t const len= pack_row(table, table->write_set, row_data, record);
+  size_t const len= pack_row(table, table->rpl_write_set, row_data, record);
 
   /* Ensure that all events in a GTID group are in the same cache */
   if (variables.option_bits & OPTION_GTID_BEGIN)
@@ -6197,13 +6197,6 @@ int THD::binlog_update_row(TABLE* table, bool is_trans,
   DBUG_ASSERT(is_current_stmt_binlog_format_row() &&
             ((WSREP(this) && wsrep_emulate_bin_log) || mysql_bin_log.is_open()));
 
-  /**
-    Save a reference to the original read and write set bitmaps.
-    We will need this to restore the bitmaps at the end.
-   */
-  MY_BITMAP *old_read_set= table->read_set;
-  MY_BITMAP *old_write_set= table->write_set;
-
   size_t const before_maxlen = max_row_length(table, before_record);
   size_t const after_maxlen  = max_row_length(table, after_record);
 
@@ -6216,7 +6209,7 @@ int THD::binlog_update_row(TABLE* table, bool is_trans,
 
   size_t const before_size= pack_row(table, table->read_set, before_row,
                                         before_record);
-  size_t const after_size= pack_row(table, table->write_set, after_row,
+  size_t const after_size= pack_row(table, table->rpl_write_set, after_row,
                                        after_record);
 
   /* Ensure that all events in a GTID group are in the same cache */
@@ -6245,10 +6238,6 @@ int THD::binlog_update_row(TABLE* table, bool is_trans,
   int error=  ev->add_row_data(before_row, before_size) ||
               ev->add_row_data(after_row, after_size);
 
-  /* restore read/write set for the rest of execution */
-  table->column_bitmaps_set_no_signal(old_read_set,
-                                      old_write_set);
-
   return error;
 
 }
@@ -6259,11 +6248,13 @@ int THD::binlog_delete_row(TABLE* table, bool is_trans,
   DBUG_ASSERT(is_current_stmt_binlog_format_row() &&
             ((WSREP(this) && wsrep_emulate_bin_log) || mysql_bin_log.is_open()));
   /**
-    Save a reference to the original read and write set bitmaps.
+    Save a reference to the original read bitmaps
-    We will need this to restore the bitmaps at the end.
+    We will need this to restore the bitmaps at the end as
-   */
+    binlog_prepare_row_images() may change table->read_set.
+    table->read_set is used by pack_row and deep in
+    binlog_prepare_pending_events().
+  */
   MY_BITMAP *old_read_set= table->read_set;
-  MY_BITMAP *old_write_set= table->write_set;
 
   /** 
      This will remove spurious fields required during execution but
@@ -6300,9 +6291,9 @@ int THD::binlog_delete_row(TABLE* table, bool is_trans,
 
   int error= ev->add_row_data(row_data, len);
 
-  /* restore read/write set for the rest of execution */
+  /* restore read set for the rest of execution */
   table->column_bitmaps_set_no_signal(old_read_set,
-                                      old_write_set);
+                                      table->write_set);
 
   return error;
 }

sql/sql_insert.cc

@@ -2867,6 +2867,8 @@ pthread_handler_t handle_delayed_insert(void *arg)
     /* Tell client that the thread is initialized */
     mysql_cond_signal(&di->cond_client);
 
+    di->table->mark_columns_needed_for_insert();
+
     /* Now wait until we get an insert or lock to handle */
     /* We will not abort as long as a client thread uses this thread */
 

sql/table.cc

@@ -2885,20 +2885,28 @@ partititon_err:
   /* Allocate bitmaps */
 
   bitmap_size= share->column_bitmap_size;
-  if (!(bitmaps= (uchar*) alloc_root(&outparam->mem_root, bitmap_size*6)))
+  if (!(bitmaps= (uchar*) alloc_root(&outparam->mem_root, bitmap_size*7)))
     goto err;
   my_bitmap_init(&outparam->def_read_set,
-              (my_bitmap_map*) bitmaps, share->fields, FALSE);
+                 (my_bitmap_map*) bitmaps, share->fields, FALSE);
   my_bitmap_init(&outparam->def_write_set,
-              (my_bitmap_map*) (bitmaps+bitmap_size), share->fields, FALSE);
+                 (my_bitmap_map*) (bitmaps+bitmap_size), share->fields,
+                 FALSE);
   my_bitmap_init(&outparam->def_vcol_set,
-              (my_bitmap_map*) (bitmaps+bitmap_size*2), share->fields, FALSE);
+                 (my_bitmap_map*) (bitmaps+bitmap_size*2), share->fields,
+                 FALSE);
   my_bitmap_init(&outparam->tmp_set,
-              (my_bitmap_map*) (bitmaps+bitmap_size*3), share->fields, FALSE);
+                 (my_bitmap_map*) (bitmaps+bitmap_size*3), share->fields,
+                 FALSE);
   my_bitmap_init(&outparam->eq_join_set,
-              (my_bitmap_map*) (bitmaps+bitmap_size*4), share->fields, FALSE);
+                 (my_bitmap_map*) (bitmaps+bitmap_size*4), share->fields,
+                 FALSE);
   my_bitmap_init(&outparam->cond_set,
-              (my_bitmap_map*) (bitmaps+bitmap_size*5), share->fields, FALSE);
+                 (my_bitmap_map*) (bitmaps+bitmap_size*5), share->fields,
+                 FALSE);
+  my_bitmap_init(&outparam->def_rpl_write_set,
+                 (my_bitmap_map*) (bitmaps+bitmap_size*6), share->fields,
+                 FALSE);
   outparam->default_column_bitmaps();
 
   outparam->cond_selectivity= 1.0;
@@ -5679,6 +5687,7 @@ void TABLE::clear_column_bitmaps()
   */
   bzero((char*) def_read_set.bitmap, s->column_bitmap_size*3);
   column_bitmaps_set(&def_read_set, &def_write_set, &def_vcol_set);
+  rpl_write_set= 0;                             // Safety
 }
 
 
@@ -5959,6 +5968,8 @@ void TABLE::mark_columns_needed_for_insert()
 /*
   Mark columns according the binlog row image option.
 
+  Columns to be written are stored in 'rpl_write_set'
+
   When logging in RBR, the user can select whether to
   log partial or full rows, depending on the table
   definition, and the value of binlog_row_image.
@@ -5970,16 +5981,16 @@ void TABLE::mark_columns_needed_for_insert()
   binlog_row_image= MINIMAL
     - This marks the PKE fields in the read_set
     - This marks all fields where a value was specified
-      in the write_set
+      in the rpl_write_set
 
   binlog_row_image= NOBLOB
     - This marks PKE + all non-blob fields in the read_set
     - This marks all fields where a value was specified
-      and all non-blob fields in the write_set
+      and all non-blob fields in the rpl_write_set
 
   binlog_row_image= FULL
     - all columns in the read_set
-    - all columns in the write_set
+    - all columns in the rpl_write_set
 
   This marking is done without resetting the original
   bitmaps. This means that we will strip extra fields in
@@ -5987,36 +5998,48 @@ void TABLE::mark_columns_needed_for_insert()
   we only want to log a PK and we needed other fields for
   execution).
 */
+
 void TABLE::mark_columns_per_binlog_row_image()
 {
+  THD *thd= in_use;
   DBUG_ENTER("mark_columns_per_binlog_row_image");
   DBUG_ASSERT(read_set->bitmap);
   DBUG_ASSERT(write_set->bitmap);
 
-  THD *thd= current_thd;
+  /* If not using row format */
+  rpl_write_set= write_set;
 
   /**
     If in RBR we may need to mark some extra columns,
     depending on the binlog-row-image command line argument.
    */
   if ((WSREP_EMULATE_BINLOG(thd) || mysql_bin_log.is_open()) &&
-      in_use &&
+      thd->is_current_stmt_binlog_format_row() &&
-      in_use->is_current_stmt_binlog_format_row() &&
       !ha_check_storage_engine_flag(s->db_type(), HTON_NO_BINLOG_ROW_OPT))
   {
     /* if there is no PK, then mark all columns for the BI. */
     if (s->primary_key >= MAX_KEY)
-      bitmap_set_all(read_set);
-
-    switch (thd->variables.binlog_row_image)
     {
+      bitmap_set_all(read_set);
+      rpl_write_set= read_set;
+    }
+    else
+    {
+      switch (thd->variables.binlog_row_image) {
       case BINLOG_ROW_IMAGE_FULL:
-        if (s->primary_key < MAX_KEY)
+        bitmap_set_all(read_set);
-          bitmap_set_all(read_set);
+        /* Set of columns that should be written (all) */
-        bitmap_set_all(write_set);
+        rpl_write_set= read_set;
         break;
       case BINLOG_ROW_IMAGE_NOBLOB:
-        /* for every field that is not set, mark it unless it is a blob */
+        /* Only write changed columns + not blobs */
+        rpl_write_set= &def_rpl_write_set;
+        bitmap_copy(rpl_write_set, write_set);
+
+        /*
+          for every field that is not set, mark it unless it is a blob or
+          part of a primary key
+        */
         for (Field **ptr=field ; *ptr ; ptr++)
         {
           Field *my_field= *ptr;
@@ -6027,24 +6050,30 @@ void TABLE::mark_columns_per_binlog_row_image()
 
             If set in the AI, then the blob is really needed, there is
             nothing we can do about it.
-           */
+          */
-          if ((s->primary_key < MAX_KEY) &&
+          if ((my_field->flags & PRI_KEY_FLAG) ||
-              ((my_field->flags & PRI_KEY_FLAG) ||
+              (my_field->type() != MYSQL_TYPE_BLOB))
-              (my_field->type() != MYSQL_TYPE_BLOB)))
+          {
             bitmap_set_bit(read_set, my_field->field_index);
-
+            bitmap_set_bit(rpl_write_set, my_field->field_index);
-          if (my_field->type() != MYSQL_TYPE_BLOB)
+          }
-            bitmap_set_bit(write_set, my_field->field_index);
         }
         break;
       case BINLOG_ROW_IMAGE_MINIMAL:
-        /* mark the primary key if available in the read_set */
+        /*
-        if (s->primary_key < MAX_KEY)
+          mark the primary key in the read set so that we can find the row
-          mark_columns_used_by_index_no_reset(s->primary_key, read_set);
+          that is updated / deleted.
+          We don't need to mark the primary key in the rpl_write_set as the
+          binary log will include all columns read anyway.
+        */
+        mark_columns_used_by_index_no_reset(s->primary_key, read_set);
+        /* Only write columns that have changed */
+        rpl_write_set= write_set;
         break;
 
       default:
         DBUG_ASSERT(FALSE);
+      }
     }
     file->column_bitmaps_signal();
   }
@@ -7389,4 +7418,3 @@ double KEY::actual_rec_per_key(uint i)
   return (is_statistics_from_stat_tables ?
           read_stats->get_avg_frequency(i) : (double) rec_per_key[i]);
 }
-

sql/table.h

@@ -1065,10 +1065,12 @@ public:
   ORDER		*group;
   String	alias;            	  /* alias or table name */
   uchar		*null_flags;
-  MY_BITMAP     def_read_set, def_write_set, def_vcol_set, tmp_set; 
+  MY_BITMAP     def_read_set, def_write_set, def_vcol_set, tmp_set;
+  MY_BITMAP     def_rpl_write_set;
   MY_BITMAP     eq_join_set;         /* used to mark equi-joined fields */
   MY_BITMAP     cond_set;   /* used to mark fields from sargable conditions*/
-  MY_BITMAP     *read_set, *write_set, *vcol_set; /* Active column sets */
+  /* Active column sets */
+  MY_BITMAP     *read_set, *write_set, *vcol_set, *rpl_write_set;
   /*
    The ID of the query that opened and is using this table. Has different
    meanings depending on the table type.
@@ -1321,6 +1323,7 @@ public:
     read_set= &def_read_set;
     write_set= &def_write_set;
     vcol_set= &def_vcol_set;
+    rpl_write_set= 0;
   }
   /** Should this instance of the table be reopened? */
   inline bool needs_reopen()