MDEV-27862 Galera should replicate nextval()-related changes in sequences with INCREMENT <> 0, at least NOCACHE ones with engine=InnoDB

Sequence storage engine is not transactionl so cache will be written in
stmt_cache that is not replicated in cluster. To fix this replicate
what is available in both trans_cache and stmt_cache.

Sequences will only work when NOCACHE keyword is used when sequnce is
created. If WSREP is enabled and we don't have this keyword report error
indicting that sequence will not work correctly in cluster.

When binlog is enabled statement cache will be cleared in transaction
before COMMIT so cache generated from sequence will not be replicated.
We need to keep cache until replication.

Tests are re-recorded because of replication changes that were
introducted with this PR.

Reviewed-by: Jan Lindström <jan.lindstrom@mariadb.com>

6 files changed, 51 insertions, 12 deletions

diff --git a/sql/ha_sequence.cc b/sql/ha_sequence.cc
index b0611c1505a..1331fea74d1 100644
--- a/sql/ha_sequence.cc
+++ b/sql/ha_sequence.cc
@@ -434,6 +434,9 @@ static int sequence_initialize(void *p)
                                HTON_HIDDEN |
                                HTON_TEMPORARY_NOT_SUPPORTED |
                                HTON_ALTER_NOT_SUPPORTED |
+#ifdef WITH_WSREP
+                               HTON_WSREP_REPLICATION |
+#endif
                                HTON_NO_PARTITION);
   DBUG_RETURN(0);
 }
diff --git a/sql/log.cc b/sql/log.cc
index 25cf0831469..ec96a2f9b23 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -2013,7 +2013,13 @@ static int binlog_commit(handlerton *hton, THD *thd, bool all)
 
   thd->backup_stage(&org_stage);
   THD_STAGE_INFO(thd, stage_binlog_write);
+#ifdef WITH_WSREP
+  // DON'T clear stmt cache in case we are in transaction
+  if (!cache_mngr->stmt_cache.empty() &&
+      (!wsrep_on(thd) || ending_trans(thd, all)))
+#else
   if (!cache_mngr->stmt_cache.empty())
+#endif
   {
     error= binlog_commit_flush_stmt_cache(thd, all, cache_mngr);
   }
@@ -10777,13 +10783,13 @@ maria_declare_plugin_end;
 #ifdef WITH_WSREP
 #include "wsrep_mysqld.h"
 
-IO_CACHE *wsrep_get_trans_cache(THD * thd)
+IO_CACHE *wsrep_get_cache(THD * thd, bool is_transactional)
 {
   DBUG_ASSERT(binlog_hton->slot != HA_SLOT_UNDEF);
   binlog_cache_mngr *cache_mngr = (binlog_cache_mngr*)
     thd_get_ha_data(thd, binlog_hton);
   if (cache_mngr)
-    return cache_mngr->get_binlog_cache_log(true);
+    return cache_mngr->get_binlog_cache_log(is_transactional);
 
   WSREP_DEBUG("binlog cache not initialized, conn: %llu",
 	      thd->thread_id);
diff --git a/sql/log.h b/sql/log.h
index 5cb75c97b5e..2520c8fede8 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -1239,7 +1239,7 @@ static inline TC_LOG *get_tc_log_implementation()
 }
 
 #ifdef WITH_WSREP
-IO_CACHE* wsrep_get_trans_cache(THD *);
+IO_CACHE* wsrep_get_cache(THD *, bool);
 void wsrep_thd_binlog_trx_reset(THD * thd);
 void wsrep_thd_binlog_stmt_rollback(THD * thd);
 #endif /* WITH_WSREP */
diff --git a/sql/sql_sequence.cc b/sql/sql_sequence.cc
index 367fbad144c..8488fa67ecb 100644
--- a/sql/sql_sequence.cc
+++ b/sql/sql_sequence.cc
@@ -308,6 +308,11 @@ bool sequence_insert(THD *thd, LEX *lex, TABLE_LIST *org_table_list)
       DBUG_RETURN(TRUE);
   }
 
+#ifdef WITH_WSREP
+  if (WSREP_ON && seq->cache != 0)
+    WSREP_WARN("CREATE SEQUENCES declared without `NOCACHE` will not behave correctly in galera cluster.");
+#endif
+
   /* If not temporary table */
   if (!temporary_table)
   {
@@ -904,12 +909,18 @@ bool Sql_cmd_alter_sequence::execute(THD *thd)
   No_such_table_error_handler no_such_table_handler;
   DBUG_ENTER("Sql_cmd_alter_sequence::execute");
 
+
   if (check_access(thd, ALTER_ACL, first_table->db.str,
                    &first_table->grant.privilege,
                    &first_table->grant.m_internal,
                    0, 0))
     DBUG_RETURN(TRUE);                  /* purecov: inspected */
 
+#ifdef WITH_WSREP
+  if (WSREP_ON && new_seq->cache != 0)
+    WSREP_WARN("ALTER SEQUENCES declared without `NOCACHE` will not behave correctly in galera cluster.");
+#endif
+
   if (check_grant(thd, ALTER_ACL, first_table, FALSE, 1, FALSE))
     DBUG_RETURN(TRUE);                  /* purecov: inspected */
 
diff --git a/sql/wsrep_binlog.cc b/sql/wsrep_binlog.cc
index d04d9989e99..5c1c899fcde 100644
--- a/sql/wsrep_binlog.cc
+++ b/sql/wsrep_binlog.cc
@@ -155,10 +155,10 @@ static int wsrep_write_cache_inc(THD*      const thd,
         goto cleanup;
       cache->read_pos= cache->read_end;
     } while ((cache->file >= 0) && (length= my_b_fill(cache)));
-  }
-  if (ret == 0)
-  {
-    assert(total_length + thd->wsrep_sr().log_position() == saved_pos);
+    if (ret == 0)
+    {
+      assert(total_length + thd->wsrep_sr().log_position() == saved_pos);
+    }
   }
 
 cleanup:
diff --git a/sql/wsrep_client_service.cc b/sql/wsrep_client_service.cc
index 464296ea6cf..59ba8c30e43 100644
--- a/sql/wsrep_client_service.cc
+++ b/sql/wsrep_client_service.cc
@@ -86,18 +86,37 @@ int Wsrep_client_service::prepare_data_for_replication()
   DBUG_ASSERT(m_thd == current_thd);
   DBUG_ENTER("Wsrep_client_service::prepare_data_for_replication");
   size_t data_len= 0;
-  IO_CACHE* cache= wsrep_get_trans_cache(m_thd);
+  IO_CACHE* transactional_cache= wsrep_get_cache(m_thd, true);
+  IO_CACHE* stmt_cache= wsrep_get_cache(m_thd, false);
 
-  if (cache)
+  if (transactional_cache || stmt_cache)
   {
     m_thd->binlog_flush_pending_rows_event(true);
-    if (wsrep_write_cache(m_thd, cache, &data_len))
+
+    size_t transactional_data_len= 0;
+    size_t stmt_data_len= 0;
+
+    // Write transactional cache
+    if (transactional_cache &&
+        wsrep_write_cache(m_thd, transactional_cache, &transactional_data_len))
     {
       WSREP_ERROR("rbr write fail, data_len: %zu",
                   data_len);
       // wsrep_override_error(m_thd, ER_ERROR_DURING_COMMIT);
       DBUG_RETURN(1);
     }
+
+    // Write stmt cache
+    if (stmt_cache && wsrep_write_cache(m_thd, stmt_cache, &stmt_data_len))
+    {
+      WSREP_ERROR("rbr write fail, data_len: %zu",
+                  data_len);
+      // wsrep_override_error(m_thd, ER_ERROR_DURING_COMMIT);
+      DBUG_RETURN(1);
+    }
+
+    // Complete data written from both caches
+    data_len = transactional_data_len + stmt_data_len;
   }
 
   if (data_len == 0)
@@ -139,7 +158,7 @@ int Wsrep_client_service::prepare_fragment_for_replication(
   DBUG_ASSERT(m_thd == current_thd);
   THD* thd= m_thd;
   DBUG_ENTER("Wsrep_client_service::prepare_fragment_for_replication");
-  IO_CACHE* cache= wsrep_get_trans_cache(thd);
+  IO_CACHE* cache= wsrep_get_cache(thd, true);
   thd->binlog_flush_pending_rows_event(true);
 
   if (!cache)
@@ -221,7 +240,7 @@ bool Wsrep_client_service::statement_allowed_for_streaming() const
 
 size_t Wsrep_client_service::bytes_generated() const
 {
-  IO_CACHE* cache= wsrep_get_trans_cache(m_thd);
+  IO_CACHE* cache= wsrep_get_cache(m_thd, true);
   if (cache)
   {
     size_t pending_rows_event_length= 0;