BUG#17138: Error in stored procedure due to fatal error when not really

a fatal error. New handling of ignore error in place.


mysql-test/t/partition.test:
  New test case
sql/ha_ndbcluster.h:
  New handler method to check if error can be ignored
sql/ha_partition.h:
  New handler method to check if error can be ignored
sql/handler.h:
  New handler method to check if error can be ignored
sql/sql_acl.cc:
  Use new handler method
sql/sql_insert.cc:
  Use new handler method
sql/sql_table.cc:
  Use new handler method
sql/sql_union.cc:
  Use new handler method
sql/sql_update.cc:
  Use new handler method

mysql-test/t/partition.test

@@ -1135,4 +1135,27 @@ alter table t1 drop partition p2;
 use test;
 drop database db99;
 
+#
+#BUG 17138 Problem with stored procedure and analyze partition
+#
+create table t1 (a int)
+partition by list (a)
+(partition p0 values in (0));
+
+insert into t1 values (0);
+delimiter //;
+
+create procedure po ()
+begin
+  begin
+    declare continue handler for sqlexception begin end;
+    update ignore t1 set a = 1 where a = 0;
+  end;
+  prepare stmt1 from 'alter table t1';
+  execute stmt1;
+end//
+
+call po()//
+delimiter ;//
+drop table t1;
 --echo End of 5.1 tests

sql/ha_ndbcluster.h

@@ -654,6 +654,14 @@ class ha_ndbcluster: public handler
   int get_default_no_partitions(ulonglong max_rows);
   bool get_no_parts(const char *name, uint *no_parts);
   void set_auto_partitions(partition_info *part_info);
+  virtual bool cannot_ignore_error(int error, uint flags)
+  {
+    if (!handler::cannot_ignore_error(error, flags))
+      return FALSE;
+    if (error == HA_ERR_NO_PARTITION_FOUND)
+      return FALSE;
+    return TRUE;
+  }
 
   THR_LOCK_DATA **store_lock(THD *thd,
                              THR_LOCK_DATA **to,

sql/ha_partition.h

@@ -302,6 +302,14 @@ class ha_partition :public handler
   virtual void start_bulk_insert(ha_rows rows);
   virtual int end_bulk_insert();
 
+  virtual bool cannot_ignore_error(int error, uint flags)
+  {
+    if (!handler::cannot_ignore_error(error, flags))
+      return FALSE;
+    if (error == HA_ERR_NO_PARTITION_FOUND)
+      return FALSE;
+    return TRUE;
+  }
   /*
     -------------------------------------------------------------------------
     MODULE full table scan

sql/handler.h

@@ -974,7 +974,28 @@ class handler :public Sql_alloc
   bool has_transactions()
   { return (ha_table_flags() & HA_NO_TRANSACTIONS) == 0; }
   virtual uint extra_rec_buf_length() const { return 0; }
-  
+
+  /*
+    This method is used to analyse the error to see whether the error
+    is ignorable or not, certain handlers can have more error that are
+    ignorable than others. E.g. the partition handler can get inserts
+    into a range where there is no partition and this is an ignorable
+    error.
+  */
+#define HA_CHECK_DUPP_KEY 1
+#define HA_CHECK_DUPP_UNIQUE 2
+#define HA_CHECK_DUPP (CHECK_DUPP_KEY + CHECK_DUPP_UNIQUE)
+  virtual bool cannot_ignore_error(int error, uint flags)
+  {
+    if (!error ||
+        ((flags & HA_CHECK_DUPP_KEY) &&
+         error == HA_ERR_FOUND_DUPP_KEY) ||
+         ((flags & HA_CHECK_DUPP_UNIQUE) &&
+          error == HA_ERR_FOUND_DUPP_UNIQUE))
+      return FALSE;
+    return TRUE;
+  }
+
   /*
     Number of rows in table. It will only be called if
     (table_flags() & (HA_HAS_RECORDS | HA_STATS_RECORDS_IS_EXACT)) != 0
@@ -1026,7 +1047,7 @@ class handler :public Sql_alloc
     DBUG_RETURN(rnd_end());
   }
   int ha_reset();
-    
+
   /* this is necessary in many places, e.g. in HANDLER command */
   int ha_index_or_rnd_end()
   {

sql/sql_acl.cc

@@ -2049,8 +2049,7 @@ static int replace_user_table(THD *thd, TABLE *table, const LEX_USER &combo,
   }
   else if ((error=table->file->ha_write_row(table->record[0]))) // insert
   {						// This should never happen
-    if (error && error != HA_ERR_FOUND_DUPP_KEY &&
-	error != HA_ERR_FOUND_DUPP_UNIQUE)	/* purecov: inspected */
+    if (table->file->cannot_ignore_error(error, HA_CHECK_DUPP))
     {
       table->file->print_error(error,MYF(0));	/* purecov: deadcode */
       error= -1;				/* purecov: deadcode */
@@ -2172,7 +2171,7 @@ static int replace_db_table(TABLE *table, const char *db,
   }
   else if (rights && (error= table->file->ha_write_row(table->record[0])))
   {
-    if (error && error != HA_ERR_FOUND_DUPP_KEY) /* purecov: inspected */
+    if (table->file->cannot_ignore_error(error, HA_CHECK_DUPP_KEY))
       goto table_error; /* purecov: deadcode */
   }
 
@@ -2744,7 +2743,7 @@ static int replace_table_table(THD *thd, GRANT_TABLE *grant_table,
   else
   {
     error=table->file->ha_write_row(table->record[0]);
-    if (error && error != HA_ERR_FOUND_DUPP_KEY)
+    if (table->file->cannot_ignore_error(error, HA_CHECK_DUPP_KEY))
       goto table_error;				/* purecov: deadcode */
   }
 
@@ -2862,7 +2861,7 @@ static int replace_routine_table(THD *thd, GRANT_NAME *grant_name,
   else
   {
     error=table->file->ha_write_row(table->record[0]);
-    if (error && error != HA_ERR_FOUND_DUPP_KEY)
+    if (table->file->cannot_ignore_error(error, HA_CHECK_DUPP_KEY))
       goto table_error;
   }
 

sql/sql_insert.cc

@@ -979,6 +979,9 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
       if (error != HA_WRITE_SKIP)
 	goto err;
       table->file->restore_auto_increment(); // it's too early here! BUG#20188
+      if (info->ignore &&
+          !table->file->cannot_ignore_error(error, 0))
+        goto ok_or_after_trg_err;
       if ((int) (key_nr = table->file->get_dup_key(error)) < 0)
       {
 	error=HA_WRITE_SKIP;			/* Database can't find key */
@@ -1062,7 +1065,8 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
         if ((error=table->file->ha_update_row(table->record[1],
                                               table->record[0])))
 	{
-	  if ((error == HA_ERR_FOUND_DUPP_KEY) && info->ignore)
+          if (info->ignore &&
+              !table->file->cannot_ignore_error(error, HA_CHECK_DUPP_KEY))
             goto ok_or_after_trg_err;
           goto err;
 	}
@@ -1148,7 +1152,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
   else if ((error=table->file->ha_write_row(table->record[0])))
   {
     if (!info->ignore ||
-	(error != HA_ERR_FOUND_DUPP_KEY && error != HA_ERR_FOUND_DUPP_UNIQUE))
+        table->file->cannot_ignore_error(error, HA_CHECK_DUPP))
       goto err;
     table->file->restore_auto_increment();
   }

sql/sql_table.cc

@@ -6235,10 +6235,8 @@ copy_data_between_tables(TABLE *from,TABLE *to,
     }
     if ((error=to->file->ha_write_row((byte*) to->record[0])))
     {
-      if ((!ignore &&
-	   handle_duplicates != DUP_REPLACE) ||
-	  (error != HA_ERR_FOUND_DUPP_KEY &&
-	   error != HA_ERR_FOUND_DUPP_UNIQUE))
+      if (!ignore ||
+          to->file->cannot_ignore_error(error, HA_CHECK_DUPP))
       {
          if (error == HA_ERR_FOUND_DUPP_KEY)
          {

sql/sql_union.cc

@@ -65,7 +65,7 @@ bool select_union::send_data(List<Item> &values)
   if ((error= table->file->ha_write_row(table->record[0])))
   {
     /* create_myisam_from_heap will generate error if needed */
-    if (error != HA_ERR_FOUND_DUPP_KEY && error != HA_ERR_FOUND_DUPP_UNIQUE &&
+    if (table->file->cannot_ignore_error(error, HA_CHECK_DUPP) &&
         create_myisam_from_heap(thd, table, &tmp_table_param, error, 1))
       return 1;
   }

sql/sql_update.cc

@@ -541,13 +541,14 @@ int mysql_update(THD *thd,
             break;
           }
 	}
- 	else if (!ignore || error != HA_ERR_FOUND_DUPP_KEY)
+ 	else if (!ignore ||
+                 table->file->cannot_ignore_error(error, HA_CHECK_DUPP_KEY))
 	{
           /*
-            If (ignore && error == HA_ERR_FOUND_DUPP_KEY) we don't have to
+            If (ignore && error is ignorable) we don't have to
             do anything; otherwise...
           */
-          if (error != HA_ERR_FOUND_DUPP_KEY)
+          if (table->file->cannot_ignore_error(error, HA_CHECK_DUPP_KEY))
             thd->fatal_error(); /* Other handler errors are fatal */
 	  table->file->print_error(error,MYF(0));
 	  error= 1;
@@ -1417,13 +1418,14 @@ bool multi_update::send_data(List<Item> &not_used_values)
 					      table->record[0])))
 	{
 	  updated--;
-          if (!ignore || error != HA_ERR_FOUND_DUPP_KEY)
+          if (!ignore ||
+              table->file->cannot_ignore_error(error, HA_CHECK_DUPP_KEY))
 	  {
             /*
-              If (ignore && error == HA_ERR_FOUND_DUPP_KEY) we don't have to
+              If (ignore && error == is ignorable) we don't have to
               do anything; otherwise...
             */
-            if (error != HA_ERR_FOUND_DUPP_KEY)
+            if (table->file->cannot_ignore_error(error, HA_CHECK_DUPP_KEY))
               thd->fatal_error(); /* Other handler errors are fatal */
 	    table->file->print_error(error,MYF(0));
 	    DBUG_RETURN(1);
@@ -1452,8 +1454,7 @@ bool multi_update::send_data(List<Item> &not_used_values)
       /* Write row, ignoring duplicated updates to a row */
       if ((error= tmp_table->file->ha_write_row(tmp_table->record[0])))
       {
-        if (error != HA_ERR_FOUND_DUPP_KEY &&
-            error != HA_ERR_FOUND_DUPP_UNIQUE &&
+        if (tmp_table->file->cannot_ignore_error(error, HA_CHECK_DUPP) &&
             create_myisam_from_heap(thd, tmp_table,
                                          tmp_table_param + offset, error, 1))
 	{
@@ -1576,7 +1577,8 @@ int multi_update::do_updates(bool from_send_error)
 	if ((local_error=table->file->ha_update_row(table->record[1],
 						    table->record[0])))
 	{
-	  if (!ignore || local_error != HA_ERR_FOUND_DUPP_KEY)
+	  if (!ignore ||
+              table->file->cannot_ignore_error(local_error, HA_CHECK_DUPP_KEY))
 	    goto err;
 	}
 	updated++;