MDEV-30046 wrong row targeted with "insert ... on duplicate" and "rep…

…lace"

When HA_DUPLICATE_POS is not supported, the row to replace was navigated by
ha_index_read_idx_map, which uses only hash to navigate.

Suchwise, given a hash collision it may choose an incorrect row.

handler::position would be correct and very convenient to use here.
The code is updated to set dup_ref by a handler independently of engine
capabilities, when extra lookup is made (for long unique or something else,
for example WITHOUT OVERLAPS)

REPLACE works correctly with this regard, however there are other cases yet
uncovered: LOAD DATA and replication with IDEMPOTENT mode. They will be
fixed by a follow-up, after refactoring.

sql/handler.cc

@@ -4549,6 +4549,12 @@ uint handler::get_dup_key(int error)
   DBUG_RETURN(errkey);
 }
 
+bool handler::has_dup_ref() const
+{
+  DBUG_ASSERT(lookup_errkey != (uint)-1 || errkey != (uint)-1);
+  return ha_table_flags() & HA_DUPLICATE_POS || lookup_errkey != (uint)-1;
+}
+
 
 /**
   Delete all files with extension from bas_ext().
@@ -6978,11 +6984,8 @@ int handler::check_duplicate_long_entry_key(const uchar *new_rec, uint key_no)
   if (error == HA_ERR_FOUND_DUPP_KEY)
   {
     table->file->lookup_errkey= key_no;
-    if (ha_table_flags() & HA_DUPLICATE_POS)
-    {
-      lookup_handler->position(table->record[0]);
-      memcpy(table->file->dup_ref, lookup_handler->ref, ref_length);
-    }
+    lookup_handler->position(table->record[0]);
+    memcpy(table->file->dup_ref, lookup_handler->ref, ref_length);
   }
   restore_record(table, file->lookup_buffer);
   table->restore_blob_values(blob_storage);

sql/handler.h

@@ -3464,6 +3464,7 @@ class handler :public Sql_alloc
   virtual void print_error(int error, myf errflag);
   virtual bool get_error_message(int error, String *buf);
   uint get_dup_key(int error);
+  bool has_dup_ref() const;
   /**
     Retrieves the names of the table and the key for which there was a
     duplicate entry in the case of HA_ERR_FOREIGN_DUPLICATE_KEY.

sql/sql_insert.cc

@@ -918,7 +918,8 @@ bool mysql_insert(THD *thd, TABLE_LIST *table_list,
     {
       create_lookup_handler= true;
       table->file->extra(HA_EXTRA_IGNORE_DUP_KEY);
-      if (table->file->ha_table_flags() & HA_DUPLICATE_POS)
+      if (table->file->ha_table_flags() & HA_DUPLICATE_POS ||
+          table->s->long_unique_table)
       {
         if (table->file->ha_rnd_init_with_error(0))
           goto abort;
@@ -1179,7 +1180,8 @@ bool mysql_insert(THD *thd, TABLE_LIST *table_list,
     if (duplic != DUP_ERROR || ignore)
     {
       table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
-      if (table->file->ha_table_flags() & HA_DUPLICATE_POS)
+      if (table->file->ha_table_flags() & HA_DUPLICATE_POS ||
+          table->s->long_unique_table)
         table->file->ha_rnd_end();
     }
 
@@ -1884,7 +1886,7 @@ int write_record(THD *thd, TABLE *table, COPY_INFO *info, select_result *sink)
       if (info->handle_duplicates == DUP_REPLACE && table->next_number_field &&
           key_nr == table->s->next_number_index && insert_id_for_cur_row > 0)
 	goto err;
-      if (table->file->ha_table_flags() & HA_DUPLICATE_POS)
+      if (table->file->has_dup_ref())
       {
         DBUG_ASSERT(table->file->inited == handler::RND);
 	if (table->file->ha_rnd_pos(table->record[1],table->file->dup_ref))