bug#19793475 - NO WAIT FOR EPOCHS THAT HAVE NO DB CHANGES MADE BY THIS NDB THD TO BE IN BINLOG

Ensure the most recent data changes made by the current thread to Ndb
(session_last_committed_epoch) instead of those made by all threads
(server_last_committed_epoch) to be flushed to Binlog.

Author: Maitrayi Sabaratnam

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

@@ -43,5 +43,6 @@ select * from t1;
 --source include/show_binlog_events2.inc
 PURGE MASTER LOGS TO 'mysqld-bin.000002';
 
+--source include/wait_for_ndb_committed_to_binlog.inc
 --source include/show_binlog_events2.inc
 drop table t1;

mysql-test/suite/ndb_rpl/r/ndb_rpl_circular.result

@@ -102,7 +102,6 @@ insert into t1 values (1,1);
 select server_id, log_name from mysql.ndb_apply_status order by server_id;
 server_id	log_name
 1	master-bin.000001
-show binlog events;
 'Slave' has following ndb_binlog_index entries
 select inserts, updates, deletes, schemaops, orig_server_id from mysql.ndb_binlog_index order by position;
 inserts	updates	deletes	schemaops	orig_server_id

mysql-test/suite/ndb_rpl/t/ndb_rpl_circular.test

@@ -105,7 +105,7 @@ insert into t1 values (1,1);
 select server_id, log_name from mysql.ndb_apply_status order by server_id;
 
 --disable_result_log
-show binlog events;
+--source include/wait_for_ndb_committed_to_binlog.inc
 --enable_result_log
 --echo 'Slave' has following ndb_binlog_index entries
 select inserts, updates, deletes, schemaops, orig_server_id from mysql.ndb_binlog_index order by position;

sql/ha_ndbcluster_binlog.cc

@@ -483,9 +483,11 @@ get_ndb_blobs_value(TABLE* table, NdbValue* value_array,
 
 /*
   called in mysql_show_binlog_events and reset_logs to make sure we wait for
-  all events originating from this mysql server to arrive in the binlog
+  all events originating from the 'thd' to arrive in the binlog.
 
-  Wait for the last epoch in which the last transaction is a part of.
+  'thd' is expected to be non-NULL.
+
+  Wait for the epoch in which the last transaction of the 'thd' is a part of.
 
   Wait a maximum of 30 seconds.
 */
@@ -494,31 +496,47 @@ static void ndbcluster_binlog_wait(THD *thd)
   if (ndb_binlog_running)
   {
     DBUG_ENTER("ndbcluster_binlog_wait");
-    ulonglong wait_epoch= ndb_get_latest_trans_gci();
-    /*
-      cluster not connected or no transactions done
-      so nothing to wait for
-    */
-    if (!wait_epoch)
-      DBUG_VOID_RETURN;
+    DBUG_ASSERT(thd);
 
     /*
       Binlog Injector should not wait for itself
     */
-    if (thd && 
-        thd->system_thread == SYSTEM_THREAD_NDBCLUSTER_BINLOG)
+    if (thd->system_thread == SYSTEM_THREAD_NDBCLUSTER_BINLOG)
       DBUG_VOID_RETURN;
 
-    const char *save_info= thd ? thd->proc_info : 0;
-    int count= 30;
-    if (thd)
-      thd->proc_info= "Waiting for ndbcluster binlog update to "
+    Thd_ndb *thd_ndb = get_thd_ndb(thd);
+    if (!thd_ndb)
+    {
+      /*
+       thd has not interfaced with ndb before
+       so there is no need for waiting
+      */
+       DBUG_VOID_RETURN;
+    }
+
+    const char *save_info = thd->proc_info;
+    thd->proc_info = "Waiting for ndbcluster binlog update to "
 	"reach current position";
+
     const Uint64 start_handled_epoch = ndb_latest_handled_binlog_epoch;
+   /*
+     Highest epoch that a transaction against Ndb has received
+     as part of commit processing *in this thread*. This is a
+     per-session 'most recent change' indicator.
+    */
+    const Uint64 session_last_committed_epoch =
+      thd_ndb->m_last_commit_epoch_session;
+
+    /*
+     * Wait until the last committed epoch from the session enters Binlog.
+     * Break any possible deadlock after 30s.
+     */
+    int count = 30;
+
     pthread_mutex_lock(&injector_mutex);
-    while (!(thd && thd->killed) && count && ndb_binlog_running &&
+    while (!thd->killed && count && ndb_binlog_running &&
            (ndb_latest_handled_binlog_epoch == 0 ||
-            ndb_latest_handled_binlog_epoch < wait_epoch))
+            ndb_latest_handled_binlog_epoch < session_last_committed_epoch))
     {
       count--;
       struct timespec abstime;
@@ -532,16 +550,15 @@ static void ndbcluster_binlog_wait(THD *thd)
       sql_print_warning("NDB: Thread id %llu timed out (30s) waiting for epoch %u/%u "
                         "to be handled.  Progress : %u/%u -> %u/%u.",
                         (ulonglong) thd->thread_id,
-                        Uint32((wait_epoch >> 32) & 0xffffffff),
-                        Uint32(wait_epoch & 0xffffffff),
+                        Uint32((session_last_committed_epoch >> 32) & 0xffffffff),
+                        Uint32(session_last_committed_epoch & 0xffffffff),
                         Uint32((start_handled_epoch >> 32) & 0xffffffff),
                         Uint32(start_handled_epoch & 0xffffffff),
                         Uint32((ndb_latest_handled_binlog_epoch >> 32) & 0xffffffff),
                         Uint32(ndb_latest_handled_binlog_epoch & 0xffffffff));
     }
 
-    if (thd)
-      thd->proc_info= save_info;
+    thd->proc_info= save_info;
     DBUG_VOID_RETURN;
   }
 }
@@ -913,8 +930,9 @@ static void ndbcluster_reset_slave(THD *thd)
 
 /**
   Upon the sql command flush logs, we need to ensure that all outstanding
-  ndb data to be logged has made it to the binary log to get a deterministic
-  behavior on the rotation of the log.
+  changes made to ndb by the current thread have entered the binary log
+  in order to get a deterministic behavior on the rotation of the log.
+  'current_thd' should not be NULL.
  */
 static bool ndbcluster_flush_logs(handlerton *hton)
 {