This fix+comments was originally made by Alexey Kopytov

LP bug #1035225 / MySQL bug #66301: INSERT ... ON DUPLICATE KEY UPDATE +
innodb_autoinc_lock_mode=1 is broken

The problem was that when certain INSERT ... ON DUPLICATE KEY UPDATE
were executed concurrently on a table containing an AUTO_INCREMENT
column as a primary key, InnoDB would correctly reserve non-overlapping
AUTO_INCREMENT intervals for each statement, but when the server
encountered the first duplicate key error on the secondary key in one of
the statements and performed an UPDATE, it also updated the internal
AUTO_INCREMENT value to the one from the existing row that caused a
duplicate key error, even though the AUTO_INCREMENT value was not
specified explicitly in the UPDATE clause. It would then proceed with
using AUTO_INCREMENT values the range reserved previously by another
statement, causing duplicate key errors on the AUTO_INCREMENT column.

Fixed by changing write_record() to ensure that in case of a duplicate
key error the internal AUTO_INCREMENT counter is only updated when the
AUTO_INCREMENT value was explicitly updated by the UPDATE
clause. Otherwise it is restored to what it was before the duplicate key
error, as that value is unused and can be reused for subsequent
successfully inserted rows.

sql/sql_insert.cc:
  Don't update next_insert_id to the value of a row found during ON DUPLICATE KEY UPDATE.
sql/sql_parse.cc:
  Added DBUG_SYNC
sql/table.h:
  Added next_number_field_updated flag to detect changing of auto increment fields.
  Moved fields a bit to get bool fields after each other (better alignment)

5 files changed, 142 insertions, 7 deletions

diff --git a/mysql-test/suite/innodb/r/auto_increment_dup.result b/mysql-test/suite/innodb/r/auto_increment_dup.result
new file mode 100644
index 00000000000..5bf901cb212
--- /dev/null
+++ b/mysql-test/suite/innodb/r/auto_increment_dup.result
@@ -0,0 +1,33 @@
+drop table if exists t1;
+CREATE TABLE t1(
+id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+k INT,
+c CHAR(1),
+UNIQUE KEY(k)) ENGINE=InnoDB;
+#
+# Connection 1
+#
+SET DEBUG_SYNC='ha_write_row_end SIGNAL continue2 WAIT_FOR continue1';
+affected rows: 0
+INSERT INTO t1(k) VALUES (1), (2), (3) ON DUPLICATE KEY UPDATE c='1';
+#
+# Connection 2
+#
+SET DEBUG_SYNC='start_ha_write_row WAIT_FOR continue2';
+affected rows: 0
+SET DEBUG_SYNC='after_mysql_insert SIGNAL continue1';
+affected rows: 0
+INSERT INTO t1(k) VALUES (2), (4), (5) ON DUPLICATE KEY UPDATE c='2';
+affected rows: 3
+info: Records: 3  Duplicates: 0  Warnings: 0
+affected rows: 4
+info: Records: 3  Duplicates: 1  Warnings: 0
+SET DEBUG_SYNC='RESET';
+SELECT * FROM t1 ORDER BY k;
+id	k	c
+1	1	NULL
+4	2	1
+2	3	NULL
+5	4	NULL
+6	5	NULL
+DROP TABLE t1;
diff --git a/mysql-test/suite/innodb/t/auto_increment_dup.test b/mysql-test/suite/innodb/t/auto_increment_dup.test
new file mode 100644
index 00000000000..ad439024f65
--- /dev/null
+++ b/mysql-test/suite/innodb/t/auto_increment_dup.test
@@ -0,0 +1,51 @@
+##########################################################################
+# LP bug #1035225 / MySQL bug #66301: INSERT ... ON DUPLICATE KEY UPDATE +
+# innodb_autoinc_lock_mode=1 is broken
+##########################################################################
+
+--source include/have_innodb.inc
+--source include/have_debug_sync.inc
+
+--disable_warnings
+drop table if exists t1;
+--enable_warnings
+
+CREATE TABLE t1(
+       id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+       k INT,
+       c CHAR(1),
+       UNIQUE KEY(k)) ENGINE=InnoDB;
+
+--enable_info
+
+--connect(con1, localhost, root)
+--connect(con2, localhost, root)
+
+--connection con1
+
+--echo #
+--echo # Connection 1
+--echo #
+SET DEBUG_SYNC='ha_write_row_end SIGNAL continue2 WAIT_FOR continue1';
+--send INSERT INTO t1(k) VALUES (1), (2), (3) ON DUPLICATE KEY UPDATE c='1'
+
+--connection con2
+--echo #
+--echo # Connection 2
+--echo #
+SET DEBUG_SYNC='start_ha_write_row WAIT_FOR continue2';
+SET DEBUG_SYNC='after_mysql_insert SIGNAL continue1';
+INSERT INTO t1(k) VALUES (2), (4), (5) ON DUPLICATE KEY UPDATE c='2';
+
+--connection con1
+--reap
+--disable_info
+SET DEBUG_SYNC='RESET';
+SELECT * FROM t1 ORDER BY k;
+
+--disconnect con1
+--disconnect con2
+
+--connection default
+
+DROP TABLE t1;
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 42704862c16..3c3b9f85727 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -329,7 +329,7 @@ static int check_insert_fields(THD *thd, TABLE_LIST *table_list,
 
 
 /*
-  Check update fields for the timestamp field.
+  Check update fields for the timestamp and auto_increment fields.
 
   SYNOPSIS
     check_update_fields()
@@ -342,6 +342,9 @@ static int check_insert_fields(THD *thd, TABLE_LIST *table_list,
     If the update fields include the timestamp field,
     remove TIMESTAMP_AUTO_SET_ON_UPDATE from table->timestamp_field_type.
 
+   If the update fields include an autoinc field, set the
+   table->next_number_field_updated flag.
+
   RETURN
     0           OK
     -1          Error
@@ -353,7 +356,9 @@ static int check_update_fields(THD *thd, TABLE_LIST *insert_table_list,
 {
   TABLE *table= insert_table_list->table;
   my_bool timestamp_mark;
+  my_bool autoinc_mark;
   LINT_INIT(timestamp_mark);
+  LINT_INIT(autoinc_mark);
 
   if (table->timestamp_field)
   {
@@ -365,6 +370,19 @@ static int check_update_fields(THD *thd, TABLE_LIST *insert_table_list,
                                           table->timestamp_field->field_index);
   }
 
+  table->next_number_field_updated= FALSE;
+
+  if (table->found_next_number_field)
+  {
+    /*
+      Unmark the auto_increment field so that we can check if this is modified
+      by update_fields
+    */
+    autoinc_mark= bitmap_test_and_clear(table->write_set,
+                                        table->found_next_number_field->
+                                        field_index);
+  }
+
   /* Check the fields we are going to modify */
   if (setup_fields(thd, 0, update_fields, MARK_COLUMNS_WRITE, 0, 0))
     return -1;
@@ -386,6 +404,18 @@ static int check_update_fields(THD *thd, TABLE_LIST *insert_table_list,
       bitmap_set_bit(table->write_set,
                      table->timestamp_field->field_index);
   }
+
+  if (table->found_next_number_field)
+  {
+    if (bitmap_is_set(table->write_set,
+                      table->found_next_number_field->field_index))
+      table->next_number_field_updated= TRUE;
+
+    if (autoinc_mark)
+      bitmap_set_bit(table->write_set,
+                     table->found_next_number_field->field_index);
+  }
+
   return 0;
 }
 
@@ -1563,6 +1593,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
   MY_BITMAP *save_read_set, *save_write_set;
   ulonglong prev_insert_id= table->file->next_insert_id;
   ulonglong insert_id_for_cur_row= 0;
+  ulonglong prev_insert_id_for_cur_row= 0;
   DBUG_ENTER("write_record");
 
   info->records++;
@@ -1709,6 +1740,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
             INSERT query, which is handled separately by
             THD::arg_of_last_insert_id_function.
           */
+          prev_insert_id_for_cur_row= table->file->insert_id_for_cur_row;
           insert_id_for_cur_row= table->file->insert_id_for_cur_row= 0;
           trg_error= (table->triggers &&
                       table->triggers->process_triggers(thd, TRG_EVENT_UPDATE,
@@ -1716,9 +1748,22 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
           info->copied++;
         }
 
-        if (table->next_number_field)
-          table->file->adjust_next_insert_id_after_explicit_value(
-            table->next_number_field->val_int());
+        /*
+          Only update next_insert_id if the AUTO_INCREMENT value was explicitly
+          updated, so we don't update next_insert_id with the value from the
+          row being updated. Otherwise reset next_insert_id to what it was
+          before the duplicate key error, since that value is unused.
+        */
+        if (table->next_number_field_updated)
+        {
+          DBUG_ASSERT(table->next_number_field != NULL);
+
+          table->file->adjust_next_insert_id_after_explicit_value(table->next_number_field->val_int());
+        }
+        else
+        {
+          table->file->restore_auto_increment(prev_insert_id_for_cur_row);
+        }
         goto ok_or_after_trg_err;
       }
       else /* DUP_REPLACE */
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index f5124d60f3f..d2eb6ae8d1f 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2938,6 +2938,7 @@ end_with_restore_list:
                       DBUG_ASSERT(!debug_sync_set_action(thd,
                                                          STRING_WITH_LEN(act2)));
                     };);
+    DEBUG_SYNC(thd, "after_mysql_insert");
     break;
   }
   case SQLCOM_REPLACE_SELECT:
diff --git a/sql/table.h b/sql/table.h
index f5ae3bcebb3..0259e063748 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1053,15 +1053,20 @@ public:
   uint		db_stat;		/* mode of file as in handler.h */
   /* number of select if it is derived table */
   uint          derived_select_number;
-  int		current_lock;           /* Type of lock on table */
-  bool copy_blobs;			/* copy_blobs when storing */
-
   /*
     0 or JOIN_TYPE_{LEFT|RIGHT}. Currently this is only compared to 0.
     If maybe_null !=0, this table is inner w.r.t. some outer join operation,
     and null_row may be true.
   */
   uint maybe_null;
+  int		current_lock;           /* Type of lock on table */
+  bool copy_blobs;			/* copy_blobs when storing */
+  /*
+    Set if next_number_field is in the UPDATE fields of INSERT ... ON DUPLICATE
+    KEY UPDATE.
+  */
+  bool next_number_field_updated;
+
   /*
     If true, the current table row is considered to have all columns set to 
     NULL, including columns declared as "not null" (see maybe_null).