MDEV-18727 improve DML operation of System Versioning

MDEV-18957 UPDATE with LIMIT clause is wrong for versioned partitioned tables

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

Additional DML cases in view.test

19 files changed, 271 insertions, 87 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 93f240272c1..93f240272c1 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 2163ebeb1a0..315413fbd7d 100644
--- a/mysql-test/suite/versioning/r/partition.result
+++ b/mysql-test/suite/versioning/r/partition.result
@@ -566,3 +566,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 8a7f8e84a76..8a7f8e84a76 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 eca322d9ef4..ce8c2e5ec1a 100644
--- a/mysql-test/suite/versioning/t/partition.test
+++ b/mysql-test/suite/versioning/t/partition.test
@@ -517,4 +517,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 232111d5a98..380fc48e915 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -4538,7 +4538,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 11871155355..dc0641502ce 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -189,7 +189,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 e3d17174b8b..4245c843968 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;
@@ -305,29 +297,7 @@ 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)
-  {
-    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;
 
   if (thd->lex->handle_list_of_derived(table_list, DT_MERGE_FOR_INSERT))
     DBUG_RETURN(TRUE);
@@ -940,16 +910,36 @@ 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);
+  }
+
+/* 10.4:
+  if (table_list->has_period())
+  {
+    if (table_list->is_view_or_derived())
     {
       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))
+
+    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(),
@@ -1238,11 +1228,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 44ea6a18663..44595746614 100644
--- a/sql/sql_derived.cc
+++ b/sql/sql_derived.cc
@@ -689,7 +689,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_parse.cc b/sql/sql_parse.cc
index a8e66d2a230..861d50e8872 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -4710,8 +4710,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,
@@ -4733,6 +4735,7 @@ mysql_execute_command(THD *thd)
           if (lex->describe || lex->analyze_stmt)
             res= thd->lex->explain->send_explain(thd);
         }
+      multi_delete_error:
         delete result;
       }
     }
@@ -9483,7 +9486,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 c7501e4fd1d..46e21822cc4 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -677,6 +677,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)
   {
@@ -709,6 +710,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:
@@ -776,9 +778,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= 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;
@@ -808,7 +823,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);
@@ -834,7 +849,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
@@ -861,10 +876,14 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
       switch (vers_conditions.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, row_end, curr);
+        if (vers_conditions.type == SYSTEM_TIME_UNSPECIFIED)
+          cond1= newx Item_func_eq(thd, row_end, curr);
+        else
+          cond1= newx Item_func_lt(thd, row_end, curr);
         break;
       case SYSTEM_TIME_AS_OF:
         cond1= newx Item_func_le(thd, row_start, point_in_time1);
@@ -896,8 +915,12 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
       switch (vers_conditions.type)
       {
       case SYSTEM_TIME_UNSPECIFIED:
+      case SYSTEM_TIME_HISTORY:
         curr= newx Item_int(thd, ULONGLONG_MAX);
-        cond1= newx Item_func_eq(thd, row_end, curr);
+        if (vers_conditions.type == SYSTEM_TIME_UNSPECIFIED)
+          cond1= newx Item_func_eq(thd, row_end, curr);
+        else
+          cond1= newx Item_func_lt(thd, row_end, curr);
         break;
       case SYSTEM_TIME_AS_OF:
         trx_id0= vers_conditions.start.unit == VERS_TIMESTAMP
@@ -938,7 +961,19 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
     {
       cond1= and_items(thd, cond2, cond1);
       cond1= and_items(thd, cond3, cond1);
-      table->on_expr= and_items(thd, table->on_expr, cond1);
+      if (is_select)
+        table->on_expr= and_items(thd, table->on_expr, cond1);
+      else
+      {
+        if (join)
+        {
+          where= and_items(thd, join->conds, cond1);
+          join->conds= where;
+        }
+        else
+          where= and_items(thd, where, cond1);
+        table->where= and_items(thd, table->where, cond1);
+      }
     }
 
     table->vers_conditions.type= SYSTEM_TIME_ALL;
diff --git a/sql/sql_union.cc b/sql/sql_union.cc
index c119f1e0116..b71b62b35ed 100644
--- a/sql/sql_union.cc
+++ b/sql/sql_union.cc
@@ -976,9 +976,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 32adc4420b3..39d28bfbe50 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -878,11 +878,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)
@@ -1266,6 +1261,21 @@ bool mysql_prepare_update(THD *thd, TABLE_LIST *table_list,
 
   thd->lex->allow_sum_func.clear_all();
 
+/* 10.4:
+  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
     (not multi-) update
@@ -1786,6 +1796,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,
@@ -2345,11 +2358,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 235be9b0f89..bddbec26d67 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -13785,7 +13785,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 278423ec0c2..e008e6a3ded 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -9172,6 +9172,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 bec0fdd10ba..1dda70ae0da 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1863,6 +1863,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;
 
@@ -1870,6 +1871,7 @@ struct vers_select_conds_t
   {
     type= SYSTEM_TIME_UNSPECIFIED;
     used= false;
+    delete_history= false;
     start.empty();
     end.empty();
   }
@@ -1880,6 +1882,8 @@ struct vers_select_conds_t
   {
     type= _type;
     used= false;
+    delete_history= (type == SYSTEM_TIME_HISTORY ||
+      type == SYSTEM_TIME_BEFORE);
     start= _start;
     end= _end;
   }