MDEV-25468 DELETE HISTORY may delete current data on system-versioned table

Item_func_history (is_history()) is a bool function that checks if the
row is the history row by checking row_end->is_max(). The argument to
this function must be row_end system field.

Added the above function to conjunction with SYSTEM_TIME_BEFORE
versioning condition.

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

@@ -129,3 +129,36 @@ select * from t1;
 a
 1
 drop table t1;
+#
+# MDEV-25468 DELETE HISTORY may delete current data on system-versioned table
+#
+create or replace table t1 (x int) with system versioning;
+insert into t1 values (1);
+delete history from t1 before system_time '2039-01-01 23:00';
+select * from t1;
+x
+1
+explain extended delete history from t1 before system_time '2039-01-01 23:00';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	1	100.00	Using where
+create or replace procedure p() delete history from t1 before system_time '2039-01-01 23:00';
+call p;
+select * from t1;
+x
+1
+call p;
+select * from t1;
+x
+1
+drop procedure p;
+prepare stmt from "delete history from t1 before system_time '2039-01-01 23:00'";
+execute stmt;
+select * from t1;
+x
+1
+execute stmt;
+select * from t1;
+x
+1
+drop prepare stmt;
+drop table t1;

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

@@ -141,4 +141,26 @@ insert into t1 values (1);
 select * from t1;
 drop table t1;
 
+--echo #
+--echo # MDEV-25468 DELETE HISTORY may delete current data on system-versioned table
+--echo #
+create or replace table t1 (x int) with system versioning;
+insert into t1 values (1);
+delete history from t1 before system_time '2039-01-01 23:00';
+select * from t1;
+explain extended delete history from t1 before system_time '2039-01-01 23:00';
+create or replace procedure p() delete history from t1 before system_time '2039-01-01 23:00';
+call p;
+select * from t1;
+call p;
+select * from t1;
+drop procedure p;
+prepare stmt from "delete history from t1 before system_time '2039-01-01 23:00'";
+execute stmt;
+select * from t1;
+execute stmt;
+select * from t1;
+drop prepare stmt;
+drop table t1;
+
 --source suite/versioning/common_finish.inc

sql/item_vers.cc

@@ -26,6 +26,22 @@
 #include "tztime.h"
 #include "item.h"
 
+bool Item_func_history::val_bool()
+{
+  Item_field *f= static_cast<Item_field *>(args[0]);
+  DBUG_ASSERT(f->fixed);
+  DBUG_ASSERT(f->field->flags & VERS_SYS_END_FLAG);
+  return !f->field->is_max();
+}
+
+void Item_func_history::print(String *str, enum_query_type query_type)
+{
+  str->append(func_name());
+  str->append('(');
+  args[0]->print(str, query_type);
+  str->append(')');
+}
+
 Item_func_trt_ts::Item_func_trt_ts(THD *thd, Item* a, TR_table::field_id_t _trt_field) :
   Item_datetimefunc(thd, a),
   trt_field(_trt_field)

sql/item_vers.h

@@ -22,6 +22,36 @@
 #pragma interface			/* gcc class implementation */
 #endif
 
+class Item_func_history: public Item_bool_func
+{
+public:
+  /*
+     @param    a  Item_field for row_end system field
+  */
+  Item_func_history(THD *thd, Item *a): Item_bool_func(thd, a)
+  {
+    DBUG_ASSERT(a->type() == Item::FIELD_ITEM);
+  }
+
+  virtual bool val_bool();
+  virtual longlong val_int()
+  {
+    return (val_bool() ? 1 : 0);
+  }
+  bool fix_length_and_dec()
+  {
+    maybe_null= 0;
+    null_value= 0;
+    decimals= 0;
+    max_length= 1;
+    return FALSE;
+  }
+  virtual const char* func_name() const { return "is_history"; }
+  virtual void print(String *str, enum_query_type query_type);
+  Item *get_copy(THD *thd)
+  { return get_item_copy<Item_func_history>(thd, this); }
+};
+
 class Item_func_trt_ts: public Item_datetimefunc
 {
   TR_table::field_id_t trt_field;

sql/sql_select.cc

@@ -928,7 +928,8 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
         cond3= newx Item_func_le(thd, point_in_time1, point_in_time2);
         break;
       case SYSTEM_TIME_BEFORE:
-        cond1= newx Item_func_lt(thd, row_end, point_in_time1);
+        cond1= newx Item_func_history(thd, row_end);
+        cond2= newx Item_func_lt(thd, row_end, point_in_time1);
         break;
       default:
         DBUG_ASSERT(0);
@@ -978,7 +979,8 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
         trx_id0= vers_conditions.start.unit == VERS_TIMESTAMP
           ? newx Item_func_trt_id(thd, point_in_time1, TR_table::FLD_TRX_ID, true)
           : point_in_time1;
-        cond1= newx Item_func_trt_trx_sees(thd, trx_id0, row_end);
+        cond1= newx Item_func_history(thd, row_end);
+        cond2= newx Item_func_trt_trx_sees(thd, trx_id0, row_end);
         break;
       default:
         DBUG_ASSERT(0);