MDEV-17333 Assertion in update_auto_increment() upon exotic LOAD

While `handler::next_insert_id` is restored on duplicate key errors
`part_share->next_auto_inc_val` is not restored which causes
discrepancy.

mysql-test/main/auto_increment_ranges.inc

@@ -238,3 +238,23 @@ delete from t1 where a=32767;
 --error HA_ERR_AUTOINC_ERANGE
 insert into t1 values(NULL);
 drop table t1;
+
+--echo #
+--echo # MDEV-17333 Assertion in update_auto_increment() upon exotic LOAD
+--echo #
+--source include/have_partition.inc
+let $mysqld_datadir= `select @@datadir`;
+--write_file $mysqld_datadir/test/load.data
+1	1
+0	2
+3	3
+4	1
+0	1
+6	6
+EOF
+create or replace table t1 (pk int auto_increment, x int, primary key(pk), unique key(x))
+with system versioning partition by system_time interval 2 day
+(partition p1 history, partition pn current);
+load data infile 'load.data' ignore into table t1;
+--remove_file $mysqld_datadir/test/load.data
+drop table t1;

mysql-test/main/auto_increment_ranges_innodb.result

@@ -264,6 +264,17 @@ delete from t1 where a=32767;
 insert into t1 values(NULL);
 ERROR 22003: Out of range value for column 'a' at row 1
 drop table t1;
+#
+# MDEV-17333 Assertion in update_auto_increment() upon exotic LOAD
+#
+create or replace table t1 (pk int auto_increment, x int, primary key(pk), unique key(x))
+with system versioning partition by system_time interval 2 day
+(partition p1 history, partition pn current);
+load data infile 'load.data' ignore into table t1;
+Warnings:
+Warning	1062	Duplicate entry '1' for key 'x'
+Warning	1062	Duplicate entry '1' for key 'x'
+drop table t1;
 create table t1 (pk int auto_increment primary key, f varchar(20));
 insert t1 (f) values ('a'), ('b'), ('c'), ('d');
 select null, f into outfile 'load.data' from t1 limit 1;

mysql-test/main/auto_increment_ranges_myisam.result

@@ -270,3 +270,14 @@ delete from t1 where a=32767;
 insert into t1 values(NULL);
 ERROR 22003: Out of range value for column 'a' at row 1
 drop table t1;
+#
+# MDEV-17333 Assertion in update_auto_increment() upon exotic LOAD
+#
+create or replace table t1 (pk int auto_increment, x int, primary key(pk), unique key(x))
+with system versioning partition by system_time interval 2 day
+(partition p1 history, partition pn current);
+load data infile 'load.data' ignore into table t1;
+Warnings:
+Warning	1062	Duplicate entry '1' for key 'x'
+Warning	1062	Duplicate entry '1' for key 'x'
+drop table t1;

sql/ha_partition.h

@@ -92,6 +92,7 @@ class Partition_share : public Handler_share
   bool auto_inc_initialized;
   mysql_mutex_t auto_inc_mutex;                /**< protecting auto_inc val */
   ulonglong next_auto_inc_val;                 /**< first non reserved value */
+  ulonglong prev_auto_inc_val;                 /**< stored next_auto_inc_val */
   /**
     Hash of partition names. Initialized in the first ha_partition::open()
     for the table_share. After that it is read-only, i.e. no locking required.
@@ -103,6 +104,7 @@ class Partition_share : public Handler_share
   Partition_share()
     : auto_inc_initialized(false),
     next_auto_inc_val(0),
+    prev_auto_inc_val(0),
     partition_name_hash_initialized(false),
     partition_names(NULL)
   {
@@ -371,6 +373,19 @@ class ha_partition :public handler
   MY_BITMAP m_locked_partitions;
   /** Stores shared auto_increment etc. */
   Partition_share *part_share;
+  /** Store and restore next_auto_inc_val over duplicate key errors. */
+  virtual void store_auto_increment()
+  {
+    DBUG_ASSERT(part_share);
+    part_share->prev_auto_inc_val= part_share->next_auto_inc_val;
+    handler::store_auto_increment();
+  }
+  virtual void restore_auto_increment()
+  {
+    DBUG_ASSERT(part_share);
+    part_share->next_auto_inc_val= part_share->prev_auto_inc_val;
+    handler::restore_auto_increment();
+  }
   /** Temporary storage for new partitions Handler_shares during ALTER */
   List<Parts_share_refs> m_new_partitions_share_refs;
   /** Sorted array of partition ids in descending order of number of rows. */

sql/handler.h

@@ -3008,6 +3008,10 @@ class handler :public Sql_alloc
   */
   Handler_share **ha_share;
 
+  /** Stores next_insert_id for handling duplicate key errors. */
+  ulonglong m_prev_insert_id;
+
+
 public:
   handler(handlerton *ht_arg, TABLE_SHARE *share_arg)
     :table_share(share_arg), table(0),
@@ -3030,7 +3034,7 @@ class handler :public Sql_alloc
     auto_inc_intervals_count(0),
     m_psi(NULL), set_top_table_fields(FALSE), top_table(0),
     top_table_field(0), top_table_fields(0),
-    m_lock_type(F_UNLCK), ha_share(NULL)
+    m_lock_type(F_UNLCK), ha_share(NULL), m_prev_insert_id(0)
   {
     DBUG_PRINT("info",
                ("handler created F_UNLCK %d F_RDLCK %d F_WRLCK %d",
@@ -3700,6 +3704,16 @@ class handler :public Sql_alloc
       insert_id_for_cur_row;
   }
 
+  /** Store and restore next_insert_id over duplicate key errors. */
+  virtual void store_auto_increment()
+  {
+    m_prev_insert_id= next_insert_id;
+  }
+  virtual void restore_auto_increment()
+  {
+    restore_auto_increment(m_prev_insert_id);
+  }
+
   virtual void update_create_info(HA_CREATE_INFO *create_info) {}
   int check_old_types();
   virtual int assign_to_keycache(THD* thd, HA_CHECK_OPT* check_opt)

sql/sql_insert.cc

@@ -1691,7 +1691,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
   int error, trg_error= 0;
   char *key=0;
   MY_BITMAP *save_read_set, *save_write_set;
-  ulonglong prev_insert_id= table->file->next_insert_id;
+  table->file->store_auto_increment();
   ulonglong insert_id_for_cur_row= 0;
   ulonglong prev_insert_id_for_cur_row= 0;
   DBUG_ENTER("write_record");
@@ -1844,7 +1844,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
         if (res == VIEW_CHECK_ERROR)
           goto before_trg_err;
 
-        table->file->restore_auto_increment(prev_insert_id);
+        table->file->restore_auto_increment();
         info->touched++;
         if (different_records)
         {
@@ -2038,7 +2038,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
     if (!(thd->variables.old_behavior &
           OLD_MODE_NO_DUP_KEY_WARNINGS_WITH_IGNORE))
       table->file->print_error(error, MYF(ME_JUST_WARNING));
-    table->file->restore_auto_increment(prev_insert_id);
+    table->file->restore_auto_increment();
     goto ok_or_after_trg_err;
   }
 
@@ -2061,7 +2061,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
   table->file->print_error(error,MYF(0));
 
 before_trg_err:
-  table->file->restore_auto_increment(prev_insert_id);
+  table->file->restore_auto_increment();
   if (key)
     my_safe_afree(key, table->s->max_unique_length);
   table->column_bitmaps_set(save_read_set, save_write_set);