diff --git a/mysql-test/extra/rpl_tests/rpl_row_blob.test b/mysql-test/extra/rpl_tests/rpl_row_blob.test index 14449e0a018..75a2dc2cf9e 100644 --- a/mysql-test/extra/rpl_tests/rpl_row_blob.test +++ b/mysql-test/extra/rpl_tests/rpl_row_blob.test @@ -183,3 +183,9 @@ connection master; DROP TABLE IF EXISTS test.t1; DROP TABLE IF EXISTS test.t2; +# ensure cleanup on slave as well: +# ndb blob tables consist of several tables +# if cluster is shutdown while not all tables are +# properly dropped, the table becomes inconsistent +# and wrecks later test cases +--sync_slave_with_master diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc index a4193e4eb33..570ac48238c 100644 --- a/sql/ha_ndbcluster.cc +++ b/sql/ha_ndbcluster.cc @@ -4480,6 +4480,21 @@ int ha_ndbcluster::create(const char *name, DBUG_RETURN(my_errno); } +#ifdef HAVE_NDB_BINLOG + /* + Don't allow table creation unless + schema distribution table is setup + ( unless it is a creation of the schema dist table itself ) + */ + if (!schema_share && + !(strcmp(m_dbname, NDB_REP_DB) == 0 && + strcmp(m_tabname, NDB_SCHEMA_TABLE) == 0)) + { + DBUG_PRINT("info", ("Schema distribution table not setup")); + DBUG_RETURN(HA_ERR_NO_CONNECTION); + } +#endif /* HAVE_NDB_BINLOG */ + DBUG_PRINT("table", ("name: %s", m_tabname)); tab.setName(m_tabname); tab.setLogging(!(info->options & HA_LEX_CREATE_TMP_TABLE)); @@ -5004,7 +5019,8 @@ int ha_ndbcluster::rename_table(const char *from, const char *to) is_old_table_tmpfile= 0; String event_name(INJECTOR_EVENT_LEN); ndb_rep_event_name(&event_name, from + sizeof(share_prefix) - 1, 0); - ndbcluster_handle_drop_table(ndb, event_name.c_ptr(), share); + ndbcluster_handle_drop_table(ndb, event_name.c_ptr(), share, + "rename table"); } if (!result && !IS_TMP_PREFIX(new_tabname)) @@ -5088,6 +5104,15 @@ ha_ndbcluster::delete_table(ha_ndbcluster *h, Ndb *ndb, DBUG_ENTER("ha_ndbcluster::ndbcluster_delete_table"); NDBDICT *dict= ndb->getDictionary(); #ifdef HAVE_NDB_BINLOG + /* + Don't allow drop table unless + schema distribution table is setup + */ + if (!schema_share) + { + DBUG_PRINT("info", ("Schema distribution table not setup")); + DBUG_RETURN(HA_ERR_NO_CONNECTION); + } NDB_SHARE *share= get_share(path, 0, false); #endif @@ -5156,7 +5181,7 @@ ha_ndbcluster::delete_table(ha_ndbcluster *h, Ndb *ndb, ndb_rep_event_name(&event_name, path + sizeof(share_prefix) - 1, 0); ndbcluster_handle_drop_table(ndb, table_dropped ? event_name.c_ptr() : 0, - share); + share, "delete table"); } if (share) @@ -5185,6 +5210,18 @@ int ha_ndbcluster::delete_table(const char *name) set_dbname(name); set_tabname(name); +#ifdef HAVE_NDB_BINLOG + /* + Don't allow drop table unless + schema distribution table is setup + */ + if (!schema_share) + { + DBUG_PRINT("info", ("Schema distribution table not setup")); + DBUG_RETURN(HA_ERR_NO_CONNECTION); + } +#endif + if (check_ndb_connection()) DBUG_RETURN(HA_ERR_NO_CONNECTION); @@ -5406,6 +5443,11 @@ int ha_ndbcluster::open(const char *name, int mode, uint test_if_locked) if (!res) info(HA_STATUS_VARIABLE | HA_STATUS_CONST); +#ifdef HAVE_NDB_BINLOG + if (!ndb_binlog_tables_inited && ndb_binlog_running) + table->db_stat|= HA_READ_ONLY; +#endif + DBUG_RETURN(res); } @@ -5704,6 +5746,19 @@ int ndbcluster_drop_database_impl(const char *path) static void ndbcluster_drop_database(char *path) { + DBUG_ENTER("ndbcluster_drop_database"); +#ifdef HAVE_NDB_BINLOG + /* + Don't allow drop database unless + schema distribution table is setup + */ + if (!schema_share) + { + DBUG_PRINT("info", ("Schema distribution table not setup")); + DBUG_VOID_RETURN; + //DBUG_RETURN(HA_ERR_NO_CONNECTION); + } +#endif ndbcluster_drop_database_impl(path); #ifdef HAVE_NDB_BINLOG char db[FN_REFLEN]; @@ -5712,6 +5767,7 @@ static void ndbcluster_drop_database(char *path) current_thd->query, current_thd->query_length, db, "", 0, 0, SOT_DROP_DB); #endif + DBUG_VOID_RETURN; } /* find all tables in ndb and discover those needed @@ -5740,29 +5796,30 @@ int ndbcluster_find_all_files(THD *thd) NDBDICT *dict= ndb->getDictionary(); - int unhandled, retries= 5; + int unhandled, retries= 5, skipped; do { if (dict->listObjects(list, NdbDictionary::Object::UserTable) != 0) ERR_RETURN(dict->getNdbError()); unhandled= 0; + skipped= 0; + retries--; for (uint i= 0 ; i < list.count ; i++) { NDBDICT::List::Element& elmt= list.elements[i]; - int do_handle_table= 0; if (IS_TMP_PREFIX(elmt.name) || IS_NDB_BLOB_PREFIX(elmt.name)) { DBUG_PRINT("info", ("Skipping %s.%s in NDB", elmt.database, elmt.name)); continue; } DBUG_PRINT("info", ("Found %s.%s in NDB", elmt.database, elmt.name)); - if (elmt.state == NDBOBJ::StateOnline || - elmt.state == NDBOBJ::StateBackup) - do_handle_table= 1; - else if (!(elmt.state == NDBOBJ::StateBuilding)) + if (elmt.state != NDBOBJ::StateOnline && + elmt.state != NDBOBJ::StateBackup && + elmt.state != NDBOBJ::StateBuilding) { sql_print_information("NDB: skipping setup table %s.%s, in state %d", elmt.database, elmt.name, elmt.state); + skipped++; continue; } @@ -5771,7 +5828,7 @@ int ndbcluster_find_all_files(THD *thd) if (!(ndbtab= dict->getTable(elmt.name))) { - if (do_handle_table) + if (retries == 0) sql_print_error("NDB: failed to setup table %s.%s, error: %d, %s", elmt.database, elmt.name, dict->getNdbError().code, @@ -5840,9 +5897,9 @@ int ndbcluster_find_all_files(THD *thd) pthread_mutex_unlock(&LOCK_open); } } - while (unhandled && retries--); + while (unhandled && retries); - DBUG_RETURN(0); + DBUG_RETURN(-(skipped + unhandled)); } int ndbcluster_find_files(THD *thd,const char *db,const char *path, @@ -7706,6 +7763,8 @@ pthread_handler_t ndb_util_thread_func(void *arg __attribute__((unused))) pthread_cond_wait(&COND_server_started, &LOCK_server_started); pthread_mutex_unlock(&LOCK_server_started); + ndbcluster_util_inited= 1; + /* Wait for cluster to start */ @@ -7737,6 +7796,8 @@ pthread_handler_t ndb_util_thread_func(void *arg __attribute__((unused))) } #ifdef HAVE_NDB_BINLOG + if (ndb_extra_logging && ndb_binlog_running) + sql_print_information("NDB Binlog: Ndb tables initially read only."); /* create tables needed by the replication */ ndbcluster_setup_binlog_table_shares(thd); #else @@ -7746,17 +7807,9 @@ pthread_handler_t ndb_util_thread_func(void *arg __attribute__((unused))) ndbcluster_find_all_files(thd); #endif - ndbcluster_util_inited= 1; - -#ifdef HAVE_NDB_BINLOG - /* Signal injector thread that all is setup */ - pthread_cond_signal(&injector_cond); -#endif - set_timespec(abstime, 0); for (;!abort_loop;) { - pthread_mutex_lock(&LOCK_ndb_util_thread); pthread_cond_timedwait(&COND_ndb_util_thread, &LOCK_ndb_util_thread, @@ -7774,7 +7827,7 @@ pthread_handler_t ndb_util_thread_func(void *arg __attribute__((unused))) Check that the apply_status_share and schema_share has been created. If not try to create it */ - if (!apply_status_share || !schema_share) + if (!ndb_binlog_tables_inited) ndbcluster_setup_binlog_table_shares(thd); #endif diff --git a/sql/ha_ndbcluster_binlog.cc b/sql/ha_ndbcluster_binlog.cc index 60ccb661703..9e4a7dcdef0 100644 --- a/sql/ha_ndbcluster_binlog.cc +++ b/sql/ha_ndbcluster_binlog.cc @@ -48,6 +48,7 @@ int ndb_binlog_thread_running= 0; FALSE if not */ my_bool ndb_binlog_running= FALSE; +my_bool ndb_binlog_tables_inited= FALSE; /* Global reference to the ndb injector thread THD oject @@ -775,32 +776,50 @@ static int ndbcluster_create_schema_table(THD *thd) DBUG_RETURN(0); } -void ndbcluster_setup_binlog_table_shares(THD *thd) +int ndbcluster_setup_binlog_table_shares(THD *thd) { - int done_find_all_files= 0; if (!schema_share && ndbcluster_check_schema_share() == 0) { - if (!done_find_all_files) - { - ndbcluster_find_all_files(thd); - done_find_all_files= 1; - } - ndbcluster_create_schema_table(thd); - // always make sure we create the 'schema' first + pthread_mutex_lock(&LOCK_open); + ndb_create_table_from_engine(thd, NDB_REP_DB, NDB_SCHEMA_TABLE); + pthread_mutex_unlock(&LOCK_open); if (!schema_share) - return; + { + ndbcluster_create_schema_table(thd); + // always make sure we create the 'schema' first + if (!schema_share) + return 1; + } } if (!apply_status_share && ndbcluster_check_apply_status_share() == 0) { - if (!done_find_all_files) + pthread_mutex_lock(&LOCK_open); + ndb_create_table_from_engine(thd, NDB_REP_DB, NDB_APPLY_TABLE); + pthread_mutex_unlock(&LOCK_open); + if (!apply_status_share) { - ndbcluster_find_all_files(thd); - done_find_all_files= 1; + ndbcluster_create_apply_status_table(thd); + if (!apply_status_share) + return 1; } - ndbcluster_create_apply_status_table(thd); } + if (!ndbcluster_find_all_files(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); + } + return 0; } /* @@ -936,6 +955,31 @@ static char *ndb_pack_varchar(const NDBCOL *col, char *buf, /* log query in schema table */ +static void ndb_report_waiting(const char *key, + int the_time, + const char *op, + const char *obj) +{ + ulonglong ndb_latest_epoch= 0; + const char *proc_info= ""; + pthread_mutex_lock(&injector_mutex); + if (injector_ndb) + ndb_latest_epoch= injector_ndb->getLatestGCI(); + if (injector_thd) + proc_info= injector_thd->proc_info; + pthread_mutex_unlock(&injector_mutex); + sql_print_information("NDB %s:" + " waiting max %u sec for %s %s." + " epochs: (%u,%u,%u)" + " injector proc_info: %s" + ,key, the_time, op, obj + ,(uint)ndb_latest_handled_binlog_epoch + ,(uint)ndb_latest_received_binlog_epoch + ,(uint)ndb_latest_epoch + ,proc_info + ); +} + int ndbcluster_log_schema_op(THD *thd, NDB_SHARE *share, const char *query, int query_length, const char *db, const char *table_name, @@ -965,6 +1009,7 @@ int ndbcluster_log_schema_op(THD *thd, NDB_SHARE *share, } char tmp_buf2[FN_REFLEN]; + const char *type_str; switch (type) { case SOT_DROP_TABLE: @@ -975,6 +1020,7 @@ int ndbcluster_log_schema_op(THD *thd, NDB_SHARE *share, query= tmp_buf2; query_length= (uint) (strxmov(tmp_buf2, "drop table `", table_name, "`", NullS) - tmp_buf2); + type_str= "drop table"; break; case SOT_RENAME_TABLE: /* redo the rename table query as is may contain several tables */ @@ -982,20 +1028,28 @@ int ndbcluster_log_schema_op(THD *thd, NDB_SHARE *share, query_length= (uint) (strxmov(tmp_buf2, "rename table `", old_db, ".", old_table_name, "` to `", db, ".", table_name, "`", NullS) - tmp_buf2); + type_str= "rename table"; break; case SOT_CREATE_TABLE: - // fall through + type_str= "create table"; + break; case SOT_ALTER_TABLE: + type_str= "create table"; break; case SOT_DROP_DB: + type_str= "drop db"; break; case SOT_CREATE_DB: + type_str= "create db"; break; case SOT_ALTER_DB: + type_str= "alter db"; break; case SOT_TABLESPACE: + type_str= "tablespace"; break; case SOT_LOGFILE_GROUP: + type_str= "logfile group"; break; default: abort(); /* should not happen, programming error */ @@ -1201,13 +1255,13 @@ end: max_timeout--; if (max_timeout == 0) { - sql_print_error("NDB create table: timed out. Ignoring..."); + sql_print_error("NDB %s: distibuting %s timed out. Ignoring...", + type_str, ndb_schema_object->key); break; } if (ndb_extra_logging) - sql_print_information("NDB create table: " - "waiting max %u sec for create table %s.", - max_timeout, ndb_schema_object->key); + ndb_report_waiting(type_str, max_timeout, + "distributing", ndb_schema_object->key); } (void) pthread_mutex_unlock(&ndb_schema_object->mutex); } @@ -1689,9 +1743,15 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb, // skip break; case NDBEVENT::TE_CLUSTER_FAILURE: + // fall through case NDBEVENT::TE_DROP: + if (ndb_extra_logging && + ndb_binlog_tables_inited && ndb_binlog_running) + sql_print_information("NDB Binlog: ndb tables initially " + "read only on reconnect."); free_share(&schema_share); schema_share= 0; + ndb_binlog_tables_inited= FALSE; // fall through case NDBEVENT::TE_ALTER: ndb_handle_schema_change(thd, ndb, pOp, tmp_share); @@ -2494,9 +2554,15 @@ ndbcluster_create_event_ops(NDB_SHARE *share, const NDBTAB *ndbtab, get_share(share); if (do_apply_status_share) + { apply_status_share= get_share(share); + (void) pthread_cond_signal(&injector_cond); + } else if (do_schema_share) + { schema_share= get_share(share); + (void) pthread_cond_signal(&injector_cond); + } DBUG_PRINT("info",("%s share->op: 0x%lx, share->use_count: %u", share->key, share->op, share->use_count)); @@ -2513,7 +2579,7 @@ ndbcluster_create_event_ops(NDB_SHARE *share, const NDBTAB *ndbtab, */ int ndbcluster_handle_drop_table(Ndb *ndb, const char *event_name, - NDB_SHARE *share) + NDB_SHARE *share, const char *type_str) { DBUG_ENTER("ndbcluster_handle_drop_table"); @@ -2581,9 +2647,8 @@ ndbcluster_handle_drop_table(Ndb *ndb, const char *event_name, break; } if (ndb_extra_logging) - sql_print_information("NDB delete table: " - "waiting max %u sec for drop table %s.", - max_timeout, share->key); + ndb_report_waiting(type_str, max_timeout, + type_str, share->key); } (void) pthread_mutex_unlock(&share->mutex); #else @@ -2660,13 +2725,18 @@ ndb_binlog_thread_handle_non_data_event(Ndb *ndb, NdbEventOperation *pOp, switch (type) { case NDBEVENT::TE_CLUSTER_FAILURE: - if (apply_status_share == share) - { - free_share(&apply_status_share); - apply_status_share= 0; - } if (ndb_extra_logging) sql_print_information("NDB Binlog: cluster failure for %s.", share->key); + if (apply_status_share == share) + { + if (ndb_extra_logging && + ndb_binlog_tables_inited && ndb_binlog_running) + sql_print_information("NDB Binlog: ndb tables initially " + "read only on reconnect."); + free_share(&apply_status_share); + apply_status_share= 0; + ndb_binlog_tables_inited= FALSE; + } DBUG_PRINT("info", ("CLUSTER FAILURE EVENT: " "%s received share: 0x%lx op: %lx share op: %lx " "op_old: %lx", @@ -2675,8 +2745,13 @@ ndb_binlog_thread_handle_non_data_event(Ndb *ndb, NdbEventOperation *pOp, case NDBEVENT::TE_DROP: if (apply_status_share == share) { + if (ndb_extra_logging && + ndb_binlog_tables_inited && ndb_binlog_running) + sql_print_information("NDB Binlog: ndb tables initially " + "read only on reconnect."); free_share(&apply_status_share); apply_status_share= 0; + ndb_binlog_tables_inited= FALSE; } /* ToDo: remove printout */ if (ndb_extra_logging) @@ -3087,7 +3162,7 @@ pthread_handler_t ndb_binlog_thread_func(void *arg) thd->proc_info= "Waiting for ndbcluster to start"; pthread_mutex_lock(&injector_mutex); - while (!ndbcluster_util_inited) + while (!schema_share || !apply_status_share) { /* ndb not connected yet */ struct timespec abstime; @@ -3119,10 +3194,6 @@ pthread_handler_t ndb_binlog_thread_func(void *arg) thd->db= db; if (ndb_binlog_running) open_binlog_index(thd, &binlog_tables, &binlog_index); - if (!apply_status_share) - { - sql_print_error("NDB: Could not get apply status share"); - } thd->db= db; } @@ -3184,6 +3255,7 @@ pthread_handler_t ndb_binlog_thread_func(void *arg) if (unlikely(schema_res > 0)) { + thd->proc_info= "Processing events from schema table"; schema_ndb-> setReportThreshEventGCISlip(ndb_report_thresh_binlog_epoch_slip); schema_ndb-> @@ -3379,6 +3451,7 @@ pthread_handler_t ndb_binlog_thread_func(void *arg) if (trans.good()) { //DBUG_ASSERT(row.n_inserts || row.n_updates || row.n_deletes); + thd->proc_info= "Committing events to binlog"; injector::transaction::binlog_pos start= trans.start_pos(); if (int r= trans.commit()) { @@ -3418,6 +3491,7 @@ pthread_handler_t ndb_binlog_thread_func(void *arg) } err: DBUG_PRINT("info",("Shutting down cluster binlog thread")); + thd->proc_info= "Shutting down"; close_thread_tables(thd); pthread_mutex_lock(&injector_mutex); /* don't mess with the injector_ndb anymore from other threads */ diff --git a/sql/ha_ndbcluster_binlog.h b/sql/ha_ndbcluster_binlog.h index fda025842a0..9d15016568b 100644 --- a/sql/ha_ndbcluster_binlog.h +++ b/sql/ha_ndbcluster_binlog.h @@ -101,7 +101,8 @@ int ndbcluster_log_schema_op(THD *thd, NDB_SHARE *share, const char *old_db= 0, const char *old_table_name= 0); int ndbcluster_handle_drop_table(Ndb *ndb, const char *event_name, - NDB_SHARE *share); + NDB_SHARE *share, + const char *type_str); void ndb_rep_event_name(String *event_name, const char *db, const char *tbl); int ndb_create_table_from_engine(THD *thd, const char *db, @@ -112,12 +113,13 @@ pthread_handler_t ndb_binlog_thread_func(void *arg); /* table cluster_replication.apply_status */ -void ndbcluster_setup_binlog_table_shares(THD *thd); +int ndbcluster_setup_binlog_table_shares(THD *thd); extern NDB_SHARE *apply_status_share; extern NDB_SHARE *schema_share; extern THD *injector_thd; extern my_bool ndb_binlog_running; +extern my_bool ndb_binlog_tables_inited; bool ndbcluster_show_status_binlog(THD* thd, stat_print_fn *stat_print,