Bug #24754897 RPL_INJECTOR API FUNCTION FOR UPDATE_ROW WITH BEFORE AND AFTER COLUMN BITMAPS

Priyanka Sangam · Priyanka Sangam · commit 061d482738f0 · 2017-04-11T14:34:43.000+05:30
An UPDATE_ROW binlog event contains 2 versions of the updated
row: a Before Image which has the row before the update,
and an After Image of the row after the update. The Before
and After Image do not always include all the columns in the
row. The binlog format decides which columns will be included.

There is a requirement in ndbcluster to reduce the size of the
binlog event while retaining the UPDATE_ROW format of
BeforeImage/AfterImage. This can be done by writing only
primary key columns from the Before Image and only modified
columns from the After Image. However, the existing binlog
formats do not support different sets of columns in the Before
and After Images.

The --binlog-format=MINIMAL option is not usable for ndbcluster
because enabling the binlog format code breaks the
--ndb-log-updated-only option in ndb replication.

This patch implements 2 new ndb binlog formats for UPDATE_ROW
where the Before Image and After Image are minimised to remove
duplicate information.

UPDATED_ONLY_USE_UPDATE_MINIMAL: log update as UPDATE_ROW, log
only primary key columns in Before Image and only updated columns
in After Image

FULL_USE_UPDATE_MINIMAL: log update as UPDATE_ROW, log only
primary key columns in Before Image and all non primary key
columns in After Image

This means that each column is contained only once in each event,
bringing efficiency close to that achieved by the variants logging
with WRITE_ROW events.

The UPDATED_ONLY_USE_UPDATE_MINIMAL and FULL_USE_UPDATE_MINIMAL
formats can be used with the conflict resolution algorithms
NDB$EPOCH(), NDB$EPOCH2() and NDB$EPOCH_TRANS(), because these
do not require the 'Before' values of non-primary-key-columns.
Conflict resolution algorithms like NDB$MAX() and NDB$OLD()
require the old values of non-PK columns, and will not work
correctly with these new variants.
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
@@ -1,4 +1,4 @@
-/* Copyright (c) 2004, 2016, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2004, 2017, Oracle and/or its affiliates. All rights reserved.
 
   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
@@ -19038,13 +19038,27 @@ static MYSQL_SYSVAR_BOOL(
   opt_ndb_log_update_as_write,       /* var */
   PLUGIN_VAR_OPCMDARG,
   "For efficiency log only after image as a write event. "
-  "Ignore before image. This may cause compatability problems if "
+  "Ignore before image. This may cause compatibility problems if "
   "replicating to other storage engines than ndbcluster.",
   NULL,                              /* check func. */
   NULL,                              /* update func. */
   1                                  /* default */
 );
 
+my_bool opt_ndb_log_update_minimal;
+static MYSQL_SYSVAR_BOOL(
+  log_update_minimal,                  /* name */
+  opt_ndb_log_update_minimal,          /* var */
+  PLUGIN_VAR_OPCMDARG,
+  "For efficiency, log updates in a minimal format"
+  "Log only the primary key value(s) in the before "
+  "image. Log only the changed columns in the after "
+  "image. This may cause compatibility problems if "
+  "replicating to other storage engines than ndbcluster.",
+  NULL,                              /* check func. */
+  NULL,                              /* update func. */
+  0                                  /* default */
+);
 
 my_bool opt_ndb_log_updated_only;
 static MYSQL_SYSVAR_BOOL(
@@ -19053,7 +19067,7 @@ static MYSQL_SYSVAR_BOOL(
   PLUGIN_VAR_OPCMDARG,
   "For efficiency log only updated columns. Columns are considered "
   "as \"updated\" even if they are updated with the same value. "
-  "This may cause compatability problems if "
+  "This may cause compatibility problems if "
   "replicating to other storage engines than ndbcluster.",
   NULL,                              /* check func. */
   NULL,                              /* update func. */
@@ -19375,6 +19389,7 @@ static struct st_mysql_sys_var* system_variables[]= {
   MYSQL_SYSVAR(eventbuffer_free_percent),
   MYSQL_SYSVAR(log_update_as_write),
   MYSQL_SYSVAR(log_updated_only),
+  MYSQL_SYSVAR(log_update_minimal),
   MYSQL_SYSVAR(log_empty_update),
   MYSQL_SYSVAR(log_orig),
   MYSQL_SYSVAR(distribution),
diff --git a/sql/ha_ndbcluster_binlog.cc b/sql/ha_ndbcluster_binlog.cc
@@ -1,5 +1,5 @@
 /*
-   Copyright (c) 2000, 2015, Oracle and/or its affiliates. All rights reserved.
+   Copyright (c) 2000, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -44,6 +44,7 @@ extern my_bool opt_ndb_log_orig;
 extern my_bool opt_ndb_log_bin;
 extern my_bool opt_ndb_log_update_as_write;
 extern my_bool opt_ndb_log_updated_only;
+extern my_bool opt_ndb_log_update_minimal;
 extern my_bool opt_ndb_log_binlog_index;
 extern my_bool opt_ndb_log_apply_status;
 extern ulong opt_ndb_extra_logging;
@@ -356,6 +357,8 @@ ndb_binlog_open_shadow_table(THD *thd, NDB_SHARE *share)
   if (shadow_table->s->blob_fields != 0)
     share->flags|= NSF_BLOB_FLAG;
 
+  event_data->init_pk_bitmap();
+
 #ifndef DBUG_OFF
   dbug_print_table("table", shadow_table);
 #endif
@@ -4283,6 +4286,10 @@ set_binlog_flags(NDB_SHARE *share,
     {
       set_binlog_use_update(share);
     }
+    if (opt_ndb_log_update_minimal)
+    {
+      set_binlog_update_minimal(share);
+    }
     break;
   case NBT_UPDATED_ONLY:
     DBUG_PRINT("info", ("NBT_UPDATED_ONLY"));
@@ -4306,6 +4313,20 @@ set_binlog_flags(NDB_SHARE *share,
     set_binlog_full(share);
     set_binlog_use_update(share);
     break;
+  case NBT_UPDATED_ONLY_MINIMAL:
+    DBUG_PRINT("info", ("NBT_UPDATED_ONLY_MINIMAL"));
+    set_binlog_updated_only(share);
+    set_binlog_use_update(share);
+    set_binlog_update_minimal(share);
+    break;
+  case NBT_UPDATED_FULL_MINIMAL:
+    DBUG_PRINT("info", ("NBT_UPDATED_FULL_MINIMAL"));
+    set_binlog_full(share);
+    set_binlog_use_update(share);
+    set_binlog_update_minimal(share);
+    break;
+  default:
+    DBUG_VOID_RETURN;
   }
   set_binlog_logging(share);
   DBUG_VOID_RETURN;
@@ -5806,10 +5827,10 @@ handle_data_event(THD* thd, Ndb *ndb, NdbEventOperation *pOp,
   DBUG_PRINT("info", ("Assuming %u columns for table %s",
                       n_fields, table->s->table_name.str));
   MY_BITMAP b;
-  /* Potential buffer for the bitmap */
-  uint32 bitbuf[128 / (sizeof(uint32) * 8)];
-  const bool own_buffer = n_fields <= sizeof(bitbuf) * 8;
-  bitmap_init(&b, own_buffer ? bitbuf : NULL, n_fields, FALSE); 
+  my_bitmap_map bitbuf[(NDB_MAX_ATTRIBUTES_IN_TABLE +
+                            8*sizeof(my_bitmap_map) - 1) /
+                           (8*sizeof(my_bitmap_map))];
+  bitmap_init(&b, bitbuf, n_fields, FALSE);
   bitmap_set_all(&b);
 
   /*
@@ -5954,9 +5975,24 @@ handle_data_event(THD* thd, Ndb *ndb, NdbEventOperation *pOp,
         }
         ndb_unpack_record(table, event_data->ndb_value[1], &b, table->record[1]);
         DBUG_EXECUTE("info", print_records(table, table->record[1]););
+
+        MY_BITMAP col_bitmap_before_update;
+        my_bitmap_map bitbuf[(NDB_MAX_ATTRIBUTES_IN_TABLE +
+                                  8*sizeof(my_bitmap_map) - 1) /
+                                 (8*sizeof(my_bitmap_map))];
+        bitmap_init(&col_bitmap_before_update, bitbuf, n_fields, FALSE);
+        if (get_binlog_update_minimal(share))
+        {
+          event_data->generate_minimal_bitmap(&col_bitmap_before_update, &b);
+        }
+        else
+        {
+          bitmap_copy(&col_bitmap_before_update, &b);
+        }
+
         ret = trans.update_row(logged_server_id,
                                injector::transaction::table(table, true),
-                               &b, n_fields,
+                               &col_bitmap_before_update, &b, n_fields,
                                table->record[1], // before values
                                table->record[0], // after values
                                extra_row_info_ptr);
@@ -5976,11 +6012,6 @@ handle_data_event(THD* thd, Ndb *ndb, NdbEventOperation *pOp,
     my_free(blobs_buffer[1], MYF(MY_ALLOW_ZERO_PTR));
   }
 
-  if (!own_buffer)
-  {
-    bitmap_free(&b);
-  }
-
   return 0;
 }
 
diff --git a/sql/ndb_event_data.cc b/sql/ndb_event_data.cc
@@ -1,5 +1,5 @@
 /*
-   Copyright (c) 2011, 2015, Oracle and/or its affiliates. All rights reserved.
+   Copyright (c) 2011, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -22,7 +22,8 @@
 
 Ndb_event_data::Ndb_event_data(NDB_SHARE *the_share) :
   shadow_table(NULL),
-  share(the_share)
+  share(the_share),
+  pk_bitmap(NULL)
 {
   ndb_value[0]= NULL;
   ndb_value[1]= NULL;
@@ -34,6 +35,10 @@ Ndb_event_data::~Ndb_event_data()
   if (shadow_table)
     closefrm(shadow_table, 1);
   shadow_table= NULL;
+
+  delete pk_bitmap;
+  pk_bitmap = NULL;
+
   free_root(&mem_root, MYF(0));
   share= NULL;
   /*
@@ -84,3 +89,66 @@ void Ndb_event_data::print(const char* where, FILE* file) const
     fprintf(file, "  - mem_root used: %lu\n", mem_root_used);
   }
 }
+
+/*
+ * While writing an UPDATE_ROW event to the binlog, a bitmap is
+ * used to indicate which columns should be written. An
+ * UPDATE_ROW event contains 2 versions of the row: a Before Image
+ * of the row before the update was done, and an After Image of
+ * the row after the update. Column bitmaps are used to decide
+ * which columns will be written to both images. The Before
+ * Image and After Image can contain different columns.
+ *
+ * For the binlog formats UPDATED_ONLY_USE_UPDATE_MINIMAL and
+ * FULL_USE_UPDATE_MINIMAL, it is necessary to write only primary
+ * key columns to the Before Image, and to remove all primary key
+ * columns from the After Image. A bitmap of primary key columns is
+ * created for this purpose.
+ */
+void Ndb_event_data::init_pk_bitmap()
+{
+  if (shadow_table->s->primary_key == MAX_KEY)
+  {
+    // Table without pk, no need for pk_bitmap since minimal is full
+    return;
+  }
+  pk_bitmap = new MY_BITMAP();
+  bitmap_init(pk_bitmap, pk_bitbuf, shadow_table->s->fields, FALSE);
+  KEY* key = shadow_table->key_info + shadow_table->s->primary_key;
+  KEY_PART_INFO* key_part_info = key->key_part;
+  const uint key_parts = key->user_defined_key_parts;
+  for (uint i = 0; i < key_parts; i++, key_part_info++)
+  {
+    bitmap_set_bit(pk_bitmap, key_part_info->fieldnr - 1);
+  }
+  assert(!bitmap_is_clear_all(pk_bitmap));
+}
+
+/*
+ * Modify the column bitmaps generated for UPDATE_ROW as per
+ * the MINIMAL binlog format type. Expected arguments:
+ *
+ * @before: empty bitmap to be populated with PK columns
+ * @after: bitmap with updated cols, if ndb_log_updated_only=TRUE
+ *         bitmap with all cols, if ndb_log_updated_only=FALSE
+ *
+ * If no PK is defined, bitmaps revert to default behaviour:
+ *  - before and after bitmaps are identical
+ *  - bitmaps contain all/updated cols as per ndb_log_updated_only
+ */
+void Ndb_event_data::generate_minimal_bitmap(MY_BITMAP *before, MY_BITMAP *after)
+{
+  if (pk_bitmap)
+  {
+    assert(!bitmap_is_clear_all(pk_bitmap));
+    // set Before Image to contain only primary keys
+    bitmap_copy(before, pk_bitmap);
+    // remove primary keys from After Image
+    bitmap_subtract(after, pk_bitmap);
+  }
+  else
+  {
+    // no usable PK bitmap, set Before Image = After Image
+    bitmap_copy(before, after);
+  }
+}
diff --git a/sql/ndb_event_data.h b/sql/ndb_event_data.h
@@ -1,5 +1,5 @@
 /*
-   Copyright (c) 2011, Oracle and/or its affiliates. All rights reserved.
+   Copyright (c) 2011, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -20,6 +20,9 @@
 
 #include <my_global.h> // my_alloc.h
 #include <my_alloc.h> // MEM_ROOT
+#include <my_bitmap.h>
+
+#include <ndbapi/ndbapi_limits.h>
 
 class Ndb_event_data
 {
@@ -34,8 +37,15 @@ class Ndb_event_data
   struct TABLE *shadow_table;
   struct NDB_SHARE *share;
   union NdbValue *ndb_value[2];
+  /* Bitmap with bit set for all primary key columns. */
+  MY_BITMAP *pk_bitmap;
+  my_bitmap_map pk_bitbuf[(NDB_MAX_ATTRIBUTES_IN_TABLE +
+                            8*sizeof(my_bitmap_map) - 1) /
+                           (8*sizeof(my_bitmap_map))];
 
   void print(const char* where, FILE* file) const;
+  void init_pk_bitmap();
+  void generate_minimal_bitmap(MY_BITMAP *before, MY_BITMAP *after);
 };
 
 #endif
diff --git a/sql/ndb_share.h b/sql/ndb_share.h
@@ -1,5 +1,5 @@
 /*
-   Copyright (c) 2011, 2015, Oracle and/or its affiliates. All rights reserved.
+   Copyright (c) 2011, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -38,9 +38,11 @@ enum Ndb_binlog_type
   ,NBT_NO_LOGGING               = 1
   ,NBT_UPDATED_ONLY             = 2
   ,NBT_FULL                     = 3
-  ,NBT_USE_UPDATE               = 4 /* bit 0x4 indicates USE_UPDATE */
-  ,NBT_UPDATED_ONLY_USE_UPDATE  = NBT_UPDATED_ONLY | NBT_USE_UPDATE
-  ,NBT_FULL_USE_UPDATE          = NBT_FULL         | NBT_USE_UPDATE
+  ,NBT_USE_UPDATE               = 4
+  ,NBT_UPDATED_ONLY_USE_UPDATE  = 6
+  ,NBT_FULL_USE_UPDATE          = 7
+  ,NBT_UPDATED_ONLY_MINIMAL     = 8
+  ,NBT_UPDATED_FULL_MINIMAL     = 9
 };
 #endif
 
@@ -138,6 +140,8 @@ set_ndb_share_state(NDB_SHARE *share, NDB_SHARE_STATE state)
 #define NSF_BINLOG_FULL 8u /* table should be binlogged with full rows */
 #define NSF_BINLOG_USE_UPDATE 16u  /* table update should be binlogged using
                                      update log event */
+#define NSF_BINLOG_MINIMAL_UPDATE 32u  /* table update should be binlogged using
+                              minimal format: before(PK):after(changed cols) */
 inline void set_binlog_logging(NDB_SHARE *share)
 {
   DBUG_PRINT("info", ("set_binlog_logging"));
@@ -175,6 +179,16 @@ inline void set_binlog_use_update(NDB_SHARE *share)
 inline my_bool get_binlog_use_update(NDB_SHARE *share)
 { return (share->flags & NSF_BINLOG_USE_UPDATE) != 0; }
 
+static inline void set_binlog_update_minimal(NDB_SHARE *share)
+{
+  DBUG_PRINT("info", ("set_binlog_update_minimal"));
+  share->flags|= NSF_BINLOG_MINIMAL_UPDATE;
+}
+
+static inline bool get_binlog_update_minimal(const NDB_SHARE *share)
+{
+  return (share->flags & NSF_BINLOG_MINIMAL_UPDATE) != 0;
+}
 
 NDB_SHARE *ndbcluster_get_share(const char *key,
                                 struct TABLE *table,
diff --git a/sql/rpl_injector.cc b/sql/rpl_injector.cc
@@ -1,4 +1,4 @@
-/* Copyright (c) 2006, 2015, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2006, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -229,10 +229,11 @@ int injector::transaction::delete_row(server_id_type sid, table tbl,
   return delete_row(sid, tbl, cols, colcnt, record, NULL);
 }
 
-
-int injector::transaction::update_row(server_id_type sid, table tbl, 
-				      MY_BITMAP const* cols, size_t colcnt,
-				      record_type before, record_type after,
+int injector::transaction::update_row(server_id_type sid, table tbl,
+                                      MY_BITMAP const* before_cols,
+                                      MY_BITMAP const* after_cols,
+                                      size_t colcnt,
+                                      record_type before, record_type after,
                                       const uchar* extra_row_info)
 {
    DBUG_ENTER("injector::transaction::update_row(...)");
@@ -244,7 +245,7 @@ int injector::transaction::update_row(server_id_type sid, table tbl,
    server_id_type save_id= m_thd->server_id;
    m_thd->set_server_id(sid);
    // The read- and write sets with autorestore (in the destructor)
-   table::save_sets saveset(tbl, cols, cols);
+   table::save_sets saveset(tbl, before_cols, after_cols);
 
    error= m_thd->binlog_update_row(tbl.get_table(), tbl.is_transactional(), 
                                    before, after, extra_row_info);
@@ -253,10 +254,10 @@ int injector::transaction::update_row(server_id_type sid, table tbl,
 }
 
 int injector::transaction::update_row(server_id_type sid, table tbl,
-				      MY_BITMAP const* cols, size_t colcnt,
-				      record_type before, record_type after)
+                                      MY_BITMAP const* cols, size_t colcnt,
+                                      record_type before, record_type after)
 {
-  return update_row(sid, tbl, cols, colcnt, before, after, NULL);
+  return update_row(sid, tbl, cols, cols, colcnt, before, after, NULL);
 }
 
 injector::transaction::binlog_pos injector::transaction::start_pos() const
diff --git a/sql/rpl_injector.h b/sql/rpl_injector.h
@@ -1,4 +1,4 @@
-/* Copyright (c) 2006, 2015, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2006, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -224,7 +224,9 @@ class injector
         Add an 'update row' entry to the transaction.
       */
       int update_row(server_id_type sid, table tbl, 
-                     MY_BITMAP const *cols, size_t colcnt,
+                     MY_BITMAP const *before_cols,
+                     MY_BITMAP const *after_cols,
+                     size_t colcnt,
                      record_type before, record_type after,
                      const uchar* extra_row_info);
       int update_row(server_id_type sid, table tbl,
