Add a new config variable wsrep_certification_rules

This is used for controlling whether to use a new/optimized
certification rules or the old/classic ones that could cause more
certification failures - when foreign keys are used and two INSERTs are
done concurrently to the child table from different nodes.

(cherry picked from commit 815d73e6af8daace6262ab63ca6c043ffc4204b3)

sql/sys_vars.cc

@@ -3887,6 +3887,19 @@ static Sys_var_mybool Sys_wsrep_certify_nonPK(
        GLOBAL_VAR(wsrep_certify_nonPK), 
        CMD_LINE(OPT_ARG), DEFAULT(TRUE));
 
+static const char *wsrep_certification_rules_names[]= { "strict", "optimized", NullS };
+static Sys_var_enum Sys_wsrep_certification_rules(
+       "wsrep_certification_rules",
+       "Certification rules to use in the cluster. Possible values are: "
+       "\"strict\": stricter rules that could result in more certification "
+       "failures. "
+       "\"optimized\": relaxed rules that allow more concurrency and "
+       "cause less certification failures.",
+       READ_ONLY GLOBAL_VAR(wsrep_certification_rules), CMD_LINE(REQUIRED_ARG),
+       wsrep_certification_rules_names, DEFAULT(WSREP_CERTIFICATION_RULES_STRICT),
+       NO_MUTEX_GUARD, NOT_IN_BINLOG, ON_CHECK(0),
+       ON_UPDATE(0));
+
 static Sys_var_mybool Sys_wsrep_causal_reads(
        "wsrep_causal_reads", "(DEPRECATED) Setting this variable is equivalent "
        "to setting wsrep_sync_wait READ flag",

sql/wsrep_mysqld.cc

@@ -51,6 +51,7 @@ ulong   wsrep_max_ws_size              = 1073741824UL;//max ws (RBR buffer) size
 ulong   wsrep_max_ws_rows              = 65536; // max number of rows in ws
 int     wsrep_to_isolation             = 0; // # of active TO isolation threads
 my_bool wsrep_certify_nonPK            = 1; // certify, even when no primary key
+ulong   wsrep_certification_rules      = WSREP_CERTIFICATION_RULES_STRICT;
 long    wsrep_max_protocol_version     = 3; // maximum protocol version to use
 ulong   wsrep_forced_binlog_format     = BINLOG_FORMAT_UNSPEC;
 my_bool wsrep_recovery                 = 0; // recovery

sql/wsrep_mysqld.h

@@ -20,6 +20,7 @@
 typedef struct st_mysql_show_var SHOW_VAR;
 #include <sql_priv.h>
 #include "../wsrep/wsrep_api.h"
+#include "wsrep_mysqld_c.h"
 
 #define WSREP_UNDEFINED_TRX_ID ULONGLONG_MAX
 
@@ -58,7 +59,6 @@ enum wsrep_consistency_check_mode {
     CONSISTENCY_CHECK_RUNNING,
 };
 
-
 // Global wsrep parameters
 extern wsrep_t*    wsrep;
 

storage/innobase/row/row0ins.c

@@ -58,6 +58,7 @@ Created 4/20/1996 Heikki Tuuri
 
 #ifdef WITH_WSREP
 #include "../../../wsrep/wsrep_api.h"
+#include "wsrep_mysqld_c.h"
 #endif /* WITH_WSREP */
 
 /*************************************************************************
@@ -1435,15 +1436,26 @@ row_ins_check_foreign_constraint(
 				if (check_ref) {
 					err = DB_SUCCESS;
 #ifdef WITH_WSREP
+					enum wsrep_key_type key_type = WSREP_KEY_EXCLUSIVE;
+					if (upd_node != NULL) {
+						key_type = WSREP_KEY_SHARED;
+					} else {
+						switch (wsrep_certification_rules) {
+						case WSREP_CERTIFICATION_RULES_STRICT:
+							key_type = WSREP_KEY_EXCLUSIVE;
+							break;
+						case WSREP_CERTIFICATION_RULES_OPTIMIZED:
+							key_type = WSREP_KEY_SEMI;
+							break;
+						}
+					}
 					err = wsrep_append_foreign_key(
 						thr_get_trx(thr),
 						foreign,
 						rec,
 						check_index,
 						check_ref,
-						upd_node != NULL
-							? WSREP_KEY_SHARED
-							: WSREP_KEY_EXCLUSIVE);
+						key_type);
 #endif /* WITH_WSREP */
 					goto end_scan;
 				} else if (foreign->type != 0) {

storage/xtradb/row/row0ins.c

@@ -58,6 +58,7 @@ Created 4/20/1996 Heikki Tuuri
 
 #ifdef WITH_WSREP
 #include "../../../wsrep/wsrep_api.h"
+#include "wsrep_mysqld_c.h"
 #endif /* WITH_WSREP */
 
 /*************************************************************************