Bug#33763 Mysqld leaves readonly mode for cluster tables too early - part2

 - Open tables readonly while not properly initialized
 - Close cached tables when coming out of readonly mode
 - Apply basically same patch as for 32275

mysql-test/include/ndb_not_readonly.inc

@@ -11,13 +11,13 @@ let $counter= 600;
 while ($mysql_errno)
 {
   # Table is readonly until the mysqld has connected properly
-  --error 0,ER_NO_SUCH_TABLE,ER_OPEN_AS_READONLY,ER_GET_ERRMSG
+  --error 0,ER_NO_SUCH_TABLE,ER_OPEN_AS_READONLY
   replace into mysql.ndb_apply_status values(0,0,"",0,0);
   if ($mysql_errno)
   {
     if (!$counter)
     {
-      die("Failed while waiting for mysqld to come out of readonly mode");
+      die Failed while waiting for mysqld to come out of readonly mode;
     }
     dec $counter;
     --sleep 0.1

sql/ha_ndbcluster.cc

@@ -5138,6 +5138,7 @@ int ha_ndbcluster::create(const char *name,
           strcmp(m_tabname, NDB_SCHEMA_TABLE) == 0))
     {
       DBUG_PRINT("info", ("Schema distribution table not setup"));
+      DBUG_ASSERT(ndb_schema_share);
       DBUG_RETURN(HA_ERR_NO_CONNECTION);
     }
     single_user_mode = NdbDictionary::Table::SingleUserModeReadWrite;
@@ -5963,6 +5964,7 @@ ha_ndbcluster::delete_table(ha_ndbcluster *h, Ndb *ndb,
   if (!ndb_schema_share)
   {
     DBUG_PRINT("info", ("Schema distribution table not setup"));
+    DBUG_ASSERT(ndb_schema_share);
     DBUG_RETURN(HA_ERR_NO_CONNECTION);
   }
   /* ndb_share reference temporary */
@@ -6144,6 +6146,7 @@ int ha_ndbcluster::delete_table(const char *name)
   if (!ndb_schema_share)
   {
     DBUG_PRINT("info", ("Schema distribution table not setup"));
+    DBUG_ASSERT(ndb_schema_share);
     DBUG_RETURN(HA_ERR_NO_CONNECTION);
   }
 #endif
@@ -6424,8 +6427,11 @@ int ha_ndbcluster::open(const char *name, int mode, uint test_if_locked)
     DBUG_RETURN(res);
   }
 #ifdef HAVE_NDB_BINLOG
-  if (!ndb_binlog_tables_inited && ndb_binlog_running)
+  if (!ndb_binlog_tables_inited)
+  {
     table->db_stat|= HA_READ_ONLY;
+    sql_print_information("table '%s' opened read only", name);
+  }
 #endif
   DBUG_RETURN(0);
 }
@@ -6784,8 +6790,8 @@ static void ndbcluster_drop_database(handlerton *hton, char *path)
   if (!ndb_schema_share)
   {
     DBUG_PRINT("info", ("Schema distribution table not setup"));
+    DBUG_ASSERT(ndb_schema_share);
     DBUG_VOID_RETURN;
-    //DBUG_RETURN(HA_ERR_NO_CONNECTION);
   }
 #endif
   ndbcluster_drop_database_impl(path);

sql/ha_ndbcluster_binlog.cc

@@ -879,12 +879,9 @@ int ndbcluster_setup_binlog_table_shares(THD *thd)
   {
     pthread_mutex_lock(&LOCK_open);
     ndb_binlog_tables_inited= TRUE;
-    if (ndb_binlog_running)
-    {
     if (ndb_extra_logging)
       sql_print_information("NDB Binlog: ndb tables writable");
     close_cached_tables((THD*) 0, 0, (TABLE_LIST*) 0, TRUE);
-    }
     pthread_mutex_unlock(&LOCK_open);
     /* Signal injector thread that all is setup */
     pthread_cond_signal(&injector_cond);
@@ -2029,6 +2026,7 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
                                ndb_schema_share->use_count));
       free_share(&ndb_schema_share);
       ndb_schema_share= 0;
+      ndb_binlog_tables_inited= 0;
       pthread_mutex_unlock(&ndb_schema_share_mutex);
       /* end protect ndb_schema_share */
 
@@ -3226,6 +3224,7 @@ ndb_binlog_thread_handle_non_data_event(THD *thd, Ndb *ndb,
                                share->key, share->use_count));
       free_share(&ndb_apply_status_share);
       ndb_apply_status_share= 0;
+      ndb_binlog_tables_inited= 0;
     }
     DBUG_PRINT("error", ("CLUSTER FAILURE EVENT: "
                         "%s  received share: 0x%lx  op: 0x%lx  share op: 0x%lx  "
@@ -3245,6 +3244,7 @@ ndb_binlog_thread_handle_non_data_event(THD *thd, Ndb *ndb,
                                share->key, share->use_count));
       free_share(&ndb_apply_status_share);
       ndb_apply_status_share= 0;
+      ndb_binlog_tables_inited= 0;
     }
     /* ToDo: remove printout */
     if (ndb_extra_logging)
@@ -4220,6 +4220,7 @@ err:
                              ndb_schema_share->use_count));
     free_share(&ndb_schema_share);
     ndb_schema_share= 0;
+    ndb_binlog_tables_inited= 0;
     pthread_mutex_unlock(&ndb_schema_share_mutex);
     /* end protect ndb_schema_share */
   }