diff options
Diffstat (limited to 'sql/log_event.cc')
| -rw-r--r-- | sql/log_event.cc | 2056 |
1 files changed, 1679 insertions, 377 deletions
diff --git a/sql/log_event.cc b/sql/log_event.cc index c30d03adaf5..52e97310d75 100644 --- a/sql/log_event.cc +++ b/sql/log_event.cc @@ -20,39 +20,390 @@ #pragma implementation // gcc: Class implementation #endif #include "mysql_priv.h" +#include "slave.h" +#include <my_dir.h> #endif /* MYSQL_CLIENT */ +#include <assert.h> -static void pretty_print_char(FILE* file, int c) +inline int my_b_safe_write(IO_CACHE* file, const byte *buf, + int len) { + /* + Sasha: We are not writing this with the ? operator to avoid hitting + a possible compiler bug. At least gcc 2.95 cannot deal with + several layers of ternary operators that evaluated comma(,) operator + expressions inside - I do have a test case if somebody wants it + */ + if (file->type == SEQ_READ_APPEND) + return my_b_append(file, buf,len); + return my_b_write(file, buf,len); +} + +#ifdef MYSQL_CLIENT +static void pretty_print_str(FILE* file, char* str, int len) +{ + char* end = str + len; fputc('\'', file); - switch(c) { - case '\n': fprintf(file, "\\n"); break; - case '\r': fprintf(file, "\\r"); break; - case '\\': fprintf(file, "\\\\"); break; - case '\b': fprintf(file, "\\b"); break; - case '\'': fprintf(file, "\\'"); break; - case 0 : fprintf(file, "\\0"); break; - default: - fputc(c, file); - break; + while (str < end) + { + char c; + switch ((c=*str++)) { + case '\n': fprintf(file, "\\n"); break; + case '\r': fprintf(file, "\\r"); break; + case '\\': fprintf(file, "\\\\"); break; + case '\b': fprintf(file, "\\b"); break; + case '\t': fprintf(file, "\\t"); break; + case '\'': fprintf(file, "\\'"); break; + case 0 : fprintf(file, "\\0"); break; + default: + fputc(c, file); + break; + } } fputc('\'', file); } +#endif + +#ifndef MYSQL_CLIENT + +inline int ignored_error_code(int err_code) +{ + return use_slave_mask && bitmap_is_set(&slave_error_mask, err_code); +} + + +static void pretty_print_str(String* packet, char* str, int len) +{ + char* end = str + len; + packet->append('\''); + while (str < end) + { + char c; + switch ((c=*str++)) { + case '\n': packet->append( "\\n"); break; + case '\r': packet->append( "\\r"); break; + case '\\': packet->append( "\\\\"); break; + case '\b': packet->append( "\\b"); break; + case '\t': packet->append( "\\t"); break; + case '\'': packet->append( "\\'"); break; + case 0 : packet->append( "\\0"); break; + default: + packet->append((char)c); + break; + } + } + packet->append('\''); +} + + +static inline char* slave_load_file_stem(char*buf, uint file_id, + int event_server_id) +{ + fn_format(buf,"SQL_LOAD-",slave_load_tmpdir,"",4+32); + buf = strend(buf); + buf = int10_to_str(::server_id, buf, 10); + *buf++ = '-'; + buf = int10_to_str(event_server_id, buf, 10); + *buf++ = '-'; + return int10_to_str(file_id, buf, 10); +} + +#endif + +const char* Log_event::get_type_str() +{ + switch(get_type_code()) { + case START_EVENT: return "Start"; + case STOP_EVENT: return "Stop"; + case QUERY_EVENT: return "Query"; + case ROTATE_EVENT: return "Rotate"; + case INTVAR_EVENT: return "Intvar"; + case LOAD_EVENT: return "Load"; + case NEW_LOAD_EVENT: return "New_load"; + case SLAVE_EVENT: return "Slave"; + case CREATE_FILE_EVENT: return "Create_file"; + case APPEND_BLOCK_EVENT: return "Append_block"; + case DELETE_FILE_EVENT: return "Delete_file"; + case EXEC_LOAD_EVENT: return "Exec_load"; + default: /* impossible */ return "Unknown"; + } +} + +#ifndef MYSQL_CLIENT +Log_event::Log_event(THD* thd_arg, uint16 flags_arg) + :exec_time(0), flags(flags_arg), cached_event_len(0), + temp_buf(0), thd(thd_arg) +{ + if (thd) + { + server_id = thd->server_id; + when = thd->start_time; + log_pos = thd->log_pos; + } + else + { + server_id = ::server_id; + when = time(NULL); + log_pos=0; + } +} + + +static void cleanup_load_tmpdir() +{ + MY_DIR *dirp; + FILEINFO *file; + uint i; + if (!(dirp=my_dir(slave_load_tmpdir,MYF(MY_WME)))) + return; + + for (i=0 ; i < (uint)dirp->number_off_files; i++) + { + file=dirp->dir_entry+i; + if (is_prefix(file->name,"SQL_LOAD-")) + my_delete(file->name, MYF(0)); + } + + my_dirend(dirp); +} + +#endif + +Log_event::Log_event(const char* buf, bool old_format) + :cached_event_len(0), temp_buf(0) +{ + when = uint4korr(buf); + server_id = uint4korr(buf + SERVER_ID_OFFSET); + if (old_format) + { + log_pos=0; + flags=0; + } + else + { + log_pos = uint4korr(buf + LOG_POS_OFFSET); + flags = uint2korr(buf + FLAGS_OFFSET); + } +#ifndef MYSQL_CLIENT + thd = 0; +#endif +} + + +#ifndef MYSQL_CLIENT + +int Log_event::exec_event(struct st_relay_log_info* rli) +{ + if (rli) + { + rli->inc_pos(get_event_len(),log_pos); + DBUG_ASSERT(rli->sql_thd != 0); + flush_relay_log_info(rli); + } + return 0; +} + +void Log_event::pack_info(String* packet) +{ + net_store_data(packet, "", 0); +} + +void Query_log_event::pack_info(String* packet) +{ + char buf[256]; + String tmp(buf, sizeof(buf)); + tmp.length(0); + if (db && db_len) + { + tmp.append("use "); + tmp.append(db, db_len); + tmp.append("; ", 2); + } + + if (query && q_len) + tmp.append(query, q_len); + net_store_data(packet, (char*)tmp.ptr(), tmp.length()); +} + +void Start_log_event::pack_info(String* packet) +{ + char buf1[256]; + String tmp(buf1, sizeof(buf1)); + tmp.length(0); + char buf[22]; + + tmp.append("Server ver: "); + tmp.append(server_version); + tmp.append(", Binlog ver: "); + tmp.append(llstr(binlog_version, buf)); + net_store_data(packet, tmp.ptr(), tmp.length()); +} + +void Load_log_event::pack_info(String* packet) +{ + char buf[256]; + String tmp(buf, sizeof(buf)); + tmp.length(0); + if (db && db_len) + { + tmp.append("use "); + tmp.append(db, db_len); + tmp.append("; ", 2); + } + + tmp.append("LOAD DATA INFILE '"); + tmp.append(fname, fname_len); + tmp.append("' ", 2); + if (sql_ex.opt_flags && REPLACE_FLAG ) + tmp.append(" REPLACE "); + else if (sql_ex.opt_flags && IGNORE_FLAG ) + tmp.append(" IGNORE "); + + tmp.append("INTO TABLE "); + tmp.append(table_name); + if (sql_ex.field_term_len) + { + tmp.append(" FIELDS TERMINATED BY "); + pretty_print_str(&tmp, sql_ex.field_term, sql_ex.field_term_len); + } + + if (sql_ex.enclosed_len) + { + if (sql_ex.opt_flags && OPT_ENCLOSED_FLAG ) + tmp.append(" OPTIONALLY "); + tmp.append( " ENCLOSED BY "); + pretty_print_str(&tmp, sql_ex.enclosed, sql_ex.enclosed_len); + } + + if (sql_ex.escaped_len) + { + tmp.append( " ESCAPED BY "); + pretty_print_str(&tmp, sql_ex.escaped, sql_ex.escaped_len); + } + + if (sql_ex.line_term_len) + { + tmp.append(" LINES TERMINATED BY "); + pretty_print_str(&tmp, sql_ex.line_term, sql_ex.line_term_len); + } + + if (sql_ex.line_start_len) + { + tmp.append(" LINES STARTING BY "); + pretty_print_str(&tmp, sql_ex.line_start, sql_ex.line_start_len); + } + + if ((int)skip_lines > 0) + tmp.append( " IGNORE %ld LINES ", (long) skip_lines); + + if (num_fields) + { + uint i; + const char* field = fields; + tmp.append(" ("); + for (i = 0; i < num_fields; i++) + { + if (i) + tmp.append(" ,"); + tmp.append( field); + + field += field_lens[i] + 1; + } + tmp.append(')'); + } + + net_store_data(packet, tmp.ptr(), tmp.length()); +} + +void Rotate_log_event::pack_info(String* packet) +{ + char buf1[256], buf[22]; + String tmp(buf1, sizeof(buf1)); + tmp.length(0); + tmp.append(new_log_ident, ident_len); + tmp.append(";pos="); + tmp.append(llstr(pos,buf)); + if (flags & LOG_EVENT_FORCED_ROTATE_F) + tmp.append("; forced by master"); + net_store_data(packet, tmp.ptr(), tmp.length()); +} + +void Intvar_log_event::pack_info(String* packet) +{ + char buf1[256], buf[22]; + String tmp(buf1, sizeof(buf1)); + tmp.length(0); + tmp.append(get_var_type_name()); + tmp.append('='); + tmp.append(llstr(val, buf)); + net_store_data(packet, tmp.ptr(), tmp.length()); +} + +void Slave_log_event::pack_info(String* packet) +{ + char buf1[256], buf[22], *end; + String tmp(buf1, sizeof(buf1)); + tmp.length(0); + tmp.append("host="); + tmp.append(master_host); + tmp.append(",port="); + end= int10_to_str((long) master_port, buf, 10); + tmp.append(buf, (uint32) (end-buf)); + tmp.append(",log="); + tmp.append(master_log); + tmp.append(",pos="); + tmp.append(llstr(master_pos,buf)); + net_store_data(packet, tmp.ptr(), tmp.length()); +} + + +void Log_event::init_show_field_list(List<Item>* field_list) +{ + field_list->push_back(new Item_empty_string("Log_name", 20)); + field_list->push_back(new Item_empty_string("Pos", 20)); + field_list->push_back(new Item_empty_string("Event_type", 20)); + field_list->push_back(new Item_empty_string("Server_id", 20)); + field_list->push_back(new Item_empty_string("Orig_log_pos", 20)); + field_list->push_back(new Item_empty_string("Info", 20)); +} + +int Log_event::net_send(THD* thd, const char* log_name, my_off_t pos) +{ + String* packet = &thd->packet; + const char* p = strrchr(log_name, FN_LIBCHAR); + const char* event_type; + if (p) + log_name = p + 1; + + packet->length(0); + net_store_data(packet, log_name, strlen(log_name)); + net_store_data(packet, (longlong) pos); + event_type = get_type_str(); + net_store_data(packet, event_type, strlen(event_type)); + net_store_data(packet, server_id); + net_store_data(packet, (longlong) log_pos); + pack_info(packet); + return my_net_write(&thd->net, (char*) packet->ptr(), packet->length()); +} + +#endif /* MYSQL_CLIENT */ + int Query_log_event::write(IO_CACHE* file) { return query ? Log_event::write(file) : -1; } + int Log_event::write(IO_CACHE* file) { return (write_header(file) || write_data(file)) ? -1 : 0; } + int Log_event::write_header(IO_CACHE* file) { - // make sure to change this when the header gets bigger char buf[LOG_EVENT_HEADER_LEN]; char* pos = buf; int4store(pos, (ulong) when); // timestamp @@ -63,7 +414,11 @@ int Log_event::write_header(IO_CACHE* file) long tmp=get_data_size() + LOG_EVENT_HEADER_LEN; int4store(pos, tmp); pos += 4; - return (my_b_write(file, (byte*) buf, (uint) (pos - buf))); + int4store(pos, log_pos); + pos += 4; + int2store(pos, flags); + pos += 2; + return (my_b_safe_write(file, (byte*) buf, (uint) (pos - buf))); } #ifndef MYSQL_CLIENT @@ -72,215 +427,210 @@ int Log_event::read_log_event(IO_CACHE* file, String* packet, pthread_mutex_t* log_lock) { ulong data_len; + int result=0; char buf[LOG_EVENT_HEADER_LEN]; + if (log_lock) pthread_mutex_lock(log_lock); if (my_b_read(file, (byte*) buf, sizeof(buf))) { - if (log_lock) pthread_mutex_unlock(log_lock); - // if the read hits eof, we must report it as eof - // so the caller will know it can go into cond_wait to be woken up - // on the next update to the log - if(!file->error) return LOG_READ_EOF; - return file->error > 0 ? LOG_READ_TRUNC: LOG_READ_IO; + /* + If the read hits eof, we must report it as eof so the caller + will know it can go into cond_wait to be woken up on the next + update to the log. + */ + if (!file->error) + result= LOG_READ_EOF; + else + result= (file->error > 0 ? LOG_READ_TRUNC: LOG_READ_IO); + goto end; } - data_len = uint4korr(buf + EVENT_LEN_OFFSET); - if (data_len < LOG_EVENT_HEADER_LEN || data_len > max_allowed_packet) + data_len= uint4korr(buf + EVENT_LEN_OFFSET); + if (data_len < LOG_EVENT_HEADER_LEN || + data_len > current_thd->variables.max_allowed_packet) { - if (log_lock) pthread_mutex_unlock(log_lock); - return (data_len < LOG_EVENT_HEADER_LEN) ? LOG_READ_BOGUS : - LOG_READ_TOO_LARGE; + + result= ((data_len < LOG_EVENT_HEADER_LEN) ? LOG_READ_BOGUS : + LOG_READ_TOO_LARGE); + goto end; } packet->append(buf, sizeof(buf)); - data_len -= LOG_EVENT_HEADER_LEN; + data_len-= LOG_EVENT_HEADER_LEN; if (data_len) { if (packet->append(file, data_len)) { - if(log_lock) - pthread_mutex_unlock(log_lock); - // here we should never hit eof in a non-error condtion - // eof means we are reading the event partially, which should - // never happen - return file->error >= 0 ? LOG_READ_TRUNC: LOG_READ_IO; + /* + Here we should never hit eof in a non-error condtion + eof means we are reading the event partially, which should + never happen. + */ + result= file->error >= 0 ? LOG_READ_TRUNC: LOG_READ_IO; + /* Implicit goto end; */ } } - if (log_lock) pthread_mutex_unlock(log_lock); - return 0; + +end: + if (log_lock) + pthread_mutex_unlock(log_lock); + return result; } #endif // MYSQL_CLIENT #ifndef MYSQL_CLIENT -#define UNLOCK_MUTEX if(log_lock) pthread_mutex_unlock(log_lock); +#define UNLOCK_MUTEX if (log_lock) pthread_mutex_unlock(log_lock); +#define LOCK_MUTEX if (log_lock) pthread_mutex_lock(log_lock); +#define max_allowed_packet current_thd->variables.max_allowed_packet #else #define UNLOCK_MUTEX +#define LOCK_MUTEX #endif // allocates memory - the caller is responsible for clean-up #ifndef MYSQL_CLIENT -Log_event* Log_event::read_log_event(IO_CACHE* file, pthread_mutex_t* log_lock) +Log_event* Log_event::read_log_event(IO_CACHE* file, + pthread_mutex_t* log_lock, + bool old_format) #else -Log_event* Log_event::read_log_event(IO_CACHE* file) +Log_event* Log_event::read_log_event(IO_CACHE* file, bool old_format) #endif { - time_t timestamp; - uint32 server_id; - - char buf[LOG_EVENT_HEADER_LEN-4]; -#ifndef MYSQL_CLIENT - if(log_lock) pthread_mutex_lock(log_lock); -#endif - if (my_b_read(file, (byte *) buf, sizeof(buf))) - { - UNLOCK_MUTEX - return NULL; - } - timestamp = uint4korr(buf); - server_id = uint4korr(buf + 5); - - switch(buf[EVENT_TYPE_OFFSET]) - { - case QUERY_EVENT: + char head[LOG_EVENT_HEADER_LEN]; + uint header_size= old_format ? OLD_HEADER_LEN : LOG_EVENT_HEADER_LEN; + + LOCK_MUTEX; + if (my_b_read(file, (byte *) head, header_size)) { - Query_log_event* q = new Query_log_event(file, timestamp, server_id); - UNLOCK_MUTEX - if (!q->query) - { - delete q; - q=NULL; - } - return q; + UNLOCK_MUTEX; + return 0; } - - case LOAD_EVENT: + + uint data_len = uint4korr(head + EVENT_LEN_OFFSET); + char *buf= 0; + const char *error= 0; + Log_event *res= 0; + + if (data_len > max_allowed_packet) { - Load_log_event* l = new Load_log_event(file, timestamp, server_id); - UNLOCK_MUTEX - if (!l->table_name) - { - delete l; - l=NULL; - } - return l; + error = "Event too big"; + goto err; } - - case ROTATE_EVENT: + if (data_len < header_size) { - Rotate_log_event* r = new Rotate_log_event(file, timestamp, server_id); - UNLOCK_MUTEX - if (!r->new_log_ident) - { - delete r; - r=NULL; - } - return r; + error = "Event too small"; + goto err; } - case INTVAR_EVENT: + // some events use the extra byte to null-terminate strings + if (!(buf = my_malloc(data_len+1, MYF(MY_WME)))) { - Intvar_log_event* e = new Intvar_log_event(file, timestamp, server_id); - UNLOCK_MUTEX - if (e->type == INVALID_INT_EVENT) - { - delete e; - e=NULL; - } - return e; + error = "Out of memory"; + goto err; } - - case START_EVENT: - { - Start_log_event* e = new Start_log_event(file, timestamp, server_id); - UNLOCK_MUTEX - return e; - } - case STOP_EVENT: - { - Stop_log_event* e = new Stop_log_event(file, timestamp, server_id); - UNLOCK_MUTEX - return e; - } - default: - break; + buf[data_len] = 0; + memcpy(buf, head, header_size); + if (my_b_read(file, (byte*) buf + header_size, data_len - header_size)) + { + error = "read error"; + goto err; } + if ((res = read_log_event(buf, data_len, &error, old_format))) + res->register_temp_buf(buf); - // default - UNLOCK_MUTEX - return NULL; +err: + UNLOCK_MUTEX; + if (error) + { + sql_print_error("Error in Log_event::read_log_event(): '%s', \ +data_len=%d,event_type=%d",error,data_len,head[EVENT_TYPE_OFFSET]); + my_free(buf, MYF(MY_ALLOW_ZERO_PTR)); + } + return res; } -Log_event* Log_event::read_log_event(const char* buf, int event_len) + +Log_event* Log_event::read_log_event(const char* buf, int event_len, + const char **error, bool old_format) { - if(event_len < EVENT_LEN_OFFSET || - (uint)event_len != uint4korr(buf+EVENT_LEN_OFFSET)) + if (event_len < EVENT_LEN_OFFSET || + (uint) event_len != uint4korr(buf+EVENT_LEN_OFFSET)) + { + *error="Sanity check failed"; // Needed to free buffer return NULL; // general sanity check - will fail on a partial read + } - switch(buf[EVENT_TYPE_OFFSET]) - { + Log_event* ev = NULL; + + switch(buf[EVENT_TYPE_OFFSET]) { case QUERY_EVENT: - { - Query_log_event* q = new Query_log_event(buf, event_len); - if (!q->query) - { - delete q; - return NULL; - } - - return q; - } - + ev = new Query_log_event(buf, event_len, old_format); + break; case LOAD_EVENT: - { - Load_log_event* l = new Load_log_event(buf, event_len); - if (!l->table_name) - { - delete l; - return NULL; - } - - return l; - } - + ev = new Create_file_log_event(buf, event_len, old_format); + break; + case NEW_LOAD_EVENT: + ev = new Load_log_event(buf, event_len, old_format); + break; case ROTATE_EVENT: - { - Rotate_log_event* r = new Rotate_log_event(buf, event_len); - if (!r->new_log_ident) - { - delete r; - return NULL; - } - - return r; - } - case START_EVENT: return new Start_log_event(buf); - case STOP_EVENT: return new Stop_log_event(buf); - case INTVAR_EVENT: return new Intvar_log_event(buf); + ev = new Rotate_log_event(buf, event_len, old_format); + break; + case SLAVE_EVENT: + ev = new Slave_log_event(buf, event_len); + break; + case CREATE_FILE_EVENT: + ev = new Create_file_log_event(buf, event_len, old_format); + break; + case APPEND_BLOCK_EVENT: + ev = new Append_block_log_event(buf, event_len); + break; + case DELETE_FILE_EVENT: + ev = new Delete_file_log_event(buf, event_len); + break; + case EXEC_LOAD_EVENT: + ev = new Execute_load_log_event(buf, event_len); + break; + case START_EVENT: + ev = new Start_log_event(buf, old_format); + break; + case STOP_EVENT: + ev = new Stop_log_event(buf, old_format); + break; + case INTVAR_EVENT: + ev = new Intvar_log_event(buf, old_format); + break; default: break; } - return NULL; // default value + if (!ev || !ev->is_valid()) + { + *error= "Found invalid event in binary log"; + delete ev; + return 0; + } + ev->cached_event_len = event_len; + return ev; } + +#ifdef MYSQL_CLIENT void Log_event::print_header(FILE* file) { + char llbuff[22]; fputc('#', file); print_timestamp(file); - fprintf(file, " server id %d ", server_id); + fprintf(file, " server id %d log_pos %s ", server_id, + llstr(log_pos,llbuff)); } void Log_event::print_timestamp(FILE* file, time_t* ts) { -#ifdef MYSQL_SERVER - struct tm tm_tmp; -#endif struct tm *res; if (!ts) - { ts = &when; - } -#ifdef MYSQL_SERVER +#ifdef MYSQL_SERVER // This is always false + struct tm tm_tmp; localtime_r(ts,(res= &tm_tmp)); #else res=localtime(ts); @@ -321,6 +671,7 @@ void Stop_log_event::print(FILE* file, bool short_form, char* last_db) void Rotate_log_event::print(FILE* file, bool short_form, char* last_db) { + char buf[22]; if (short_form) return; @@ -329,134 +680,128 @@ void Rotate_log_event::print(FILE* file, bool short_form, char* last_db) if (new_log_ident) my_fwrite(file, (byte*) new_log_ident, (uint)ident_len, MYF(MY_NABP | MY_WME)); - fprintf(file, "\n"); + fprintf(file, " pos: %s\n", llstr(pos, buf)); fflush(file); } -Rotate_log_event::Rotate_log_event(IO_CACHE* 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]; +#endif /* #ifdef MYSQL_CLIENT */ - if (my_b_read(file, (byte*) buf, sizeof(buf))) - return; - ulong event_len; - event_len = uint4korr(buf); - if (event_len < ROTATE_EVENT_OVERHEAD) - return; - ident_len = (uchar)(event_len - ROTATE_EVENT_OVERHEAD); - if (!(tmp_ident = (char*) my_malloc((uint)ident_len, MYF(MY_WME)))) - return; - if (my_b_read( file, (byte*) tmp_ident, (uint) ident_len)) - { - my_free((gptr) tmp_ident, MYF(0)); - return; - } - - new_log_ident = tmp_ident; - alloced = 1; -} - -Start_log_event::Start_log_event(const char* buf) :Log_event(buf) +Start_log_event::Start_log_event(const char* buf, + bool old_format) + :Log_event(buf, old_format) { - 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)); + buf += (old_format) ? OLD_HEADER_LEN : LOG_EVENT_HEADER_LEN; + binlog_version = uint2korr(buf+ST_BINLOG_VER_OFFSET); + memcpy(server_version, buf+ST_SERVER_VER_OFFSET, + ST_SERVER_VER_LEN); + created = uint4korr(buf+ST_CREATED_OFFSET); } int Start_log_event::write_data(IO_CACHE* file) { - char buff[sizeof(server_version)+2+4]; - int2store(buff,binlog_version); - memcpy(buff+2,server_version,sizeof(server_version)); - int4store(buff+2+sizeof(server_version),created); - return (my_b_write(file, (byte*) buff, sizeof(buff)) ? -1 : 0); + char buff[START_HEADER_LEN]; + int2store(buff + ST_BINLOG_VER_OFFSET,binlog_version); + memcpy(buff + ST_SERVER_VER_OFFSET,server_version,ST_SERVER_VER_LEN); + int4store(buff + ST_CREATED_OFFSET,created); + return (my_b_safe_write(file, (byte*) buff, sizeof(buff)) ? -1 : 0); } -Rotate_log_event::Rotate_log_event(const char* buf, int event_len): - Log_event(buf),new_log_ident(NULL),alloced(0) + +Rotate_log_event::Rotate_log_event(const char* buf, int event_len, + bool old_format) + :Log_event(buf, old_format),new_log_ident(NULL),alloced(0) { - // the caller will ensure that event_len is what we have at - // EVENT_LEN_OFFSET - if(event_len < ROTATE_EVENT_OVERHEAD) + // The caller will ensure that event_len is what we have at EVENT_LEN_OFFSET + int header_size = (old_format) ? OLD_HEADER_LEN : LOG_EVENT_HEADER_LEN; + uint ident_offset; + if (event_len < header_size) return; - - ident_len = (uchar)(event_len - ROTATE_EVENT_OVERHEAD); - if (!(new_log_ident = (char*) my_memdup((byte*) buf + LOG_EVENT_HEADER_LEN, + buf += header_size; + if (old_format) + { + ident_len = (uchar)(event_len - OLD_HEADER_LEN); + pos = 4; + ident_offset = 0; + } + else + { + ident_len = (uchar)(event_len - ROTATE_EVENT_OVERHEAD); + pos = uint8korr(buf + R_POS_OFFSET); + ident_offset = ROTATE_HEADER_LEN; + } + if (!(new_log_ident = (char*) my_memdup((byte*) buf + ident_offset, (uint) ident_len, MYF(MY_WME)))) return; alloced = 1; } + int Rotate_log_event::write_data(IO_CACHE* file) { - return my_b_write(file, (byte*) new_log_ident, (uint) ident_len) ? -1 :0; + char buf[ROTATE_HEADER_LEN]; + int8store(buf, pos + R_POS_OFFSET); + return (my_b_safe_write(file, (byte*)buf, ROTATE_HEADER_LEN) || + my_b_safe_write(file, (byte*)new_log_ident, (uint) ident_len)); } -Query_log_event::Query_log_event(IO_CACHE* 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; - if (my_b_read(file, (byte*) buf, sizeof(buf))) - return; // query == NULL will tell the - // caller there was a problem - data_len = uint4korr(buf); - if (data_len < QUERY_EVENT_OVERHEAD) - return; // tear-drop attack protection :) - - data_len -= QUERY_EVENT_OVERHEAD; - exec_time = uint4korr(buf + 8); - db_len = (uint)buf[12]; - error_code = uint2korr(buf + 13); - - /* Allocate one byte extra for end \0 */ - if (!(data_buf = (char*) my_malloc(data_len+1, MYF(MY_WME)))) - return; - if (my_b_read( file, (byte*) data_buf, data_len)) - { - my_free((gptr) data_buf, MYF(0)); - data_buf = 0; - return; - } - thread_id = uint4korr(buf + 4); - db = data_buf; - query=data_buf + db_len + 1; - q_len = data_len - 1 - db_len; - *((char*) query + q_len) = 0; // Safety +#ifndef MYSQL_CLIENT +Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg, + bool using_trans) + :Log_event(thd_arg), data_buf(0), query(query_arg), db(thd_arg->db), + q_len(thd_arg->query_length), + error_code(thd_arg->killed ? ER_SERVER_SHUTDOWN: thd_arg->net.last_errno), + thread_id(thd_arg->thread_id), + cache_stmt(using_trans && + (thd_arg->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))) +{ + time_t end_time; + time(&end_time); + exec_time = (ulong) (end_time - thd->start_time); + db_len = (db) ? (uint32) strlen(db) : 0; } +#endif -Query_log_event::Query_log_event(const char* buf, int event_len): - Log_event(buf),data_buf(0), query(NULL), db(NULL) +Query_log_event::Query_log_event(const char* buf, int event_len, + bool old_format) + :Log_event(buf, old_format),data_buf(0), query(NULL), db(NULL) { - if ((uint)event_len < QUERY_EVENT_OVERHEAD) - return; ulong data_len; - buf += EVENT_LEN_OFFSET; - data_len = event_len - QUERY_EVENT_OVERHEAD; + if (old_format) + { + if ((uint)event_len < OLD_HEADER_LEN + QUERY_HEADER_LEN) + return; + data_len = event_len - (QUERY_HEADER_LEN + OLD_HEADER_LEN); + buf += OLD_HEADER_LEN; + } + else + { + if ((uint)event_len < QUERY_EVENT_OVERHEAD) + return; + data_len = event_len - QUERY_EVENT_OVERHEAD; + buf += LOG_EVENT_HEADER_LEN; + } - exec_time = uint4korr(buf + 8); - error_code = uint2korr(buf + 13); + exec_time = uint4korr(buf + Q_EXEC_TIME_OFFSET); + error_code = uint2korr(buf + Q_ERR_CODE_OFFSET); if (!(data_buf = (char*) my_malloc(data_len + 1, MYF(MY_WME)))) return; - memcpy(data_buf, buf + QUERY_HEADER_LEN + 4, data_len); - thread_id = uint4korr(buf + 4); + memcpy(data_buf, buf + Q_DATA_OFFSET, data_len); + thread_id = uint4korr(buf + Q_THREAD_ID_OFFSET); db = data_buf; - db_len = (uint)buf[12]; + db_len = (uint)buf[Q_DB_LEN_OFFSET]; query=data_buf + db_len + 1; q_len = data_len - 1 - db_len; *((char*)query+q_len) = 0; } + +#ifdef MYSQL_CLIENT + void Query_log_event::print(FILE* file, bool short_form, char* last_db) { char buff[40],*end; // Enough for SET TIMESTAMP @@ -469,11 +814,11 @@ void Query_log_event::print(FILE* file, bool short_form, char* last_db) bool same_db = 0; - if(db && last_db) - { - if(!(same_db = !memcmp(last_db, db, db_len + 1))) - memcpy(last_db, db, db_len + 1); - } + if (db && last_db) + { + if (!(same_db = !memcmp(last_db, db, db_len + 1))) + memcpy(last_db, db, db_len + 1); + } if (db && db[0] && !same_db) fprintf(file, "use %s;\n", db); @@ -484,170 +829,334 @@ void Query_log_event::print(FILE* file, bool short_form, char* last_db) my_fwrite(file, (byte*) query, q_len, MYF(MY_NABP | MY_WME)); fprintf(file, ";\n"); } +#endif + int Query_log_event::write_data(IO_CACHE* file) { - if (!query) return -1; + if (!query) + return -1; char buf[QUERY_HEADER_LEN]; - char* pos = buf; - int4store(pos, thread_id); - pos += 4; - int4store(pos, exec_time); - pos += 4; - *pos++ = (char)db_len; - int2store(pos, error_code); - pos += 2; - - return (my_b_write(file, (byte*) buf, (uint)(pos - buf)) || - my_b_write(file, (db) ? (byte*) db : (byte*)"", db_len + 1) || - my_b_write(file, (byte*) query, q_len)) ? -1 : 0; + int4store(buf + Q_THREAD_ID_OFFSET, thread_id); + int4store(buf + Q_EXEC_TIME_OFFSET, exec_time); + buf[Q_DB_LEN_OFFSET] = (char) db_len; + int2store(buf + Q_ERR_CODE_OFFSET, error_code); + + return (my_b_safe_write(file, (byte*) buf, QUERY_HEADER_LEN) || + my_b_safe_write(file, (db) ? (byte*) db : (byte*)"", db_len + 1) || + my_b_safe_write(file, (byte*) query, q_len)) ? -1 : 0; } -Intvar_log_event:: Intvar_log_event(IO_CACHE* file, time_t when_arg, - uint32 server_id) - :Log_event(when_arg,0,0,server_id), type(INVALID_INT_EVENT) +Intvar_log_event::Intvar_log_event(const char* buf, bool old_format) + :Log_event(buf, old_format) { - char buf[9+4]; - if (!my_b_read(file, (byte*) buf, sizeof(buf))) - { - type = buf[4]; - val = uint8korr(buf+1+4); - } + buf += (old_format) ? OLD_HEADER_LEN : LOG_EVENT_HEADER_LEN; + type = buf[I_TYPE_OFFSET]; + val = uint8korr(buf+I_VAL_OFFSET); } -Intvar_log_event::Intvar_log_event(const char* buf):Log_event(buf) +const char* Intvar_log_event::get_var_type_name() { - buf += LOG_EVENT_HEADER_LEN; - type = buf[0]; - val = uint8korr(buf+1); + switch(type) { + case LAST_INSERT_ID_EVENT: return "LAST_INSERT_ID"; + case INSERT_ID_EVENT: return "INSERT_ID"; + default: /* impossible */ return "UNKNOWN"; + } } int Intvar_log_event::write_data(IO_CACHE* file) { char buf[9]; - buf[0] = type; - int8store(buf + 1, val); - return my_b_write(file, (byte*) buf, sizeof(buf)); + buf[I_TYPE_OFFSET] = type; + int8store(buf + I_VAL_OFFSET, val); + return my_b_safe_write(file, (byte*) buf, sizeof(buf)); } +#ifdef MYSQL_CLIENT void Intvar_log_event::print(FILE* file, bool short_form, char* last_db) { char llbuff[22]; - if(!short_form) + const char *msg; + LINT_INIT(msg); + + if (!short_form) { print_header(file); fprintf(file, "\tIntvar\n"); } fprintf(file, "SET "); - switch(type) - { + switch (type) { case LAST_INSERT_ID_EVENT: - fprintf(file, "LAST_INSERT_ID = "); + msg="LAST_INSERT_ID"; break; case INSERT_ID_EVENT: - fprintf(file, "INSERT_ID = "); + msg="INSERT_ID"; break; } - fprintf(file, "%s;\n", llstr(val,llbuff)); + fprintf(file, "%s=%s;\n", msg, llstr(val,llbuff)); fflush(file); - } +#endif -int Load_log_event::write_data(IO_CACHE* file) + +int Load_log_event::write_data_header(IO_CACHE* file) { char buf[LOAD_HEADER_LEN]; - int4store(buf, thread_id); - int4store(buf + 4, exec_time); - int4store(buf + 8, skip_lines); - buf[12] = (char)table_name_len; - buf[13] = (char)db_len; - int4store(buf + 14, num_fields); - - if(my_b_write(file, (byte*)buf, sizeof(buf)) || - my_b_write(file, (byte*)&sql_ex, sizeof(sql_ex))) - return 1; + int4store(buf + L_THREAD_ID_OFFSET, thread_id); + int4store(buf + L_EXEC_TIME_OFFSET, exec_time); + int4store(buf + L_SKIP_LINES_OFFSET, skip_lines); + buf[L_TBL_LEN_OFFSET] = (char)table_name_len; + buf[L_DB_LEN_OFFSET] = (char)db_len; + int4store(buf + L_NUM_FIELDS_OFFSET, num_fields); + return my_b_safe_write(file, (byte*)buf, LOAD_HEADER_LEN); +} +int Load_log_event::write_data_body(IO_CACHE* file) +{ + if (sql_ex.write_data(file)) + return 1; if (num_fields && fields && field_lens) { - if(my_b_write(file, (byte*)field_lens, num_fields) || - my_b_write(file, (byte*)fields, field_block_len)) + if (my_b_safe_write(file, (byte*)field_lens, num_fields) || + my_b_safe_write(file, (byte*)fields, field_block_len)) return 1; } - if(my_b_write(file, (byte*)table_name, table_name_len + 1) || - my_b_write(file, (byte*)db, db_len + 1) || - my_b_write(file, (byte*)fname, fname_len)) + return (my_b_safe_write(file, (byte*)table_name, table_name_len + 1) || + my_b_safe_write(file, (byte*)db, db_len + 1) || + my_b_safe_write(file, (byte*)fname, fname_len)); +} + + + +static bool write_str(IO_CACHE *file, char *str, byte length) +{ + return (my_b_safe_write(file, &length, 1) || + my_b_safe_write(file, (byte*) str, (int) length)); +} + + +int sql_ex_info::write_data(IO_CACHE* file) +{ + if (new_format()) + { + return (write_str(file, field_term, field_term_len) || + write_str(file, enclosed, enclosed_len) || + write_str(file, line_term, line_term_len) || + write_str(file, line_start, line_start_len) || + write_str(file, escaped, escaped_len) || + my_b_safe_write(file,(byte*) &opt_flags,1)); + } + else + { + old_sql_ex old_ex; + old_ex.field_term= *field_term; + old_ex.enclosed= *enclosed; + old_ex.line_term= *line_term; + old_ex.line_start= *line_start; + old_ex.escaped= *escaped; + old_ex.opt_flags= opt_flags; + old_ex.empty_flags=empty_flags; + return my_b_safe_write(file, (byte*) &old_ex, sizeof(old_ex)); + } +} + + +static inline int read_str(char * &buf, char *buf_end, char * &str, + uint8 &len) +{ + if (buf + (uint) (uchar) *buf >= buf_end) return 1; + len = (uint8) *buf; + str= buf+1; + buf+= (uint) len+1; return 0; } -Load_log_event::Load_log_event(IO_CACHE* 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) + +char* sql_ex_info::init(char* buf,char* buf_end,bool use_new_format) { - char buf[LOAD_HEADER_LEN + 4]; - ulong data_len; - if (my_b_read(file, (byte*)buf, sizeof(buf)) || - my_b_read(file, (byte*)&sql_ex, sizeof(sql_ex))) - return; + cached_new_format = use_new_format; + if (use_new_format) + { + empty_flags=0; + /* + The code below assumes that buf will not disappear from + under our feet during the lifetime of the event. This assumption + holds true in the slave thread if the log is in new format, but is not + the case when we have old format because we will be reusing net buffer + to read the actual file before we write out the Create_file event. + */ + if (read_str(buf, buf_end, field_term, field_term_len) || + read_str(buf, buf_end, enclosed, enclosed_len) || + read_str(buf, buf_end, line_term, line_term_len) || + read_str(buf, buf_end, line_start, line_start_len) || + read_str(buf, buf_end, escaped, escaped_len)) + return 0; + opt_flags = *buf++; + } + else + { + field_term_len= enclosed_len= line_term_len= line_start_len= escaped_len=1; + field_term = buf++; // Use first byte in string + enclosed= buf++; + line_term= buf++; + line_start= buf++; + escaped= buf++; + opt_flags = *buf++; + empty_flags= *buf++; + if (empty_flags & FIELD_TERM_EMPTY) + field_term_len=0; + if (empty_flags & ENCLOSED_EMPTY) + enclosed_len=0; + if (empty_flags & LINE_TERM_EMPTY) + line_term_len=0; + if (empty_flags & LINE_START_EMPTY) + line_start_len=0; + if (empty_flags & ESCAPED_EMPTY) + escaped_len=0; + } + return buf; +} - data_len = uint4korr(buf) - LOAD_EVENT_OVERHEAD; - if (!(data_buf = (char*)my_malloc(data_len + 1, MYF(MY_WME)))) - return; - if (my_b_read(file, (byte*)data_buf, data_len)) - return; - copy_log_event(buf,data_len); + +#ifndef MYSQL_CLIENT +Load_log_event::Load_log_event(THD* thd, sql_exchange* ex, + const char* db_arg, const char* table_name_arg, + List<Item>& fields_arg, + enum enum_duplicates handle_dup) + :Log_event(thd),thread_id(thd->thread_id), num_fields(0),fields(0), + field_lens(0),field_block_len(0), + table_name(table_name_arg ? table_name_arg : ""), + db(db_arg), fname(ex->file_name) +{ + time_t end_time; + time(&end_time); + exec_time = (ulong) (end_time - thd->start_time); + /* db can never be a zero pointer in 4.0 */ + db_len = (uint32) strlen(db); + table_name_len = (uint32) strlen(table_name); + fname_len = (fname) ? (uint) strlen(fname) : 0; + sql_ex.field_term = (char*) ex->field_term->ptr(); + sql_ex.field_term_len = (uint8) ex->field_term->length(); + sql_ex.enclosed = (char*) ex->enclosed->ptr(); + sql_ex.enclosed_len = (uint8) ex->enclosed->length(); + sql_ex.line_term = (char*) ex->line_term->ptr(); + sql_ex.line_term_len = (uint8) ex->line_term->length(); + sql_ex.line_start = (char*) ex->line_start->ptr(); + sql_ex.line_start_len = (uint8) ex->line_start->length(); + sql_ex.escaped = (char*) ex->escaped->ptr(); + sql_ex.escaped_len = (uint8) ex->escaped->length(); + sql_ex.opt_flags = 0; + sql_ex.cached_new_format = -1; + + if (ex->dumpfile) + sql_ex.opt_flags |= DUMPFILE_FLAG; + if (ex->opt_enclosed) + sql_ex.opt_flags |= OPT_ENCLOSED_FLAG; + + sql_ex.empty_flags = 0; + + switch (handle_dup) { + case DUP_IGNORE: sql_ex.opt_flags |= IGNORE_FLAG; break; + case DUP_REPLACE: sql_ex.opt_flags |= REPLACE_FLAG; break; + case DUP_ERROR: break; + } + + if (!ex->field_term->length()) + sql_ex.empty_flags |= FIELD_TERM_EMPTY; + if (!ex->enclosed->length()) + sql_ex.empty_flags |= ENCLOSED_EMPTY; + if (!ex->line_term->length()) + sql_ex.empty_flags |= LINE_TERM_EMPTY; + if (!ex->line_start->length()) + sql_ex.empty_flags |= LINE_START_EMPTY; + if (!ex->escaped->length()) + sql_ex.empty_flags |= ESCAPED_EMPTY; + + skip_lines = ex->skip_lines; + + List_iterator<Item> li(fields_arg); + field_lens_buf.length(0); + fields_buf.length(0); + Item* item; + while ((item = li++)) + { + num_fields++; + uchar len = (uchar) strlen(item->name); + field_block_len += len + 1; + fields_buf.append(item->name, len + 1); + field_lens_buf.append((char*)&len, 1); + } + + field_lens = (const uchar*)field_lens_buf.ptr(); + fields = fields_buf.ptr(); } -Load_log_event::Load_log_event(const char* buf, int event_len): - Log_event(buf),data_buf(0),num_fields(0),fields(0), +#endif + +/* + The caller must do buf[event_len] = 0 before he starts using the + constructed event. +*/ + +Load_log_event::Load_log_event(const char* buf, int event_len, + bool old_format): + Log_event(buf, old_format),num_fields(0),fields(0), field_lens(0),field_block_len(0), table_name(0),db(0),fname(0) { - ulong data_len; - - if((uint)event_len < (LOAD_EVENT_OVERHEAD + LOG_EVENT_HEADER_LEN)) - return; - buf += EVENT_LEN_OFFSET; - memcpy(&sql_ex, buf + LOAD_HEADER_LEN + 4, sizeof(sql_ex)); - data_len = event_len; - - if(!(data_buf = (char*)my_malloc(data_len + 1, MYF(MY_WME)))) + if (!event_len) // derived class, will call copy_log_event() itself return; - memcpy(data_buf, buf + 22 + sizeof(sql_ex), data_len); - copy_log_event(buf, data_len); + copy_log_event(buf, event_len, old_format); } -void Load_log_event::copy_log_event(const char *buf, ulong data_len) +int Load_log_event::copy_log_event(const char *buf, ulong event_len, + bool old_format) { - thread_id = uint4korr(buf+4); - exec_time = uint4korr(buf+8); - skip_lines = uint4korr(buf + 12); - table_name_len = (uint)buf[16]; - db_len = (uint)buf[17]; - num_fields = uint4korr(buf + 18); + uint data_len; + char* buf_end = (char*)buf + event_len; + const char* data_head = buf + ((old_format) ? + OLD_HEADER_LEN : LOG_EVENT_HEADER_LEN); + thread_id = uint4korr(data_head + L_THREAD_ID_OFFSET); + exec_time = uint4korr(data_head + L_EXEC_TIME_OFFSET); + skip_lines = uint4korr(data_head + L_SKIP_LINES_OFFSET); + table_name_len = (uint)data_head[L_TBL_LEN_OFFSET]; + db_len = (uint)data_head[L_DB_LEN_OFFSET]; + num_fields = uint4korr(data_head + L_NUM_FIELDS_OFFSET); + int body_offset = ((buf[EVENT_TYPE_OFFSET] == LOAD_EVENT) ? + LOAD_HEADER_LEN + OLD_HEADER_LEN : + get_data_body_offset()); + + if ((int) event_len < body_offset) + return 1; + /* + Sql_ex.init() on success returns the pointer to the first byte after + the sql_ex structure, which is the start of field lengths array. + */ + if (!(field_lens=(uchar*)sql_ex.init((char*)buf + body_offset, + buf_end, + buf[EVENT_TYPE_OFFSET] != LOAD_EVENT))) + return 1; + + data_len = event_len - body_offset; if (num_fields > data_len) // simple sanity check against corruption - return; - - field_lens = (uchar*) data_buf; - uint i; - for (i = 0; i < num_fields; i++) - { + return 1; + for (uint i = 0; i < num_fields; i++) field_block_len += (uint)field_lens[i] + 1; - } + fields = (char*)field_lens + num_fields; - - *((char*)data_buf+data_len) = 0; table_name = fields + field_block_len; db = table_name + table_name_len + 1; fname = db + db_len + 1; - fname_len = data_len - 2 - db_len - table_name_len - num_fields - - field_block_len; + fname_len = strlen(fname); + // null termination is accomplished by the caller doing buf[event_len]=0 + return 0; } +#ifdef MYSQL_CLIENT void Load_log_event::print(FILE* file, bool short_form, char* last_db) { @@ -659,67 +1168,66 @@ void Load_log_event::print(FILE* file, bool short_form, char* last_db) } bool same_db = 0; - - if(db && last_db) - { - if(!(same_db = !memcmp(last_db, db, db_len + 1))) - memcpy(last_db, db, db_len + 1); - } + if (db && last_db) + { + if (!(same_db = !memcmp(last_db, db, db_len + 1))) + memcpy(last_db, db, db_len + 1); + } - if(db && db[0] && !same_db) + if (db && db[0] && !same_db) fprintf(file, "use %s;\n", db); - fprintf(file, "LOAD DATA INFILE '%s' ", fname); + fprintf(file, "LOAD DATA INFILE '%-*s' ", fname_len, fname); - if(sql_ex.opt_flags && REPLACE_FLAG ) + if (sql_ex.opt_flags && REPLACE_FLAG ) fprintf(file," REPLACE "); - else if(sql_ex.opt_flags && IGNORE_FLAG ) + else if (sql_ex.opt_flags && IGNORE_FLAG ) fprintf(file," IGNORE "); fprintf(file, "INTO TABLE %s ", table_name); - if(!(sql_ex.empty_flags & FIELD_TERM_EMPTY)) + if (sql_ex.field_term) { fprintf(file, " FIELDS TERMINATED BY "); - pretty_print_char(file, sql_ex.field_term); + pretty_print_str(file, sql_ex.field_term, sql_ex.field_term_len); } - if(!(sql_ex.empty_flags & ENCLOSED_EMPTY)) + if (sql_ex.enclosed) { - if(sql_ex.opt_flags && OPT_ENCLOSED_FLAG ) + if (sql_ex.opt_flags && OPT_ENCLOSED_FLAG ) fprintf(file," OPTIONALLY "); fprintf(file, " ENCLOSED BY "); - pretty_print_char(file, sql_ex.enclosed); + pretty_print_str(file, sql_ex.enclosed, sql_ex.enclosed_len); } - if(!(sql_ex.empty_flags & ESCAPED_EMPTY)) + if (sql_ex.escaped) { fprintf(file, " ESCAPED BY "); - pretty_print_char(file, sql_ex.escaped); + pretty_print_str(file, sql_ex.escaped, sql_ex.escaped_len); } - if(!(sql_ex.empty_flags & LINE_TERM_EMPTY)) + if (sql_ex.line_term) { fprintf(file," LINES TERMINATED BY "); - pretty_print_char(file, sql_ex.line_term); + pretty_print_str(file, sql_ex.line_term, sql_ex.line_term_len); } - if(!(sql_ex.empty_flags & LINE_START_EMPTY)) + if (sql_ex.line_start) { fprintf(file," LINES STARTING BY "); - pretty_print_char(file, sql_ex.line_start); + pretty_print_str(file, sql_ex.line_start, sql_ex.line_start_len); } - if((int)skip_lines > 0) + if ((int)skip_lines > 0) fprintf(file, " IGNORE %ld LINES ", (long) skip_lines); if (num_fields) { uint i; const char* field = fields; - fprintf( file, " ("); - for(i = 0; i < num_fields; i++) + fprintf(file, " ("); + for (i = 0; i < num_fields; i++) { - if(i) + if (i) fputc(',', file); fprintf(file, field); @@ -731,18 +1239,812 @@ void Load_log_event::print(FILE* file, bool short_form, char* last_db) fprintf(file, ";\n"); } +#endif /* #ifdef MYSQL_CLIENT */ + #ifndef MYSQL_CLIENT +void Log_event::set_log_pos(MYSQL_LOG* log) +{ + if (!log_pos) + log_pos = my_b_tell(&log->log_file); +} + + void Load_log_event::set_fields(List<Item> &fields) { uint i; const char* field = this->fields; - for(i = 0; i < num_fields; i++) + for (i = 0; i < num_fields; i++) + { + fields.push_back(new Item_field(db, table_name, field)); + field += field_lens[i] + 1; + } +} + + +Slave_log_event::Slave_log_event(THD* thd_arg, + struct st_relay_log_info* rli): + Log_event(thd_arg),mem_pool(0),master_host(0) +{ + DBUG_ENTER("Slave_log_event"); + if (!rli->inited) + DBUG_VOID_RETURN; + + MASTER_INFO* mi = rli->mi; + // TODO: re-write this better without holding both locks at the same time + pthread_mutex_lock(&mi->data_lock); + pthread_mutex_lock(&rli->data_lock); + master_host_len = strlen(mi->host); + master_log_len = strlen(rli->master_log_name); + // on OOM, just do not initialize the structure and print the error + if ((mem_pool = (char*)my_malloc(get_data_size() + 1, + MYF(MY_WME)))) + { + master_host = mem_pool + SL_MASTER_HOST_OFFSET ; + memcpy(master_host, mi->host, master_host_len + 1); + master_log = master_host + master_host_len + 1; + memcpy(master_log, rli->master_log_name, master_log_len + 1); + master_port = mi->port; + master_pos = rli->master_log_pos; + DBUG_PRINT("info", ("master_log: %s pos: %d", master_log, + (ulong) master_pos)); + } + else + sql_print_error("Out of memory while recording slave event"); + pthread_mutex_unlock(&rli->data_lock); + pthread_mutex_unlock(&mi->data_lock); + DBUG_VOID_RETURN; +} + +#endif /* ! MYSQL_CLIENT */ + + +Slave_log_event::~Slave_log_event() +{ + my_free(mem_pool, MYF(MY_ALLOW_ZERO_PTR)); +} + +#ifdef MYSQL_CLIENT + +void Slave_log_event::print(FILE* file, bool short_form, char* last_db) +{ + char llbuff[22]; + if (short_form) + return; + print_header(file); + fputc('\n', file); + fprintf(file, "Slave: master_host: '%s' master_port: %d \ +master_log: '%s' master_pos: %s\n", + master_host, master_port, master_log, llstr(master_pos, llbuff)); +} + +#endif /* MYSQL_CLIENT */ + +int Slave_log_event::get_data_size() +{ + return master_host_len + master_log_len + 1 + SL_MASTER_HOST_OFFSET; +} + +int Slave_log_event::write_data(IO_CACHE* file) +{ + int8store(mem_pool + SL_MASTER_POS_OFFSET, master_pos); + int2store(mem_pool + SL_MASTER_PORT_OFFSET, master_port); + // log and host are already there + return my_b_safe_write(file, (byte*)mem_pool, get_data_size()); +} + + +void Slave_log_event::init_from_mem_pool(int data_size) +{ + master_pos = uint8korr(mem_pool + SL_MASTER_POS_OFFSET); + master_port = uint2korr(mem_pool + SL_MASTER_PORT_OFFSET); + master_host = mem_pool + SL_MASTER_HOST_OFFSET; + master_host_len = strlen(master_host); + // safety + master_log = master_host + master_host_len + 1; + if (master_log > mem_pool + data_size) + { + master_host = 0; + return; + } + master_log_len = strlen(master_log); +} + +Slave_log_event::Slave_log_event(const char* buf, int event_len) + :Log_event(buf,0),mem_pool(0),master_host(0) +{ + event_len -= LOG_EVENT_HEADER_LEN; + if (event_len < 0) + return; + if (!(mem_pool = (char*) my_malloc(event_len + 1, MYF(MY_WME)))) + return; + memcpy(mem_pool, buf + LOG_EVENT_HEADER_LEN, event_len); + mem_pool[event_len] = 0; + init_from_mem_pool(event_len); +} + +#ifndef MYSQL_CLIENT +Create_file_log_event::Create_file_log_event(THD* thd_arg, sql_exchange* ex, + const char* db_arg, const char* table_name_arg, + List<Item>& fields_arg, enum enum_duplicates handle_dup, + char* block_arg, uint block_len_arg) + :Load_log_event(thd_arg,ex,db_arg,table_name_arg,fields_arg,handle_dup), + fake_base(0),block(block_arg),block_len(block_len_arg), + file_id(thd_arg->file_id = mysql_bin_log.next_file_id()) +{ + sql_ex.force_new_format(); +} +#endif + +int Create_file_log_event::write_data_body(IO_CACHE* file) +{ + int res; + if ((res = Load_log_event::write_data_body(file)) || fake_base) + return res; + return (my_b_safe_write(file, (byte*) "", 1) || + my_b_safe_write(file, (byte*) block, block_len)); +} + +int Create_file_log_event::write_data_header(IO_CACHE* file) +{ + int res; + if ((res = Load_log_event::write_data_header(file)) || fake_base) + return res; + byte buf[CREATE_FILE_HEADER_LEN]; + int4store(buf + CF_FILE_ID_OFFSET, file_id); + return my_b_safe_write(file, buf, CREATE_FILE_HEADER_LEN); +} + +int Create_file_log_event::write_base(IO_CACHE* file) +{ + int res; + fake_base = 1; // pretend we are Load event + res = write(file); + fake_base = 0; + return res; +} + +Create_file_log_event::Create_file_log_event(const char* buf, int len, + bool old_format) + :Load_log_event(buf,0,old_format),fake_base(0),block(0),inited_from_old(0) +{ + int block_offset; + if (copy_log_event(buf,len,old_format)) + return; + if (!old_format) + { + file_id = uint4korr(buf + LOG_EVENT_HEADER_LEN + + + LOAD_HEADER_LEN + CF_FILE_ID_OFFSET); + // + 1 for \0 terminating fname + block_offset = (LOG_EVENT_HEADER_LEN + Load_log_event::get_data_size() + + CREATE_FILE_HEADER_LEN + 1); + if (len < block_offset) + return; + block = (char*)buf + block_offset; + block_len = len - block_offset; + } + else + { + sql_ex.force_new_format(); + inited_from_old = 1; + } +} + + +#ifdef MYSQL_CLIENT +void Create_file_log_event::print(FILE* file, bool short_form, + char* last_db) +{ + if (short_form) + return; + Load_log_event::print(file, 1, last_db); + fprintf(file, " file_id: %d block_len: %d\n", file_id, block_len); +} +#endif + +#ifndef MYSQL_CLIENT +void Create_file_log_event::pack_info(String* packet) +{ + char buf1[256],buf[22], *end; + String tmp(buf1, sizeof(buf1)); + tmp.length(0); + tmp.append("db="); + tmp.append(db, db_len); + tmp.append(";table="); + tmp.append(table_name, table_name_len); + tmp.append(";file_id="); + end= int10_to_str((long) file_id, buf, 10); + tmp.append(buf, (uint32) (end-buf)); + tmp.append(";block_len="); + end= int10_to_str((long) block_len, buf, 10); + tmp.append(buf, (uint32) (end-buf)); + net_store_data(packet, (char*) tmp.ptr(), tmp.length()); +} +#endif + +#ifndef MYSQL_CLIENT +Append_block_log_event::Append_block_log_event(THD* thd_arg, char* block_arg, + uint block_len_arg) + :Log_event(thd_arg), block(block_arg),block_len(block_len_arg), + file_id(thd_arg->file_id) +{ +} +#endif + +Append_block_log_event::Append_block_log_event(const char* buf, int len) + :Log_event(buf, 0),block(0) +{ + if ((uint)len < APPEND_BLOCK_EVENT_OVERHEAD) + return; + file_id = uint4korr(buf + LOG_EVENT_HEADER_LEN + AB_FILE_ID_OFFSET); + block = (char*)buf + APPEND_BLOCK_EVENT_OVERHEAD; + block_len = len - APPEND_BLOCK_EVENT_OVERHEAD; +} + +int Append_block_log_event::write_data(IO_CACHE* file) +{ + byte buf[APPEND_BLOCK_HEADER_LEN]; + int4store(buf + AB_FILE_ID_OFFSET, file_id); + return (my_b_safe_write(file, buf, APPEND_BLOCK_HEADER_LEN) || + my_b_safe_write(file, (byte*) block, block_len)); +} + +#ifdef MYSQL_CLIENT +void Append_block_log_event::print(FILE* file, bool short_form, + char* last_db) +{ + if (short_form) + return; + print_header(file); + fputc('\n', file); + fprintf(file, "#Append_block: file_id: %d block_len: %d\n", + file_id, block_len); +} +#endif + +#ifndef MYSQL_CLIENT +void Append_block_log_event::pack_info(String* packet) +{ + char buf1[256]; + sprintf(buf1, ";file_id=%u;block_len=%u", file_id, block_len); + net_store_data(packet, buf1); +} + +Delete_file_log_event::Delete_file_log_event(THD* thd_arg) + :Log_event(thd_arg),file_id(thd_arg->file_id) +{ +} +#endif + + +Delete_file_log_event::Delete_file_log_event(const char* buf, int len) + :Log_event(buf, 0),file_id(0) +{ + if ((uint)len < DELETE_FILE_EVENT_OVERHEAD) + return; + file_id = uint4korr(buf + LOG_EVENT_HEADER_LEN + AB_FILE_ID_OFFSET); +} + + +int Delete_file_log_event::write_data(IO_CACHE* file) +{ + byte buf[DELETE_FILE_HEADER_LEN]; + int4store(buf + DF_FILE_ID_OFFSET, file_id); + return my_b_safe_write(file, buf, DELETE_FILE_HEADER_LEN); +} + +#ifdef MYSQL_CLIENT +void Delete_file_log_event::print(FILE* file, bool short_form, + char* last_db) +{ + if (short_form) + return; + print_header(file); + fputc('\n', file); + fprintf(file, "#Delete_file: file_id=%u\n", file_id); +} +#endif + +#ifndef MYSQL_CLIENT +void Delete_file_log_event::pack_info(String* packet) +{ + char buf1[64]; + sprintf(buf1, ";file_id=%u", (uint) file_id); + net_store_data(packet, buf1); +} +#endif + + +#ifndef MYSQL_CLIENT +Execute_load_log_event::Execute_load_log_event(THD* thd_arg) + :Log_event(thd_arg),file_id(thd_arg->file_id) +{ +} +#endif + +Execute_load_log_event::Execute_load_log_event(const char* buf,int len) + :Log_event(buf, 0),file_id(0) +{ + if ((uint)len < EXEC_LOAD_EVENT_OVERHEAD) + return; + file_id = uint4korr(buf + LOG_EVENT_HEADER_LEN + EL_FILE_ID_OFFSET); +} + +int Execute_load_log_event::write_data(IO_CACHE* file) +{ + byte buf[EXEC_LOAD_HEADER_LEN]; + int4store(buf + EL_FILE_ID_OFFSET, file_id); + return my_b_safe_write(file, buf, EXEC_LOAD_HEADER_LEN); +} + +#ifdef MYSQL_CLIENT +void Execute_load_log_event::print(FILE* file, bool short_form, + char* last_db) +{ + if (short_form) + return; + print_header(file); + fputc('\n', file); + fprintf(file, "#Exec_load: file_id=%d\n", + file_id); +} +#endif +#ifndef MYSQL_CLIENT +void Execute_load_log_event::pack_info(String* packet) +{ + char buf[64]; + sprintf(buf, ";file_id=%u", (uint) file_id); + net_store_data(packet, buf); +} +#endif + +#ifndef MYSQL_CLIENT +int Query_log_event::exec_event(struct st_relay_log_info* rli) +{ + int expected_error,actual_error = 0; + init_sql_alloc(&thd->mem_root, 8192,0); + thd->db = rewrite_db((char*)db); + DBUG_ASSERT(q_len == strlen(query)); + if (db_ok(thd->db, replicate_do_db, replicate_ignore_db)) + { + thd->query = (char*)query; + thd->set_time((time_t)when); + thd->current_tablenr = 0; + VOID(pthread_mutex_lock(&LOCK_thread_count)); + thd->query_id = query_id++; + VOID(pthread_mutex_unlock(&LOCK_thread_count)); + thd->query_error = 0; // clear error + thd->net.last_errno = 0; + thd->net.last_error[0] = 0; + thd->slave_proxy_id = thread_id; // for temp tables + + /* + Sanity check to make sure the master did not get a really bad + error on the query. + */ + if (ignored_error_code((expected_error = error_code)) || + !check_expected_error(thd,rli,expected_error)) { - fields.push_back(new Item_field(db, table_name, field)); - field += field_lens[i] + 1; + mysql_parse(thd, thd->query, q_len); + if ((expected_error != (actual_error = thd->net.last_errno)) && + expected_error && + !ignored_error_code(actual_error) && + !ignored_error_code(expected_error)) + { + const char* errmsg = "Slave: did not get the expected error\ + running query from master - expected: '%s' (%d), got '%s' (%d)"; + sql_print_error(errmsg, ER_SAFE(expected_error), + expected_error, + actual_error ? thd->net.last_error: "no error", + actual_error); + thd->query_error = 1; + } + else if (expected_error == actual_error || + ignored_error_code(actual_error)) + { + thd->query_error = 0; + *rli->last_slave_error = 0; + rli->last_slave_errno = 0; + } + } + else + { + // master could be inconsistent, abort and tell DBA to check/fix it + thd->db = thd->query = 0; + thd->variables.convert_set = 0; + close_thread_tables(thd); + free_root(&thd->mem_root,0); + return 1; } + } + thd->db= 0; // prevent db from being freed + thd->query= 0; // just to be sure + // assume no convert for next query unless set explictly + thd->variables.convert_set = 0; + close_thread_tables(thd); + + if (thd->query_error || thd->fatal_error) + { + slave_print_error(rli,actual_error, "error '%s' on query '%s'", + actual_error ? thd->net.last_error : + "unexpected success or fatal error", query); + free_root(&thd->mem_root,0); + return 1; + } + free_root(&thd->mem_root,0); + return Log_event::exec_event(rli); +} + + +int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli) +{ + init_sql_alloc(&thd->mem_root, 8192,0); + thd->db = rewrite_db((char*)db); + thd->query = 0; + thd->query_error = 0; + + if (db_ok(thd->db, replicate_do_db, replicate_ignore_db)) + { + thd->set_time((time_t)when); + thd->current_tablenr = 0; + VOID(pthread_mutex_lock(&LOCK_thread_count)); + thd->query_id = query_id++; + VOID(pthread_mutex_unlock(&LOCK_thread_count)); + + TABLE_LIST tables; + bzero((char*) &tables,sizeof(tables)); + tables.db = thd->db; + tables.name = tables.real_name = (char*)table_name; + tables.lock_type = TL_WRITE; + // the table will be opened in mysql_load + if (table_rules_on && !tables_ok(thd, &tables)) + { + // TODO: this is a bug - this needs to be moved to the I/O thread + if (net) + skip_load_data_infile(net); + } + else + { + char llbuff[22]; + enum enum_duplicates handle_dup = DUP_IGNORE; + if (sql_ex.opt_flags && REPLACE_FLAG) + handle_dup = DUP_REPLACE; + sql_exchange ex((char*)fname, sql_ex.opt_flags && + DUMPFILE_FLAG ); + String field_term(sql_ex.field_term,sql_ex.field_term_len); + String enclosed(sql_ex.enclosed,sql_ex.enclosed_len); + String line_term(sql_ex.line_term,sql_ex.line_term_len); + String line_start(sql_ex.line_start,sql_ex.line_start_len); + String escaped(sql_ex.escaped,sql_ex.escaped_len); + + ex.opt_enclosed = (sql_ex.opt_flags & OPT_ENCLOSED_FLAG); + if (sql_ex.empty_flags & FIELD_TERM_EMPTY) + ex.field_term->length(0); + + ex.skip_lines = skip_lines; + List<Item> fields; + set_fields(fields); + thd->slave_proxy_id = thd->thread_id; + if (net) + { + // mysql_load will use thd->net to read the file + thd->net.vio = net->vio; + /* + Make sure the client does not get confused about the packet sequence + */ + thd->net.pkt_nr = net->pkt_nr; + } + if (mysql_load(thd, &ex, &tables, fields, handle_dup, net != 0, + TL_WRITE)) + thd->query_error = 1; + if (thd->cuted_fields) + sql_print_error("Slave: load data infile at position %s in log \ +'%s' produced %d warning(s)", llstr(rli->master_log_pos,llbuff), RPL_LOG_NAME, + thd->cuted_fields ); + if (net) + net->pkt_nr= thd->net.pkt_nr; + } + } + else + { + /* + We will just ask the master to send us /dev/null if we do not + want to load the data. + TODO: this a bug - needs to be done in I/O thread + */ + if (net) + skip_load_data_infile(net); + } + + thd->net.vio = 0; + thd->db= 0; // prevent db from being freed + close_thread_tables(thd); + if (thd->query_error) + { + int sql_error = thd->net.last_errno; + if (!sql_error) + sql_error = ER_UNKNOWN_ERROR; + + slave_print_error(rli,sql_error, + "Slave: Error '%s' running load data infile ", + ER_SAFE(sql_error)); + free_root(&thd->mem_root,0); + return 1; + } + free_root(&thd->mem_root,0); + + if (thd->fatal_error) + { + sql_print_error("Slave: Fatal error running LOAD DATA INFILE "); + return 1; + } + + return Log_event::exec_event(rli); +} + + +int Start_log_event::exec_event(struct st_relay_log_info* rli) +{ + close_temporary_tables(thd); + /* + If we have old format, load_tmpdir is cleaned up by the I/O thread + + TODO: cleanup_load_tmpdir() needs to remove only the files associated + with the server id that has just started + */ + if (!rli->mi->old_format) + cleanup_load_tmpdir(); + return Log_event::exec_event(rli); +} + + +int Stop_log_event::exec_event(struct st_relay_log_info* rli) +{ + // do not clean up immediately after rotate event + if (rli->master_log_pos > BIN_LOG_HEADER_SIZE) + { + close_temporary_tables(thd); + cleanup_load_tmpdir(); + } + /* + We do not want to update master_log pos because we get a rotate event + before stop, so by now master_log_name is set to the next log + if we updated it, we will have incorrect master coordinates and this + could give false triggers in MASTER_POS_WAIT() that we have reached + the targed position when in fact we have not + */ + rli->inc_pos(get_event_len(), 0); + flush_relay_log_info(rli); + return 0; +} + +int Rotate_log_event::exec_event(struct st_relay_log_info* rli) +{ + bool rotate_binlog = 0, write_slave_event = 0; + char* log_name = rli->master_log_name; + DBUG_ENTER("Rotate_log_event::exec_event"); + + pthread_mutex_lock(&rli->data_lock); + /* + TODO: probably needs re-write + rotate local binlog only if the name of remote has changed + */ + if (!*log_name || !(log_name[ident_len] == 0 && + !memcmp(log_name, new_log_ident, ident_len))) + { + write_slave_event = (!(flags & LOG_EVENT_FORCED_ROTATE_F) + && mysql_bin_log.is_open()); + rotate_binlog = (*log_name && write_slave_event); + if (ident_len >= sizeof(rli->master_log_name)) + { + pthread_mutex_unlock(&rli->data_lock); + DBUG_RETURN(1); + } + memcpy(log_name, new_log_ident,ident_len); + log_name[ident_len] = 0; + } + rli->master_log_pos = pos; + rli->relay_log_pos += get_event_len(); + if (rotate_binlog) + { + mysql_bin_log.new_file(); + rli->master_log_pos = 4; + } + DBUG_PRINT("info", ("master_log_pos: %d", (ulong) rli->master_log_pos)); + pthread_cond_broadcast(&rli->data_cond); + pthread_mutex_unlock(&rli->data_lock); + flush_relay_log_info(rli); + + if (write_slave_event) + { + Slave_log_event s(thd, rli); + if (s.master_host) + { + s.set_log_pos(&mysql_bin_log); + s.server_id = ::server_id; + mysql_bin_log.write(&s); + } + } + DBUG_RETURN(0); +} + +int Intvar_log_event::exec_event(struct st_relay_log_info* rli) +{ + switch (type) { + case LAST_INSERT_ID_EVENT: + thd->last_insert_id_used = 1; + thd->last_insert_id = val; + break; + case INSERT_ID_EVENT: + thd->next_insert_id = val; + break; + } + rli->inc_pending(get_event_len()); + return 0; +} + +int Slave_log_event::exec_event(struct st_relay_log_info* rli) +{ + if (mysql_bin_log.is_open()) + mysql_bin_log.write(this); + return Log_event::exec_event(rli); +} + +int Create_file_log_event::exec_event(struct st_relay_log_info* rli) +{ + char fname_buf[FN_REFLEN+10]; + char *p; + int fd = -1; + IO_CACHE file; + int error = 1; + + bzero((char*)&file, sizeof(file)); + p = slave_load_file_stem(fname_buf, file_id, server_id); + strmov(p, ".info"); // strmov takes less code than memcpy + if ((fd = my_open(fname_buf, O_WRONLY|O_CREAT|O_BINARY|O_TRUNC, + MYF(MY_WME))) < 0 || + init_io_cache(&file, fd, IO_SIZE, WRITE_CACHE, (my_off_t)0, 0, + MYF(MY_WME|MY_NABP))) + { + slave_print_error(rli,my_errno, "Could not open file '%s'", fname_buf); + goto err; + } + // a trick to avoid allocating another buffer + strmov(p, ".data"); + fname = fname_buf; + fname_len = (uint)(p-fname) + 5; + if (write_base(&file)) + { + strmov(p, ".info"); // to have it right in the error message + slave_print_error(rli,my_errno, "Could not write to file '%s'", fname_buf); + goto err; + } + end_io_cache(&file); + my_close(fd, MYF(0)); + + // fname_buf now already has .data, not .info, because we did our trick + if ((fd = my_open(fname_buf, O_WRONLY|O_CREAT|O_BINARY|O_TRUNC, + MYF(MY_WME))) < 0) + { + slave_print_error(rli,my_errno, "Could not open file '%s'", fname_buf); + goto err; + } + if (my_write(fd, (byte*) block, block_len, MYF(MY_WME+MY_NABP))) + { + slave_print_error(rli,my_errno, "Write to '%s' failed", fname_buf); + goto err; + } + if (mysql_bin_log.is_open()) + mysql_bin_log.write(this); + error=0; // Everything is ok + +err: + if (error) + end_io_cache(&file); + if (fd >= 0) + my_close(fd, MYF(0)); + return error ? 1 : Log_event::exec_event(rli); } -#endif +int Delete_file_log_event::exec_event(struct st_relay_log_info* rli) +{ + char fname[FN_REFLEN+10]; + char *p= slave_load_file_stem(fname, file_id, server_id); + memcpy(p, ".data", 6); + (void) my_delete(fname, MYF(MY_WME)); + memcpy(p, ".info", 6); + (void) my_delete(fname, MYF(MY_WME)); + if (mysql_bin_log.is_open()) + mysql_bin_log.write(this); + return Log_event::exec_event(rli); +} + +int Append_block_log_event::exec_event(struct st_relay_log_info* rli) +{ + char fname[FN_REFLEN+10]; + char *p= slave_load_file_stem(fname, file_id, server_id); + int fd; + int error = 1; + + memcpy(p, ".data", 6); + if ((fd = my_open(fname, O_WRONLY|O_APPEND|O_BINARY, MYF(MY_WME))) < 0) + { + slave_print_error(rli,my_errno, "Could not open file '%s'", fname); + goto err; + } + if (my_write(fd, (byte*) block, block_len, MYF(MY_WME+MY_NABP))) + { + slave_print_error(rli,my_errno, "Write to '%s' failed", fname); + goto err; + } + if (mysql_bin_log.is_open()) + mysql_bin_log.write(this); + error=0; + +err: + if (fd >= 0) + my_close(fd, MYF(0)); + return error ? error : Log_event::exec_event(rli); +} + +int Execute_load_log_event::exec_event(struct st_relay_log_info* rli) +{ + char fname[FN_REFLEN+10]; + char *p= slave_load_file_stem(fname, file_id, server_id); + int fd; + int error = 1; + ulong save_options; + IO_CACHE file; + Load_log_event* lev = 0; + + memcpy(p, ".info", 6); + if ((fd = my_open(fname, O_RDONLY|O_BINARY, MYF(MY_WME))) < 0 || + init_io_cache(&file, fd, IO_SIZE, READ_CACHE, (my_off_t)0, 0, + MYF(MY_WME|MY_NABP))) + { + slave_print_error(rli,my_errno, "Could not open file '%s'", fname); + goto err; + } + if (!(lev = (Load_log_event*)Log_event::read_log_event(&file, + (pthread_mutex_t*)0, + (bool)0)) || + lev->get_type_code() != NEW_LOAD_EVENT) + { + slave_print_error(rli,0, "File '%s' appears corrupted", fname); + goto err; + } + /* + We want to disable binary logging in slave thread because we need the file + events to appear in the same order as they do on the master relative to + other events, so that we can preserve ascending order of log sequence + numbers - needed to handle failover . + */ + save_options = thd->options; + thd->options &= ~ (ulong) (OPTION_BIN_LOG); + lev->thd = thd; + if (lev->exec_event(0,0)) + { + slave_print_error(rli,my_errno, "Failed executing load from '%s'", fname); + thd->options = save_options; + goto err; + } + thd->options = save_options; + (void) my_delete(fname, MYF(MY_WME)); + memcpy(p, ".data", 6); + (void) my_delete(fname, MYF(MY_WME)); + if (mysql_bin_log.is_open()) + mysql_bin_log.write(this); + error = 0; + +err: + delete lev; + if (fd >= 0) + { + my_close(fd, MYF(0)); + end_io_cache(&file); + } + return error ? error : Log_event::exec_event(rli); +} + +#endif /* !MYSQL_CLIENT */ |