new option --local-load in mysqlbinlog

parent abf49f32
...@@ -56,6 +56,9 @@ static short binlog_flags = 0; ...@@ -56,6 +56,9 @@ static short binlog_flags = 0;
static MYSQL* mysql = NULL; static MYSQL* mysql = NULL;
static const char* table = 0; static const char* table = 0;
static bool use_local_load= 0;
static const char* dirname_for_local_load= 0;
static void dump_local_log_entries(const char* logname); static void dump_local_log_entries(const char* logname);
static void dump_remote_log_entries(const char* logname); static void dump_remote_log_entries(const char* logname);
static void dump_log_entries(const char* logname); static void dump_log_entries(const char* logname);
...@@ -64,6 +67,129 @@ static void dump_remote_table(NET* net, const char* db, const char* table); ...@@ -64,6 +67,129 @@ static void dump_remote_table(NET* net, const char* db, const char* table);
static void die(const char* fmt, ...); static void die(const char* fmt, ...);
static MYSQL* safe_connect(); static MYSQL* safe_connect();
class Load_log_processor {
char target_dir_name[MY_NFILE];
int target_dir_name_len;
DYNAMIC_ARRAY file_names;
const char* create_file(Create_file_log_event *ce)
{
const char *bname= ce->fname + ce->fname_len -1;
while (bname>ce->fname && bname[-1]!=FN_LIBCHAR)
bname--;
uint blen= ce->fname_len - (bname-ce->fname);
uint full_len= target_dir_name_len + blen;
char *tmp;
if (!(tmp= my_malloc(full_len + 9 + 1,MYF(MY_WME))) ||
set_dynamic(&file_names,(gptr)&ce,ce->file_id))
{
die("Could not construct local filename %s%s",target_dir_name,bname);
return 0;
}
char *ptr= tmp;
memcpy(ptr,target_dir_name,target_dir_name_len);
ptr+= target_dir_name_len;
memcpy(ptr,bname,blen);
ptr+= blen;
sprintf(ptr,"-%08x",ce->file_id);
ce->set_fname_outside_temp_buf(tmp,full_len);
return tmp;
}
void append_to_file(const char* fname, int flags,
gptr data, uint size)
{
FILE *file;
if(!(file= my_fopen(fname,flags,MYF(MY_WME))))
exit(1);
if (my_fwrite(file,data,size,MYF(MY_WME|MY_NABP)))
exit(1);
if (my_fclose(file,MYF(MY_WME)))
exit(1);
}
public:
Load_log_processor()
{
init_dynamic_array(&file_names,sizeof(Create_file_log_event*),
100,100 CALLER_INFO);
}
~Load_log_processor()
{
destroy();
delete_dynamic(&file_names);
}
void init_by_dir_name(const char *atarget_dir_name)
{
char *end= strmov(target_dir_name,atarget_dir_name);
if (end[-1]!=FN_LIBCHAR)
*end++= FN_LIBCHAR;
target_dir_name_len= end-target_dir_name;
}
void init_by_file_name(const char *file_name)
{
int len= strlen(file_name);
const char *end= file_name + len - 1;
while (end>file_name && *end!=FN_LIBCHAR)
end--;
if (*end!=FN_LIBCHAR)
target_dir_name_len= 0;
else
{
target_dir_name_len= end - file_name + 1;
memmove(target_dir_name,file_name,target_dir_name_len);
}
}
void init_by_cur_dir()
{
target_dir_name_len= 0;
}
void destroy()
{
Create_file_log_event **ptr= (Create_file_log_event**)file_names.buffer;
Create_file_log_event **end= ptr + file_names.elements;
for (; ptr<end; ptr++)
{
if (*ptr)
{
my_free((char*)(*ptr)->fname,MYF(MY_WME));
delete *ptr;
*ptr= 0;
}
}
}
Create_file_log_event *grab_event(uint file_id)
{
Create_file_log_event **ptr=
(Create_file_log_event**)file_names.buffer + file_id;
Create_file_log_event *res= *ptr;
*ptr= 0;
return res;
}
void process(Create_file_log_event *ce)
{
const char *fname= create_file(ce);
append_to_file (fname,O_CREAT|O_BINARY,ce->block,ce->block_len);
}
void process(Append_block_log_event *ae)
{
if (ae->file_id >= file_names.elements)
die("Skiped CreateFile event for file_id: %u",ae->file_id);
Create_file_log_event* ce=
*((Create_file_log_event**)file_names.buffer + ae->file_id);
append_to_file(ce->fname,O_APPEND|O_BINARY,ae->block,ae->block_len);
}
};
Load_log_processor load_processor;
static struct my_option my_long_options[] = static struct my_option my_long_options[] =
{ {
#ifndef DBUG_OFF #ifndef DBUG_OFF
...@@ -97,6 +223,9 @@ static struct my_option my_long_options[] = ...@@ -97,6 +223,9 @@ static struct my_option my_long_options[] =
{"user", 'u', "Connect to the remote server as username", {"user", 'u', "Connect to the remote server as username",
(gptr*) &user, (gptr*) &user, 0, GET_STR_ALLOC, REQUIRED_ARG, 0, 0, 0, 0, (gptr*) &user, (gptr*) &user, 0, GET_STR_ALLOC, REQUIRED_ARG, 0, 0, 0, 0,
0, 0}, 0, 0},
{"local-load", 'l', "Prepare files for local load in directory",
(gptr*) &dirname_for_local_load, (gptr*) &dirname_for_local_load, 0,
GET_STR_ALLOC, OPT_ARG, 0, 0, 0, 0, 0, 0},
{"version", 'V', "Print version and exit.", 0, 0, 0, GET_NO_ARG, NO_ARG, 0, {"version", 'V', "Print version and exit.", 0, 0, 0, GET_NO_ARG, NO_ARG, 0,
0, 0, 0, 0, 0}, 0, 0, 0, 0, 0},
{0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0} {0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
...@@ -209,6 +338,9 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)), ...@@ -209,6 +338,9 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
case 'V': case 'V':
print_version(); print_version();
exit(0); exit(0);
case 'l':
use_local_load= 1;
break;
case '?': case '?':
usage(); usage();
exit(0); exit(0);
...@@ -420,6 +552,8 @@ static void dump_local_log_entries(const char* logname) ...@@ -420,6 +552,8 @@ static void dump_local_log_entries(const char* logname)
MYF(MY_WME | MY_NABP))) MYF(MY_WME | MY_NABP)))
exit(1); exit(1);
old_format = check_header(file); old_format = check_header(file);
if (use_local_load && !dirname_for_local_load)
load_processor.init_by_file_name(logname);
} }
else else
{ {
...@@ -441,6 +575,8 @@ static void dump_local_log_entries(const char* logname) ...@@ -441,6 +575,8 @@ static void dump_local_log_entries(const char* logname)
} }
file->pos_in_file=position; file->pos_in_file=position;
file->seek_not_done=0; file->seek_not_done=0;
if (use_local_load && !dirname_for_local_load)
load_processor.init_by_cur_dir();
} }
if (!position) if (!position)
...@@ -495,11 +631,43 @@ Could not read entry at offset %s : Error in log format or read error", ...@@ -495,11 +631,43 @@ Could not read entry at offset %s : Error in log format or read error",
} }
if (!short_form) if (!short_form)
fprintf(result_file, "# at %s\n",llstr(old_off,llbuff)); fprintf(result_file, "# at %s\n",llstr(old_off,llbuff));
ev->print(result_file, short_form, last_db); if (!use_local_load)
ev->print(result_file, short_form, last_db);
else
{
switch(ev->get_type_code())
{
case CREATE_FILE_EVENT:
{
Create_file_log_event* ce= (Create_file_log_event*)ev;
ce->print(result_file, short_form, last_db,true);
load_processor.process(ce);
ev= 0;
break;
}
case APPEND_BLOCK_EVENT:
ev->print(result_file, short_form, last_db);
load_processor.process((Append_block_log_event*)ev);
break;
case EXEC_LOAD_EVENT:
{
ev->print(result_file, short_form, last_db);
Execute_load_log_event *exv= (Execute_load_log_event*)ev;
Create_file_log_event *ce= load_processor.grab_event(exv->file_id);
ce->print(result_file, short_form, last_db,true);
my_free((char*)ce->fname,MYF(MY_WME));
delete ce;
break;
}
default:
ev->print(result_file, short_form, last_db);
}
}
} }
rec_count++; rec_count++;
delete ev; if (ev)
delete ev;
} }
if (fd >= 0) if (fd >= 0)
my_close(fd, MYF(MY_WME)); my_close(fd, MYF(MY_WME));
...@@ -520,6 +688,8 @@ int main(int argc, char** argv) ...@@ -520,6 +688,8 @@ int main(int argc, char** argv)
if (use_remote) if (use_remote)
mysql = safe_connect(); mysql = safe_connect();
if (dirname_for_local_load)
load_processor.init_by_dir_name(dirname_for_local_load);
if (table) if (table)
{ {
......
...@@ -1333,7 +1333,10 @@ void Load_log_event::print(FILE* file, bool short_form, char* last_db) ...@@ -1333,7 +1333,10 @@ void Load_log_event::print(FILE* file, bool short_form, char* last_db)
if (db && db[0] && !same_db) if (db && db[0] && !same_db)
fprintf(file, "use %s;\n", db); fprintf(file, "use %s;\n", db);
fprintf(file, "LOAD DATA INFILE '%-*s' ", fname_len, fname); fprintf(file, "LOAD ");
if (check_fname_outside_temp_buf())
fprintf(file, "LOCAL ");
fprintf(file, "DATA INFILE '%-*s' ", fname_len, fname);
if (sql_ex.opt_flags && REPLACE_FLAG ) if (sql_ex.opt_flags && REPLACE_FLAG )
fprintf(file," REPLACE "); fprintf(file," REPLACE ");
...@@ -2220,14 +2223,32 @@ Create_file_log_event::Create_file_log_event(const char* buf, int len, ...@@ -2220,14 +2223,32 @@ Create_file_log_event::Create_file_log_event(const char* buf, int len,
****************************************************************************/ ****************************************************************************/
#ifdef MYSQL_CLIENT #ifdef MYSQL_CLIENT
void Create_file_log_event::print(FILE* file, bool short_form, void Create_file_log_event::print(FILE* file, bool short_form,
char* last_db) char* last_db, bool enable_local)
{ {
if (short_form) if (short_form)
{
if (enable_local && check_fname_outside_temp_buf())
Load_log_event::print(file, 1, last_db);
return; return;
Load_log_event::print(file, 1, last_db); }
if (enable_local)
{
if (!check_fname_outside_temp_buf())
fprintf(file, "#");
Load_log_event::print(file, 1, last_db);
fprintf(file, "#");
}
fprintf(file, " file_id: %d block_len: %d\n", file_id, block_len); fprintf(file, " file_id: %d block_len: %d\n", file_id, block_len);
} }
void Create_file_log_event::print(FILE* file, bool short_form,
char* last_db)
{
print(file,short_form,last_db,0);
}
#endif // MYSQL_CLIENT #endif // MYSQL_CLIENT
/***************************************************************************** /*****************************************************************************
......
...@@ -445,6 +445,13 @@ class Load_log_event: public Log_event ...@@ -445,6 +445,13 @@ class Load_log_event: public Log_event
uint32 skip_lines; uint32 skip_lines;
sql_ex_info sql_ex; sql_ex_info sql_ex;
/* fname doesn't point to memory inside Log_event::temp_buf */
void set_fname_outside_temp_buf(const char *afname, uint alen)
{fname=afname;fname_len=alen;}
/* fname doesn't point to memory inside Log_event::temp_buf */
int check_fname_outside_temp_buf()
{return fname<temp_buf || fname>temp_buf+cached_event_len;}
#ifndef MYSQL_CLIENT #ifndef MYSQL_CLIENT
String field_lens_buf; String field_lens_buf;
String fields_buf; String fields_buf;
...@@ -687,6 +694,7 @@ class Create_file_log_event: public Load_log_event ...@@ -687,6 +694,7 @@ class Create_file_log_event: public Load_log_event
int exec_event(struct st_relay_log_info* rli); int exec_event(struct st_relay_log_info* rli);
#else #else
void print(FILE* file, bool short_form = 0, char* last_db = 0); void print(FILE* file, bool short_form = 0, char* last_db = 0);
void print(FILE* file, bool short_form, char* last_db, bool enable_local);
#endif #endif
Create_file_log_event(const char* buf, int event_len, bool old_format); Create_file_log_event(const char* buf, int event_len, bool old_format);
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment