WL#7640 Ndb Active Active Delete-Delete handling

Implement mechanisms for extra signalling in the Binlog...

 - New optional conflict flags, transported by v2 Binlog events in
   the Binlog extra info area.
 - Two values currently : Refresh_op and Reflect_op
 - New reserved AnyValue codes, transferred through a cluster from
   NdbApi -> NdbApi event Api.
 - Test to show that the Binlog flag setting works

The new signalling is not yet used.

Author: frazerclement

mysql-test/suite/ndb_binlog/r/ndb_binlog_test_conflict_flags.result

@@ -0,0 +1,44 @@
+use test;
+reset master;
+create table t1 (a int primary key, b int) engine=ndb;
+Low level transport test
+set global debug='+d,ndb_injector_set_event_conflict_flags';
+Two inserts to the same table, with nothing else in the epoch,
+will be grouped together in one binlog event, as the 
+extra row info is the same.
+insert into test.t1 values (1,1), (2,2);
+A separate delete.
+delete from test.t1 where a=2;
+txt
+### Extra row data format: 0, len: 4 :0x0200FAFA
+### Extra row data format: 0, len: 4 :0x0200FAFA
+set global debug='';
+delete from test.t1;
+reset master;
+Specific flag settings test : Reflect
+set global debug='+d,ndb_set_reflect_anyvalue';
+Two inserts to the same table, with nothing else in the epoch,
+will be grouped together in one binlog event, as the 
+extra row info is the same.
+insert into test.t1 values (1,1), (2,2);
+A separate delete.
+delete from test.t1 where a=2;
+txt
+### Extra row data format: 0, len: 4 :0x02000100
+### Extra row data format: 0, len: 4 :0x02000100
+set global debug='';
+delete from test.t1;
+reset master;
+Specific flag settings test : Refresh
+set global debug='+d,ndb_set_refresh_anyvalue';
+Two inserts to the same table, with nothing else in the epoch,
+will be grouped together in one binlog event, as the 
+extra row info is the same.
+insert into test.t1 values (1,1), (2,2);
+A separate delete.
+delete from test.t1 where a=2;
+txt
+### Extra row data format: 0, len: 4 :0x02000200
+### Extra row data format: 0, len: 4 :0x02000200
+set global debug='';
+drop table t1;

mysql-test/suite/ndb_binlog/t/ndb_binlog_test_conflict_flags.test

@@ -0,0 +1,142 @@
+--source include/have_ndb.inc
+--source include/have_log_bin.inc
+--source include/have_debug.inc
+
+use test;
+
+reset master;
+
+create table t1 (a int primary key, b int) engine=ndb;
+
+--echo Low level transport test
+set global debug='+d,ndb_injector_set_event_conflict_flags';
+
+--echo Two inserts to the same table, with nothing else in the epoch,
+--echo will be grouped together in one binlog event, as the 
+--echo extra row info is the same.
+insert into test.t1 values (1,1), (2,2);
+
+# Wait for epoch boundary
+--disable_query_log
+--disable_result_log
+show binlog events;
+--enable_result_log
+--enable_query_log
+
+--echo A separate delete.
+delete from test.t1 where a=2;
+
+# Wait for epoch boundary
+--disable_query_log
+--disable_result_log
+show binlog events;
+--enable_result_log
+--enable_query_log
+
+
+--disable_query_log
+let $MYSQLD_DATADIR= `select @@datadir;`;
+--exec $MYSQL_BINLOG --verbose $MYSQLD_DATADIR/mysqld-bin.000001 > $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql
+
+create table raw_binlog_rows (txt varchar(1000));
+
+--eval load data local infile '$MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql' into table raw_binlog_rows columns terminated by '\n';
+
+select txt from raw_binlog_rows where txt like '### Extra row data %';
+
+
+drop table raw_binlog_rows;
+--enable_query_log
+
+set global debug='';
+delete from test.t1;
+reset master;
+
+--echo Specific flag settings test : Reflect
+
+set global debug='+d,ndb_set_reflect_anyvalue';
+
+--echo Two inserts to the same table, with nothing else in the epoch,
+--echo will be grouped together in one binlog event, as the 
+--echo extra row info is the same.
+insert into test.t1 values (1,1), (2,2);
+
+# Wait for epoch boundary
+--disable_query_log
+--disable_result_log
+show binlog events;
+--enable_result_log
+--enable_query_log
+
+--echo A separate delete.
+delete from test.t1 where a=2;
+
+# Wait for epoch boundary
+--disable_query_log
+--disable_result_log
+show binlog events;
+--enable_result_log
+--enable_query_log
+
+
+--disable_query_log
+let $MYSQLD_DATADIR= `select @@datadir;`;
+--exec $MYSQL_BINLOG --verbose $MYSQLD_DATADIR/mysqld-bin.000001 > $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql
+
+create table raw_binlog_rows (txt varchar(1000));
+
+--eval load data local infile '$MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql' into table raw_binlog_rows columns terminated by '\n';
+
+select txt from raw_binlog_rows where txt like '### Extra row data %';
+
+
+drop table raw_binlog_rows;
+--enable_query_log
+
+set global debug='';
+delete from test.t1;
+reset master;
+
+--echo Specific flag settings test : Refresh
+set global debug='+d,ndb_set_refresh_anyvalue';
+
+--echo Two inserts to the same table, with nothing else in the epoch,
+--echo will be grouped together in one binlog event, as the 
+--echo extra row info is the same.
+insert into test.t1 values (1,1), (2,2);
+
+# Wait for epoch boundary
+--disable_query_log
+--disable_result_log
+show binlog events;
+--enable_result_log
+--enable_query_log
+
+--echo A separate delete.
+delete from test.t1 where a=2;
+
+# Wait for epoch boundary
+--disable_query_log
+--disable_result_log
+show binlog events;
+--enable_result_log
+--enable_query_log
+
+
+--disable_query_log
+let $MYSQLD_DATADIR= `select @@datadir;`;
+--exec $MYSQL_BINLOG --verbose $MYSQLD_DATADIR/mysqld-bin.000001 > $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql
+
+create table raw_binlog_rows (txt varchar(1000));
+
+--eval load data local infile '$MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql' into table raw_binlog_rows columns terminated by '\n';
+
+select txt from raw_binlog_rows where txt like '### Extra row data %';
+
+
+drop table raw_binlog_rows;
+--enable_query_log
+
+set global debug='';
+
+drop table t1;

sql/ha_ndbcluster.cc

@@ -4518,6 +4518,19 @@ ha_ndbcluster::eventSetAnyValue(THD *thd,
     }
   }
 #ifndef DBUG_OFF
+  DBUG_EXECUTE_IF("ndb_set_reflect_anyvalue",
+                  {
+                    fprintf(stderr, "Ndb forcing reflect AnyValue\n");
+                    options->optionsPresent |= NdbOperation::OperationOptions::OO_ANYVALUE;
+                    ndbcluster_anyvalue_set_reflect_op(options->anyValue);
+                  });
+  DBUG_EXECUTE_IF("ndb_set_refresh_anyvalue",
+                  {
+                    fprintf(stderr, "Ndb forcing refresh AnyValue\n");
+                    options->optionsPresent |= NdbOperation::OperationOptions::OO_ANYVALUE;
+                    ndbcluster_anyvalue_set_refresh_op(options->anyValue);
+                  });
+  
   /*
     MySQLD will set the user-portion of AnyValue (if any) to all 1s
     This tests code filtering ServerIds on the value of server-id-bits.
@@ -4590,10 +4603,28 @@ ha_ndbcluster::prepare_conflict_detection(enum_conflicting_op_type op_type,
   if (thd->binlog_row_event_extra_data)
   {
     Ndb_binlog_extra_row_info extra_row_info;
-    extra_row_info.loadFromBuffer(thd->binlog_row_event_extra_data);
+    if (extra_row_info.loadFromBuffer(thd->binlog_row_event_extra_data) != 0)
+    {
+      sql_print_warning("NDB Slave : Malformed event received on table %s "
+                        "cannot parse.  Stopping Slave.",
+                        m_share->key);
+      DBUG_RETURN( ER_SLAVE_CORRUPT_EVENT );
+    }
+    
     if (extra_row_info.getFlags() &
         Ndb_binlog_extra_row_info::NDB_ERIF_TRANSID)
       transaction_id = extra_row_info.getTransactionId();
+
+#ifndef DBUG_OFF
+    if (extra_row_info.getFlags() &
+        Ndb_binlog_extra_row_info::NDB_ERIF_CFT_FLAGS)
+    {
+      DBUG_PRINT("info", 
+                 ("Slave : have conflict flags : %x\n",
+                  extra_row_info.getConflictFlags()));
+      /* No effect currently */
+    }
+#endif
   }
 
   {

sql/ha_ndbcluster_binlog.cc

@@ -163,6 +163,8 @@ static pthread_mutex_t ndb_schema_share_mutex;
 extern my_bool opt_log_slave_updates;
 static my_bool g_ndb_log_slave_updates;
 
+static bool g_injector_v1_warning_emitted = false;
+
 #ifndef DBUG_OFF
 static void print_records(TABLE *table, const uchar *record)
 {
@@ -6250,6 +6252,9 @@ handle_data_event(THD* thd, Ndb *ndb, NdbEventOperation *pOp,
   Ndb_event_data *event_data= (Ndb_event_data *) pOp->getCustomData();
   TABLE *table= event_data->shadow_table;
   NDB_SHARE *share= event_data->share;
+  bool reflected_op = false;
+  bool refresh_op = false;
+
   if (pOp != share->op)
   {
     return 0;
@@ -6258,12 +6263,30 @@ handle_data_event(THD* thd, Ndb *ndb, NdbEventOperation *pOp,
   uint32 anyValue= pOp->getAnyValue();
   if (ndbcluster_anyvalue_is_reserved(anyValue))
   {
-    if (!ndbcluster_anyvalue_is_nologging(anyValue))
+    if (ndbcluster_anyvalue_is_nologging(anyValue))
+      return 0;
+    
+    if (ndbcluster_anyvalue_is_reflect_op(anyValue))
+    {
+      DBUG_PRINT("info", ("Anyvalue -> Reflect"));
+      reflected_op = true;
+      anyValue = 0;
+    }
+    else if (ndbcluster_anyvalue_is_refresh_op(anyValue))
+    {
+      DBUG_PRINT("info", ("Anyvalue -> Refresh"));
+      refresh_op = true;
+      anyValue = 0;
+    }
+    else
+    {
       sql_print_warning("NDB: unknown value for binlog signalling 0x%X, "
                         "event not logged",
                         anyValue);
-    return 0;
+      return 0;
+    }
   }
+
   uint32 originating_server_id= ndbcluster_anyvalue_get_serverid(anyValue);
   bool log_this_slave_update = g_ndb_log_slave_updates;
   bool count_this_event = true;
@@ -6364,6 +6387,7 @@ handle_data_event(THD* thd, Ndb *ndb, NdbEventOperation *pOp,
     originating_server_id= ::server_id;
   else 
   {
+    assert(!reflected_op && !refresh_op);
     /* Track that we received a replicated row event */
     if (likely( count_this_event ))
       trans_slave_row_count++;
@@ -6397,7 +6421,54 @@ handle_data_event(THD* thd, Ndb *ndb, NdbEventOperation *pOp,
   {
     extra_row_info.setFlags(Ndb_binlog_extra_row_info::NDB_ERIF_TRANSID);
     extra_row_info.setTransactionId(pOp->getTransId());
-    extra_row_info_ptr = extra_row_info.generateBuffer();
+  }
+
+  /* Set conflict flags member if necessary */
+  Uint16 event_conflict_flags = 0;
+  assert(! (reflected_op && refresh_op));
+  if (reflected_op)
+  {
+    event_conflict_flags |= NDB_ERIF_CFT_REFLECT_OP;
+  }
+  else if (refresh_op)
+  {
+    event_conflict_flags |= NDB_ERIF_CFT_REFRESH_OP;
+  }
+    
+  DBUG_EXECUTE_IF("ndb_injector_set_event_conflict_flags",
+                  {
+                    event_conflict_flags = 0xfafa;
+                  });
+  if (event_conflict_flags != 0)
+  {
+    extra_row_info.setFlags(Ndb_binlog_extra_row_info::NDB_ERIF_CFT_FLAGS);
+    extra_row_info.setConflictFlags(event_conflict_flags);
+  }
+
+  if (opt_ndb_log_transaction_id ||
+      event_conflict_flags != 0)
+  {
+    if (likely(!log_bin_use_v1_row_events))
+    {
+      extra_row_info_ptr = extra_row_info.generateBuffer();
+    }
+    else
+    {
+      /**
+       * Can't put the metadata in a v1 event
+       * Produce 1 warning at most
+       */
+      if (!g_injector_v1_warning_emitted)
+      {
+        sql_print_error("NDB: Binlog Injector discarding row event "
+                        "meta data as server is using v1 row events. "
+                        "(%u %x)",
+                        opt_ndb_log_transaction_id,
+                        event_conflict_flags);
+
+        g_injector_v1_warning_emitted = true;
+      }
+    }
   }
 
   DBUG_ASSERT(trans.good());