MDEV-31272 Statement rollback causes empty writeset replication

This patch fixes cases where a transaction caused empty writeset to be
replicated. This could happen in the case where a transaction executes
a statement that initially manages to modify some data and therefore
appended keys some for  certification. The statement is however rolled
back at some later stage due to some error (for example, a duplicate
key error). After statement rollback the transaction is still alive,
has no other changes. When committing such transaction, an empty
writeset was replicated through Galera.

The fix is to avoid calling into commit hook only when transaction
has appended one or keys for certification *and* has some data in
binlog cache to replicate. Otherwise, the commit is considered empty,
and goes through usual empty commit path.
Signed-off-by: Julius Goryavsky <julius.goryavsky@mariadb.com>

mysql-test/suite/galera/r/MDEV-31272.result

+connection node_2;
+connection node_1;
+connection node_1;
+CREATE TABLE t1 (f1 int primary key, f2 int);
+INSERT INTO t1 VALUES (1,0);
+BEGIN;
+INSERT INTO t1 VALUES (2,4),(1,1);
+ERROR 23000: Duplicate entry '1' for key 'PRIMARY'
+COMMIT;
+Writesets replicated (expect 0)
+0
+connection node_1;
+SELECT * FROM t1;
+f1	f2
+1	0
+connection node_2;
+SELECT * FROM t1;
+f1	f2
+1	0
+DROP TABLE t1;
+connection node_1;
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY, f2 INTEGER);
+INSERT INTO t1 VALUES (1,0);
+INSERT INTO t1 VALUES (2,4), (1,1);
+ERROR 23000: Duplicate entry '1' for key 'PRIMARY'
+Writesets replicated (expect 0)
+0
+connection node_1;
+SELECT * FROM t1;
+f1	f2
+1	0
+connection node_2;
+SELECT * FROM t1;
+f1	f2
+1	0
+DROP TABLE t1;

mysql-test/suite/galera/t/MDEV-31272.combinations

+[binlogon]
+log-bin
+log-slave-updates=ON
+
+[binlogoff]

mysql-test/suite/galera/t/MDEV-31272.test

+#
+# MDEV-31272: Statement rollback causes empty writeset replication
+#
+
+--source include/galera_cluster.inc
+
+#
+# Case 1: Multi statement transaction
+#
+--connection node_1
+CREATE TABLE t1 (f1 int primary key, f2 int);
+INSERT INTO t1 VALUES (1,0);
+
+--let $replicated_old = `SELECT VARIABLE_VALUE FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_replicated'`
+
+BEGIN;
+--error ER_DUP_ENTRY
+INSERT INTO t1 VALUES (2,4),(1,1);
+COMMIT;
+
+--let $replicated_new = `SELECT VARIABLE_VALUE FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_replicated'`
+
+--disable_query_log
+--eval SELECT $replicated_new - $replicated_old AS 'Writesets replicated (expect 0)';
+--enable_query_log
+
+--connection node_1
+SELECT * FROM t1;
+--connection node_2
+SELECT * FROM t1;
+
+DROP TABLE t1;
+
+
+#
+# Case 2: autocommit statement
+#
+--connection node_1
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY, f2 INTEGER);
+INSERT INTO t1 VALUES (1,0);
+
+--let $replicated_old = `SELECT VARIABLE_VALUE FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_replicated'`
+
+--error ER_DUP_ENTRY
+INSERT INTO t1 VALUES (2,4), (1,1);
+
+--let $replicated_new = `SELECT VARIABLE_VALUE FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_replicated'`
+
+--disable_query_log
+--eval SELECT $replicated_new - $replicated_old AS 'Writesets replicated (expect 0)';
+--enable_query_log
+
+--connection node_1
+SELECT * FROM t1;
+--connection node_2
+SELECT * FROM t1;
+
+DROP TABLE t1;

sql/log.cc

@@ -10894,6 +10894,15 @@ IO_CACHE *wsrep_get_cache(THD * thd, bool is_transactional)
   return NULL;
 }
 
+bool wsrep_is_binlog_cache_empty(THD *thd)
+{
+  binlog_cache_mngr *cache_mngr=
+      (binlog_cache_mngr *) thd_get_ha_data(thd, binlog_hton);
+  if (cache_mngr)
+    return cache_mngr->trx_cache.empty() && cache_mngr->stmt_cache.empty();
+  return true;
+}
+
 void wsrep_thd_binlog_trx_reset(THD * thd)
 {
   DBUG_ENTER("wsrep_thd_binlog_trx_reset");

sql/log.h

@@ -1240,6 +1240,7 @@ static inline TC_LOG *get_tc_log_implementation()
 
 #ifdef WITH_WSREP
 IO_CACHE* wsrep_get_cache(THD *, bool);
+bool wsrep_is_binlog_cache_empty(THD *);
 void wsrep_thd_binlog_trx_reset(THD * thd);
 void wsrep_thd_binlog_stmt_rollback(THD * thd);
 #endif /* WITH_WSREP */

sql/wsrep_trans_observer.h

@@ -88,7 +88,13 @@ static inline bool wsrep_is_real(THD* thd, bool all)
  */
 static inline bool wsrep_has_changes(THD* thd)
 {
-  return (thd->wsrep_trx().is_empty() == false);
+  // Transaction has changes to replicate if it
+  // has appended one or more certification keys,
+  // and has actual changes to replicate in binlog
+  // cache. Except for streaming replication,
+  // where commit message may have no payload.
+  return !thd->wsrep_trx().is_empty() &&
+    (!wsrep_is_binlog_cache_empty(thd) || thd->wsrep_trx().is_streaming());
 }
 
 /*