MDEV-15990 REPLACE on a precise-versioned table returns ER_DUP_ENTRY

We had a protection against it, by allowing versioned delete if:
trx->id != table->vers_start_id()

For replace this check fails: replace calls ha_delete_row(record[2]), but
table->vers_start_id() returns the value from record[0], which is irrelevant.

The same problem hits Field::is_max, which may have checked the wrong record.

Fix:
* Refactor Field::is_max to optionally accept a pointer as an argument.
* Refactor vers_start_id and vers_end_id to always accept a pointer to the
record. there is a difference with is_max is that is_max accepts the pointer to
the
field data, rather than to the record.

Method val_int() would be too effortful to refactor to accept the argument, so
instead the value in record is fetched directly, like it is done in
Field_longlong.

Author: FooBarrior

mysql-test/suite/versioning/r/replace.result

@@ -63,9 +63,11 @@ connection default;
 replace into t1 values (1),(2);
 connection con1;
 replace into t1 values (1),(2);
-ERROR 23000: Duplicate entry '1' for key 'PRIMARY'
 drop table t1;
 #
+# MDEV-14794 Limitations which the row end as a part of PK imposes due to
+# CURRENT_TIMESTAMP behavior and otherwise
+#
 # MDEV-15330 Server crash or assertion `table->insert_values' failure in write_record upon LOAD DATA
 #
 create table t1 (a int, b int, c int, vc int as (c), unique(a), unique(b)) with system versioning;
@@ -88,3 +90,18 @@ Note	1831	Duplicate index `data_2`. This is deprecated and will be disallowed in
 alter table t1 add system versioning;
 replace into t1 values ('o'), ('o');
 drop table t1;
+#
+# MDEV-15990 REPLACE on a precise-versioned table returns duplicate key
+# error (ER_DUP_ENTRY)
+#
+create or replace table t1 (
+pk int primary key,  i int,
+row_start bigint unsigned as row start,
+row_end bigint unsigned as row end,
+period for system_time(row_start, row_end)
+) engine=InnoDB with system versioning;
+replace into t1 (pk,i) values (1,10),(1,100),(1,1000);
+select pk, i, row_end from t1 for system_time all;
+pk	i	row_end
+1	1000	18446744073709551615
+drop table t1;

mysql-test/suite/versioning/t/replace.test

@@ -78,6 +78,7 @@ replace into t1 values (1),(2);
 --connection con1
 --error ER_DUP_ENTRY
 replace into t1 values (1),(2);
+drop table t1;
 
 drop table t1;
 
@@ -116,4 +117,20 @@ alter table t1 add system versioning;
 replace into t1 values ('o'), ('o');
 drop table t1;
 
+--echo #
+--echo # MDEV-15990 REPLACE on a precise-versioned table returns duplicate key
+--echo # error (ER_DUP_ENTRY)
+--echo #
+
+create or replace table t1 (
+    pk int primary key,  i int,
+    row_start bigint unsigned as row start,
+    row_end bigint unsigned as row end,
+    period for system_time(row_start, row_end)
+) engine=InnoDB with system versioning;
+replace into t1 (pk,i) values (1,10),(1,100),(1,1000);
+select pk, i, row_end from t1 for system_time all;
+drop table t1;
+
+
 --source suite/versioning/common_finish.inc

sql/field.cc

@@ -4723,17 +4723,17 @@ void Field_longlong::set_max()
   int8store(ptr, unsigned_flag ? ULONGLONG_MAX : LONGLONG_MAX);
 }
 
-bool Field_longlong::is_max()
+bool Field_longlong::is_max(const uchar *ptr_arg) const
 {
   DBUG_ASSERT(marked_for_read());
   if (unsigned_flag)
   {
     ulonglong j;
-    j= uint8korr(ptr);
+    j= uint8korr(ptr_arg);
     return j == ULONGLONG_MAX;
   }
   longlong j;
-  j= sint8korr(ptr);
+  j= sint8korr(ptr_arg);
   return j == LONGLONG_MAX;
 }
 
@@ -5792,13 +5792,13 @@ void Field_timestampf::set_max()
   DBUG_VOID_RETURN;
 }
 
-bool Field_timestampf::is_max()
+bool Field_timestampf::is_max(const uchar *ptr_arg) const
 {
   DBUG_ENTER("Field_timestampf::is_max");
   DBUG_ASSERT(marked_for_read());
 
-  DBUG_RETURN(mi_sint4korr(ptr) == TIMESTAMP_MAX_VALUE &&
-              mi_sint3korr(ptr + 4) == TIME_MAX_SECOND_PART);
+  DBUG_RETURN(mi_sint4korr(ptr_arg) == TIMESTAMP_MAX_VALUE &&
+              mi_sint3korr(ptr_arg + 4) == TIME_MAX_SECOND_PART);
 }
 
 my_time_t Field_timestampf::get_timestamp(const uchar *pos,

sql/field.h

@@ -803,8 +803,9 @@ class Field: public Value_source
    */
   virtual void set_max()
   { DBUG_ASSERT(0); }
-  virtual bool is_max()
+  virtual bool is_max(const uchar *ptr_arg) const
   { DBUG_ASSERT(0); return false; }
+  bool is_max() const { return is_max(ptr); }
 
   uchar		*ptr;			// Position to field in record
 
@@ -2857,7 +2858,7 @@ class Field_longlong :public Field_int
     return unpack_int64(to, from, from_end);
   }
   void set_max() override;
-  bool is_max() override;
+  bool is_max(const uchar *ptr_arg) const override;
   ulonglong get_max_int_value() const override
   {
     return unsigned_flag ? 0xFFFFFFFFFFFFFFFFULL : 0x7FFFFFFFFFFFFFFFULL;
@@ -3437,7 +3438,7 @@ class Field_timestampf :public Field_timestamp_with_dec {
     return memcmp(a_ptr, b_ptr, pack_length());
   }
   void set_max() override;
-  bool is_max() override;
+  bool is_max(const uchar *ptr_arg) const override;
   my_time_t get_timestamp(const uchar *pos, ulong *sec_part) const override;
   bool val_native(Native *to) override;
   uint size_of() const override { return sizeof *this; }
@@ -5973,17 +5974,21 @@ bool check_expression(Virtual_column_info *vcol, const LEX_CSTRING *name,
 #define f_visibility(x)         (static_cast<field_visibility_t> ((x) & INVISIBLE_MAX_BITS))
 
 inline
-ulonglong TABLE::vers_end_id() const
+ulonglong TABLE::vers_end_id(const uchar *record_arg) const
 {
   DBUG_ASSERT(versioned(VERS_TRX_ID));
-  return static_cast<ulonglong>(vers_end_field()->val_int());
+  DBUG_ASSERT(dynamic_cast<Field_longlong*>(vers_end_field()));
+  const uchar *ptr= vers_end_field()->ptr_in_record(record_arg);
+  return static_cast<ulonglong>(sint8korr(ptr));
 }
 
 inline
-ulonglong TABLE::vers_start_id() const
+ulonglong TABLE::vers_start_id(const uchar *record_arg) const
 {
   DBUG_ASSERT(versioned(VERS_TRX_ID));
-  return static_cast<ulonglong>(vers_start_field()->val_int());
+  DBUG_ASSERT(dynamic_cast<Field_longlong*>(vers_start_field()));
+  const uchar *ptr= vers_start_field()->ptr_in_record(record_arg);
+  return static_cast<ulonglong>(sint8korr(ptr));
 }
 
 double pos_in_interval_for_string(CHARSET_INFO *cset,

sql/handler.cc

@@ -7551,8 +7551,12 @@ int handler::ha_check_overlaps(const uchar *old_data, const uchar* new_data)
   DBUG_ASSERT(this == table->file);
   if (!table_share->period.unique_keys)
     return 0;
-  if (table->versioned() && !table->vers_end_field()->is_max())
-    return 0;
+  if (table->versioned())
+  {
+    Field *end= table->vers_end_field();
+    if (!end->is_max(end->ptr_in_record(new_data)))
+      return 0;
+  }
 
   const bool after_write= ha_table_flags() & HA_CHECK_UNIQUE_AFTER_WRITE;
   const bool is_update= !after_write && old_data;

sql/log_event_server.cc

@@ -8720,8 +8720,6 @@ Update_rows_log_event::do_exec_row(rpl_group_info *rgi)
     return error;
   }
 
-  const bool history_change= m_table->versioned() ?
-    !m_table->vers_end_field()->is_max() : false;
   TABLE_LIST *tl= m_table->pos_in_table_list;
   uint8 trg_event_map_save= tl->trg_event_map;
 
@@ -8772,8 +8770,12 @@ Update_rows_log_event::do_exec_row(rpl_group_info *rgi)
   {
     if (m_vers_from_plain && m_table->versioned(VERS_TIMESTAMP))
       m_table->vers_update_fields();
-    if (!history_change && !m_table->vers_end_field()->is_max())
+    Field *end= m_table->vers_end_field();
+    const uchar *old_ptr= end->ptr_in_record(m_table->record[1]);
+
+    if (end->is_max(old_ptr) && !end->is_max())
     {
+      // This is a versioned delete, and we'll have to invoke ON DELETE actions
       tl->trg_event_map|= trg2bit(TRG_EVENT_DELETE);
     }
   }

sql/sql_delete.cc

@@ -288,7 +288,8 @@ int update_portion_of_time(THD *thd, TABLE *table,
 inline
 int TABLE::delete_row()
 {
-  if (!versioned(VERS_TIMESTAMP) || !vers_end_field()->is_max())
+  if (!versioned(VERS_TIMESTAMP) ||
+      !vers_end_field()->is_max())
     return file->ha_delete_row(record[0]);
 
   store_record(this, record[1]);

sql/sql_insert.cc

@@ -2114,25 +2114,12 @@ int Write_record::replace_row(ha_rows *inserted, ha_rows *deleted)
     if (can_optimize && key_nr == last_unique_key)
     {
       DBUG_PRINT("info", ("Updating row using ha_update_row()"));
-      if (table->versioned(VERS_TRX_ID))
-        table->vers_start_field()->store(0, false);
-
       error= table->file->ha_update_row(table->record[1], table->record[0]);
 
       if (likely(!error))
         ++*deleted;
       else if (error != HA_ERR_RECORD_IS_THE_SAME)
         return on_ha_error(error);
-
-      if (versioned)
-      {
-        store_record(table, record[2]);
-        error= vers_insert_history_row(table);
-        restore_record(table, record[2]);
-        if (unlikely(error))
-          return on_ha_error(error);
-      }
-
       break;
     }
     else

sql/sql_insert.h

@@ -124,7 +124,8 @@ class Write_record
       bool has_delete_triggers= use_triggers &&
                                 table->triggers->has_delete_triggers();
       bool referenced_by_fk= table->file->referenced_by_foreign_key();
-      can_optimize= !referenced_by_fk && !has_delete_triggers;
+      can_optimize= !referenced_by_fk && !has_delete_triggers
+                    && !versioned && !table->versioned();
     }
   }
   Write_record(THD *thd, TABLE *table, COPY_INFO *info,

sql/table.cc

@@ -7668,6 +7668,11 @@ static void do_mark_index_columns(TABLE *table, uint index,
       table->s->primary_key != MAX_KEY && table->s->primary_key != index)
     do_mark_index_columns(table, table->s->primary_key, bitmap, read);
 
+  if (table->versioned(VERS_TRX_ID))
+  {
+    table->vers_start_field()->register_field_in_read_map();
+    table->vers_end_field()->register_field_in_read_map();
+  }
 }
 /*
   mark columns used by key, but don't reset other fields