mariadb/sql/log_event.cc

/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 2 of the License, or
   (at your option) any later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */


#ifndef MYSQL_CLIENT
#ifdef __GNUC__
#pragma implementation				// gcc: Class implementation
#endif
#include  "mysql_priv.h"
#include "slave.h"
#include <my_dir.h>
#endif /* MYSQL_CLIENT */

#ifdef MYSQL_CLIENT
static void pretty_print_str(FILE* file, char* str, int len)
{
  char* end = str + len;
  fputc('\'', file);
  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


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;
    log_seq = thd->log_seq;
    when = thd->start_time;
  }
  else
  {
    server_id = ::server_id;
    log_seq = 0;
    when = time(NULL);
  }
}

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 (!memcmp(file->name,"SQL_LOAD-",9))
      my_delete(file->name,MYF(MY_WME));
  }

  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_seq=0;
    flags=0;
  }
  else
  {
    log_seq = uint4korr(buf + LOG_SEQ_OFFSET);
    flags = uint2korr(buf + FLAGS_OFFSET);
  }
#ifndef MYSQL_CLIENT
  thd = 0;
#endif  
}


#ifndef MYSQL_CLIENT

int Log_event::exec_event(struct st_master_info* mi)
{
  if (mi)
  {
    thd->log_seq = 0;
    mi->inc_pos(get_event_len(), log_seq);
    flush_master_info(mi);
  }
  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)
  {
    const char* field = fields;
    tmp.append(" (");
    for (uint 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];
  String tmp(buf1, sizeof(buf1));
  tmp.length(0);
  char buf[22];
  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];
  String tmp(buf1, sizeof(buf1));
  tmp.length(0);
  char buf[22];
  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];
  String tmp(buf1, sizeof(buf1));
  tmp.length(0);
  char buf[22];
  tmp.append("host=");
  tmp.append(master_host);
  tmp.append(",port=");
  tmp.append(llstr(master_port,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("Log_seq", 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, log_seq);
  pack_info(packet);
  return my_net_write(&thd->net, (char*)packet->ptr(), packet->length());
}

#endif

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)
{
  char buf[LOG_EVENT_HEADER_LEN];
  char* pos = buf;
  int4store(pos, (ulong) when); // timestamp
  pos += 4;
  *pos++ = get_type_code(); // event type code
  int4store(pos, server_id);
  pos += 4;
  long tmp=get_data_size() + LOG_EVENT_HEADER_LEN;
  int4store(pos, tmp);
  pos += 4;
  int4store(pos, log_seq);
  pos += 4;
  int2store(pos, flags);
  pos += 2;
  return (my_b_write(file, (byte*) buf, (uint) (pos - buf)));
}

#ifndef MYSQL_CLIENT

int Log_event::read_log_event(IO_CACHE* file, String* packet,
			      pthread_mutex_t* log_lock)
{
  ulong data_len;
  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;
  }
  data_len = uint4korr(buf + EVENT_LEN_OFFSET);
  if (data_len < LOG_EVENT_HEADER_LEN || data_len > 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;
  }
  packet->append(buf, sizeof(buf));
  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;
    }
  }
  if (log_lock) pthread_mutex_unlock(log_lock);
  return 0;
}

#endif // MYSQL_CLIENT

#ifndef MYSQL_CLIENT
#define UNLOCK_MUTEX if (log_lock) pthread_mutex_unlock(log_lock);
#else
#define UNLOCK_MUTEX
#endif

#ifndef MYSQL_CLIENT
#define LOCK_MUTEX if (log_lock) pthread_mutex_lock(log_lock);
#else
#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,
				     bool old_format)
#else
Log_event* Log_event::read_log_event(IO_CACHE* file, bool old_format)
#endif  
{
  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 ))
  {
    UNLOCK_MUTEX;
    return 0;
  }

  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)
  {
    error = "Event too big";
    goto err;
  }

  if (data_len < header_size)
  {
    error = "Event too small";
    goto err;
  }

  // some events use the extra byte to null-terminate strings
  if (!(buf = my_malloc(data_len+1, MYF(MY_WME))))
  {
    error = "Out of memory";
    goto err;
  }
  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);
err:
  UNLOCK_MUTEX;
  if (error)
  {
    sql_print_error(error);
    my_free(buf, MYF(MY_ALLOW_ZERO_PTR));
  }
  return res;
}

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))
    return NULL; // general sanity check - will fail on a partial read

  Log_event* ev = NULL;

  switch(buf[EVENT_TYPE_OFFSET])
  {
  case QUERY_EVENT:
    ev  = new Query_log_event(buf, event_len, old_format);
    break;
  case LOAD_EVENT:
  case NEW_LOAD_EVENT:
    ev = new Load_log_event(buf, event_len, old_format);
    break;
  case ROTATE_EVENT:
    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);
    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;
  }
  if (!ev) return 0;
  if (!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)
{
  fputc('#', file);
  print_timestamp(file);
  fprintf(file, " server id  %d ", server_id); 
}

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
  localtime_r(ts,(res= &tm_tmp));
#else
  res=localtime(ts);
#endif

  fprintf(file,"%02d%02d%02d %2d:%02d:%02d",
	  res->tm_year % 100,
	  res->tm_mon+1,
	  res->tm_mday,
	  res->tm_hour,
	  res->tm_min,
	  res->tm_sec);
}


void Start_log_event::print(FILE* file, bool short_form, char* last_db)
{
  if (short_form)
    return;

  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);
}

void Stop_log_event::print(FILE* file, bool short_form, char* last_db)
{
  if (short_form)
    return;

  print_header(file);
  fprintf(file, "\tStop\n");
  fflush(file);
}

void Rotate_log_event::print(FILE* file, bool short_form, char* last_db)
{
  char buf[22];
  if (short_form)
    return;

  print_header(file);
  fprintf(file, "\tRotate to ");
  if (new_log_ident)
    my_fwrite(file, (byte*) new_log_ident, (uint)ident_len, 
	      MYF(MY_NABP | MY_WME));
  fprintf(file, "pos=%s\n", llstr(pos, buf));
  fflush(file);
}

#endif /* #ifdef MYSQL_CLIENT */

Start_log_event::Start_log_event(const char* buf,
				 bool old_format) :Log_event(buf, old_format)
{
  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[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_write(file, (byte*) buff, sizeof(buff)) ? -1 : 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
  int header_size = (old_format) ? OLD_HEADER_LEN : LOG_EVENT_HEADER_LEN;
  uint ident_offset;
  if (event_len < header_size)
    return;
  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)
{
  char buf[ROTATE_HEADER_LEN];
  int8store(buf, pos + R_POS_OFFSET);
  return my_b_write(file, (byte*)buf, ROTATE_HEADER_LEN) ||
    my_b_write(file, (byte*)new_log_ident, (uint) ident_len);
}

#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_AUTO_COMMIT | 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,
				 bool old_format):
  Log_event(buf, old_format),data_buf(0), query(NULL), db(NULL)
{
  ulong data_len;
  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 + 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 + Q_DATA_OFFSET, data_len);
  thread_id = uint4korr(buf + Q_THREAD_ID_OFFSET);
  db = data_buf;
  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
  if (!short_form)
  {
    print_header(file);
    fprintf(file, "\tQuery\tthread_id=%lu\texec_time=%lu\terror_code=%d\n",
	    (ulong) thread_id, (ulong) exec_time, error_code);
  }

  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 && db[0] && !same_db)
    fprintf(file, "use %s;\n", db);
  end=int10_to_str((long) when, strmov(buff,"SET TIMESTAMP="),10);
  *end++=';';
  *end++='\n';
  my_fwrite(file, (byte*) buff, (uint) (end-buff),MYF(MY_NABP | MY_WME));
  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;

  char buf[QUERY_HEADER_LEN]; 
  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_write(file, (byte*) buf, QUERY_HEADER_LEN) ||
	  my_b_write(file, (db) ? (byte*) db : (byte*)"", db_len + 1) ||
	  my_b_write(file, (byte*) query, q_len)) ? -1 : 0;
}

Intvar_log_event::Intvar_log_event(const char* buf, bool old_format):
  Log_event(buf, old_format)
{
  buf += (old_format) ? OLD_HEADER_LEN : LOG_EVENT_HEADER_LEN;
  type = buf[I_TYPE_OFFSET];
  val = uint8korr(buf+I_VAL_OFFSET);
}

const char* Intvar_log_event::get_var_type_name()
{
  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[I_TYPE_OFFSET] = type;
  int8store(buf + I_VAL_OFFSET, val);
  return my_b_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)
  {
    print_header(file);
    fprintf(file, "\tIntvar\n");
  }

  fprintf(file, "SET ");
  switch(type)
  {
  case LAST_INSERT_ID_EVENT:
    fprintf(file, "LAST_INSERT_ID = ");
    break;
  case INSERT_ID_EVENT:
    fprintf(file, "INSERT_ID = ");
    break;
  }
  fprintf(file, "%s;\n", llstr(val,llbuff));
  fflush(file);

}
#endif

int Load_log_event::write_data_header(IO_CACHE* file)
{
  char buf[LOAD_HEADER_LEN];
  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_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))
      return 1;
  }
  return (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));
}


static bool write_str(IO_CACHE *file, char *str, byte length)
{
  return (my_b_write(file, &length, 1) ||
	  my_b_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_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_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;
}

char* sql_ex_info::init(char* buf,char* buf_end,bool use_new_format)
{
  cached_new_format = use_new_format;
  if (use_new_format)
  {
    empty_flags=0;
    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++;
    *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;
}


#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), fields(0), field_lens(0),
  table_name(table_name_arg), db(db_arg), fname(ex->file_name),
  thread_id(thd->thread_id), num_fields(0), field_block_len(0)
{
  time_t end_time;
  time(&end_time);
  exec_time = (ulong) (end_time  - thd->start_time);
  db_len = (db) ? (uint32) strlen(db) : 0;
  table_name_len = (table_name) ? (uint32) strlen(table_name) : 0;
  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();
}

#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),fields(0),
  field_lens(0),  num_fields(0), field_block_len(0)
{
  if (!event_len) // derived class, will call copy_log_event() itself
    return;
  copy_log_event(buf, event_len, old_format);
}

int Load_log_event::copy_log_event(const char *buf, ulong event_len,
				   bool old_format)
{
  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 = 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 1;
  uint i;
  for (i = 0; i < num_fields; i++)
  {
    field_block_len += (uint)field_lens[i] + 1;
  }
  fields = (char*)field_lens + num_fields;
  table_name  = fields + field_block_len;
  db = table_name + table_name_len + 1;
  fname = db + db_len + 1;
  int type_code = get_type_code();
  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)
{
  if (!short_form)
  {
    print_header(file);
    fprintf(file, "\tQuery\tthread_id=%ld\texec_time=%ld\n",
	    thread_id, exec_time);
  }

  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 && db[0] && !same_db)
    fprintf(file, "use %s;\n", db);

  fprintf(file, "LOAD DATA INFILE '%-*s' ", fname_len, fname);

  if (sql_ex.opt_flags && REPLACE_FLAG )
    fprintf(file," REPLACE ");
  else if (sql_ex.opt_flags && IGNORE_FLAG )
    fprintf(file," IGNORE ");

  fprintf(file, "INTO TABLE %s ", table_name);
  if (sql_ex.field_term)
  {
    fprintf(file, " FIELDS TERMINATED BY ");
    pretty_print_str(file, sql_ex.field_term, sql_ex.field_term_len);
  }

  if (sql_ex.enclosed)
  {
    if (sql_ex.opt_flags && OPT_ENCLOSED_FLAG )
      fprintf(file," OPTIONALLY ");
    fprintf(file, " ENCLOSED BY ");
    pretty_print_str(file, sql_ex.enclosed, sql_ex.enclosed_len);
  }

  if (sql_ex.escaped)
  {
    fprintf(file, " ESCAPED BY ");
    pretty_print_str(file, sql_ex.escaped, sql_ex.escaped_len);
  }

  if (sql_ex.line_term)
  {
    fprintf(file," LINES TERMINATED BY ");
    pretty_print_str(file, sql_ex.line_term, sql_ex.line_term_len);
  }

  if (sql_ex.line_start)
  {
    fprintf(file," LINES STARTING BY ");
    pretty_print_str(file, sql_ex.line_start, sql_ex.line_start_len);
  }

  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++)
    {
      if (i)
	fputc(',', file);
      fprintf(file, field);

      field += field_lens[i]  + 1;
    }
    fputc(')', file);
  }

  fprintf(file, ";\n");
}

#endif /* #ifdef MYSQL_CLIENT */

#ifndef MYSQL_CLIENT

void Log_event::set_log_seq(THD* thd, MYSQL_LOG* log)
 {
   log_seq = (thd && thd->log_seq) ? thd->log_seq++ : log->log_seq++;
 }


void Load_log_event::set_fields(List<Item> &fields)
{
  uint i;
  const char* field = this->fields;
  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_master_info* mi):
  Log_event(thd_arg),mem_pool(0),master_host(0)
{
  if (!mi->inited)
    return;
  pthread_mutex_lock(&mi->lock);
  master_host_len = strlen(mi->host);
  master_log_len = strlen(mi->log_file_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, mi->log_file_name, master_log_len + 1);
    master_port = mi->port;
    master_pos = mi->pos;
  }
  else
    sql_print_error("Out of memory while recording slave event");
  pthread_mutex_unlock(&mi->lock);
}


#endif


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

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_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_write(file, (byte*) "", 1) ||
	  my_b_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_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):
  Load_log_event(buf,0,0),fake_base(0),block(0)
{
  int block_offset;
  if (copy_log_event(buf,len,0))
    return;
  file_id = uint4korr(buf + LOG_EVENT_HEADER_LEN +
		      + LOAD_HEADER_LEN + CF_FILE_ID_OFFSET);
  block_offset = LOG_EVENT_HEADER_LEN + Load_log_event::get_data_size() +
    CREATE_FILE_HEADER_LEN + 1; // 1 for \0 terminating fname  
  if (len < block_offset)
    return;
  block = (char*)buf + block_offset;
  block_len = len - block_offset;
}


#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];
  String tmp(buf1, sizeof(buf1));
  tmp.length(0);
  char buf[22];
  tmp.append("db=");
  tmp.append(db, db_len);
  tmp.append(";table=");
  tmp.append(table_name, table_name_len);
  tmp.append(";file_id=");
  tmp.append(llstr(file_id,buf));
  tmp.append(";block_len=");
  tmp.append(llstr(block_len,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_write(file, buf, APPEND_BLOCK_HEADER_LEN) ||
	  my_b_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];
  String tmp(buf1, sizeof(buf1));
  tmp.length(0);
  char buf[22];
  tmp.append(";file_id=");
  tmp.append(llstr(file_id,buf));
  tmp.append(";block_len=");
  tmp.append(llstr(block_len,buf));
  net_store_data(packet, (char*)tmp.ptr(), tmp.length());
}
#endif  

#ifndef MYSQL_CLIENT  
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_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=%d\n",
	  file_id);
}
#endif  
#ifndef MYSQL_CLIENT
void Delete_file_log_event::pack_info(String* packet)
{
  char buf1[256];
  String tmp(buf1, sizeof(buf1));
  tmp.length(0);
  char buf[22];
  tmp.append(";file_id=");
  tmp.append(llstr(file_id,buf));
  net_store_data(packet, (char*)tmp.ptr(), tmp.length());
}
#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_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 buf1[256];
  String tmp(buf1, sizeof(buf1));
  tmp.length(0);
  char buf[22];
  tmp.append(";file_id=");
  tmp.append(llstr(file_id,buf));
  net_store_data(packet, (char*)tmp.ptr(), tmp.length());
}
#endif

#ifndef MYSQL_CLIENT
int Query_log_event::exec_event(struct st_master_info* mi)
{
  int expected_error,actual_error = 0;
  init_sql_alloc(&thd->mem_root, 8192,0);
  thd->db = rewrite_db((char*)db);
  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 (!check_expected_error(thd, (expected_error = error_code)))
    {
      mysql_parse(thd, thd->query, q_len);
      if (expected_error !=
	  (actual_error = thd->net.last_errno) && 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)
      {
	thd->query_error = 0;
	*last_slave_error = 0;
	last_slave_errno = 0;
      }
    }
    else
    {
      // master could be inconsistent, abort and tell DBA to check/fix it
      thd->db = thd->query = 0;
      thd->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->convert_set = 0;
  close_thread_tables(thd);

  if (thd->query_error || thd->fatal_error)
  {
    slave_print_error(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(mi); 
}

int Load_log_event::exec_event(NET* net, struct st_master_info* mi)
{
  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))
    {
      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(mi->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
    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(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(mi); 
}

int Start_log_event::exec_event(struct st_master_info* mi)
{
  if (!mi->old_format)
  {
    close_temporary_tables(thd);
    cleanup_load_tmpdir();
  }
  return Log_event::exec_event(mi);
}

int Stop_log_event::exec_event(struct st_master_info* mi)
{
  if (mi->pos > 4) // stop event should be ignored after rotate event
  {
    close_temporary_tables(thd);
    cleanup_load_tmpdir();
    mi->inc_pos(get_event_len(), log_seq);
    flush_master_info(mi);
  }
  thd->log_seq = 0;
  return 0;
}

int Rotate_log_event::exec_event(struct st_master_info* mi)
{
  bool rotate_binlog = 0, write_slave_event = 0;
  char* log_name = mi->log_file_name;
  pthread_mutex_lock(&mi->lock);

  // 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);
    memcpy(log_name, new_log_ident,ident_len );
    log_name[ident_len] = 0;
  }
  mi->pos = pos; 
  mi->last_log_seq = log_seq;
#ifndef DBUG_OFF
  if (abort_slave_event_count)
    ++events_till_abort;
#endif
  if (rotate_binlog)
  {
    mysql_bin_log.new_file();
    mi->last_log_seq = 0;
  }
  pthread_cond_broadcast(&mi->cond);
  pthread_mutex_unlock(&mi->lock);
  flush_master_info(mi);

  if (write_slave_event)
  {
    Slave_log_event s(thd, mi);
    if (s.master_host)
    {
      s.set_log_seq(0, &mysql_bin_log);
      s.server_id = ::server_id;
      mysql_bin_log.write(&s);
    }
  }
  thd->log_seq = 0;
  return 0;
}

int Intvar_log_event::exec_event(struct st_master_info* mi)
{
  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;
  }
  mi->inc_pending(get_event_len());
  return 0;
}

int Slave_log_event::exec_event(struct st_master_info* mi)
{
  if (mysql_bin_log.is_open())
    mysql_bin_log.write(this);
  return Log_event::exec_event(mi);
}

int Create_file_log_event::exec_event(struct st_master_info* mi)
{
  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(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(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(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(my_errno, "Write to '%s' failed", fname_buf);
    goto err;
  }
  if (mysql_bin_log.is_open())
    mysql_bin_log.write(this);
  error=0;
err:
  if (error)
    end_io_cache(&file);
  if (fd >= 0)
    my_close(fd, MYF(0));
  return error ? 1 : Log_event::exec_event(mi);
}

int Delete_file_log_event::exec_event(struct st_master_info* mi)
{
  char fname[FN_REFLEN+10];
  char* p;
  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(mi);
}

int Append_block_log_event::exec_event(struct st_master_info* mi)
{
  char fname[FN_REFLEN+10];
  char* p;
  int fd = -1;
  int error = 1;
  p = slave_load_file_stem(fname, file_id, server_id);
  memcpy(p, ".data", 6);
  if ((fd = my_open(fname, O_WRONLY|O_APPEND|O_BINARY, MYF(MY_WME))) < 0)
  {
    slave_print_error(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(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(mi);
}

int Execute_load_log_event::exec_event(struct st_master_info* mi)
{
  char fname[FN_REFLEN+10];
  char* p;
  int fd = -1;
  int error = 1;
  ulong save_options;
  IO_CACHE file;
  Load_log_event* lev = 0;
  p = slave_load_file_stem(fname, file_id, server_id);
  memcpy(p, ".info", 6);
  bzero((char*)&file, sizeof(file));
  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(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(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(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;
  end_io_cache(&file);
  if (fd >= 0)
    my_close(fd, MYF(0));
  return error ? error : Log_event::exec_event(mi);
}


#endif