Merge from mysql-trunk of:

BUG#35949017 Schema dist setup lockup
Bug#35948153 Problem setting up events due to stale NdbApi dictionary cache [#2]
Bug#35948153 Problem setting up events due to stale NdbApi dictionary cache [#1]
Bug#32550019 Missing check for ndb_schema_result leads to schema dist timeout

Change-Id: I4a32197992bf8b6899892f21587580788f828f34

mysql-test/suite/ndb_ddl/check_util_table_drop.inc

@@ -26,17 +26,15 @@ RESET BINARY LOGS AND GTIDS;
 --echo # Drop mysql.$util_table_name from NDB
 --exec $NDB_DROP_TABLE -d mysql $util_table_name
 
---connection mysqld1
---echo # mysqld1: Wait until ready again...
---source include/ndb_not_readonly.inc
-
---connection mysqld3
---echo # mysqld3: Wait until ready again...
---source include/ndb_not_readonly.inc
-
---connection mysqld5
---echo # mysqld5: Wait until ready again...
---source include/ndb_not_readonly.inc
+# Wait for all servers connected again
+let $i = 1;
+while($i <= $NUM_MYSQLDS)
+{
+  --connection mysqld$i
+  --echo # mysqld$i: Wait until ready again...
+  --source include/ndb_not_readonly.inc
+  inc $i;
+}
 
 --connection mysqld5
 # NOTE! Neither drop of ndb_schema or ndb_schema_result should have a LOST_EVENT

mysql-test/suite/ndb_ddl/util_tables_drop.result

@@ -14,8 +14,11 @@ Dropping table ndb_schema...OK
 NDBT_ProgramExit: 0 - OK
 
 # mysqld1: Wait until ready again...
+# mysqld2: Wait until ready again...
 # mysqld3: Wait until ready again...
+# mysqld4: Wait until ready again...
 # mysqld5: Wait until ready again...
+# mysqld6: Wait until ready again...
 # Check if a LOST_EVENTS event was written to the binlog
 include/show_binlog_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
@@ -44,8 +47,11 @@ Dropping table ndb_apply_status...OK
 NDBT_ProgramExit: 0 - OK
 
 # mysqld1: Wait until ready again...
+# mysqld2: Wait until ready again...
 # mysqld3: Wait until ready again...
+# mysqld4: Wait until ready again...
 # mysqld5: Wait until ready again...
+# mysqld6: Wait until ready again...
 # Check if a LOST_EVENTS event was written to the binlog
 include/show_binlog_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
@@ -76,8 +82,11 @@ Dropping table ndb_schema_result...OK
 NDBT_ProgramExit: 0 - OK
 
 # mysqld1: Wait until ready again...
+# mysqld2: Wait until ready again...
 # mysqld3: Wait until ready again...
+# mysqld4: Wait until ready again...
 # mysqld5: Wait until ready again...
+# mysqld6: Wait until ready again...
 # Check if a LOST_EVENTS event was written to the binlog
 include/show_binlog_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info

storage/ndb/include/ndbapi/Ndb.hpp

@@ -2215,6 +2215,9 @@ class Ndb {
 
   static const char *externalizeTableName(const char *internalTableName,
                                           bool fullyQualifiedNames);
+  static BaseString internalize_table_name(const char *db_name,
+                                           const char *schema,
+                                           const char *table_name);
   const BaseString internalize_table_name(const char *external_name) const;
 
   static const char *externalizeIndexName(const char *internalIndexName,

storage/ndb/include/ndbapi/NdbDictionary.hpp

@@ -2917,6 +2917,9 @@ class NdbDictionary {
     int removeIndexGlobal(const Index &ndbidx, int invalidate) const;
     int removeTableGlobal(const Table &ndbtab, int invalidate) const;
     void invalidateDbGlobal(const char *dbname);
+    // Invalidate table by name if it exist in the cache
+    void invalidateTableGlobal(const char *dbName, const char *schemaName,
+                               const char *tableName);
 #endif
 
     /*

storage/ndb/plugin/ha_ndbcluster_binlog.cc

@@ -1159,7 +1159,8 @@ bool Ndb_schema_dist_client::log_schema_op_impl(
     }
 
     // Check if schema distribution is still ready.
-    if (m_share->have_event_operation() == false) {
+    if (m_share->have_event_operation() == false ||
+        m_result_share->have_event_operation() == false) {
       // This case is unlikely, but there is small race between
       // clients first check for schema distribution ready and schema op
       // registered in the coordinator(since the message is passed
@@ -1221,19 +1222,9 @@ static void ndbcluster_binlog_event_operation_teardown(THD *thd, Ndb *is_ndb,
       Ndb_event_data::get_event_data(pOp->getCustomData());
   NDB_SHARE *const share = event_data->share;
 
-  // Invalidate any cached NdbApi table if object version is lower
-  // than what was used when setting up the NdbEventOperation
-  // NOTE! This functionality need to be explained further
-  {
-    Thd_ndb *thd_ndb = get_thd_ndb(thd);
-    Ndb *ndb = thd_ndb->ndb;
-    Ndb_table_guard ndbtab_g(ndb, share->db, share->table_name);
-    const NDBTAB *ev_tab = pOp->getTable();
-    const NDBTAB *cache_tab = ndbtab_g.get_table();
-    if (cache_tab && cache_tab->getObjectId() == ev_tab->getObjectId() &&
-        cache_tab->getObjectVersion() <= ev_tab->getObjectVersion())
-      ndbtab_g.invalidate();
-  }
+  // Since table has been dropped or cluster connection lost the NdbApi table
+  // should be invalidated in the global dictionary cache
+  Ndb_table_guard::invalidate_table(is_ndb, share->db, share->table_name);
 
   // Close the table in MySQL Server
   ndb_tdc_close_cached_table(thd, share->db, share->table_name);
@@ -2270,9 +2261,8 @@ class Ndb_schema_event_handler {
   void ndbapi_invalidate_table(const std::string &db_name,
                                const std::string &table_name) const {
     DBUG_TRACE;
-    Ndb_table_guard ndbtab_g(m_thd_ndb->ndb, db_name.c_str(),
-                             table_name.c_str());
-    ndbtab_g.invalidate();
+    Ndb_table_guard::invalidate_table(m_thd_ndb->ndb, db_name.c_str(),
+                                      table_name.c_str());
   }
 
   NDB_SHARE *acquire_reference(const std::string &db, const std::string &name,
@@ -3163,6 +3153,8 @@ class Ndb_schema_event_handler {
     if (schema->node_id == own_nodeid()) return;
 
     write_schema_op_to_binlog(m_thd, schema);
+    ndbapi_invalidate_table(schema->db, schema->name);
+    ndb_tdc_close_cached_table(m_thd, schema->db.c_str(), schema->name.c_str());
 
     if (!create_table_from_engine(schema->db.c_str(), schema->name.c_str(),
                                   true, /* force_overwrite */
@@ -4221,14 +4213,9 @@ class Ndb_schema_event_handler {
     // take the GSL properly
     assert(!m_thd_ndb->check_option(Thd_ndb::IS_SCHEMA_DIST_PARTICIPANT));
 
-    // Sleep here will make other mysql server in same cluster setup to create
-    // the schema result table in NDB before this mysql server. This also makes
-    // the create table in the connection thread to acquire GSL before the
-    // Binlog thread
-    DBUG_EXECUTE_IF("ndb_bi_sleep_before_gsl", sleep(1););
     // Protect the setup with GSL(Global Schema Lock)
     Ndb_global_schema_lock_guard global_schema_lock_guard(m_thd);
-    if (global_schema_lock_guard.lock()) {
+    if (!global_schema_lock_guard.try_lock()) {
       ndb_log_info(" - failed to lock GSL");
       return true;
     }
@@ -5003,6 +4990,12 @@ static int ndbcluster_setup_binlog_for_share(THD *thd, Ndb *ndb,
         return -1;
       }
     }
+    // The function that check if event exist will silently mark the NDB table
+    // definition as 'Invalid' when the event's table version does not match the
+    // cached NDB table definitions version. This indicates that the caller have
+    // used a stale version of the NDB table definition and is a problem which
+    // has to be fixed by the caller of this function.
+    assert(ndbtab->getObjectStatus() != NdbDictionary::Object::Invalid);
 
     if (share->have_event_operation()) {
       DBUG_PRINT("info", ("binlogging already setup"));

storage/ndb/plugin/ndb_dd_sync.cc

@@ -1187,6 +1187,9 @@ bool Ndb_dd_sync::synchronize_table(const char *schema_name,
                                     const char *table_name) const {
   ndb_log_verbose(1, "Synchronizing table '%s.%s'", schema_name, table_name);
 
+  // Invalidate potentially stale cached table
+  Ndb_table_guard::invalidate_table(m_thd_ndb->ndb, schema_name, table_name);
+
   Ndb_table_guard ndbtab_g(m_thd_ndb->ndb, schema_name, table_name);
   const NdbDictionary::Table *ndbtab = ndbtab_g.get_table();
   if (!ndbtab) {

storage/ndb/plugin/ndb_metadata_sync.cc

@@ -913,8 +913,8 @@ bool Ndb_metadata_sync::sync_table(THD *thd, const std::string &schema_name,
                  table_name.c_str());
 
     // Invalidate the table in NdbApi
-    Ndb_table_guard ndbtab_guard(ndb, schema_name.c_str(), table_name.c_str());
-    ndbtab_guard.invalidate();
+    Ndb_table_guard::invalidate_table(ndb, schema_name.c_str(),
+                                      table_name.c_str());
     return true;
   }
 

storage/ndb/plugin/ndb_schema_dist.cc

@@ -139,6 +139,19 @@ bool Ndb_schema_dist_client::prepare(const char *db, const char *tabname) {
     return false;
   }
 
+  // Acquire reference also on mysql.ndb_schema_result
+  m_result_share = NDB_SHARE::acquire_reference(
+      Ndb_schema_result_table::DB_NAME.c_str(),
+      Ndb_schema_result_table::TABLE_NAME.c_str(), m_share_reference.c_str());
+  if (m_result_share == nullptr ||
+      m_result_share->have_event_operation() == false) {
+    // The mysql.ndb_schema_result hasn't been created or not setup yet ->
+    // schema distribution is not ready
+    push_warning(m_thd, Sql_condition::SL_WARNING, ER_GET_ERRMSG,
+                 "Schema distribution is not ready (ndb_schema_result)");
+    return false;
+  }
+
   if (unlikely(m_ddl_blocked)) {
     // If a data node gets upgraded after this MySQL Server is upgraded, this
     // MySQL Server will not be aware of the upgrade due to Bug#30930132.
@@ -287,6 +300,10 @@ Ndb_schema_dist_client::~Ndb_schema_dist_client() {
     // Release the reference to mysql.ndb_schema table
     NDB_SHARE::release_reference(m_share, m_share_reference.c_str());
   }
+  if (m_result_share) {
+    // Release the reference to mysql.ndb_schema_result table
+    NDB_SHARE::release_reference(m_result_share, m_share_reference.c_str());
+  }
 
   if (m_thd_ndb) {
     // Inform Applier that one schema distribution has completed
@@ -374,9 +391,9 @@ bool Ndb_schema_dist_client::log_schema_op(const char *query,
     return false;
   }
 
-  // Require that m_share has been initialized to reference the
-  // schema distribution table
+  // Require that references to schema distribution tables has been initialized
   ndbcluster::ndbrequire(m_share);
+  ndbcluster::ndbrequire(m_result_share);
 
   // Check that prepared keys match
   if (!m_prepared_keys.check_key(db, table_name)) {

storage/ndb/plugin/ndb_schema_dist.h

@@ -130,6 +130,7 @@ class Ndb_schema_dist_client {
   class THD *const m_thd;
   class Thd_ndb *const m_thd_ndb;
   struct NDB_SHARE *m_share{nullptr};
+  struct NDB_SHARE *m_result_share{nullptr};
   const std::string m_share_reference;
   class Prepared_keys {
     using Key = std::pair<std::string, std::string>;

storage/ndb/plugin/ndb_table_guard.h

@@ -186,6 +186,18 @@ class Ndb_table_guard {
   }
 
   const NdbError &getNdbError() const { return m_ndberror; }
+
+  /**
+     @brief Invalidate table in the global dictionary cache (if it exists)
+     without fetching table from NDB on cache miss, this saves one roundtrip.
+
+     This function is to be used to invalidate table when it's not already being
+     held open.
+  */
+  static void invalidate_table(Ndb *ndb, const char *dbname,
+                               const char *tabname) {
+    ndb->getDictionary()->invalidateTableGlobal(dbname, "def", tabname);
+  }
 };
 
 #endif

storage/ndb/src/ndbapi/DictCache.cpp

@@ -360,6 +360,35 @@ void GlobalDictCache::invalidateDb(const char *name, size_t len) {
   DBUG_VOID_RETURN;
 }
 
+void GlobalDictCache::invalidateTable(const BaseString &internal_name) {
+  DBUG_TRACE;
+  DBUG_PRINT("enter", ("internal_name: '%s'", internal_name.c_str()));
+  Vector<TableVersion> *vers =
+      m_tableHash.getData(internal_name.c_str(), internal_name.length());
+  if (vers == nullptr) {
+    DBUG_PRINT("exit", ("nothing cached"));
+    return;
+  }
+
+  if (vers->size() == 0) {
+    DBUG_PRINT("exit", ("no cached versions"));
+    return;
+  }
+
+  TableVersion *const ver = &vers->back();
+  if (ver->m_status == RETREIVING) {
+    DBUG_PRINT("exit", ("fresh version on the way"));
+    return;
+  }
+
+  ver->m_impl->m_status = NdbDictionary::Object::Invalid;
+  ver->m_status = DROPPED;
+  if (ver->m_refCount == 0) {
+    delete ver->m_impl;
+    vers->erase(vers->size() - 1);
+  }
+}
+
 void GlobalDictCache::release(const NdbTableImpl *tab, int invalidate) {
   DBUG_ENTER("GlobalDictCache::release");
   DBUG_PRINT("enter",

storage/ndb/src/ndbapi/DictCache.hpp

@@ -94,6 +94,7 @@ class GlobalDictCache : public NdbLockable {
   }
 
   void invalidateDb(const char *name, size_t len);
+  void invalidateTable(const BaseString &name);
 
  public:
   enum Status { OK = 0, DROPPED = 1, RETREIVING = 2 };