MDEV-9519: Data corruption will happen on the Galera cluster size change

If we have a 2+ node cluster which is replicating from an async master
and the binlog_format is set to STATEMENT and multi-row inserts are executed
on a table with an auto_increment column such that values are automatically
generated by MySQL, then the server node generates wrong auto_increment
values, which are different from what was generated on the async master.

In the title of the MDEV-9519 it was proposed to ban start slave on a Galera
if master binlog_format = statement and wsrep_auto_increment_control = 1,
but the problem can be solved without such a restriction.

The causes and fixes:

1. We need to improve processing of changing the auto-increment values
after changing the cluster size.

2. If wsrep auto_increment_control switched on during operation of
the node, then we should immediately update the auto_increment_increment
and auto_increment_offset global variables, without waiting of the next
invocation of the wsrep_view_handler_cb() callback. In the current version
these variables retain its initial values if wsrep_auto_increment_control
is switched on during operation of the node, which leads to inconsistent
results on the different nodes in some scenarios.

3. If wsrep auto_increment_control switched off during operation of the node,
then we must return the original values of the auto_increment_increment and
auto_increment_offset global variables, as the user has set. To make this
possible, we need to add a "shadow copies" of these variables (which stores
the latest values set by the user).

https://jira.mariadb.org/browse/MDEV-9519

17 files changed, 392 insertions, 81 deletions

diff --git a/include/mysql/service_wsrep.h b/include/mysql/service_wsrep.h
index b51f154422f..267c8cb4e90 100644
--- a/include/mysql/service_wsrep.h
+++ b/include/mysql/service_wsrep.h
@@ -107,6 +107,7 @@ extern struct wsrep_service_st {
   bool                        (*wsrep_thd_ignore_table_func)(THD *thd);
   long long                   (*wsrep_thd_trx_seqno_func)(THD *thd);
   struct wsrep_ws_handle *    (*wsrep_thd_ws_handle_func)(THD *thd);
+  void                        (*wsrep_thd_auto_increment_variables_func)(THD *thd, unsigned long long *offset, unsigned long long *increment);
   int                         (*wsrep_trx_is_aborting_func)(MYSQL_THD thd);
   int                         (*wsrep_trx_order_before_func)(MYSQL_THD, MYSQL_THD);
   void                        (*wsrep_unlock_rollback_func)();
@@ -149,6 +150,7 @@ extern struct wsrep_service_st {
 #define wsrep_thd_ignore_table(T) wsrep_service->wsrep_thd_ignore_table_func(T)
 #define wsrep_thd_trx_seqno(T) wsrep_service->wsrep_thd_trx_seqno_func(T)
 #define wsrep_thd_ws_handle(T) wsrep_service->wsrep_thd_ws_handle_func(T)
+#define wsrep_thd_auto_increment_variables(T,O,I) wsrep_service->wsrep_thd_auto_increment_variables_func(T,O,I)
 #define wsrep_trx_is_aborting(T) wsrep_service->wsrep_trx_is_aborting_func(T)
 #define wsrep_trx_order_before(T1,T2) wsrep_service->wsrep_trx_order_before_func(T1,T2)
 #define wsrep_unlock_rollback() wsrep_service->wsrep_unlock_rollback_func()
@@ -201,6 +203,7 @@ my_bool wsrep_thd_is_BF(MYSQL_THD thd, my_bool sync);
 my_bool wsrep_thd_is_wsrep(MYSQL_THD thd);
 struct wsrep *get_wsrep();
 struct wsrep_ws_handle *wsrep_thd_ws_handle(THD *thd);
+void wsrep_thd_auto_increment_variables(THD *thd, unsigned long long *offset, unsigned long long *increment);
 void wsrep_aborting_thd_enqueue(THD *thd);
 void wsrep_lock_rollback();
 void wsrep_post_commit(THD* thd, bool all);
diff --git a/mysql-test/suite/galera/disabled.def b/mysql-test/suite/galera/disabled.def
index 1d8f8ad5949..d0bc20b0de7 100644
--- a/mysql-test/suite/galera/disabled.def
+++ b/mysql-test/suite/galera/disabled.def
@@ -25,7 +25,6 @@ galera.MW-329 : wsrep_local_replays not stable
 MW-416 : MDEV-13549 Galera test failures
 MW-388 : MDEV-13549 Galera test failures
 galera_sst_mysqldump_with_key : MDEV-16890 Galera test failure
-galera_binlog_stmt_autoinc: MDEV-17106 Test failure on galera.galera_binlog_stmt_autoinc
 galera_gc_fc_limit : MDEV-17061 Test failure on galera.galera_gc_fc_limit
 galera_as_slave_replication_budle : MDEV-15785 Test case galera_as_slave_replication_bundle caused debug assertion
 galera_wan : MDEV-17259: Test failure on galera.galera_wan
diff --git a/mysql-test/suite/galera/r/galera_binlog_stmt_autoinc.result b/mysql-test/suite/galera/r/galera_binlog_stmt_autoinc.result
index 8e8b79b168f..542bd156816 100644
--- a/mysql-test/suite/galera/r/galera_binlog_stmt_autoinc.result
+++ b/mysql-test/suite/galera/r/galera_binlog_stmt_autoinc.result
@@ -8,20 +8,20 @@ PRIMARY KEY (i)
 insert into t1(i) values(null);
 select * from t1;
 i	c
-3	dummy_text
+1	dummy_text
 insert into t1(i) values(null), (null), (null);
 select * from t1;
 i	c
+1	dummy_text
 3	dummy_text
 5	dummy_text
 7	dummy_text
-9	dummy_text
 select * from t1;
 i	c
+1	dummy_text
 3	dummy_text
 5	dummy_text
 7	dummy_text
-9	dummy_text
 SET GLOBAL wsrep_forced_binlog_format='none';
 SET GLOBAL wsrep_forced_binlog_format='none';
 drop table t1;
@@ -40,20 +40,20 @@ PRIMARY KEY (i)
 insert into t1(i) values(null);
 select * from t1;
 i	c
-4	dummy_text
+1	dummy_text
 insert into t1(i) values(null), (null), (null);
 select * from t1;
 i	c
+1	dummy_text
 4	dummy_text
 7	dummy_text
 10	dummy_text
-13	dummy_text
 select * from t1;
 i	c
+1	dummy_text
 4	dummy_text
 7	dummy_text
 10	dummy_text
-13	dummy_text
 SET GLOBAL wsrep_auto_increment_control='ON';
 SET SESSION binlog_format='ROW';
 show variables like 'binlog_format';
@@ -67,7 +67,7 @@ wsrep_auto_increment_control	ON
 SET GLOBAL wsrep_auto_increment_control='OFF';
 show variables like '%auto_increment%';
 Variable_name	Value
-auto_increment_increment	2
+auto_increment_increment	3
 auto_increment_offset	1
 wsrep_auto_increment_control	OFF
 SET GLOBAL wsrep_auto_increment_control='ON';
@@ -82,20 +82,20 @@ PRIMARY KEY (i)
 insert into t1(i) values(null);
 select * from t1;
 i	c
-3	dummy_text
+1	dummy_text
 insert into t1(i) values(null), (null), (null);
 select * from t1;
 i	c
+1	dummy_text
 3	dummy_text
 5	dummy_text
 7	dummy_text
-9	dummy_text
 select * from t1;
 i	c
+1	dummy_text
 3	dummy_text
 5	dummy_text
 7	dummy_text
-9	dummy_text
 SET GLOBAL wsrep_forced_binlog_format='none';
 SET GLOBAL wsrep_forced_binlog_format='none';
 drop table t1;
@@ -114,20 +114,20 @@ PRIMARY KEY (i)
 insert into t1(i) values(null);
 select * from t1;
 i	c
-4	dummy_text
+1	dummy_text
 insert into t1(i) values(null), (null), (null);
 select * from t1;
 i	c
+1	dummy_text
 4	dummy_text
 7	dummy_text
 10	dummy_text
-13	dummy_text
 select * from t1;
 i	c
+1	dummy_text
 4	dummy_text
 7	dummy_text
 10	dummy_text
-13	dummy_text
 SET GLOBAL wsrep_auto_increment_control='ON';
 show variables like 'binlog_format';
 Variable_name	Value
@@ -140,7 +140,7 @@ wsrep_auto_increment_control	ON
 SET GLOBAL wsrep_auto_increment_control='OFF';
 show variables like '%auto_increment%';
 Variable_name	Value
-auto_increment_increment	2
+auto_increment_increment	3
 auto_increment_offset	1
 wsrep_auto_increment_control	OFF
 SET GLOBAL wsrep_auto_increment_control='ON';
diff --git a/mysql-test/suite/parts/r/partition_auto_increment_max.result b/mysql-test/suite/parts/r/partition_auto_increment_max.result
new file mode 100644
index 00000000000..65a3900e8e6
--- /dev/null
+++ b/mysql-test/suite/parts/r/partition_auto_increment_max.result
@@ -0,0 +1,7 @@
+CREATE TABLE t1 (pk INT AUTO_INCREMENT PRIMARY KEY) PARTITION BY KEY (pk) PARTITIONS 2;
+INSERT INTO t1 VALUES (NULL),(NULL);
+UPDATE t1 SET pk = 2147483647;
+ERROR 23000: Duplicate entry '2147483647' for key 'PRIMARY'
+REPLACE INTO t1 VALUES (NULL);
+ERROR 22003: Out of range value for column 'pk' at row 1
+DROP TABLE t1;
diff --git a/mysql-test/suite/parts/t/partition_auto_increment_max.test b/mysql-test/suite/parts/t/partition_auto_increment_max.test
new file mode 100644
index 00000000000..74e6139131d
--- /dev/null
+++ b/mysql-test/suite/parts/t/partition_auto_increment_max.test
@@ -0,0 +1,12 @@
+--source include/have_partition.inc
+--source include/have_log_bin.inc
+
+CREATE TABLE t1 (pk INT AUTO_INCREMENT PRIMARY KEY) PARTITION BY KEY (pk) PARTITIONS 2;
+INSERT INTO t1 VALUES (NULL),(NULL);
+
+--error ER_DUP_ENTRY
+UPDATE t1 SET pk = 2147483647;
+--error HA_ERR_AUTOINC_ERANGE
+REPLACE INTO t1 VALUES (NULL);
+
+DROP TABLE t1;
diff --git a/sql/field.h b/sql/field.h
index c21bba6936f..86853f7d9d9 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -1329,6 +1329,17 @@ public:
   /* Hash value */
   virtual void hash(ulong *nr, ulong *nr2);
 
+  /**
+    Get the upper limit of the MySQL integral and floating-point type.
+
+    @return maximum allowed value for the field
+  */
+  virtual ulonglong get_max_int_value() const
+  {
+    DBUG_ASSERT(false);
+    return 0ULL;
+  }
+
 /**
   Checks whether a string field is part of write_set.
 
@@ -1796,6 +1807,11 @@ public:
     *to= *from;
     return from + 1;
   }
+
+  virtual ulonglong get_max_int_value() const
+  {
+    return unsigned_flag ? 0xFFULL : 0x7FULL;
+  }
 };
 
 
@@ -1837,6 +1853,11 @@ public:
   virtual const uchar *unpack(uchar* to, const uchar *from,
                               const uchar *from_end, uint param_data)
   { return unpack_int16(to, from, from_end); }
+
+  virtual ulonglong get_max_int_value() const
+  {
+    return unsigned_flag ? 0xFFFFULL : 0x7FFFULL;
+  }
 };
 
 class Field_medium :public Field_num {
@@ -1870,6 +1891,11 @@ public:
   {
     return Field::pack(to, from, max_length);
   }
+
+  virtual ulonglong get_max_int_value() const
+  {
+    return unsigned_flag ? 0xFFFFFFULL : 0x7FFFFFULL;
+  }
 };
 
 
@@ -1915,6 +1941,11 @@ public:
   {
     return unpack_int32(to, from, from_end);
   }
+
+  virtual ulonglong get_max_int_value() const
+  {
+    return unsigned_flag ? 0xFFFFFFFFULL : 0x7FFFFFFFULL;
+  }
 };
 
 
@@ -1964,6 +1995,10 @@ public:
   {
     return unpack_int64(to, from, from_end);
   }
+  virtual ulonglong get_max_int_value() const
+  {
+    return unsigned_flag ? 0xFFFFFFFFFFFFFFFFULL : 0x7FFFFFFFFFFFFFFFULL;
+  }
 };
 
 
@@ -1997,6 +2032,13 @@ public:
   uint32 pack_length() const { return sizeof(float); }
   uint row_pack_length() const { return pack_length(); }
   void sql_type(String &str) const;
+  virtual ulonglong get_max_int_value() const
+  {
+    /*
+      We use the maximum as per IEEE754-2008 standard, 2^24
+    */
+    return 0x1000000ULL;
+  }
 private:
   int do_save_field_metadata(uchar *first_byte);
 };
@@ -2039,6 +2081,13 @@ public:
   uint32 pack_length() const { return sizeof(double); }
   uint row_pack_length() const { return pack_length(); }
   void sql_type(String &str) const;
+  virtual ulonglong get_max_int_value() const
+  {
+    /*
+      We use the maximum as per IEEE754-2008 standard, 2^53
+    */
+    return 0x20000000000000ULL;
+  }
 private:
   int do_save_field_metadata(uchar *first_byte);
 };
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index d0ae8f47007..2167bea8d7c 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -8706,31 +8706,37 @@ void ha_partition::release_auto_increment()
       m_file[i]->ha_release_auto_increment();
     }
   }
-  else if (next_insert_id)
+  else
   {
-    ulonglong next_auto_inc_val;
     lock_auto_increment();
-    next_auto_inc_val= part_share->next_auto_inc_val;
-    /*
-      If the current auto_increment values is lower than the reserved
-      value, and the reserved value was reserved by this thread,
-      we can lower the reserved value.
-    */
-    if (next_insert_id < next_auto_inc_val &&
-        auto_inc_interval_for_cur_row.maximum() >= next_auto_inc_val)
+    if (next_insert_id)
     {
-      THD *thd= ha_thd();
+      ulonglong next_auto_inc_val= part_share->next_auto_inc_val;
       /*
-        Check that we do not lower the value because of a failed insert
-        with SET INSERT_ID, i.e. forced/non generated values.
+        If the current auto_increment values is lower than the reserved
+        value, and the reserved value was reserved by this thread,
+        we can lower the reserved value.
       */
-      if (thd->auto_inc_intervals_forced.maximum() < next_insert_id)
-        part_share->next_auto_inc_val= next_insert_id;
+      if (next_insert_id < next_auto_inc_val &&
+          auto_inc_interval_for_cur_row.maximum() >= next_auto_inc_val)
+      {
+        THD *thd= ha_thd();
+        /*
+          Check that we do not lower the value because of a failed insert
+          with SET INSERT_ID, i.e. forced/non generated values.
+        */
+        if (thd->auto_inc_intervals_forced.maximum() < next_insert_id)
+          part_share->next_auto_inc_val= next_insert_id;
+      }
+      DBUG_PRINT("info", ("part_share->next_auto_inc_val: %lu",
+                          (ulong) part_share->next_auto_inc_val));
     }
-    DBUG_PRINT("info", ("part_share->next_auto_inc_val: %lu",
-                        (ulong) part_share->next_auto_inc_val));
-
-    /* Unlock the multi row statement lock taken in get_auto_increment */
+    /*
+      Unlock the multi-row statement lock taken in get_auto_increment.
+      These actions must be performed even if the next_insert_id field
+      contains zero, otherwise if the update_auto_increment fails then
+      an unnecessary lock will remain:
+    */
     if (auto_increment_safe_stmt_log_lock)
     {
       auto_increment_safe_stmt_log_lock= FALSE;
diff --git a/sql/handler.cc b/sql/handler.cc
index 497409d0f53..ba3f7f3e16a 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -2854,11 +2854,17 @@ compute_next_insert_id(ulonglong nr,struct system_variables *variables)
     nr= nr + 1; // optimization of the formula below
   else
   {
-    nr= (((nr+ variables->auto_increment_increment -
-           variables->auto_increment_offset)) /
-         (ulonglong) variables->auto_increment_increment);
-    nr= (nr* (ulonglong) variables->auto_increment_increment +
-         variables->auto_increment_offset);
+    /*
+       Calculating the number of complete auto_increment_increment extents:
+    */
+    nr= (nr + variables->auto_increment_increment -
+         variables->auto_increment_offset) /
+        (ulonglong) variables->auto_increment_increment;
+    /*
+       Adding an offset to the auto_increment_increment extent boundary:
+    */
+    nr= nr * (ulonglong) variables->auto_increment_increment +
+        variables->auto_increment_offset;
   }
 
   if (unlikely(nr <= save_nr))
@@ -2912,8 +2918,14 @@ prev_insert_id(ulonglong nr, struct system_variables *variables)
   }
   if (variables->auto_increment_increment == 1)
     return nr; // optimization of the formula below
-  nr= (((nr - variables->auto_increment_offset)) /
-       (ulonglong) variables->auto_increment_increment);
+  /*
+     Calculating the number of complete auto_increment_increment extents:
+  */
+  nr= (nr - variables->auto_increment_offset) /
+      (ulonglong) variables->auto_increment_increment;
+  /*
+     Adding an offset to the auto_increment_increment extent boundary:
+  */
   return (nr * (ulonglong) variables->auto_increment_increment +
           variables->auto_increment_offset);
 }
@@ -3004,7 +3016,7 @@ int handler::update_auto_increment()
   bool append= FALSE;
   THD *thd= table->in_use;
   struct system_variables *variables= &thd->variables;
-  int result=0, tmp;
+  int tmp;
   enum enum_check_fields save_count_cuted_fields;
   DBUG_ENTER("handler::update_auto_increment");
 
@@ -3135,10 +3147,23 @@ int handler::update_auto_increment()
   if (unlikely(tmp))                            // Out of range value in store
   {
     /*
-      It's better to return an error here than getting a confusing
-      'duplicate key error' later.
+      First, test if the query was aborted due to strict mode constraints
+      or new field value greater than maximum integer value:
+    */
+    if (thd->killed == KILL_BAD_DATA ||
+        nr > table->next_number_field->get_max_int_value())
+      DBUG_RETURN(HA_ERR_AUTOINC_ERANGE);
+    /*
+      Field refused this value (overflow) and truncated it, use the result
+      of the truncation (which is going to be inserted); however we try to
+      decrease it to honour auto_increment_* variables.
+      That will shift the left bound of the reserved interval, we don't
+      bother shifting the right bound (anyway any other value from this
+      interval will cause a duplicate key).
     */
-    result= HA_ERR_AUTOINC_ERANGE;
+    nr= prev_insert_id(table->next_number_field->val_int(), variables);
+    if (unlikely(table->next_number_field->store((longlong)nr, TRUE)))
+      nr= table->next_number_field->val_int();
   }
   if (append)
   {
@@ -3163,9 +3188,6 @@ int handler::update_auto_increment()
   */
   insert_id_for_cur_row= nr;
 
-  if (result)                                   // overflow
-    DBUG_RETURN(result);
-
   /*
     Set next insert id to point to next auto-increment value to be able to
     handle multi-row statements.
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 5452f7164e6..813f0988b74 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -4563,6 +4563,20 @@ static int init_common_variables()
     return 1;
   }
 
+#ifdef WITH_WSREP
+  /*
+    We need to initialize auxiliary variables, that will be
+    further keep the original values of auto-increment options
+    as they set by the user. These variables used to restore
+    user-defined values of the auto-increment options after
+    setting of the wsrep_auto_increment_control to 'OFF'.
+  */
+  global_system_variables.saved_auto_increment_increment=
+    global_system_variables.auto_increment_increment;
+  global_system_variables.saved_auto_increment_offset=
+    global_system_variables.auto_increment_offset;
+#endif /* WITH_WSREP */
+
   return 0;
 }
 
diff --git a/sql/sql_class.h b/sql/sql_class.h
index fae4a683370..8dde68b7fba 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -550,6 +550,16 @@ typedef struct system_variables
   ha_rows max_join_size;
   ha_rows expensive_subquery_limit;
   ulong auto_increment_increment, auto_increment_offset;
+#ifdef WITH_WSREP
+  /*
+    Stored values of the auto_increment_increment and auto_increment_offset
+    that are will be restored when wsrep_auto_increment_control will be set
+    to 'OFF', because the setting it to 'ON' leads to overwriting of the
+    original values (which are set by the user) by calculated ones (which
+    are based on the cluster size):
+  */
+  ulong saved_auto_increment_increment, saved_auto_increment_offset;
+#endif /* WITH_WSREP */
   ulong lock_wait_timeout;
   ulong join_cache_level;
   ulong max_allowed_packet;
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index de5c3b4c409..d695c586529 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1862,7 +1862,6 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
             info->deleted++;
           else
             error= 0;
-          thd->record_first_successful_insert_id_in_cur_stmt(table->file->insert_id_for_cur_row);
           /*
             Since we pretend that we have done insert we should call
             its after triggers.
@@ -1903,7 +1902,6 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
     if (table->file->insert_id_for_cur_row == 0)
       table->file->insert_id_for_cur_row= insert_id_for_cur_row;
       
-    thd->record_first_successful_insert_id_in_cur_stmt(table->file->insert_id_for_cur_row);
     /*
       Restore column maps if they where replaced during an duplicate key
       problem.
diff --git a/sql/sql_plugin_services.ic b/sql/sql_plugin_services.ic
index 95301a5fbe8..427d8937c57 100644
--- a/sql/sql_plugin_services.ic
+++ b/sql/sql_plugin_services.ic
@@ -177,6 +177,7 @@ static struct wsrep_service_st wsrep_handler = {
   wsrep_thd_ignore_table,
   wsrep_thd_trx_seqno,
   wsrep_thd_ws_handle,
+  wsrep_thd_auto_increment_variables,
   wsrep_trx_is_aborting,
   wsrep_trx_order_before,
   wsrep_unlock_rollback,
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index d1db14d7b35..223015e81c2 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -345,13 +345,56 @@ static Sys_var_long Sys_pfs_connect_attrs_size(
 
 #endif /* WITH_PERFSCHEMA_STORAGE_ENGINE */
 
+#ifdef WITH_WSREP
+
+/*
+  We need to keep the original values set by the user, as they will
+  be lost if wsrep_auto_increment_control set to 'ON':
+*/
+static bool update_auto_increment_increment (sys_var *self, THD *thd, enum_var_type type)
+{
+  if (type == OPT_GLOBAL)
+    global_system_variables.saved_auto_increment_increment=
+      global_system_variables.auto_increment_increment;
+  else
+    thd->variables.saved_auto_increment_increment=
+      thd->variables.auto_increment_increment;
+  return false;
+}
+
+#endif /* WITH_WSREP */
+
 static Sys_var_ulong Sys_auto_increment_increment(
        "auto_increment_increment",
        "Auto-increment columns are incremented by this",
        SESSION_VAR(auto_increment_increment),
        CMD_LINE(OPT_ARG),
        VALID_RANGE(1, 65535), DEFAULT(1), BLOCK_SIZE(1),
+#ifdef WITH_WSREP
+       NO_MUTEX_GUARD, IN_BINLOG, ON_CHECK(0),
+       ON_UPDATE(update_auto_increment_increment));
+#else
        NO_MUTEX_GUARD, IN_BINLOG);
+#endif /* WITH_WSREP */
+
+#ifdef WITH_WSREP
+
+/*
+  We need to keep the original values set by the user, as they will
+  be lost if wsrep_auto_increment_control set to 'ON':
+*/
+static bool update_auto_increment_offset (sys_var *self, THD *thd, enum_var_type type)
+{
+  if (type == OPT_GLOBAL)
+    global_system_variables.saved_auto_increment_offset=
+      global_system_variables.auto_increment_offset;
+  else
+    thd->variables.saved_auto_increment_offset=
+      thd->variables.auto_increment_offset;
+  return false;
+}
+
+#endif /* WITH_WSREP */
 
 static Sys_var_ulong Sys_auto_increment_offset(
        "auto_increment_offset",
@@ -360,7 +403,12 @@ static Sys_var_ulong Sys_auto_increment_offset(
        SESSION_VAR(auto_increment_offset),
        CMD_LINE(OPT_ARG),
        VALID_RANGE(1, 65535), DEFAULT(1), BLOCK_SIZE(1),
+#ifdef WITH_WSREP
+       NO_MUTEX_GUARD, IN_BINLOG, ON_CHECK(0),
+       ON_UPDATE(update_auto_increment_offset));
+#else
        NO_MUTEX_GUARD, IN_BINLOG);
+#endif /* WITH_WSREP */
 
 static Sys_var_mybool Sys_automatic_sp_privileges(
        "automatic_sp_privileges",
@@ -4851,11 +4899,54 @@ static Sys_var_ulong Sys_wsrep_retry_autocommit(
        SESSION_VAR(wsrep_retry_autocommit), CMD_LINE(REQUIRED_ARG),
        VALID_RANGE(0, 10000), DEFAULT(1), BLOCK_SIZE(1));
 
+static bool update_wsrep_auto_increment_control (sys_var *self, THD *thd, enum_var_type type)
+{
+  if (wsrep_auto_increment_control)
+  {
+    /*
+      The variables that control auto increment shall be calculated
+      automaticaly based on the size of the cluster. This usually done
+      within the wsrep_view_handler_cb callback. However, if the user
+      manually sets the value of wsrep_auto_increment_control to 'ON',
+      then we should to re-calculate these variables again (because
+      these values may be required before wsrep_view_handler_cb will
+      be re-invoked, which is rarely invoked if the cluster stays in
+      the stable state):
+    */
+    global_system_variables.auto_increment_increment=
+       wsrep_cluster_size ? wsrep_cluster_size : 1;
+    global_system_variables.auto_increment_offset=
+       wsrep_local_index >= 0 ? wsrep_local_index + 1 : 1;
+    thd->variables.auto_increment_increment=
+      global_system_variables.auto_increment_increment;
+    thd->variables.auto_increment_offset=
+      global_system_variables.auto_increment_offset;
+  }
+  else
+  {
+    /*
+      We must restore the last values of the variables that
+      are explicitly specified by the user:
+    */
+    global_system_variables.auto_increment_increment=
+      global_system_variables.saved_auto_increment_increment;
+    global_system_variables.auto_increment_offset=
+      global_system_variables.saved_auto_increment_offset;
+    thd->variables.auto_increment_increment=
+      thd->variables.saved_auto_increment_increment;
+    thd->variables.auto_increment_offset=
+      thd->variables.saved_auto_increment_offset;
+  }
+  return false;
+}
+
 static Sys_var_mybool Sys_wsrep_auto_increment_control(
        "wsrep_auto_increment_control", "To automatically control the "
        "assignment of autoincrement variables",
        GLOBAL_VAR(wsrep_auto_increment_control), 
-       CMD_LINE(OPT_ARG), DEFAULT(TRUE));
+       CMD_LINE(OPT_ARG), DEFAULT(TRUE),
+       NO_MUTEX_GUARD, NOT_IN_BINLOG, ON_CHECK(0),
+       ON_UPDATE(update_wsrep_auto_increment_control));
 
 static Sys_var_mybool Sys_wsrep_drupal_282555_workaround(
        "wsrep_drupal_282555_workaround", "Enable a workaround to handle the "
diff --git a/sql/wsrep_dummy.cc b/sql/wsrep_dummy.cc
index 5837ab4bed5..795e2d19252 100644
--- a/sql/wsrep_dummy.cc
+++ b/sql/wsrep_dummy.cc
@@ -125,6 +125,14 @@ longlong wsrep_thd_trx_seqno(THD *)
 struct wsrep_ws_handle* wsrep_thd_ws_handle(THD *)
 { return 0; }
 
+void wsrep_thd_auto_increment_variables(THD *thd,
+                                        unsigned long long *offset,
+                                        unsigned long long *increment)
+{
+  *offset= thd->variables.auto_increment_offset;
+  *increment= thd->variables.auto_increment_increment;
+}
+
 int wsrep_trx_is_aborting(THD *)
 { return 0; }
 
diff --git a/sql/wsrep_thd.cc b/sql/wsrep_thd.cc
index eb26da61282..551e710cfeb 100644
--- a/sql/wsrep_thd.cc
+++ b/sql/wsrep_thd.cc
@@ -676,3 +676,24 @@ bool wsrep_thd_has_explicit_locks(THD *thd)
   assert(thd);
   return thd->mdl_context.has_explicit_locks();
 }
+
+/*
+  Get auto increment variables for THD. Use global settings for
+  applier threads.
+ */
+void wsrep_thd_auto_increment_variables(THD* thd,
+                                        unsigned long long* offset,
+                                        unsigned long long* increment)
+{
+  if (thd->wsrep_exec_mode == REPL_RECV &&
+      thd->wsrep_conflict_state != REPLAYING)
+  {
+    *offset= global_system_variables.auto_increment_offset;
+    *increment= global_system_variables.auto_increment_increment;
+  }
+  else
+  {
+    *offset= thd->variables.auto_increment_offset;
+    *increment= thd->variables.auto_increment_increment;
+  }
+}
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 472fb86288f..80ff2e02e13 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -8324,8 +8324,8 @@ no_commit:
 
 		/* We need the upper limit of the col type to check for
 		whether we update the table autoinc counter or not. */
-		col_max_value = innobase_get_int_col_max_value(
-			table->next_number_field);
+		col_max_value =
+			table->next_number_field->get_max_int_value();
 
 		/* Get the value that MySQL attempted to store in the table.*/
 		auto_inc = table->next_number_field->val_uint();
@@ -8402,15 +8402,33 @@ set_max_autoinc:
 				/* This should filter out the negative
 				values set explicitly by the user. */
 				if (auto_inc <= col_max_value) {
-					ut_a(prebuilt->autoinc_increment > 0);
 
 					ulonglong	offset;
 					ulonglong	increment;
 					dberr_t		err;
 
-					offset = prebuilt->autoinc_offset;
-					increment = prebuilt->autoinc_increment;
-
+#ifdef WITH_WSREP
+					/* Applier threads which are processing
+					ROW events and don't go through server
+					level autoinc processing, therefore
+					prebuilt autoinc values don't get
+					properly assigned. Fetch values from
+					server side. */
+					if (wsrep_on(user_thd) &&
+					    wsrep_thd_exec_mode(user_thd) == REPL_RECV)
+					{
+					    wsrep_thd_auto_increment_variables(
+						user_thd, &offset, &increment);
+					}
+					else
+					{
+#endif /* WITH_WSREP */
+					    ut_a(prebuilt->autoinc_increment > 0);
+					    offset = prebuilt->autoinc_offset;
+					    increment = prebuilt->autoinc_increment;
+#ifdef WITH_WSREP
+					}
+#endif /* WITH_WSREP */
 					auto_inc = innobase_next_autoinc(
 						auto_inc,
 						1, increment, offset,
@@ -8925,17 +8943,35 @@ ha_innobase::update_row(
 
 		/* We need the upper limit of the col type to check for
 		whether we update the table autoinc counter or not. */
-		col_max_value = innobase_get_int_col_max_value(
-			table->next_number_field);
+		col_max_value =
+		   table->next_number_field->get_max_int_value();
 
 		if (auto_inc <= col_max_value && auto_inc != 0) {
 
 			ulonglong	offset;
 			ulonglong	increment;
 
-			offset = prebuilt->autoinc_offset;
-			increment = prebuilt->autoinc_increment;
-
+#ifdef WITH_WSREP
+			/* Applier threads which are processing
+			ROW events and don't go through server
+			level autoinc processing, therefore
+			prebuilt autoinc values don't get
+			properly assigned. Fetch values from
+			server side. */
+			if (wsrep_on(user_thd) &&
+			    wsrep_thd_exec_mode(user_thd) == REPL_RECV)
+			{
+			    wsrep_thd_auto_increment_variables(
+				user_thd, &offset, &increment);
+			}
+			else
+			{
+#endif /* WITH_WSREP */
+			    offset = prebuilt->autoinc_offset;
+			    increment = prebuilt->autoinc_increment;
+#ifdef WITH_WSREP
+			}
+#endif /* WITH_WSREP */
 			auto_inc = innobase_next_autoinc(
 				auto_inc, 1, increment, offset, col_max_value);
 
@@ -16040,12 +16076,11 @@ ha_innobase::get_auto_increment(
 				    current, autoinc);
 			if (!wsrep_on(ha_thd()))
 			{
-			current = autoinc - prebuilt->autoinc_increment;
+				current = autoinc - prebuilt->autoinc_increment;
+				current = innobase_next_autoinc(
+					current, 1, increment, offset, col_max_value);
 			}
 
-			current = innobase_next_autoinc(
-				current, 1, increment, offset, col_max_value);
-
 			dict_table_autoinc_initialize(prebuilt->table, current);
 
 			*first_value = current;
diff --git a/storage/xtradb/handler/ha_innodb.cc b/storage/xtradb/handler/ha_innodb.cc
index c9db941b4bf..4ed9f644b45 100644
--- a/storage/xtradb/handler/ha_innodb.cc
+++ b/storage/xtradb/handler/ha_innodb.cc
@@ -8911,8 +8911,8 @@ no_commit:
 
 		/* We need the upper limit of the col type to check for
 		whether we update the table autoinc counter or not. */
-		col_max_value = innobase_get_int_col_max_value(
-			table->next_number_field);
+		col_max_value =
+			table->next_number_field->get_max_int_value();
 
 		/* Get the value that MySQL attempted to store in the table.*/
 		auto_inc = table->next_number_field->val_uint();
@@ -8989,15 +8989,33 @@ set_max_autoinc:
 				/* This should filter out the negative
 				values set explicitly by the user. */
 				if (auto_inc <= col_max_value) {
-					ut_a(prebuilt->autoinc_increment > 0);
 
 					ulonglong	offset;
 					ulonglong	increment;
 					dberr_t		err;
 
-					offset = prebuilt->autoinc_offset;
-					increment = prebuilt->autoinc_increment;
-
+#ifdef WITH_WSREP
+					/* Applier threads which are processing
+					ROW events and don't go through server
+					level autoinc processing, therefore
+					prebuilt autoinc values don't get
+					properly assigned. Fetch values from
+					server side. */
+					if (wsrep_on(user_thd) &&
+					    wsrep_thd_exec_mode(user_thd) == REPL_RECV)
+					{
+					    wsrep_thd_auto_increment_variables(
+						user_thd, &offset, &increment);
+					}
+					else
+					{
+#endif /* WITH_WSREP */
+					    ut_a(prebuilt->autoinc_increment > 0);
+					    offset = prebuilt->autoinc_offset;
+					    increment = prebuilt->autoinc_increment;
+#ifdef WITH_WSREP
+					}
+#endif /* WITH_WSREP */
 					auto_inc = innobase_next_autoinc(
 						auto_inc,
 						1, increment, offset,
@@ -9509,17 +9527,35 @@ ha_innobase::update_row(
 
 		/* We need the upper limit of the col type to check for
 		whether we update the table autoinc counter or not. */
-		col_max_value = innobase_get_int_col_max_value(
-			table->next_number_field);
+		col_max_value =
+			table->next_number_field->get_max_int_value();
 
 		if (auto_inc <= col_max_value && auto_inc != 0) {
 
 			ulonglong	offset;
 			ulonglong	increment;
 
-			offset = prebuilt->autoinc_offset;
-			increment = prebuilt->autoinc_increment;
-
+#ifdef WITH_WSREP
+			/* Applier threads which are processing
+			ROW events and don't go through server
+			level autoinc processing, therefore
+			prebuilt autoinc values don't get
+			properly assigned. Fetch values from
+			server side. */
+			if (wsrep_on(user_thd) &&
+			    wsrep_thd_exec_mode(user_thd) == REPL_RECV)
+			{
+			    wsrep_thd_auto_increment_variables(
+				user_thd, &offset, &increment);
+			}
+			else
+			{
+#endif /* WITH_WSREP */
+			    offset = prebuilt->autoinc_offset;
+			    increment = prebuilt->autoinc_increment;
+#ifdef WITH_WSREP
+			}
+#endif /* WITH_WSREP */
 			auto_inc = innobase_next_autoinc(
 				auto_inc, 1, increment, offset, col_max_value);
 
@@ -16744,12 +16780,11 @@ ha_innobase::get_auto_increment(
 				    current, autoinc);
 			if (!wsrep_on(ha_thd()))
 			{
-			current = autoinc - prebuilt->autoinc_increment;
+				current = autoinc - prebuilt->autoinc_increment;
+				current = innobase_next_autoinc(
+					current, 1, increment, offset, col_max_value);
 			}
 
-			current = innobase_next_autoinc(
-				current, 1, increment, offset, col_max_value);
-
 			dict_table_autoinc_initialize(prebuilt->table, current);
 
 			*first_value = current;