MW-265 Add support for wsrep_max_ws_rows

Variable wsrep_max_ws_rows limits the number of rows that a transaction
can insert/update/delete.

4 files changed, 41 insertions, 3 deletions

diff --git a/sql/handler.cc b/sql/handler.cc
index e84e1b52ca2..6fa937faa84 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -6098,6 +6098,17 @@ int handler::ha_write_row(uchar *buf)
   rows_changed++;
   if (unlikely(error= binlog_log_row(table, 0, buf, log_func)))
     DBUG_RETURN(error); /* purecov: inspected */
+#ifdef WITH_WSREP
+  current_thd->wsrep_affected_rows++;
+  if (wsrep_max_ws_rows &&
+      current_thd->wsrep_exec_mode != REPL_RECV &&
+      current_thd->wsrep_affected_rows > wsrep_max_ws_rows)
+  {
+    current_thd->transaction_rollback_request= TRUE;
+    my_message(ER_ERROR_DURING_COMMIT, "wsrep_max_ws_rows exceeded", MYF(0));
+    DBUG_RETURN(ER_ERROR_DURING_COMMIT);
+  }
+#endif /* WITH_WSREP */
 
   DEBUG_SYNC_C("ha_write_row_end");
   DBUG_RETURN(0);
@@ -6131,6 +6142,17 @@ int handler::ha_update_row(const uchar *old_data, uchar *new_data)
   rows_changed++;
   if (unlikely(error= binlog_log_row(table, old_data, new_data, log_func)))
     return error;
+#ifdef WITH_WSREP
+  current_thd->wsrep_affected_rows++;
+  if (wsrep_max_ws_rows &&
+      current_thd->wsrep_exec_mode != REPL_RECV &&
+      current_thd->wsrep_affected_rows > wsrep_max_ws_rows)
+  {
+    current_thd->transaction_rollback_request= TRUE;
+    my_message(ER_ERROR_DURING_COMMIT, "wsrep_max_ws_rows exceeded", MYF(0));
+    return ER_ERROR_DURING_COMMIT;
+  }
+#endif /* WITH_WSREP */
   return 0;
 }
 
@@ -6158,6 +6180,17 @@ int handler::ha_delete_row(const uchar *buf)
   rows_changed++;
   if (unlikely(error= binlog_log_row(table, buf, 0, log_func)))
     return error;
+#ifdef WITH_WSREP
+  current_thd->wsrep_affected_rows++;
+  if (wsrep_max_ws_rows &&
+      current_thd->wsrep_exec_mode != REPL_RECV &&
+      current_thd->wsrep_affected_rows > wsrep_max_ws_rows)
+  {
+    current_thd->transaction_rollback_request= TRUE;
+    my_message(ER_ERROR_DURING_COMMIT, "wsrep_max_ws_rows exceeded", MYF(0));
+    return ER_ERROR_DURING_COMMIT;
+  }
+#endif /* WITH_WSREP */
   return 0;
 }
 
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 5705694208a..4873586aba5 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -1213,7 +1213,8 @@ THD::THD()
   wsrep_mysql_replicated  = 0;
   wsrep_TOI_pre_query     = NULL;
   wsrep_TOI_pre_query_len = 0;
-  wsrep_sync_wait_gtid= WSREP_GTID_UNDEFINED;
+  wsrep_sync_wait_gtid    = WSREP_GTID_UNDEFINED;
+  wsrep_affected_rows     = 0;
 #endif
   /* Call to init() below requires fully initialized Open_tables_state. */
   reset_open_tables_state(this);
@@ -1629,7 +1630,8 @@ void THD::init(void)
   wsrep_mysql_replicated  = 0;
   wsrep_TOI_pre_query     = NULL;
   wsrep_TOI_pre_query_len = 0;
-  wsrep_sync_wait_gtid= WSREP_GTID_UNDEFINED;
+  wsrep_sync_wait_gtid    = WSREP_GTID_UNDEFINED;
+  wsrep_affected_rows     = 0;
 
   /*
     @@wsrep_causal_reads is now being handled via wsrep_sync_wait, update it
@@ -2383,6 +2385,8 @@ void THD::cleanup_after_query()
 
 #ifdef WITH_WSREP
   wsrep_sync_wait_gtid= WSREP_GTID_UNDEFINED;
+  if (!in_active_multi_stmt_transaction())
+    wsrep_affected_rows= 0;
 #endif /* WITH_WSREP */
 
   DBUG_VOID_RETURN;
diff --git a/sql/sql_class.h b/sql/sql_class.h
index b9f0c0a0ae7..9ee5a40dc99 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -3878,6 +3878,7 @@ public:
   bool                      wsrep_apply_toi; /* applier processing in TOI */
   bool                      wsrep_skip_append_keys;
   wsrep_gtid_t              wsrep_sync_wait_gtid;
+  ulong                     wsrep_affected_rows;
 #endif /* WITH_WSREP */
 };
 
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index c49a5de75e3..0bfa0bf5eb0 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -4694,7 +4694,7 @@ static Sys_var_ulong Sys_wsrep_max_ws_size (
 static Sys_var_ulong Sys_wsrep_max_ws_rows (
        "wsrep_max_ws_rows", "Max number of rows in write set",
        GLOBAL_VAR(wsrep_max_ws_rows), CMD_LINE(REQUIRED_ARG),
-       VALID_RANGE(1, 1048576), DEFAULT(131072), BLOCK_SIZE(1));
+       VALID_RANGE(0, 1048576), DEFAULT(0), BLOCK_SIZE(1));
 
 static Sys_var_charptr Sys_wsrep_notify_cmd(
        "wsrep_notify_cmd", "",