MDEV-34124: Fix streaming replication offset for binlog stmt cache

As the binlog statement cache is only replicated with the last
fragment, it's safe to pass zero offset instead of the stored
log position, which is used only for the binlog transaction cache.

Author: mariadb-DenisProtivensky

sql/wsrep_binlog.cc

@@ -118,12 +118,13 @@ int wsrep_write_cache_buf(IO_CACHE *cache, uchar **buf, size_t *buf_len)
  */
 static int wsrep_write_cache_inc(THD*      const thd,
                                  IO_CACHE* const cache,
+                                 size_t    const log_position,
                                  size_t*   const len)
 {
   DBUG_ENTER("wsrep_write_cache_inc");
   my_off_t const saved_pos(my_b_tell(cache));
 
-  if (reinit_io_cache(cache, READ_CACHE, thd->wsrep_sr().log_position(), 0, 0))
+  if (reinit_io_cache(cache, READ_CACHE, log_position, 0, 0))
   {
     WSREP_ERROR("failed to initialize io-cache");
     DBUG_RETURN(1);;
@@ -157,7 +158,7 @@ static int wsrep_write_cache_inc(THD*      const thd,
     } while ((cache->file >= 0) && (length= my_b_fill(cache)));
     if (ret == 0)
     {
-      assert(total_length + thd->wsrep_sr().log_position() == saved_pos);
+      assert(total_length + log_position == saved_pos);
     }
   }
 
@@ -178,9 +179,10 @@ static int wsrep_write_cache_inc(THD*      const thd,
  */
 int wsrep_write_cache(THD*      const thd,
                       IO_CACHE* const cache,
+                      size_t    const log_position,
                       size_t*   const len)
 {
-  return wsrep_write_cache_inc(thd, cache, len);
+  return wsrep_write_cache_inc(thd, cache, log_position, len);
 }
 
 void wsrep_dump_rbr_buf(THD *thd, const void* rbr_buf, size_t buf_len)

sql/wsrep_binlog.h

@@ -36,11 +36,13 @@ int wsrep_write_cache_buf(IO_CACHE *cache, uchar **buf, size_t *buf_len);
   This function quite the same as MYSQL_BIN_LOG::write_cache(),
   with the exception that here we write in buffer instead of log file.
 
+  @param log_position position to start writing the cache from
   @param len  total amount of data written
   @return     wsrep error status
  */
 int  wsrep_write_cache(THD*      thd,
                        IO_CACHE* cache,
+                       size_t    log_position,
                        size_t*   len);
 
 /* Dump replication buffer to disk */

sql/wsrep_client_service.cc

@@ -95,18 +95,20 @@ int Wsrep_client_service::prepare_data_for_replication()
     size_t transactional_data_len= 0;
     size_t stmt_data_len= 0;
 
-    // Write transactional cache
+    // Write transactional cache from the last remembered position
     if (transactional_cache &&
-        wsrep_write_cache(m_thd, transactional_cache, &transactional_data_len))
+        wsrep_write_cache(m_thd, transactional_cache,
+                          m_thd->wsrep_sr().log_position(),
+                          &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))
+    // Write stmt cache fully as it's only written upon transaction commit
+    if (stmt_cache && wsrep_write_cache(m_thd, stmt_cache, 0u, &stmt_data_len))
     {
       WSREP_ERROR("rbr write fail, data_len: %zu",
                   data_len);