Re: bzr commit into mysql-5.1-bugteam branch (alfranio.correia:2802) Bug#39393

[Andrei Elkin]

Alfranio, hello.

I thought about clever Zhen Xing's comments to agree that the current patch
changes are way too intrusive, not quite expectedly to me.

Indeed, we could be content with mimicing of how the statement level 
ignores errors (i might have offered that in the first place, but i
slipped, sorry).

So

Query_log_event::do_apply_event() shows what to do.
Once Rows_log_event::do_exec_row() returns with the HANDLER
error thd->is_error() must be TRUE and
thd->main_da.sql_errno() gets us the USER error.
The rest, including clearing the error, you will see in
Query_log_event::do_apply_event().

So I suggest to go this way.

It's a good refactoring you've made for idempotent error, to admit.

cheers,

Andrei




> He Zhenxing wrote:
>> Hi Alfranio,
>>
>> Thank you for the work!
>>
>> I think it is not necessary to change the handler structure, I think you
>> do not need to construct the error message from storage engines and
>> report the error as a warning, because we do not do that for SBR
>> eighter.
>>   
> Hi Jasonh,
>
> I can remove the warning messages if you want but I don't see any
> problem in having it.
> However, I think we need to change the handler's interface because the
> mapping between
> the engine errors and the server error is there. And If the RBR fails we
> want to skip
> based on the server error and thus we need to call a method in the
> handler that does the
> translation but does not print out any message.
>
> So I would keep the changes but just eliminate the warning message.
> Do you agree?
>
> If you don't agree we will need at least to change the the signature of
> the ::print_error
> and other methods in the cluster and partition to avoid printing the
> message and
> at the same return the error code.
>
> Andrei, what do you think?
> Jasonh, Andrei, Do you have comments on the core of the bug and the test
> case?
>
> Cheers.
>
>
>> Alfranio Correia wrote: 
>>   
>>> #At
> file:///home/acorreia/workspace.sun/repository.mysql/bzrwork/bug-39393/mysql-5.1-bugteam/
> based on revid:sergey.glukhov@stripped
>>>
>>>  2802 Alfranio Correia	2009-03-20
>>>       BUG#39393 slave-skip-errors does not work when using ROW based
> replication
>>>                   
>>>       RBR was not considering the option --slave-skip-errors.
>>>                   
>>>       To fix the problem, we are reporting the ignored ERROR(s) as warnings
> thus avoiding 
>>>       stopping the SQL Thread. Besides, it fixes the output of "SHOW
> VARIABLES LIKE 
>>>       'slave_skip_errors'" which was showing nothing when the value "all" was
> assigned 
>>>       to --slave-skip-errors.
>>>       
>>>       @include/my_sys.h
>>>       @mysys/my_error.c
>>>         added functions to create messages in a buffer without printing them
> out.
>>>       @sql/ha_ndbcluster.h
>>>       @sql/ha_ndbcluster.cc
>>>         refactored its interface so specific errors in the cluster are
> easily
>>>         reported through the handler.
>>>       @sql/ha_partition.cc
>>>       @sql/ha_partition.h
>>>       @sql/partition_info.cc
>>>       @sql/partition_info.h
>>>         refactored its interface so specific errors in the cluster engine 
>>>         are easily reported through the handler.
>>>       @sql/handler.h
>>>       @sql/handler.cc
>>>         refactored its interface so errors can be easily checked and
> reported
>>>         through its interfaces.
>>>       @sql/log_event.cc
>>>         skipped rbr errors when the option skip-slave-errors is set.
>>>       @sql/slave.cc
>>>         fixed the output of for SHOW VARIABLES LIKE 'slave_skip_errors'".
>>> added:
>>>   mysql-test/suite/rpl/r/rpl_skiperrors_rbr.result
>>>   mysql-test/suite/rpl/t/rpl_skiperrors_rbr-slave.opt
>>>   mysql-test/suite/rpl/t/rpl_skiperrors_rbr.test
>>> modified:
>>>   include/my_sys.h
>>>   mysys/my_error.c
>>>   sql/ha_ndbcluster.cc
>>>   sql/ha_ndbcluster.h
>>>   sql/ha_partition.cc
>>>   sql/ha_partition.h
>>>   sql/handler.cc
>>>   sql/handler.h
>>>   sql/log_event.cc
>>>   sql/partition_info.cc
>>>   sql/partition_info.h
>>>   sql/slave.cc
>>>
>>> === modified file 'include/my_sys.h'
>>> --- a/include/my_sys.h	2009-02-05 06:16:00 +0000
>>> +++ b/include/my_sys.h	2009-03-20 12:40:23 +0000
>>> @@ -645,6 +645,9 @@ extern int my_sync(File fd, myf my_flags
>>>  extern int my_sync_dir(const char *dir_name, myf my_flags);
>>>  extern int my_sync_dir_by_file(const char *file_name, myf my_flags);
>>>  extern int my_error _VARARGS((int nr,myf MyFlags, ...));
>>> +extern void my_buffer_error _VARARGS((int nr, char* ebuff, size_t size,
> ...));
>>> +extern void my_buffer_format _VARARGS((char* ebuff, size_t size,
>>> +                                       const char* format, ...));
>>>  extern int my_printf_error _VARARGS((uint my_err, const char *format,
>>>  				     myf MyFlags, ...))
>>>  				    ATTRIBUTE_FORMAT(printf, 2, 4);
>>>
>>> === added file 'mysql-test/suite/rpl/r/rpl_skiperrors_rbr.result'
>>> --- a/mysql-test/suite/rpl/r/rpl_skiperrors_rbr.result	1970-01-01 00:00:00
> +0000
>>> +++ b/mysql-test/suite/rpl/r/rpl_skiperrors_rbr.result	2009-03-20 12:40:23
> +0000
>>> @@ -0,0 +1,94 @@
>>> +stop slave;
>>> +drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
>>> +reset master;
>>> +reset slave;
>>> +drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
>>> +start slave;
>>> +CREATE TABLE t1(id INT NOT NULL PRIMARY KEY, data INT) Engine=InnoDB;
>>> +INSERT INTO t1 VALUES(1, 1);
>>> +SET SQL_LOG_BIN=0;
>>> +DELETE FROM t1 WHERE id = 1;
>>> +SET SQL_LOG_BIN=1;
>>> +INSERT INTO t1 VALUES(1, 2);
>>> +
>>> +SELECT *, "SLAVE DATA" FROM t1 ORDER BY id;
>>> +id	data	SLAVE DATA
>>> +1	1	SLAVE DATA
>>> +SELECT *, "MASTER DATA" FROM t1 ORDER BY id;
>>> +id	data	MASTER DATA
>>> +1	2	MASTER DATA
>>> +DELETE FROM t1;
>>> +start transaction;
>>> +INSERT INTO t1 VALUES(1, 1);
>>> +INSERT INTO t1 VALUES(10, 1);
>>> +SET SQL_LOG_BIN=0;
>>> +DELETE FROM t1 WHERE id = 10;
>>> +SET SQL_LOG_BIN=1;
>>> +INSERT INTO t1 VALUES(10, 2);
>>> +INSERT INTO t1 VALUES(3, 2);
>>> +commit;
>>> +
>>> +start transaction;
>>> +INSERT INTO t1 VALUES(4, 3);
>>> +INSERT INTO t1 VALUES(5, 3);
>>> +SET SQL_LOG_BIN=0;
>>> +DELETE FROM t1 WHERE id = 5;
>>> +SET SQL_LOG_BIN=1;
>>> +INSERT INTO t1 VALUES(5, 4);
>>> +INSERT INTO t1 VALUES(6, 4);
>>> +rollback;
>>> +
>>> +SELECT *, "SLAVE DATA" FROM t1 ORDER BY id;
>>> +id	data	SLAVE DATA
>>> +1	1	SLAVE DATA
>>> +3	2	SLAVE DATA
>>> +10	1	SLAVE DATA
>>> +SELECT *, "MASTER DATA" FROM t1 ORDER BY id;
>>> +id	data	MASTER DATA
>>> +1	1	MASTER DATA
>>> +3	2	MASTER DATA
>>> +10	2	MASTER DATA
>>> +DELETE FROM t1;
>>> +INSERT INTO t1 VALUES(1, 1);
>>> +INSERT INTO t1 VALUES(2, 1);
>>> +INSERT INTO t1 VALUES(3, 1);
>>> +INSERT INTO t1 VALUES(4, 1);
>>> +SET SQL_LOG_BIN=0;
>>> +DELETE FROM t1 WHERE id = 4;
>>> +SET SQL_LOG_BIN=1;
>>> +UPDATE t1 SET id= id + 3, data = 2;
>>> +
>>> +SELECT *, "SLAVE DATA" FROM t1 ORDER BY id;
>>> +id	data	SLAVE DATA
>>> +1	1	SLAVE DATA
>>> +4	1	SLAVE DATA
>>> +5	2	SLAVE DATA
>>> +6	2	SLAVE DATA
>>> +SELECT *, "MASTER DATA" FROM t1 ORDER BY id;
>>> +id	data	MASTER DATA
>>> +4	2	MASTER DATA
>>> +5	2	MASTER DATA
>>> +6	2	MASTER DATA
>>> +CREATE TABLE t2(id INT NOT NULL PRIMARY KEY, data INT)  Engine=MyIsam;
>>> +INSERT INTO t2 VALUES(1, 1);
>>> +INSERT INTO t2 VALUES(2, 1);
>>> +INSERT INTO t2 VALUES(3, 1);
>>> +INSERT INTO t2 VALUES(4, 1);
>>> +SET SQL_LOG_BIN=0;
>>> +DELETE FROM t2 WHERE id = 4;
>>> +SET SQL_LOG_BIN=1;
>>> +UPDATE t2 SET id= id + 3, data = 2;
>>> +
>>> +SELECT *, "SLAVE DATA" FROM t2 ORDER BY id;
>>> +id	data	SLAVE DATA
>>> +1	1	SLAVE DATA
>>> +4	1	SLAVE DATA
>>> +5	2	SLAVE DATA
>>> +6	2	SLAVE DATA
>>> +SELECT *, "MASTER DATA" FROM t2 ORDER BY id;
>>> +id	data	MASTER DATA
>>> +4	2	MASTER DATA
>>> +5	2	MASTER DATA
>>> +6	2	MASTER DATA
>>> +DROP TABLE t1;
>>> +DROP TABLE t2;
>>>
>>> === added file 'mysql-test/suite/rpl/t/rpl_skiperrors_rbr-slave.opt'
>>> --- a/mysql-test/suite/rpl/t/rpl_skiperrors_rbr-slave.opt	1970-01-01 00:00:00
> +0000
>>> +++ b/mysql-test/suite/rpl/t/rpl_skiperrors_rbr-slave.opt	2009-03-20 12:40:23
> +0000
>>> @@ -0,0 +1 @@
>>> +--slave_skip_errors=all
>>>
>>> === added file 'mysql-test/suite/rpl/t/rpl_skiperrors_rbr.test'
>>> --- a/mysql-test/suite/rpl/t/rpl_skiperrors_rbr.test	1970-01-01 00:00:00
> +0000
>>> +++ b/mysql-test/suite/rpl/t/rpl_skiperrors_rbr.test	2009-03-20 12:40:23
> +0000
>>> @@ -0,0 +1,163 @@
>>> +#################################################################
>>> +#   This test cases checks if slave-skip-errors works when 
>>> +#   using ROW based by generating forcing duplicate keys
>>> +#   in the SLAVE. 
>>> +#
>>> +#   The following scenarios are checked:
>>> +#
>>> +#   1 - InnoDB without transactions
>>> +#   2 - InnoDB with transactions, both commit and rollback.
>>> +#   3 - InnoDB with an UPDATE on a SET of rows.
>>> +#   4 - MyIsam with an UPDATE on a SET of rows.
>>> +#
>>> +#################################################################
>>> +
>>> +--source include/have_binlog_format_row.inc
>>> +--source include/have_innodb.inc
>>> +--source include/master-slave.inc
>>> +
>>> +#################################################################
>>> +#                  1 - (InnoDB) Duplicate key
>>> +#################################################################
>>> +connection master;
>>> +
>>> +CREATE TABLE t1(id INT NOT NULL PRIMARY KEY, data INT) Engine=InnoDB;
>>> +
>>> +INSERT INTO t1 VALUES(1, 1);
>>> +
>>> +SET SQL_LOG_BIN=0;
>>> +DELETE FROM t1 WHERE id = 1;
>>> +SET SQL_LOG_BIN=1;
>>> +INSERT INTO t1 VALUES(1, 2);
>>> +
>>> +sync_slave_with_master;
>>> +
>>> +let $error= query_get_value("SHOW SLAVE STATUS", Last_SQL_Error, 1);
>>> +echo $error;
>>> +
>>> +connection slave;
>>> +
>>> +SELECT *, "SLAVE DATA" FROM t1 ORDER BY id;
>>> +
>>> +connection master;
>>> +
>>> +SELECT *, "MASTER DATA" FROM t1 ORDER BY id;
>>> +
>>> +#################################################################
>>> +#            2 - (InnoDB with Transaction) Duplicate key
>>> +#################################################################
>>> +connection master;
>>> +
>>> +DELETE FROM t1;
>>> +
>>> +start transaction;
>>> +INSERT INTO t1 VALUES(1, 1);
>>> +INSERT INTO t1 VALUES(10, 1);
>>> +SET SQL_LOG_BIN=0;
>>> +DELETE FROM t1 WHERE id = 10;
>>> +SET SQL_LOG_BIN=1;
>>> +INSERT INTO t1 VALUES(10, 2);
>>> +INSERT INTO t1 VALUES(3, 2);
>>> +commit;
>>> +
>>> +sync_slave_with_master;
>>> +
>>> +let $error= query_get_value("SHOW SLAVE STATUS", Last_SQL_Error, 1);
>>> +echo $error;
>>> +
>>> +connection master;
>>> +
>>> +start transaction;
>>> +INSERT INTO t1 VALUES(4, 3);
>>> +INSERT INTO t1 VALUES(5, 3);
>>> +SET SQL_LOG_BIN=0;
>>> +DELETE FROM t1 WHERE id = 5;
>>> +SET SQL_LOG_BIN=1;
>>> +INSERT INTO t1 VALUES(5, 4);
>>> +INSERT INTO t1 VALUES(6, 4);
>>> +rollback;
>>> +
>>> +sync_slave_with_master;
>>> +
>>> +let $error= query_get_value("SHOW SLAVE STATUS", Last_SQL_Error, 1);
>>> +echo $error;
>>> +
>>> +connection master;
>>> +
>>> +connection slave;
>>> +
>>> +SELECT *, "SLAVE DATA" FROM t1 ORDER BY id;
>>> +
>>> +connection master;
>>> +
>>> +SELECT *, "MASTER DATA" FROM t1 ORDER BY id;
>>> +
>>> +#################################################################
>>> +#              3 - (InnoDB with sets) Duplicate key
>>> +#################################################################
>>> +connection master;
>>> +
>>> +DELETE FROM t1;
>>> +
>>> +INSERT INTO t1 VALUES(1, 1);
>>> +INSERT INTO t1 VALUES(2, 1);
>>> +INSERT INTO t1 VALUES(3, 1);
>>> +INSERT INTO t1 VALUES(4, 1);
>>> +
>>> +SET SQL_LOG_BIN=0;
>>> +DELETE FROM t1 WHERE id = 4;
>>> +SET SQL_LOG_BIN=1;
>>> +UPDATE t1 SET id= id + 3, data = 2;
>>> +
>>> +sync_slave_with_master;
>>> +
>>> +let $error= query_get_value("SHOW SLAVE STATUS", Last_SQL_Error, 1);
>>> +echo $error;
>>> +
>>> +connection slave;
>>> +
>>> +SELECT *, "SLAVE DATA" FROM t1 ORDER BY id;
>>> +
>>> +connection master;
>>> +
>>> +SELECT *, "MASTER DATA" FROM t1 ORDER BY id;
>>> +
>>> +#################################################################
>>> +#                 4 - (MyIsam with sets) Duplicate key
>>> +#################################################################
>>> +connection master;
>>> +
>>> +CREATE TABLE t2(id INT NOT NULL PRIMARY KEY, data INT)  Engine=MyIsam;
>>> +
>>> +INSERT INTO t2 VALUES(1, 1);
>>> +INSERT INTO t2 VALUES(2, 1);
>>> +INSERT INTO t2 VALUES(3, 1);
>>> +INSERT INTO t2 VALUES(4, 1);
>>> +
>>> +SET SQL_LOG_BIN=0;
>>> +DELETE FROM t2 WHERE id = 4;
>>> +SET SQL_LOG_BIN=1;
>>> +UPDATE t2 SET id= id + 3, data = 2;
>>> +
>>> +sync_slave_with_master;
>>> +
>>> +let $error= query_get_value("SHOW SLAVE STATUS", Last_SQL_Error, 1);
>>> +echo $error;
>>> +
>>> +connection slave;
>>> +
>>> +SELECT *, "SLAVE DATA" FROM t2 ORDER BY id;
>>> +
>>> +connection master;
>>> +
>>> +SELECT *, "MASTER DATA" FROM t2 ORDER BY id;
>>> +
>>> +#################################################################
>>> +#                           Clean up
>>> +#################################################################
>>> +connection master;
>>> +
>>> +DROP TABLE t1;
>>> +DROP TABLE t2;
>>> +
>>> +sync_slave_with_master;
>>>
>>> === modified file 'mysys/my_error.c'
>>> --- a/mysys/my_error.c	2009-02-05 06:16:00 +0000
>>> +++ b/mysys/my_error.c	2009-03-20 12:40:23 +0000
>>> @@ -101,6 +101,49 @@ int my_error(int nr, myf MyFlags, ...)
>>>  
>>>
>>>  /*
>>> +  Populates a buffer with the error message associated with the error (nr)
>>> +  number specified as parameter.
>>> + */
>>> +void my_buffer_error(int nr, char* ebuff, size_t size, ...)
>>> +{
>>> +  const char *format;
>>> +  struct my_err_head *meh_p;
>>> +  va_list args;
>>> +  DBUG_ENTER("my_bufer_error");
>>> +
>>> +  /* Search for the error messages array, which could contain the message.
> */
>>> +  for (meh_p= my_errmsgs_list; meh_p; meh_p= meh_p->meh_next)
>>> +    if (nr <= meh_p->meh_last)
>>> +      break;
>>> +
>>> +  /* get the error message string. Default, if NULL or empty string ("").
> */
>>> +  if (! (format= (meh_p && (nr >= meh_p->meh_first)) ?
>>> +         meh_p->meh_errmsgs[nr - meh_p->meh_first] : NULL) || !
> *format)
>>> +    (void) my_snprintf (ebuff, size, "Unknown error %d", nr);
>>> +  else
>>> +  {
>>> +    va_start(args, size);
>>> +    (void) my_vsnprintf (ebuff, size, format, args);
>>> +    va_end(args);
>>> +  }
>>> +  DBUG_VOID_RETURN;
>>> +}
>>> +
>>> +/*
>>> +  Populates a buffer with an error message specified as parameter (format).
>>> + */
>>> +void my_buffer_format(char* ebuff, size_t size, const char* format, ...)
>>> +{
>>> +  va_list args;
>>> +  DBUG_ENTER("my_buffer_format");
>>> +
>>> +  va_start(args,format);
>>> +  (void) my_vsnprintf (ebuff, size, format, args);
>>> +  va_end(args);
>>> +  DBUG_VOID_RETURN;
>>> +}
>>> +
>>> +/*
>>>    Error as printf
>>>  
>>>    SYNOPSIS
>>>
>>> === modified file 'sql/ha_ndbcluster.cc'
>>> --- a/sql/ha_ndbcluster.cc	2008-10-23 19:27:09 +0000
>>> +++ b/sql/ha_ndbcluster.cc	2009-03-20 12:40:23 +0000
>>> @@ -7545,19 +7545,17 @@ static int ndbcluster_end(handlerton *ht
>>>    DBUG_RETURN(0);
>>>  }
>>>  
>>> -void ha_ndbcluster::print_error(int error, myf errflag)
>>> +int ha_ndbcluster::process_error(int error, char* ebuff, int size, myf*
> srvflag)
>>>  {
>>> -  DBUG_ENTER("ha_ndbcluster::print_error");
>>> -  DBUG_PRINT("enter", ("error: %d", error));
>>> -
>>> +  DBUG_ENTER("ha_ndbcluster::process_error");
>>> +  int ret= error;
>>>    if (error == HA_ERR_NO_PARTITION_FOUND)
>>> -    m_part_info->print_no_partition_found(table);
>>> +    ret= m_part_info->process_no_partition_found(table, ebuff, size,
> srvflag);
>>>    else
>>> -    handler::print_error(error, errflag);
>>> -  DBUG_VOID_RETURN;
>>> +    ret= handler::process_error(error, ebuff, size, srvflag);
>>> +  DBUG_RETURN(ret);
>>>  }
>>>  
>>> -
>>>  /**
>>>    Static error print function called from static handler method
>>>    ndbcluster_commit and ndbcluster_rollback.
>>>
>>> === modified file 'sql/ha_ndbcluster.h'
>>> --- a/sql/ha_ndbcluster.h	2008-02-04 14:40:04 +0000
>>> +++ b/sql/ha_ndbcluster.h	2009-03-20 12:40:23 +0000
>>> @@ -279,7 +279,7 @@ class ha_ndbcluster: public handler
>>>    int external_lock(THD *thd, int lock_type);
>>>    void unlock_row();
>>>    int start_stmt(THD *thd, thr_lock_type lock_type);
>>> -  void print_error(int error, myf errflag);
>>> +  int process_error(int error, char* ebuff, int size, myf* errflag);
>>>    const char * table_type() const;
>>>    const char ** bas_ext() const;
>>>    ulonglong table_flags(void) const;
>>>
>>> === modified file 'sql/ha_partition.cc'
>>> --- a/sql/ha_partition.cc	2009-01-07 22:30:10 +0000
>>> +++ b/sql/ha_partition.cc	2009-03-20 12:40:23 +0000
>>> @@ -5831,22 +5831,21 @@ enum row_type ha_partition::get_row_type
>>>    return type;
>>>  }
>>>  
>>> -
>>> -void ha_partition::print_error(int error, myf errflag)
>>> +int ha_partition::process_error(int error, char* ebuff, int size, myf*
> srvflag)
>>>  {
>>> -  DBUG_ENTER("ha_partition::print_error");
>>> +  DBUG_ENTER("ha_partition::process_error");
>>> +  int ret= error;
>>>  
>>>    /* Should probably look for my own errors first */
>>>    DBUG_PRINT("enter", ("error: %d", error));
>>>  
>>> -  if (error == HA_ERR_NO_PARTITION_FOUND)
>>> -    m_part_info->print_no_partition_found(table);
>>> +  if (error == HA_ERR_NO_PARTITION_FOUND) 
>>> +    ret= m_part_info->process_no_partition_found(table, ebuff, size,
> srvflag);
>>>    else
>>> -    m_file[m_last_part]->print_error(error, errflag);
>>> -  DBUG_VOID_RETURN;
>>> +    ret= m_file[m_last_part]->process_error(error, ebuff, size,
> srvflag);
>>> +  DBUG_RETURN(ret);
>>>  }
>>>  
>>> -
>>>  bool ha_partition::get_error_message(int error, String *buf)
>>>  {
>>>    DBUG_ENTER("ha_partition::get_error_message");
>>>
>>> === modified file 'sql/ha_partition.h'
>>> --- a/sql/ha_partition.h	2009-01-05 16:10:20 +0000
>>> +++ b/sql/ha_partition.h	2009-03-20 12:40:23 +0000
>>> @@ -600,7 +600,8 @@ public:
>>>    /*
>>>       Handler specific error messages
>>>    */
>>> -  virtual void print_error(int error, myf errflag);
>>> +  int process_error(int error, char* ebuff, int size, myf* srvflag);
>>> +
>>>    virtual bool get_error_message(int error, String * buf);
>>>    /*
>>>    
> -------------------------------------------------------------------------
>>>
>>> === modified file 'sql/handler.cc'
>>> --- a/sql/handler.cc	2009-02-10 08:37:27 +0000
>>> +++ b/sql/handler.cc	2009-03-20 12:40:23 +0000
>>> @@ -2526,9 +2526,22 @@ void handler::ha_release_auto_increment(
>>>    }
>>>  }
>>>  
>>> -
>>>  void handler::print_keydup_error(uint key_nr, const char *msg)
>>>  {
>>> +  char ebuff[MYSQL_ERRMSG_SIZE];
>>> +  DBUG_ENTER("handler::print_keydup_error");
>>> +
>>> +  process_keydup_error(key_nr, msg, ebuff, MYSQL_ERRMSG_SIZE);
>>> +  my_message(ER_DUP_ENTRY, ebuff, MYF(0));
>>> +
>>> +  DBUG_VOID_RETURN;
>>> +}
>>> +
>>> +void handler::process_keydup_error(uint key_nr, const char *msg,
>>> +                                       char* ebuff, int size)
>>> +{
>>> +  DBUG_ENTER("handler::process_keydup_error");
>>> +
>>>    /* Write the duplicated key in the error message */
>>>    char key[MAX_KEY_LENGTH];
>>>    String str(key,sizeof(key),system_charset_info);
>>> @@ -2537,7 +2550,7 @@ void handler::print_keydup_error(uint ke
>>>    {
>>>      /* Key is unknown */
>>>      str.copy("", 0, system_charset_info);
>>> -    my_printf_error(ER_DUP_ENTRY, msg, MYF(0), str.c_ptr(), "*UNKNOWN*");
>>> +    my_buffer_format(ebuff, size, msg, str.c_ptr(), "*UNKNOWN*");
>>>    }
>>>    else
>>>    {
>>> @@ -2549,14 +2562,26 @@ void handler::print_keydup_error(uint ke
>>>        str.length(max_length-4);
>>>        str.append(STRING_WITH_LEN("..."));
>>>      }
>>> -    my_printf_error(ER_DUP_ENTRY, msg,
>>> -		    MYF(0), str.c_ptr_safe(), table->key_info[key_nr].name);
>>> +    my_buffer_format(ebuff, size, msg, str.c_ptr_safe(), 
>>> +                     table->key_info[key_nr].name);
>>>    }
>>> +  DBUG_VOID_RETURN;
>>>  }
>>>  
>>> +void handler::print_error(int error, myf errflag)
>>> +{
>>> +  DBUG_ENTER("handler::print_error");
>>> +
>>> +  char ebuff[MYSQL_ERRMSG_SIZE];
>>> +  myf srvflag= errflag;
>>> +  int srverror= process_error(error, ebuff, MYSQL_ERRMSG_SIZE,
> &srvflag);
>>> +  my_message(srverror, ebuff, srvflag);
>>> +
>>> +  DBUG_VOID_RETURN;
>>> +}
>>>  
>>>  /**
>>> -  Print error that we got from handler function.
>>> +  Process error that we got from handler function.
>>>  
>>>    @note
>>>      In case of delete table it's only safe to use the following parts of
>>> @@ -2564,39 +2589,42 @@ void handler::print_keydup_error(uint ke
>>>      - table->s->path
>>>      - table->alias
>>>  */
>>> -void handler::print_error(int error, myf errflag)
>>> +int handler::process_error(int error, char* ebuff, int size, myf* srvflag)
>>>  {
>>> -  DBUG_ENTER("handler::print_error");
>>> +  DBUG_ENTER("handler::process_error");
>>>    DBUG_PRINT("enter",("error: %d",error));
>>> +  DBUG_ASSERT(ebuff != 0 && size >= 0 && srvflag != 0);
>>>  
>>> -  int textno=ER_GET_ERRNO;
>>> +  int textno= ER_GET_ERRNO;
>>>    switch (error) {
>>>    case EACCES:
>>> -    textno=ER_OPEN_AS_READONLY;
>>> +    textno= ER_OPEN_AS_READONLY;
>>>      break;
>>>    case EAGAIN:
>>> -    textno=ER_FILE_USED;
>>> +    textno= ER_FILE_USED;
>>>      break;
>>>    case ENOENT:
>>> -    textno=ER_FILE_NOT_FOUND;
>>> +    textno= ER_FILE_NOT_FOUND;
>>>      break;
>>>    case HA_ERR_KEY_NOT_FOUND:
>>>    case HA_ERR_NO_ACTIVE_RECORD:
>>>    case HA_ERR_END_OF_FILE:
>>> -    textno=ER_KEY_NOT_FOUND;
>>> +    textno= ER_KEY_NOT_FOUND;
>>>      break;
>>>    case HA_ERR_WRONG_MRG_TABLE_DEF:
>>> -    textno=ER_WRONG_MRG_TABLE;
>>> +    textno= ER_WRONG_MRG_TABLE;
>>>      break;
>>>    case HA_ERR_FOUND_DUPP_KEY:
>>>    {
>>>      uint key_nr=get_dup_key(error);
>>>      if ((int) key_nr >= 0)
>>>      {
>>> -      print_keydup_error(key_nr, ER(ER_DUP_ENTRY_WITH_KEY_NAME));
>>> -      DBUG_VOID_RETURN;
>>> +      textno= ER_DUP_ENTRY;
>>> +      process_keydup_error(key_nr, ER(ER_DUP_ENTRY_WITH_KEY_NAME), ebuff,
> size);
>>> +      (*srvflag)= MYF(0);
>>> +      DBUG_RETURN(textno);
>>>      }
>>> -    textno=ER_DUP_KEY;
>>> +    textno= ER_DUP_KEY;
>>>      break;
>>>    }
>>>    case HA_ERR_FOREIGN_DUPLICATE_KEY:
>>> @@ -2604,6 +2632,7 @@ void handler::print_error(int error, myf
>>>      uint key_nr= get_dup_key(error);
>>>      if ((int) key_nr >= 0)
>>>      {
>>> +      textno= ER_FOREIGN_DUPLICATE_KEY;
>>>        uint max_length;
>>>        /* Write the key in the error message */
>>>        char key[MAX_KEY_LENGTH];
>>> @@ -2611,110 +2640,127 @@ void handler::print_error(int error, myf
>>>        /* Table is opened and defined at this point */
>>>        key_unpack(&str,table,(uint) key_nr);
>>>        max_length= (MYSQL_ERRMSG_SIZE-
>>> -                   (uint) strlen(ER(ER_FOREIGN_DUPLICATE_KEY)));
>>> +                   (uint) strlen(ER(textno)));
>>>        if (str.length() >= max_length)
>>>        {
>>>          str.length(max_length-4);
>>>          str.append(STRING_WITH_LEN("..."));
>>>        }
>>> -      my_error(ER_FOREIGN_DUPLICATE_KEY, MYF(0),
> table_share->table_name.str,
>>> -        str.c_ptr_safe(), key_nr+1);
>>> -      DBUG_VOID_RETURN;
>>> -    }
>>> +      (*srvflag)= MYF(0);
>>> +      my_buffer_error(textno, ebuff, size, table_share->table_name.str, 
>>> +                          str.c_ptr_safe(), key_nr+1);
>>> +      DBUG_RETURN(textno);
>>> +    } 
>>>      textno= ER_DUP_KEY;
>>>      break;
>>>    }
>>>    case HA_ERR_NULL_IN_SPATIAL:
>>> -    my_error(ER_CANT_CREATE_GEOMETRY_OBJECT, MYF(0));
>>> -    DBUG_VOID_RETURN;
>>> +    textno= ER_CANT_CREATE_GEOMETRY_OBJECT;
>>> +    (*srvflag)= MYF(0);
>>> +    my_buffer_error(textno, ebuff, size);
>>> +    DBUG_RETURN(textno);
>>> +    break;
>>>    case HA_ERR_FOUND_DUPP_UNIQUE:
>>> -    textno=ER_DUP_UNIQUE;
>>> +    textno= ER_DUP_UNIQUE;
>>>      break;
>>>    case HA_ERR_RECORD_CHANGED:
>>> -    textno=ER_CHECKREAD;
>>> +    textno= ER_CHECKREAD;
>>>      break;
>>>    case HA_ERR_CRASHED:
>>> -    textno=ER_NOT_KEYFILE;
>>> +    textno= ER_NOT_KEYFILE;
>>>      break;
>>>    case HA_ERR_WRONG_IN_RECORD:
>>>      textno= ER_CRASHED_ON_USAGE;
>>>      break;
>>>    case HA_ERR_CRASHED_ON_USAGE:
>>> -    textno=ER_CRASHED_ON_USAGE;
>>> +    textno= ER_CRASHED_ON_USAGE;
>>>      break;
>>>    case HA_ERR_NOT_A_TABLE:
>>>      textno= error;
>>>      break;
>>>    case HA_ERR_CRASHED_ON_REPAIR:
>>> -    textno=ER_CRASHED_ON_REPAIR;
>>> +    textno= ER_CRASHED_ON_REPAIR;
>>>      break;
>>>    case HA_ERR_OUT_OF_MEM:
>>> -    textno=ER_OUT_OF_RESOURCES;
>>> +    textno= ER_OUT_OF_RESOURCES;
>>>      break;
>>>    case HA_ERR_WRONG_COMMAND:
>>> -    textno=ER_ILLEGAL_HA;
>>> +    textno= ER_ILLEGAL_HA;
>>>      break;
>>>    case HA_ERR_OLD_FILE:
>>> -    textno=ER_OLD_KEYFILE;
>>> +    textno= ER_OLD_KEYFILE;
>>>      break;
>>>    case HA_ERR_UNSUPPORTED:
>>> -    textno=ER_UNSUPPORTED_EXTENSION;
>>> +    textno= ER_UNSUPPORTED_EXTENSION;
>>>      break;
>>>    case HA_ERR_RECORD_FILE_FULL:
>>>    case HA_ERR_INDEX_FILE_FULL:
>>>    {
>>> -    textno=ER_RECORD_FILE_FULL;
>>> +    textno= ER_RECORD_FILE_FULL;
>>>      /* Write the error message to error log */
>>> -    errflag|= ME_NOREFRESH;
>>> +    if (srvflag)
>>> +      (*srvflag)|= ME_NOREFRESH;
>>>      break;
>>>    }
>>>    case HA_ERR_LOCK_WAIT_TIMEOUT:
>>> -    textno=ER_LOCK_WAIT_TIMEOUT;
>>> +    textno= ER_LOCK_WAIT_TIMEOUT;
>>>      break;
>>>    case HA_ERR_LOCK_TABLE_FULL:
>>> -    textno=ER_LOCK_TABLE_FULL;
>>> +    textno= ER_LOCK_TABLE_FULL;
>>>      break;
>>>    case HA_ERR_LOCK_DEADLOCK:
>>> -    textno=ER_LOCK_DEADLOCK;
>>> +    textno= ER_LOCK_DEADLOCK;
>>>      break;
>>>    case HA_ERR_READ_ONLY_TRANSACTION:
>>> -    textno=ER_READ_ONLY_TRANSACTION;
>>> +    textno= ER_READ_ONLY_TRANSACTION;
>>>      break;
>>>    case HA_ERR_CANNOT_ADD_FOREIGN:
>>> -    textno=ER_CANNOT_ADD_FOREIGN;
>>> +    textno= ER_CANNOT_ADD_FOREIGN;
>>>      break;
>>>    case HA_ERR_ROW_IS_REFERENCED:
>>>    {
>>>      String str;
>>> +    textno= ER_ROW_IS_REFERENCED_2;
>>>      get_error_message(error, &str);
>>> -    my_error(ER_ROW_IS_REFERENCED_2, MYF(0), str.c_ptr_safe());
>>> -    DBUG_VOID_RETURN;
>>> +    (*srvflag)= MYF(0);
>>> +    my_buffer_error(textno, ebuff, size, str.c_ptr_safe());
>>> +    DBUG_RETURN(textno);
>>> +    break;
>>>    }
>>>    case HA_ERR_NO_REFERENCED_ROW:
>>>    {
>>>      String str;
>>> +    textno= ER_NO_REFERENCED_ROW_2;
>>>      get_error_message(error, &str);
>>> -    my_error(ER_NO_REFERENCED_ROW_2, MYF(0), str.c_ptr_safe());
>>> -    DBUG_VOID_RETURN;
>>> +    (*srvflag)= MYF(0);
>>> +    my_buffer_error(textno, ebuff, size, str.c_ptr_safe());
>>> +    DBUG_RETURN(textno);
>>> +    break;
>>>    }
>>>    case HA_ERR_TABLE_DEF_CHANGED:
>>> -    textno=ER_TABLE_DEF_CHANGED;
>>> +    textno= ER_TABLE_DEF_CHANGED;
>>>      break;
>>>    case HA_ERR_NO_SUCH_TABLE:
>>> -    my_error(ER_NO_SUCH_TABLE, MYF(0), table_share->db.str,
>>> -             table_share->table_name.str);
>>> -    DBUG_VOID_RETURN;
>>> +    textno= ER_NO_SUCH_TABLE;
>>> +    (*srvflag)= MYF(0);
>>> +    my_buffer_error(textno, ebuff, size, table_share->db.str,
>>> +                        table_share->table_name.str);
>>> +    DBUG_RETURN(textno);
>>> +    break;
>>>    case HA_ERR_RBR_LOGGING_FAILED:
>>>      textno= ER_BINLOG_ROW_LOGGING_FAILED;
>>>      break;
>>>    case HA_ERR_DROP_INDEX_FK:
>>>    {
>>> +    textno= ER_DROP_INDEX_FK;
>>>      const char *ptr= "???";
>>>      uint key_nr= get_dup_key(error);
>>>      if ((int) key_nr >= 0)
>>>        ptr= table->key_info[key_nr].name;
>>> -    my_error(ER_DROP_INDEX_FK, MYF(0), ptr);
>>> -    DBUG_VOID_RETURN;
>>> +    (*srvflag)= MYF(0);
>>> +    my_buffer_error(textno, ebuff, size, ptr);
>>> +    DBUG_RETURN(textno);
>>> +    break;
>>>    }
>>>    case HA_ERR_TABLE_NEEDS_UPGRADE:
>>>      textno=ER_TABLE_NEEDS_UPGRADE;
>>> @@ -2732,24 +2778,26 @@ void handler::print_error(int error, myf
>>>      {
>>>        /* The error was "unknown" to this function.
>>>  	 Ask handler if it has got a message for this error */
>>> -      bool temporary= FALSE;
>>>        String str;
>>> -      temporary= get_error_message(error, &str);
>>> -      if (!str.is_empty())
>>> +      bool temporary= get_error_message(error, &str);
>>> +      textno= (str.is_empty() ? ER_GET_ERRNO :
>>> +              (temporary ? ER_GET_TEMPORARY_ERRMSG : ER_GET_ERRMSG));
>>> +      if (textno != ER_GET_ERRNO)
>>>        {
>>> -	const char* engine= table_type();
>>> -	if (temporary)
>>> -	  my_error(ER_GET_TEMPORARY_ERRMSG, MYF(0), error, str.ptr(), engine);
>>> -	else
>>> -	  my_error(ER_GET_ERRMSG, MYF(0), error, str.ptr(), engine);
>>> +        const char* engine= table_type();
>>> +        (*srvflag)= MYF(0);
>>> +        my_buffer_error(textno, ebuff, size, error, str.ptr(), engine);
>>>        }
>>>        else
>>> -	my_error(ER_GET_ERRNO,errflag,error);
>>> -      DBUG_VOID_RETURN;
>>> +      {
>>> +        my_buffer_error(textno, ebuff, size, error);
>>> +      }
>>> +      DBUG_RETURN(textno);
>>> +      break;
>>>      }
>>>    }
>>> -  my_error(textno, errflag, table_share->table_name.str, error);
>>> -  DBUG_VOID_RETURN;
>>> +  my_buffer_error(textno, ebuff, size, table_share->table_name.str,
> error);
>>> +  DBUG_RETURN(textno);
>>>  }
>>>  
>>>
>>>
>>> === modified file 'sql/handler.h'
>>> --- a/sql/handler.h	2008-12-10 20:14:50 +0000
>>> +++ b/sql/handler.h	2009-03-20 12:40:23 +0000
>>> @@ -1262,8 +1262,11 @@ public:
>>>  
>>>    void adjust_next_insert_id_after_explicit_value(ulonglong nr);
>>>    int update_auto_increment();
>>> +  void process_keydup_error(uint key_nr, const char *msg,
>>> +                            char* ebuff, int size);
>>>    void print_keydup_error(uint key_nr, const char *msg);
>>> -  virtual void print_error(int error, myf errflag);
>>> +  virtual int process_error(int error, char* ebuff, int size, myf*
> srvflag);
>>> +  void print_error(int error, myf errflag);
>>>    virtual bool get_error_message(int error, String *buf);
>>>    uint get_dup_key(int error);
>>>    virtual void change_table_ptr(TABLE *table_arg, TABLE_SHARE *share)
>>>
>>> === modified file 'sql/log_event.cc'
>>> --- a/sql/log_event.cc	2009-02-04 11:08:27 +0000
>>> +++ b/sql/log_event.cc	2009-03-20 12:40:23 +0000
>>> @@ -277,6 +277,47 @@ static void clear_all_errors(THD *thd, R
>>>    rli->clear_error();
>>>  }
>>>  
>>> +inline int idempotent_error_code(int err_code)
>>> +{
>>> +  int ret= 0;
>>> +
>>> +  switch (err_code)
>>> +  {
>>> +    case 0:
>>> +      ret= 1;
>>> +    break;
>>> +    /*
>>> +      The following list of "idempotent" errors
>>> +      means that an error from the list might happen
>>> +      because of idempotent (more than once)
>>> +      applying of a binlog file.
>>> +      Notice, that binlog has a  ddl operation its
>>> +      second applying may cause
>>> +
>>> +      case HA_ERR_TABLE_DEF_CHANGED:
>>> +      case HA_ERR_CANNOT_ADD_FOREIGN:
>>> +
>>> +      which are not included into to the list.
>>> +
>>> +      Note that HA_ERR_RECORD_DELETED is not in the list since
>>> +      do_exec_row() should not return that error code.
>>> +    */
>>> +    case HA_ERR_RECORD_CHANGED:
>>> +    case HA_ERR_KEY_NOT_FOUND:
>>> +    case HA_ERR_END_OF_FILE:
>>> +    case HA_ERR_FOUND_DUPP_KEY:
>>> +    case HA_ERR_FOUND_DUPP_UNIQUE:
>>> +    case HA_ERR_FOREIGN_DUPLICATE_KEY:
>>> +    case HA_ERR_NO_REFERENCED_ROW:
>>> +    case HA_ERR_ROW_IS_REFERENCED:
>>> +      ret= 1;
>>> +    break;
>>> +    default:
>>> +      ret= 0;
>>> +    break;
>>> +  }
>>> +  return (ret);
>>> +}
>>>  
>>>  /**
>>>    Ignore error code specified on command line.
>>> @@ -7151,7 +7192,9 @@ int Rows_log_event::do_apply_event(Relay
>>>        {
>>>          /*
>>>            Error reporting borrowed from Query_log_event with many excessive
>>> -          simplifications (we don't honour --slave-skip-errors)
>>> +          simplifications. 
>>> +          We should not honour --slave-skip-errors at this point as we are
>>> +          having severe errors which should not be skiped.
>>>          */
>>>          rli->report(ERROR_LEVEL, actual_error,
>>>                      "Error '%s' on opening tables",
>>> @@ -7177,6 +7220,10 @@ int Rows_log_event::do_apply_event(Relay
>>>        {
>>>          if (ptr->m_tabledef.compatible_with(rli, ptr->table))
>>>          {
>>> +          /*
>>> +            We should not honour --slave-skip-errors at this point as we
> are
>>> +            having severe errors which should not be skiped.
>>> +          */
>>>            mysql_unlock_tables(thd, thd->lock);
>>>            thd->lock= 0;
>>>            thd->is_slave_error= 1;
>>> @@ -7214,7 +7261,7 @@ int Rows_log_event::do_apply_event(Relay
>>>      m_table=
> const_cast<Relay_log_info*>(rli)->m_table_map.get_table(m_table_id);
>>>  
>>>    DBUG_PRINT("debug", ("m_table: 0x%lx, m_table_id: %lu", (ulong) m_table,
> m_table_id));
>>> -
>>> +  
>>>    if (table)
>>>    {
>>>      /*
>>> @@ -7284,48 +7331,32 @@ int Rows_log_event::do_apply_event(Relay
>>>        DBUG_ASSERT(error != HA_ERR_RECORD_DELETED);
>>>  
>>>        table->in_use = old_thd;
>>> -      switch (error)
>>> -      {
>>> -      case 0:
>>> -	break;
>>> -      /*
>>> -        The following list of "idempotent" errors
>>> -        means that an error from the list might happen
>>> -        because of idempotent (more than once) 
>>> -        applying of a binlog file.
>>> -        Notice, that binlog has a  ddl operation its
>>> -        second applying may cause
>>> -
>>> -        case HA_ERR_TABLE_DEF_CHANGED:
>>> -        case HA_ERR_CANNOT_ADD_FOREIGN:
>>>  
>>> -        which are not included into to the list.
>>> -
>>> -        Note that HA_ERR_RECORD_DELETED is not in the list since
>>> -        do_exec_row() should not return that error code.
>>> -      */
>>> -      case HA_ERR_RECORD_CHANGED:
>>> -      case HA_ERR_KEY_NOT_FOUND:
>>> -      case HA_ERR_END_OF_FILE:
>>> -      case HA_ERR_FOUND_DUPP_KEY:
>>> -      case HA_ERR_FOUND_DUPP_UNIQUE:
>>> -      case HA_ERR_FOREIGN_DUPLICATE_KEY:
>>> -      case HA_ERR_NO_REFERENCED_ROW:
>>> -      case HA_ERR_ROW_IS_REFERENCED:
>>> -
>>> -        if (bit_is_set(slave_exec_mode, SLAVE_EXEC_MODE_IDEMPOTENT) == 1)
>>> -        {
>>> -          if (global_system_variables.log_warnings)
>>> -            slave_rows_error_report(WARNING_LEVEL, error, rli, thd, table,
>>> -                                    get_type_str(),
>>> -                                    RPL_LOG_NAME, (ulong) log_pos);
>>> -          error= 0;
>>> -        }
>>> -        break;
>>> -        
>>> -      default:
>>> -	thd->is_slave_error= 1;
>>> -	break;
>>> +      DBUG_PRINT("info",("error (0) %d %d \n", error, thd->is_error() ?
> thd->main_da.sql_errno() : 0));
>>> +      if (error)
>>> +      {
>>> +          int srvflag;
>>> +          char ebuff[MYSQL_ERRMSG_SIZE];
>>> +          uint actual_error_code = (thd->is_error() ?
> thd->main_da.sql_errno() :
>>> +                                    table->file->process_error(error,
> ebuff,
>>> +                                                              
> MYSQL_ERRMSG_SIZE,
>>> +                                                              
> &srvflag));
>>> +          if (ignored_error_code(actual_error_code) ||
>>> +              (idempotent_error_code(error) &&
>>> +               bit_is_set(slave_exec_mode, SLAVE_EXEC_MODE_IDEMPOTENT)))
>>> +          {
>>> +            if (global_system_variables.log_warnings)
>>> +            {
>>> +              sql_print_warning(ebuff);
>>> +              slave_rows_error_report(WARNING_LEVEL, error, rli, thd,
> table,
>>> +                                      get_type_str(),
>>> +                                      RPL_LOG_NAME, (ulong) log_pos);
>>> +            }
>>> +            clear_all_errors(thd, const_cast<Relay_log_info*>(rli));
>>> +            thd->killed= THD::NOT_KILLED;
>>> +            thd->is_slave_error= 0;
>>> +            error= 0;
>>> +          }
>>>        }
>>>  
>>>        /*
>>> @@ -7353,6 +7384,9 @@ int Rows_log_event::do_apply_event(Relay
>>>      DBUG_EXECUTE_IF("STOP_SLAVE_after_first_Rows_event",
>>>                      const_cast<Relay_log_info*>(rli)->abort_slave=
> 1;);
>>>      error= do_after_row_operations(rli, error);
>>> +
>>> +    DBUG_PRINT("info",("error (1) %d %d \n", error, thd->is_error() ?
> thd->main_da.sql_errno() : 0));
>>> +
>>>      if (!cache_stmt)
>>>      {
>>>        DBUG_PRINT("info", ("Marked that we need to keep log"));
>>> @@ -7366,29 +7400,39 @@ int Rows_log_event::do_apply_event(Relay
>>>    */
>>>    if (rli->tables_to_lock && get_flags(STMT_END_F))
>>>      const_cast<Relay_log_info*>(rli)->clear_tables_to_lock();
>>> -  
>>> -  if (error)
>>> -  {                     /* error has occured during the transaction */
>>> -    slave_rows_error_report(ERROR_LEVEL, error, rli, thd, table,
>>> -                            get_type_str(), RPL_LOG_NAME, (ulong) log_pos);
>>> -  }
>>> +
>>>    if (error)
>>>    {
>>> -    /*
>>> -      If one day we honour --skip-slave-errors in row-based replication,
> and
>>> -      the error should be skipped, then we would clear mappings, rollback,
>>> -      close tables, but the slave SQL thread would not stop and then may
>>> -      assume the mapping is still available, the tables are still open...
>>> -      So then we should clear mappings/rollback/close here only if this is
> a
>>> -      STMT_END_F.
>>> -      For now we code, knowing that error is not skippable and so slave SQL
>>> -      thread is certainly going to stop.
>>> -      rollback at the caller along with sbr.
>>> -    */
>>> -    thd->reset_current_stmt_binlog_row_based();
>>> -    const_cast<Relay_log_info*>(rli)->cleanup_context(thd, error);
>>> -    thd->is_slave_error= 1;
>>> -    DBUG_RETURN(error);
>>> +    DBUG_PRINT("info",("error (2) %d %d \n", error, thd->is_error() ?
> thd->main_da.sql_errno() : 0));
>>> +    int srvflag;
>>> +    char ebuff[MYSQL_ERRMSG_SIZE];
>>> +    uint actual_error_code = (thd->is_error() ?
> thd->main_da.sql_errno() :
>>> +                              table->file->process_error(error,
> ebuff,
>>> +                                                         MYSQL_ERRMSG_SIZE,
>>> +                                                         &srvflag));
>>> +    if (ignored_error_code(actual_error_code))
>>> +    {
>>> +      if (global_system_variables.log_warnings)
>>> +      {
>>> +        sql_print_warning(ebuff);
>>> +        slave_rows_error_report(WARNING_LEVEL, error, rli, thd, table,
>>> +                                get_type_str(),
>>> +                                RPL_LOG_NAME, (ulong) log_pos);
>>> +      }
>>> +      clear_all_errors(thd, const_cast<Relay_log_info*>(rli));
>>> +      thd->killed= THD::NOT_KILLED;
>>> +      error= 0;
>>> +      thd->is_slave_error= 0;
>>> +    }
>>> +    else 
>>> +    {
>>> +      slave_rows_error_report(ERROR_LEVEL, error, rli, thd, table,
>>> +                               get_type_str(),
>>> +                               RPL_LOG_NAME, (ulong) log_pos);
>>> +      thd->reset_current_stmt_binlog_row_based();
>>> +      const_cast<Relay_log_info*>(rli)->cleanup_context(thd,
> error);
>>> +      thd->is_slave_error= 1;
>>> +    }
>>>    }
>>>  
>>>    /*
>>> @@ -7396,7 +7440,8 @@ int Rows_log_event::do_apply_event(Relay
>>>      since we have no access to table there, we do the setting of
>>>      last_event_start_time here instead.
>>>    */
>>> -  if (table && (table->s->primary_key == MAX_KEY) &&
>>> +  if (!error && 
>>> +      table && (table->s->primary_key == MAX_KEY) &&
>>>        !cache_stmt && get_flags(STMT_END_F) == RLE_NO_FLAGS)
>>>    {
>>>      /*
>>> @@ -7418,7 +7463,7 @@ int Rows_log_event::do_apply_event(Relay
>>>      const_cast<Relay_log_info*>(rli)->last_event_start_time=
> my_time(0);
>>>    }
>>>  
>>> -  DBUG_RETURN(0);
>>> +  DBUG_RETURN(error);
>>>  }
>>>  
>>>  Log_event::enum_skip_reason
>>>
>>> === modified file 'sql/partition_info.cc'
>>> --- a/sql/partition_info.cc	2008-12-02 10:18:01 +0000
>>> +++ b/sql/partition_info.cc	2009-03-20 12:40:23 +0000
>>> @@ -1075,22 +1075,41 @@ end:
>>>    DBUG_RETURN(result);
>>>  }
>>>  
>>> +void partition_info::print_no_partition_found(TABLE *table) 
>>> +{
>>> +  DBUG_ENTER("partition_info::print_no_partition_found");
>>> +
>>> +  char ebuff[MYSQL_ERRMSG_SIZE];
>>> +  myf srvflag= MYF(0);
>>> +  int srverror= process_no_partition_found(table, ebuff, MYSQL_ERRMSG_SIZE,
>>> +                                           &srvflag);
>>> +  my_message(srverror, ebuff, srvflag);
>>> +
>>> +  DBUG_VOID_RETURN;
>>> +}
>>>  
>>>  /*
>>> -  Print error for no partition found
>>> +  Process error for no partition found
>>>  
>>>    SYNOPSIS
>>> -    print_no_partition_found()
>>> +    process_no_partition_found()
>>>      table                        Table object
>>> +    ebuff                        char* buffer
>>> +    srvflag                      MYF flags
>>>  
>>>    RETURN VALUES
>>>  */
>>> -
>>> -void partition_info::print_no_partition_found(TABLE *table)
>>> +int partition_info::process_no_partition_found(TABLE *table, char* ebuff, 
>>> +                                               int size, myf* srvflag)
>>>  {
>>>    char buf[100];
>>>    char *buf_ptr= (char*)&buf;
>>>    TABLE_LIST table_list;
>>> +  int textno= ER_NO_PARTITION_FOR_GIVEN_VALUE;
>>> +
>>> +  DBUG_ENTER("partition_info::process_no_partition_found");
>>> +
>>> +  DBUG_ASSERT(ebuff != 0 && size >= 0 && srvflag != 0);
>>>  
>>>    bzero(&table_list, sizeof(table_list));
>>>    table_list.db= table->s->db.str;
>>> @@ -1098,8 +1117,7 @@ void partition_info::print_no_partition_
>>>  
>>>    if (check_single_table_access(current_thd,
>>>                                  SELECT_ACL, &table_list, TRUE))
>>> -    my_message(ER_NO_PARTITION_FOR_GIVEN_VALUE,
>>> -               ER(ER_NO_PARTITION_FOR_GIVEN_VALUE_SILENT), MYF(0));
>>> +    my_buffer_format(ebuff, size,
> ER(ER_NO_PARTITION_FOR_GIVEN_VALUE_SILENT));
>>>    else
>>>    {
>>>      my_bitmap_map *old_map= dbug_tmp_use_all_columns(table,
> table->read_set);
>>> @@ -1108,9 +1126,13 @@ void partition_info::print_no_partition_
>>>      else
>>>        longlong2str(err_value, buf,
>>>                     part_expr->unsigned_flag ? 10 : -10);
>>> -    my_error(ER_NO_PARTITION_FOR_GIVEN_VALUE, MYF(0), buf_ptr);
>>> +
>>> +    my_buffer_error(textno, ebuff, size, buf_ptr);
>>>      dbug_tmp_restore_column_map(table->read_set, old_map);
>>>    }
>>> +  (*srvflag)= MYF(0);
>>> +
>>> +  DBUG_RETURN(textno);
>>>  }
>>>  /*
>>>    Set up buffers and arrays for fields requiring preparation
>>>
>>> === modified file 'sql/partition_info.h'
>>> --- a/sql/partition_info.h	2008-11-10 20:21:49 +0000
>>> +++ b/sql/partition_info.h	2009-03-20 12:40:23 +0000
>>> @@ -276,6 +276,8 @@ public:
>>>    bool check_partition_info(THD *thd, handlerton **eng_type,
>>>                              handler *file, HA_CREATE_INFO *info,
>>>                              bool check_partition_function);
>>> +  int process_no_partition_found(TABLE *table, char* ebuff, int size,
>>> +                                 myf* srvflag);
>>>    void print_no_partition_found(TABLE *table);
>>>    bool set_up_charset_field_preps();
>>>  private:
>>>
>>> === modified file 'sql/slave.cc'
>>> --- a/sql/slave.cc	2009-01-09 12:49:24 +0000
>>> +++ b/sql/slave.cc	2009-03-20 12:40:23 +0000
>>> @@ -361,6 +361,7 @@ void init_slave_skip_errors(const char* 
>>>    if (!my_strnncoll(system_charset_info,(uchar*)arg,4,(const
> uchar*)"all",4))
>>>    {
>>>      bitmap_set_all(&slave_error_mask);
>>> +    print_slave_skip_errors();
>>>      DBUG_VOID_RETURN;
>>>    }
>>>    for (p= arg ; *p; )