Check for certification keys

sql/CMakeLists.txt

@@ -34,6 +34,7 @@ IF(WITH_WSREP AND NOT EMBEDDED_LIBRARY)
     wsrep_thd.cc
     wsrep_schema.cc
     wsrep_plugin.cc
+    wsrep_key_check.cc
     service_wsrep.cc
   )
   MYSQL_ADD_PLUGIN(wsrep ${WSREP_SOURCES} MANDATORY NOT_EMBEDDED LINK_LIBRARIES wsrep-lib wsrep_api_v26)

sql/wsrep_key_check.cc

@@ -0,0 +1,48 @@
+#include "wsrep_key_check.h"
+#include "my_global.h"
+#include "sql_class.h"
+#include "mdl.h"
+
+static int check_key_for_ticket(MDL_ticket *mdl_ticket, void *arg,
+                                bool granted)
+{
+  THD *thd= (THD *) arg;
+  MDL_context *mdl_ctx= mdl_ticket->get_ctx();
+  MDL_key *mdl_key= mdl_ticket->get_key();
+
+  if (!granted)
+    return 0;
+
+  if (mdl_key->mdl_namespace() != MDL_key::TABLE)
+    return 0;
+
+  if (thd != mdl_ctx->get_thd())
+    return 0;
+
+  if (std::strncmp("performance_schema", mdl_key->db_name(),
+                   mdl_key->db_name_length()) == 0)
+    return 0;
+
+  if (std::strncmp("gtid_slave_pos", mdl_key->name(),
+                   mdl_key->name_length()) == 0)
+    return 0;
+
+  wsrep::key key(wsrep::key::shared);
+  key.append_key_part(mdl_key->db_name(), mdl_key->db_name_length());
+  key.append_key_part(mdl_key->name(), mdl_key->name_length());
+
+  if (!thd->wsrep_trx().has_key(key))
+  {
+    WSREP_WARN("No certification key for MDL lock "
+               "db: %s name: %s type: %s query: %s",
+               mdl_key->db_name(), mdl_key->name(),
+               mdl_ticket->get_type_name()->str, thd->query());
+  }
+
+  return 0;
+}
+
+void wsrep_check_keys(THD *thd)
+{
+  mdl_iterate(check_key_for_ticket, thd);
+}

sql/wsrep_key_check.h

@@ -0,0 +1,7 @@
+#ifndef WSREP_KEY_CHECK_H
+#define WSREP_KEY_CHECK_H
+
+class THD;
+void wsrep_check_keys(THD *thd);
+
+#endif

sql/wsrep_trans_observer.h

@@ -23,6 +23,7 @@
 #include "wsrep_thd.h"
 #include "wsrep_binlog.h" /* register/deregister group commit */
 #include "my_dbug.h"
+#include "wsrep_key_check.h"
 
 class THD;
 
@@ -343,6 +344,11 @@ static inline int wsrep_before_commit(THD* thd, bool all)
                    thd->wsrep_trx().ws_meta().gtid(),
                    wsrep_gtid_server.gtid());
     wsrep_register_for_group_commit(thd);
+
+    if (wsrep_thd_is_local(thd))
+    {
+      wsrep_check_keys(thd);
+    }
   }
 
   mysql_mutex_lock(&thd->LOCK_thd_kill);