Merge work:/home/bk/mysql (79f12db0) · Commits · Software / OSDI20 Artifacts / mariadb

repl-tests/test-dump/run.test

+8 −0

Original line number	Diff line number	Diff line
		source ../include/master-slave.inc;
		connection slave;
		!slave stop;
		flush slave;
		connection master;
		flush master;
		connection slave;
		slave start;
		connection master;
		use test;
		drop table if exists words;
		@@ -7,6 +14,7 @@ load data infile '/usr/dict/words' into table words;
		drop table if exists words1;
		create table words1 (word char(20) not null);
		load data infile '/usr/dict/words' into table words1;
		sleep 5;
		connection slave;
		use test;
		drop table if exists words;

sql/Makefile.am

+2 −2

Original line number	Diff line number	Diff line
		@@ -48,7 +48,7 @@ noinst_HEADERS = item.h item_func.h item_sum.h item_cmpfunc.h \
		opt_range.h \
		sql_select.h structs.h table.h sql_udf.h hash_filo.h\
		lex.h lex_symbol.h sql_acl.h sql_crypt.h md5.h \
		log_event.h mini_client.h
		log_event.h mini_client.h sql_repl.h
		mysqld_SOURCES = sql_lex.cc \
		item.cc item_sum.cc item_buff.cc item_func.cc \
		item_cmpfunc.cc item_strfunc.cc item_timefunc.cc \
		@@ -69,7 +69,7 @@ mysqld_SOURCES = sql_lex.cc \
		sql_db.cc sql_table.cc sql_rename.cc sql_crypt.cc \
		sql_load.cc mf_iocache.cc field_conv.cc sql_show.cc \
		sql_udf.cc sql_analyse.cc sql_analyse.h sql_cache.cc \
		slave.cc \
		slave.cc sql_repl.cc \
		md5.c log_event.cc mini_client.cc mini_client_errors.c
		gen_lex_hash_SOURCES = gen_lex_hash.cc
		gen_lex_hash_LDADD = $(LDADD) $(CXXLDFLAGS)

sql/log_event.cc

+59 −37

Original line number	Diff line number	Diff line
		@@ -22,16 +22,6 @@
		#include "mysql_priv.h"
		#endif /* MYSQL_CLIENT */

		#define LOG_EVENT_HEADER_LEN 9
		#define QUERY_HEADER_LEN (sizeof(uint32) + sizeof(uint32) + sizeof(uchar))
		#define LOAD_HEADER_LEN (sizeof(uint32) + sizeof(uint32) + \
		+ sizeof(uint32) + 2 + sizeof(uint32))
		#define EVENT_LEN_OFFSET 5
		#define EVENT_TYPE_OFFSET 4
		#define MAX_EVENT_LEN 410241024
		#define QUERY_EVENT_OVERHEAD LOG_EVENT_HEADER_LEN+QUERY_HEADER_LEN
		#define ROTATE_EVENT_OVERHEAD LOG_EVENT_HEADER_LEN
		#define LOAD_EVENT_OVERHEAD (LOG_EVENT_HEADER_LEN+LOAD_HEADER_LEN+sizeof(sql_ex_info))

		static void pretty_print_char(FILE* file, int c)
		{
		@@ -71,6 +61,8 @@ int Log_event::write_header(FILE* file)
		int4store(pos, when); // timestamp
		pos += 4;
		*pos++ = get_type_code(); // event type code
		int4store(pos, server_id);
		pos += 4;
		int4store(pos, get_data_size() + LOG_EVENT_HEADER_LEN);
		pos += 4;
		return (my_fwrite(file, (byte*) buf, (uint) (pos - buf),
		@@ -106,16 +98,19 @@ int Log_event::read_log_event(FILE* file, String* packet)
		Log_event* Log_event::read_log_event(FILE* file)
		{
		time_t timestamp;
		char buf[5];
		uint32 server_id;

		char buf[LOG_EVENT_HEADER_LEN-4];
		if (my_fread(file, (byte *) buf, sizeof(buf), MY_NABP))
		return NULL;
		timestamp = uint4korr(buf);
		server_id = uint4korr(buf + 5);

		switch(buf[EVENT_TYPE_OFFSET])
		{
		case QUERY_EVENT:
		{
		Query_log_event* q = new Query_log_event(file, timestamp);
		Query_log_event* q = new Query_log_event(file, timestamp, server_id);
		if (!q->query)
		{
		delete q;
		@@ -127,7 +122,7 @@ Log_event* Log_event::read_log_event(FILE* file)

		case LOAD_EVENT:
		{
		Load_log_event* l = new Load_log_event(file, timestamp);
		Load_log_event* l = new Load_log_event(file, timestamp, server_id);
		if (!l->table_name)
		{
		delete l;
		@@ -140,7 +135,7 @@ Log_event* Log_event::read_log_event(FILE* file)

		case ROTATE_EVENT:
		{
		Rotate_log_event* r = new Rotate_log_event(file, timestamp);
		Rotate_log_event* r = new Rotate_log_event(file, timestamp, server_id);
		if (!r->new_log_ident)
		{
		delete r;
		@@ -152,7 +147,7 @@ Log_event* Log_event::read_log_event(FILE* file)

		case INTVAR_EVENT:
		{
		Intvar_log_event* e = new Intvar_log_event(file, timestamp);
		Intvar_log_event* e = new Intvar_log_event(file, timestamp, server_id);
		if (e->type == INVALID_INT_EVENT)
		{
		delete e;
		@@ -162,8 +157,8 @@ Log_event* Log_event::read_log_event(FILE* file)
		return e;
		}

		case START_EVENT: return new Start_log_event(file, timestamp);
		case STOP_EVENT: return new Stop_log_event(file, timestamp);
		case START_EVENT: return new Start_log_event(file, timestamp, server_id);
		case STOP_EVENT: return new Stop_log_event(file, timestamp, server_id);
		default: return NULL;
		}

		@@ -221,12 +216,23 @@ Log_event* Log_event::read_log_event(const char* buf, int max_buf)
		return NULL;
		}

		void Log_event::print_timestamp(FILE* file)
		void Log_event::print_header(FILE* file)
		{
		fputc('#', file);
		print_timestamp(file);
		fprintf(file, " server id %d ", server_id);
		}

		void Log_event::print_timestamp(FILE* file, time_t* ts = 0)
		{
		struct tm tm_tmp;
		localtime_r(&when,&tm_tmp);
		if(!ts)
		{
		ts = &when;
		}
		localtime_r(ts,&tm_tmp);

		fprintf(file,"#%02d%02d%02d %2d:%02d:%02d",
		fprintf(file,"%02d%02d%02d %2d:%02d:%02d",
		tm_tmp.tm_year % 100,
		tm_tmp.tm_mon+1,
		tm_tmp.tm_mday,
		@@ -241,8 +247,11 @@ void Start_log_event::print(FILE* file, bool short_form)
		if (short_form)
		return;

		print_timestamp(file);
		fprintf(file, "\tStart\n");
		print_header(file);
		fprintf(file, "\tStart: binlog v %d, server v %s created ", binlog_version,
		server_version);
		print_timestamp(file, (time_t*)&created);
		fputc('\n', file);
		fflush(file);
		}

		@@ -251,7 +260,7 @@ void Stop_log_event::print(FILE* file, bool short_form)
		if (short_form)
		return;

		print_timestamp(file);
		print_header(file);
		fprintf(file, "\tStop\n");
		fflush(file);
		}
		@@ -261,7 +270,7 @@ void Rotate_log_event::print(FILE* file, bool short_form)
		if (short_form)
		return;

		print_timestamp(file);
		print_header(file);
		fprintf(file, "\tRotate to ");
		if (new_log_ident)
		my_fwrite(file, (byte*) new_log_ident, (uint)ident_len,
		@@ -270,8 +279,9 @@ void Rotate_log_event::print(FILE* file, bool short_form)
		fflush(file);
		}

		Rotate_log_event::Rotate_log_event(FILE* file, time_t when_arg):
		Log_event(when_arg),new_log_ident(NULL),alloced(0)
		Rotate_log_event::Rotate_log_event(FILE* file, time_t when_arg,
		uint32 server_id):
		Log_event(when_arg, 0, 0, server_id),new_log_ident(NULL),alloced(0)
		{
		char *tmp_ident;
		char buf[4];
		@@ -298,6 +308,14 @@ Rotate_log_event::Rotate_log_event(FILE* file, time_t when_arg):
		alloced = 1;
		}

		Start_log_event::Start_log_event(const char* buf) :Log_event(buf)
		{
		buf += EVENT_LEN_OFFSET + 4; // skip even length
		binlog_version = uint2korr(buf);
		memcpy(server_version, buf + 2, sizeof(server_version));
		created = uint4korr(buf + 2 + sizeof(server_version));
		}

		Rotate_log_event::Rotate_log_event(const char* buf, int max_buf):
		Log_event(buf),new_log_ident(NULL),alloced(0)
		{
		@@ -322,8 +340,9 @@ int Rotate_log_event::write_data(FILE* file)
		return 0;
		}

		Query_log_event::Query_log_event(FILE* file, time_t when_arg):
		Log_event(when_arg),data_buf(0),query(NULL),db(NULL)
		Query_log_event::Query_log_event(FILE* file, time_t when_arg,
		uint32 server_id):
		Log_event(when_arg,0,0,server_id),data_buf(0),query(NULL),db(NULL)
		{
		char buf[QUERY_HEADER_LEN + 4];
		ulong data_len;
		@@ -382,7 +401,7 @@ void Query_log_event::print(FILE* file, bool short_form)
		{
		if (!short_form)
		{
		print_timestamp(file);
		print_header(file);
		fprintf(file, "\tQuery\tthread_id=%lu\texec_time=%lu\n",
		(ulong) thread_id, (ulong) exec_time);
		}
		@@ -414,8 +433,9 @@ int Query_log_event::write_data(FILE* file)
		return 0;
		}

		Intvar_log_event:: Intvar_log_event(FILE* file, time_t when_arg)
		:Log_event(when_arg), type(INVALID_INT_EVENT)
		Intvar_log_event:: Intvar_log_event(FILE* file, time_t when_arg,
		uint32 server_id)
		:Log_event(when_arg,0,0,server_id), type(INVALID_INT_EVENT)
		{
		my_fseek(file, 4L, MY_SEEK_CUR, MYF(MY_WME)); // skip the event length
		char buf[9];
		@@ -444,7 +464,7 @@ void Intvar_log_event::print(FILE* file, bool short_form)
		char llbuff[22];
		if(!short_form)
		{
		print_timestamp(file);
		print_header(file);
		fprintf(file, "\tIntvar\n");
		}

		@@ -493,8 +513,9 @@ int Load_log_event::write_data(FILE* file __attribute__((unused)))
		return 0;
		}

		Load_log_event::Load_log_event(FILE* file, time_t when):
		Log_event(when),data_buf(0),num_fields(0),fields(0),field_lens(0),field_block_len(0),
		Load_log_event::Load_log_event(FILE* file, time_t when, uint32 server_id):
		Log_event(when,0,0,server_id),data_buf(0),num_fields(0),
		fields(0),field_lens(0),field_block_len(0),
		table_name(0),db(0),fname(0)

		{
		@@ -539,7 +560,8 @@ Load_log_event::Load_log_event(FILE* file, time_t when):
		}

		Load_log_event::Load_log_event(const char* buf, int max_buf):
		Log_event(when),data_buf(0),num_fields(0),fields(0),field_lens(0),field_block_len(0),
		Log_event(when,0,0,server_id),data_buf(0),num_fields(0),fields(0),
		field_lens(0),field_block_len(0),
		table_name(0),db(0),fname(0)

		{
		@@ -594,7 +616,7 @@ void Load_log_event::print(FILE* file, bool short_form)
		{
		if (!short_form)
		{
		print_timestamp(file);
		print_header(file);
		fprintf(file, "\tQuery\tthread_id=%d\texec_time=%ld\n",
		thread_id, exec_time);
		}

sql/log_event.h

+69 −17

Original line number	Diff line number	Diff line
		@@ -28,6 +28,18 @@
		#define LOG_READ_MEM -5

		#define LOG_EVENT_OFFSET 4
		#define BINLOG_VERSION 1

		#define LOG_EVENT_HEADER_LEN 13
		#define QUERY_HEADER_LEN (sizeof(uint32) + sizeof(uint32) + sizeof(uchar))
		#define LOAD_HEADER_LEN (sizeof(uint32) + sizeof(uint32) + \
		+ sizeof(uint32) + 2 + sizeof(uint32))
		#define EVENT_LEN_OFFSET 9
		#define EVENT_TYPE_OFFSET 4
		#define MAX_EVENT_LEN 410241024
		#define QUERY_EVENT_OVERHEAD LOG_EVENT_HEADER_LEN+QUERY_HEADER_LEN
		#define ROTATE_EVENT_OVERHEAD LOG_EVENT_HEADER_LEN
		#define LOAD_EVENT_OVERHEAD (LOG_EVENT_HEADER_LEN+LOAD_HEADER_LEN+sizeof(sql_ex_info))


		enum Log_event_type { START_EVENT = 1, QUERY_EVENT =2,
		@@ -40,25 +52,32 @@ enum Int_event_type { INVALID_INT_EVENT = 0, LAST_INSERT_ID_EVENT = 1, INSERT_ID
		class String;
		#endif

		extern uint32 server_id;

		class Log_event
		{
		public:
		time_t when;
		ulong exec_time;
		int valid_exec_time; // if false, the exec time setting is bogus and needs
		int valid_exec_time; // if false, the exec time setting is bogus
		uint32 server_id;

		int write(FILE* file);
		int write_header(FILE* file);
		virtual int write_data(FILE* file __attribute__((unused))) { return 0; }
		virtual Log_event_type get_type_code() = 0;
		Log_event(time_t when_arg, ulong exec_time_arg = 0,
		int valid_exec_time_arg = 0): when(when_arg),
		exec_time(exec_time_arg), valid_exec_time(valid_exec_time_arg) {}
		int valid_exec_time_arg = 0, uint32 server_id = 0): when(when_arg),
		exec_time(exec_time_arg), valid_exec_time(valid_exec_time_arg)
		{
		if(!server_id) this->server_id = ::server_id;
		else this->server_id = server_id;
		}

		Log_event(const char* buf): valid_exec_time(1)
		Log_event(const char* buf): valid_exec_time(0)
		{
		when = uint4korr(buf);
		exec_time = uint4korr(buf + 5);
		server_id = uint4korr(buf + 5);
		}

		virtual ~Log_event() {}
		@@ -66,7 +85,8 @@ class Log_event
		virtual int get_data_size() { return 0;}
		virtual void print(FILE* file, bool short_form = 0) = 0;

		void print_timestamp(FILE* file);
		void print_timestamp(FILE* file, time_t *ts = 0);
		void print_header(FILE* file);

		static Log_event* read_log_event(FILE* file);
		static Log_event* read_log_event(const char* buf, int max_buf);
		@@ -93,7 +113,7 @@ class Query_log_event: public Log_event
		#if !defined(MYSQL_CLIENT)
		THD* thd;
		Query_log_event(THD* thd_arg, const char* query_arg):
		Log_event(thd_arg->start_time), data_buf(0),
		Log_event(thd_arg->start_time,0,0,thd_arg->server_id), data_buf(0),
		query(query_arg), db(thd_arg->db), q_len(thd_arg->query_length),
		thread_id(thd_arg->thread_id), thd(thd_arg)
		{
		@@ -105,7 +125,7 @@ class Query_log_event: public Log_event
		}
		#endif

		Query_log_event(FILE* file, time_t when);
		Query_log_event(FILE* file, time_t when, uint32 server_id);
		Query_log_event(const char* buf, int max_buf);
		~Query_log_event()
		{
		@@ -244,7 +264,7 @@ class Load_log_event: public Log_event
		void set_fields(List<Item> &fields);
		#endif

		Load_log_event(FILE* file, time_t when);
		Load_log_event(FILE* file, time_t when, uint32 server_id);
		Load_log_event(const char* buf, int max_buf);
		~Load_log_event()
		{
		@@ -269,21 +289,52 @@ class Load_log_event: public Log_event
		void print(FILE* file, bool short_form = 0);
		};

		extern char server_version[50];

		class Start_log_event: public Log_event
		{
		public:
		Start_log_event() :Log_event(time(NULL))
		{}
		Start_log_event(FILE* file, time_t when_arg) :Log_event(when_arg)
		uint16 binlog_version;
		char server_version[50];
		uint32 created;

		Start_log_event() :Log_event(time(NULL)),binlog_version(BINLOG_VERSION)
		{
		my_fseek(file, 4L, MY_SEEK_CUR, MYF(MY_WME)); // skip the event length
		created = when;
		memcpy(server_version, ::server_version, sizeof(server_version));
		}
		Start_log_event(const char* buf) :Log_event(buf)
		Start_log_event(FILE* file, time_t when_arg, uint32 server_id) :
		Log_event(when_arg, 0, 0, server_id)
		{
		char buf[sizeof(server_version) + sizeof(binlog_version) +
		sizeof(created)];
		my_fseek(file, 4L, MY_SEEK_CUR, MYF(MY_WME)); // skip the event length
		if (my_fread(file, (byte*) buf, sizeof(buf), MYF(MY_NABP \| MY_WME)))
		return;
		binlog_version = uint2korr(buf);
		memcpy(server_version, buf + 2, sizeof(server_version));
		created = uint4korr(buf + 2 + sizeof(server_version));
		}
		Start_log_event(const char* buf);

		~Start_log_event() {}
		Log_event_type get_type_code() { return START_EVENT;}
		int write_data(FILE* file)
		{
		if(my_fwrite(file, (byte*) &binlog_version, sizeof(binlog_version),
		MYF(MY_NABP \| MY_WME)) \|\|
		my_fwrite(file, (byte*) server_version, sizeof(server_version),
		MYF(MY_NABP \| MY_WME)) \|\|
		my_fwrite(file, (byte*) &created, sizeof(created),
		MYF(MY_NABP \| MY_WME)))
		return -1;
		return 0;
		}
		int get_data_size()
		{
		return sizeof(binlog_version) + sizeof(server_version) +
		sizeof(created);
		}
		void print(FILE* file, bool short_form = 0);
		};

		@@ -295,7 +346,7 @@ class Intvar_log_event: public Log_event
		Intvar_log_event(uchar type_arg, ulonglong val_arg)
		:Log_event(time(NULL)),val(val_arg),type(type_arg)
		{}
		Intvar_log_event(FILE* file, time_t when);
		Intvar_log_event(FILE* file, time_t when, uint32 server_id);
		Intvar_log_event(const char* buf);
		~Intvar_log_event() {}
		Log_event_type get_type_code() { return INTVAR_EVENT;}
		@@ -311,7 +362,8 @@ class Stop_log_event: public Log_event
		public:
		Stop_log_event() :Log_event(time(NULL))
		{}
		Stop_log_event(FILE* file, time_t when_arg): Log_event(when_arg)
		Stop_log_event(FILE* file, time_t when_arg, uint32 server_id):
		Log_event(when_arg,0,0,server_id)
		{
		my_fseek(file, 4L, MY_SEEK_CUR, MYF(MY_WME)); // skip the event length
		}
		@@ -337,7 +389,7 @@ class Rotate_log_event: public Log_event
		alloced(0)
		{}

		Rotate_log_event(FILE* file, time_t when) ;
		Rotate_log_event(FILE* file, time_t when, uint32 server_id) ;
		Rotate_log_event(const char* buf, int max_buf);
		~Rotate_log_event()
		{

sql/mysql_priv.h

+2 −1

Original line number	Diff line number	Diff line
		@@ -437,6 +437,7 @@ void sql_perror(const char *message);
		void sql_print_error(const char *format,...)
		__attribute__ ((format (printf, 1, 2)));

		extern uint32 server_id;
		extern char mysql_data_home[2],server_version[50],max_sort_char,
		mysql_real_data_home[];
		extern my_string mysql_unix_port,mysql_tmpdir;
		@@ -463,7 +464,7 @@ extern pthread_mutex_t LOCK_mysql_create_db,LOCK_Acl,LOCK_open,
		LOCK_thread_count,LOCK_mapped_file,LOCK_user_locks, LOCK_status,
		LOCK_grant, LOCK_error_log, LOCK_delayed_insert,
		LOCK_delayed_status, LOCK_delayed_create, LOCK_crypt, LOCK_timezone,
		LOCK_binlog_update, LOCK_slave;
		LOCK_binlog_update, LOCK_slave, LOCK_server_id;
		extern pthread_cond_t COND_refresh,COND_thread_count, COND_binlog_update,
		COND_slave_stopped;
		extern pthread_attr_t connection_attrib;