MDEV-15990 REPLACE on a precise-versioned table returns ER_DUP_ENTRYbb-10.3-nikita

8 files changed, 212 insertions, 49 deletions

diff --git a/mysql-test/suite/versioning/r/delete.result b/mysql-test/suite/versioning/r/delete.result
index 0f9e2c22130..6f8c8921790 100644
--- a/mysql-test/suite/versioning/r/delete.result
+++ b/mysql-test/suite/versioning/r/delete.result
@@ -146,6 +146,5 @@ delete from t1;
 select f1, f3, check_row_ts(row_start, row_end) from t1 for system_time all;
 f1	f3	check_row_ts(row_start, row_end)
 1	1	HISTORICAL ROW
-1	NULL	ERROR: row_end == row_start
 1	1	HISTORICAL ROW
 drop table t1;
diff --git a/mysql-test/suite/versioning/r/replace.result b/mysql-test/suite/versioning/r/replace.result
index 57a992cce49..af23235e2c9 100644
--- a/mysql-test/suite/versioning/r/replace.result
+++ b/mysql-test/suite/versioning/r/replace.result
@@ -60,4 +60,81 @@ replace into t1 values (1),(2);
 connection con1;
 replace into t1 values (1),(2);
 ERROR 23000: Duplicate entry '1' for key 'PRIMARY'
+connection default;
+set timestamp=default;
 drop table t1;
+#
+# MDEV-15990: REPLACE on a precise-versioned table
+# returns duplicate key error (ER_DUP_ENTRY)
+create or replace table t(
+id int,
+KEY_TYPE(id),
+x int,
+row_start SYS_DATATYPE as row start invisible,
+row_end SYS_DATATYPE as row end invisible,
+period for system_time(row_start, row_end)
+) with system versioning;
+replace t values (1, 3), (1, 30), (1, 300);
+create temporary table log(id int key auto_increment, s char(30));
+create trigger tr1del before delete on t
+for each row insert log(s) select 'DEL BEFORE' union
+select concat(id, ' ', x, ' ',
+if(row_start = (select VERS_CURRENT_VALUE),
+"CURRENT", "OTHER"),
+' ', check_row(row_start, row_end))
+from t for system_time all;
+create trigger tr2del after delete on t
+for each row insert log(s) select 'DEL AFTER' union
+select concat(id, ' ', x, ' ',
+if(row_start = (select VERS_CURRENT_VALUE),
+"CURRENT", "OTHER"),
+' ', check_row(row_start, row_end))
+from t for system_time all;
+replace t values (1, 4), (1, 40), (1, 400);
+select *, check_row(row_start, row_end), row_start != 0 as start_nonzero
+from t for system_time all order by x, row_end;
+id	x	check_row(row_start, row_end)	start_nonzero
+1	300	HISTORICAL ROW	1
+1	400	CURRENT ROW	1
+select s from log order by id;
+s
+DEL BEFORE
+1 300 OTHER CURRENT ROW
+DEL AFTER
+1 300 OTHER HISTORICAL ROW
+DEL BEFORE
+1 300 OTHER HISTORICAL ROW
+1 4 CURRENT CURRENT ROW
+DEL AFTER
+1 300 OTHER HISTORICAL ROW
+DEL BEFORE
+1 300 OTHER HISTORICAL ROW
+1 40 CURRENT CURRENT ROW
+DEL AFTER
+1 300 OTHER HISTORICAL ROW
+# ensure that row_start is not duplicated
+select count(row_start) as empty from t for system_time all
+group by row_start having count(row_start) > 1;
+empty
+drop table t;
+# Replace on unversioned field should create history record
+create table t(
+id int primary key,
+x int without system versioning,
+row_start SYS_DATATYPE as row start invisible,
+row_end SYS_DATATYPE as row end invisible,
+period for system_time(row_start, row_end)
+) with system versioning;
+insert t values (1, 2);
+replace t values (1, 3);
+select *, check_row(row_start, row_end), row_start != 0 as start_nonzero
+from t for system_time all order by x;
+id	x	check_row(row_start, row_end)	start_nonzero
+1	2	HISTORICAL ROW	1
+1	3	CURRENT ROW	1
+# ensure that row_start is not duplicated
+select count(row_start) as empty from t for system_time all
+group by row_start having count(row_start) > 1;
+empty
+drop table t;
+drop table log;
diff --git a/mysql-test/suite/versioning/t/replace.test b/mysql-test/suite/versioning/t/replace.test
index 83489f4a4b9..9c4f5b194c9 100644
--- a/mysql-test/suite/versioning/t/replace.test
+++ b/mysql-test/suite/versioning/t/replace.test
@@ -75,6 +75,89 @@ replace into t1 values (1),(2);
 --error ER_DUP_ENTRY
 replace into t1 values (1),(2);
 
+--connection default
+set timestamp=default;
 drop table t1;
 
+
+--echo #
+--echo # MDEV-15990: REPLACE on a precise-versioned table
+--echo # returns duplicate key error (ER_DUP_ENTRY)
+
+--replace_result $sys_datatype_expl SYS_DATATYPE "$KEY_TYPE" KEY_TYPE
+eval create or replace table t(
+  id int,
+  $KEY_TYPE(id),
+  x int,
+  row_start $sys_datatype_expl as row start invisible,
+  row_end $sys_datatype_expl as row end invisible,
+  period for system_time(row_start, row_end)
+) with system versioning;
+
+replace t values (1, 3), (1, 30), (1, 300);
+
+
+if ($MTR_COMBINATION_TIMESTAMP)
+{
+  let $vers_current_value= FROM_UNIXTIME(@@timestamp);
+}
+if ($MTR_COMBINATION_TRX_ID)
+{
+  let $vers_current_value= trx_id from information_schema.innodb_trx
+                           where trx_mysql_thread_id = connection_id();
+}
+
+
+create temporary table log(id int key auto_increment, s char(30));
+--replace_result $vers_current_value VERS_CURRENT_VALUE
+eval create trigger tr1del before delete on t
+  for each row insert log(s) select 'DEL BEFORE' union
+                              select concat(id, ' ', x, ' ',
+                                   if(row_start = (select $vers_current_value),
+                                      "CURRENT", "OTHER"),
+                                   ' ', check_row(row_start, row_end))
+                              from t for system_time all;
+
+--replace_result $vers_current_value VERS_CURRENT_VALUE
+eval create trigger tr2del after delete on t
+  for each row insert log(s) select 'DEL AFTER' union
+                     select concat(id, ' ', x, ' ',
+                                   if(row_start = (select $vers_current_value),
+                                      "CURRENT", "OTHER"),
+                                   ' ', check_row(row_start, row_end))
+                     from t for system_time all;
+replace t values (1, 4), (1, 40), (1, 400);
+
+select *, check_row(row_start, row_end), row_start != 0 as start_nonzero
+  from t for system_time all order by x, row_end;
+select s from log order by id;
+
+--echo # ensure that row_start is not duplicated
+select count(row_start) as empty from t for system_time all
+                                 group by row_start having count(row_start) > 1;
+
+drop table t;
+
+--echo # Replace on unversioned field should create history record
+--replace_result $sys_datatype_expl SYS_DATATYPE
+eval create table t(
+  id int primary key,
+  x int without system versioning,
+  row_start $sys_datatype_expl as row start invisible,
+  row_end $sys_datatype_expl as row end invisible,
+  period for system_time(row_start, row_end)
+) with system versioning;
+
+insert t values (1, 2);
+replace t values (1, 3);
+select *, check_row(row_start, row_end), row_start != 0 as start_nonzero
+  from t for system_time all order by x;
+
+--echo # ensure that row_start is not duplicated
+select count(row_start) as empty from t for system_time all
+                                 group by row_start having count(row_start) > 1;
+
+drop table t;
+drop table log;
+
 --source suite/versioning/common_finish.inc
diff --git a/sql/field.h b/sql/field.h
index 7979ffe2980..c4c653f1415 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -4964,10 +4964,18 @@ ulonglong TABLE::vers_end_id() const
 }
 
 inline
-ulonglong TABLE::vers_start_id() const
+ulonglong TABLE::vers_start_id(const uchar* rec) const
 {
   DBUG_ASSERT(versioned(VERS_TRX_ID));
-  return static_cast<ulonglong>(vers_start_field()->val_int());
+  uchar *ptr= rec == NULL ? vers_start_field()->ptr
+          : const_cast<uchar*>(vers_start_field()->ptr_in_record(rec));
+
+  uchar *old_ptr= vers_start_field()->ptr;
+  vers_start_field()->ptr= ptr;
+  auto ret= static_cast<ulonglong>(vers_start_field()->val_int());
+  vers_start_field()->ptr= old_ptr;
+
+  return ret;
 }
 
 #endif /* FIELD_INCLUDED */
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 95de5d6ad61..9a4cd4a22b4 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -239,14 +239,39 @@ static bool record_should_be_deleted(THD *thd, TABLE *table, SQL_SELECT *sel,
 
 
 inline
-int TABLE::delete_row()
+int TABLE::delete_row(uchar *buf)
 {
-  if (!versioned(VERS_TIMESTAMP) || !vers_end_field()->is_max())
-    return file->ha_delete_row(record[0]);
+  ulong sec_part;
+  Field *row_start;
+  const uchar *pos;
+  if (versioned(VERS_TIMESTAMP))
+  {
+    row_start= vers_start_field();
+    pos= row_start->ptr_in_record(buf);
+  }
 
-  store_record(this, record[1]);
+  if (!versioned(VERS_TIMESTAMP) || !vers_end_field()->is_max() ||
+      (row_start->get_timestamp(pos, &sec_part) == in_use->query_start() &&
+       sec_part == in_use->query_start_sec_part()))
+    return file->ha_delete_row(buf);
+
+  if (buf != record[0])
+  {
+    DBUG_ASSERT(buf == record[1]);
+    store_record(this, record[2]);
+    restore_record(this, record[1]);
+  }
+  else
+  {
+    store_record(this, record[1]);
+  }
   vers_update_end();
+
   int err= file->ha_update_row(record[1], record[0]);
+
+  if (buf != record[0])
+    restore_record(this, record[2]);
+
   /*
      MDEV-23644: we get HA_ERR_FOREIGN_DUPLICATE_KEY iff we already got history
      row with same trx_id which is the result of foreign key action, so we
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 43007b2243a..d84a0be6bc1 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1958,34 +1958,15 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
           For system versioning wa also use path through delete since we would
           save nothing through this cheating.
         */
-        if (last_uniq_key(table,key_nr) &&
+        if (last_uniq_key(table,key_nr) && !table->versioned() &&
             !table->file->referenced_by_foreign_key() &&
             (!table->triggers || !table->triggers->has_delete_triggers()))
         {
-          if (table->versioned(VERS_TRX_ID))
-          {
-            bitmap_set_bit(table->write_set, table->vers_start_field()->field_index);
-            table->file->column_bitmaps_signal();
-            table->vers_start_field()->store(0, false);
-          }
-          if (unlikely(error= table->file->ha_update_row(table->record[1],
-                                                         table->record[0])) &&
-              error != HA_ERR_RECORD_IS_THE_SAME)
+          error= table->file->ha_update_row(table->record[1], table->record[0]);
+          if (unlikely(error && error != HA_ERR_RECORD_IS_THE_SAME))
             goto err;
           if (likely(!error))
-          {
             info->deleted++;
-            if (!table->file->has_transactions())
-              thd->transaction.stmt.modified_non_trans_table= TRUE;
-            if (table->versioned(VERS_TIMESTAMP))
-            {
-              store_record(table, record[2]);
-              error= vers_insert_history_row(table);
-              restore_record(table, record[2]);
-              if (unlikely(error))
-                goto err;
-            }
-          }
           else
             error= 0;   // error was HA_ERR_RECORD_IS_THE_SAME
           /*
@@ -2001,23 +1982,11 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
                                                 TRG_ACTION_BEFORE, TRUE))
             goto before_trg_err;
 
-          if (!table->versioned(VERS_TIMESTAMP))
-            error= table->file->ha_delete_row(table->record[1]);
-          else
-          {
-            store_record(table, record[2]);
-            restore_record(table, record[1]);
-            table->vers_update_end();
-            error= table->file->ha_update_row(table->record[1],
-                                              table->record[0]);
-            restore_record(table, record[2]);
-          }
+          error= table->delete_row(table->record[1]);
           if (unlikely(error))
             goto err;
-          if (!table->versioned(VERS_TIMESTAMP))
-            info->deleted++;
-          else
-            info->updated++;
+
+          info->deleted++;
           if (!table->file->has_transactions())
             thd->transaction.stmt.modified_non_trans_table= TRUE;
           if (table->triggers &&
@@ -2042,7 +2011,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
      */
     if (table->file->insert_id_for_cur_row == 0)
       table->file->insert_id_for_cur_row= insert_id_for_cur_row;
-      
+
     /*
       Restore column maps if they where replaced during an duplicate key
       problem.
diff --git a/sql/table.h b/sql/table.h
index 577c11c37ee..69e97c263d8 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1681,12 +1681,13 @@ public:
     return field[s->row_end_field];
   }
 
-  ulonglong vers_start_id() const;
+  ulonglong vers_start_id(const uchar* rec=NULL) const;
   ulonglong vers_end_id() const;
 
   bool vers_check_update(List<Item> &items);
 
-  int delete_row();
+  inline int delete_row(){ return delete_row(record[0]); }
+  int delete_row(uchar *buf);
   /* Used in majority of DML (called from fill_record()) */
   void vers_update_fields();
   /* Used in DELETE, DUP REPLACE and insert history row */
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index a1f58091a90..0ffdaf70b5d 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -9080,8 +9080,9 @@ ha_innobase::delete_row(
 
 	/* This is a delete */
 	m_prebuilt->upd_node->is_delete = table->versioned_write(VERS_TRX_ID)
-		&& table->vers_end_field()->is_max()
-		&& trx->id != table->vers_start_id()
+		&& table->vers_end_field()->is_max(
+			table->vers_end_field()->ptr_in_record(record))
+		&& trx->id != table->vers_start_id(record)
 		? VERSIONED_DELETE
 		: PLAIN_DELETE;