MDEV-18727 improve DML operation of System Versioning (10.4)

UPDATE, DELETE: replace linear search of current/historical records
with vers_setup_conds().

Additional DML cases in view.test

20 files changed, 267 insertions, 107 deletions

diff --git a/mysql-test/suite/versioning/r/delete.result b/mysql-test/suite/versioning/r/delete.result
index 77b7fc80286..26ade83acd7 100644
--- a/mysql-test/suite/versioning/r/delete.result
+++ b/mysql-test/suite/versioning/r/delete.result
@@ -1,3 +1,4 @@
+# Basic + delete from view
 create or replace table t1(
 XNo int unsigned,
 sys_start SYS_DATATYPE as row start invisible,
@@ -44,6 +45,7 @@ XNo_vt1
 5
 drop view vt1;
 drop table t1;
+# Check sys_start, sys_end
 create or replace table t1(
 x int,
 sys_start SYS_DATATYPE as row start invisible,
@@ -59,6 +61,7 @@ select x = 1 as A, sys_start = @sys_start as B, sys_end > sys_start as C from t1
 A	B	C
 1	1	1
 drop table t1;
+# Multi-delete
 create or replace table t1(
 x int,
 y int,
@@ -103,9 +106,6 @@ t2_x_all
 14
 drop table t1;
 drop table t2;
-# Basic + delete from view
-# Check sys_start, sys_end
-# Multi-delete
 # Update + delete
 create or replace table t1 (x int) with system versioning;
 insert into t1 values (1);
diff --git a/mysql-test/suite/versioning/r/truncate.result b/mysql-test/suite/versioning/r/delete_history.result
index 46ef271db3d..46ef271db3d 100644
--- a/mysql-test/suite/versioning/r/truncate.result
+++ b/mysql-test/suite/versioning/r/delete_history.result
diff --git a/mysql-test/suite/versioning/r/partition.result b/mysql-test/suite/versioning/r/partition.result
index 690617551de..76268af9c3c 100644
--- a/mysql-test/suite/versioning/r/partition.result
+++ b/mysql-test/suite/versioning/r/partition.result
@@ -574,3 +574,20 @@ execute immediate 'select * from t1 for update';
 pk
 drop view v1;
 drop tables t, t1, t2, t3, t4;
+#
+# MDEV-18957 UPDATE with LIMIT clause is wrong for versioned partitioned tables
+#
+create or replace table t1 (
+x int,
+a varchar(255)
+) with system versioning partition by system_time (partition p1 history, partition pn current);
+insert into t1 (x) values (1), (2), (3), (4);
+update t1 set a= 'foo' limit 3;
+update t1 set a= 'bar' limit 4;
+select * from t1;
+x	a
+1	bar
+2	bar
+3	bar
+4	bar
+drop table t1;
diff --git a/mysql-test/suite/versioning/r/view.result b/mysql-test/suite/versioning/r/view.result
index 850eba32c0d..8b23e87d6a4 100644
--- a/mysql-test/suite/versioning/r/view.result
+++ b/mysql-test/suite/versioning/r/view.result
@@ -64,13 +64,13 @@ select * from vt1;
 x
 1
 2
-# VIEW with parameters [#151]
+# VIEW with parameters [tempesta-tech/mariadb#151]
 create or replace table t1 (x int) with system versioning;
 create or replace view vt1(c) as select x from t1;
 show create view vt1;
 View	Create View	character_set_client	collation_connection
 vt1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `vt1` AS select `t1`.`x` AS `c` from `t1`	latin1	latin1_swedish_ci
-# VIEW over JOIN of versioned tables [#153]
+# VIEW over JOIN of versioned tables [tempesta-tech/mariadb#153]
 create or replace table t1 (a int) with system versioning;
 create or replace table t2 (b int) with system versioning;
 insert into t1 values (1);
@@ -82,7 +82,7 @@ a	b
 create or replace view vt12 as select * from t1 for system_time as of timestamp ('0-0-0') cross join t2;
 select * from vt12;
 a	b
-# VIEW improvements [#183]
+# VIEW improvements [tempesta-tech/mariadb#183]
 create or replace table t3 (x int);
 create or replace view vt1 as select * from t1, t2, t3;
 show create view vt1;
@@ -96,12 +96,12 @@ create or replace view vt1 as select a, t2.row_end as endo from t3, t1, t2;
 show create view vt1;
 View	Create View	character_set_client	collation_connection
 vt1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `vt1` AS select `t1`.`a` AS `a`,`t2`.`row_end` AS `endo` from ((`t3` join `t1`) join `t2`)	latin1	latin1_swedish_ci
-# VIEW over UNION [#269]
+# VIEW over UNION [tempesta-tech/mariadb#269]
 create or replace view vt1 as select * from t1 union select * from t1;
 select * from vt1;
 a
 1
-# VIEW over UNION with non-versioned [#393]
+# VIEW over UNION with non-versioned [tempesta-tech/mariadb#393]
 create or replace table t2 (a int);
 create or replace view vt1 as select * from t1 union select * from t2;
 select * from vt1;
@@ -123,10 +123,10 @@ drop tables t1, t2;
 #
 # MDEV-15146 SQLError[4122]: View is not system versioned
 #
-create table t1 (a int) with system versioning;
+create or replace table t1 (a int) with system versioning;
 insert t1 values (1),(2);
 set @a=now(6);
-create view v1 as select * from t1;
+create or replace view v1 as select * from t1;
 delete from t1;
 select * from v1;
 a
@@ -149,3 +149,67 @@ View	Create View	character_set_client	collation_connection
 v1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v1` AS select `t1`.`i` AS `i` from `t1` FOR SYSTEM_TIME AS OF current_timestamp() - interval 6 second	latin1	latin1_swedish_ci
 drop view v1, vt1, vt12;
 drop tables t1, t3;
+#
+# MDEV-18727 improve DML operation of System Versioning
+#
+create or replace table t1 (
+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;
+insert into t1 values (1), (2);
+create or replace view v1 as select * from t1 where x > 1;
+update v1 set x= x + 1;
+select *, check_row(row_start, row_end) from t1 for system_time all order by x;
+x	check_row(row_start, row_end)
+1	CURRENT ROW
+2	HISTORICAL ROW
+3	CURRENT ROW
+insert v1 values (4);
+select *, check_row(row_start, row_end) from t1 for system_time all order by x;
+x	check_row(row_start, row_end)
+1	CURRENT ROW
+2	HISTORICAL ROW
+3	CURRENT ROW
+4	CURRENT ROW
+delete from v1 where x < 4;
+select *, check_row(row_start, row_end) from t1 for system_time all order by x;
+x	check_row(row_start, row_end)
+1	CURRENT ROW
+2	HISTORICAL ROW
+3	HISTORICAL ROW
+4	CURRENT ROW
+# multi-update
+create or replace table t2 like t1;
+insert into t2 values (1), (2);
+create or replace view v2 as select * from t2 where x > 1;
+update v1, v2 set v1.x= v1.x + 1, v2.x= v2.x + 1 where v1.x = v2.x + 2;
+select *, check_row(row_start, row_end) from t1 for system_time all order by x;
+x	check_row(row_start, row_end)
+1	CURRENT ROW
+2	HISTORICAL ROW
+3	HISTORICAL ROW
+4	HISTORICAL ROW
+5	CURRENT ROW
+select *, check_row(row_start, row_end) from t2 for system_time all order by x;
+x	check_row(row_start, row_end)
+1	CURRENT ROW
+2	HISTORICAL ROW
+3	CURRENT ROW
+# multi-delete
+delete v1, v2 from v1 join v2 where v1.x = v2.x + 2;
+select *, check_row(row_start, row_end) from t1 for system_time all order by x;
+x	check_row(row_start, row_end)
+1	CURRENT ROW
+2	HISTORICAL ROW
+3	HISTORICAL ROW
+4	HISTORICAL ROW
+5	HISTORICAL ROW
+select *, check_row(row_start, row_end) from t2 for system_time all order by x;
+x	check_row(row_start, row_end)
+1	CURRENT ROW
+2	HISTORICAL ROW
+3	HISTORICAL ROW
+drop view v1, v2;
+drop tables t1, t2;
diff --git a/mysql-test/suite/versioning/t/delete.test b/mysql-test/suite/versioning/t/delete.test
index b9045898bb0..4f1ba4b1d8e 100644
--- a/mysql-test/suite/versioning/t/delete.test
+++ b/mysql-test/suite/versioning/t/delete.test
@@ -1,6 +1,7 @@
 source suite/versioning/engines.inc;
 source suite/versioning/common.inc;
 
+--echo # Basic + delete from view
 replace_result $sys_datatype_expl SYS_DATATYPE;
 eval create or replace table t1(
     XNo int unsigned,
@@ -31,7 +32,7 @@ select XNo as XNo_vt1 from vt1;
 drop view vt1;
 drop table t1;
 
-
+--echo # Check sys_start, sys_end
 replace_result $sys_datatype_expl SYS_DATATYPE;
 eval create or replace table t1(
     x int,
@@ -47,6 +48,7 @@ select * from t1;
 select x = 1 as A, sys_start = @sys_start as B, sys_end > sys_start as C from t1 for system_time all;
 drop table t1;
 
+--echo # Multi-delete
 replace_result $sys_datatype_expl SYS_DATATYPE;
 eval create or replace table t1(
     x int,
@@ -69,12 +71,6 @@ select x as t2_x_all from t2 for system_time all;
 drop table t1;
 drop table t2;
 
---echo # Basic + delete from view
-
---echo # Check sys_start, sys_end
-
---echo # Multi-delete
-
 --echo # Update + delete
 create or replace table t1 (x int) with system versioning;
 insert into t1 values (1);
diff --git a/mysql-test/suite/versioning/t/truncate.test b/mysql-test/suite/versioning/t/delete_history.test
index 07fb4eb187a..07fb4eb187a 100644
--- a/mysql-test/suite/versioning/t/truncate.test
+++ b/mysql-test/suite/versioning/t/delete_history.test
diff --git a/mysql-test/suite/versioning/t/partition.test b/mysql-test/suite/versioning/t/partition.test
index e3acf53686e..bf6dd18118e 100644
--- a/mysql-test/suite/versioning/t/partition.test
+++ b/mysql-test/suite/versioning/t/partition.test
@@ -508,4 +508,18 @@ execute immediate 'select * from t1 for update';
 drop view v1;
 drop tables t, t1, t2, t3, t4;
 
+--echo #
+--echo # MDEV-18957 UPDATE with LIMIT clause is wrong for versioned partitioned tables
+--echo #
+create or replace table t1 (
+  x int,
+  a varchar(255)
+) with system versioning partition by system_time (partition p1 history, partition pn current);
+
+insert into t1 (x) values (1), (2), (3), (4);
+update t1 set a= 'foo' limit 3;
+update t1 set a= 'bar' limit 4;
+select * from t1;
+drop table t1;
+
 --source suite/versioning/common_finish.inc
diff --git a/mysql-test/suite/versioning/t/view.test b/mysql-test/suite/versioning/t/view.test
index 5a03a50f1d3..c05fbfd3866 100644
--- a/mysql-test/suite/versioning/t/view.test
+++ b/mysql-test/suite/versioning/t/view.test
@@ -52,13 +52,13 @@ prepare stmt from @tmp; execute stmt; drop prepare stmt;
 
 select * from vt1;
 
---echo # VIEW with parameters [#151]
+--echo # VIEW with parameters [tempesta-tech/mariadb#151]
 create or replace table t1 (x int) with system versioning;
 create or replace view vt1(c) as select x from t1;
 --replace_result 18446744073709551615 MAX_RESULT "TIMESTAMP'2038-01-19 03:14:07.999999'" MAX_RESULT
 show create view vt1;
 
---echo # VIEW over JOIN of versioned tables [#153]
+--echo # VIEW over JOIN of versioned tables [tempesta-tech/mariadb#153]
 create or replace table t1 (a int) with system versioning;
 create or replace table t2 (b int) with system versioning;
 insert into t1 values (1);
@@ -68,7 +68,7 @@ select * from vt12;
 create or replace view vt12 as select * from t1 for system_time as of timestamp ('0-0-0') cross join t2;
 select * from vt12;
 
---echo # VIEW improvements [#183]
+--echo # VIEW improvements [tempesta-tech/mariadb#183]
 create or replace table t3 (x int);
 create or replace view vt1 as select * from t1, t2, t3;
 --replace_result 18446744073709551615 MAX_RESULT "TIMESTAMP'2038-01-19 03:14:07.999999'" MAX_RESULT
@@ -80,11 +80,11 @@ create or replace view vt1 as select a, t2.row_end as endo from t3, t1, t2;
 --replace_result 18446744073709551615 MAX_RESULT "TIMESTAMP'2038-01-19 03:14:07.999999'" MAX_RESULT
 show create view vt1;
 
---echo # VIEW over UNION [#269]
+--echo # VIEW over UNION [tempesta-tech/mariadb#269]
 create or replace view vt1 as select * from t1 union select * from t1;
 select * from vt1;
 
---echo # VIEW over UNION with non-versioned [#393]
+--echo # VIEW over UNION with non-versioned [tempesta-tech/mariadb#393]
 create or replace table t2 (a int);
 create or replace view vt1 as select * from t1 union select * from t2;
 select * from vt1;
@@ -104,10 +104,10 @@ drop tables t1, t2;
 --echo #
 --echo # MDEV-15146 SQLError[4122]: View is not system versioned
 --echo #
-create table t1 (a int) with system versioning;
+create or replace table t1 (a int) with system versioning;
 insert t1 values (1),(2);
 set @a=now(6);
-create view v1 as select * from t1;
+create or replace view v1 as select * from t1;
 delete from t1;
 select * from v1;
 select * from v1 for system_time as of @a;
@@ -124,4 +124,37 @@ show create view v1;
 drop view v1, vt1, vt12;
 drop tables t1, t3;
 
+--echo #
+--echo # MDEV-18727 improve DML operation of System Versioning
+--echo #
+--replace_result $sys_datatype_expl SYS_DATATYPE
+eval create or replace table t1 (
+  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;
+insert into t1 values (1), (2);
+create or replace view v1 as select * from t1 where x > 1;
+update v1 set x= x + 1;
+select *, check_row(row_start, row_end) from t1 for system_time all order by x;
+insert v1 values (4);
+select *, check_row(row_start, row_end) from t1 for system_time all order by x;
+delete from v1 where x < 4;
+select *, check_row(row_start, row_end) from t1 for system_time all order by x;
+--echo # multi-update
+create or replace table t2 like t1;
+insert into t2 values (1), (2);
+create or replace view v2 as select * from t2 where x > 1;
+update v1, v2 set v1.x= v1.x + 1, v2.x= v2.x + 1 where v1.x = v2.x + 2;
+select *, check_row(row_start, row_end) from t1 for system_time all order by x;
+select *, check_row(row_start, row_end) from t2 for system_time all order by x;
+--echo # multi-delete
+delete v1, v2 from v1 join v2 where v1.x = v2.x + 2;
+select *, check_row(row_start, row_end) from t1 for system_time all order by x;
+select *, check_row(row_start, row_end) from t2 for system_time all order by x;
+
+drop view v1, v2;
+drop tables t1, t2;
+
 --source suite/versioning/common_finish.inc
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 9ca6408007e..8dd808065ce 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -4540,7 +4540,7 @@ int ha_partition::delete_row(const uchar *buf)
     or last historical partition, but DELETE HISTORY can delete from any
     historical partition. So, skip the check in this case.
   */
-  if (!thd->lex->vers_conditions.is_set()) // if not DELETE HISTORY
+  if (!thd->lex->vers_conditions.delete_history)
   {
     uint32 part_id;
     error= get_part_for_buf(buf, m_rec0, m_part_info, &part_id);
diff --git a/sql/mysqld.h b/sql/mysqld.h
index ad19ce13db8..44db6c6216b 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -187,7 +187,8 @@ enum vers_system_time_t
   SYSTEM_TIME_AS_OF,
   SYSTEM_TIME_FROM_TO,
   SYSTEM_TIME_BETWEEN,
-  SYSTEM_TIME_BEFORE,
+  SYSTEM_TIME_BEFORE,  // used for DELETE HISTORY ... BEFORE
+  SYSTEM_TIME_HISTORY, // used for DELETE HISTORY
   SYSTEM_TIME_ALL
 };
 
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index c4bc32262e8..622b24292f3 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -225,19 +225,11 @@ bool Update_plan::save_explain_data_intern(MEM_ROOT *mem_root,
 static bool record_should_be_deleted(THD *thd, TABLE *table, SQL_SELECT *sel,
                                      Explain_delete *explain, bool truncate_history)
 {
-  bool check_delete= true;
-
-  if (table->versioned())
-  {
-    bool historical= !table->vers_end_field()->is_max();
-    check_delete= truncate_history ? historical : !historical;
-  }
-
   explain->tracker.on_record_read();
   thd->inc_examined_row_count(1);
   if (table->vfield)
     (void) table->update_virtual_fields(table->file, VCOL_UPDATE_FOR_DELETE);
-  if (check_delete && (!sel || sel->skip_record(thd) > 0))
+  if (!sel || sel->skip_record(thd) > 0)
   {
     explain->tracker.on_record_after_where();
     return true;
@@ -351,30 +343,8 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
 
   THD_STAGE_INFO(thd, stage_init_update);
 
-  bool delete_history= table_list->vers_conditions.is_set();
-  if (delete_history)
-  {
-    DBUG_ASSERT(!table_list->period_conditions.is_set());
-
-    if (table_list->is_view_or_derived())
-    {
-      my_error(ER_IT_IS_A_VIEW, MYF(0), table_list->table_name.str);
-      DBUG_RETURN(true);
-    }
-
-    DBUG_ASSERT(table_list->table);
-    DBUG_ASSERT(!conds || thd->stmt_arena->is_stmt_execute());
-
-    // conds could be cached from previous SP call
-    if (!conds)
-    {
-      if (select_lex->vers_setup_conds(thd, table_list))
-        DBUG_RETURN(TRUE);
-
-      conds= table_list->on_expr;
-      table_list->on_expr= NULL;
-    }
-  }
+  const bool delete_history= table_list->vers_conditions.delete_history;
+  DBUG_ASSERT(!(delete_history && table_list->period_conditions.is_set()));
 
   if (thd->lex->handle_list_of_derived(table_list, DT_MERGE_FOR_INSERT))
     DBUG_RETURN(TRUE);
@@ -1036,15 +1006,11 @@ int mysql_prepare_delete(THD *thd, TABLE_LIST *table_list,
                                     select_lex->leaf_tables, FALSE, 
                                     DELETE_ACL, SELECT_ACL, TRUE))
     DBUG_RETURN(TRUE);
-  if (table_list->vers_conditions.is_set())
+
+  if (table_list->vers_conditions.is_set() && table_list->is_view_or_derived())
   {
-    if (table_list->is_view())
-    {
-      my_error(ER_IT_IS_A_VIEW, MYF(0), table_list->table_name.str);
-      DBUG_RETURN(true);
-    }
-    if (select_lex->vers_setup_conds(thd, table_list))
-      DBUG_RETURN(true);
+    my_error(ER_IT_IS_A_VIEW, MYF(0), table_list->table_name.str);
+    DBUG_RETURN(true);
   }
 
   if (table_list->has_period())
@@ -1055,11 +1021,19 @@ int mysql_prepare_delete(THD *thd, TABLE_LIST *table_list,
       DBUG_RETURN(true);
     }
 
-    *conds= select_lex->period_setup_conds(thd, table_list, *conds);
-    if (!*conds)
+    if (select_lex->period_setup_conds(thd, table_list))
       DBUG_RETURN(true);
   }
 
+  DBUG_ASSERT(table_list->table);
+  // conds could be cached from previous SP call
+  DBUG_ASSERT(!table_list->vers_conditions.is_set() ||
+              !*conds || thd->stmt_arena->is_stmt_execute());
+  if (select_lex->vers_setup_conds(thd, table_list))
+    DBUG_RETURN(TRUE);
+
+  *conds= select_lex->where;
+
   if ((wild_num && setup_wild(thd, table_list, field_list, NULL, wild_num,
                               &select_lex->hidden_bit_fields)) ||
       setup_fields(thd, Ref_ptr_array(),
@@ -1356,11 +1330,6 @@ int multi_delete::send_data(List<Item> &values)
     if (table->status & (STATUS_NULL_ROW | STATUS_DELETED))
       continue;
 
-    if (table->versioned() && !table->vers_end_field()->is_max())
-    {
-      continue;
-    }
-
     table->file->position(table->record[0]);
     found++;
 
diff --git a/sql/sql_derived.cc b/sql/sql_derived.cc
index 06e82263524..06a3e8a108f 100644
--- a/sql/sql_derived.cc
+++ b/sql/sql_derived.cc
@@ -723,7 +723,17 @@ bool mysql_derived_prepare(THD *thd, LEX *lex, TABLE_LIST *derived)
        !(derived->is_multitable() &&
          (thd->lex->sql_command == SQLCOM_UPDATE_MULTI ||
           thd->lex->sql_command == SQLCOM_DELETE_MULTI))))
+  {
+    /*
+       System versioned tables may still require to get versioning conditions
+       (when updating view). See vers_setup_conds().
+    */
+    if (!unit->prepared &&
+        derived->table->versioned() &&
+        (res= unit->prepare(derived, derived->derived_result, 0)))
+      goto exit;
     DBUG_RETURN(FALSE);
+  }
 
   /* prevent name resolving out of derived table */
   for (SELECT_LEX *sl= first_select; sl; sl= sl->next_select())
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 159c05fb98f..9d22eb1a8cc 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -1291,7 +1291,7 @@ public:
   /* push new Item_field into item_list */
   bool vers_push_field(THD *thd, TABLE_LIST *table, const LEX_CSTRING field_name);
 
-  Item* period_setup_conds(THD *thd, TABLE_LIST *table, Item *where);
+  int period_setup_conds(THD *thd, TABLE_LIST *table);
   void init_query();
   void init_select();
   st_select_lex_unit* master_unit() { return (st_select_lex_unit*) master; }
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 516813a0264..d89cae39330 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -4778,8 +4778,10 @@ mysql_execute_command(THD *thd)
     {
       result= new (thd->mem_root) multi_delete(thd, aux_tables,
                                                lex->table_count);
-      if (unlikely(result))
+      if (likely(result))
       {
+        if (unlikely(select_lex->vers_setup_conds(thd, aux_tables)))
+          goto multi_delete_error;
         res= mysql_select(thd,
                           select_lex->get_table_list(),
                           select_lex->with_wild,
@@ -4801,6 +4803,7 @@ mysql_execute_command(THD *thd)
           if (lex->describe || lex->analyze_stmt)
             res= thd->lex->explain->send_explain(thd);
         }
+      multi_delete_error:
         delete result;
       }
     }
@@ -9595,7 +9598,7 @@ bool update_precheck(THD *thd, TABLE_LIST *tables)
 bool delete_precheck(THD *thd, TABLE_LIST *tables)
 {
   DBUG_ENTER("delete_precheck");
-  if (tables->vers_conditions.is_set())
+  if (tables->vers_conditions.delete_history)
   {
     if (check_one_table_access(thd, DELETE_HISTORY_ACL, tables))
       DBUG_RETURN(TRUE);
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 1ee2a175579..9022252c34c 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -715,6 +715,7 @@ bool vers_select_conds_t::init_from_sysvar(THD *thd)
 {
   vers_asof_timestamp_t &in= thd->variables.vers_asof_timestamp;
   type= (vers_system_time_t) in.type;
+  delete_history= false;
   start.unit= VERS_TIMESTAMP;
   if (type != SYSTEM_TIME_UNSPECIFIED && type != SYSTEM_TIME_ALL)
   {
@@ -747,6 +748,7 @@ void vers_select_conds_t::print(String *str, enum_query_type query_type) const
     end.print(str, query_type, STRING_WITH_LEN(" AND "));
     break;
   case SYSTEM_TIME_BEFORE:
+  case SYSTEM_TIME_HISTORY:
     DBUG_ASSERT(0);
     break;
   case SYSTEM_TIME_ALL:
@@ -782,10 +784,15 @@ Item* period_get_condition(THD *thd, TABLE_LIST *table, SELECT_LEX *select,
     switch (conds->type)
     {
     case SYSTEM_TIME_UNSPECIFIED:
+    case SYSTEM_TIME_HISTORY:
       thd->variables.time_zone->gmt_sec_to_TIME(&max_time, TIMESTAMP_MAX_VALUE);
       max_time.second_part= TIME_MAX_SECOND_PART;
       curr= newx Item_datetime_literal(thd, &max_time, TIME_SECOND_PART_DIGITS);
-      cond1= newx Item_func_eq(thd, conds->field_end, curr);
+      if (conds->type == SYSTEM_TIME_UNSPECIFIED)
+        cond1= newx Item_func_eq(thd, conds->field_end, curr);
+      else
+        cond1= newx Item_func_lt(thd, conds->field_end, curr);
+      break;
       break;
     case SYSTEM_TIME_AS_OF:
       cond1= newx Item_func_le(thd, conds->field_start, conds->start.item);
@@ -829,8 +836,13 @@ Item* period_get_condition(THD *thd, TABLE_LIST *table, SELECT_LEX *select,
     switch (conds->type)
     {
     case SYSTEM_TIME_UNSPECIFIED:
+    case SYSTEM_TIME_HISTORY:
       curr= newx Item_int(thd, ULONGLONG_MAX);
-      cond1= newx Item_func_eq(thd, conds->field_end, curr);
+      if (conds->type == SYSTEM_TIME_UNSPECIFIED)
+        cond1= newx Item_func_eq(thd, conds->field_end, curr);
+      else
+        cond1= newx Item_func_lt(thd, conds->field_end, curr);
+      break;
       DBUG_ASSERT(!conds->start.item);
       DBUG_ASSERT(!conds->end.item);
       break;
@@ -873,12 +885,12 @@ bool skip_setup_conds(THD *thd)
          || thd->lex->is_view_context_analysis();
 }
 
-Item* SELECT_LEX::period_setup_conds(THD *thd, TABLE_LIST *tables, Item *where)
+int SELECT_LEX::period_setup_conds(THD *thd, TABLE_LIST *tables)
 {
   DBUG_ENTER("SELECT_LEX::period_setup_conds");
 
   if (skip_setup_conds(thd))
-    DBUG_RETURN(where);
+    DBUG_RETURN(0);
 
   Query_arena backup;
   Query_arena *arena= thd->activate_stmt_arena_if_needed(&backup);
@@ -896,19 +908,19 @@ Item* SELECT_LEX::period_setup_conds(THD *thd, TABLE_LIST *tables, Item *where)
       my_error(ER_PERIOD_NOT_FOUND, MYF(0), conds.name.str);
       if (arena)
         thd->restore_active_arena(arena, &backup);
-      DBUG_RETURN(NULL);
+      DBUG_RETURN(-1);
     }
 
     conds.period= &table->table->s->period;
     result= and_items(thd, result,
                       period_get_condition(thd, table, this, &conds, true));
   }
-  result= and_items(thd, where, result);
+  where= and_items(thd, where, result);
 
   if (arena)
     thd->restore_active_arena(arena, &backup);
 
-  DBUG_RETURN(result);
+  DBUG_RETURN(0);
 }
 
 int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
@@ -960,9 +972,22 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
     }
   }
 
+  bool is_select= false;
+  switch (thd->lex->sql_command)
+  {
+  case SQLCOM_SELECT:
+  case SQLCOM_INSERT_SELECT:
+  case SQLCOM_REPLACE_SELECT:
+  case SQLCOM_DELETE_MULTI:
+  case SQLCOM_UPDATE_MULTI:
+    is_select= true;
+  default:
+    break;
+  }
+
   for (TABLE_LIST *table= tables; table; table= table->next_local)
   {
-    if (!table->table || !table->table->versioned())
+    if (!table->table || table->is_view() || !table->table->versioned())
       continue;
 
     vers_select_conds_t &vers_conditions= table->vers_conditions;
@@ -992,7 +1017,7 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
     }
 
     // propagate system_time from sysvar
-    if (!vers_conditions.is_set())
+    if (!vers_conditions.is_set() && is_select)
     {
       if (vers_conditions.init_from_sysvar(thd))
         DBUG_RETURN(-1);
@@ -1008,7 +1033,7 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
 
     bool timestamps_only= table->table->versioned(VERS_TIMESTAMP);
 
-    if (vers_conditions.is_set())
+    if (vers_conditions.is_set() && vers_conditions.type != SYSTEM_TIME_HISTORY)
     {
       thd->where= "FOR SYSTEM_TIME";
       /* TODO: do resolve fix_length_and_dec(), fix_fields(). This requires
@@ -1028,10 +1053,21 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
     vers_conditions.period = &table->table->s->vers;
     Item *cond= period_get_condition(thd, table, this, &vers_conditions,
                                      timestamps_only);
-    if (cond)
+    if (is_select)
       table->on_expr= and_items(thd, table->on_expr, cond);
-    table->vers_conditions.type= SYSTEM_TIME_ALL;
+    else
+    {
+      if (join)
+      {
+        where= and_items(thd, join->conds, cond);
+        join->conds= where;
+      }
+      else
+        where= and_items(thd, where, cond);
+      table->where= and_items(thd, table->where, cond);
+    }
 
+    table->vers_conditions.type= SYSTEM_TIME_ALL;
   } // for (table= tables; ...)
 
   DBUG_RETURN(0);
diff --git a/sql/sql_union.cc b/sql/sql_union.cc
index da25fa774d0..5ff88e02f5d 100644
--- a/sql/sql_union.cc
+++ b/sql/sql_union.cc
@@ -977,9 +977,21 @@ bool st_select_lex_unit::prepare(TABLE_LIST *derived_arg,
       if (sl->tvc->prepare(thd, sl, tmp_result, this))
 	goto err;
     }
-    else if (prepare_join(thd, first_sl, tmp_result, additional_options,
+    else
+    {
+      if (prepare_join(thd, first_sl, tmp_result, additional_options,
                      is_union_select))
-      goto err;
+        goto err;
+
+      if (derived_arg && derived_arg->table &&
+          derived_arg->derived_type == VIEW_ALGORITHM_MERGE &&
+          derived_arg->table->versioned())
+      {
+        /* Got versioning conditions (see vers_setup_conds()), need to update
+           derived_arg. */
+        derived_arg->where= first_sl->where;
+      }
+    }
     types= first_sl->item_list;
     goto cont;
   }
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index f5bb298fdba..590a4f8bae1 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -960,11 +960,6 @@ update_begin:
   THD_STAGE_INFO(thd, stage_updating);
   while (!(error=info.read_record()) && !thd->killed)
   {
-    if (table->versioned() && !table->vers_end_field()->is_max())
-    {
-      continue;
-    }
-
     explain->tracker.on_record_read();
     thd->inc_examined_row_count(1);
     if (!select || select->skip_record(thd) > 0)
@@ -1373,12 +1368,18 @@ bool mysql_prepare_update(THD *thd, TABLE_LIST *table_list,
 
   thd->lex->allow_sum_func.clear_all();
 
-  if (table_list->has_period())
-  {
-    *conds= select_lex->period_setup_conds(thd, table_list, *conds);
-    if (!*conds)
+  if (table_list->has_period() &&
+      select_lex->period_setup_conds(thd, table_list))
       DBUG_RETURN(true);
-  }
+
+  DBUG_ASSERT(table_list->table);
+  // conds could be cached from previous SP call
+  DBUG_ASSERT(!table_list->vers_conditions.is_set() ||
+              !*conds || thd->stmt_arena->is_stmt_execute());
+  if (select_lex->vers_setup_conds(thd, table_list))
+    DBUG_RETURN(TRUE);
+
+  *conds= select_lex->where;
 
   /*
     We do not call DT_MERGE_FOR_INSERT because it has no sense for simple
@@ -1900,6 +1901,9 @@ bool mysql_multi_update(THD *thd, TABLE_LIST *table_list, List<Item> *fields,
   thd->abort_on_warning= !ignore && thd->is_strict_mode();
   List<Item> total_list;
 
+  if (select_lex->vers_setup_conds(thd, table_list))
+    DBUG_RETURN(1);
+
   res= mysql_select(thd,
                     table_list, select_lex->with_wild, total_list, conds,
                     select_lex->order_list.elements,
@@ -2459,11 +2463,6 @@ int multi_update::send_data(List<Item> &not_used_values)
     if (table->status & (STATUS_NULL_ROW | STATUS_UPDATED))
       continue;
 
-    if (table->versioned() && !table->vers_end_field()->is_max())
-    {
-      continue;
-    }
-
     if (table == table_to_update)
     {
       /*
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 3fa2c85f160..476196f38ea 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -13866,7 +13866,7 @@ delete:
 opt_delete_system_time:
             /* empty */
           {
-            Lex->vers_conditions.init(SYSTEM_TIME_ALL);
+            Lex->vers_conditions.init(SYSTEM_TIME_HISTORY);
           }
           | BEFORE_SYM SYSTEM_TIME_SYM history_point
           {
diff --git a/sql/table.cc b/sql/table.cc
index 0a7d2c9547a..055f02ef267 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -9511,6 +9511,8 @@ bool vers_select_conds_t::eq(const vers_select_conds_t &conds) const
     return true;
   case SYSTEM_TIME_BEFORE:
     break;
+  case SYSTEM_TIME_HISTORY:
+    break;
   case SYSTEM_TIME_AS_OF:
     return start.eq(conds.start);
   case SYSTEM_TIME_FROM_TO:
diff --git a/sql/table.h b/sql/table.h
index 700c5f036fd..ae46b192854 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1930,6 +1930,7 @@ struct vers_select_conds_t
 {
   vers_system_time_t type;
   bool used:1;
+  bool delete_history:1;
   Vers_history_point start;
   Vers_history_point end;
   Lex_ident name;
@@ -1943,6 +1944,7 @@ struct vers_select_conds_t
   {
     type= SYSTEM_TIME_UNSPECIFIED;
     used= false;
+    delete_history= false;
     start.empty();
     end.empty();
   }
@@ -1954,6 +1956,8 @@ struct vers_select_conds_t
   {
     type= _type;
     used= false;
+    delete_history= (type == SYSTEM_TIME_HISTORY ||
+      type == SYSTEM_TIME_BEFORE);
     start= _start;
     end= _end;
     name= _name;