Merge 10.4 into 10.5

47 files changed, 1240 insertions, 108 deletions

diff --git a/cmake/cpack_rpm.cmake b/cmake/cpack_rpm.cmake
index f001ff11387..90b4cab53f8 100644
--- a/cmake/cpack_rpm.cmake
+++ b/cmake/cpack_rpm.cmake
@@ -195,7 +195,7 @@ MACRO(ALTERNATIVE_NAME real alt)
   SET(p "CPACK_RPM_${real}_PACKAGE_PROVIDES")
   SET(${p} "${${p}} ${alt} = ${ver} ${alt}%{?_isa} = ${ver} config(${alt}) = ${ver}")
   SET(o "CPACK_RPM_${real}_PACKAGE_OBSOLETES")
-  SET(${o} "${${o}} ${alt} ${alt}%{?_isa}")
+  SET(${o} "${${o}} ${alt}")
 ENDMACRO(ALTERNATIVE_NAME)
 
 ALTERNATIVE_NAME("devel"  "mysql-devel")
@@ -215,8 +215,9 @@ ELSEIF(RPM MATCHES "fedora" OR RPM MATCHES "(rhel|centos)7")
   ALTERNATIVE_NAME("server" "mariadb-server")
   ALTERNATIVE_NAME("server" "mysql-compat-server")
   ALTERNATIVE_NAME("test"   "mariadb-test")
-ELSEIF(RPM MATCHES "(rhel|centos)8")
-  SET(PYTHON_SHEBANG "/usr/bin/python3")
+ENDIF()
+IF(RPM MATCHES "fedora31" OR RPM MATCHES "(rhel|centos)8")
+  SET(PYTHON_SHEBANG "/usr/bin/python3" CACHE STRING "python shebang")
 ENDIF()
 
 # If we want to build build MariaDB-shared-compat,
@@ -249,6 +250,7 @@ IF(compat53 AND compat101)
 
   STRING(REPLACE "\n" " " compat_provides "${compat_provides}")
   STRING(REPLACE "\n" " " compat_obsoletes "${compat_obsoletes}")
+  STRING(REGEX REPLACE "[^ ]+\\([^ ]+ *" "" compat_obsoletes "${compat_obsoletes}")
   SETA(CPACK_RPM_compat_PACKAGE_PROVIDES "${compat_provides}")
   SETA(CPACK_RPM_compat_PACKAGE_OBSOLETES "${compat_obsoletes}")
 
diff --git a/include/lf.h b/include/lf.h
index fa8c2d3570c..933eb4938f8 100644
--- a/include/lf.h
+++ b/include/lf.h
@@ -65,7 +65,6 @@ typedef struct {
 typedef struct {
   void * volatile pin[LF_PINBOX_PINS];
   LF_PINBOX *pinbox;
-  void  **stack_ends_here;
   void  *purgatory;
   uint32 purgatory_count;
   uint32 volatile link;
diff --git a/mysql-test/suite/period/r/delete.result b/mysql-test/suite/period/r/delete.result
index 428200a4564..451017e5340 100644
--- a/mysql-test/suite/period/r/delete.result
+++ b/mysql-test/suite/period/r/delete.result
@@ -353,6 +353,18 @@ id	s	e	datediff(e, s)
 1	1999-01-01	1999-01-03	2
 1	2018-12-10	2018-12-12	2
 2	1999-01-01	1999-01-03	2
+#
+# MDEV-18929 2nd execution of SP does not detect ER_VERS_NOT_VERSIONED
+#
+create or replace table t1 (id int, s date, e date, period for apptime(s,e));
+create or replace procedure sp()
+delete from t1 for portion of othertime from '2000-01-01' to '2018-01-01';
+call sp;
+ERROR HY000: Period `othertime` is not found in table
+call sp;
+ERROR HY000: Period `othertime` is not found in table
+drop table t1;
+drop procedure sp;
 drop table t,t2,t3,log_tbl;
 drop view v;
 drop procedure log;
diff --git a/mysql-test/suite/period/t/delete.test b/mysql-test/suite/period/t/delete.test
index 00bc314160f..738d77d2c19 100644
--- a/mysql-test/suite/period/t/delete.test
+++ b/mysql-test/suite/period/t/delete.test
@@ -181,6 +181,20 @@ delete from t for portion of apptime from '1999-01-03' to '2018-12-10';
 --sorted_result
 select *, datediff(e, s) from t;
 
+--echo #
+--echo # MDEV-18929 2nd execution of SP does not detect ER_VERS_NOT_VERSIONED
+--echo #
+create or replace table t1 (id int, s date, e date, period for apptime(s,e));
+create or replace procedure sp()
+delete from t1 for portion of othertime from '2000-01-01' to '2018-01-01';
+--error ER_PERIOD_NOT_FOUND
+call sp;
+--error ER_PERIOD_NOT_FOUND
+call sp;
+drop table t1;
+drop procedure sp;
+
+
 drop table t,t2,t3,log_tbl;
 drop view v;
 drop procedure log;
diff --git a/mysql-test/suite/versioning/r/partition.result b/mysql-test/suite/versioning/r/partition.result
index a286cdf23be..84f31b55775 100644
--- a/mysql-test/suite/versioning/r/partition.result
+++ b/mysql-test/suite/versioning/r/partition.result
@@ -472,6 +472,14 @@ delete from t1 where a is not null;
 create or replace table t1 (i int) with system versioning partition by system_time limit 10;
 select * from t1 partition (p0) for system_time all;
 ERROR HY000: SYSTEM_TIME partitions in table `t1` does not support historical query
+# MDEV-18929 2nd execution of SP does not detect ER_VERS_NOT_VERSIONED
+create or replace procedure sp()
+select * from t1 partition (p0) for system_time all;
+call sp;
+ERROR HY000: SYSTEM_TIME partitions in table `t1` does not support historical query
+call sp;
+ERROR HY000: SYSTEM_TIME partitions in table `t1` does not support historical query
+drop procedure sp;
 # MDEV-15380 Index for versioned table gets corrupt after partitioning and DELETE
 create or replace table t1 (pk int primary key)
 engine=myisam
@@ -583,3 +591,17 @@ x	a
 3	bar
 4	bar
 drop table t1;
+#
+# MDEV-21011 Table corruption reported for versioned partitioned table after DELETE: "Found a misplaced row"
+#
+create table t1 (a int) with system versioning
+partition by system_time limit 3
+(partition p1 history, partition p2 history, partition pn current);
+insert into t1 values (1),(2),(3),(4);
+delete from t1;
+delete from t1;
+check table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	note	Not supported for non-INTERVAL history partitions
+test.t1	check	note	The storage engine for the table doesn't support check
+drop table t1;
diff --git a/mysql-test/suite/versioning/r/select.result b/mysql-test/suite/versioning/r/select.result
index 66e22de9b0e..87b09d8dff7 100644
--- a/mysql-test/suite/versioning/r/select.result
+++ b/mysql-test/suite/versioning/r/select.result
@@ -356,6 +356,21 @@ ERROR HY000: Table `t` is not system-versioned
 create or replace table t1 (x int) with system versioning engine myisam;
 select * from t1 for system_time as of transaction 1;
 ERROR HY000: Transaction-precise system versioning for `t1` is not supported
+# MDEV-18929 2nd execution of SP does not detect ER_VERS_NOT_VERSIONED
+create or replace procedure sp()
+select * from t1 for system_time as of transaction 1;
+call sp;
+ERROR HY000: Transaction-precise system versioning for `t1` is not supported
+call sp;
+ERROR HY000: Transaction-precise system versioning for `t1` is not supported
+create or replace table t1 (a int);
+create or replace procedure sp()
+select * from t1 for system_time all;
+call sp;
+ERROR HY000: Table `t1` is not system-versioned
+call sp;
+ERROR HY000: Table `t1` is not system-versioned
+drop procedure sp;
 create or replace table t1 (
 x int,
 sys_trx_start bigint unsigned as row start invisible,
diff --git a/mysql-test/suite/versioning/r/update.result b/mysql-test/suite/versioning/r/update.result
index eaa8549b38a..64aa7c7068c 100644
--- a/mysql-test/suite/versioning/r/update.result
+++ b/mysql-test/suite/versioning/r/update.result
@@ -276,3 +276,24 @@ update t1 set a= '2012-12-12';
 update v set a= '2000-01-01' order by b limit 1;
 drop view v;
 drop table t1, t2;
+#
+# MDEV-20441 ER_CRASHED_ON_USAGE upon update on versioned Aria table
+#
+create or replace table t1 (a varchar(8))
+engine=aria row_format=fixed
+with system versioning;
+insert into t1 (a) values ('foo');
+update t1 set a = 'bar';
+drop table t1;
+#
+# MDEV-21147 Assertion `marked_for_read()' upon UPDATE on versioned table via view
+#
+create or replace table t1 (
+pk int, a char(8), b char(8),
+primary key (pk)
+) with system versioning;
+create or replace view v1 as select * from t1;
+insert into t1 values  (1, null, 'd') ,  (2, null, 'i') ;
+update v1 set a= null where b = '';
+drop view v1;
+drop table t1;
diff --git a/mysql-test/suite/versioning/r/view.result b/mysql-test/suite/versioning/r/view.result
index b33602f7336..3b3fe580af4 100644
--- a/mysql-test/suite/versioning/r/view.result
+++ b/mysql-test/suite/versioning/r/view.result
@@ -160,6 +160,7 @@ 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, delete
 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)
@@ -211,5 +212,70 @@ x	check_row(row_start, row_end)
 1	CURRENT ROW
 2	HISTORICAL ROW
 3	HISTORICAL ROW
+# replace
+create or replace table t1 (
+x int primary key, y 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, 0), (2, 0);
+create or replace view v1 as select * from t1 where x > 1;
+replace v1 values (1, 1);
+replace v1 values (2, 1);
+replace v1 values (3, 1);
+# REPLACE ignores VIEW condition because itself doesn't use WHERE
+select *, check_row(row_start, row_end) from t1 for system_time all order by x, row_end;
+x	y	check_row(row_start, row_end)
+1	0	HISTORICAL ROW
+1	1	CURRENT ROW
+2	0	HISTORICAL ROW
+2	1	CURRENT ROW
+3	1	CURRENT ROW
+# insert-select, on duplicate key
+insert v1 select * from t1 where x = 1 on duplicate key update x = v1.x - 1;
+select *, check_row(row_start, row_end) from t1 for system_time all order by x, row_end;
+x	y	check_row(row_start, row_end)
+0	1	CURRENT ROW
+1	0	HISTORICAL ROW
+1	1	HISTORICAL ROW
+2	0	HISTORICAL ROW
+2	1	CURRENT ROW
+3	1	CURRENT ROW
 drop view v1, v2;
 drop tables t1, t2;
+#
+# MDEV-21146 Assertion `m_lock_type == 2' in handler::ha_drop_table upon LOAD DATA
+#
+create table t1 (a int);
+create view v1 as select * from t1;
+create or replace table t1 (b int) with system versioning;
+load data infile 'xx' into table v1;
+ERROR HY000: View 'test.v1' references invalid table(s) or column(s) or function(s) or definer/invoker of view lack rights to use them
+drop view v1;
+drop table t1;
+#
+# MDEV-21155 Assertion with versioned table upon DELETE from view of view after replacing first view
+#
+create table t1 (a int);
+insert into t1 values (1);
+create table t2 (
+b 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 t2 values (2);
+create view v1 as select * from t1;
+create view v2 as select * from v1;
+create or replace view v1 as select * from t2;
+delete from v2;
+select * from t1;
+a
+1
+select *, check_row(row_start, row_end) from t2 for system_time all;
+b	check_row(row_start, row_end)
+2	HISTORICAL ROW
+drop view v2;
+drop view v1;
+drop table t1, t2;
diff --git a/mysql-test/suite/versioning/t/partition.test b/mysql-test/suite/versioning/t/partition.test
index 64c4bc247c0..9fedc4fbf1a 100644
--- a/mysql-test/suite/versioning/t/partition.test
+++ b/mysql-test/suite/versioning/t/partition.test
@@ -411,6 +411,14 @@ delete from t1 where a is not null;
 create or replace table t1 (i int) with system versioning partition by system_time limit 10;
 --error ER_VERS_QUERY_IN_PARTITION
 select * from t1 partition (p0) for system_time all;
+--echo # MDEV-18929 2nd execution of SP does not detect ER_VERS_NOT_VERSIONED
+create or replace procedure sp()
+select * from t1 partition (p0) for system_time all;
+--error ER_VERS_QUERY_IN_PARTITION
+call sp;
+--error ER_VERS_QUERY_IN_PARTITION
+call sp;
+drop procedure sp;
 
 --echo # MDEV-15380 Index for versioned table gets corrupt after partitioning and DELETE
 create or replace table t1 (pk int primary key)
@@ -501,4 +509,19 @@ update t1 set a= 'bar' limit 4;
 select * from t1;
 drop table t1;
 
+--echo #
+--echo # MDEV-21011 Table corruption reported for versioned partitioned table after DELETE: "Found a misplaced row"
+--echo #
+create table t1 (a int) with system versioning
+partition by system_time limit 3
+(partition p1 history, partition p2 history, partition pn current);
+insert into t1 values (1),(2),(3),(4);
+delete from t1;
+delete from t1;
+check table t1;
+
+# cleanup
+drop table t1;
+
+
 --source suite/versioning/common_finish.inc
diff --git a/mysql-test/suite/versioning/t/select.test b/mysql-test/suite/versioning/t/select.test
index e8d34e55b16..1767457401e 100644
--- a/mysql-test/suite/versioning/t/select.test
+++ b/mysql-test/suite/versioning/t/select.test
@@ -228,6 +228,21 @@ for system_time all as t;
 create or replace table t1 (x int) with system versioning engine myisam;
 --error ER_VERS_ENGINE_UNSUPPORTED
 select * from t1 for system_time as of transaction 1;
+--echo # MDEV-18929 2nd execution of SP does not detect ER_VERS_NOT_VERSIONED
+create or replace procedure sp()
+select * from t1 for system_time as of transaction 1;
+--error ER_VERS_ENGINE_UNSUPPORTED
+call sp;
+--error ER_VERS_ENGINE_UNSUPPORTED
+call sp;
+create or replace table t1 (a int);
+create or replace procedure sp()
+select * from t1 for system_time all;
+--error ER_VERS_NOT_VERSIONED
+call sp;
+--error ER_VERS_NOT_VERSIONED
+call sp;
+drop procedure sp;
 
 create or replace table t1 (
   x int,
diff --git a/mysql-test/suite/versioning/t/update.test b/mysql-test/suite/versioning/t/update.test
index e41c7d15995..baf3c1ec876 100644
--- a/mysql-test/suite/versioning/t/update.test
+++ b/mysql-test/suite/versioning/t/update.test
@@ -157,7 +157,6 @@ replace t1 values (1,2),(1,3),(2,4);
 --echo #
 --echo # MDEV-14829 Assertion `0' failed in Protocol::end_statement upon concurrent UPDATE
 --echo #
-
 create or replace table t1 (pk int, a char(3), b char(3), primary key(pk))
   engine=innodb with system versioning;
 
@@ -192,4 +191,31 @@ drop view v;
 drop table t1, t2;
 --enable_warnings
 
+--echo #
+--echo # MDEV-20441 ER_CRASHED_ON_USAGE upon update on versioned Aria table
+--echo #
+create or replace table t1 (a varchar(8))
+engine=aria row_format=fixed
+with system versioning;
+
+insert into t1 (a) values ('foo');
+update t1 set a = 'bar';
+drop table t1;
+
+--echo #
+--echo # MDEV-21147 Assertion `marked_for_read()' upon UPDATE on versioned table via view
+--echo #
+create or replace table t1 (
+    pk int, a char(8), b char(8),
+    primary key (pk)
+) with system versioning;
+
+create or replace view v1 as select * from t1;
+insert into t1 values  (1, null, 'd') ,  (2, null, 'i') ;
+update v1 set a= null where b = '';
+
+# cleanup
+drop view v1;
+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 c05fbfd3866..288f1eb6e21 100644
--- a/mysql-test/suite/versioning/t/view.test
+++ b/mysql-test/suite/versioning/t/view.test
@@ -136,6 +136,7 @@ eval create or replace table t1 (
 ) with system versioning;
 insert into t1 values (1), (2);
 create or replace view v1 as select * from t1 where x > 1;
+--echo # update, delete
 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);
@@ -153,8 +154,63 @@ select *, check_row(row_start, row_end) from t2 for system_time all order by x;
 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;
-
+--echo # replace
+--replace_result $sys_datatype_expl SYS_DATATYPE
+eval create or replace table t1 (
+  x int primary key, y 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, 0), (2, 0);
+create or replace view v1 as select * from t1 where x > 1;
+replace v1 values (1, 1);
+replace v1 values (2, 1);
+replace v1 values (3, 1);
+--echo # REPLACE ignores VIEW condition because itself doesn't use WHERE
+select *, check_row(row_start, row_end) from t1 for system_time all order by x, row_end;
+--echo # insert-select, on duplicate key
+insert v1 select * from t1 where x = 1 on duplicate key update x = v1.x - 1;
+select *, check_row(row_start, row_end) from t1 for system_time all order by x, row_end;
 drop view v1, v2;
 drop tables t1, t2;
 
+--echo #
+--echo # MDEV-21146 Assertion `m_lock_type == 2' in handler::ha_drop_table upon LOAD DATA
+--echo #
+create table t1 (a int);
+create view v1 as select * from t1;
+create or replace table t1 (b int) with system versioning;
+--error ER_VIEW_INVALID
+load data infile 'xx' into table v1;
+
+# cleanup
+drop view v1;
+drop table t1;
+
+--echo #
+--echo # MDEV-21155 Assertion with versioned table upon DELETE from view of view after replacing first view
+--echo #
+create table t1 (a int);
+insert into t1 values (1);
+--replace_result $sys_datatype_expl SYS_DATATYPE
+eval create table t2 (
+  b 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 t2 values (2);
+create view v1 as select * from t1;
+create view v2 as select * from v1;
+create or replace view v1 as select * from t2;
+delete from v2;
+select * from t1;
+select *, check_row(row_start, row_end) from t2 for system_time all;
+
+# cleanup
+drop view v2;
+drop view v1;
+drop table t1, t2;
+
 --source suite/versioning/common_finish.inc
diff --git a/mysys/lf_alloc-pin.c b/mysys/lf_alloc-pin.c
index 2a0ee7fddf9..b98684556c3 100644
--- a/mysys/lf_alloc-pin.c
+++ b/mysys/lf_alloc-pin.c
@@ -151,7 +151,6 @@ void lf_pinbox_destroy(LF_PINBOX *pinbox)
 */
 LF_PINS *lf_pinbox_get_pins(LF_PINBOX *pinbox)
 {
-  struct st_my_thread_var *var;
   uint32 pins, next, top_ver;
   LF_PINS *el;
   /*
@@ -194,12 +193,7 @@ LF_PINS *lf_pinbox_get_pins(LF_PINBOX *pinbox)
   el->link= pins;
   el->purgatory_count= 0;
   el->pinbox= pinbox;
-  var= my_thread_var;
-  /*
-    Threads that do not call my_thread_init() should still be
-    able to use the LF_HASH.
-  */
-  el->stack_ends_here= (var ? & var->stack_ends_here : NULL);
+
   return el;
 }
 
@@ -335,16 +329,18 @@ static void lf_pinbox_real_free(LF_PINS *pins)
   void *list;
   void **addr= NULL;
   void *first= NULL, *last= NULL;
+  struct st_my_thread_var *var= my_thread_var;
+  void *stack_ends_here= var ? var->stack_ends_here : NULL;
   LF_PINBOX *pinbox= pins->pinbox;
 
   npins= pinbox->pins_in_array+1;
 
 #ifdef HAVE_ALLOCA
-  if (pins->stack_ends_here != NULL)
+  if (stack_ends_here != NULL)
   {
     int alloca_size= sizeof(void *)*LF_PINBOX_PINS*npins;
     /* create a sorted list of pinned addresses, to speed up searches */
-    if (available_stack_size(&pinbox, *pins->stack_ends_here) >
+    if (available_stack_size(&pinbox, stack_ends_here) >
         alloca_size + ALLOCA_SAFETY_MARGIN)
     {
       struct st_harvester hv;
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index d40913c21d1..77a53668638 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -4821,6 +4821,42 @@ ha_rows ha_partition::guess_bulk_insert_rows()
 }
 
 
+void ha_partition::sum_copy_info(handler *file)
+{
+  copy_info.records+= file->copy_info.records;
+  copy_info.touched+= file->copy_info.touched;
+  copy_info.copied+=  file->copy_info.copied;
+  copy_info.deleted+= file->copy_info.deleted;
+  copy_info.updated+= file->copy_info.updated;
+}
+
+
+void ha_partition::sum_copy_infos()
+{
+  handler **file_array;
+  bzero(&copy_info, sizeof(copy_info));
+  file_array= m_file;
+  do
+  {
+    if (bitmap_is_set(&(m_opened_partitions), (uint)(file_array - m_file)))
+      sum_copy_info(*file_array);
+  } while (*(++file_array));
+}
+
+void ha_partition::reset_copy_info()
+{
+  handler **file_array;
+  bzero(&copy_info, sizeof(copy_info));
+  file_array= m_file;
+  do
+  {
+    if (bitmap_is_set(&(m_opened_partitions), (uint)(file_array - m_file)))
+      bzero(&(*file_array)->copy_info, sizeof(copy_info));
+  } while (*(++file_array));
+}
+
+
+
 /*
   Finish a large batch of insert rows
 
@@ -4852,6 +4888,7 @@ int ha_partition::end_bulk_insert()
     int tmp;
     if ((tmp= m_file[i]->ha_end_bulk_insert()))
       error= tmp;
+    sum_copy_info(m_file[i]);
   }
   bitmap_clear_all(&m_bulk_insert_started);
   DBUG_RETURN(error);
@@ -10779,8 +10816,8 @@ int ha_partition::indexes_are_disabled(void)
   @param repair        If true, move misplaced rows to correct partition.
 
   @return Operation status.
-    @retval 0     Success
-    @retval != 0  Error
+    @retval HA_ADMIN_OK     Success
+    @retval != HA_ADMIN_OK  Error
 */
 
 int ha_partition::check_misplaced_rows(uint read_part_id, bool do_repair)
@@ -10794,6 +10831,17 @@ int ha_partition::check_misplaced_rows(uint read_part_id, bool do_repair)
 
   DBUG_ASSERT(m_file);
 
+  if (m_part_info->vers_info &&
+      read_part_id != m_part_info->vers_info->now_part->id &&
+      !m_part_info->vers_info->interval.is_set())
+  {
+    print_admin_msg(ha_thd(), MYSQL_ERRMSG_SIZE, "note",
+                    table_share->db.str, table->alias,
+                    opt_op_name[CHECK_PARTS],
+                    "Not supported for non-INTERVAL history partitions");
+    DBUG_RETURN(HA_ADMIN_NOT_IMPLEMENTED);
+  }
+
   if (do_repair)
   {
     /* We must read the full row, if we need to move it! */
@@ -11150,6 +11198,7 @@ bool ha_partition::start_bulk_update()
 
   do
   {
+    bzero(&(*file)->copy_info, sizeof((*file)->copy_info));
     if ((*file)->start_bulk_update())
       DBUG_RETURN(TRUE);
   } while (*(++file));
@@ -11207,6 +11256,7 @@ int ha_partition::end_bulk_update()
     if ((tmp= (*file)->end_bulk_update()))
       error= tmp;
   } while (*(++file));
+  sum_copy_infos();
   DBUG_RETURN(error);
 }
 
@@ -11303,6 +11353,7 @@ int ha_partition::end_bulk_delete()
     if ((tmp= (*file)->end_bulk_delete()))
       error= tmp;
   } while (*(++file));
+  sum_copy_infos();
   DBUG_RETURN(error);
 }
 
@@ -11419,11 +11470,13 @@ int ha_partition::pre_direct_update_rows_init(List<Item> *update_fields)
     0                         Success
 */
 
-int ha_partition::direct_update_rows(ha_rows *update_rows_result)
+int ha_partition::direct_update_rows(ha_rows *update_rows_result,
+                                     ha_rows *found_rows_result)
 {
   int error;
   bool rnd_seq= FALSE;
   ha_rows update_rows= 0;
+  ha_rows found_rows= 0;
   uint32 i;
   DBUG_ENTER("ha_partition::direct_update_rows");
 
@@ -11435,6 +11488,7 @@ int ha_partition::direct_update_rows(ha_rows *update_rows_result)
   }
 
   *update_rows_result= 0;
+  *found_rows_result= 0;
   for (i= m_part_spec.start_part; i <= m_part_spec.end_part; i++)
   {
     handler *file= m_file[i];
@@ -11450,7 +11504,8 @@ int ha_partition::direct_update_rows(ha_rows *update_rows_result)
       }
       if (unlikely((error= (m_pre_calling ?
                             (file)->pre_direct_update_rows() :
-                            (file)->ha_direct_update_rows(&update_rows)))))
+                            (file)->ha_direct_update_rows(&update_rows,
+                                                          &found_rows)))))
       {
         if (rnd_seq)
         {
@@ -11462,6 +11517,7 @@ int ha_partition::direct_update_rows(ha_rows *update_rows_result)
         DBUG_RETURN(error);
       }
       *update_rows_result+= update_rows;
+      *found_rows_result+= found_rows;
     }
     if (rnd_seq)
     {
@@ -11497,7 +11553,7 @@ int ha_partition::pre_direct_update_rows()
   DBUG_ENTER("ha_partition::pre_direct_update_rows");
   save_m_pre_calling= m_pre_calling;
   m_pre_calling= TRUE;
-  error= direct_update_rows(&not_used);
+  error= direct_update_rows(&not_used, &not_used);
   m_pre_calling= save_m_pre_calling;
   DBUG_RETURN(error);
 }
diff --git a/sql/ha_partition.h b/sql/ha_partition.h
index ff02ea5aa1c..62adca8c72f 100644
--- a/sql/ha_partition.h
+++ b/sql/ha_partition.h
@@ -391,6 +391,9 @@ private:
     part_share->next_auto_inc_val= part_share->prev_auto_inc_val;
     handler::restore_auto_increment();
   }
+  void sum_copy_info(handler *file);
+  void sum_copy_infos();
+  void reset_copy_info();
   /** Temporary storage for new partitions Handler_shares during ALTER */
   List<Parts_share_refs> m_new_partitions_share_refs;
   /** Sorted array of partition ids in descending order of number of rows. */
@@ -657,7 +660,7 @@ public:
   virtual int update_row(const uchar * old_data, const uchar * new_data);
   virtual int direct_update_rows_init(List<Item> *update_fields);
   virtual int pre_direct_update_rows_init(List<Item> *update_fields);
-  virtual int direct_update_rows(ha_rows *update_rows);
+  virtual int direct_update_rows(ha_rows *update_rows, ha_rows *found_rows);
   virtual int pre_direct_update_rows();
   virtual bool start_bulk_delete();
   virtual int end_bulk_delete();
diff --git a/sql/handler.cc b/sql/handler.cc
index 46a0c313c80..624eece9b09 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -6795,14 +6795,14 @@ int handler::ha_delete_row(const uchar *buf)
   @retval != 0          Failure.
 */
 
-int handler::ha_direct_update_rows(ha_rows *update_rows)
+int handler::ha_direct_update_rows(ha_rows *update_rows, ha_rows *found_rows)
 {
   int error;
 
   MYSQL_UPDATE_ROW_START(table_share->db.str, table_share->table_name.str);
   mark_trx_read_write();
 
-  error = direct_update_rows(update_rows);
+  error = direct_update_rows(update_rows, found_rows);
   MYSQL_UPDATE_ROW_DONE(error);
   return error;
 }
diff --git a/sql/handler.h b/sql/handler.h
index 6c9b8f4edcb..e22361b3449 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -931,6 +931,16 @@ enum tablespace_access_mode
   TS_NOT_ACCESSIBLE = 2
 };
 
+/* Statistics about batch operations like bulk_insert */
+struct ha_copy_info
+{
+  ha_rows records;        /* Used to check if rest of variables can be used */
+  ha_rows touched;
+  ha_rows copied;
+  ha_rows deleted;
+  ha_rows updated;
+};
+
 struct handlerton;
 class st_alter_tablespace : public Sql_alloc
 {
@@ -3065,6 +3075,7 @@ public:
   ulonglong rows_changed;
   /* One bigger than needed to avoid to test if key == MAX_KEY */
   ulonglong index_rows_read[MAX_KEY+1];
+  ha_copy_info copy_info;
 
 private:
   /* ANALYZE time tracker, if present */
@@ -3280,6 +3291,7 @@ public:
   {
     DBUG_ENTER("handler::ha_start_bulk_insert");
     estimation_rows_to_insert= rows;
+    bzero(&copy_info,sizeof(copy_info));
     start_bulk_insert(rows, flags);
     DBUG_VOID_RETURN;
   }
@@ -3351,6 +3363,13 @@ public:
   {
     rows_read= rows_changed= rows_tmp_read= 0;
     bzero(index_rows_read, sizeof(index_rows_read));
+    bzero(&copy_info, sizeof(copy_info));
+  }
+  virtual void reset_copy_info() {}
+  void ha_reset_copy_info()
+  {
+    bzero(&copy_info, sizeof(copy_info));
+    reset_copy_info();
   }
   virtual void change_table_ptr(TABLE *table_arg, TABLE_SHARE *share)
   {
@@ -4584,7 +4603,7 @@ private:
 
   /* Perform initialization for a direct update request */
 public:
-  int ha_direct_update_rows(ha_rows *update_rows);
+  int ha_direct_update_rows(ha_rows *update_rows, ha_rows *found_rows);
   virtual int direct_update_rows_init(List<Item> *update_fields)
   {
     return HA_ERR_WRONG_COMMAND;
@@ -4594,7 +4613,8 @@ private:
   {
     return HA_ERR_WRONG_COMMAND;
   }
-  virtual int direct_update_rows(ha_rows *update_rows __attribute__((unused)))
+  virtual int direct_update_rows(ha_rows *update_rows __attribute__((unused)),
+                                 ha_rows *found_rows __attribute__((unused)))
   {
     return HA_ERR_WRONG_COMMAND;
   }
diff --git a/sql/proxy_protocol.cc b/sql/proxy_protocol.cc
index b54af619487..550813c6457 100644
--- a/sql/proxy_protocol.cc
+++ b/sql/proxy_protocol.cc
@@ -470,7 +470,7 @@ static int compare_bits(const void *s1, const void *s2, int bit_count)
   int byte_count= bit_count / 8;
   if (byte_count && (result= memcmp(s1, s2, byte_count)))
     return result;
-  int rem= byte_count % 8;
+  int rem= bit_count % 8;
   if (rem)
   {
     // compare remaining bits i.e partial bytes.
diff --git a/sql/sql_derived.cc b/sql/sql_derived.cc
index 09f8acc536c..fc6bf44bbb1 100644
--- a/sql/sql_derived.cc
+++ b/sql/sql_derived.cc
@@ -726,12 +726,27 @@ bool mysql_derived_prepare(THD *thd, LEX *lex, TABLE_LIST *derived)
   {
     /*
        System versioned tables may still require to get versioning conditions
-       (when updating view). See vers_setup_conds().
+       when modifying view (see vers_setup_conds()). Only UPDATE and DELETE are
+       affected because they use WHERE condition.
     */
     if (!unit->prepared &&
         derived->table->versioned() &&
-        (res= unit->prepare(derived, derived->derived_result, 0)))
-      goto exit;
+        derived->merge_underlying_list &&
+        /* choose only those merged views that do not select from other views */
+        !derived->merge_underlying_list->merge_underlying_list)
+    {
+      switch (thd->lex->sql_command)
+      {
+      case SQLCOM_DELETE:
+      case SQLCOM_DELETE_MULTI:
+      case SQLCOM_UPDATE:
+      case SQLCOM_UPDATE_MULTI:
+        if ((res= unit->prepare(derived, derived->derived_result, 0)))
+          goto exit;
+      default:
+        break;
+      }
+    }
     DBUG_RETURN(FALSE);
   }
 
@@ -881,7 +896,7 @@ exit:
   {
     if (!derived->is_with_table_recursive_reference())
     {
-      if (derived->table)
+      if (derived->table && derived->table->s->tmp_table)
         free_tmp_table(thd, derived->table);
       delete derived->derived_result;
     }
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 329b6a130b0..63ce6ef30cf 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -897,6 +897,8 @@ bool mysql_insert(THD *thd, TABLE_LIST *table_list,
       using_bulk_insert= 1;
       table->file->ha_start_bulk_insert(values_list.elements);
     }
+    else
+      table->file->ha_reset_copy_info();
   }
 
   thd->abort_on_warning= !ignore && thd->is_strict_mode();
@@ -1110,11 +1112,23 @@ values_loop_end:
       auto_inc values from the delayed_insert thread as they share TABLE.
     */
     table->file->ha_release_auto_increment();
-    if (using_bulk_insert && unlikely(table->file->ha_end_bulk_insert()) &&
-        !error)
+    if (using_bulk_insert)
+    {
+      if (unlikely(table->file->ha_end_bulk_insert()) &&
+          !error)
+      {
+        table->file->print_error(my_errno,MYF(0));
+        error=1;
+      }
+    }
+    /* Get better status from handler if handler supports it */
+    if (table->file->copy_info.records)
     {
-      table->file->print_error(my_errno,MYF(0));
-      error=1;
+      DBUG_ASSERT(info.copied >= table->file->copy_info.copied);
+      info.touched= table->file->copy_info.touched;
+      info.copied=  table->file->copy_info.copied;
+      info.deleted= table->file->copy_info.deleted;
+      info.updated= table->file->copy_info.updated;
     }
     if (duplic != DUP_ERROR || ignore)
     {
@@ -1237,9 +1251,12 @@ values_loop_end:
     retval= 0;
     goto abort;
   }
+  DBUG_PRINT("info", ("touched: %llu  copied: %llu  updated: %llu  deleted: %llu",
+                      (ulonglong) info.touched, (ulonglong) info.copied,
+                      (ulonglong) info.updated, (ulonglong) info.deleted));
 
-  if ((iteration * values_list.elements) == 1 && (!(thd->variables.option_bits & OPTION_WARNINGS) ||
-				    !thd->cuted_fields))
+  if ((iteration * values_list.elements) == 1 &&
+      (!(thd->variables.option_bits & OPTION_WARNINGS) || !thd->cuted_fields))
   {
     /*
       Client expects an EOF/OK packet if result set metadata was sent. If
@@ -1652,6 +1669,8 @@ static int last_uniq_key(TABLE *table,uint keynr)
 int vers_insert_history_row(TABLE *table)
 {
   DBUG_ASSERT(table->versioned(VERS_TIMESTAMP));
+  if (!table->vers_write)
+    return 0;
   restore_record(table,record[1]);
 
   // Set Sys_end to now()
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 20dd8fe222e..220abd31187 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -896,9 +896,7 @@ bool skip_setup_conds(THD *thd)
 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(0);
+  const bool update_conds= !skip_setup_conds(thd);
 
   Query_arena backup;
   Query_arena *arena= thd->activate_stmt_arena_if_needed(&backup);
@@ -919,11 +917,15 @@ int SELECT_LEX::period_setup_conds(THD *thd, TABLE_LIST *tables)
       DBUG_RETURN(-1);
     }
 
-    conds.period= &table->table->s->period;
-    result= and_items(thd, result,
-                      period_get_condition(thd, table, this, &conds, true));
+    if (update_conds)
+    {
+      conds.period= &table->table->s->period;
+      result= and_items(thd, result,
+                        period_get_condition(thd, table, this, &conds, true));
+    }
   }
-  where= and_items(thd, where, result);
+  if (update_conds)
+    where= and_items(thd, where, result);
 
   if (arena)
     thd->restore_active_arena(arena, &backup);
@@ -934,9 +936,7 @@ int SELECT_LEX::period_setup_conds(THD *thd, TABLE_LIST *tables)
 int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
 {
   DBUG_ENTER("SELECT_LEX::vers_setup_conds");
-
-  if (skip_setup_conds(thd))
-    DBUG_RETURN(0);
+  const bool update_conds= !skip_setup_conds(thd);
 
   if (!versioned_tables)
   {
@@ -1007,13 +1007,15 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
       */
       if (table->partition_names && table->table->part_info->vers_info)
       {
-        if (vers_conditions.is_set())
+        /* If the history is stored in partitions, then partitions
+            themselves are not versioned. */
+        if (vers_conditions.was_set())
         {
           my_error(ER_VERS_QUERY_IN_PARTITION, MYF(0), table->alias.str);
           DBUG_RETURN(-1);
         }
-        else
-          vers_conditions.init(SYSTEM_TIME_ALL);
+        else if (!vers_conditions.is_set())
+          vers_conditions.set_all();
       }
 #endif
 
@@ -1058,24 +1060,27 @@ 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 (is_select)
-      table->on_expr= and_items(thd, table->on_expr, cond);
-    else
+    if (update_conds)
     {
-      if (join)
+      vers_conditions.period = &table->table->s->vers;
+      Item *cond= period_get_condition(thd, table, this, &vers_conditions,
+                                      timestamps_only);
+      if (is_select)
+        table->on_expr= and_items(thd, table->on_expr, cond);
+      else
       {
-        where= and_items(thd, join->conds, cond);
-        join->conds= where;
+        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);
       }
-      else
-        where= and_items(thd, where, cond);
-      table->where= and_items(thd, table->where, cond);
-    }
 
-    table->vers_conditions.type= SYSTEM_TIME_ALL;
+      table->vers_conditions.set_all();
+    }
   } // for (table= tables; ...)
 
   DBUG_RETURN(0);
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 739febf5e87..76cbb257fea 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -206,10 +206,10 @@ static bool check_fields(THD *thd, TABLE_LIST *table, List<Item> &items,
   return FALSE;
 }
 
-static bool check_has_vers_fields(TABLE *table, List<Item> &items)
+bool TABLE::vers_check_update(List<Item> &items)
 {
   List_iterator<Item> it(items);
-  if (!table->versioned())
+  if (!versioned_write())
     return false;
 
   while (Item *item= it++)
@@ -217,8 +217,11 @@ static bool check_has_vers_fields(TABLE *table, List<Item> &items)
     if (Item_field *item_field= item->field_for_view_update())
     {
       Field *field= item_field->field;
-      if (field->table == table && !field->vers_update_unversioned())
+      if (field->table == this && !field->vers_update_unversioned())
+      {
+        no_cache= true;
         return true;
+      }
     }
   }
   return false;
@@ -480,7 +483,7 @@ int mysql_update(THD *thd,
   {
     DBUG_RETURN(1);
   }
-  bool has_vers_fields= check_has_vers_fields(table, fields);
+  bool has_vers_fields= table->vers_check_update(fields);
   if (check_key_in_view(thd, table_list))
   {
     my_error(ER_NON_UPDATABLE_TABLE, MYF(0), table_list->alias.str, "UPDATE");
@@ -717,6 +720,11 @@ int mysql_update(THD *thd,
 
     Later we also ensure that we are only using one table (no sub queries)
   */
+  DBUG_PRINT("info", ("HA_CAN_DIRECT_UPDATE_AND_DELETE: %s", (table->file->ha_table_flags() & HA_CAN_DIRECT_UPDATE_AND_DELETE) ? "TRUE" : "FALSE"));
+  DBUG_PRINT("info", ("using_io_buffer: %s", query_plan.using_io_buffer ? "TRUE" : "FALSE"));
+  DBUG_PRINT("info", ("ignore: %s", ignore ? "TRUE" : "FALSE"));
+  DBUG_PRINT("info", ("virtual_columns_marked_for_read: %s", table->check_virtual_columns_marked_for_read() ? "TRUE" : "FALSE"));
+  DBUG_PRINT("info", ("virtual_columns_marked_for_write: %s", table->check_virtual_columns_marked_for_write() ? "TRUE" : "FALSE"));
   if ((table->file->ha_table_flags() & HA_CAN_DIRECT_UPDATE_AND_DELETE) &&
       !has_triggers && !binlog_is_row &&
       !query_plan.using_io_buffer && !ignore &&
@@ -927,11 +935,16 @@ update_begin:
   if (do_direct_update)
   {
     /* Direct updating is supported */
+    ha_rows update_rows= 0, found_rows= 0;
     DBUG_PRINT("info", ("Using direct update"));
     table->reset_default_fields();
-    if (unlikely(!(error= table->file->ha_direct_update_rows(&updated))))
+    if (unlikely(!(error= table->file->ha_direct_update_rows(&update_rows,
+                                                             &found_rows))))
       error= -1;
-    found= updated;
+    updated= update_rows;
+    found= found_rows;
+    if (found < updated)
+      found= updated;
     goto update_end;
   }
 
@@ -2243,7 +2256,7 @@ multi_update::initialize_tables(JOIN *join)
       if (safe_update_on_fly(thd, join->join_tab, table_ref, all_tables))
       {
 	table_to_update= table;			// Update table on the fly
-        has_vers_fields= check_has_vers_fields(table, *fields);
+        has_vers_fields= table->vers_check_update(*fields);
 	continue;
       }
     }
@@ -2712,7 +2725,7 @@ int multi_update::do_updates()
     if (table->vfield)
       empty_record(table);
 
-    has_vers_fields= check_has_vers_fields(table, *fields);
+    has_vers_fields= table->vers_check_update(*fields);
 
     check_opt_it.rewind();
     while(TABLE *tbl= check_opt_it++)
diff --git a/sql/table.cc b/sql/table.cc
index 2b1bb16ef24..d56981ce114 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -5249,6 +5249,7 @@ void TABLE::init(THD *thd, TABLE_LIST *tl)
   check_unique_buf= NULL;
   vers_write= s->versioned;
   quick_condition_rows=0;
+  no_cache= false;
   initialize_quick_structures();
 #ifdef HAVE_REPLICATION
   /* used in RBR Triggers */
@@ -7168,12 +7169,8 @@ void TABLE::mark_columns_needed_for_update()
     /*
       For System Versioning we have to read all columns since we store
       a copy of previous row with modified row_end back to a table.
-
-      Without write_set versioning.rpl,row is unstable until MDEV-16370 is
-      applied.
     */
     bitmap_union(read_set, &s->all_set);
-    bitmap_union(write_set, &s->all_set);
     need_signal= true;
   }
   if (check_constraints)
@@ -7336,8 +7333,16 @@ void TABLE::mark_columns_per_binlog_row_image()
           binary log will include all columns read anyway.
         */
         mark_columns_used_by_index_no_reset(s->primary_key, read_set);
-        /* Only write columns that have changed */
-        rpl_write_set= write_set;
+        if (versioned())
+        {
+          // TODO: After MDEV-18432 we don't pass history rows, so remove this:
+          rpl_write_set= &s->all_set;
+        }
+        else
+        {
+          /* Only write columns that have changed */
+          rpl_write_set= write_set;
+        }
         break;
 
       default:
diff --git a/sql/table.h b/sql/table.h
index 0c6917e7b62..5391177f625 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -326,7 +326,7 @@ typedef struct st_grant_info
 
 enum tmp_table_type
 {
-  NO_TMP_TABLE, NON_TRANSACTIONAL_TMP_TABLE, TRANSACTIONAL_TMP_TABLE,
+  NO_TMP_TABLE= 0, NON_TRANSACTIONAL_TMP_TABLE, TRANSACTIONAL_TMP_TABLE,
   INTERNAL_TMP_TABLE, SYSTEM_TMP_TABLE
 };
 enum release_type { RELEASE_NORMAL, RELEASE_WAIT_FOR_DROP };
@@ -1595,9 +1595,16 @@ public:
     return s->versioned == type;
   }
 
-  bool versioned_write(vers_kind_t type= VERS_UNDEFINED) const
+  bool versioned_write() const
   {
     DBUG_ASSERT(versioned() || !vers_write);
+    return versioned() ? vers_write : false;
+  }
+
+  bool versioned_write(vers_kind_t type) const
+  {
+    DBUG_ASSERT(type);
+    DBUG_ASSERT(versioned() || !vers_write);
     return versioned(type) ? vers_write : false;
   }
 
@@ -1620,6 +1627,8 @@ public:
   int period_make_insert(Item *src, Field *dst);
   int insert_portion_of_time(THD *thd, const vers_select_conds_t &period_conds,
                              ha_rows *rows_inserted);
+  bool vers_check_update(List<Item> &items);
+
   int delete_row();
   void vers_update_fields();
   void vers_update_end();
@@ -1858,6 +1867,7 @@ public:
 struct vers_select_conds_t
 {
   vers_system_time_t type;
+  vers_system_time_t orig_type;
   bool used:1;
   bool delete_history:1;
   Vers_history_point start;
@@ -1872,6 +1882,7 @@ struct vers_select_conds_t
   void empty()
   {
     type= SYSTEM_TIME_UNSPECIFIED;
+    orig_type= SYSTEM_TIME_UNSPECIFIED;
     used= false;
     delete_history= false;
     start.empty();
@@ -1884,6 +1895,7 @@ struct vers_select_conds_t
             Lex_ident          _name= "SYSTEM_TIME")
   {
     type= _type;
+    orig_type= _type;
     used= false;
     delete_history= (type == SYSTEM_TIME_HISTORY ||
       type == SYSTEM_TIME_BEFORE);
@@ -1892,6 +1904,12 @@ struct vers_select_conds_t
     name= _name;
   }
 
+  void set_all()
+  {
+    type= SYSTEM_TIME_ALL;
+    name= "SYSTEM_TIME";
+  }
+
   void print(String *str, enum_query_type query_type) const;
 
   bool init_from_sysvar(THD *thd);
@@ -1901,6 +1919,10 @@ struct vers_select_conds_t
     return type != SYSTEM_TIME_UNSPECIFIED;
   }
   bool check_units(THD *thd);
+  bool was_set() const
+  {
+    return orig_type != SYSTEM_TIME_UNSPECIFIED;
+  }
   bool eq(const vers_select_conds_t &conds) const;
 };
 
diff --git a/storage/innobase/trx/trx0undo.cc b/storage/innobase/trx/trx0undo.cc
index ac61e41da01..60622582480 100644
--- a/storage/innobase/trx/trx0undo.cc
+++ b/storage/innobase/trx/trx0undo.cc
@@ -1616,7 +1616,7 @@ trx_undo_commit_cleanup(trx_undo_t* undo, bool is_temp)
 
 		/* Delete first the undo log segment in the file */
 		mutex_exit(&rseg->mutex);
-		trx_undo_seg_free(undo, true);
+		trx_undo_seg_free(undo, is_temp);
 		mutex_enter(&rseg->mutex);
 
 		ut_ad(rseg->curr_size > undo->size);
diff --git a/storage/maria/ma_check.c b/storage/maria/ma_check.c
index dd4163f11eb..79c78628065 100644
--- a/storage/maria/ma_check.c
+++ b/storage/maria/ma_check.c
@@ -6195,7 +6195,7 @@ end:
 }
 
 
-	/* write suffix to data file if neaded */
+/* Write suffix to data file if needed */
 
 int maria_write_data_suffix(MARIA_SORT_INFO *sort_info, my_bool fix_datafile)
 {
diff --git a/storage/spider/ha_spider.cc b/storage/spider/ha_spider.cc
index 5920f802c9e..1cec6c894c2 100644
--- a/storage/spider/ha_spider.cc
+++ b/storage/spider/ha_spider.cc
@@ -9787,6 +9787,7 @@ void ha_spider::start_bulk_insert(
   bulk_insert = TRUE;
   bulk_size = -1;
   store_last_insert_id = 0;
+  bzero(&copy_info, sizeof(copy_info));
   DBUG_VOID_RETURN;
 }
 
@@ -9799,7 +9800,7 @@ int ha_spider::end_bulk_insert()
   bulk_insert = FALSE;
   if (bulk_size == -1)
     DBUG_RETURN(0);
-  if ((error_num = spider_db_bulk_insert(this, table, TRUE)))
+  if ((error_num = spider_db_bulk_insert(this, table, &copy_info, TRUE)))
     DBUG_RETURN(check_error_mode(error_num));
   DBUG_RETURN(0);
 }
@@ -9933,7 +9934,7 @@ int ha_spider::write_row(
     else
       bulk_size = 0;
   }
-  if ((error_num = spider_db_bulk_insert(this, table, FALSE)))
+  if ((error_num = spider_db_bulk_insert(this, table, &copy_info, FALSE)))
     DBUG_RETURN(check_error_mode(error_num));
 
 #ifdef HA_CAN_BULK_ACCESS
@@ -10558,7 +10559,8 @@ int ha_spider::direct_update_rows(
   uint range_count,
   bool sorted,
   uchar *new_data,
-  ha_rows *update_rows
+  ha_rows *update_rows,
+  ha_rows *found_rows
 ) {
   int error_num;
   THD *thd = ha_thd();
@@ -10585,17 +10587,17 @@ int ha_spider::direct_update_rows(
     if (is_bulk_access_clone)
     {
       bulk_access_pre_called = FALSE;
-      DBUG_RETURN(spider_db_bulk_direct_update(this, update_rows));
+      DBUG_RETURN(spider_db_bulk_direct_update(this, update_rows, found_rows));
     }
     DBUG_RETURN(bulk_access_link_exec_tgt->spider->ha_direct_update_rows(
-      ranges, range_count, sorted, new_data, update_rows));
+      ranges, range_count, sorted, new_data, update_rows, found_rows));
   }
 #endif
   if (
     (active_index != MAX_KEY && (error_num = index_handler_init())) ||
     (active_index == MAX_KEY && (error_num = rnd_handler_init())) ||
     (error_num = spider_db_direct_update(this, table, ranges, range_count,
-      update_rows))
+      update_rows, found_rows))
   )
     DBUG_RETURN(check_error_mode(error_num));
 
@@ -10603,14 +10605,15 @@ int ha_spider::direct_update_rows(
   if (bulk_access_executing && is_bulk_access_clone)
   {
     bulk_req_exec();
-    DBUG_RETURN(spider_db_bulk_direct_update(this, update_rows));
+    DBUG_RETURN(spider_db_bulk_direct_update(this, update_rows, found_rows));
   }
 #endif
   DBUG_RETURN(0);
 }
 #else
 int ha_spider::direct_update_rows(
-  ha_rows *update_rows
+  ha_rows *update_rows,
+  ha_rows *found_rows
 ) {
   int error_num;
   THD *thd = ha_thd();
@@ -10637,16 +10640,16 @@ int ha_spider::direct_update_rows(
     if (is_bulk_access_clone)
     {
       bulk_access_pre_called = FALSE;
-      DBUG_RETURN(spider_db_bulk_direct_update(this, update_rows));
+      DBUG_RETURN(spider_db_bulk_direct_update(this, update_rows, found_rows));
     }
     DBUG_RETURN(bulk_access_link_exec_tgt->spider->ha_direct_update_rows(
-      update_rows));
+      update_rows, found_rows));
   }
 #endif
   if (
     (active_index != MAX_KEY && (error_num = index_handler_init())) ||
     (active_index == MAX_KEY && (error_num = rnd_handler_init())) ||
-    (error_num = spider_db_direct_update(this, table, update_rows))
+    (error_num = spider_db_direct_update(this, table, update_rows, found_rows))
   )
     DBUG_RETURN(check_error_mode(error_num));
 
@@ -10654,7 +10657,7 @@ int ha_spider::direct_update_rows(
   if (bulk_access_executing && is_bulk_access_clone)
   {
     bulk_req_exec();
-    DBUG_RETURN(spider_db_bulk_direct_update(this, update_rows));
+    DBUG_RETURN(spider_db_bulk_direct_update(this, update_rows, found_rows));
   }
 #endif
   DBUG_RETURN(0);
@@ -10668,21 +10671,23 @@ int ha_spider::pre_direct_update_rows(
   uint range_count,
   bool sorted,
   uchar *new_data,
-  ha_rows *update_rows
+  ha_rows *update_rows,
+  ha_rows *found_rows
 ) {
   DBUG_ENTER("ha_spider::pre_direct_update_rows");
   DBUG_PRINT("info",("spider this=%p", this));
   DBUG_RETURN(bulk_access_link_current->spider->ha_direct_update_rows(ranges,
-    range_count, sorted, new_data, update_rows));
+    range_count, sorted, new_data, update_rows, found_rows));
 }
 #else
 int ha_spider::pre_direct_update_rows()
 {
   uint update_rows;
+  uint found_rows;
   DBUG_ENTER("ha_spider::pre_direct_update_rows");
   DBUG_PRINT("info",("spider this=%p", this));
   DBUG_RETURN(bulk_access_link_current->spider->ha_direct_update_rows(
-    &update_rows));
+    &update_rows, &found_rows));
 }
 #endif
 #endif
diff --git a/storage/spider/ha_spider.h b/storage/spider/ha_spider.h
index a146745aa97..edd7b8c881f 100644
--- a/storage/spider/ha_spider.h
+++ b/storage/spider/ha_spider.h
@@ -685,20 +685,22 @@ public:
 #endif
 #endif
 #ifdef HANDLER_HAS_DIRECT_UPDATE_ROWS_WITH_HS
-  inline int direct_update_rows(ha_rows *update_rows)
+  inline int direct_update_rows(ha_rows *update_rows, ha_rows *found_rows)
   {
-    return direct_update_rows(NULL, 0, FALSE, NULL, update_rows);
+    return direct_update_rows(NULL, 0, FALSE, NULL, update_rows, found_rows);
   }
   int direct_update_rows(
     KEY_MULTI_RANGE *ranges,
     uint range_count,
     bool sorted,
     uchar *new_data,
-    ha_rows *update_rows
+    ha_rows *update_rows,
+    ha_rows *found_row
   );
 #else
   int direct_update_rows(
-    ha_rows *update_rows
+    ha_rows *update_rows,
+    ha_rows *found_row
   );
 #endif
 #ifdef HA_CAN_BULK_ACCESS
@@ -706,15 +708,18 @@ public:
   inline int pre_direct_update_rows()
   {
     ha_rows update_rows;
+    ha_rows found_rows;
 
-    return pre_direct_update_rows(NULL, 0, FALSE, NULL, &update_rows);
+    return pre_direct_update_rows(NULL, 0, FALSE, NULL, &update_rows,
+      &found_rows);
   }
   int pre_direct_update_rows(
     KEY_MULTI_RANGE *ranges,
     uint range_count,
     bool sorted,
     uchar *new_data,
-    ha_rows *update_rows
+    ha_rows *update_rows,
+    ha_rows *found_row
   );
 #else
   int pre_direct_update_rows();
diff --git a/storage/spider/mysql-test/spider/bugfix/include/return_found_rows_insert_deinit.inc b/storage/spider/mysql-test/spider/bugfix/include/return_found_rows_insert_deinit.inc
new file mode 100644
index 00000000000..76b7582abfe
--- /dev/null
+++ b/storage/spider/mysql-test/spider/bugfix/include/return_found_rows_insert_deinit.inc
@@ -0,0 +1,11 @@
+--let $MASTER_1_COMMENT_2_1= $MASTER_1_COMMENT_2_1_BACKUP
+--let $CHILD2_1_DROP_TABLES= $CHILD2_1_DROP_TABLES_BACKUP
+--let $CHILD2_1_CREATE_TABLES= $CHILD2_1_CREATE_TABLES_BACKUP
+--let $CHILD2_1_SELECT_TABLES= $CHILD2_1_SELECT_TABLES_BACKUP
+--disable_warnings
+--disable_query_log
+--disable_result_log
+--source ../t/test_deinit.inc
+--enable_result_log
+--enable_query_log
+--enable_warnings
diff --git a/storage/spider/mysql-test/spider/bugfix/include/return_found_rows_insert_init.inc b/storage/spider/mysql-test/spider/bugfix/include/return_found_rows_insert_init.inc
new file mode 100644
index 00000000000..da6778de504
--- /dev/null
+++ b/storage/spider/mysql-test/spider/bugfix/include/return_found_rows_insert_init.inc
@@ -0,0 +1,27 @@
+--disable_warnings
+--disable_query_log
+--disable_result_log
+--source ../t/test_init.inc
+--enable_result_log
+--enable_query_log
+--enable_warnings
+--let $MASTER_1_COMMENT_2_1_BACKUP= $MASTER_1_COMMENT_2_1
+let $MASTER_1_COMMENT_2_1=
+  COMMENT='table "tbl_a", srv "s_2_1"';
+--let $CHILD2_1_DROP_TABLES_BACKUP= $CHILD2_1_DROP_TABLES
+let $CHILD2_1_DROP_TABLES=
+  DROP TABLE IF EXISTS tbl_a;
+--let $CHILD2_1_CREATE_TABLES_BACKUP= $CHILD2_1_CREATE_TABLES
+let $CHILD2_1_CREATE_TABLES=
+  CREATE TABLE tbl_a (
+    skey int NOT NULL,
+    dt date NOT NULL,
+    tm time NOT NULL,
+    PRIMARY KEY (skey)
+  ) $CHILD2_1_ENGINE $CHILD2_1_CHARSET;
+--let $CHILD2_1_SELECT_TABLES_BACKUP= $CHILD2_1_SELECT_TABLES
+let $CHILD2_1_SELECT_TABLES=
+  SELECT skey, dt, tm FROM tbl_a ORDER BY skey;
+let $CHILD2_1_SELECT_ARGUMENT1=
+  SELECT argument FROM mysql.general_log WHERE argument LIKE '%insert %' OR argument LIKE '%replace %';
+--let $MASTER_1_SET_COMMAND=set session spider_direct_dup_insert=1 $STR_SEMICOLON
diff --git a/storage/spider/mysql-test/spider/bugfix/include/return_found_rows_update_deinit.inc b/storage/spider/mysql-test/spider/bugfix/include/return_found_rows_update_deinit.inc
new file mode 100644
index 00000000000..76b7582abfe
--- /dev/null
+++ b/storage/spider/mysql-test/spider/bugfix/include/return_found_rows_update_deinit.inc
@@ -0,0 +1,11 @@
+--let $MASTER_1_COMMENT_2_1= $MASTER_1_COMMENT_2_1_BACKUP
+--let $CHILD2_1_DROP_TABLES= $CHILD2_1_DROP_TABLES_BACKUP
+--let $CHILD2_1_CREATE_TABLES= $CHILD2_1_CREATE_TABLES_BACKUP
+--let $CHILD2_1_SELECT_TABLES= $CHILD2_1_SELECT_TABLES_BACKUP
+--disable_warnings
+--disable_query_log
+--disable_result_log
+--source ../t/test_deinit.inc
+--enable_result_log
+--enable_query_log
+--enable_warnings
diff --git a/storage/spider/mysql-test/spider/bugfix/include/return_found_rows_update_init.inc b/storage/spider/mysql-test/spider/bugfix/include/return_found_rows_update_init.inc
new file mode 100644
index 00000000000..884ef74c47e
--- /dev/null
+++ b/storage/spider/mysql-test/spider/bugfix/include/return_found_rows_update_init.inc
@@ -0,0 +1,26 @@
+--disable_warnings
+--disable_query_log
+--disable_result_log
+--source ../t/test_init.inc
+--enable_result_log
+--enable_query_log
+--enable_warnings
+--let $MASTER_1_COMMENT_2_1_BACKUP= $MASTER_1_COMMENT_2_1
+let $MASTER_1_COMMENT_2_1=
+  COMMENT='table "tbl_a", srv "s_2_1"';
+--let $CHILD2_1_DROP_TABLES_BACKUP= $CHILD2_1_DROP_TABLES
+let $CHILD2_1_DROP_TABLES=
+  DROP TABLE IF EXISTS tbl_a;
+--let $CHILD2_1_CREATE_TABLES_BACKUP= $CHILD2_1_CREATE_TABLES
+let $CHILD2_1_CREATE_TABLES=
+  CREATE TABLE tbl_a (
+    skey int NOT NULL,
+    dt date NOT NULL,
+    tm time NOT NULL,
+    KEY idx1 (skey,dt)
+  ) $CHILD2_1_ENGINE $CHILD2_1_CHARSET;
+--let $CHILD2_1_SELECT_TABLES_BACKUP= $CHILD2_1_SELECT_TABLES
+let $CHILD2_1_SELECT_TABLES=
+  SELECT skey, dt, tm FROM tbl_a ORDER BY skey;
+let $CHILD2_1_SELECT_ARGUMENT1=
+  SELECT argument FROM mysql.general_log WHERE argument LIKE '%update %';
diff --git a/storage/spider/mysql-test/spider/bugfix/r/return_found_rows_insert.result b/storage/spider/mysql-test/spider/bugfix/r/return_found_rows_insert.result
new file mode 100644
index 00000000000..df88d7a5165
--- /dev/null
+++ b/storage/spider/mysql-test/spider/bugfix/r/return_found_rows_insert.result
@@ -0,0 +1,179 @@
+for master_1
+for child2
+child2_1
+child2_2
+child2_3
+for child3
+
+this test is for MDEV-18973
+
+drop and create databases
+connection master_1;
+CREATE DATABASE auto_test_local;
+USE auto_test_local;
+connection child2_1;
+SET @old_log_output = @@global.log_output;
+SET GLOBAL log_output = 'TABLE,FILE';
+CREATE DATABASE auto_test_remote;
+USE auto_test_remote;
+
+create table and insert
+connection child2_1;
+CHILD2_1_CREATE_TABLES
+TRUNCATE TABLE mysql.general_log;
+connection master_1;
+CREATE TABLE tbl_a (
+skey int NOT NULL,
+dt date NOT NULL,
+tm time NOT NULL,
+PRIMARY KEY (skey)
+) MASTER_1_ENGINE MASTER_1_CHARSET MASTER_1_COMMENT_2_1
+INSERT INTO tbl_a (skey, dt, tm) VALUES (0, '2012-01-01', '12:00:00'),(1, '2012-02-01', '12:00:00'),(2, '2012-03-01', '12:00:00'),(3, '2012-04-01', '12:00:00'),(4, '2012-05-01', '12:00:00'),(5, '2012-06-01', '12:00:00'),(6, '2012-07-01', '12:00:00'),(7, '2012-08-01', '12:00:00'),(8, '2012-09-01', '12:00:00'),(9, '2012-10-01', '12:00:00');
+FLUSH TABLES;
+
+select test 1
+connection child2_1;
+TRUNCATE TABLE mysql.general_log;
+EXE_MYSQL -v -v -u root -h localhost -P MASTER_1_MYPORT -S MASTER_1_MYSOCK -e "MASTER_1_SET_COMMAND INSERT IGNORE INTO tbl_a (skey, dt, tm) VALUES (0, '2013-01-01', '13:00:00'),(2, '2013-02-01', '13:00:00'),(4, '2013-03-01', '13:00:00'),(7, '2013-04-01', '13:00:00'),(8, '2013-05-01', '13:00:00'),(10, '2013-06-01', '13:00:00'),(11, '2013-07-01', '13:00:00'),(12, '2013-08-01', '13:00:00'),(13, '2013-09-01', '13:00:00'),(14, '2013-10-01', '13:00:00')" auto_test_local
+--------------
+set session spider_direct_dup_insert=1
+--------------
+
+Query OK, 0 rows affected
+
+--------------
+INSERT IGNORE INTO tbl_a (skey, dt, tm) VALUES (0, '2013-01-01', '13:00:00'),(2, '2013-02-01', '13:00:00'),(4, '2013-03-01', '13:00:00'),(7, '2013-04-01', '13:00:00'),(8, '2013-05-01', '13:00:00'),(10, '2013-06-01', '13:00:00'),(11, '2013-07-01', '13:00:00'),(12, '2013-08-01', '13:00:00'),(13, '2013-09-01', '13:00:00'),(14, '2013-10-01', '13:00:00')
+--------------
+
+Query OK, 5 rows affected
+Records: 10  Duplicates: 5  Warnings: 0
+
+Bye
+connection child2_1;
+SELECT argument FROM mysql.general_log WHERE argument LIKE '%insert %' OR argument LIKE '%replace %';
+argument
+insert ignore into `auto_test_remote`.`tbl_a`(`skey`,`dt`,`tm`)values(0,_latin1'2013-01-01',_latin1'13:00:00'),(2,_latin1'2013-02-01',_latin1'13:00:00'),(4,_latin1'2013-03-01',_latin1'13:00:00'),(7,_latin1'2013-04-01',_latin1'13:00:00'),(8,_latin1'2013-05-01',_latin1'13:00:00'),(10,_latin1'2013-06-01',_latin1'13:00:00'),(11,_latin1'2013-07-01',_latin1'13:00:00'),(12,_latin1'2013-08-01',_latin1'13:00:00'),(13,_latin1'2013-09-01',_latin1'13:00:00'),(14,_latin1'2013-10-01',_latin1'13:00:00')
+SELECT argument FROM mysql.general_log WHERE argument LIKE '%insert %' OR argument LIKE '%replace %'
+SELECT skey, dt, tm FROM tbl_a ORDER BY skey;
+skey	dt	tm
+0	2012-01-01	12:00:00
+1	2012-02-01	12:00:00
+2	2012-03-01	12:00:00
+3	2012-04-01	12:00:00
+4	2012-05-01	12:00:00
+5	2012-06-01	12:00:00
+6	2012-07-01	12:00:00
+7	2012-08-01	12:00:00
+8	2012-09-01	12:00:00
+9	2012-10-01	12:00:00
+10	2013-06-01	13:00:00
+11	2013-07-01	13:00:00
+12	2013-08-01	13:00:00
+13	2013-09-01	13:00:00
+14	2013-10-01	13:00:00
+TRUNCATE TABLE mysql.general_log;
+EXE_MYSQL -v -v -u root -h localhost -P MASTER_1_MYPORT -S MASTER_1_MYSOCK -e "MASTER_1_SET_COMMAND REPLACE INTO tbl_a (skey, dt, tm) VALUES (1, '2012-02-01', '12:00:00'),(3, '2012-12-01', '11:00:00'),(8, '2012-11-30', '11:00:00'),(9, '2012-11-29', '11:00:00'),(10, '2012-11-28', '11:00:00'),(15, '2012-11-27', '11:00:00'),(16, '2012-11-26', '11:00:00'),(17, '2012-11-25', '11:00:00'),(18, '2012-11-24', '11:00:00'),(19, '2012-11-23', '11:00:00')" auto_test_local
+--------------
+set session spider_direct_dup_insert=1
+--------------
+
+Query OK, 0 rows affected
+
+--------------
+REPLACE INTO tbl_a (skey, dt, tm) VALUES (1, '2012-02-01', '12:00:00'),(3, '2012-12-01', '11:00:00'),(8, '2012-11-30', '11:00:00'),(9, '2012-11-29', '11:00:00'),(10, '2012-11-28', '11:00:00'),(15, '2012-11-27', '11:00:00'),(16, '2012-11-26', '11:00:00'),(17, '2012-11-25', '11:00:00'),(18, '2012-11-24', '11:00:00'),(19, '2012-11-23', '11:00:00')
+--------------
+
+Query OK, 14 rows affected
+Records: 10  Duplicates: 4  Warnings: 0
+
+Bye
+connection child2_1;
+SELECT argument FROM mysql.general_log WHERE argument LIKE '%insert %' OR argument LIKE '%replace %';
+argument
+replace into `auto_test_remote`.`tbl_a`(`skey`,`dt`,`tm`)values(1,_latin1'2012-02-01',_latin1'12:00:00'),(3,_latin1'2012-12-01',_latin1'11:00:00'),(8,_latin1'2012-11-30',_latin1'11:00:00'),(9,_latin1'2012-11-29',_latin1'11:00:00'),(10,_latin1'2012-11-28',_latin1'11:00:00'),(15,_latin1'2012-11-27',_latin1'11:00:00'),(16,_latin1'2012-11-26',_latin1'11:00:00'),(17,_latin1'2012-11-25',_latin1'11:00:00'),(18,_latin1'2012-11-24',_latin1'11:00:00'),(19,_latin1'2012-11-23',_latin1'11:00:00')
+SELECT argument FROM mysql.general_log WHERE argument LIKE '%insert %' OR argument LIKE '%replace %'
+SELECT skey, dt, tm FROM tbl_a ORDER BY skey;
+skey	dt	tm
+0	2012-01-01	12:00:00
+1	2012-02-01	12:00:00
+2	2012-03-01	12:00:00
+3	2012-12-01	11:00:00
+4	2012-05-01	12:00:00
+5	2012-06-01	12:00:00
+6	2012-07-01	12:00:00
+7	2012-08-01	12:00:00
+8	2012-11-30	11:00:00
+9	2012-11-29	11:00:00
+10	2012-11-28	11:00:00
+11	2013-07-01	13:00:00
+12	2013-08-01	13:00:00
+13	2013-09-01	13:00:00
+14	2013-10-01	13:00:00
+15	2012-11-27	11:00:00
+16	2012-11-26	11:00:00
+17	2012-11-25	11:00:00
+18	2012-11-24	11:00:00
+19	2012-11-23	11:00:00
+TRUNCATE TABLE mysql.general_log;
+EXE_MYSQL -v -v -u root -h localhost -P MASTER_1_MYPORT -S MASTER_1_MYSOCK -e "MASTER_1_SET_COMMAND INSERT INTO tbl_a (skey, dt, tm) VALUES (1, '2012-11-01', '11:00:00'),(3, '2012-12-01', '11:00:00'),(11, '2012-11-30', '11:00:00'),(15, '2012-11-29', '11:00:00'),(16, '2012-11-28', '11:00:00'),(20, '2012-11-27', '11:00:00'),(21, '2012-11-26', '11:00:00'),(22, '2012-11-25', '11:00:00'),(23, '2012-11-24', '11:00:00'),(24, '2012-11-23', '11:00:00') ON DUPLICATE KEY UPDATE dt=VALUE(dt), tm=VALUE(tm)" auto_test_local
+--------------
+set session spider_direct_dup_insert=1
+--------------
+
+Query OK, 0 rows affected
+
+--------------
+INSERT INTO tbl_a (skey, dt, tm) VALUES (1, '2012-11-01', '11:00:00'),(3, '2012-12-01', '11:00:00'),(11, '2012-11-30', '11:00:00'),(15, '2012-11-29', '11:00:00'),(16, '2012-11-28', '11:00:00'),(20, '2012-11-27', '11:00:00'),(21, '2012-11-26', '11:00:00'),(22, '2012-11-25', '11:00:00'),(23, '2012-11-24', '11:00:00'),(24, '2012-11-23', '11:00:00') ON DUPLICATE KEY UPDATE dt=VALUE(dt), tm=VALUE(tm)
+--------------
+
+Query OK, 13 rows affected
+Records: 10  Duplicates: 4  Warnings: 0
+
+Bye
+connection child2_1;
+SELECT argument FROM mysql.general_log WHERE argument LIKE '%insert %' OR argument LIKE '%replace %';
+argument
+insert high_priority into `auto_test_remote`.`tbl_a`(`skey`,`dt`,`tm`)values(1,_latin1'2012-11-01',_latin1'11:00:00'),(3,_latin1'2012-12-01',_latin1'11:00:00'),(11,_latin1'2012-11-30',_latin1'11:00:00'),(15,_latin1'2012-11-29',_latin1'11:00:00'),(16,_latin1'2012-11-28',_latin1'11:00:00'),(20,_latin1'2012-11-27',_latin1'11:00:00'),(21,_latin1'2012-11-26',_latin1'11:00:00'),(22,_latin1'2012-11-25',_latin1'11:00:00'),(23,_latin1'2012-11-24',_latin1'11:00:00'),(24,_latin1'2012-11-23',_latin1'11:00:00') on duplicate key update `dt` = values(`dt`),`tm` = values(`tm`)
+SELECT argument FROM mysql.general_log WHERE argument LIKE '%insert %' OR argument LIKE '%replace %'
+SELECT skey, dt, tm FROM tbl_a ORDER BY skey;
+skey	dt	tm
+0	2012-01-01	12:00:00
+1	2012-11-01	11:00:00
+2	2012-03-01	12:00:00
+3	2012-12-01	11:00:00
+4	2012-05-01	12:00:00
+5	2012-06-01	12:00:00
+6	2012-07-01	12:00:00
+7	2012-08-01	12:00:00
+8	2012-11-30	11:00:00
+9	2012-11-29	11:00:00
+10	2012-11-28	11:00:00
+11	2012-11-30	11:00:00
+12	2013-08-01	13:00:00
+13	2013-09-01	13:00:00
+14	2013-10-01	13:00:00
+15	2012-11-29	11:00:00
+16	2012-11-28	11:00:00
+17	2012-11-25	11:00:00
+18	2012-11-24	11:00:00
+19	2012-11-23	11:00:00
+20	2012-11-27	11:00:00
+21	2012-11-26	11:00:00
+22	2012-11-25	11:00:00
+23	2012-11-24	11:00:00
+24	2012-11-23	11:00:00
+
+deinit
+connection master_1;
+DROP DATABASE IF EXISTS auto_test_local;
+connection child2_1;
+DROP DATABASE IF EXISTS auto_test_remote;
+SET GLOBAL log_output = @old_log_output;
+for master_1
+for child2
+child2_1
+child2_2
+child2_3
+for child3
+
+end of test
diff --git a/storage/spider/mysql-test/spider/bugfix/r/return_found_rows_update.result b/storage/spider/mysql-test/spider/bugfix/r/return_found_rows_update.result
new file mode 100644
index 00000000000..2ebc2693dc5
--- /dev/null
+++ b/storage/spider/mysql-test/spider/bugfix/r/return_found_rows_update.result
@@ -0,0 +1,99 @@
+for master_1
+for child2
+child2_1
+child2_2
+child2_3
+for child3
+
+this test is for MDEV-18973
+
+drop and create databases
+connection master_1;
+CREATE DATABASE auto_test_local;
+USE auto_test_local;
+connection child2_1;
+SET @old_log_output = @@global.log_output;
+SET GLOBAL log_output = 'TABLE,FILE';
+CREATE DATABASE auto_test_remote;
+USE auto_test_remote;
+
+create table and insert
+connection child2_1;
+CHILD2_1_CREATE_TABLES
+TRUNCATE TABLE mysql.general_log;
+connection master_1;
+CREATE TABLE tbl_a (
+skey int NOT NULL,
+dt date NOT NULL,
+tm time NOT NULL,
+KEY idx1 (skey,dt)
+) MASTER_1_ENGINE MASTER_1_CHARSET MASTER_1_COMMENT_2_1
+INSERT INTO tbl_a (skey, dt, tm) VALUES (0, '2012-01-01', '12:00:00'),(1, '2012-02-01', '12:00:00'),(2, '2012-03-01', '12:00:00'),(3, '2012-04-01', '12:00:00'),(4, '2012-05-01', '12:00:00'),(5, '2012-06-01', '12:00:00'),(6, '2012-07-01', '12:00:00'),(7, '2012-08-01', '12:00:00'),(8, '2012-09-01', '12:00:00'),(9, '2012-10-01', '12:00:00');
+INSERT INTO tbl_a (skey, dt, tm) VALUES (0, '2013-01-01', '13:00:00'),(1, '2013-02-01', '13:00:00'),(2, '2013-03-01', '13:00:00'),(3, '2013-04-01', '13:00:00'),(4, '2013-05-01', '13:00:00'),(5, '2013-06-01', '13:00:00'),(6, '2013-07-01', '13:00:00'),(7, '2013-08-01', '13:00:00'),(8, '2013-09-01', '13:00:00'),(9, '2013-10-01', '13:00:00');
+INSERT INTO tbl_a (skey, dt, tm) VALUES (0, '2012-11-01', '11:00:00'),(1, '2012-12-01', '11:00:00'),(2, '2012-11-30', '11:00:00'),(3, '2012-11-29', '11:00:00'),(4, '2012-11-28', '11:00:00'),(5, '2012-11-27', '11:00:00'),(6, '2012-11-26', '11:00:00'),(7, '2012-11-25', '11:00:00'),(8, '2012-11-24', '11:00:00'),(9, '2012-11-23', '11:00:00');
+FLUSH TABLES;
+
+select test 1
+connection child2_1;
+TRUNCATE TABLE mysql.general_log;
+EXE_MYSQL -v -v -u root -h localhost -P MASTER_1_MYPORT -S MASTER_1_MYSOCK -e "UPDATE tbl_a SET tm = '12:00:00' WHERE skey = 0" auto_test_local
+--------------
+UPDATE tbl_a SET tm = '12:00:00' WHERE skey = 0
+--------------
+
+Query OK, 2 rows affected
+Rows matched: 3  Changed: 2  Warnings: 0
+
+Bye
+connection child2_1;
+SELECT argument FROM mysql.general_log WHERE argument LIKE '%update %';
+argument
+update `auto_test_remote`.`tbl_a` set `tm` = _latin1'12:00:00' where (`skey` = 0)
+SELECT argument FROM mysql.general_log WHERE argument LIKE '%update %'
+SELECT skey, dt, tm FROM tbl_a ORDER BY skey;
+skey	dt	tm
+0	2012-01-01	12:00:00
+0	2013-01-01	12:00:00
+0	2012-11-01	12:00:00
+1	2012-12-01	11:00:00
+1	2013-02-01	13:00:00
+1	2012-02-01	12:00:00
+2	2013-03-01	13:00:00
+2	2012-11-30	11:00:00
+2	2012-03-01	12:00:00
+3	2012-11-29	11:00:00
+3	2013-04-01	13:00:00
+3	2012-04-01	12:00:00
+4	2012-11-28	11:00:00
+4	2012-05-01	12:00:00
+4	2013-05-01	13:00:00
+5	2012-11-27	11:00:00
+5	2012-06-01	12:00:00
+5	2013-06-01	13:00:00
+6	2013-07-01	13:00:00
+6	2012-11-26	11:00:00
+6	2012-07-01	12:00:00
+7	2012-11-25	11:00:00
+7	2012-08-01	12:00:00
+7	2013-08-01	13:00:00
+8	2012-09-01	12:00:00
+8	2013-09-01	13:00:00
+8	2012-11-24	11:00:00
+9	2012-10-01	12:00:00
+9	2013-10-01	13:00:00
+9	2012-11-23	11:00:00
+
+deinit
+connection master_1;
+DROP DATABASE IF EXISTS auto_test_local;
+connection child2_1;
+DROP DATABASE IF EXISTS auto_test_remote;
+SET GLOBAL log_output = @old_log_output;
+for master_1
+for child2
+child2_1
+child2_2
+child2_3
+for child3
+
+end of test
diff --git a/storage/spider/mysql-test/spider/bugfix/t/return_found_rows_insert.cnf b/storage/spider/mysql-test/spider/bugfix/t/return_found_rows_insert.cnf
new file mode 100644
index 00000000000..05dfd8a0bce
--- /dev/null
+++ b/storage/spider/mysql-test/spider/bugfix/t/return_found_rows_insert.cnf
@@ -0,0 +1,3 @@
+!include include/default_mysqld.cnf
+!include ../my_1_1.cnf
+!include ../my_2_1.cnf
diff --git a/storage/spider/mysql-test/spider/bugfix/t/return_found_rows_insert.test b/storage/spider/mysql-test/spider/bugfix/t/return_found_rows_insert.test
new file mode 100644
index 00000000000..ea2a2147910
--- /dev/null
+++ b/storage/spider/mysql-test/spider/bugfix/t/return_found_rows_insert.test
@@ -0,0 +1,98 @@
+--source ../include/return_found_rows_insert_init.inc
+--echo
+--echo this test is for MDEV-18973
+--echo
+--echo drop and create databases
+
+--connection master_1
+--disable_warnings
+CREATE DATABASE auto_test_local;
+USE auto_test_local;
+
+--connection child2_1
+SET @old_log_output = @@global.log_output;
+SET GLOBAL log_output = 'TABLE,FILE';
+CREATE DATABASE auto_test_remote;
+USE auto_test_remote;
+--enable_warnings
+
+--echo
+--echo create table and insert
+
+--connection child2_1
+--disable_query_log
+echo CHILD2_1_CREATE_TABLES;
+eval $CHILD2_1_CREATE_TABLES;
+--enable_query_log
+TRUNCATE TABLE mysql.general_log;
+
+--connection master_1
+--disable_query_log
+echo CREATE TABLE tbl_a (
+    skey int NOT NULL,
+    dt date NOT NULL,
+    tm time NOT NULL,
+    PRIMARY KEY (skey)
+) MASTER_1_ENGINE MASTER_1_CHARSET MASTER_1_COMMENT_2_1;
+eval CREATE TABLE tbl_a (
+    skey int NOT NULL,
+    dt date NOT NULL,
+    tm time NOT NULL,
+    PRIMARY KEY (skey)
+) $MASTER_1_ENGINE $MASTER_1_CHARSET $MASTER_1_COMMENT_2_1;
+--enable_query_log
+INSERT INTO tbl_a (skey, dt, tm) VALUES (0, '2012-01-01', '12:00:00'),(1, '2012-02-01', '12:00:00'),(2, '2012-03-01', '12:00:00'),(3, '2012-04-01', '12:00:00'),(4, '2012-05-01', '12:00:00'),(5, '2012-06-01', '12:00:00'),(6, '2012-07-01', '12:00:00'),(7, '2012-08-01', '12:00:00'),(8, '2012-09-01', '12:00:00'),(9, '2012-10-01', '12:00:00');
+FLUSH TABLES;
+
+--echo
+--echo select test 1
+
+--connection child2_1
+TRUNCATE TABLE mysql.general_log;
+
+--disable_query_log
+echo EXE_MYSQL -v -v -u root -h localhost -P MASTER_1_MYPORT -S MASTER_1_MYSOCK -e "MASTER_1_SET_COMMAND INSERT IGNORE INTO tbl_a (skey, dt, tm) VALUES (0, '2013-01-01', '13:00:00'),(2, '2013-02-01', '13:00:00'),(4, '2013-03-01', '13:00:00'),(7, '2013-04-01', '13:00:00'),(8, '2013-05-01', '13:00:00'),(10, '2013-06-01', '13:00:00'),(11, '2013-07-01', '13:00:00'),(12, '2013-08-01', '13:00:00'),(13, '2013-09-01', '13:00:00'),(14, '2013-10-01', '13:00:00')" auto_test_local;
+exec $EXE_MYSQL -v -v -u root -h localhost -P $MASTER_1_MYPORT -S $MASTER_1_MYSOCK -e "$MASTER_1_SET_COMMAND INSERT IGNORE INTO tbl_a (skey, dt, tm) VALUES (0, '2013-01-01', '13:00:00'),(2, '2013-02-01', '13:00:00'),(4, '2013-03-01', '13:00:00'),(7, '2013-04-01', '13:00:00'),(8, '2013-05-01', '13:00:00'),(10, '2013-06-01', '13:00:00'),(11, '2013-07-01', '13:00:00'),(12, '2013-08-01', '13:00:00'),(13, '2013-09-01', '13:00:00'),(14, '2013-10-01', '13:00:00')" auto_test_local;
+--enable_query_log
+
+--connection child2_1
+eval $CHILD2_1_SELECT_ARGUMENT1;
+eval $CHILD2_1_SELECT_TABLES;
+
+TRUNCATE TABLE mysql.general_log;
+
+--disable_query_log
+echo EXE_MYSQL -v -v -u root -h localhost -P MASTER_1_MYPORT -S MASTER_1_MYSOCK -e "MASTER_1_SET_COMMAND REPLACE INTO tbl_a (skey, dt, tm) VALUES (1, '2012-02-01', '12:00:00'),(3, '2012-12-01', '11:00:00'),(8, '2012-11-30', '11:00:00'),(9, '2012-11-29', '11:00:00'),(10, '2012-11-28', '11:00:00'),(15, '2012-11-27', '11:00:00'),(16, '2012-11-26', '11:00:00'),(17, '2012-11-25', '11:00:00'),(18, '2012-11-24', '11:00:00'),(19, '2012-11-23', '11:00:00')" auto_test_local;
+exec $EXE_MYSQL -v -v -u root -h localhost -P $MASTER_1_MYPORT -S $MASTER_1_MYSOCK -e "$MASTER_1_SET_COMMAND REPLACE INTO tbl_a (skey, dt, tm) VALUES (1, '2012-02-01', '12:00:00'),(3, '2012-12-01', '11:00:00'),(8, '2012-11-30', '11:00:00'),(9, '2012-11-29', '11:00:00'),(10, '2012-11-28', '11:00:00'),(15, '2012-11-27', '11:00:00'),(16, '2012-11-26', '11:00:00'),(17, '2012-11-25', '11:00:00'),(18, '2012-11-24', '11:00:00'),(19, '2012-11-23', '11:00:00')" auto_test_local;
+--enable_query_log
+
+--connection child2_1
+eval $CHILD2_1_SELECT_ARGUMENT1;
+eval $CHILD2_1_SELECT_TABLES;
+
+TRUNCATE TABLE mysql.general_log;
+
+--disable_query_log
+echo EXE_MYSQL -v -v -u root -h localhost -P MASTER_1_MYPORT -S MASTER_1_MYSOCK -e "MASTER_1_SET_COMMAND INSERT INTO tbl_a (skey, dt, tm) VALUES (1, '2012-11-01', '11:00:00'),(3, '2012-12-01', '11:00:00'),(11, '2012-11-30', '11:00:00'),(15, '2012-11-29', '11:00:00'),(16, '2012-11-28', '11:00:00'),(20, '2012-11-27', '11:00:00'),(21, '2012-11-26', '11:00:00'),(22, '2012-11-25', '11:00:00'),(23, '2012-11-24', '11:00:00'),(24, '2012-11-23', '11:00:00') ON DUPLICATE KEY UPDATE dt=VALUE(dt), tm=VALUE(tm)" auto_test_local;
+exec $EXE_MYSQL -v -v -u root -h localhost -P $MASTER_1_MYPORT -S $MASTER_1_MYSOCK -e "$MASTER_1_SET_COMMAND INSERT INTO tbl_a (skey, dt, tm) VALUES (1, '2012-11-01', '11:00:00'),(3, '2012-12-01', '11:00:00'),(11, '2012-11-30', '11:00:00'),(15, '2012-11-29', '11:00:00'),(16, '2012-11-28', '11:00:00'),(20, '2012-11-27', '11:00:00'),(21, '2012-11-26', '11:00:00'),(22, '2012-11-25', '11:00:00'),(23, '2012-11-24', '11:00:00'),(24, '2012-11-23', '11:00:00') ON DUPLICATE KEY UPDATE dt=VALUE(dt), tm=VALUE(tm)" auto_test_local;
+--enable_query_log
+
+--connection child2_1
+eval $CHILD2_1_SELECT_ARGUMENT1;
+eval $CHILD2_1_SELECT_TABLES;
+
+--echo
+--echo deinit
+--disable_warnings
+
+--connection master_1
+DROP DATABASE IF EXISTS auto_test_local;
+
+--connection child2_1
+DROP DATABASE IF EXISTS auto_test_remote;
+SET GLOBAL log_output = @old_log_output;
+
+--enable_warnings
+--source ../include/return_found_rows_insert_deinit.inc
+--echo
+--echo end of test
diff --git a/storage/spider/mysql-test/spider/bugfix/t/return_found_rows_update.cnf b/storage/spider/mysql-test/spider/bugfix/t/return_found_rows_update.cnf
new file mode 100644
index 00000000000..05dfd8a0bce
--- /dev/null
+++ b/storage/spider/mysql-test/spider/bugfix/t/return_found_rows_update.cnf
@@ -0,0 +1,3 @@
+!include include/default_mysqld.cnf
+!include ../my_1_1.cnf
+!include ../my_2_1.cnf
diff --git a/storage/spider/mysql-test/spider/bugfix/t/return_found_rows_update.test b/storage/spider/mysql-test/spider/bugfix/t/return_found_rows_update.test
new file mode 100644
index 00000000000..31f1d9c4c6c
--- /dev/null
+++ b/storage/spider/mysql-test/spider/bugfix/t/return_found_rows_update.test
@@ -0,0 +1,78 @@
+--source ../include/return_found_rows_update_init.inc
+--echo
+--echo this test is for MDEV-18973
+--echo
+--echo drop and create databases
+
+--connection master_1
+--disable_warnings
+CREATE DATABASE auto_test_local;
+USE auto_test_local;
+
+--connection child2_1
+SET @old_log_output = @@global.log_output;
+SET GLOBAL log_output = 'TABLE,FILE';
+CREATE DATABASE auto_test_remote;
+USE auto_test_remote;
+--enable_warnings
+
+--echo
+--echo create table and insert
+
+--connection child2_1
+--disable_query_log
+echo CHILD2_1_CREATE_TABLES;
+eval $CHILD2_1_CREATE_TABLES;
+--enable_query_log
+TRUNCATE TABLE mysql.general_log;
+
+--connection master_1
+--disable_query_log
+echo CREATE TABLE tbl_a (
+    skey int NOT NULL,
+    dt date NOT NULL,
+    tm time NOT NULL,
+    KEY idx1 (skey,dt)
+) MASTER_1_ENGINE MASTER_1_CHARSET MASTER_1_COMMENT_2_1;
+eval CREATE TABLE tbl_a (
+    skey int NOT NULL,
+    dt date NOT NULL,
+    tm time NOT NULL,
+    KEY idx1 (skey,dt)
+) $MASTER_1_ENGINE $MASTER_1_CHARSET $MASTER_1_COMMENT_2_1;
+--enable_query_log
+INSERT INTO tbl_a (skey, dt, tm) VALUES (0, '2012-01-01', '12:00:00'),(1, '2012-02-01', '12:00:00'),(2, '2012-03-01', '12:00:00'),(3, '2012-04-01', '12:00:00'),(4, '2012-05-01', '12:00:00'),(5, '2012-06-01', '12:00:00'),(6, '2012-07-01', '12:00:00'),(7, '2012-08-01', '12:00:00'),(8, '2012-09-01', '12:00:00'),(9, '2012-10-01', '12:00:00');
+INSERT INTO tbl_a (skey, dt, tm) VALUES (0, '2013-01-01', '13:00:00'),(1, '2013-02-01', '13:00:00'),(2, '2013-03-01', '13:00:00'),(3, '2013-04-01', '13:00:00'),(4, '2013-05-01', '13:00:00'),(5, '2013-06-01', '13:00:00'),(6, '2013-07-01', '13:00:00'),(7, '2013-08-01', '13:00:00'),(8, '2013-09-01', '13:00:00'),(9, '2013-10-01', '13:00:00');
+INSERT INTO tbl_a (skey, dt, tm) VALUES (0, '2012-11-01', '11:00:00'),(1, '2012-12-01', '11:00:00'),(2, '2012-11-30', '11:00:00'),(3, '2012-11-29', '11:00:00'),(4, '2012-11-28', '11:00:00'),(5, '2012-11-27', '11:00:00'),(6, '2012-11-26', '11:00:00'),(7, '2012-11-25', '11:00:00'),(8, '2012-11-24', '11:00:00'),(9, '2012-11-23', '11:00:00');
+FLUSH TABLES;
+
+--echo
+--echo select test 1
+
+--connection child2_1
+TRUNCATE TABLE mysql.general_log;
+
+--disable_query_log
+echo EXE_MYSQL -v -v -u root -h localhost -P MASTER_1_MYPORT -S MASTER_1_MYSOCK -e "UPDATE tbl_a SET tm = '12:00:00' WHERE skey = 0" auto_test_local;
+exec $EXE_MYSQL -v -v -u root -h localhost -P $MASTER_1_MYPORT -S $MASTER_1_MYSOCK -e "UPDATE tbl_a SET tm = '12:00:00' WHERE skey = 0" auto_test_local;
+--enable_query_log
+
+--connection child2_1
+eval $CHILD2_1_SELECT_ARGUMENT1;
+eval $CHILD2_1_SELECT_TABLES;
+
+--echo
+--echo deinit
+--disable_warnings
+
+--connection master_1
+DROP DATABASE IF EXISTS auto_test_local;
+
+--connection child2_1
+DROP DATABASE IF EXISTS auto_test_remote;
+SET GLOBAL log_output = @old_log_output;
+
+--enable_warnings
+--source ../include/return_found_rows_update_deinit.inc
+--echo
+--echo end of test
diff --git a/storage/spider/spd_db_conn.cc b/storage/spider/spd_db_conn.cc
index aa82f0b0fb1..8d635296854 100644
--- a/storage/spider/spd_db_conn.cc
+++ b/storage/spider/spd_db_conn.cc
@@ -5975,6 +5975,7 @@ int spider_db_bulk_insert_init(
 int spider_db_bulk_insert(
   ha_spider *spider,
   TABLE *table,
+  ha_copy_info *copy_info,
   bool bulk_end
 ) {
   int error_num, first_insert_link_idx = -1;
@@ -6021,6 +6022,7 @@ int spider_db_bulk_insert(
     if (!spider->is_bulk_access_clone)
     {
 #endif
+      bool insert_info = FALSE;
       for (
         roop_count2 = spider_conn_link_idx_next(share->link_statuses,
           spider->conn_link_idx, -1, share->link_count,
@@ -6169,6 +6171,11 @@ int spider_db_bulk_insert(
         }
         conn->mta_conn_mutex_lock_already = mta_conn_mutex_lock_already_backup;
         conn->mta_conn_mutex_unlock_later = mta_conn_mutex_unlock_later_backup;
+        if (!insert_info && copy_info)
+        {
+          insert_info =
+            conn->db_conn->inserted_info(dbton_handler, copy_info);
+        }
 #if defined(HS_HAS_SQLCOM) && defined(HAVE_HANDLERSOCKET)
         if (conn->conn_kind != SPIDER_CONN_KIND_MYSQL)
         {
@@ -6922,7 +6929,8 @@ int spider_db_direct_update(
   TABLE *table,
   KEY_MULTI_RANGE *ranges,
   uint range_count,
-  ha_rows *update_rows
+  ha_rows *update_rows,
+  ha_rows *found_rows
 ) {
   int error_num, roop_count;
   SPIDER_SHARE *share = spider->share;
@@ -7195,6 +7203,8 @@ int spider_db_direct_update(
         {
           *update_rows = spider->conns[roop_count]->db_conn->affected_rows();
           DBUG_PRINT("info", ("spider update_rows = %llu", *update_rows));
+          *found_rows = spider->conns[roop_count]->db_conn->matched_rows();
+          DBUG_PRINT("info", ("spider found_rows = %llu", *found_rows));
           counted = TRUE;
         }
 #if defined(HS_HAS_SQLCOM) && defined(HAVE_HANDLERSOCKET)
@@ -7216,6 +7226,8 @@ int spider_db_direct_update(
           {
             *update_rows = conn->db_conn->affected_rows();
             DBUG_PRINT("info", ("spider update_rows = %llu", *update_rows));
+            *found_rows = conn->db_conn->matched_rows();
+            DBUG_PRINT("info", ("spider found_rows = %llu", *found_rows));
             counted = TRUE;
           }
           result->free_result();
@@ -7253,7 +7265,8 @@ int spider_db_direct_update(
 int spider_db_direct_update(
   ha_spider *spider,
   TABLE *table,
-  ha_rows *update_rows
+  ha_rows *update_rows,
+  ha_rows *found_rows
 ) {
   int error_num, roop_count;
   SPIDER_SHARE *share = spider->share;
@@ -7446,6 +7459,8 @@ int spider_db_direct_update(
       {
         *update_rows = spider->conns[roop_count]->db_conn->affected_rows();
         DBUG_PRINT("info", ("spider update_rows = %llu", *update_rows));
+        *found_rows = spider->conns[roop_count]->db_conn->matched_rows();
+        DBUG_PRINT("info", ("spider found_rows = %llu", *found_rows));
         counted = TRUE;
       }
 #ifdef HA_CAN_BULK_ACCESS
@@ -7465,7 +7480,8 @@ int spider_db_direct_update(
 #ifdef HA_CAN_BULK_ACCESS
 int spider_db_bulk_direct_update(
   ha_spider *spider,
-  ha_rows *update_rows
+  ha_rows *update_rows,
+  ha_rows *found_rows
 ) {
   int error_num = 0, roop_count, tmp_error_num;
   SPIDER_SHARE *share = spider->share;
@@ -7510,6 +7526,8 @@ int spider_db_bulk_direct_update(
       {
         *update_rows = spider->conns[roop_count]->db_conn->affected_rows();
         DBUG_PRINT("info", ("spider update_rows = %llu", *update_rows));
+        *found_rows = spider->conns[roop_count]->db_conn->matched_rows();
+        DBUG_PRINT("info", ("spider found_rows = %llu", *found_rows));
         counted = TRUE;
       }
 #if defined(HS_HAS_SQLCOM) && defined(HAVE_HANDLERSOCKET)
@@ -7531,6 +7549,8 @@ int spider_db_bulk_direct_update(
         {
           *update_rows = conn->db_conn->affected_rows();
           DBUG_PRINT("info", ("spider update_rows = %llu", *update_rows));
+          *found_rows = conn->db_conn->matched_rows();
+          DBUG_PRINT("info", ("spider found_rows = %llu", *found_rows));
           counted = TRUE;
         }
         result->free_result();
diff --git a/storage/spider/spd_db_conn.h b/storage/spider/spd_db_conn.h
index 0300dc6c407..6fdb4b694ae 100644
--- a/storage/spider/spd_db_conn.h
+++ b/storage/spider/spd_db_conn.h
@@ -1,4 +1,5 @@
-/* Copyright (C) 2008-2018 Kentoku Shiba
+/* Copyright (C) 2008-2019 Kentoku Shiba
+   Copyright (C) 2019 MariaDB corp
 
   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
@@ -745,6 +746,7 @@ int spider_db_bulk_insert_init(
 int spider_db_bulk_insert(
   ha_spider *spider,
   TABLE *table,
+  ha_copy_info *copy_info,
   bool bulk_end
 );
 
@@ -788,13 +790,15 @@ int spider_db_direct_update(
   TABLE *table,
   KEY_MULTI_RANGE *ranges,
   uint range_count,
-  ha_rows *update_rows
+  ha_rows *update_rows,
+  ha_rows *found_rows
 );
 #else
 int spider_db_direct_update(
   ha_spider *spider,
   TABLE *table,
-  ha_rows *update_rows
+  ha_rows *update_rows,
+  ha_rows *found_rows
 );
 #endif
 #endif
diff --git a/storage/spider/spd_db_handlersocket.cc b/storage/spider/spd_db_handlersocket.cc
index 091f48a2460..7408ec33cee 100644
--- a/storage/spider/spd_db_handlersocket.cc
+++ b/storage/spider/spd_db_handlersocket.cc
@@ -1655,6 +1655,22 @@ uint spider_db_handlersocket::affected_rows()
   DBUG_RETURN((uint) my_strtoll10(hs_row->begin(), (char**) NULL, &error_num));
 }
 
+uint spider_db_handlersocket::matched_rows()
+{
+  DBUG_ENTER("spider_db_handlersocket::matched_rows");
+  DBUG_PRINT("info",("spider this=%p", this));
+  DBUG_RETURN(0);
+}
+
+bool spider_db_handlersocket::inserted_info(
+  spider_db_handler *handler,
+  ha_copy_info *copy_info
+) {
+  DBUG_ENTER("spider_db_handlersocket::inserted_info");
+  DBUG_PRINT("info",("spider this=%p", this));
+  DBUG_RETURN(FALSE);
+}
+
 ulonglong spider_db_handlersocket::last_insert_id()
 {
   DBUG_ENTER("spider_db_handlersocket::last_insert_id");
diff --git a/storage/spider/spd_db_handlersocket.h b/storage/spider/spd_db_handlersocket.h
index d2beb2124c0..19a4a391ed6 100644
--- a/storage/spider/spd_db_handlersocket.h
+++ b/storage/spider/spd_db_handlersocket.h
@@ -335,6 +335,11 @@ public:
   );
   int next_result();
   uint affected_rows();
+  uint matched_rows();
+  bool inserted_info(
+    spider_db_handler *handler,
+    spider_copy_info *copy_info
+  );
   ulonglong last_insert_id();
   int set_character_set(
     const char *csname
diff --git a/storage/spider/spd_db_include.h b/storage/spider/spd_db_include.h
index 56a88a2b7bc..4daa5c71b93 100644
--- a/storage/spider/spd_db_include.h
+++ b/storage/spider/spd_db_include.h
@@ -259,6 +259,7 @@ typedef struct st_spider_transaction SPIDER_TRX;
 typedef struct st_spider_share SPIDER_SHARE;
 class ha_spider;
 class spider_db_copy_table;
+class spider_db_handler;
 
 class spider_string
 {
@@ -1109,6 +1110,11 @@ public:
   ) = 0;
   virtual int next_result() = 0;
   virtual uint affected_rows() = 0;
+  virtual uint matched_rows() = 0;
+  virtual bool inserted_info(
+    spider_db_handler *handler,
+    ha_copy_info *copy_info
+  ) = 0;
   virtual ulonglong last_insert_id() = 0;
   virtual int set_character_set(
     const char *csname
diff --git a/storage/spider/spd_db_mysql.cc b/storage/spider/spd_db_mysql.cc
index 6b551804c87..2c35bd7bf28 100644
--- a/storage/spider/spd_db_mysql.cc
+++ b/storage/spider/spd_db_mysql.cc
@@ -184,6 +184,11 @@ static uchar SPIDER_SQL_LINESTRING_HEAD_STR[] =
   {0x00,0x00,0x00,0x00,0x01,0x02,0x00,0x00,0x00,0x02,0x00,0x00,0x00};
 #define SPIDER_SQL_LINESTRING_HEAD_LEN sizeof(SPIDER_SQL_LINESTRING_HEAD_STR)
 
+#define SPIDER_SQL_DIRECT_INSERT_KIND_INSERT     0
+#define SPIDER_SQL_DIRECT_INSERT_KIND_REPLACE    1
+#define SPIDER_SQL_DIRECT_INSERT_KIND_IGNORE     2
+#define SPIDER_SQL_DIRECT_INSERT_KIND_DUP_UPDATE 3
+
 static const char *spider_db_table_lock_str[] =
 {
   " read local,",
@@ -2444,6 +2449,81 @@ uint spider_db_mbase::affected_rows()
   DBUG_RETURN((uint) last_used_con->affected_rows);
 }
 
+uint spider_db_mbase::matched_rows()
+{
+  MYSQL *last_used_con;
+  DBUG_ENTER("spider_db_mysql::matched_rows");
+  DBUG_PRINT("info", ("spider this=%p", this));
+#if MYSQL_VERSION_ID < 50500
+  last_used_con = db_conn->last_used_con;
+#else
+  last_used_con = db_conn;
+#endif
+  /* Rows matched: 65 Changed: 65 Warnings: 0 */
+  const char *info = last_used_con->info;
+  if (!info)
+    DBUG_RETURN(0);
+  DBUG_PRINT("info", ("spider info=%s", info));
+  const char *begin = strstr(info, "Rows matched: ");
+  if (!begin)
+    DBUG_RETURN(0);
+  DBUG_RETURN(atoi(begin + strlen("Rows matched: ")));
+}
+
+bool spider_db_mbase::inserted_info(
+  spider_db_handler *handler,
+  ha_copy_info *copy_info
+) {
+  MYSQL *last_used_con;
+  uchar direct_insert_kind =
+    ((spider_mbase_handler *) handler)->direct_insert_kind;
+  DBUG_ENTER("spider_db_mysql::inserted_info");
+  DBUG_PRINT("info", ("spider this=%p", this));
+  if (direct_insert_kind == SPIDER_SQL_DIRECT_INSERT_KIND_INSERT)
+  {
+    DBUG_RETURN(TRUE);
+  }
+#if MYSQL_VERSION_ID < 50500
+  last_used_con = db_conn->last_used_con;
+#else
+  last_used_con = db_conn;
+#endif
+  /* Records: 10  Duplicates: 4  Warnings: 0 */
+  const char *info = last_used_con->info;
+  if (!info)
+    DBUG_RETURN(FALSE);
+  DBUG_PRINT("info", ("spider info=%s", info));
+  const char *begin = strstr(info, "Records: ");
+  if (!begin)
+    DBUG_RETURN(FALSE);
+  begin += strlen("Records: ");
+  uint records = atoi(begin);
+  begin = strstr(begin, "Duplicates: ");
+  if (!begin)
+    DBUG_RETURN(FALSE);
+  uint duplicates = atoi(begin + strlen("Duplicates: "));
+  copy_info->records+= records;
+  switch (direct_insert_kind)
+  {
+    case SPIDER_SQL_DIRECT_INSERT_KIND_IGNORE:
+      copy_info->copied+= duplicates;
+      break;
+    case SPIDER_SQL_DIRECT_INSERT_KIND_REPLACE:
+      copy_info->copied+= records;
+      copy_info->deleted+= duplicates;
+      break;
+    case SPIDER_SQL_DIRECT_INSERT_KIND_DUP_UPDATE:
+      copy_info->touched+= (last_used_con->affected_rows - (duplicates * 2));
+      copy_info->copied+= (last_used_con->affected_rows - duplicates);
+      copy_info->updated+= duplicates;
+      break;
+    default:
+      DBUG_ASSERT(0);
+      DBUG_RETURN(FALSE);
+  }
+  DBUG_RETURN(TRUE);
+}
+
 ulonglong spider_db_mbase::last_insert_id()
 {
   MYSQL *last_used_con;
@@ -9017,6 +9097,7 @@ int spider_mbase_handler::append_insert(
 ) {
   SPIDER_SHARE *share = spider->share;
   DBUG_ENTER("spider_mbase_handler::append_insert");
+  direct_insert_kind = SPIDER_SQL_DIRECT_INSERT_KIND_INSERT;
   if (
     (
       spider->write_can_replace ||
@@ -9026,6 +9107,7 @@ int spider_mbase_handler::append_insert(
     ) &&
     spider->direct_dup_insert
   ) {
+    direct_insert_kind = SPIDER_SQL_DIRECT_INSERT_KIND_REPLACE;
     if (str->reserve(SPIDER_SQL_REPLACE_LEN))
       DBUG_RETURN(HA_ERR_OUT_OF_MEM);
     str->q_append(SPIDER_SQL_REPLACE_STR, SPIDER_SQL_REPLACE_LEN);
@@ -9073,6 +9155,7 @@ int spider_mbase_handler::append_insert(
     spider->sql_command != SQLCOM_REPLACE &&
     spider->sql_command != SQLCOM_REPLACE_SELECT
   ) {
+    direct_insert_kind = SPIDER_SQL_DIRECT_INSERT_KIND_IGNORE;
     if (str->reserve(SPIDER_SQL_SQL_IGNORE_LEN))
       DBUG_RETURN(HA_ERR_OUT_OF_MEM);
     str->q_append(SPIDER_SQL_SQL_IGNORE_STR, SPIDER_SQL_SQL_IGNORE_LEN);
@@ -11911,6 +11994,7 @@ int spider_mbase_handler::append_insert_terminator(
     dup_update_sql.length()
   ) {
     DBUG_PRINT("info",("spider add duplicate key update"));
+    direct_insert_kind = SPIDER_SQL_DIRECT_INSERT_KIND_DUP_UPDATE;
     str->length(str->length() - SPIDER_SQL_COMMA_LEN);
     if (str->reserve(SPIDER_SQL_DUPLICATE_KEY_UPDATE_LEN +
       dup_update_sql.length()))
diff --git a/storage/spider/spd_db_mysql.h b/storage/spider/spd_db_mysql.h
index 51db3b1f2fa..626bb4d5624 100644
--- a/storage/spider/spd_db_mysql.h
+++ b/storage/spider/spd_db_mysql.h
@@ -453,6 +453,11 @@ public:
   );
   int next_result();
   uint affected_rows();
+  uint matched_rows();
+  bool inserted_info(
+    spider_db_handler *handler,
+    ha_copy_info *copy_info
+  );
   ulonglong last_insert_id();
   int set_character_set(
     const char *csname
@@ -801,6 +806,7 @@ public:
   spider_mbase_share      *mysql_share;
   SPIDER_LINK_FOR_HASH    *link_for_hash;
   uchar                   *minimum_select_bitmap;
+  uchar                   direct_insert_kind;
   spider_mbase_handler(
     ha_spider *spider,
     spider_mbase_share *share,
diff --git a/storage/spider/spd_db_oracle.cc b/storage/spider/spd_db_oracle.cc
index e56cb31a32c..1a14537cd9f 100644
--- a/storage/spider/spd_db_oracle.cc
+++ b/storage/spider/spd_db_oracle.cc
@@ -1731,6 +1731,22 @@ uint spider_db_oracle::affected_rows()
   DBUG_RETURN(update_rows);
 }
 
+uint spider_db_oracle::matched_rows()
+{
+  DBUG_ENTER("spider_db_oracle::matched_rows");
+  DBUG_PRINT("info",("spider this=%p", this));
+  DBUG_RETURN(0);
+}
+
+bool spider_db_oracle::inserted_info(
+  spider_db_handler *handler,
+  ha_copy_info *copy_info
+) {
+  DBUG_ENTER("spider_db_oracle::inserted_info");
+  DBUG_PRINT("info",("spider this=%p", this));
+  DBUG_RETURN(FALSE);
+}
+
 ulonglong spider_db_oracle::last_insert_id()
 {
   DBUG_ENTER("spider_db_oracle::last_insert_id");
diff --git a/storage/spider/spd_db_oracle.h b/storage/spider/spd_db_oracle.h
index a4be417bc67..ebdc23a9bfa 100644
--- a/storage/spider/spd_db_oracle.h
+++ b/storage/spider/spd_db_oracle.h
@@ -385,6 +385,11 @@ public:
   );
   int next_result();
   uint affected_rows();
+  uint matched_rows();
+  bool inserted_info(
+    spider_db_handler *handler,
+    ha_copy_info *copy_info
+  );
   ulonglong last_insert_id();
   int set_character_set(
     const char *csname