Merge tulin@bk-internal.mysql.com:/home/bk/mysql-5.1-new (475acacb) · Commits · Software / OSDI20 Artifacts / mariadb

mysql-test/extra/rpl_tests/rpl_row_blob.test

+6 −0

Original line number	Diff line number	Diff line
		@@ -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

sql/ha_ndbcluster.cc

+73 −20

Original line number	Diff line number	Diff line
		@@ -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

sql/ha_ndbcluster_binlog.cc

+103 −29

Original line number	Diff line number	Diff line
		@@ -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,33 +776,51 @@ 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)
		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)
		{
		ndbcluster_find_all_files(thd);
		done_find_all_files= 1;
		}
		ndbcluster_create_schema_table(thd);
		// always make sure we create the 'schema' first
		if (!schema_share)
		return;
		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;
		}
		}
		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;
		}

		/*
		Defines and struct for schema table.
		@@ -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= "<no 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 @@ int ndbcluster_log_schema_op(THD thd, NDB_SHARE share,
		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 (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;
		}
		if (ndb_extra_logging)
		sql_print_information("NDB Binlog: cluster failure for %s.", share->key);
		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 */

sql/ha_ndbcluster_binlog.h

+4 −2

Original line number	Diff line number	Diff line
		@@ -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,