BUG#32019842: WRITE_SET VECTOR HAS NO UPPER MEMORY LIMIT

When the option is active, each server transaction will collect write
sets for each row, allowing to identify what changes the transaction
made, important for replication parallelization in general and Group
Replication conflict detection.

This patch improves the handling of write set collection for each
transaction in 2 ways.

First is that write sets are discarded if their size grow above
binlog_transaction_dependency_history_size. As a consequence the
transaction will no longer be marked as concurrent with previous ones
even if they touch different data.

The second one is that components like Group Replication can set
memory limits for the collection of write sets. Components like GR are
dependent on write sets being present, so they will never be
discarded, but breaking this memory limit will cause transactions to
abort.

ReviewBoard: 25520
ReviewBoard: 25592

Author: pedrolgomes

include/mysql/service_rpl_transaction_write_set.h

@@ -1,4 +1,4 @@
-/* Copyright (c) 2014, 2016, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2014, 2020, 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, version 2.0,
@@ -66,17 +66,32 @@ typedef struct st_trans_write_set Transaction_write_set;
 
 extern struct transaction_write_set_service_st {
   Transaction_write_set* (*get_transaction_write_set)(unsigned long m_thread_id);
+  void (*require_full_write_set)(int requires_ws);
+  void (*set_write_set_memory_size_limit)(long long size_limit);
+  void (*update_write_set_memory_size_limit)(long long  size_limit);
 } *transaction_write_set_service;
 
 #ifdef MYSQL_DYNAMIC_PLUGIN
 
 #define get_transaction_write_set(m_thread_id) \
   (transaction_write_set_service->get_transaction_write_set((m_thread_id)))
+#define require_full_write_set(requires_ws) \
+  transaction_write_set_service->require_full_write_set(requires_ws)
+#define set_write_set_memory_size_limit(size_limit) \
+  transaction_write_set_service->set_write_set_memory_size_limit(size_limit)
+#define update_write_set_memory_size_limit(size_limit) \
+  transaction_write_set_service->update_write_set_memory_size_limit(size_limit)
 
 #else
 
 Transaction_write_set* get_transaction_write_set(unsigned long m_thread_id);
 
+void require_full_write_set(int requires_ws);
+
+void set_write_set_memory_size_limit(long long size_limit);
+
+void update_write_set_memory_size_limit(long long size_limit);
+
 #endif
 
 #ifdef __cplusplus

include/mysql/services.h.pp

@@ -444,8 +444,14 @@
 typedef struct st_trans_write_set Transaction_write_set;
 extern struct transaction_write_set_service_st {
   Transaction_write_set* (*get_transaction_write_set)(unsigned long m_thread_id);
+  void (*require_full_write_set)(int requires_ws);
+  void (*set_write_set_memory_size_limit)(long long size_limit);
+  void (*update_write_set_memory_size_limit)(long long  size_limit);
 } *transaction_write_set_service;
 Transaction_write_set* get_transaction_write_set(unsigned long m_thread_id);
+void require_full_write_set(int requires_ws);
+void set_write_set_memory_size_limit(long long size_limit);
+void update_write_set_memory_size_limit(long long size_limit);
 #include <mysql/service_security_context.h>
 typedef char my_svc_bool;
 extern struct security_context_service_st {

mysql-test/suite/rpl/r/rpl_binlog_transaction_dependency_history_size.result

@@ -3,6 +3,7 @@ Warnings:
 Note	####	Sending passwords in plain text without SSL/TLS is extremely insecure.
 Note	####	Storing MySQL user name or password information in the master info repository is not secure and is therefore not recommended. Please consider using the USER and PASSWORD connection options for START SLAVE; see the 'START SLAVE Syntax' in the MySQL Manual for more information.
 [connection master]
+SET @binlog_transaction_dependency_tracking_save = @@GLOBAL.binlog_transaction_dependency_tracking;
 
 SET GLOBAL binlog_transaction_dependency_tracking = WRITESET;
 ######## 1. WRITESET ########
@@ -105,4 +106,5 @@ FLUSH LOGS;
 include/include/assert_logical_timestamps.inc [0 1;1 2;2 3;3 4;4 5;5 6;6 7;7 8;8 9;9 10;10 11;11 12;9 13;13 14]
 # Verify that replication is correct
 include/sync_slave_sql_with_master.inc
+SET GLOBAL binlog_transaction_dependency_tracking = @binlog_transaction_dependency_tracking_save;
 include/rpl_end.inc

mysql-test/suite/rpl/t/rpl_binlog_transaction_dependency_history_size.test

@@ -25,11 +25,15 @@
 # ==== References ====
 # WL#9556: Writeset-based MTS dependency tracking on master
 ################################################################################
+
+--source include/have_binlog_format_row.inc
 --let $rpl_extra_connections_per_server= 7
 --source include/master-slave.inc
---source include/have_binlog_format_row.inc
 
 --connection master
+
+SET @binlog_transaction_dependency_tracking_save = @@GLOBAL.binlog_transaction_dependency_tracking;
+
 # Loop between binlog_transaction_dependency_tracking modes
 --let $type= 0
 while ($type < 3)
@@ -132,4 +136,7 @@ while ($type < 3)
 }
 
 --connection master
+
+SET GLOBAL binlog_transaction_dependency_tracking = @binlog_transaction_dependency_tracking_save;
+
 --source include/rpl_end.inc

rapid/plugin/group_replication/include/replication_threads_api.h

@@ -1,4 +1,4 @@
-/* Copyright (c) 2014, 2017, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2014, 2020, 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, version 2.0,
@@ -75,6 +75,8 @@ class Replication_thread_api
     @param priority      The channel priority on event application
     @param retry_count   The number of retries when connecting
     @param preserve_logs If logs should be always preserved
+    @param ignore_ws_mem_limit Shall ignore write set mem limits
+    @param allow_drop_write_set Shall not require write set to be preserved
 
     @return the operation status
       @retval 0      OK
@@ -93,7 +95,9 @@ class Replication_thread_api
                          bool ssl_verify_server_cert,
                          int priority,
                          int retry_count,
-                         bool preserve_logs);
+                         bool preserve_logs,
+                         bool ignore_ws_mem_limit,
+                         bool allow_drop_write_set);
 
   /**
     Start the Applier/Receiver threads according to the given options.

rapid/plugin/group_replication/src/handlers/applier_handler.cc

@@ -1,4 +1,4 @@
-/* Copyright (c) 2014, 2017, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2014, 2020, 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, version 2.0,
@@ -77,7 +77,7 @@ Applier_handler::initialize_repositories(bool reset_logs,
                                               NULL,
                                               false,
                                               GROUP_REPLICATION_APPLIER_THREAD_PRIORITY,
-                                              0, true);
+                                              0, true, true, true);
 
   if (error)
   {

rapid/plugin/group_replication/src/plugin.cc

@@ -21,6 +21,7 @@
    51 Franklin Street, Suite 500, Boston, MA 02110-1335 USA */
 
 #include <sstream>
+#include <mysql/service_rpl_transaction_write_set.h>
 
 #include "observer_server_actions.h"
 #include "observer_server_state.h"
@@ -207,7 +208,8 @@ int flow_control_applier_threshold_var= DEFAULT_FLOW_CONTROL_THRESHOLD;
 #define DEFAULT_TRANSACTION_SIZE_LIMIT 0
 #define MAX_TRANSACTION_SIZE_LIMIT 2147483647
 #define MIN_TRANSACTION_SIZE_LIMIT 0
-ulong transaction_size_limit_var= DEFAULT_TRANSACTION_SIZE_LIMIT;
+ulong transaction_size_limit_base_var= DEFAULT_TRANSACTION_SIZE_LIMIT;
+int64 transaction_size_limit_var;
 
 /* Member Weight limits */
 #define DEFAULT_MEMBER_WEIGHT 50
@@ -459,6 +461,7 @@ int initialize_plugin_and_join(enum_plugin_con_isolation sql_api_isolation,
   //Avoid unnecessary operations
   bool enabled_super_read_only= false;
   bool read_only_mode= false, super_read_only_mode=false;
+  bool write_set_limits_set = false;
 
   st_server_ssl_variables server_ssl_variables=
     {false,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL};
@@ -512,6 +515,10 @@ int initialize_plugin_and_join(enum_plugin_con_isolation sql_api_isolation,
   if (delayed_init_thd)
     delayed_init_thd->signal_read_mode_ready();
 
+  require_full_write_set(1);
+  set_write_set_memory_size_limit(get_transaction_size_limit());
+  write_set_limits_set = true;
+
   get_server_parameters(&hostname, &port, &uuid, &server_version,
                         &server_ssl_variables);
 
@@ -623,6 +630,12 @@ int initialize_plugin_and_join(enum_plugin_con_isolation sql_api_isolation,
     leave_group();
     terminate_plugin_modules();
 
+    if (write_set_limits_set) {
+      // Remove server constraints on write set collection
+      update_write_set_memory_size_limit(0);
+      require_full_write_set(0);
+    }
+
     if (!server_shutdown_status && server_engine_initialized()
         && enabled_super_read_only)
     {
@@ -913,6 +926,10 @@ int plugin_group_replication_stop()
     plugin_is_waiting_to_set_server_read_mode= false;
   }
 
+  // Remove server constraints on write set collection
+  update_write_set_memory_size_limit(0);
+  require_full_write_set(0);
+
   DBUG_RETURN(error);
 }
 
@@ -1075,6 +1092,9 @@ int plugin_group_replication_init(MYSQL_PLUGIN plugin_info)
   //Initialize the compatibility module before starting
   init_compatibility_manager();
 
+  // Set the atomic var to the value of the base plugin variable
+  transaction_size_limit_var = transaction_size_limit_base_var;
+
   plugin_is_auto_starting= start_group_replication_at_boot_var;
   if (start_group_replication_at_boot_var && plugin_group_replication_start())
   {
@@ -1573,7 +1593,9 @@ bool get_allow_local_disjoint_gtids_join()
 ulong get_transaction_size_limit()
 {
   DBUG_ENTER("get_transaction_size_limit");
-  DBUG_RETURN(transaction_size_limit_var);
+  DBUG_ASSERT(my_atomic_load64(&transaction_size_limit_var)>=0);
+  ulong limit = static_cast<ulong>(my_atomic_load64(&transaction_size_limit_var));
+  DBUG_RETURN(limit);
 }
 
 bool is_plugin_waiting_to_set_server_read_mode()
@@ -2364,6 +2386,20 @@ update_member_weight(MYSQL_THD, SYS_VAR*,
   DBUG_VOID_RETURN;
 }
 
+static void update_transaction_size_limit(MYSQL_THD, SYS_VAR *, void *var_ptr,
+                                          const void *save) {
+
+  ulong in_val = *static_cast<const ulong *>(save);
+  *static_cast<ulong *>(var_ptr) = in_val;
+  my_atomic_store64(&transaction_size_limit_var, in_val);
+
+  transaction_size_limit_var = in_val;
+
+  if (plugin_is_group_replication_running()) {
+    update_write_set_memory_size_limit(transaction_size_limit_var);
+  }
+}
+
 //Base plugin variables
 
 static MYSQL_SYSVAR_STR(
@@ -2791,11 +2827,11 @@ static MYSQL_SYSVAR_INT(
 
 static MYSQL_SYSVAR_ULONG(
   transaction_size_limit,              /* name */
-  transaction_size_limit_var,          /* var */
+  transaction_size_limit_base_var,          /* var */
   PLUGIN_VAR_OPCMDARG,                 /* optional var */
   "Specifies the limit of transaction size that can be transferred over network.",
   NULL,                                /* check func. */
-  NULL,                                /* update func. */
+  update_transaction_size_limit,       /* update func. */
   DEFAULT_TRANSACTION_SIZE_LIMIT,      /* default */
   MIN_TRANSACTION_SIZE_LIMIT,          /* min */
   MAX_TRANSACTION_SIZE_LIMIT,          /* max */

rapid/plugin/group_replication/src/recovery_state_transfer.cc

@@ -1,4 +1,4 @@
-/* Copyright (c) 2015, 2019, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2015, 2020, 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, version 2.0,
@@ -529,7 +529,7 @@ int Recovery_state_transfer::initialize_donor_connection()
                                                        recovery_ssl_crlpath,
                                                        recovery_ssl_verify_server_cert,
                                                        DEFAULT_THREAD_PRIORITY,
-                                                       1, false);
+                                                       1, false, true, true);
 
   if (!error)
   {
@@ -721,7 +721,7 @@ int Recovery_state_transfer::purge_recovery_slave_threads_repos()
                                                   NULL,
                                                   NULL,
                                                   DEFAULT_THREAD_PRIORITY,
-                                                  1, false);
+                                                  1, false, true, true);
 
   DBUG_RETURN(error);
 }

rapid/plugin/group_replication/src/replication_threads_api.cc

@@ -1,4 +1,4 @@
-/* Copyright (c) 2014, 2017, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2014, 2020, 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, version 2.0,
@@ -40,7 +40,9 @@ Replication_thread_api::initialize_channel(char* hostname,
                                            bool  ssl_verify_server_cert,
                                            int   priority,
                                            int   retry_count,
-                                           bool  preserve_logs)
+                                           bool  preserve_logs,
+                                           bool ignore_ws_mem_limit,
+                                           bool allow_drop_write_set)
 {
   DBUG_ENTER("Replication_thread_api::initialize");
   int error= 0;
@@ -67,6 +69,9 @@ Replication_thread_api::initialize_channel(char* hostname,
 
   info.preserve_relay_logs= preserve_logs;
 
+  info.m_ignore_write_set_memory_limit = ignore_ws_mem_limit;
+  info.m_allow_drop_write_set = allow_drop_write_set;
+
   if( use_ssl || ssl_ca != NULL || ssl_capath != NULL || ssl_cert != NULL ||
       ssl_cipher!= NULL || ssl_key != NULL || ssl_crl != NULL ||
       ssl_crlpath != NULL || ssl_verify_server_cert)

rapid/plugin/group_replication/tests/mtr/r/gr_transaction_limit.result

@@ -9,13 +9,14 @@ Note	####	Storing MySQL user name or password information in the master info rep
 [connection server2]
 set session sql_log_bin=0;
 call mtr.add_suppression("Run function 'before_commit' in plugin 'group_replication' failed");
-call mtr.add_suppression("Error on session .*. Transaction of size .* exceeds specified limit 1. To increase the limit please adjust group_replication_transaction_size_limit option.");
+call mtr.add_suppression("Error on session .*. Transaction of size .* exceeds specified limit 16. To increase the limit please adjust group_replication_transaction_size_limit option.");
+call mtr.add_suppression("Error on session .*. Transaction of size .* exceeds specified limit 1600. To increase the limit please adjust group_replication_transaction_size_limit option.");
 set session sql_log_bin=1;
 
 [connection server3]
 set session sql_log_bin=0;
 call mtr.add_suppression("Run function 'before_commit' in plugin 'group_replication' failed");
-call mtr.add_suppression("Error on session .*. Transaction of size .* exceeds specified limit 1024. To increase the limit please adjust group_replication_transaction_size_limit option.");
+call mtr.add_suppression("Error on session .*. Transaction of size .* exceeds specified limit 1600. To increase the limit please adjust group_replication_transaction_size_limit option.");
 set session sql_log_bin=1;
 
 include/rpl_gr_wait_for_number_of_members.inc
@@ -169,9 +170,12 @@ include/assert.inc ['Checking the number of records in table t1']
 ## 3.6 Delete all data from table t1 on server 2,
 ##     transaction should be rolledback
 [connection server2]
+SET @group_replication_transaction_size_limit_save= @@GLOBAL.group_replication_transaction_size_limit;
+SET GLOBAL group_replication_transaction_size_limit= 1600;
 DELETE FROM t1;
 ERROR HY000: Error on observer while running replication hook 'before_commit'.
 include/rpl_sync.inc
+SET GLOBAL group_replication_transaction_size_limit= @group_replication_transaction_size_limit_save;
 
 [connection server1]
 include/assert.inc ['Checking the number of records in table t1']
@@ -185,9 +189,12 @@ include/assert.inc ['Checking the number of records in table t1']
 ## 3.7 Delete all data from table t1 on server 3,
 ##     transaction should be rolledback
 [connection server3]
+SET @group_replication_transaction_size_limit_save= @@GLOBAL.group_replication_transaction_size_limit;
+SET GLOBAL group_replication_transaction_size_limit= 1600;
 DELETE FROM t1;
 ERROR HY000: Error on observer while running replication hook 'before_commit'.
 include/rpl_sync.inc
+SET GLOBAL group_replication_transaction_size_limit= @group_replication_transaction_size_limit_save;
 
 [connection server1]
 include/assert.inc ['Checking the number of records in table t1']