WL7639 Ndb Active Active Manageability improvements

Implement status variables exposing last commit epochs for the server
and each session.

New status variables Ndb_last_commit_epoch_server and 
Ndb_last_commit_epoch_session are added to the MySQL server, allowing
SQL layer code to determine e.g.
  - What was the epoch of the last transaction committed by this server
    (Same info as embedded in SHOW ENGINE NDB STATUS output)
  - What was the epoch of the last transaction committed by this session

The session variant particularly is useful when determining when various
post-commit activities have completed on a transaction.  e.g. it could be 
used to determine k-safety round a replication ring by comparing with
the Max_replicated_epoch, or if a latest-disk-durable epoch were exposed, it
could be used to determine when a transaction becomes disk-durable.

A new testcase is added to verify the behaviour of the variables.

Author: frazerclement

mysql-test/suite/ndb/r/ndb_basic.result

@@ -89,6 +89,8 @@ Ndb_execute_count	#
 Ndb_index_stat_cache_clean	#
 Ndb_index_stat_cache_query	#
 Ndb_index_stat_status	#
+Ndb_last_commit_epoch_server	#
+Ndb_last_commit_epoch_session	#
 Ndb_number_of_data_nodes	#
 Ndb_number_of_ready_data_nodes	#
 Ndb_pruned_scan_count	#

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

@@ -0,0 +1,63 @@
+Reset binlog and show no state held
+reset master;
+select count(1) from mysql.ndb_binlog_index;
+count(1)
+0
+Create table for experiments
+use test;
+create table test.t1 (a int primary key, b varchar(1000), c text) engine=ndb;
+Basic test that an insert affects Server + Session epochs
+insert into test.t1 values (1, repeat("B", 1000), repeat("E", 500));
+init_server_epoch_includes_session_epoch	curr_server_epoch_includes_session_epoch	server_epoch_does_not_regress	session_epoch_does_not_regress
+1	1	1	1
+binlog_contains_last_session_epoch
+1
+Basic test that a read with exclusive lock has no effect
+begin;
+select count(1) from test.t1 where a=1 for update;
+count(1)
+1
+commit;
+init_server_epoch_includes_session_epoch	curr_server_epoch_includes_session_epoch	server_epoch_does_not_regress	session_epoch_does_not_regress
+1	1	1	1
+binlog_contains_last_session_epoch
+0
+Basic test that a read with shared lock has no effect
+begin;
+select count(1) from test.t1 where a=1 lock in share mode;
+count(1)
+1
+commit;
+init_server_epoch_includes_session_epoch	curr_server_epoch_includes_session_epoch	server_epoch_does_not_regress	session_epoch_does_not_regress
+1	1	1	1
+binlog_contains_last_session_epoch
+0
+Basic test that a committedread has no effect
+begin;
+select count(1) from test.t1 where a=1;
+count(1)
+1
+commit;
+init_server_epoch_includes_session_epoch	curr_server_epoch_includes_session_epoch	server_epoch_does_not_regress	session_epoch_does_not_regress
+1	1	1	1
+binlog_contains_last_session_epoch
+0
+Basic test that an update affects Server + Session epochs
+update test.t1 set b=repeat("E", 1000), c=repeat("A", 5000) where a=1;
+init_server_epoch_includes_session_epoch	curr_server_epoch_includes_session_epoch	server_epoch_does_not_regress	session_epoch_does_not_regress
+1	1	1	1
+binlog_contains_last_session_epoch
+1
+Basic test that a delete affects Server + Session epochs
+delete from test.t1 where a=1;
+init_server_epoch_includes_session_epoch	curr_server_epoch_includes_session_epoch	server_epoch_does_not_regress	session_epoch_does_not_regress
+1	1	1	1
+binlog_contains_last_session_epoch
+1
+Show that server epoch is preserved across connections
+but session epoch is not
+server_epoch_preserved
+1
+session_epoch_reset
+1
+drop table test.t1;

mysql-test/suite/ndb_binlog/t/ndb_binlog_cmp_epoch_vals.inc

@@ -0,0 +1,31 @@
+--disable_query_log
+--disable_result_log
+
+select @curr_server_epoch:=Variable_value 
+from information_schema.global_status 
+where Variable_name ="ndb_last_commit_epoch_server";
+
+select @curr_session_epoch:=Variable_value 
+from information_schema.global_status 
+where Variable_name ="ndb_last_commit_epoch_session";
+
+show binlog events;
+
+select @binlog_max_epoch:= max(epoch) 
+from mysql.ndb_binlog_index;
+
+--enable_result_log
+
+select (@init_server_epoch + 0) >= (@init_session_epoch + 0) 
+  as init_server_epoch_includes_session_epoch,
+  (@curr_server_epoch + 0) >= (@curr_session_epoch + 0) 
+  as curr_server_epoch_includes_session_epoch,
+  (@curr_server_epoch + 0) >= (@init_server_epoch + 0) 
+  as server_epoch_does_not_regress,
+  (@curr_session_epoch + 0) >= (@init_session_epoch + 0) 
+  as session_epoch_does_not_regress;
+
+select @binlog_max_epoch = @curr_session_epoch as
+  binlog_contains_last_session_epoch;
+
+--enable_query_log

mysql-test/suite/ndb_binlog/t/ndb_binlog_init_epoch_vals.inc

@@ -0,0 +1,13 @@
+--disable_query_log
+--disable_result_log
+
+select @init_server_epoch:=Variable_value 
+from information_schema.global_status 
+where Variable_name ="ndb_last_commit_epoch_server";
+
+select @init_session_epoch:=Variable_value 
+from information_schema.global_status 
+where Variable_name ="ndb_last_commit_epoch_session";
+
+--enable_query_log
+--enable_result_log

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

@@ -0,0 +1,65 @@
+-- source include/have_ndb.inc
+-- source include/have_binlog_format_mixed_or_row.inc
+
+--echo Reset binlog and show no state held
+reset master;
+select count(1) from mysql.ndb_binlog_index;
+
+--echo Create table for experiments
+use test;
+create table test.t1 (a int primary key, b varchar(1000), c text) engine=ndb;
+
+--echo Basic test that an insert affects Server + Session epochs
+--source ndb_binlog_init_epoch_vals.inc
+insert into test.t1 values (1, repeat("B", 1000), repeat("E", 500));
+--source ndb_binlog_cmp_epoch_vals.inc
+
+--echo Basic test that a read with exclusive lock has no effect
+--source ndb_binlog_init_epoch_vals.inc
+begin;
+select count(1) from test.t1 where a=1 for update;
+commit;
+--source ndb_binlog_cmp_epoch_vals.inc
+
+--echo Basic test that a read with shared lock has no effect
+--source ndb_binlog_init_epoch_vals.inc
+begin;
+select count(1) from test.t1 where a=1 lock in share mode;
+commit;
+--source ndb_binlog_cmp_epoch_vals.inc
+
+--echo Basic test that a committedread has no effect
+--source ndb_binlog_init_epoch_vals.inc
+begin;
+select count(1) from test.t1 where a=1;
+commit;
+--source ndb_binlog_cmp_epoch_vals.inc
+
+--echo Basic test that an update affects Server + Session epochs
+--source ndb_binlog_init_epoch_vals.inc
+update test.t1 set b=repeat("E", 1000), c=repeat("A", 5000) where a=1;
+--source ndb_binlog_cmp_epoch_vals.inc
+
+--echo Basic test that a delete affects Server + Session epochs
+--source ndb_binlog_init_epoch_vals.inc
+delete from test.t1 where a=1;
+--source ndb_binlog_cmp_epoch_vals.inc
+
+--echo Show that server epoch is preserved across connections
+--echo but session epoch is not
+
+--source ndb_binlog_init_epoch_vals.inc
+let $server_epoch= query_get_value(select @init_server_epoch as e, e, 1);
+let $session_epoch= query_get_value(select @init_session_epoch as e, e, 1);
+connect('ExtraConn', '127.0.0.1', 'root',,,$MASTER_MYPORT);
+--source ndb_binlog_init_epoch_vals.inc
+let $new_server_epoch= query_get_value(select @init_server_epoch as e, e, 1);
+let $new_session_epoch= query_get_value(select @init_session_epoch as e, e, 1);
+
+--disable_query_log
+eval select $new_server_epoch >= $server_epoch as server_epoch_preserved;
+eval select $new_session_epoch < $session_epoch as session_epoch_reset;
+--enable_query_log
+
+drop table test.t1;
+

sql/ha_ndbcluster.cc

@@ -595,6 +595,7 @@ static int update_status_variables(Thd_ndb *thd_ndb,
   }
   ns->number_of_data_nodes= c->no_db_nodes();
   ns->connect_count= c->get_connect_count();
+  ns->last_commit_epoch_server= ndb_get_latest_trans_gci();
   if (thd_ndb)
   {
     ns->execute_count= thd_ndb->m_execute_count;
@@ -605,6 +606,7 @@ static int update_status_variables(Thd_ndb *thd_ndb,
     ns->pushed_queries_dropped= thd_ndb->m_pushed_queries_dropped;
     ns->pushed_queries_executed= thd_ndb->m_pushed_queries_executed;
     ns->pushed_reads= thd_ndb->m_pushed_reads;
+    ns->last_commit_epoch_session = thd_ndb->m_last_commit_epoch_session;
     for (int i= 0; i < MAX_NDB_NODES; i++)
     {
       ns->transaction_no_hint_count[i]= thd_ndb->m_transaction_no_hint_count[i];
@@ -708,6 +710,12 @@ SHOW_VAR ndb_status_variables_dynamic[]= {
   {"pushed_queries_executed", (char*) &g_ndb_status.pushed_queries_executed,
    SHOW_LONG},
   {"pushed_reads",       (char*) &g_ndb_status.pushed_reads,          SHOW_LONG},
+  {"last_commit_epoch_server", 
+                         (char*) &g_ndb_status.last_commit_epoch_server,
+                                                                      SHOW_LONGLONG},
+  {"last_commit_epoch_session", 
+                         (char*) &g_ndb_status.last_commit_epoch_session, 
+                                                                      SHOW_LONGLONG},
   {NullS, NullS, SHOW_LONG}
 };
 
@@ -1143,6 +1151,20 @@ execute_commit(Thd_ndb *thd_ndb, NdbTransaction *trans,
                                    ignore_count);
   } while (0);
 
+  if (likely(rc == 0))
+  {
+    /* Committed ok, update session GCI, if it's available
+     * (Not available for reads, empty transactions etc...)
+     */
+    Uint64 reportedGCI;
+    if (trans->getGCI(&reportedGCI) == 0 &&
+        reportedGCI != 0)
+    {
+      assert(reportedGCI >= thd_ndb->m_last_commit_epoch_session);
+      thd_ndb->m_last_commit_epoch_session = reportedGCI;
+    }
+  }
+
   if (thd_ndb->is_slave_thread())
   {
     if (likely(rc == 0))
@@ -1193,7 +1215,8 @@ Thd_ndb::Thd_ndb(THD* thd) :
   m_thd(thd),
   m_slave_thread(thd->slave_thread),
   m_skip_binlog_setup_in_find_files(false),
-  schema_locks_count(0)
+  schema_locks_count(0),
+  m_last_commit_epoch_session(0)
 {
   connection= ndb_get_cluster_connection();
   m_connect_count= connection->get_connect_count();

sql/ha_ndbcluster.h

@@ -124,6 +124,8 @@ struct st_ndb_status {
   long pushed_queries_dropped;
   long pushed_queries_executed;
   long pushed_reads;
+  long last_commit_epoch_server;
+  long last_commit_epoch_session;
   long transaction_no_hint_count[MAX_NDB_NODES];
   long transaction_hint_count[MAX_NDB_NODES];
   long long api_client_stats[Ndb::NumClientStatistics];

sql/ndb_thd_ndb.cc

@@ -1,5 +1,5 @@
 /*
-   Copyright (c) 2011, 2013 Oracle and/or its affiliates. All rights reserved.
+   Copyright (c) 2011, 2014 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
@@ -89,6 +89,10 @@ Thd_ndb::recycle_ndb(void)
   {
     ndb->setCustomData64(thd_get_thread_id(m_thd));
   }
+
+  /* Reset last commit epoch for this 'session'. */
+  m_last_commit_epoch_session = 0;
+
   DBUG_RETURN(true);
 }
 

sql/ndb_thd_ndb.h

@@ -1,5 +1,5 @@
 /*
-   Copyright (c) 2011, Oracle and/or its affiliates. All rights reserved.
+   Copyright (c) 2011, 2014 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
@@ -135,6 +135,11 @@ class Thd_ndb
   uint schema_locks_count; // Number of global schema locks taken by thread
   bool has_required_global_schema_lock(const char* func);
 
+  /**
+     Epoch of last committed transaction in this session, 0 if none so far
+   */
+  Uint64 m_last_commit_epoch_session;
+
   unsigned m_connect_count;
   bool valid_ndb(void) const;
   bool recycle_ndb(void);