diff --git a/mysql-test/main/delete_use_source.result b/mysql-test/main/delete_use_source.result
index 608d20b383d0a..e4d0c554ccd91 100644
--- a/mysql-test/main/delete_use_source.result
+++ b/mysql-test/main/delete_use_source.result
@@ -350,7 +350,7 @@ id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
 1	PRIMARY	t2	ALL	NULL	NULL	NULL	NULL	8	100.00	Using where
 1	PRIMARY	t2	ref	a	a	5	test.t2.c1	1	100.00	Using index; FirstMatch(t2)
 Warnings:
-Note	1003	delete  from `test`.`t1` `t2` using (`test`.`t2`) where `test`.`t2`.`a` = `test`.`t2`.`c1`
+Note	1003	delete  from `t2` using (`test`.`t2`) where `test`.`t2`.`a` = `test`.`t2`.`c1`
 delete from t1 t2 where t2.c1 in (select a from t2);
 select * from t1;
 c1	c2	c3
@@ -368,7 +368,7 @@ id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
 1	PRIMARY	t_x	ALL	NULL	NULL	NULL	NULL	7	100.00	Using where
 1	PRIMARY	t_y	ref	a	a	5	test.t_x.c2	1	100.00	Using index; FirstMatch(t_x)
 Warnings:
-Note	1003	delete  from `test`.`t1` `t_x` using (`test`.`t2` `t_y`) where `test`.`t_y`.`a` = `test`.`t_x`.`c2`
+Note	1003	delete  from `t_x` using (`test`.`t2` `t_y`) where `test`.`t_y`.`a` = `test`.`t_x`.`c2`
 delete from t1 t_x where t_x.c2 IN (select a from t2 as t_y);
 select * from t1;
 c1	c2	c3
@@ -383,7 +383,7 @@ id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
 1	PRIMARY	t_x	ALL	NULL	NULL	NULL	NULL	4	100.00	Using where
 1	PRIMARY	t_x	ref	a	a	5	test.t_x.c3	1	100.00	Using index; FirstMatch(t_x)
 Warnings:
-Note	1003	delete  from `test`.`t1` `t_x` using (`test`.`t2` `t_x`) where `test`.`t_x`.`a` = `test`.`t_x`.`c3`
+Note	1003	delete  from `t_x` using (`test`.`t2` `t_x`) where `test`.`t_x`.`a` = `test`.`t_x`.`c3`
 delete from t1 as t_x where t_x.c3 IN (select a from t2 t_x);
 select * from t1;
 c1	c2	c3
@@ -397,7 +397,7 @@ id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
 1	PRIMARY	t2	ALL	NULL	NULL	NULL	NULL	3	100.00	
 1	PRIMARY	t1	index	NULL	a	5	NULL	3	33.33	Using where; Using index; FirstMatch(t2)
 Warnings:
-Note	1003	delete  from `test`.`t1` `t2` using (`test`.`t2` `t1`) where `test`.`t2`.`c1` = `test`.`t1`.`a` - 1
+Note	1003	delete  from `t2` using (`test`.`t2` `t1`) where `test`.`t2`.`c1` = `test`.`t1`.`a` - 1
 delete from t1 t2 where t2.c1 IN (select a -1  from t2 as t1);
 select * from t1;
 c1	c2	c3
diff --git a/mysql-test/main/opt_hints_join_order.result b/mysql-test/main/opt_hints_join_order.result
index 70df5ca241434..e2924f02d893f 100644
--- a/mysql-test/main/opt_hints_join_order.result
+++ b/mysql-test/main/opt_hints_join_order.result
@@ -893,7 +893,7 @@ id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
 1	PRIMARY	ta1	eq_ref	PRIMARY	PRIMARY	4	test.t2.f1	1	100.00	Using where
 1	PRIMARY	ta2	ALL	NULL	NULL	NULL	NULL	3	100.00	
 Warnings:
-Note	1003	delete  from `test`.`t1` `ta1` using `test`.`t2` semi join (`test`.`t3`) left join (`test`.`t1` `ta1` join `test`.`t1` `ta2`) on(`test`.`ta1`.`f1` = `test`.`t2`.`f1` and 1 and `test`.`t2`.`f1` is not null) where `test`.`t3`.`f1` = 9
+Note	1003	delete  from `ta1` using `test`.`t2` semi join (`test`.`t3`) left join (`test`.`t1` `ta1` join `test`.`t1` `ta2`) on(`test`.`ta1`.`f1` = `test`.`t2`.`f1` and 1 and `test`.`t2`.`f1` is not null) where `test`.`t3`.`f1` = 9
 # Applicable hint
 EXPLAIN EXTENDED DELETE /*+ JOIN_PREFIX(t2, t3, ta2) */
 FROM ta1.* USING t1 AS ta1 JOIN t1 AS ta2 ON 1
@@ -906,7 +906,7 @@ id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
 1	PRIMARY	ta1	eq_ref	PRIMARY	PRIMARY	4	test.t2.f1	1	100.00	
 2	MATERIALIZED	t3	ALL	NULL	NULL	NULL	NULL	3	33.33	Using where
 Warnings:
-Note	1003	delete /*+ JOIN_PREFIX(@`select#1` `t2`,`t3`,`ta2`) */  from `test`.`t1` `ta1` using `test`.`t2` semi join (`test`.`t3`) left join (`test`.`t1` `ta1` join `test`.`t1` `ta2`) on(`test`.`ta1`.`f1` = `test`.`t2`.`f1` and 1 and `test`.`t2`.`f1` is not null) where `test`.`t3`.`f1` = 9
+Note	1003	delete /*+ JOIN_PREFIX(@`select#1` `t2`,`t3`,`ta2`) */  from `ta1` using `test`.`t2` semi join (`test`.`t3`) left join (`test`.`t1` `ta1` join `test`.`t1` `ta2`) on(`test`.`ta1`.`f1` = `test`.`t2`.`f1` and 1 and `test`.`t2`.`f1` is not null) where `test`.`t3`.`f1` = 9
 # Applicable hint
 EXPLAIN EXTENDED DELETE /*+ JOIN_PREFIX(t2, t3, ta1, ta2) */
 FROM ta1.* USING t1 AS ta1 JOIN t1 AS ta2 ON 1
@@ -919,7 +919,7 @@ id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
 1	PRIMARY	ta2	ALL	NULL	NULL	NULL	NULL	3	100.00	
 2	MATERIALIZED	t3	ALL	NULL	NULL	NULL	NULL	3	33.33	Using where
 Warnings:
-Note	1003	delete /*+ JOIN_PREFIX(@`select#1` `t2`,`t3`,`ta1`,`ta2`) */  from `test`.`t1` `ta1` using `test`.`t2` semi join (`test`.`t3`) left join (`test`.`t1` `ta1` join `test`.`t1` `ta2`) on(`test`.`ta1`.`f1` = `test`.`t2`.`f1` and 1 and `test`.`t2`.`f1` is not null) where `test`.`t3`.`f1` = 9
+Note	1003	delete /*+ JOIN_PREFIX(@`select#1` `t2`,`t3`,`ta1`,`ta2`) */  from `ta1` using `test`.`t2` semi join (`test`.`t3`) left join (`test`.`t1` `ta1` join `test`.`t1` `ta2`) on(`test`.`ta1`.`f1` = `test`.`t2`.`f1` and 1 and `test`.`t2`.`f1` is not null) where `test`.`t3`.`f1` = 9
 # Applicable hint
 EXPLAIN EXTENDED DELETE /*+ JOIN_PREFIX(t2, t3, ta2, ta1) */
 FROM ta1.* USING t1 AS ta1 JOIN t1 AS ta2 ON 1
@@ -932,7 +932,7 @@ id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
 1	PRIMARY	ta1	eq_ref	PRIMARY	PRIMARY	4	test.t2.f1	1	100.00	
 2	MATERIALIZED	t3	ALL	NULL	NULL	NULL	NULL	3	33.33	Using where
 Warnings:
-Note	1003	delete /*+ JOIN_PREFIX(@`select#1` `t2`,`t3`,`ta2`,`ta1`) */  from `test`.`t1` `ta1` using `test`.`t2` semi join (`test`.`t3`) left join (`test`.`t1` `ta1` join `test`.`t1` `ta2`) on(`test`.`ta1`.`f1` = `test`.`t2`.`f1` and 1 and `test`.`t2`.`f1` is not null) where `test`.`t3`.`f1` = 9
+Note	1003	delete /*+ JOIN_PREFIX(@`select#1` `t2`,`t3`,`ta2`,`ta1`) */  from `ta1` using `test`.`t2` semi join (`test`.`t3`) left join (`test`.`t1` `ta1` join `test`.`t1` `ta2`) on(`test`.`ta1`.`f1` = `test`.`t2`.`f1` and 1 and `test`.`t2`.`f1` is not null) where `test`.`t3`.`f1` = 9
 # Hint should be ignored
 EXPLAIN EXTENDED DELETE /*+ JOIN_SUFFIX(ta2, t3, ta1) */
 FROM ta1.* USING t1 AS ta1 JOIN t1 AS ta2 ON 1
@@ -945,7 +945,7 @@ id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
 1	PRIMARY	ta2	ALL	NULL	NULL	NULL	NULL	3	100.00	
 Warnings:
 Warning	4219	Hint JOIN_SUFFIX(`ta2`,`t3`,`ta1`) is ignored as conflicting/duplicated
-Note	1003	delete  from `test`.`t1` `ta1` using `test`.`t2` semi join (`test`.`t3`) left join (`test`.`t1` `ta1` join `test`.`t1` `ta2`) on(`test`.`ta1`.`f1` = `test`.`t2`.`f1` and 1 and `test`.`t2`.`f1` is not null) where `test`.`t3`.`f1` = 9
+Note	1003	delete  from `ta1` using `test`.`t2` semi join (`test`.`t3`) left join (`test`.`t1` `ta1` join `test`.`t1` `ta2`) on(`test`.`ta1`.`f1` = `test`.`t2`.`f1` and 1 and `test`.`t2`.`f1` is not null) where `test`.`t3`.`f1` = 9
 # Both hints are ignored
 EXPLAIN EXTENDED DELETE /*+ JOIN_PREFIX(ta1, t2, t3) JOIN_SUFFIX(t3, ta2) */
 FROM ta1.* USING t1 AS ta1 JOIN t1 AS ta2 ON 1
@@ -959,7 +959,7 @@ id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
 Warnings:
 Warning	4219	Hint JOIN_PREFIX(`ta1`,`t2`,`t3`) is ignored as conflicting/duplicated
 Warning	4219	Hint JOIN_SUFFIX(`t3`,`ta2`) is ignored as conflicting/duplicated
-Note	1003	delete  from `test`.`t1` `ta1` using `test`.`t2` semi join (`test`.`t3`) left join (`test`.`t1` `ta1` join `test`.`t1` `ta2`) on(`test`.`ta1`.`f1` = `test`.`t2`.`f1` and 1 and `test`.`t2`.`f1` is not null) where `test`.`t3`.`f1` = 9
+Note	1003	delete  from `ta1` using `test`.`t2` semi join (`test`.`t3`) left join (`test`.`t1` `ta1` join `test`.`t1` `ta2`) on(`test`.`ta1`.`f1` = `test`.`t2`.`f1` and 1 and `test`.`t2`.`f1` is not null) where `test`.`t3`.`f1` = 9
 DROP TABLE t1, t2, t3;
 # Const table behavior, table order is not changed, hint is applicable.
 # Note: Const tables are excluded from the process of dependency setting
diff --git a/mysql-test/suite/federated/federatedx_pushdown_upd_del.result b/mysql-test/suite/federated/federatedx_pushdown_upd_del.result
new file mode 100644
index 0000000000000..80a9844223cd1
--- /dev/null
+++ b/mysql-test/suite/federated/federatedx_pushdown_upd_del.result
@@ -0,0 +1,404 @@
+connect  master,127.0.0.1,root,,test,$MASTER_MYPORT,;
+connect  slave,127.0.0.1,root,,test,$SLAVE_MYPORT,;
+connection master;
+CREATE DATABASE federated;
+connection slave;
+CREATE DATABASE federated;
+connection default;
+ALTER DATABASE federated CHARACTER SET latin1;
+set global federated_pushdown=1;
+connection slave;
+DROP TABLE IF EXISTS federated.t1, federated.t2, federated.t3;
+Warnings:
+Note	1051	Unknown table 'federated.t1,federated.t2,federated.t3'
+CREATE TABLE federated.t1 (a int, b varchar(16)) DEFAULT CHARSET=latin1;
+CREATE TABLE federated.t2 (a int, c varchar(16)) DEFAULT CHARSET=latin1;
+CREATE TABLE federated.t3 (a int, b int, UNIQUE KEY(b));
+SET @old_log_output= @@global.log_output;
+SET @old_general_log= @@global.general_log;
+SET global log_output='TABLE';
+connection master;
+DROP TABLE IF EXISTS federated.t1, federated.t2, federated.t3;
+Warnings:
+Note	1051	Unknown table 'federated.t1,federated.t2,federated.t3'
+CREATE TABLE federated.t1 (a int, b varchar(16))
+ENGINE="FEDERATED" DEFAULT CHARSET=latin1
+CONNECTION='mysql://root@127.0.0.1:SLAVE_PORT/federated/t1';
+CREATE TABLE federated.t2 (a int, c varchar(16))
+ENGINE="FEDERATED" DEFAULT CHARSET=latin1
+CONNECTION='mysql://root@127.0.0.1:SLAVE_PORT/federated/t2';
+CREATE TABLE federated.t3 (a int, b int, UNIQUE KEY(b))
+ENGINE="FEDERATED"
+CONNECTION='mysql://root@127.0.0.1:SLAVE_PORT/federated/t3';
+USE federated;
+CREATE PROCEDURE fill_tables()
+BEGIN
+DELETE FROM t1;
+DELETE FROM t2;
+DELETE FROM t3;
+INSERT INTO t1 VALUES (1,'one'),(2,'two'),(3,'three'),(4,'four');
+INSERT INTO t2 VALUES (1,'I'),(2,'II'),(3,'III');
+INSERT INTO t3 VALUES (1,10),(2,20),(3,30);
+END|
+#
+# ==================================================================
+# The statement the remote server receives
+# ==================================================================
+#
+# A pushed down statement is printed back by SELECT_LEX::print() with
+# QT_SELECT_ONLY cleared, which produces
+#   update <join> set <assignments> where <cond>
+# for an UPDATE and
+#   delete from <targets> using <join> where <cond>
+# for a DELETE. Both are understood by the remote server.
+#
+connection master;
+CALL fill_tables();
+connection slave;
+TRUNCATE TABLE mysql.general_log;
+SET global general_log=1;
+connection master;
+UPDATE t1, t2 SET t1.b= 'updated' WHERE t1.a= t2.a AND t1.a < 3;
+UPDATE t1 JOIN t2 ON t1.a= t2.a SET t1.b= CONCAT(t2.c,'!') WHERE t2.a > 2;
+UPDATE t1 AS x JOIN t2 AS y ON x.a= y.a SET x.b= y.c WHERE y.a= 1;
+UPDATE t1, t2 SET t1.b= 'both', t2.c= 'both' WHERE t1.a= t2.a;
+UPDATE t1, t2 SET t1.b= 'sub' WHERE t1.a= t2.a AND
+t1.a IN (SELECT a FROM t2 WHERE c= 'both');
+DELETE t1 FROM t1 JOIN t2 ON t1.a= t2.a WHERE t1.a= 3;
+DELETE FROM x USING t1 AS x JOIN t2 AS y ON x.a= y.a WHERE y.a= 2;
+DELETE t1, t2 FROM t1 JOIN t2 ON t1.a= t2.a;
+connection slave;
+SET global general_log=0;
+SELECT argument FROM mysql.general_log
+WHERE argument LIKE 'update %' OR argument LIKE 'delete %';
+argument
+update `federated`.`t1` join `federated`.`t2` set `federated`.`t1`.`b` = 'updated' where `federated`.`t1`.`a` = `federated`.`t2`.`a` and `federated`.`t1`.`a` < 3
+update (`federated`.`t1` join `federated`.`t2` on(`federated`.`t1`.`a` = `federated`.`t2`.`a`)) set `federated`.`t1`.`b` = concat(`federated`.`t2`.`c`,'!') where `federated`.`t2`.`a` > 2
+update (`federated`.`t1` `x` join `federated`.`t2` `y` on(`federated`.`x`.`a` = `federated`.`y`.`a`)) set `federated`.`x`.`b` = `federated`.`y`.`c` where `federated`.`y`.`a` = 1
+update `federated`.`t1` join `federated`.`t2` set `federated`.`t1`.`b` = 'both',`federated`.`t2`.`c` = 'both' where `federated`.`t1`.`a` = `federated`.`t2`.`a`
+update `federated`.`t1` join `federated`.`t2` set `federated`.`t1`.`b` = 'sub' where `federated`.`t1`.`a` = `federated`.`t2`.`a` and `federated`.`t1`.`a` in (select `federated`.`t2`.`a` from `federated`.`t2` where `federated`.`t2`.`c` = 'both')
+delete  from `federated`.`t1` using (`federated`.`t1` join `federated`.`t2` on(`federated`.`t1`.`a` = `federated`.`t2`.`a`)) where `federated`.`t1`.`a` = 3
+delete  from `x` using (`federated`.`t1` `x` join `federated`.`t2` `y` on(`federated`.`x`.`a` = `federated`.`y`.`a`)) where `federated`.`y`.`a` = 2
+delete  from `federated`.`t1`,`federated`.`t2` using (`federated`.`t1` join `federated`.`t2` on(`federated`.`t1`.`a` = `federated`.`t2`.`a`))
+#
+# ==================================================================
+# Row counts reported by batch_update_delete()
+# ==================================================================
+#
+connection master;
+# All matched rows are changed
+CALL fill_tables();
+UPDATE t1, t2 SET t1.b= 'updated' WHERE t1.a= t2.a AND t1.a < 3;
+affected rows: 2
+info: Rows matched: 2  Changed: 2  Warnings: 0
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	updated
+2	updated
+3	three
+4	four
+# A matched row that keeps its value is counted as matched but not as
+# changed. The count comes from federatedx_io_mysql::matched_rows(),
+# which parses it out of the remote server's info string
+CALL fill_tables();
+UPDATE t1, t2 SET t1.b= 'two' WHERE t1.a= t2.a AND t1.a < 3;
+affected rows: 1
+info: Rows matched: 2  Changed: 1  Warnings: 0
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	two
+2	two
+3	three
+4	four
+# Two tables updated at once
+CALL fill_tables();
+UPDATE t1, t2 SET t1.b= 'upd1', t2.c= 'upd2' WHERE t1.a= t2.a;
+affected rows: 6
+info: Rows matched: 6  Changed: 6  Warnings: 0
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	upd1
+2	upd1
+3	upd1
+4	four
+SELECT * FROM t2 ORDER BY a;
+a	c
+1	upd2
+2	upd2
+3	upd2
+# No row matches
+CALL fill_tables();
+UPDATE t1, t2 SET t1.b= 'nope' WHERE t1.a= t2.a AND t1.a > 100;
+affected rows: 0
+info: Rows matched: 0  Changed: 0  Warnings: 0
+DELETE t1 FROM t1 JOIN t2 ON t1.a= t2.a WHERE t1.a > 100;
+affected rows: 0
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	one
+2	two
+3	three
+4	four
+# For a DELETE the remote server reports no matched count, so
+# matched_rows() falls back to the number of deleted rows
+CALL fill_tables();
+DELETE t1 FROM t1 JOIN t2 ON t1.a= t2.a WHERE t2.c <> 'II';
+affected rows: 2
+SELECT * FROM t1 ORDER BY a;
+a	b
+2	two
+4	four
+# Two tables deleted from at once
+CALL fill_tables();
+DELETE t1, t2 FROM t1 JOIN t2 ON t1.a= t2.a WHERE t1.a= 2;
+affected rows: 2
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	one
+3	three
+4	four
+SELECT * FROM t2 ORDER BY a;
+a	c
+1	I
+3	III
+# Aliased delete target
+CALL fill_tables();
+DELETE FROM x USING t1 AS x JOIN t2 AS y ON x.a= y.a WHERE y.a= 2;
+affected rows: 1
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	one
+3	three
+4	four
+#
+# ==================================================================
+# An error on the remote server is reported to the client
+# ==================================================================
+#
+CALL fill_tables();
+UPDATE t1, t2 SET t1.a= 1/0 WHERE t1.a= t2.a;
+ERROR HY000: There was a problem processing the query on the foreign data source. Data source error: Division by 0
+UPDATE t3, t2 SET t3.b= 30 WHERE t3.a= t2.a AND t3.a < 3;
+ERROR HY000: There was a problem processing the query on the foreign data source. Data source error: Duplicate entry '30' for key 'b'
+SELECT * FROM t3 ORDER BY a;
+a	b
+1	10
+2	20
+3	30
+#
+# ==================================================================
+# Inside a transaction the statement is not auto-committed by itself
+# ==================================================================
+#
+CALL fill_tables();
+BEGIN;
+UPDATE t1, t2 SET t1.b= 'in trx' WHERE t1.a= t2.a;
+affected rows: 3
+info: Rows matched: 3  Changed: 3  Warnings: 0
+DELETE t1 FROM t1 JOIN t2 ON t1.a= t2.a WHERE t1.a= 3;
+affected rows: 1
+COMMIT;
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	in trx
+2	in trx
+4	four
+#
+# ==================================================================
+# Statements that are not pushed down
+# ==================================================================
+#
+# EXPLAIN builds the local query plan and does not execute anything
+CALL fill_tables();
+EXPLAIN UPDATE t1, t2 SET t1.b= 'explained' WHERE t1.a= t2.a;
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t2	ALL	NULL	NULL	NULL	NULL	3	
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	4	Using where
+EXPLAIN DELETE t1 FROM t1 JOIN t2 ON t1.a= t2.a;
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t2	ALL	NULL	NULL	NULL	NULL	3	
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	4	Using where
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	one
+2	two
+3	three
+4	four
+# ANALYZE needs the local query plan, so the statement is executed
+# locally, row by row
+CALL fill_tables();
+connection slave;
+TRUNCATE TABLE mysql.general_log;
+SET global general_log=1;
+connection master;
+ANALYZE UPDATE t1, t2 SET t1.b= 'analyzed' WHERE t1.a= t2.a AND t1.a= 1;
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	r_rows	filtered	r_filtered	Extra
+1	SIMPLE	t2	ALL	NULL	NULL	NULL	NULL	3	3.00	100.00	33.33	Using where
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	4	4.00	100.00	25.00	Using where
+connection slave;
+SET global general_log=0;
+SELECT argument FROM mysql.general_log WHERE argument LIKE 'update %';
+argument
+UPDATE `t1` SET `b` = 'analyzed' WHERE `a` = 1 AND `b` = 'one' LIMIT 1
+connection master;
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	analyzed
+2	two
+3	three
+4	four
+# IGNORE is not printed by SELECT_LEX::print(), so an IGNORE statement
+# is executed locally and keeps ignoring the duplicate key errors
+CALL fill_tables();
+connection slave;
+TRUNCATE TABLE mysql.general_log;
+SET global general_log=1;
+connection master;
+UPDATE IGNORE t3, t2 SET t3.b= 30 WHERE t3.a= t2.a AND t3.a < 3;
+SELECT * FROM t3 ORDER BY a;
+a	b
+1	10
+2	20
+3	30
+connection slave;
+SET global general_log=0;
+SELECT argument FROM mysql.general_log WHERE argument LIKE 'update %';
+argument
+UPDATE IGNORE `t3` SET `b` = 30 WHERE `a` = 1 AND `b` = 10 LIMIT 1
+UPDATE IGNORE `t3` SET `b` = 30 WHERE `a` = 2 AND `b` = 20 LIMIT 1
+# A statement that mixes engines has no FederatedX-only table list
+connection master;
+CREATE TABLE t4 (a int, d varchar(16)) ENGINE=MyISAM;
+INSERT INTO t4 VALUES (1,'x'),(2,'y');
+CALL fill_tables();
+UPDATE t1, t4 SET t1.b= t4.d WHERE t1.a= t4.a;
+affected rows: 2
+info: Rows matched: 2  Changed: 2  Warnings: 0
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	x
+2	y
+3	three
+4	four
+DROP TABLE t4;
+# A single-table UPDATE/DELETE does not go through the select handler
+# at all, it is pushed down by direct_update_rows()/direct_delete_rows()
+CALL fill_tables();
+UPDATE t1 SET b= 'single' WHERE a= 1;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Warnings: 0
+DELETE FROM t1 WHERE a= 4;
+affected rows: 1
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	single
+2	two
+3	three
+#
+# ==================================================================
+# Without pushdown the same statements give the same results
+# ==================================================================
+#
+connection default;
+set global federated_pushdown=0;
+connection master;
+CALL fill_tables();
+connection slave;
+TRUNCATE TABLE mysql.general_log;
+SET global general_log=1;
+connection master;
+UPDATE t1, t2 SET t1.b= 'updated' WHERE t1.a= t2.a AND t1.a < 3;
+affected rows: 2
+info: Rows matched: 2  Changed: 2  Warnings: 0
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	updated
+2	updated
+3	three
+4	four
+connection slave;
+SET global general_log=0;
+SELECT argument FROM mysql.general_log WHERE argument LIKE 'update %';
+argument
+UPDATE `t1` SET `b` = 'updated' WHERE `a` = 1 AND `b` = 'one' LIMIT 1
+UPDATE `t1` SET `b` = 'updated' WHERE `a` = 2 AND `b` = 'two' LIMIT 1
+connection master;
+CALL fill_tables();
+UPDATE t1, t2 SET t1.b= 'two' WHERE t1.a= t2.a AND t1.a < 3;
+affected rows: 1
+info: Rows matched: 2  Changed: 1  Warnings: 0
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	two
+2	two
+3	three
+4	four
+CALL fill_tables();
+UPDATE t1, t2 SET t1.b= 'upd1', t2.c= 'upd2' WHERE t1.a= t2.a;
+affected rows: 6
+info: Rows matched: 6  Changed: 6  Warnings: 0
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	upd1
+2	upd1
+3	upd1
+4	four
+SELECT * FROM t2 ORDER BY a;
+a	c
+1	upd2
+2	upd2
+3	upd2
+CALL fill_tables();
+UPDATE t1, t2 SET t1.b= 'nope' WHERE t1.a= t2.a AND t1.a > 100;
+affected rows: 0
+info: Rows matched: 0  Changed: 0  Warnings: 0
+DELETE t1 FROM t1 JOIN t2 ON t1.a= t2.a WHERE t1.a > 100;
+affected rows: 0
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	one
+2	two
+3	three
+4	four
+CALL fill_tables();
+DELETE t1 FROM t1 JOIN t2 ON t1.a= t2.a WHERE t2.c <> 'II';
+affected rows: 2
+SELECT * FROM t1 ORDER BY a;
+a	b
+2	two
+4	four
+CALL fill_tables();
+DELETE t1, t2 FROM t1 JOIN t2 ON t1.a= t2.a WHERE t1.a= 2;
+affected rows: 2
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	one
+3	three
+4	four
+SELECT * FROM t2 ORDER BY a;
+a	c
+1	I
+3	III
+CALL fill_tables();
+DELETE FROM x USING t1 AS x JOIN t2 AS y ON x.a= y.a WHERE y.a= 2;
+affected rows: 1
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	one
+3	three
+4	four
+DROP PROCEDURE fill_tables;
+DROP TABLE federated.t1, federated.t2, federated.t3;
+connection slave;
+TRUNCATE TABLE mysql.general_log;
+SET global general_log= @old_general_log;
+SET global log_output= @old_log_output;
+DROP TABLE federated.t1, federated.t2, federated.t3;
+connection default;
+connection master;
+DROP TABLE IF EXISTS federated.t1;
+DROP DATABASE IF EXISTS federated;
+connection slave;
+DROP TABLE IF EXISTS federated.t1;
+DROP DATABASE IF EXISTS federated;
+# End of 13.2 tests
diff --git a/mysql-test/suite/federated/federatedx_pushdown_upd_del.test b/mysql-test/suite/federated/federatedx_pushdown_upd_del.test
new file mode 100644
index 0000000000000..ed8f642c7ae9d
--- /dev/null
+++ b/mysql-test/suite/federated/federatedx_pushdown_upd_del.test
@@ -0,0 +1,338 @@
+#
+# Pushdown of multi-table UPDATE/DELETE into the FederatedX engine.
+#
+# There is no "primary" handler object for a multi-table UPDATE/DELETE, so
+# handler::direct_update_rows_init()/direct_delete_rows_init() cannot be used
+# for them. Instead the whole statement is pushed down through the
+# select_handler interface: create_federatedx_select_handler() builds a
+# handler of type select_handler_type::UPDATE_DELETE, which prints the whole
+# statement back and executes it on the remote server in
+# ha_federatedx_select_handler::batch_update_delete().
+#
+# The remote server's general log is used throughout to tell a pushed down
+# statement (a single statement, printed by SELECT_LEX::print(), so lowercase)
+# from the ordinary row-by-row execution (several "UPDATE ... LIMIT 1"
+# statements sent by the ha_federatedx handler methods).
+#
+--source have_federatedx.inc
+--source include/federated.inc
+--source include/no_valgrind_without_big.inc
+
+# FederatedX select handler doesn't support views
+--source include/no_view_protocol.inc
+
+connection default;
+ALTER DATABASE federated CHARACTER SET latin1;
+set global federated_pushdown=1;
+
+connection slave;
+DROP TABLE IF EXISTS federated.t1, federated.t2, federated.t3;
+CREATE TABLE federated.t1 (a int, b varchar(16)) DEFAULT CHARSET=latin1;
+CREATE TABLE federated.t2 (a int, c varchar(16)) DEFAULT CHARSET=latin1;
+CREATE TABLE federated.t3 (a int, b int, UNIQUE KEY(b));
+SET @old_log_output= @@global.log_output;
+SET @old_general_log= @@global.general_log;
+SET global log_output='TABLE';
+
+connection master;
+DROP TABLE IF EXISTS federated.t1, federated.t2, federated.t3;
+--replace_result $SLAVE_MYPORT SLAVE_PORT
+eval
+CREATE TABLE federated.t1 (a int, b varchar(16))
+ENGINE="FEDERATED" DEFAULT CHARSET=latin1
+CONNECTION='mysql://root@127.0.0.1:$SLAVE_MYPORT/federated/t1';
+
+--replace_result $SLAVE_MYPORT SLAVE_PORT
+eval
+CREATE TABLE federated.t2 (a int, c varchar(16))
+ENGINE="FEDERATED" DEFAULT CHARSET=latin1
+CONNECTION='mysql://root@127.0.0.1:$SLAVE_MYPORT/federated/t2';
+
+--replace_result $SLAVE_MYPORT SLAVE_PORT
+eval
+CREATE TABLE federated.t3 (a int, b int, UNIQUE KEY(b))
+ENGINE="FEDERATED"
+CONNECTION='mysql://root@127.0.0.1:$SLAVE_MYPORT/federated/t3';
+
+USE federated;
+
+delimiter |;
+CREATE PROCEDURE fill_tables()
+BEGIN
+  DELETE FROM t1;
+  DELETE FROM t2;
+  DELETE FROM t3;
+  INSERT INTO t1 VALUES (1,'one'),(2,'two'),(3,'three'),(4,'four');
+  INSERT INTO t2 VALUES (1,'I'),(2,'II'),(3,'III');
+  INSERT INTO t3 VALUES (1,10),(2,20),(3,30);
+END|
+delimiter ;|
+
+--echo #
+--echo # ==================================================================
+--echo # The statement the remote server receives
+--echo # ==================================================================
+--echo #
+--echo # A pushed down statement is printed back by SELECT_LEX::print() with
+--echo # QT_SELECT_ONLY cleared, which produces
+--echo #   update <join> set <assignments> where <cond>
+--echo # for an UPDATE and
+--echo #   delete from <targets> using <join> where <cond>
+--echo # for a DELETE. Both are understood by the remote server.
+--echo #
+
+connection master;
+CALL fill_tables();
+
+connection slave;
+TRUNCATE TABLE mysql.general_log;
+SET global general_log=1;
+
+connection master;
+UPDATE t1, t2 SET t1.b= 'updated' WHERE t1.a= t2.a AND t1.a < 3;
+UPDATE t1 JOIN t2 ON t1.a= t2.a SET t1.b= CONCAT(t2.c,'!') WHERE t2.a > 2;
+UPDATE t1 AS x JOIN t2 AS y ON x.a= y.a SET x.b= y.c WHERE y.a= 1;
+UPDATE t1, t2 SET t1.b= 'both', t2.c= 'both' WHERE t1.a= t2.a;
+UPDATE t1, t2 SET t1.b= 'sub' WHERE t1.a= t2.a AND
+                                    t1.a IN (SELECT a FROM t2 WHERE c= 'both');
+DELETE t1 FROM t1 JOIN t2 ON t1.a= t2.a WHERE t1.a= 3;
+DELETE FROM x USING t1 AS x JOIN t2 AS y ON x.a= y.a WHERE y.a= 2;
+DELETE t1, t2 FROM t1 JOIN t2 ON t1.a= t2.a;
+
+connection slave;
+SET global general_log=0;
+SELECT argument FROM mysql.general_log
+  WHERE argument LIKE 'update %' OR argument LIKE 'delete %';
+
+--echo #
+--echo # ==================================================================
+--echo # Row counts reported by batch_update_delete()
+--echo # ==================================================================
+--echo #
+
+connection master;
+--echo # All matched rows are changed
+CALL fill_tables();
+--enable_info
+UPDATE t1, t2 SET t1.b= 'updated' WHERE t1.a= t2.a AND t1.a < 3;
+--disable_info
+SELECT * FROM t1 ORDER BY a;
+
+--echo # A matched row that keeps its value is counted as matched but not as
+--echo # changed. The count comes from federatedx_io_mysql::matched_rows(),
+--echo # which parses it out of the remote server's info string
+CALL fill_tables();
+--enable_info
+UPDATE t1, t2 SET t1.b= 'two' WHERE t1.a= t2.a AND t1.a < 3;
+--disable_info
+SELECT * FROM t1 ORDER BY a;
+
+--echo # Two tables updated at once
+CALL fill_tables();
+--enable_info
+UPDATE t1, t2 SET t1.b= 'upd1', t2.c= 'upd2' WHERE t1.a= t2.a;
+--disable_info
+SELECT * FROM t1 ORDER BY a;
+SELECT * FROM t2 ORDER BY a;
+
+--echo # No row matches
+CALL fill_tables();
+--enable_info
+UPDATE t1, t2 SET t1.b= 'nope' WHERE t1.a= t2.a AND t1.a > 100;
+DELETE t1 FROM t1 JOIN t2 ON t1.a= t2.a WHERE t1.a > 100;
+--disable_info
+SELECT * FROM t1 ORDER BY a;
+
+--echo # For a DELETE the remote server reports no matched count, so
+--echo # matched_rows() falls back to the number of deleted rows
+CALL fill_tables();
+--enable_info
+DELETE t1 FROM t1 JOIN t2 ON t1.a= t2.a WHERE t2.c <> 'II';
+--disable_info
+SELECT * FROM t1 ORDER BY a;
+
+--echo # Two tables deleted from at once
+CALL fill_tables();
+--enable_info
+DELETE t1, t2 FROM t1 JOIN t2 ON t1.a= t2.a WHERE t1.a= 2;
+--disable_info
+SELECT * FROM t1 ORDER BY a;
+SELECT * FROM t2 ORDER BY a;
+
+--echo # Aliased delete target
+CALL fill_tables();
+--enable_info
+DELETE FROM x USING t1 AS x JOIN t2 AS y ON x.a= y.a WHERE y.a= 2;
+--disable_info
+SELECT * FROM t1 ORDER BY a;
+
+--echo #
+--echo # ==================================================================
+--echo # An error on the remote server is reported to the client
+--echo # ==================================================================
+--echo #
+CALL fill_tables();
+--error ER_QUERY_ON_FOREIGN_DATA_SOURCE
+UPDATE t1, t2 SET t1.a= 1/0 WHERE t1.a= t2.a;
+--error ER_QUERY_ON_FOREIGN_DATA_SOURCE
+UPDATE t3, t2 SET t3.b= 30 WHERE t3.a= t2.a AND t3.a < 3;
+SELECT * FROM t3 ORDER BY a;
+
+--echo #
+--echo # ==================================================================
+--echo # Inside a transaction the statement is not auto-committed by itself
+--echo # ==================================================================
+--echo #
+CALL fill_tables();
+BEGIN;
+--enable_info
+UPDATE t1, t2 SET t1.b= 'in trx' WHERE t1.a= t2.a;
+DELETE t1 FROM t1 JOIN t2 ON t1.a= t2.a WHERE t1.a= 3;
+--disable_info
+COMMIT;
+SELECT * FROM t1 ORDER BY a;
+
+--echo #
+--echo # ==================================================================
+--echo # Statements that are not pushed down
+--echo # ==================================================================
+--echo #
+
+--echo # EXPLAIN builds the local query plan and does not execute anything
+CALL fill_tables();
+EXPLAIN UPDATE t1, t2 SET t1.b= 'explained' WHERE t1.a= t2.a;
+EXPLAIN DELETE t1 FROM t1 JOIN t2 ON t1.a= t2.a;
+SELECT * FROM t1 ORDER BY a;
+
+--echo # ANALYZE needs the local query plan, so the statement is executed
+--echo # locally, row by row
+CALL fill_tables();
+
+connection slave;
+TRUNCATE TABLE mysql.general_log;
+SET global general_log=1;
+
+connection master;
+ANALYZE UPDATE t1, t2 SET t1.b= 'analyzed' WHERE t1.a= t2.a AND t1.a= 1;
+
+connection slave;
+SET global general_log=0;
+SELECT argument FROM mysql.general_log WHERE argument LIKE 'update %';
+
+connection master;
+SELECT * FROM t1 ORDER BY a;
+
+--echo # IGNORE is not printed by SELECT_LEX::print(), so an IGNORE statement
+--echo # is executed locally and keeps ignoring the duplicate key errors
+CALL fill_tables();
+
+connection slave;
+TRUNCATE TABLE mysql.general_log;
+SET global general_log=1;
+
+connection master;
+UPDATE IGNORE t3, t2 SET t3.b= 30 WHERE t3.a= t2.a AND t3.a < 3;
+SELECT * FROM t3 ORDER BY a;
+
+connection slave;
+SET global general_log=0;
+SELECT argument FROM mysql.general_log WHERE argument LIKE 'update %';
+
+--echo # A statement that mixes engines has no FederatedX-only table list
+connection master;
+CREATE TABLE t4 (a int, d varchar(16)) ENGINE=MyISAM;
+INSERT INTO t4 VALUES (1,'x'),(2,'y');
+CALL fill_tables();
+--enable_info
+UPDATE t1, t4 SET t1.b= t4.d WHERE t1.a= t4.a;
+--disable_info
+SELECT * FROM t1 ORDER BY a;
+DROP TABLE t4;
+
+--echo # A single-table UPDATE/DELETE does not go through the select handler
+--echo # at all, it is pushed down by direct_update_rows()/direct_delete_rows()
+CALL fill_tables();
+--enable_info
+UPDATE t1 SET b= 'single' WHERE a= 1;
+DELETE FROM t1 WHERE a= 4;
+--disable_info
+SELECT * FROM t1 ORDER BY a;
+
+--echo #
+--echo # ==================================================================
+--echo # Without pushdown the same statements give the same results
+--echo # ==================================================================
+--echo #
+connection default;
+set global federated_pushdown=0;
+
+connection master;
+CALL fill_tables();
+
+connection slave;
+TRUNCATE TABLE mysql.general_log;
+SET global general_log=1;
+
+connection master;
+--enable_info
+UPDATE t1, t2 SET t1.b= 'updated' WHERE t1.a= t2.a AND t1.a < 3;
+--disable_info
+SELECT * FROM t1 ORDER BY a;
+
+connection slave;
+SET global general_log=0;
+SELECT argument FROM mysql.general_log WHERE argument LIKE 'update %';
+
+connection master;
+CALL fill_tables();
+--enable_info
+UPDATE t1, t2 SET t1.b= 'two' WHERE t1.a= t2.a AND t1.a < 3;
+--disable_info
+SELECT * FROM t1 ORDER BY a;
+
+CALL fill_tables();
+--enable_info
+UPDATE t1, t2 SET t1.b= 'upd1', t2.c= 'upd2' WHERE t1.a= t2.a;
+--disable_info
+SELECT * FROM t1 ORDER BY a;
+SELECT * FROM t2 ORDER BY a;
+
+CALL fill_tables();
+--enable_info
+UPDATE t1, t2 SET t1.b= 'nope' WHERE t1.a= t2.a AND t1.a > 100;
+DELETE t1 FROM t1 JOIN t2 ON t1.a= t2.a WHERE t1.a > 100;
+--disable_info
+SELECT * FROM t1 ORDER BY a;
+
+CALL fill_tables();
+--enable_info
+DELETE t1 FROM t1 JOIN t2 ON t1.a= t2.a WHERE t2.c <> 'II';
+--disable_info
+SELECT * FROM t1 ORDER BY a;
+
+CALL fill_tables();
+--enable_info
+DELETE t1, t2 FROM t1 JOIN t2 ON t1.a= t2.a WHERE t1.a= 2;
+--disable_info
+SELECT * FROM t1 ORDER BY a;
+SELECT * FROM t2 ORDER BY a;
+
+CALL fill_tables();
+--enable_info
+DELETE FROM x USING t1 AS x JOIN t2 AS y ON x.a= y.a WHERE y.a= 2;
+--disable_info
+SELECT * FROM t1 ORDER BY a;
+
+DROP PROCEDURE fill_tables;
+DROP TABLE federated.t1, federated.t2, federated.t3;
+
+connection slave;
+TRUNCATE TABLE mysql.general_log;
+SET global general_log= @old_general_log;
+SET global log_output= @old_log_output;
+DROP TABLE federated.t1, federated.t2, federated.t3;
+
+connection default;
+source include/federated_cleanup.inc;
+
+--echo # End of 13.2 tests
diff --git a/sql/select_handler.cc b/sql/select_handler.cc
index ab2a6850a6e70..99785cf0fe6a0 100644
--- a/sql/select_handler.cc
+++ b/sql/select_handler.cc
@@ -38,24 +38,27 @@
 
 
 select_handler::select_handler(THD *thd_arg, handlerton *ht_arg,
-                               SELECT_LEX *sel_lex)
+                               SELECT_LEX *sel_lex,
+                               select_handler_type type_arg)
   : select_lex(sel_lex), lex_unit(nullptr), table(nullptr),
     thd(thd_arg), ht(ht_arg), result(sel_lex->join->result),
-    is_analyze(thd_arg->lex->analyze_stmt)
+    is_analyze(thd_arg->lex->analyze_stmt), handler_type(type_arg)
 {}
 
 select_handler::select_handler(THD *thd_arg, handlerton *ht_arg,
                                SELECT_LEX_UNIT *sel_unit)
   : select_lex(nullptr), lex_unit(sel_unit), table(nullptr),
     thd(thd_arg), ht(ht_arg), result(sel_unit->result),
-    is_analyze(thd_arg->lex->analyze_stmt)
+    is_analyze(thd_arg->lex->analyze_stmt),
+    handler_type(select_handler_type::SELECT)
 {}
 
 select_handler::select_handler(THD *thd_arg, handlerton *ht_arg,
                                SELECT_LEX *sel_lex, SELECT_LEX_UNIT *sel_unit)
     : select_lex(sel_lex), lex_unit(sel_unit), table(nullptr), thd(thd_arg),
       ht(ht_arg), result(sel_lex->join->result),
-      is_analyze(thd_arg->lex->analyze_stmt)
+      is_analyze(thd_arg->lex->analyze_stmt),
+      handler_type(select_handler_type::SELECT)
 {}
 
 select_handler::~select_handler()
@@ -106,6 +109,12 @@ TABLE *select_handler::create_tmp_table(THD *thd)
 bool select_handler::prepare()
 {
   DBUG_ENTER("select_handler::prepare");
+  /*
+    A multi-table UPDATE/DELETE handler produces no rows, so it needs
+    neither a record buffer nor a list of result columns.
+  */
+  if (is_update_delete_handler())
+    DBUG_RETURN(false);
   /*
     Some engines (e.g. XPand) initialize "table" on their own.
     So we need to create a temporary table only if "table" is NULL.
@@ -152,10 +161,66 @@ bool select_handler::send_eof()
 
 
 int select_handler::execute()
+{
+  DBUG_ENTER("select_handler::execute");
+  DBUG_RETURN(is_update_delete_handler() ? execute_update_delete() :
+                                           execute_select());
+}
+
+
+/*
+  Default implementation of the batch_update_delete() interface function.
+  An engine that creates a select_handler_type::UPDATE_DELETE handler must
+  provide its own.
+*/
+
+int select_handler::batch_update_delete(ha_rows *found_rows,
+                                        ha_rows *affected_rows)
+{
+  DBUG_ASSERT(0);
+  return HA_ERR_WRONG_COMMAND;
+}
+
+
+/*
+  Execute a multi-table UPDATE/DELETE that was pushed down into the engine.
+
+  The engine does the whole job and only reports the row counts. The SQL
+  layer still has to do the final part of the statement: invalidate the
+  query cache, write the statement to the binary log and send the OK packet
+  to the client. All of that is done by multi_update::send_eof() /
+  multi_delete::send_eof(), so the counters are passed to the result object
+  and send_eof() is called as if the rows had been updated locally.
+*/
+
+int select_handler::execute_update_delete()
 {
   int err;
+  ha_rows found_rows= 0, affected_rows= 0;
 
-  DBUG_ENTER("select_handler::execute");
+  DBUG_ENTER("select_handler::execute_update_delete");
+
+  if ((err= batch_update_delete(&found_rows, &affected_rows)))
+  {
+    if (!thd->is_error())
+      print_error(err, MYF(0));
+    DBUG_RETURN(-1);
+  }
+
+  result->direct_update_delete_done(found_rows, affected_rows);
+
+  if (send_eof())
+    DBUG_RETURN(-1);
+
+  DBUG_RETURN(0);
+}
+
+
+int select_handler::execute_select()
+{
+  int err;
+
+  DBUG_ENTER("select_handler::execute_select");
 
   if ((err= init_scan()))
     goto error;
diff --git a/sql/select_handler.h b/sql/select_handler.h
index 207af43f56ca0..eb0e69b3bbcd1 100644
--- a/sql/select_handler.h
+++ b/sql/select_handler.h
@@ -26,6 +26,29 @@ enum class select_pushdown_type {
   WHOLE_UNIT
 };
 
+
+/*
+  What kind of statement a select_handler object takes over.
+
+  SELECT
+    The handler produces a result set. The SQL layer retrieves the rows
+    one by one with init_scan()/next_row()/end_scan() and sends them to
+    the client.
+
+  UPDATE_DELETE
+    The handler performs a multi-table UPDATE or DELETE. There is no
+    result set and no "primary" handler object to call
+    handler::direct_update_rows_init()/direct_delete_rows_init() on, so
+    the whole statement is handed over to the engine at once with
+    batch_update_delete(). init_scan()/next_row()/end_scan() are never
+    called for such a handler.
+*/
+enum class select_handler_type
+{
+  SELECT,
+  UPDATE_DELETE
+};
+
 /**
   @class select_handler
 
@@ -37,7 +60,8 @@ class select_handler
 {
  public:
    // Constructor for a single SELECT_LEX (not a part of a unit)
-  select_handler(THD *thd_arg, handlerton *ht_arg, SELECT_LEX *sel_lex);
+  select_handler(THD *thd_arg, handlerton *ht_arg, SELECT_LEX *sel_lex,
+                 select_handler_type type_arg= select_handler_type::SELECT);
 
   // Constructor for a unit (UNION/EXCEPT/INTERSECT)
   select_handler(THD *thd_arg, handlerton *ht_arg, SELECT_LEX_UNIT *sel_unit);
@@ -55,6 +79,9 @@ class select_handler
 
   virtual bool prepare();
 
+  bool is_update_delete_handler() const
+  { return handler_type == select_handler_type::UPDATE_DELETE; }
+
   /*
     Select_handler processes these cases:
     - single SELECT
@@ -81,6 +108,7 @@ class select_handler
   /*
     Functions to scan the select result set.
     All these returns 0 if ok, error code in case of error.
+    They are called only for select_handler_type::SELECT handlers.
   */
 
   /* Initialize the process of producing rows of result set */
@@ -96,6 +124,20 @@ class select_handler
   /* Finish scanning */
   virtual int end_scan() = 0;
 
+  /*
+    Perform the whole multi-table UPDATE/DELETE statement and report how many
+    rows matched the WHERE clause (*found_rows) and how many rows were really
+    changed or deleted (*affected_rows).
+
+    This is called instead of the init_scan()/next_row()/end_scan() loop for
+    select_handler_type::UPDATE_DELETE handlers, so an engine that creates
+    such a handler must implement it.
+
+    @retval 0        ok
+    @retval != 0     error code
+  */
+  virtual int batch_update_delete(ha_rows *found_rows, ha_rows *affected_rows);
+
   /* Report errors */
   virtual void print_error(int error, myf errflag);
 
@@ -114,6 +156,14 @@ class select_handler
   List<Item> result_columns;
 
   bool is_analyze;
+
+  /* Which kind of statement this handler takes over */
+  select_handler_type handler_type;
+
+private:
+
+  int execute_select();
+  int execute_update_delete();
 };
 
 #endif /* SELECT_HANDLER_INCLUDED */
diff --git a/sql/select_result.h b/sql/select_result.h
index 4bc1050388570..bfe723022498c 100644
--- a/sql/select_result.h
+++ b/sql/select_result.h
@@ -106,6 +106,17 @@ class select_result :public select_result_sink
     @retval TRUE      error, an error message is set
   */
   virtual bool check_simple_select() const;
+  /*
+    A multi-table UPDATE/DELETE was performed by the storage engine itself,
+    see select_handler_type::UPDATE_DELETE. Tell the result object how many
+    rows matched the WHERE clause and how many rows were really changed, so
+    that the following send_eof() reports them instead of running the
+    SQL-layer update/delete loop.
+    Only multi_update and multi_delete can receive this call.
+  */
+  virtual void direct_update_delete_done(ha_rows found_rows,
+                                         ha_rows affected_rows)
+  { DBUG_ASSERT(0); }
   virtual void abort_result_set() {}
   virtual void reset_for_next_ps_execution();
   void set_thd(THD *thd_arg) { thd= thd_arg; }
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 25ffcf49d3bab..02d681a718a5f 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -7832,6 +7832,8 @@ class multi_delete :public select_result_interceptor
      so that afterward abort_result_set() needs to find out that.
   */
   bool error_handled;
+  /* True if the engine has performed the whole DELETE, see select_handler */
+  bool direct_dml_done;
 
 public:
   // Methods used by ColumnStore
@@ -7849,6 +7851,8 @@ class multi_delete :public select_result_interceptor
   int do_table_deletes(TABLE *table, SORT_INFO *sort_info, bool ignore);
   bool send_eof() override;
   inline ha_rows num_deleted() const { return deleted; }
+  void direct_update_delete_done(ha_rows found_rows,
+                                 ha_rows affected_rows) override;
   void abort_result_set() override;
   void prepare_to_read_rows() override;
 };
@@ -7883,7 +7887,9 @@ class multi_update :public select_result_interceptor
      so that afterward  abort_result_set() needs to find out that.
   */
   bool error_handled;
-  
+  /* True if the engine has performed the whole UPDATE, see select_handler */
+  bool direct_dml_done;
+
   /* Need this to protect against multiple prepare() calls */
   bool prepared;
 
@@ -7910,6 +7916,8 @@ class multi_update :public select_result_interceptor
   inline ha_rows num_updated() const { return updated; }
   inline void set_found (ha_rows n) { found= n; }
   inline void set_updated (ha_rows n) { updated= n; }
+  void direct_update_delete_done(ha_rows found_rows,
+                                 ha_rows affected_rows) override;
   virtual void abort_result_set() override;
   void update_used_tables() override;
   void prepare_to_read_rows() override;
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index b6bc6827a44ed..89f71135b0033 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -1174,7 +1174,8 @@ multi_delete::multi_delete(THD *thd_arg,
     do_delete(0),
     transactional_tables(0),
     normal_tables(0),
-    error_handled(0)
+    error_handled(0),
+    direct_dml_done(0)
 {
   tmp_tables = thd->calloc<TABLE*>(table_count);
   tmp_table_param = thd->calloc<TMP_TABLE_PARAM>(table_count);
@@ -1522,7 +1523,7 @@ void multi_delete::abort_result_set()
     The same if all tables are transactional, regardless of where we are.
     In all other cases do attempt deletes ...
   */
-  if (do_delete && normal_tables &&
+  if (do_delete && normal_tables && !direct_dml_done &&
       (table_being_deleted != delete_tables ||
        !table_being_deleted->table->file->has_transactions_and_rollback()))
   {
@@ -1734,6 +1735,42 @@ int multi_delete::rowid_table_deletes(TABLE *table, bool ignore)
 }
 
 
+/*
+  The engine has performed the whole multi-table DELETE on its own, the
+  statement was pushed down through the select_handler interface. Remember
+  the row counts reported by the engine and mark the tables we deleted from,
+  so that send_eof() can binlog the statement and send the OK packet without
+  running the SQL-layer delete loop.
+*/
+
+void multi_delete::direct_update_delete_done(ha_rows found_rows,
+                                             ha_rows affected_rows)
+{
+  DBUG_ENTER("multi_delete::direct_update_delete_done");
+
+  found= found_rows;
+  deleted= affected_rows;
+  direct_dml_done= true;
+
+  for (TABLE_LIST *walk= delete_tables; walk; walk= walk->next_local)
+  {
+    TABLE_LIST *tbl= walk->table ? walk :
+                     walk->correspondent_table->find_table_for_update();
+    if (!tbl || !tbl->table)
+      continue;
+    if (tbl->table->file->has_transactions())
+      transactional_tables= 1;
+    else
+    {
+      normal_tables= 1;
+      if (deleted)
+        thd->transaction->stmt.modified_non_trans_table= TRUE;
+    }
+  }
+  DBUG_VOID_RETURN;
+}
+
+
 /*
   Send ok to the client
 
@@ -1747,7 +1784,7 @@ bool multi_delete::send_eof()
   THD_STAGE_INFO(thd, stage_deleting_from_reference_tables);
 
   /* Does deletes for the last n - 1 tables, returns 0 if ok */
-  int local_error= do_deletes();		// returns 0 if success
+  int local_error= direct_dml_done ? 0 : do_deletes(); // returns 0 if success
 
   /* compute a total error to know if something failed */
   local_error= local_error || error;
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 948f7b3744d16..8390d3add7baf 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -32347,6 +32347,16 @@ enum explainable_cmd_type get_explainable_cmd_type(THD *thd)
 }
 
 
+/*
+  Print a table of the target list of a multi-table DELETE, i.e. of the list
+  of the tables the rows are deleted from.
+
+  The grammar accepts only a (possibly qualified) table name or an alias
+  there, and not a table reference with an alias like the USING clause does.
+  So a table that is given an alias in the USING clause has to be referred to
+  by this alias alone, otherwise the printed statement cannot be parsed back.
+*/
+
 void TABLE_LIST::print_leaf_tables(THD *thd, String *str,
                               enum_query_type query_type)
 {
@@ -32354,9 +32364,17 @@ void TABLE_LIST::print_leaf_tables(THD *thd, String *str,
   {
     for (TABLE_LIST *tbl= merge_underlying_list; tbl; tbl= tbl->next_local)
       tbl->print_leaf_tables(thd, str, query_type);
+    return;
   }
-  else
-    print(thd, 0, str, query_type);
+
+  const Lex_ident_table name= view_name.str ? Lex_ident_table(view_name)
+                                            : table_name;
+  if (!name.streq(alias))
+  {
+    append_identifier_opt_casedn(thd, str, alias, lower_case_table_names == 1);
+    return;
+  }
+  print(thd, 0, str, query_type);
 }
 
 
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 847d073b48d78..c33888747ee0a 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -1846,6 +1846,7 @@ multi_update::multi_update(THD *thd_arg, TABLE_LIST *table_list,
     transactional_tables(0),
     ignore(ignore_arg),
     error_handled(0),
+    direct_dml_done(0),
     prepared(0),
     updated_sys_ver(0),
     tables_to_update(get_table_map(fields))
@@ -2614,7 +2615,7 @@ void multi_update::abort_result_set()
   if (! trans_safe)
   {
     DBUG_ASSERT(thd->transaction->stmt.modified_non_trans_table);
-    if (do_update && table_count > 1)
+    if (do_update && table_count > 1 && !direct_dml_done)
     {
       /* Add warning here */
       (void) do_updates();
@@ -2930,6 +2931,41 @@ int multi_update::do_updates()
 }
 
 
+/*
+  The engine has performed the whole multi-table UPDATE on its own, the
+  statement was pushed down through the select_handler interface. Remember
+  the row counts reported by the engine and mark the updated tables as
+  modified, so that send_eof() can binlog the statement and send the OK
+  packet without running the SQL-layer update loop.
+*/
+
+void multi_update::direct_update_delete_done(ha_rows found_rows,
+                                             ha_rows affected_rows)
+{
+  DBUG_ENTER("multi_update::direct_update_delete_done");
+
+  found= found_rows;
+  updated= affected_rows;
+  direct_dml_done= true;
+
+  for (TABLE_LIST *cur_table= update_tables; cur_table;
+       cur_table= cur_table->next_local)
+  {
+    TABLE *table= cur_table->table;
+    if (!table)
+      continue;
+    if (table->file->has_transactions_and_rollback())
+      transactional_tables= TRUE;
+    else if (updated)
+    {
+      trans_safe= FALSE;
+      thd->transaction->stmt.modified_non_trans_table= TRUE;
+    }
+  }
+  DBUG_VOID_RETURN;
+}
+
+
 /* out: 1 if error, 0 if success */
 
 bool multi_update::send_eof()
@@ -2940,13 +2976,13 @@ bool multi_update::send_eof()
   DBUG_ENTER("multi_update::send_eof");
   THD_STAGE_INFO(thd, stage_updating_reference_tables);
 
-  /* 
+  /*
      Does updates for the last n - 1 tables, returns 0 if ok;
      error takes into account killed status gained in do_updates()
   */
   int local_error= thd->is_error();
   if (likely(!local_error))
-    local_error = (table_count) ? do_updates() : 0;
+    local_error= (table_count && !direct_dml_done) ? do_updates() : 0;
   /*
     if local_error is not set ON until after do_updates() then
     later carried out killing should not affect binlogging.
diff --git a/storage/federatedx/federatedx_io_mysql.cc b/storage/federatedx/federatedx_io_mysql.cc
index a7f1e5135b82c..18bdac3ef4e34 100644
--- a/storage/federatedx/federatedx_io_mysql.cc
+++ b/storage/federatedx/federatedx_io_mysql.cc
@@ -78,6 +78,7 @@ class federatedx_io_mysql :public federatedx_io
   size_t max_query_size() const override;
 
   my_ulonglong affected_rows() const override;
+  my_ulonglong matched_rows() const override;
   my_ulonglong last_insert_id() const override;
 
   int error_code() override;
@@ -468,6 +469,26 @@ my_ulonglong federatedx_io_mysql::affected_rows() const
 }
 
 
+my_ulonglong federatedx_io_mysql::matched_rows() const
+{
+  /*
+    For an UPDATE the server returns an info string of the form
+      "Rows matched: N  Changed: M  Warnings: W"
+    Parse the "matched" count out of it. For statements that do not report
+    matched rows (e.g. DELETE) mysql_info() is NULL, fall back to the number
+    of affected rows.
+  */
+  const char *matched;
+  if (mysql.info && (matched= strstr(mysql.info, "matched: ")))
+  {
+    int error= 0;
+    return (my_ulonglong) my_strtoll10(matched + strlen("matched: "),
+                                       NULL, &error);
+  }
+  return mysql.affected_rows;
+}
+
+
 my_ulonglong federatedx_io_mysql::last_insert_id() const
 {
   return mysql.insert_id;
diff --git a/storage/federatedx/federatedx_io_null.cc b/storage/federatedx/federatedx_io_null.cc
index 5c62ea6ab5271..f15c16a757dcf 100644
--- a/storage/federatedx/federatedx_io_null.cc
+++ b/storage/federatedx/federatedx_io_null.cc
@@ -59,6 +59,7 @@ class federatedx_io_null :public federatedx_io
   size_t max_query_size() const override;
 
   my_ulonglong affected_rows() const override;
+  my_ulonglong matched_rows() const override;
   my_ulonglong last_insert_id() const override;
 
   int error_code() override;
@@ -190,6 +191,12 @@ my_ulonglong federatedx_io_null::affected_rows() const
 }
 
 
+my_ulonglong federatedx_io_null::matched_rows() const
+{
+  return 0;
+}
+
+
 my_ulonglong federatedx_io_null::last_insert_id() const
 {
   return 0;
diff --git a/storage/federatedx/federatedx_pushdown.cc b/storage/federatedx/federatedx_pushdown.cc
index 3fc7423eee9a0..88d3502df82b1 100644
--- a/storage/federatedx/federatedx_pushdown.cc
+++ b/storage/federatedx/federatedx_pushdown.cc
@@ -212,17 +212,37 @@ static bool is_supported_by_select_handler(enum_sql_command sql_command)
 }
 
 
+/*
+  Check whether this is a multi-table UPDATE/DELETE.
+
+  The SQL layer looks for a select_handler for an UPDATE/DELETE only when the
+  statement is processed as a multi-table one (see Sql_cmd_dml::execute_inner()
+  and Sql_cmd_update::execute_inner()/Sql_cmd_delete::execute_inner()).
+  Single-table UPDATE/DELETE never reach this point, they are pushed into the
+  engine with handler::direct_update_rows()/direct_delete_rows() instead.
+*/
+static bool is_supported_update_delete(enum_sql_command sql_command)
+{
+  return sql_command == SQLCOM_UPDATE || sql_command == SQLCOM_UPDATE_MULTI ||
+         sql_command == SQLCOM_DELETE || sql_command == SQLCOM_DELETE_MULTI;
+}
+
+
 /*
   Create FederatedX select handler for processing either a single select
   (in this case sel_lex is initialized and lex_unit==NULL)
   or a select that is part of a unit
   (in this case both sel_lex and lex_unit are initialized)
+  or a whole multi-table UPDATE/DELETE statement
 */
 static select_handler *
 create_federatedx_select_handler(THD *thd, SELECT_LEX *sel_lex,
                                  SELECT_LEX_UNIT *lex_unit)
 {
-  if (!use_pushdown || !is_supported_by_select_handler(thd->lex->sql_command))
+  const bool is_upd_del= is_supported_update_delete(thd->lex->sql_command);
+
+  if (!use_pushdown ||
+      !(is_upd_del || is_supported_by_select_handler(thd->lex->sql_command)))
     return nullptr;
 
   if (lex_unit && sel_lex->master_unit()->with_clause)
@@ -235,6 +255,28 @@ create_federatedx_select_handler(THD *thd, SELECT_LEX *sel_lex,
   if (sel_lex->uncacheable & UNCACHEABLE_SIDEEFFECT)
     return NULL;
 
+  if (is_upd_del)
+  {
+    /*
+      For EXPLAIN the statement is not executed at all, and ANALYZE would need
+      the local query plan that is never built for a pushed down statement.
+    */
+    if (thd->lex->describe || thd->lex->analyze_stmt)
+      return nullptr;
+
+    /*
+      SELECT_LEX::print() does not print the IGNORE modifier, so a pushed down
+      UPDATE IGNORE/DELETE IGNORE would turn an ignored error on the remote
+      server into a real one. Let such statements be executed locally.
+    */
+    if (thd->lex->ignore)
+      return nullptr;
+
+    DBUG_ASSERT(!lex_unit);
+    return new ha_federatedx_select_handler(
+        thd, sel_lex, tbl, select_handler_type::UPDATE_DELETE);
+  }
+
   return new ha_federatedx_select_handler(thd, sel_lex, lex_unit, tbl);
 }
 
@@ -315,6 +357,77 @@ ha_federatedx_select_handler::ha_federatedx_select_handler(
   }
 }
 
+ha_federatedx_select_handler::ha_federatedx_select_handler(
+    THD *thd, SELECT_LEX *select_lex, TABLE *tbl, select_handler_type type_arg)
+    : select_handler(thd, federatedx_hton, select_lex, type_arg),
+      federatedx_handler_base(thd, tbl)
+{
+  DBUG_ASSERT(type_arg == select_handler_type::UPDATE_DELETE);
+  query.length(0);
+  /*
+    Print the whole statement back. SELECT_LEX::print() produces
+      update <join> set <assignments> where <cond>
+    for an UPDATE and
+      delete from <targets> using <join> where <cond>
+    for a DELETE, both of which the remote server understands
+  */
+  select_lex->print(thd, &query, PRINT_UPD_DEL_QUERY_TYPE);
+}
+
+
+/*
+  Execute a multi-table UPDATE/DELETE on the remote server and report
+  how many rows it has changed
+*/
+
+int ha_federatedx_select_handler::batch_update_delete(ha_rows *found_rows,
+                                                      ha_rows *affected_rows)
+{
+  THD *thd= query_table->in_use;
+  int rc;
+  DBUG_ENTER("ha_federatedx_select_handler::batch_update_delete");
+
+  ha_federatedx *h= (ha_federatedx *) query_table->file;
+  iop= &h->io;
+  share= get_share(query_table->s->table_name.str, query_table,
+                   h->option_struct);
+  txn= h->get_txn(thd);
+
+  /* no need for savepoint in autocommit mode */
+  if (!(thd->variables.option_bits & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)))
+    txn->stmt_autocommit();
+
+  if ((rc= txn->acquire(share, thd, FALSE, iop)))
+  {
+    free_share(txn, share);
+    share= NULL;
+    DBUG_RETURN(rc);
+  }
+
+  if ((*iop)->query(query.ptr(), query.length()))
+  {
+    my_error(ER_QUERY_ON_FOREIGN_DATA_SOURCE, MYF(0), (*iop)->error_str());
+    rc= HA_FEDERATEDX_ERROR_WITH_REMOTE_SYSTEM;
+  }
+  else
+  {
+    /*
+      For an UPDATE the remote server reports both the number of matched and
+      of changed rows (matched_rows() parses it out of the info string), for a
+      DELETE matched_rows() returns the number of deleted rows just like
+      affected_rows() does.
+    */
+    *affected_rows= (ha_rows) (*iop)->affected_rows();
+    *found_rows= (ha_rows) (*iop)->matched_rows();
+    rc= 0;
+  }
+
+  free_share(txn, share);
+  share= NULL;
+  DBUG_RETURN(rc);
+}
+
+
 int federatedx_handler_base::init_scan_()
 {
   THD *thd= query_table->in_use;
diff --git a/storage/federatedx/federatedx_pushdown.h b/storage/federatedx/federatedx_pushdown.h
index cea1192b9193b..0273db80a3264 100644
--- a/storage/federatedx/federatedx_pushdown.h
+++ b/storage/federatedx/federatedx_pushdown.h
@@ -66,13 +66,29 @@ class ha_federatedx_select_handler: public select_handler, public federatedx_han
                                TABLE *tbl);
   ha_federatedx_select_handler(THD *thd_arg, SELECT_LEX *sel_lex,
                                SELECT_LEX_UNIT *sel_unit, TABLE *tbl);
+  /*
+    Constructor for a multi-table UPDATE/DELETE that is pushed down as
+    a whole. Such a handler is of select_handler_type::UPDATE_DELETE, it
+    produces no rows and is executed by batch_update_delete()
+  */
+  ha_federatedx_select_handler(THD *thd_arg, SELECT_LEX *sel_lex, TABLE *tbl,
+                               select_handler_type type_arg);
   ~ha_federatedx_select_handler();
   int init_scan() override { return federatedx_handler_base::init_scan_(); }
   int next_row() override { return federatedx_handler_base::next_row_(table); }
   int end_scan() override;
+  int batch_update_delete(ha_rows *found_rows, ha_rows *affected_rows) override;
 
 private:
   static constexpr auto PRINT_QUERY_TYPE=
       enum_query_type(QT_VIEW_INTERNAL | QT_SELECT_ONLY |
                       QT_ITEM_ORIGINAL_FUNC_NULLIF | QT_PARSABLE);
+  /*
+    Same as PRINT_QUERY_TYPE, but without QT_SELECT_ONLY, so that
+    SELECT_LEX::print() prints the whole UPDATE/DELETE statement and not
+    just its SELECT part
+  */
+  static constexpr auto PRINT_UPD_DEL_QUERY_TYPE=
+      enum_query_type(QT_VIEW_INTERNAL | QT_ITEM_ORIGINAL_FUNC_NULLIF |
+                      QT_PARSABLE);
 };
diff --git a/storage/federatedx/ha_federatedx.h b/storage/federatedx/ha_federatedx.h
index 7831f394a26b7..88cc519c77808 100644
--- a/storage/federatedx/ha_federatedx.h
+++ b/storage/federatedx/ha_federatedx.h
@@ -177,6 +177,13 @@ class federatedx_io
   virtual size_t max_query_size() const=0;
 
   virtual my_ulonglong affected_rows() const=0;
+  /*
+    Number of rows that matched the WHERE clause of the last statement.
+    For an UPDATE this can be larger than affected_rows() (rows that matched
+    but were left unchanged). When the server does not report it (e.g. for a
+    DELETE) this returns the same value as affected_rows().
+  */
+  virtual my_ulonglong matched_rows() const=0;
   virtual my_ulonglong last_insert_id() const=0;
 
   virtual int error_code()=0;