Merge 10.4 into 10.5

36 files changed, 504 insertions, 232 deletions

diff --git a/mysql-test/main/invisible_field.result b/mysql-test/main/invisible_field.result
index a57298dd5c5..9b42b043ec6 100644
--- a/mysql-test/main/invisible_field.result
+++ b/mysql-test/main/invisible_field.result
@@ -621,3 +621,9 @@ a	b
 1	3
 2	4
 drop table t1;
+#
+# MDEV-23467 SIGSEGV in fill_record/fill_record_n_invoke_before_triggers on INSERT DELAYED
+#
+create table t1 (a int, b int invisible);
+insert delayed into t1 values (1);
+drop table t1;
diff --git a/mysql-test/main/invisible_field.test b/mysql-test/main/invisible_field.test
index 0e3994a78ce..7a48347ec29 100644
--- a/mysql-test/main/invisible_field.test
+++ b/mysql-test/main/invisible_field.test
@@ -271,3 +271,11 @@ select a,b from t1;
 
 #cleanup
 drop table t1;
+
+--echo #
+--echo # MDEV-23467 SIGSEGV in fill_record/fill_record_n_invoke_before_triggers on INSERT DELAYED
+--echo #
+create table t1 (a int, b int invisible);
+insert delayed into t1 values (1);
+# cleanup
+drop table t1;
diff --git a/mysql-test/main/subselect4.result b/mysql-test/main/subselect4.result
index 944515413d3..3f12322f6dc 100644
--- a/mysql-test/main/subselect4.result
+++ b/mysql-test/main/subselect4.result
@@ -2686,6 +2686,37 @@ SELECT * FROM t2;
 f
 bar
 DROP TABLE t1, t2;
+#
+# MDEV-18335: Assertion `!error || error == 137' failed in subselect_rowid_merge_engine::init
+#
+CREATE TABLE t1 (i1 int,v1 varchar(1),KEY (v1,i1));
+INSERT INTO t1 VALUES
+(9,'y'),(4,'q'),(0,null),(0,'p'),(null,'j');
+CREATE TABLE t2 (pk int);
+INSERT INTO t2 VALUES (1),(2);
+CREATE TABLE t3 (v2 varchar(1));
+INSERT INTO t3 VALUES
+('p'),('j'),('y'),('q');
+CREATE TABLE t4 (v2 varchar(1));
+INSERT INTO t4 VALUES
+('a'),('a'),('b'),('b'),('c'),('c'),
+('d'),('d'),('e'),('e'),('f'),('f'),
+('g'),('g'),('h'),('h'),('i'),('i'),
+('j'),('j'),('k'),('k'),('l'),('l'),
+('m'),('m'),('n'),('n'),('o'),('o'),
+('p'),('p'),('q'),('q'),('r'),('r'),
+('s'),('s'),('t'),('t'),('u'),('u'),('v'),('v'),
+('w'),('w'),('x'),('x'), (NULL),(NULL);
+SET @save_join_cache_level=@@join_cache_level;
+SET join_cache_level=0;
+select 1
+from t2 join t1 on
+('i','w') not in (select t1.v1,t4.v2 from t4,t1,t3 where t3.v2 = t1.v1) LIMIT ROWS EXAMINED 500;
+1
+Warnings:
+Warning	1931	Query execution was interrupted. The query examined at least 3020 rows, which exceeds LIMIT ROWS EXAMINED (500). The query result may be incomplete
+SET join_cache_level= @save_join_cache_level;
+DROP TABLE t1,t2,t3,t4;
 # End of 10.2 tests
 #
 # MDEV-19134: EXISTS() slower if ORDER BY is defined
diff --git a/mysql-test/main/subselect4.test b/mysql-test/main/subselect4.test
index 8d2ddaab09b..c9b05c4f015 100644
--- a/mysql-test/main/subselect4.test
+++ b/mysql-test/main/subselect4.test
@@ -2201,6 +2201,42 @@ SELECT * FROM t2;
 
 DROP TABLE t1, t2;
 
+--echo #
+--echo # MDEV-18335: Assertion `!error || error == 137' failed in subselect_rowid_merge_engine::init
+--echo #
+
+CREATE TABLE t1 (i1 int,v1 varchar(1),KEY (v1,i1));
+INSERT INTO t1 VALUES
+(9,'y'),(4,'q'),(0,null),(0,'p'),(null,'j');
+
+CREATE TABLE t2 (pk int);
+INSERT INTO t2 VALUES (1),(2);
+
+CREATE TABLE t3 (v2 varchar(1));
+INSERT INTO t3 VALUES
+('p'),('j'),('y'),('q');
+
+CREATE TABLE t4 (v2 varchar(1));
+INSERT INTO t4 VALUES
+('a'),('a'),('b'),('b'),('c'),('c'),
+('d'),('d'),('e'),('e'),('f'),('f'),
+('g'),('g'),('h'),('h'),('i'),('i'),
+('j'),('j'),('k'),('k'),('l'),('l'),
+('m'),('m'),('n'),('n'),('o'),('o'),
+('p'),('p'),('q'),('q'),('r'),('r'),
+('s'),('s'),('t'),('t'),('u'),('u'),('v'),('v'),
+('w'),('w'),('x'),('x'), (NULL),(NULL);
+
+SET @save_join_cache_level=@@join_cache_level;
+SET join_cache_level=0;
+
+select 1
+from t2 join t1 on
+('i','w') not in (select t1.v1,t4.v2 from t4,t1,t3 where t3.v2 = t1.v1) LIMIT ROWS EXAMINED 500;
+SET join_cache_level= @save_join_cache_level;
+
+DROP TABLE t1,t2,t3,t4;
+
 --echo # End of 10.2 tests
 
 --echo #
diff --git a/mysql-test/suite/galera/r/MDEV-22055.result b/mysql-test/suite/galera/r/MDEV-22055.result
new file mode 100644
index 00000000000..651f8501a0a
--- /dev/null
+++ b/mysql-test/suite/galera/r/MDEV-22055.result
@@ -0,0 +1,18 @@
+connection node_2;
+connection node_1;
+ROLLBACK AND CHAIN;
+CREATE TABLE t1(a int not null primary key) engine=innodb;
+INSERT INTO t1 values (1);
+BEGIN;
+INSERT INTO t1 values (2);
+ROLLBACK AND CHAIN;
+SELECT * FROM t1;
+a
+1
+connection node_2;
+SET SESSION wsrep_sync_wait=15;
+SELECT * FROM t1;
+a
+1
+connection node_1;
+DROP TABLE t1;
diff --git a/mysql-test/suite/galera/t/MDEV-22055.test b/mysql-test/suite/galera/t/MDEV-22055.test
new file mode 100644
index 00000000000..ae29c456bf0
--- /dev/null
+++ b/mysql-test/suite/galera/t/MDEV-22055.test
@@ -0,0 +1,19 @@
+--source include/galera_cluster.inc
+
+ROLLBACK AND CHAIN;
+
+CREATE TABLE t1(a int not null primary key) engine=innodb;
+INSERT INTO t1 values (1);
+
+BEGIN;
+INSERT INTO t1 values (2);
+ROLLBACK AND CHAIN;
+
+SELECT * FROM t1;
+
+--connection node_2
+SET SESSION wsrep_sync_wait=15;
+SELECT * FROM t1;
+
+--connection node_1
+DROP TABLE t1;
diff --git a/mysql-test/suite/galera/t/mdev-22543.test b/mysql-test/suite/galera/t/mdev-22543.test
index 53662e36942..1e7d3712639 100644
--- a/mysql-test/suite/galera/t/mdev-22543.test
+++ b/mysql-test/suite/galera/t/mdev-22543.test
@@ -5,15 +5,15 @@
 --source include/galera_cluster.inc
 --source include/have_debug.inc
 --source include/have_debug_sync.inc
- 
+
 --let $node_1 = node_1
 --let $node_2 = node_2
 --source include/auto_increment_offset_save.inc
- 
+
 --let $galera_connection_name = node_1_ctrl
 --let $galera_server_number = 1
 --source include/galera_connect.inc
- 
+
 #
 # Run UPDATE on node_1 and make it block before table locks are taken.
 # This should block FTWRL.
@@ -23,10 +23,10 @@ CREATE TABLE t1 (f1 INT PRIMARY KEY, f2 INT);
 INSERT INTO t1 VALUES (1, 1);
 SET DEBUG_SYNC = "before_lock_tables_takes_lock SIGNAL sync_point_reached WAIT_FOR sync_point_continue";
 --send UPDATE t1 SET f2 = 2 WHERE f1 = 1
- 
+
 --connection node_1_ctrl
 SET DEBUG_SYNC = "now WAIT_FOR sync_point_reached";
- 
+
 #
 # Restart node_2, force SST.
 #
@@ -40,19 +40,19 @@ SET DEBUG_SYNC = "now WAIT_FOR sync_point_reached";
 # If the bug is present, FTWRL times out on node_1 in couple of
 # seconds and node_2 fails to join.
 --sleep 10
- 
+
 --connection node_1_ctrl
 SET DEBUG_SYNC = "now SIGNAL sync_point_continue";
- 
+
 --connection node_1
 --reap
 SET DEBUG_SYNC = "RESET";
- 
+
 --connection node_2
 --enable_reconnect
 --source include/wait_until_connected_again.inc
- 
+
 --connection node_1
 DROP TABLE t1;
- 
+
 --source include/auto_increment_offset_restore.inc
diff --git a/mysql-test/suite/galera_3nodes/disabled.def b/mysql-test/suite/galera_3nodes/disabled.def
index 1683485981b..415da407cf7 100644
--- a/mysql-test/suite/galera_3nodes/disabled.def
+++ b/mysql-test/suite/galera_3nodes/disabled.def
@@ -1,2 +1,21 @@
+##############################################################################
+#
+#  List the test cases that are to be disabled temporarily.
+#
+#  Separate the test case name and the comment with ':'.
+#
+#    <testcasename> : MDEV-<xxxx> <comment>
+# 
+#  Do not use any TAB characters for whitespace.
+#
+##############################################################################
+
+galera_ipv6_mariabackup : MDEV-23573 Could not open '../galera/include/have_mariabackup.inc'
+galera_ipv6_mariabackup_section : MDEV-23574 Could not open '../galera/include/have_mariabackup.inc'
+galera_ipv6_mysqldump : MDEV-23576 WSREP_SST: [ERROR] rsync daemon port '16008' has been taken
+galera_ipv6_rsyn : MDEV-23581 WSREP_SST: [ERROR] rsync daemon port '16008' has been taken
+galera_ipv6_rsync_section : MDEV-23580 WSREP_SST: [ERROR] rsync daemon port '16008' has been taken
+galera_ipv6_xtrabackup-v2 : MDEV-23575 WSREP_SST: [ERROR] innobackupex not in path
+galera_ist_gcache_rollover : MDEV-23578 WSREP: exception caused by message: {v=0,t=1,ut=255,o=4,s=0,sr=0,as=1,f=6,src=50524cfe,srcvid=view_id(REG,50524cfe,4),insvid=view_id(UNKNOWN,00000000,0),ru=00000000,r=[-1,-1],fs=75,nl=(}
 galera_slave_options_do :MDEV-8798
 galera_slave_options_ignore : MDEV-8798
diff --git a/mysql-test/suite/innodb/r/foreign_key.result b/mysql-test/suite/innodb/r/foreign_key.result
index 854b53b3083..936c7bab54f 100644
--- a/mysql-test/suite/innodb/r/foreign_key.result
+++ b/mysql-test/suite/innodb/r/foreign_key.result
@@ -168,7 +168,6 @@ PRIMARY KEY  (store_id),
 UNIQUE KEY idx_unique_manager (manager_staff_id),
 CONSTRAINT fk_store_staff FOREIGN KEY (manager_staff_id) REFERENCES staff (staff_id) ON DELETE RESTRICT ON UPDATE CASCADE
 ) ENGINE=InnoDB;
-SET FOREIGN_KEY_CHECKS=DEFAULT;
 LOCK TABLE staff WRITE;
 UNLOCK TABLES;
 DROP TABLES staff, store;
diff --git a/mysql-test/suite/innodb/t/foreign_key.test b/mysql-test/suite/innodb/t/foreign_key.test
index c595cf92690..3d4c7c5004d 100644
--- a/mysql-test/suite/innodb/t/foreign_key.test
+++ b/mysql-test/suite/innodb/t/foreign_key.test
@@ -136,7 +136,6 @@ CREATE TABLE store (
   UNIQUE KEY idx_unique_manager (manager_staff_id),
   CONSTRAINT fk_store_staff FOREIGN KEY (manager_staff_id) REFERENCES staff (staff_id) ON DELETE RESTRICT ON UPDATE CASCADE
 ) ENGINE=InnoDB;
-SET FOREIGN_KEY_CHECKS=DEFAULT;
 
 LOCK TABLE staff WRITE;
 UNLOCK TABLES;
diff --git a/mysql-test/suite/sys_vars/r/foreign_key_checks_basic.result b/mysql-test/suite/sys_vars/r/foreign_key_checks_basic.result
index 834d693edb8..0bf9d859d40 100644
--- a/mysql-test/suite/sys_vars/r/foreign_key_checks_basic.result
+++ b/mysql-test/suite/sys_vars/r/foreign_key_checks_basic.result
@@ -7,7 +7,7 @@ SET @@session.foreign_key_checks = 1;
 SET @@session.foreign_key_checks = DEFAULT;
 SELECT @@session.foreign_key_checks;
 @@session.foreign_key_checks
-0
+1
 '#---------------------FN_DYNVARS_032_02-------------------------#'
 SET foreign_key_checks = 1;
 SELECT @@foreign_key_checks;
diff --git a/mysql-test/suite/sys_vars/r/unique_checks_basic.result b/mysql-test/suite/sys_vars/r/unique_checks_basic.result
index 60cf2795309..83fb9edb4cd 100644
--- a/mysql-test/suite/sys_vars/r/unique_checks_basic.result
+++ b/mysql-test/suite/sys_vars/r/unique_checks_basic.result
@@ -7,7 +7,7 @@ SET @@session.unique_checks= 1;
 SET @@session.unique_checks= DEFAULT;
 SELECT @@session.unique_checks;
 @@session.unique_checks
-0
+1
 '#--------------------FN_DYNVARS_005_04-------------------------#'
 SET @@session.unique_checks =1;
 SELECT @@session.unique_checks;
diff --git a/mysql-test/suite/versioning/r/sysvars.result b/mysql-test/suite/versioning/r/sysvars.result
index ac0a1237902..66513741631 100644
--- a/mysql-test/suite/versioning/r/sysvars.result
+++ b/mysql-test/suite/versioning/r/sysvars.result
@@ -186,4 +186,13 @@ SELECT @@global.system_versioning_asof;
 @@global.system_versioning_asof
 2002-01-01 00:00:00.000000
 SET @@global.system_versioning_asof= DEFAULT;
+#
+# MDEV-23562 Assertion `time_type == MYSQL_TIMESTAMP_DATETIME' failed upon SELECT from versioned table
+#
+CREATE TABLE t1 (a INT) WITH SYSTEM VERSIONING;
+SET system_versioning_asof= DATE(NOW());
+SELECT * FROM t1;
+a
+DROP TABLE t1;
+SET system_versioning_asof= DEFAULT;
 # End of 10.4 tests
diff --git a/mysql-test/suite/versioning/t/sysvars.test b/mysql-test/suite/versioning/t/sysvars.test
index 7c5e818ec81..a1026418e98 100644
--- a/mysql-test/suite/versioning/t/sysvars.test
+++ b/mysql-test/suite/versioning/t/sysvars.test
@@ -138,4 +138,14 @@ SET @@global.system_versioning_asof= timestamp'2001-12-31 23:59:59.9999999';
 SELECT @@global.system_versioning_asof;
 SET @@global.system_versioning_asof= DEFAULT;
 
+--echo #
+--echo # MDEV-23562 Assertion `time_type == MYSQL_TIMESTAMP_DATETIME' failed upon SELECT from versioned table
+--echo #
+
+CREATE TABLE t1 (a INT) WITH SYSTEM VERSIONING;
+SET system_versioning_asof= DATE(NOW());
+SELECT * FROM t1;
+DROP TABLE t1;
+SET system_versioning_asof= DEFAULT;
+
 --echo # End of 10.4 tests
diff --git a/sql/field.cc b/sql/field.cc
index 491cd1b5079..f8d9cbf93ce 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -5892,9 +5892,7 @@ Item *Field_temporal::get_equal_const_item_datetime(THD *thd,
         See comments about truncation in the same place in
         Field_time::get_equal_const_item().
       */
-      return new (thd->mem_root) Item_datetime_literal(thd,
-                                                       dt.get_mysql_time(),
-                                                       decimals());
+      return new (thd->mem_root) Item_datetime_literal(thd, &dt, decimals());
     }
     break;
   case ANY_SUBST:
@@ -5906,7 +5904,7 @@ Item *Field_temporal::get_equal_const_item_datetime(THD *thd,
       if (!dt.is_valid_datetime())
         return NULL;
       return new (thd->mem_root)
-        Item_datetime_literal_for_invalid_dates(thd, dt.get_mysql_time(),
+        Item_datetime_literal_for_invalid_dates(thd, &dt,
                                                 dt.get_mysql_time()->
                                                 second_part ?
                                                 TIME_SECOND_PART_DIGITS : 0);
@@ -6246,7 +6244,7 @@ Item *Field_time::get_equal_const_item(THD *thd, const Context &ctx,
 
         (assuming CURRENT_DATE is '2015-08-30'
       */
-      return new (thd->mem_root) Item_time_literal(thd, tm.get_mysql_time(),
+      return new (thd->mem_root) Item_time_literal(thd, &tm,
                                                    tm.get_mysql_time()->
                                                    second_part ?
                                                    TIME_SECOND_PART_DIGITS :
@@ -6275,8 +6273,7 @@ Item *Field_time::get_equal_const_item(THD *thd, const Context &ctx,
               decimals());
       if (!tm.is_valid_time())
         return NULL;
-      return new (thd->mem_root) Item_time_literal(thd, tm.get_mysql_time(),
-                                                   decimals());
+      return new (thd->mem_root) Item_time_literal(thd, &tm, decimals());
     }
     break;
   }
@@ -6842,12 +6839,12 @@ Item *Field_newdate::get_equal_const_item(THD *thd, const Context &ctx,
       */
       if (!dt.hhmmssff_is_zero())
         return new (thd->mem_root)
-          Item_datetime_literal_for_invalid_dates(thd, dt.get_mysql_time(),
+          Item_datetime_literal_for_invalid_dates(thd, &dt,
                                                   dt.get_mysql_time()->
                                                     second_part ?
                                                   TIME_SECOND_PART_DIGITS : 0);
-      return new (thd->mem_root)
-        Item_date_literal_for_invalid_dates(thd, Date(&dt).get_mysql_time());
+      Date d(&dt);
+      return new (thd->mem_root) Item_date_literal_for_invalid_dates(thd, &d);
     }
     break;
   case IDENTITY_SUBST:
@@ -6862,8 +6859,8 @@ Item *Field_newdate::get_equal_const_item(THD *thd, const Context &ctx,
       Datetime dt(thd, const_item, Datetime::Options(TIME_CONV_NONE, thd));
       if (!dt.is_valid_datetime())
         return NULL;
-      return new (thd->mem_root)
-        Item_date_literal(thd, Date(&dt).get_mysql_time());
+      Date d(&dt);
+      return new (thd->mem_root) Item_date_literal(thd, &d);
     }
     break;
   }
diff --git a/sql/ha_partition.h b/sql/ha_partition.h
index 4556c5f482e..8d0557f4ae4 100644
--- a/sql/ha_partition.h
+++ b/sql/ha_partition.h
@@ -1619,9 +1619,8 @@ public:
     return h;
   }
 
-  ha_rows part_records(void *_part_elem)
+  ha_rows part_records(partition_element *part_elem)
   {
-    partition_element *part_elem= reinterpret_cast<partition_element *>(_part_elem);
     DBUG_ASSERT(m_part_info);
     uint32 sub_factor= m_part_info->num_subparts ? m_part_info->num_subparts : 1;
     uint32 part_id= part_elem->id * sub_factor;
diff --git a/sql/item.cc b/sql/item.cc
index 733b5d56b1b..2d1fd24a87e 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -7067,7 +7067,7 @@ void Item_date_literal::print(String *str, enum_query_type query_type)
 {
   str->append("DATE'");
   char buf[MAX_DATE_STRING_REP_LENGTH];
-  my_date_to_str(&cached_time, buf);
+  my_date_to_str(cached_time.get_mysql_time(), buf);
   str->append(buf);
   str->append('\'');
 }
@@ -7082,7 +7082,7 @@ Item *Item_date_literal::clone_item(THD *thd)
 bool Item_date_literal::get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate)
 {
   fuzzydate |= sql_mode_for_dates(thd);
-  *ltime= cached_time;
+  cached_time.copy_to_mysql_time(ltime);
   return (null_value= check_date_with_warn(thd, ltime, fuzzydate,
                                            MYSQL_TIMESTAMP_ERROR));
 }
@@ -7092,7 +7092,7 @@ void Item_datetime_literal::print(String *str, enum_query_type query_type)
 {
   str->append("TIMESTAMP'");
   char buf[MAX_DATE_STRING_REP_LENGTH];
-  my_datetime_to_str(&cached_time, buf, decimals);
+  my_datetime_to_str(cached_time.get_mysql_time(), buf, decimals);
   str->append(buf);
   str->append('\'');
 }
@@ -7107,7 +7107,7 @@ Item *Item_datetime_literal::clone_item(THD *thd)
 bool Item_datetime_literal::get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate)
 {
   fuzzydate |= sql_mode_for_dates(thd);
-  *ltime= cached_time;
+  cached_time.copy_to_mysql_time(ltime);
   return (null_value= check_date_with_warn(thd, ltime, fuzzydate,
                                            MYSQL_TIMESTAMP_ERROR));
 }
@@ -7117,7 +7117,7 @@ void Item_time_literal::print(String *str, enum_query_type query_type)
 {
   str->append("TIME'");
   char buf[MAX_DATE_STRING_REP_LENGTH];
-  my_time_to_str(&cached_time, buf, decimals);
+  my_time_to_str(cached_time.get_mysql_time(), buf, decimals);
   str->append(buf);
   str->append('\'');
 }
@@ -7131,7 +7131,7 @@ Item *Item_time_literal::clone_item(THD *thd)
 
 bool Item_time_literal::get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate)
 {
-  *ltime= cached_time;
+  cached_time.copy_to_mysql_time(ltime);
   if (fuzzydate & TIME_TIME_ONLY)
     return (null_value= false);
   return (null_value= check_date_with_warn(thd, ltime, fuzzydate,
@@ -9990,23 +9990,20 @@ Item *Item_cache_temporal::convert_to_basic_const_item(THD *thd)
 
 Item *Item_cache_datetime::make_literal(THD *thd)
 {
-  MYSQL_TIME ltime;
-  unpack_time(val_datetime_packed(thd), &ltime, MYSQL_TIMESTAMP_DATETIME);
-  return new (thd->mem_root) Item_datetime_literal(thd, &ltime, decimals);
+  Datetime dt(thd, this, TIME_CONV_NONE | TIME_FRAC_NONE);
+  return new (thd->mem_root) Item_datetime_literal(thd, &dt, decimals);
 }
 
 Item *Item_cache_date::make_literal(THD *thd)
 {
-  MYSQL_TIME ltime;
-  unpack_time(val_datetime_packed(thd), &ltime, MYSQL_TIMESTAMP_DATE);
-  return new (thd->mem_root) Item_date_literal(thd, &ltime);
+  Date d(thd, this, TIME_CONV_NONE | TIME_FRAC_NONE);
+  return new (thd->mem_root) Item_date_literal(thd, &d);
 }
 
 Item *Item_cache_time::make_literal(THD *thd)
 {
-  MYSQL_TIME ltime;
-  unpack_time(val_time_packed(thd), &ltime, MYSQL_TIMESTAMP_TIME);
-  return new (thd->mem_root) Item_time_literal(thd, &ltime, decimals);
+  Time t(thd, this);
+  return new (thd->mem_root) Item_time_literal(thd, &t, decimals);
 }
 
 
diff --git a/sql/item.h b/sql/item.h
index e872335bb77..34ceae62c8e 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -4820,29 +4820,20 @@ public:
 
 class Item_temporal_literal :public Item_literal
 {
-protected:
-  MYSQL_TIME cached_time;
 public:
-  /**
-    Constructor for Item_date_literal.
-    @param ltime  DATE value.
-  */
-  Item_temporal_literal(THD *thd, const MYSQL_TIME *ltime)
+  Item_temporal_literal(THD *thd)
    :Item_literal(thd)
   {
     collation= DTCollation_numeric();
     decimals= 0;
-    cached_time= *ltime;
   }
-  Item_temporal_literal(THD *thd, const MYSQL_TIME *ltime, uint dec_arg):
+  Item_temporal_literal(THD *thd, uint dec_arg):
     Item_literal(thd)
   {
     collation= DTCollation_numeric();
     decimals= dec_arg;
-    cached_time= *ltime;
   }
 
-  const MYSQL_TIME *const_ptr_mysql_time() const { return &cached_time; }
   int save_in_field(Field *field, bool no_conversions)
   { return save_date_in_field(field, no_conversions); }
 };
@@ -4853,27 +4844,62 @@ public:
 */
 class Item_date_literal: public Item_temporal_literal
 {
+protected:
+  Date cached_time;
+  bool update_null()
+  {
+    return maybe_null &&
+           (null_value= cached_time.check_date_with_warn(current_thd));
+  }
 public:
-  Item_date_literal(THD *thd, const MYSQL_TIME *ltime)
-    :Item_temporal_literal(thd, ltime)
+  Item_date_literal(THD *thd, const Date *ltime)
+    :Item_temporal_literal(thd),
+     cached_time(*ltime)
   {
+    DBUG_ASSERT(cached_time.is_valid_date());
     max_length= MAX_DATE_WIDTH;
     /*
       If date has zero month or day, it can return NULL in case of
       NO_ZERO_DATE or NO_ZERO_IN_DATE.
-      We can't just check the current sql_mode here in constructor,
+      If date is `February 30`, it can return NULL in case if
+      no ALLOW_INVALID_DATES is set.
+      We can't set null_value using the current sql_mode here in constructor,
       because sql_mode can change in case of prepared statements
       between PREPARE and EXECUTE.
+      Here we only set maybe_null to true if the value has such anomalies.
+      Later (during execution time), if maybe_null is true, then the value
+      will be checked per row, according to the execution time sql_mode.
+      The check_date() below call should cover all cases mentioned.
     */
-    maybe_null= !ltime->month || !ltime->day;
+    maybe_null= cached_time.check_date(TIME_NO_ZERO_DATE | TIME_NO_ZERO_IN_DATE);
   }
   const Type_handler *type_handler() const { return &type_handler_newdate; }
   void print(String *str, enum_query_type query_type);
+  const MYSQL_TIME *const_ptr_mysql_time() const
+  {
+    return cached_time.get_mysql_time();
+  }
   Item *clone_item(THD *thd);
-  longlong val_int() { return Date(this).to_longlong(); }
-  double val_real() { return Date(this).to_double(); }
-  String *val_str(String *to) { return Date(this).to_string(to); }
-  my_decimal *val_decimal(my_decimal *to) { return Date(this).to_decimal(to); }
+  longlong val_int()
+  {
+    return update_null() ? 0 : cached_time.to_longlong();
+  }
+  double val_real()
+  {
+    return update_null() ? 0 : cached_time.to_double();
+  }
+  String *val_str(String *to)
+  {
+    return update_null() ? 0 : cached_time.to_string(to);
+  }
+  my_decimal *val_decimal(my_decimal *to)
+  {
+    return update_null() ? 0 : cached_time.to_decimal(to);
+  }
+  longlong val_datetime_packed(THD *thd)
+  {
+    return update_null() ? 0 : cached_time.valid_date_to_packed();
+  }
   bool get_date(THD *thd, MYSQL_TIME *res, date_mode_t fuzzydate);
   Item *get_copy(THD *thd)
   { return get_item_copy<Item_date_literal>(thd, this); }
@@ -4885,19 +4911,31 @@ public:
 */
 class Item_time_literal: public Item_temporal_literal
 {
+protected:
+  Time cached_time;
 public:
-  Item_time_literal(THD *thd, const MYSQL_TIME *ltime, uint dec_arg):
-    Item_temporal_literal(thd, ltime, dec_arg)
+  Item_time_literal(THD *thd, const Time *ltime, uint dec_arg):
+    Item_temporal_literal(thd, dec_arg),
+    cached_time(*ltime)
   {
+    DBUG_ASSERT(cached_time.is_valid_time());
     max_length= MIN_TIME_WIDTH + (decimals ? decimals + 1 : 0);
   }
   const Type_handler *type_handler() const { return &type_handler_time2; }
   void print(String *str, enum_query_type query_type);
+  const MYSQL_TIME *const_ptr_mysql_time() const
+  {
+    return cached_time.get_mysql_time();
+  }
   Item *clone_item(THD *thd);
-  longlong val_int() { return Time(this).to_longlong(); }
-  double val_real() { return Time(this).to_double(); }
-  String *val_str(String *to) { return Time(this).to_string(to, decimals); }
-  my_decimal *val_decimal(my_decimal *to) { return Time(this).to_decimal(to); }
+  longlong val_int() { return cached_time.to_longlong(); }
+  double val_real() { return cached_time.to_double(); }
+  String *val_str(String *to) { return cached_time.to_string(to, decimals); }
+  my_decimal *val_decimal(my_decimal *to) { return cached_time.to_decimal(to); }
+  longlong val_time_packed(THD *thd)
+  {
+    return cached_time.valid_time_to_packed();
+  }
   bool get_date(THD *thd, MYSQL_TIME *res, date_mode_t fuzzydate);
   bool val_native(THD *thd, Native *to)
   {
@@ -4913,26 +4951,49 @@ public:
 */
 class Item_datetime_literal: public Item_temporal_literal
 {
+protected:
+  Datetime cached_time;
+  bool update_null()
+  {
+    return maybe_null &&
+           (null_value= cached_time.check_date_with_warn(current_thd));
+  }
 public:
-  Item_datetime_literal(THD *thd, const MYSQL_TIME *ltime, uint dec_arg):
-    Item_temporal_literal(thd, ltime, dec_arg)
+  Item_datetime_literal(THD *thd, const Datetime *ltime, uint dec_arg):
+    Item_temporal_literal(thd, dec_arg),
+    cached_time(*ltime)
   {
+    DBUG_ASSERT(cached_time.is_valid_datetime());
     max_length= MAX_DATETIME_WIDTH + (decimals ? decimals + 1 : 0);
     // See the comment on maybe_null in Item_date_literal
-    maybe_null= !ltime->month || !ltime->day;
+    maybe_null= cached_time.check_date(TIME_NO_ZERO_DATE | TIME_NO_ZERO_IN_DATE);
   }
   const Type_handler *type_handler() const { return &type_handler_datetime2; }
   void print(String *str, enum_query_type query_type);
+  const MYSQL_TIME *const_ptr_mysql_time() const
+  {
+    return cached_time.get_mysql_time();
+  }
   Item *clone_item(THD *thd);
-  longlong val_int() { return Datetime(this).to_longlong(); }
-  double val_real() { return Datetime(this).to_double(); }
+  longlong val_int()
+  {
+    return update_null() ? 0 : cached_time.to_longlong();
+  }
+  double val_real()
+  {
+    return update_null() ? 0 : cached_time.to_double();
+  }
   String *val_str(String *to)
   {
-    return Datetime(this).to_string(to, decimals);
+    return update_null() ? NULL : cached_time.to_string(to, decimals);
   }
   my_decimal *val_decimal(my_decimal *to)
   {
-    return Datetime(this).to_decimal(to);
+    return update_null() ? NULL : cached_time.to_decimal(to);
+  }
+  longlong val_datetime_packed(THD *thd)
+  {
+    return update_null() ? 0 : cached_time.valid_datetime_to_packed();
   }
   bool get_date(THD *thd, MYSQL_TIME *res, date_mode_t fuzzydate);
   Item *get_copy(THD *thd)
@@ -4969,11 +5030,14 @@ class Item_date_literal_for_invalid_dates: public Item_date_literal
     in sql_mode=TRADITIONAL.
   */
 public:
-  Item_date_literal_for_invalid_dates(THD *thd, const MYSQL_TIME *ltime)
-   :Item_date_literal(thd, ltime) { }
+  Item_date_literal_for_invalid_dates(THD *thd, const Date *ltime)
+   :Item_date_literal(thd, ltime)
+  {
+    maybe_null= false;
+  }
   bool get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate)
   {
-    *ltime= cached_time;
+    cached_time.copy_to_mysql_time(ltime);
     return (null_value= false);
   }
 };
@@ -4987,11 +5051,14 @@ class Item_datetime_literal_for_invalid_dates: public Item_datetime_literal
 {
 public:
   Item_datetime_literal_for_invalid_dates(THD *thd,
-                                          const MYSQL_TIME *ltime, uint dec_arg)
-   :Item_datetime_literal(thd, ltime, dec_arg) { }
+                                          const Datetime *ltime, uint dec_arg)
+   :Item_datetime_literal(thd, ltime, dec_arg)
+  {
+    maybe_null= false;
+  }
   bool get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate)
   {
-    *ltime= cached_time;
+    cached_time.copy_to_mysql_time(ltime);
     return (null_value= false);
   }
 };
diff --git a/sql/item_subselect.cc b/sql/item_subselect.cc
index 9525019888d..334fec2e048 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -6361,6 +6361,9 @@ subselect_rowid_merge_engine::init(MY_BITMAP *non_null_key_parts,
   while (TRUE)
   {
     error= tmp_table->file->ha_rnd_next(tmp_table->record[0]);
+
+    if (error == HA_ERR_ABORTED_BY_USER)
+      break;
     /*
       This is a temp table that we fully own, there should be no other
       cause to stop the iteration than EOF.
diff --git a/sql/mdl.cc b/sql/mdl.cc
index 240ef97b1c4..4772dc017f9 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -25,9 +25,6 @@
 #include <mysql/plugin.h>
 #include <mysql/service_thd_wait.h>
 #include <mysql/psi/mysql_stage.h>
-#ifdef WITH_WSREP
-#include "wsrep_sst.h"
-#endif
 #include <tpool.h>
 #include <pfs_metadata_provider.h>
 #include <mysql/psi/mysql_mdl.h>
@@ -2335,26 +2332,18 @@ MDL_context::acquire_lock(MDL_request *mdl_request, double lock_wait_timeout)
     wait_status= m_wait.timed_wait(m_owner, &abs_shortwait, FALSE,
                                    mdl_request->key.get_wait_state_name());
 
-    THD* thd= m_owner->get_thd();
-
     if (wait_status != MDL_wait::EMPTY)
       break;
     /* Check if the client is gone while we were waiting. */
-    if (! thd_is_connected(thd))
+    if (! thd_is_connected(m_owner->get_thd()))
     {
-#if defined(WITH_WSREP) && !defined(EMBEDDED_LIBRARY)
-      // During SST client might not be connected
-      if (!wsrep_is_sst_progress())
-#endif
-      {
-        /*
-         * The client is disconnected. Don't wait forever:
-         * assume it's the same as a wait timeout, this
-         * ensures all error handling is correct.
-         */
-        wait_status= MDL_wait::TIMEOUT;
-        break;
-      }
+      /*
+       * The client is disconnected. Don't wait forever:
+       * assume it's the same as a wait timeout, this
+       * ensures all error handling is correct.
+       */
+      wait_status= MDL_wait::TIMEOUT;
+      break;
     }
 
     mysql_prlock_wrlock(&lock->m_rwlock);
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index d25410292ef..e1df56c6d79 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -2683,6 +2683,7 @@ TABLE *Delayed_insert::get_local_table(THD* client_thd)
     if (!(*field= (*org_field)->make_new_field(client_thd->mem_root, copy, 1)))
       goto error;
     (*field)->unireg_check= (*org_field)->unireg_check;
+    (*field)->invisible= (*org_field)->invisible;
     (*field)->orig_table= copy;			// Remove connection
     (*field)->move_field_offset(adjust_ptrs);	// Point at copy->record[0]
     (*field)->flags|= ((*org_field)->flags & LONG_UNIQUE_HASH_FIELD);
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index a4ef786776b..ac0500f0e04 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -729,8 +729,9 @@ bool vers_select_conds_t::init_from_sysvar(THD *thd)
   if (type != SYSTEM_TIME_UNSPECIFIED && type != SYSTEM_TIME_ALL)
   {
     DBUG_ASSERT(type == SYSTEM_TIME_AS_OF);
+    Datetime dt(&in.ltime);
     start.item= new (thd->mem_root)
-        Item_datetime_literal(thd, &in.ltime, TIME_SECOND_PART_DIGITS);
+        Item_datetime_literal(thd, &dt, TIME_SECOND_PART_DIGITS);
     if (!start.item)
       return true;
   }
@@ -794,15 +795,17 @@ Item* period_get_condition(THD *thd, TABLE_LIST *table, SELECT_LEX *select,
     {
     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);
+      Datetime dt(&max_time);
+      curr= newx Item_datetime_literal(thd, &dt, TIME_SECOND_PART_DIGITS);
       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);
       cond2= newx Item_func_gt(thd, conds->field_end, conds->start.item);
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 452690f3237..1e32e8b2925 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -9669,11 +9669,6 @@ int finalize_schema_table(st_plugin_int *plugin)
   DBUG_RETURN(0);
 }
 
-/*
-  This is used to create a timestamp field
-*/
-
-MYSQL_TIME zero_time={ 0,0,0,0,0,0,0,0, MYSQL_TIMESTAMP_TIME };
 
 /**
   Output trigger information (SHOW CREATE TRIGGER) to the client.
@@ -9758,8 +9753,9 @@ static bool show_create_trigger_impl(THD *thd, Trigger *trigger)
                                      MY_CS_NAME_SIZE),
                    mem_root);
 
+  static const Datetime zero_datetime(Datetime::zero());
   Item_datetime_literal *tmp= (new (mem_root) 
-                               Item_datetime_literal(thd, &zero_time, 2));
+                               Item_datetime_literal(thd, &zero_datetime, 2));
   tmp->set_name(thd, Lex_cstring(STRING_WITH_LEN("Created")));
   fields.push_back(tmp, mem_root);
 
diff --git a/sql/sql_type.cc b/sql/sql_type.cc
index fdd92d3d7bb..22375d2962c 100644
--- a/sql/sql_type.cc
+++ b/sql/sql_type.cc
@@ -8820,7 +8820,10 @@ Type_handler_date_common::create_literal_item(THD *thd,
   if (tmp.is_valid_temporal() &&
       tmp.get_mysql_time()->time_type == MYSQL_TIMESTAMP_DATE &&
       !have_important_literal_warnings(&st))
-    item= new (thd->mem_root) Item_date_literal(thd, tmp.get_mysql_time());
+  {
+    Date d(&tmp);
+    item= new (thd->mem_root) Item_date_literal(thd, &d);
+  }
   literal_warn(thd, item, str, length, cs, &st, "DATE", send_error);
   return item;
 }
@@ -8839,8 +8842,10 @@ Type_handler_temporal_with_date::create_literal_item(THD *thd,
   if (tmp.is_valid_temporal() &&
       tmp.get_mysql_time()->time_type == MYSQL_TIMESTAMP_DATETIME &&
       !have_important_literal_warnings(&st))
-    item= new (thd->mem_root) Item_datetime_literal(thd, tmp.get_mysql_time(),
-                                                    st.precision);
+  {
+    Datetime dt(&tmp);
+    item= new (thd->mem_root) Item_datetime_literal(thd, &dt, st.precision);
+  }
   literal_warn(thd, item, str, length, cs, &st, "DATETIME", send_error);
   return item;
 }
@@ -8859,8 +8864,7 @@ Type_handler_time_common::create_literal_item(THD *thd,
   Time tmp(thd, &st, str, length, cs, opt);
   if (tmp.is_valid_time() &&
       !have_important_literal_warnings(&st))
-    item= new (thd->mem_root) Item_time_literal(thd, tmp.get_mysql_time(),
-                                                st.precision);
+    item= new (thd->mem_root) Item_time_literal(thd, &tmp, st.precision);
   literal_warn(thd, item, str, length, cs, &st, "TIME", send_error);
   return item;
 }
diff --git a/sql/sql_type.h b/sql/sql_type.h
index f94d17541a4..41e840d9ed7 100644
--- a/sql/sql_type.h
+++ b/sql/sql_type.h
@@ -1201,6 +1201,13 @@ public:
   }
   // End of constuctors
 
+  bool copy_valid_value_to_mysql_time(MYSQL_TIME *ltime) const
+  {
+    DBUG_ASSERT(is_valid_temporal());
+    *ltime= *this;
+    return false;
+  }
+
   longlong to_longlong() const
   {
     if (!is_valid_temporal())
@@ -1867,6 +1874,11 @@ public:
   {
     return is_valid_time() ? Temporal::to_packed() : 0;
   }
+  longlong valid_time_to_packed() const
+  {
+    DBUG_ASSERT(is_valid_time_slow());
+    return Temporal::to_packed();
+  }
   long fraction_remainder(uint dec) const
   {
     DBUG_ASSERT(is_valid_time());
@@ -2041,6 +2053,11 @@ public:
   {
     return ::check_date_with_warn(thd, this, flags, MYSQL_TIMESTAMP_ERROR);
   }
+  bool check_date_with_warn(THD *thd)
+  {
+    return ::check_date_with_warn(thd, this, Temporal::sql_mode_for_dates(thd),
+                                  MYSQL_TIMESTAMP_ERROR);
+  }
   static date_conv_mode_t comparison_flags_for_get_date()
   { return TIME_INVALID_DATES | TIME_FUZZY_DATES; }
 };
@@ -2109,11 +2126,37 @@ public:
     datetime_to_date(this);
     DBUG_ASSERT(is_valid_date_slow());
   }
+  explicit Date(const Temporal_hybrid *from)
+  {
+    from->copy_valid_value_to_mysql_time(this);
+    DBUG_ASSERT(is_valid_date_slow());
+  }
   bool is_valid_date() const
   {
     DBUG_ASSERT(is_valid_value_slow());
     return time_type == MYSQL_TIMESTAMP_DATE;
   }
+  bool check_date(date_conv_mode_t flags, int *warnings) const
+  {
+    DBUG_ASSERT(is_valid_date_slow());
+    return ::check_date(this, (year || month || day),
+                        ulonglong(flags & TIME_MODE_FOR_XXX_TO_DATE),
+                        warnings);
+  }
+  bool check_date(THD *thd, int *warnings) const
+  {
+    return check_date(Temporal::sql_mode_for_dates(thd), warnings);
+  }
+  bool check_date(date_conv_mode_t flags) const
+  {
+    int dummy; /* unused */
+    return check_date(flags, &dummy);
+  }
+  bool check_date(THD *thd) const
+  {
+    int dummy;
+    return check_date(Temporal::sql_mode_for_dates(thd), &dummy);
+  }
   const MYSQL_TIME *get_mysql_time() const
   {
     DBUG_ASSERT(is_valid_date_slow());
@@ -2156,6 +2199,11 @@ public:
     return Temporal_with_date::yearweek(week_behaviour);
   }
 
+  longlong valid_date_to_packed() const
+  {
+    DBUG_ASSERT(is_valid_date_slow());
+    return Temporal::to_packed();
+  }
   longlong to_longlong() const
   {
     return is_valid_date() ? (longlong) TIME_to_ulonglong_date(this) : 0LL;
@@ -2342,6 +2390,16 @@ public:
   {
     round(thd, dec, time_round_mode_t(fuzzydate), warn);
   }
+  explicit Datetime(const Temporal_hybrid *from)
+  {
+    from->copy_valid_value_to_mysql_time(this);
+    DBUG_ASSERT(is_valid_datetime_slow());
+  }
+  explicit Datetime(const MYSQL_TIME *from)
+  {
+    *(static_cast<MYSQL_TIME*>(this))= *from;
+    DBUG_ASSERT(is_valid_datetime_slow());
+  }
 
   bool is_valid_datetime() const
   {
@@ -2364,6 +2422,10 @@ public:
     int dummy; /* unused */
     return check_date(flags, &dummy);
   }
+  bool check_date(THD *thd) const
+  {
+    return check_date(Temporal::sql_mode_for_dates(thd));
+  }
   bool hhmmssff_is_zero() const
   {
     DBUG_ASSERT(is_valid_datetime_slow());
@@ -2472,6 +2534,11 @@ public:
   {
     return is_valid_datetime() ? Temporal::to_packed() : 0;
   }
+  longlong valid_datetime_to_packed() const
+  {
+    DBUG_ASSERT(is_valid_datetime_slow());
+    return Temporal::to_packed();
+  }
   long fraction_remainder(uint dec) const
   {
     DBUG_ASSERT(is_valid_datetime());
diff --git a/sql/sys_vars.ic b/sql/sys_vars.ic
index ba56a7b4bfb..c21a4a83165 100644
--- a/sql/sys_vars.ic
+++ b/sql/sys_vars.ic
@@ -1861,13 +1861,16 @@ public:
     return false;
   }
   void session_save_default(THD *thd, set_var *var)
-  { var->save_result.ulonglong_value= global_var(ulonglong) & bitmask; }
+  {
+    var->save_result.ulonglong_value=
+      (reverse_semantics == !(global_var(ulonglong) & bitmask));
+  }
   void global_save_default(THD *thd, set_var *var)
   { var->save_result.ulonglong_value= option.def_value; }
 
   uchar *valptr(THD *thd, ulonglong val)
   {
-    thd->sys_var_tmp.my_bool_value= reverse_semantics ^ ((val & bitmask) != 0);
+    thd->sys_var_tmp.my_bool_value= (reverse_semantics == !(val & bitmask));
     return (uchar*) &thd->sys_var_tmp.my_bool_value;
   }
   uchar *session_value_ptr(THD *thd, const LEX_CSTRING *base)
@@ -2723,7 +2726,11 @@ private:
         Datetime::Options opt(TIME_CONV_NONE |
                               TIME_NO_ZERO_IN_DATE |
                               TIME_NO_ZERO_DATE, thd);
-        res= var->value->get_date(thd, &out.ltime, opt);
+        /*
+          var->value is allowed to return DATETIME and DATE
+          Make sure to convert DATE to DATETIME.
+        */
+        res= Datetime(thd, var->value, opt).copy_to_mysql_time(&out.ltime);
       }
       else // set DEFAULT from global var
       {
diff --git a/sql/table.cc b/sql/table.cc
index 6f0aa8d3418..779ce768eee 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -9600,7 +9600,8 @@ bool TR_table::query(MYSQL_TIME &commit_time, bool backwards)
   SELECT_LEX &slex= *(thd->lex->first_select_lex());
   Name_resolution_context_backup backup(slex.context, *this);
   Item *field= newx Item_field(thd, &slex.context, (*this)[FLD_COMMIT_TS]);
-  Item *value= newx Item_datetime_literal(thd, &commit_time, 6);
+  Datetime dt(&commit_time);
+  Item *value= newx Item_datetime_literal(thd, &dt, 6);
   COND *conds;
   if (backwards)
     conds= newx Item_func_ge(thd, field, value);
diff --git a/sql/wsrep_schema.cc b/sql/wsrep_schema.cc
index 7a3519d51f3..7eb165440c9 100644
--- a/sql/wsrep_schema.cc
+++ b/sql/wsrep_schema.cc
@@ -953,7 +953,7 @@ int Wsrep_schema::update_fragment_meta(THD* thd,
 
   Wsrep_schema_impl::binlog_off binlog_off(thd);
   int error;
-  uchar key[MAX_KEY_LENGTH];
+  uchar key[MAX_KEY_LENGTH+MAX_FIELD_WIDTH];
   key_part_map key_map= 0;
   TABLE* frag_table= 0;
 
@@ -1015,7 +1015,7 @@ static int remove_fragment(THD*                  thd,
               seqno.get());
   int ret= 0;
   int error;
-  uchar key[MAX_KEY_LENGTH];
+  uchar key[MAX_KEY_LENGTH+MAX_FIELD_WIDTH];
   key_part_map key_map= 0;
 
   DBUG_ASSERT(server_id.is_undefined() == false);
@@ -1141,7 +1141,7 @@ int Wsrep_schema::replay_transaction(THD* orig_thd,
   int ret= 1;
   int error;
   TABLE* frag_table= 0;
-  uchar key[MAX_KEY_LENGTH];
+  uchar key[MAX_KEY_LENGTH+MAX_FIELD_WIDTH];
   key_part_map key_map= 0;
 
   for (std::vector<wsrep::seqno>::const_iterator i= fragments.begin();
diff --git a/sql/wsrep_sst.cc b/sql/wsrep_sst.cc
index 444fb98c4b9..35fe5198bd9 100644
--- a/sql/wsrep_sst.cc
+++ b/sql/wsrep_sst.cc
@@ -1,4 +1,4 @@
-/* Copyright 2008-2017 Codership Oy <http://www.codership.com>
+/* Copyright 2008-2020 Codership Oy <http://www.codership.com>
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -54,7 +54,6 @@ my_bool wsrep_sst_donor_rejects_queries= FALSE;
 
 bool sst_joiner_completed            = false;
 bool sst_donor_completed             = false;
-bool sst_needed                      = false;
 
 struct sst_thread_arg
 {
@@ -308,7 +307,6 @@ bool wsrep_before_SE()
           && strcmp (wsrep_sst_method, WSREP_SST_MYSQLDUMP));
 }
 
-static bool            sst_in_progress = false;
 // Signal end of SST
 static void wsrep_sst_complete (THD*                thd,
                                 int const           rcode)
@@ -1629,12 +1627,11 @@ static void* sst_donor_thread (void* a)
   wsrep_uuid_t  ret_uuid= WSREP_UUID_UNDEFINED;
   // seqno of complete SST
   wsrep_seqno_t ret_seqno= WSREP_SEQNO_UNDEFINED;
-  // SST is now in progress
-  sst_in_progress= true;
-
-  wsp::thd thd(FALSE); // we turn off wsrep_on for this THD so that it can
-                       // operate with wsrep_ready == OFF
 
+  // We turn off wsrep_on for this THD so that it can
+  // operate with wsrep_ready == OFF
+  // We also set this SST thread THD as system thread
+  wsp::thd thd(FALSE, true);
   wsp::process proc(arg->cmd, "r", arg->env);
 
   err= -proc.error();
@@ -1736,10 +1733,7 @@ wait_signal:
   proc.wait();
 
   wsrep_donor_monitor_end();
-  sst_in_progress= false;
-
-
-  return NULL;
+  return nullptr;
 }
 
 static int sst_donate_other (const char*        method,
@@ -1891,8 +1885,3 @@ int wsrep_sst_donate(const std::string& msg,
 
   return (ret >= 0 ? 0 : 1);
 }
-
-bool wsrep_is_sst_progress()
-{
-  return (sst_in_progress);
-}
diff --git a/sql/wsrep_sst.h b/sql/wsrep_sst.h
index 50f2d362c5a..2389db4abe7 100644
--- a/sql/wsrep_sst.h
+++ b/sql/wsrep_sst.h
@@ -77,7 +77,6 @@ extern void wsrep_SE_init_grab();   /*! grab init critical section */
 extern void wsrep_SE_init_wait();   /*! wait for SE init to complete */
 extern void wsrep_SE_init_done();   /*! signal that SE init is complte */
 extern void wsrep_SE_initialized(); /*! mark SE initialization complete */
-extern bool wsrep_is_sst_progress();
 
 /**
    Return a string containing the state transfer request string.
@@ -103,6 +102,5 @@ int wsrep_sst_donate(const std::string& request,
 #define wsrep_SE_init_grab() do { } while(0)
 #define wsrep_SE_init_done() do { } while(0)
 #define wsrep_sst_continue() (0)
-#define wsrep_is_sst_progress() (0)
 
 #endif /* WSREP_SST_H */
diff --git a/sql/wsrep_trans_observer.h b/sql/wsrep_trans_observer.h
index 35f670b9af6..35b93cb4bf5 100644
--- a/sql/wsrep_trans_observer.h
+++ b/sql/wsrep_trans_observer.h
@@ -136,9 +136,11 @@ static inline size_t wsrep_fragments_certified_for_stmt(THD* thd)
 
 static inline int wsrep_start_transaction(THD* thd, wsrep_trx_id_t trx_id)
 {
-  return (thd->wsrep_cs().state() != wsrep::client_state::s_none  ?
-          thd->wsrep_cs().start_transaction(wsrep::transaction_id(trx_id)) :
-          0);
+  if (thd->wsrep_cs().state() != wsrep::client_state::s_none) {
+    if (wsrep_is_active(thd) == false)
+      return thd->wsrep_cs().start_transaction(wsrep::transaction_id(trx_id));
+  }
+  return 0;
 }
 
 /**/
diff --git a/sql/wsrep_utils.cc b/sql/wsrep_utils.cc
index 148def54aaf..a679304c40a 100644
--- a/sql/wsrep_utils.cc
+++ b/sql/wsrep_utils.cc
@@ -417,7 +417,7 @@ process::wait ()
   return err_;
 }
 
-thd::thd (my_bool won) : init(), ptr(new THD(0))
+thd::thd (my_bool won, bool system_thread) : init(), ptr(new THD(0))
 {
   if (ptr)
   {
@@ -426,6 +426,8 @@ thd::thd (my_bool won) : init(), ptr(new THD(0))
     wsrep_store_threadvars(ptr);
     ptr->variables.option_bits&= ~OPTION_BIN_LOG; // disable binlog
     ptr->variables.wsrep_on= won;
+    if (system_thread)
+      ptr->system_thread= SYSTEM_THREAD_GENERIC;
     ptr->security_ctx->master_access= ALL_KNOWN_ACL;
     lex_start(ptr);
   }
diff --git a/sql/wsrep_utils.h b/sql/wsrep_utils.h
index 4b010816a40..974c623521e 100644
--- a/sql/wsrep_utils.h
+++ b/sql/wsrep_utils.h
@@ -303,7 +303,7 @@ class thd
 
 public:
 
-  thd(my_bool wsrep_on);
+  thd(my_bool wsrep_on, bool system_thread=false);
   ~thd();
   THD* const ptr;
 };
diff --git a/storage/innobase/include/row0upd.h b/storage/innobase/include/row0upd.h
index f9d949fc2d7..58c60a0a4da 100644
--- a/storage/innobase/include/row0upd.h
+++ b/storage/innobase/include/row0upd.h
@@ -184,24 +184,18 @@ row_upd_index_replace_new_col_vals_index_pos(
 	const upd_t*		update,
 	mem_heap_t*		heap)
 	MY_ATTRIBUTE((nonnull));
-/***********************************************************//**
-Replaces the new column values stored in the update vector to the index entry
-given. */
-void
-row_upd_index_replace_new_col_vals(
-/*===============================*/
-	dtuple_t*	entry,	/*!< in/out: index entry where replaced;
-				the clustered index record must be
-				covered by a lock or a page latch to
-				prevent deletion (rollback or purge) */
-	dict_index_t*	index,	/*!< in: index; NOTE that this may also be a
-				non-clustered index */
-	const upd_t*	update,	/*!< in: an update vector built for the
-				CLUSTERED index so that the field number in
-				an upd_field is the clustered index position */
-	mem_heap_t*	heap)	/*!< in: memory heap for allocating and
-				copying the new values */
-	MY_ATTRIBUTE((nonnull));
+/** Replace the new column values stored in the update vector,
+during trx_undo_prev_version_build().
+@param entry   clustered index tuple where the values are replaced
+               (the clustered index leaf page latch must be held)
+@param index   clustered index
+@param update  update vector for the clustered index
+@param heap    memory heap for allocating and copying values
+@return whether the previous version was built successfully */
+bool
+row_upd_index_replace_new_col_vals(dtuple_t *entry, const dict_index_t &index,
+                                   const upd_t *update, mem_heap_t *heap)
+  MY_ATTRIBUTE((nonnull, warn_unused_result));
 /***********************************************************//**
 Replaces the new column values stored in the update vector. */
 void
diff --git a/storage/innobase/row/row0upd.cc b/storage/innobase/row/row0upd.cc
index 2639f3ec7ea..7c8d5601352 100644
--- a/storage/innobase/row/row0upd.cc
+++ b/storage/innobase/row/row0upd.cc
@@ -821,7 +821,9 @@ containing also the reference to the external part
 @param[in,out]	len		input - length of prefix to
 fetch; output: fetched length of the prefix
 @param[in,out]	heap		heap where to allocate
-@return BLOB prefix */
+@return BLOB prefix
+@retval NULL if the record is incomplete (should only happen
+in row_vers_vc_matches_cluster() executed concurrently with another purge) */
 static
 byte*
 row_upd_ext_fetch(
@@ -836,10 +838,7 @@ row_upd_ext_fetch(
 	*len = btr_copy_externally_stored_field_prefix(
 		buf, *len, zip_size, data, local_len);
 
-	/* We should never update records containing a half-deleted BLOB. */
-	ut_a(*len);
-
-	return(buf);
+	return *len ? buf : NULL;
 }
 
 /** Replaces the new column value stored in the update vector in
@@ -850,9 +849,11 @@ the given index entry field.
 @param[in]	uf		update field
 @param[in,out]	heap		memory heap for allocating and copying
 the new value
-@param[in]	zip_size	ROW_FORMAT=COMPRESSED page size, or 0 */
+@param[in]	zip_size	ROW_FORMAT=COMPRESSED page size, or 0
+@return whether the previous version was built successfully */
+MY_ATTRIBUTE((nonnull, warn_unused_result))
 static
-void
+bool
 row_upd_index_replace_new_col_val(
 	dfield_t*		dfield,
 	const dict_field_t*	field,
@@ -867,13 +868,13 @@ row_upd_index_replace_new_col_val(
 	dfield_copy_data(dfield, &uf->new_val);
 
 	if (dfield_is_null(dfield)) {
-		return;
+		return true;
 	}
 
 	len = dfield_get_len(dfield);
 	data = static_cast<const byte*>(dfield_get_data(dfield));
 
-	if (field && field->prefix_len > 0) {
+	if (field->prefix_len > 0) {
 		ibool		fetch_ext = dfield_is_ext(dfield)
 			&& len < (ulint) field->prefix_len
 			+ BTR_EXTERN_FIELD_REF_SIZE;
@@ -885,6 +886,9 @@ row_upd_index_replace_new_col_val(
 
 			data = row_upd_ext_fetch(data, l, zip_size,
 						 &len, heap);
+			if (UNIV_UNLIKELY(!data)) {
+				return false;
+			}
 		}
 
 		len = dtype_get_at_most_n_mbchars(col->prtype,
@@ -898,7 +902,7 @@ row_upd_index_replace_new_col_val(
 			dfield_dup(dfield, heap);
 		}
 
-		return;
+		return true;
 	}
 
 	switch (uf->orig_len) {
@@ -937,6 +941,8 @@ row_upd_index_replace_new_col_val(
 		dfield_set_ext(dfield);
 		break;
 	}
+
+	return true;
 }
 
 /** Apply an update vector to an metadata entry.
@@ -983,8 +989,11 @@ row_upd_index_replace_metadata(
 
 		f -= f > first;
 		const dict_field_t* field = dict_index_get_nth_field(index, f);
-		row_upd_index_replace_new_col_val(dfield, field, field->col,
-						  uf, heap, zip_size);
+		if (!row_upd_index_replace_new_col_val(dfield, field,
+						       field->col,
+						       uf, heap, zip_size)) {
+			ut_error;
+		}
 	}
 
 	ut_ad(found_mblob);
@@ -1035,69 +1044,58 @@ row_upd_index_replace_new_col_vals_index_pos(
 				update, i, false);
 		}
 
-		if (uf) {
-			row_upd_index_replace_new_col_val(
-				dtuple_get_nth_field(entry, i),
-				field, col, uf, heap, zip_size);
+		if (uf && UNIV_UNLIKELY(!row_upd_index_replace_new_col_val(
+						dtuple_get_nth_field(entry, i),
+						field, col, uf, heap,
+						zip_size))) {
+			ut_error;
 		}
 	}
 }
 
-/***********************************************************//**
-Replaces the new column values stored in the update vector to the index entry
-given. */
-void
-row_upd_index_replace_new_col_vals(
-/*===============================*/
-	dtuple_t*	entry,	/*!< in/out: index entry where replaced;
-				the clustered index record must be
-				covered by a lock or a page latch to
-				prevent deletion (rollback or purge) */
-	dict_index_t*	index,	/*!< in: index; NOTE that this may also be a
-				non-clustered index */
-	const upd_t*	update,	/*!< in: an update vector built for the
-				CLUSTERED index so that the field number in
-				an upd_field is the clustered index position */
-	mem_heap_t*	heap)	/*!< in: memory heap for allocating and
-				copying the new values */
+/** Replace the new column values stored in the update vector,
+during trx_undo_prev_version_build().
+@param entry   clustered index tuple where the values are replaced
+               (the clustered index leaf page latch must be held)
+@param index   clustered index
+@param update  update vector for the clustered index
+@param heap    memory heap for allocating and copying values
+@return whether the previous version was built successfully */
+bool
+row_upd_index_replace_new_col_vals(dtuple_t *entry, const dict_index_t &index,
+                                   const upd_t *update, mem_heap_t *heap)
 {
-	ulint			i;
-	const dict_index_t*	clust_index
-		= dict_table_get_first_index(index->table);
-	const ulint zip_size = index->table->space->zip_size();
+  ut_ad(index.is_primary());
+  const ulint zip_size= index.table->space->zip_size();
 
-	ut_ad(!index->table->skip_alter_undo);
+  ut_ad(!index.table->skip_alter_undo);
+  dtuple_set_info_bits(entry, update->info_bits);
 
-	dtuple_set_info_bits(entry, update->info_bits);
-
-	for (i = 0; i < dict_index_get_n_fields(index); i++) {
-		const dict_field_t*	field;
-		const dict_col_t*	col;
-		const upd_field_t*	uf;
-
-		field = dict_index_get_nth_field(index, i);
-		col = dict_field_get_col(field);
-		if (col->is_virtual()) {
-			const dict_v_col_t*	vcol = reinterpret_cast<
-							const dict_v_col_t*>(
-								col);
-
-			uf = upd_get_field_by_field_no(
-				update, vcol->v_pos, true);
-		} else {
-			uf = upd_get_field_by_field_no(
-				update, static_cast<uint16_t>(
-					dict_col_get_clust_pos(
-						col, clust_index)),
-				false);
-		}
+  for (ulint i= 0; i < index.n_fields; i++)
+  {
+   const dict_field_t *field= &index.fields[i];
+   const dict_col_t* col= dict_field_get_col(field);
+   const upd_field_t *uf;
+
+   if (col->is_virtual())
+   {
+     const dict_v_col_t *vcol= reinterpret_cast<const dict_v_col_t*>(col);
+     uf= upd_get_field_by_field_no(update, vcol->v_pos, true);
+   }
+   else
+     uf= upd_get_field_by_field_no(update, static_cast<uint16_t>
+                                   (dict_col_get_clust_pos(col, &index)),
+                                   false);
+
+   if (!uf)
+     continue;
+
+   if (!row_upd_index_replace_new_col_val(dtuple_get_nth_field(entry, i),
+                                          field, col, uf, heap, zip_size))
+     return false;
+  }
 
-		if (uf) {
-			row_upd_index_replace_new_col_val(
-				dtuple_get_nth_field(entry, i),
-				field, col, uf, heap, zip_size);
-		}
-	}
+  return true;
 }
 
 /** Replaces the virtual column values stored in the update vector.
@@ -2114,7 +2112,7 @@ row_upd_sec_index_entry(
 #ifdef UNIV_DEBUG
 		mtr_commit(&mtr);
 		mtr_start(&mtr);
-		ut_ad(btr_validate_index(index, 0));
+		ut_ad(btr_validate_index(index, 0) == DB_SUCCESS);
 		ut_ad(0);
 #endif /* UNIV_DEBUG */
 		break;
diff --git a/storage/innobase/trx/trx0rec.cc b/storage/innobase/trx/trx0rec.cc
index 040618015d4..e4dd666a1f2 100644
--- a/storage/innobase/trx/trx0rec.cc
+++ b/storage/innobase/trx/trx0rec.cc
@@ -2364,7 +2364,11 @@ trx_undo_prev_version_build(
 		/* The page containing the clustered index record
 		corresponding to entry is latched in mtr.  Thus the
 		following call is safe. */
-		row_upd_index_replace_new_col_vals(entry, index, update, heap);
+		if (!row_upd_index_replace_new_col_vals(entry, *index, update,
+							heap)) {
+			ut_a(v_status & TRX_UNDO_PREV_IN_PURGE);
+			return false;
+		}
 
 		/* Get number of externally stored columns in updated record */
 		const ulint n_ext = index->is_primary()