Merge 10.4 into HEADbb-10.4-MDEV-17520

59 files changed, 2122 insertions, 1054 deletions

diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
new file mode 100644
index 00000000000..64af450d29f
--- /dev/null
+++ b/CONTRIBUTING.md
@@ -0,0 +1,47 @@
+# How to contribute
+
+MariaDB Server has a vibrant community contributing in a wide range of areas. There are many valuable ways you can contribute to MariaDB.
+
+### Engage online with other community members
+---
+-   [MariaDB on Zulip](https://mariadb.zulipchat.com/)
+-   [maria-developers mailing list](http://launchpad.net/~maria-developers)
+-   [maria-discuss mailing list](http://launchpad.net/~maria-discuss)
+-   [maria-docs mailing list](http://launchpad.net/~maria-docs)
+-   ircs://chat.freenode.net/maria ([see the IRC page on the Knowledge Base](https://mariadb.com/kb/en/meta/irc-chat-servers-and-zulip-instance/)  for help with IRC).
+-   The MariaDB Foundation and MariaDB Corporation have a presence on Reddit, Twitter, Facebook and Google Plus. See the  [social media page](https://mariadb.com/kb/en/mariadb/social-media/).
+
+### Help document MariaDB
+----
+-   Contribute towards [documenting MariaDB Server](https://mariadb.com/kb/en/meta/writing-editing-library-articles/)  and its ecosystem by adding new content or improving existing content.
+-   [Translate](https://mariadb.com/kb/en/meta/translating-library-articles/)  existing documentation.
+
+### Help debug and develop MariaDB
+-----
+-   [Report bugs](https://jira.mariadb.org/)
+-   Test development versions
+-   Write code to fix bugs or develop new features (see  [Getting Started for Developers](https://mariadb.org/getting-started-for-developers)).See also [list of beginner friendly tasks](https://jira.mariadb.org/browse/MDEV-15736?jql=resolution%20%3D%20Unresolved%20AND%20labels%20%3D%20beginner-friendly%20ORDER%20BY%20updated%20DESC)
+-   Help with code quality control
+-   Participate in packaging for different Linux distributions
+
+### Sponsor or donate
+---
+You’re very welcome to support MariaDB Server as an individual, or talk your company into joining the Foundation as a sponsoring member. See the  [Sponsor page](https://mariadb.org/donate/).
+
+### Events, meetups and conferences
+---
+-   Attend an event
+    -   [Events and Conferences page](https://mariadb.org/events/)
+    -   [mariadb.meetup.com](http://mariadb.meetup.com/)
+
+### Live QA for beginner contributors
+----
+MariaDB has a dedicated time each week when we answer new contributor questions live on Zulip and IRC.
+From 8:00 to 10:00 UTC on Mondays, and 10:00 to 12:00 UTC on Thursdays, anyone can ask any questions they’d like,
+and a live developer will be available to assist.
+New contributors can ask questions any time, but we will provide immediate feedback during that interval.
+
+### Additional resources
+----
+ - [MariaDB Foundation ](https://mariadb.org/)
+ - [Knowledge Base](https://mariadb.com/kb/en/)
diff --git a/README.md b/README.md
index 7cdcc90471f..932ad768d62 100644
--- a/README.md
+++ b/README.md
@@ -29,22 +29,27 @@ https://mariadb.com/kb/en/
 
 https://mariadb.com/kb/en/mariadb-vs-mysql-features/
 
-https://mariadb.com/kb/en/mariadb-versus-mysql-features/
-
 https://mariadb.com/kb/en/mariadb-versus-mysql-compatibility/
 
 As MariaDB is a full replacement of MySQL, the MySQL manual at
 http://dev.mysql.com/doc is generally applicable.
 
-Help:
+Help
 -----
 
 More help is available from the Maria Discuss mailing list
 https://launchpad.net/~maria-discuss
 and the #maria IRC channel on Freenode.
 
+Live QA for beginner contributors
+----
+MariaDB has a dedicated time each week when we answer new contributor questions live on Zulip and IRC.
+From 8:00 to 10:00 UTC on Mondays, and 10:00 to 12:00 UTC on Thursdays,
+anyone can ask any questions they’d like, and a live developer will be available to assist.
+
+New contributors can ask questions any time, but we will provide immediate feedback during that interval.
 
-License:
+License
 --------
 
 ***************************************************************************
@@ -61,7 +66,7 @@ and COPYING.thirdparty files.
 
 ***************************************************************************
 
-Bug Reports:
+Bug Reports
 ------------
 
 Bug and/or error reports regarding MariaDB should be submitted at:
diff --git a/mysql-test/main/connect.test b/mysql-test/main/connect.test
index 9bc067f3236..b46913c3cb4 100644
--- a/mysql-test/main/connect.test
+++ b/mysql-test/main/connect.test
@@ -251,11 +251,8 @@ let $wait_condition =
 
 --echo
 --echo # -- Waiting for connections to close...
-let $wait_condition =
-  SELECT COUNT(*) = 1
-  FROM information_schema.processlist
-  WHERE db = 'test';
---source include/wait_condition.inc
+let $count_sessions=1;
+--source include/wait_until_count_sessions.inc
 
 --echo
 DROP USER mysqltest_u1@localhost;
diff --git a/mysql-test/main/failed_auth_3909.result b/mysql-test/main/failed_auth_3909.result
index 4c3c0aba9df..bc82492c9ed 100644
--- a/mysql-test/main/failed_auth_3909.result
+++ b/mysql-test/main/failed_auth_3909.result
@@ -10,15 +10,15 @@ Warning	1364	Field 'authentication_string' doesn't have a default value
 flush privileges;
 connect(localhost,u1,,test,MASTER_PORT,MASTER_SOCKET);
 connect  fail,localhost,u1;
-ERROR HY000: Server is running in --secure-auth mode, but 'u1'@'localhost' has a password in the old format; please change the password to the new format
+ERROR 28000: Access denied for user 'u1'@'localhost' (using password: NO)
 connect(localhost,u2,,test,MASTER_PORT,MASTER_SOCKET);
 connect  fail,localhost,u2;
-ERROR 28000: Access denied for user 'u2'@'localhost' (using password: NO)
+ERROR HY000: Server is running in --secure-auth mode, but 'u2'@'localhost' has a password in the old format; please change the password to the new format
 connect(localhost,u2,password,test,MASTER_PORT,MASTER_SOCKET);
 connect  fail,localhost,u2,password;
-ERROR 28000: Access denied for user 'u2'@'localhost' (using password: YES)
-ERROR HY000: Server is running in --secure-auth mode, but 'u1'@'localhost' has a password in the old format; please change the password to the new format
-ERROR 28000: Access denied for user 'u2'@'localhost' (using password: NO)
-ERROR 28000: Access denied for user 'u2'@'localhost' (using password: YES)
+ERROR HY000: Server is running in --secure-auth mode, but 'u2'@'localhost' has a password in the old format; please change the password to the new format
+ERROR 28000: Access denied for user 'u1'@'localhost' (using password: NO)
+ERROR HY000: Server is running in --secure-auth mode, but 'u2'@'localhost' has a password in the old format; please change the password to the new format
+ERROR HY000: Server is running in --secure-auth mode, but 'u2'@'localhost' has a password in the old format; please change the password to the new format
 delete from mysql.user where plugin = 'mysql_old_password';
 flush privileges;
diff --git a/mysql-test/main/failed_auth_3909.test b/mysql-test/main/failed_auth_3909.test
index fb104cf4b81..45267d628b2 100644
--- a/mysql-test/main/failed_auth_3909.test
+++ b/mysql-test/main/failed_auth_3909.test
@@ -11,24 +11,24 @@ insert ignore mysql.user (user,plugin) values ('foo','mysql_old_password'),('bar
 flush privileges;
 
 --replace_result $MASTER_MYSOCK MASTER_SOCKET $MASTER_MYPORT MASTER_PORT
---error ER_SERVER_IS_IN_SECURE_AUTH_MODE
+--error ER_ACCESS_DENIED_ERROR
 connect (fail,localhost,u1);
 
 --replace_result $MASTER_MYSOCK MASTER_SOCKET $MASTER_MYPORT MASTER_PORT
---error ER_ACCESS_DENIED_ERROR
+--error ER_SERVER_IS_IN_SECURE_AUTH_MODE
 connect (fail,localhost,u2);
 
 --replace_result $MASTER_MYSOCK MASTER_SOCKET $MASTER_MYPORT MASTER_PORT
---error ER_ACCESS_DENIED_ERROR
+--error ER_SERVER_IS_IN_SECURE_AUTH_MODE
 connect (fail,localhost,u2,password);
 
---error ER_SERVER_IS_IN_SECURE_AUTH_MODE
+--error ER_ACCESS_DENIED_ERROR
 change_user u1;
 
---error ER_ACCESS_DENIED_ERROR
+--error ER_SERVER_IS_IN_SECURE_AUTH_MODE
 change_user u2;
 
---error ER_ACCESS_DENIED_ERROR
+--error ER_SERVER_IS_IN_SECURE_AUTH_MODE
 change_user u2,password;
 
 delete from mysql.user where plugin = 'mysql_old_password';
diff --git a/mysql-test/main/mysql_install_db_win.result b/mysql-test/main/mysql_install_db_win.result
new file mode 100644
index 00000000000..1fba59086a4
--- /dev/null
+++ b/mysql-test/main/mysql_install_db_win.result
@@ -0,0 +1,9 @@
+Running bootstrap
+Removing default user
+Creating my.ini file
+Creation of the database was successful
+# Kill the server
+SELECT @@datadir;
+@@datadir
+DATADIR/
+# Kill the server
diff --git a/mysql-test/main/mysql_install_db_win.test b/mysql-test/main/mysql_install_db_win.test
new file mode 100644
index 00000000000..ccc08262436
--- /dev/null
+++ b/mysql-test/main/mysql_install_db_win.test
@@ -0,0 +1,22 @@
+--source include/windows.inc
+
+# Create database  in tmp directory using mysql_install_db.exe, 
+# and start server from this directory.
+let $ddir= $MYSQLTEST_VARDIR/tmp/ddir;
+exec $MYSQL_INSTALL_DB_EXE --datadir=$ddir;
+
+--source include/kill_mysqld.inc
+let $restart_parameters=--datadir=$ddir --loose-innodb;
+--source include/start_mysqld.inc
+
+# Smoke test - check that we're actually using datadir
+# we've created (i.e restart_parameters worked)
+--replace_result  $ddir DATADIR
+SELECT @@datadir;
+
+# restart in the original datadir again
+--source include/kill_mysqld.inc
+rmdir $ddir;
+let $restart_parameters=;
+--source include/start_mysqld.inc
+
diff --git a/mysql-test/main/mysqld--help.result b/mysql-test/main/mysqld--help.result
index cf25f377142..094a63cf50a 100644
--- a/mysql-test/main/mysqld--help.result
+++ b/mysql-test/main/mysqld--help.result
@@ -294,6 +294,15 @@ The following specify which files/extra groups are read (specified before remain
  --group-concat-max-len=# 
  The maximum length of the result of function
  GROUP_CONCAT()
+ --gtid-cleanup-batch-size=# 
+ Normally does not need tuning. How many old rows must
+ accumulate in the mysql.gtid_slave_pos table before a
+ background job will be run to delete them. Can be
+ increased to reduce number of commits if using many
+ different engines with --gtid_pos_auto_engines, or to
+ reduce CPU overhead if using a huge number of different
+ gtid_domain_ids. Can be decreased to reduce number of old
+ rows in the table.
  --gtid-domain-id=#  Used with global transaction ID to identify logically
  independent replication streams. When events can
  propagate through multiple parallel paths (for example
@@ -1434,6 +1443,7 @@ gdb FALSE
 general-log FALSE
 getopt-prefix-matching FALSE
 group-concat-max-len 1048576
+gtid-cleanup-batch-size 64
 gtid-domain-id 0
 gtid-ignore-duplicates FALSE
 gtid-pos-auto-engines 
diff --git a/mysql-test/main/stat_tables.result b/mysql-test/main/stat_tables.result
index 40290ca9879..3ebc3b47833 100644
--- a/mysql-test/main/stat_tables.result
+++ b/mysql-test/main/stat_tables.result
@@ -591,6 +591,25 @@ id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
 set @@optimizer_use_condition_selectivity=@save_optimizer_use_condition_selectivity;
 set use_stat_tables=@save_use_stat_tables;
 #
+# MDEV-17734: AddressSanitizer: use-after-poison in create_key_parts_for_pseudo_indexes
+#
+set @@use_stat_tables= PREFERABLY;
+set @save_optimizer_use_condition_selectivity= @@optimizer_use_condition_selectivity;
+set @@optimizer_use_condition_selectivity=4;
+set @save_use_stat_tables= @@use_stat_tables;
+create table t1 (a int, b int);
+insert into t1(a,b) values (1,2),(1,3),(1,4),(1,5),(2,6),(2,7),(3,8),(3,9),(3,9),(4,10);
+analyze table t1 persistent for columns (a) indexes ();
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	status	OK
+select * from t1 where a=1 and b=3;
+a	b
+1	3
+set @@optimizer_use_condition_selectivity= @save_optimizer_use_condition_selectivity;
+set use_stat_tables=@save_use_stat_tables;
+drop table t1;
+#
 # MDEV-16711:CREATE OR REPLACE TABLE introducing BLOB column
 #
 SET use_stat_tables= PREFERABLY;
diff --git a/mysql-test/main/stat_tables.test b/mysql-test/main/stat_tables.test
index 19bc0fa2f46..0005e48e0e7 100644
--- a/mysql-test/main/stat_tables.test
+++ b/mysql-test/main/stat_tables.test
@@ -370,6 +370,23 @@ set @@optimizer_use_condition_selectivity=@save_optimizer_use_condition_selectiv
 set use_stat_tables=@save_use_stat_tables;
 
 --echo #
+--echo # MDEV-17734: AddressSanitizer: use-after-poison in create_key_parts_for_pseudo_indexes
+--echo #
+
+set @@use_stat_tables= PREFERABLY;
+set @save_optimizer_use_condition_selectivity= @@optimizer_use_condition_selectivity;
+set @@optimizer_use_condition_selectivity=4;
+set @save_use_stat_tables= @@use_stat_tables;
+create table t1 (a int, b int);
+insert into t1(a,b) values (1,2),(1,3),(1,4),(1,5),(2,6),(2,7),(3,8),(3,9),(3,9),(4,10);
+
+analyze table t1 persistent for columns (a) indexes ();
+select * from t1 where a=1 and b=3;
+set @@optimizer_use_condition_selectivity= @save_optimizer_use_condition_selectivity;
+set use_stat_tables=@save_use_stat_tables;
+drop table t1;
+
+--echo #
 --echo # MDEV-16711:CREATE OR REPLACE TABLE introducing BLOB column
 --echo #
 
diff --git a/mysql-test/main/stat_tables_innodb.result b/mysql-test/main/stat_tables_innodb.result
index 070d13d9bb1..a6c5525a0d3 100644
--- a/mysql-test/main/stat_tables_innodb.result
+++ b/mysql-test/main/stat_tables_innodb.result
@@ -618,6 +618,25 @@ id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
 set @@optimizer_use_condition_selectivity=@save_optimizer_use_condition_selectivity;
 set use_stat_tables=@save_use_stat_tables;
 #
+# MDEV-17734: AddressSanitizer: use-after-poison in create_key_parts_for_pseudo_indexes
+#
+set @@use_stat_tables= PREFERABLY;
+set @save_optimizer_use_condition_selectivity= @@optimizer_use_condition_selectivity;
+set @@optimizer_use_condition_selectivity=4;
+set @save_use_stat_tables= @@use_stat_tables;
+create table t1 (a int, b int);
+insert into t1(a,b) values (1,2),(1,3),(1,4),(1,5),(2,6),(2,7),(3,8),(3,9),(3,9),(4,10);
+analyze table t1 persistent for columns (a) indexes ();
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	status	OK
+select * from t1 where a=1 and b=3;
+a	b
+1	3
+set @@optimizer_use_condition_selectivity= @save_optimizer_use_condition_selectivity;
+set use_stat_tables=@save_use_stat_tables;
+drop table t1;
+#
 # MDEV-16711:CREATE OR REPLACE TABLE introducing BLOB column
 #
 SET use_stat_tables= PREFERABLY;
diff --git a/mysql-test/mysql-test-run.pl b/mysql-test/mysql-test-run.pl
index 361724e5c57..b783d815222 100755
--- a/mysql-test/mysql-test-run.pl
+++ b/mysql-test/mysql-test-run.pl
@@ -2292,6 +2292,10 @@ sub environment_setup {
   $ENV{'EXE_MYSQL'}=                $exe_mysql;
   $ENV{'MYSQL_PLUGIN'}=             $exe_mysql_plugin;
   $ENV{'MYSQL_EMBEDDED'}=           $exe_mysql_embedded;
+  if(IS_WINDOWS)
+  {
+     $ENV{'MYSQL_INSTALL_DB_EXE'}=  mtr_exe_exists("$bindir/sql$opt_vs_config/mysql_install_db");
+  }
 
   my $client_config_exe=
     mtr_exe_maybe_exists(
diff --git a/mysql-test/std_data/rpl/mysql-5.7.11-stm-temporal-round-binlog.000001 b/mysql-test/std_data/rpl/mysql-5.7.11-stm-temporal-round-binlog.000001
new file mode 100644
index 00000000000..5010e164e43
--- /dev/null
+++ b/mysql-test/std_data/rpl/mysql-5.7.11-stm-temporal-round-binlog.000001
diff --git a/mysql-test/std_data/rpl/mysql-8.0.13-stm-temporal-round-binlog.000001 b/mysql-test/std_data/rpl/mysql-8.0.13-stm-temporal-round-binlog.000001
new file mode 100644
index 00000000000..4d582fdf5bb
--- /dev/null
+++ b/mysql-test/std_data/rpl/mysql-8.0.13-stm-temporal-round-binlog.000001
diff --git a/mysql-test/suite/encryption/disabled.def b/mysql-test/suite/encryption/disabled.def
index d92d3495cb8..746faf49873 100644
--- a/mysql-test/suite/encryption/disabled.def
+++ b/mysql-test/suite/encryption/disabled.def
@@ -12,3 +12,4 @@
 
 innodb_scrub            : MDEV-8139 scrubbing does not work reliably
 innodb_scrub_background : MDEV-8139 scrubbing does not work reliably
+innodb-redo-badkey      : MDEV-13893/MDEV-12699 fix recovery of corrupted pages
diff --git a/mysql-test/suite/innodb_fts/r/create.result b/mysql-test/suite/innodb_fts/r/create.result
index f3650fb3dc9..896291854a8 100644
--- a/mysql-test/suite/innodb_fts/r/create.result
+++ b/mysql-test/suite/innodb_fts/r/create.result
@@ -162,3 +162,19 @@ SELECT len,COUNT(*) FROM INFORMATION_SCHEMA.INNODB_SYS_COLUMNS where name='word'
 len	COUNT(*)
 84	6
 DROP TABLE t;
+#
+# MDEV-17923 Assertion memcmp(field, field_ref_zero, 7) failed in
+# trx_undo_page_report_modify upon optimizing table
+# under innodb_optimize_fulltext_only
+#
+CREATE TABLE t1 (f1 TEXT, f2 TEXT, FULLTEXT KEY (f2)) ENGINE=InnoDB;
+INSERT INTO t1 (f1) VALUES ('foo'),('bar');
+DELETE FROM t1 LIMIT 1;
+ALTER TABLE t1 ADD FULLTEXT KEY (f1);
+SET @optimize_fulltext.save= @@innodb_optimize_fulltext_only;
+SET GLOBAL innodb_optimize_fulltext_only= 1;
+OPTIMIZE TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	optimize	status	OK
+DROP TABLE t1;
+SET GLOBAL innodb_optimize_fulltext_only= @optimize_fulltext.save;
diff --git a/mysql-test/suite/innodb_fts/t/create.test b/mysql-test/suite/innodb_fts/t/create.test
index fe0273af750..4e522994fcc 100644
--- a/mysql-test/suite/innodb_fts/t/create.test
+++ b/mysql-test/suite/innodb_fts/t/create.test
@@ -90,3 +90,19 @@ ENGINE=InnoDB;
 # The column length should be 84 bytes (84 characters * 1 byte/character).
 SELECT len,COUNT(*) FROM INFORMATION_SCHEMA.INNODB_SYS_COLUMNS where name='word' GROUP BY len;
 DROP TABLE t;
+
+--echo #
+--echo # MDEV-17923 Assertion memcmp(field, field_ref_zero, 7) failed in
+--echo # trx_undo_page_report_modify upon optimizing table
+--echo # under innodb_optimize_fulltext_only
+--echo #
+
+CREATE TABLE t1 (f1 TEXT, f2 TEXT, FULLTEXT KEY (f2)) ENGINE=InnoDB;
+INSERT INTO t1 (f1) VALUES ('foo'),('bar');
+DELETE FROM t1 LIMIT 1;
+ALTER TABLE t1 ADD FULLTEXT KEY (f1);
+SET @optimize_fulltext.save= @@innodb_optimize_fulltext_only;
+SET GLOBAL innodb_optimize_fulltext_only= 1;
+OPTIMIZE TABLE t1;
+DROP TABLE t1;
+SET GLOBAL innodb_optimize_fulltext_only= @optimize_fulltext.save;
diff --git a/mysql-test/suite/perfschema/r/hostcache_ipv4_nameinfo_again_allow.result b/mysql-test/suite/perfschema/r/hostcache_ipv4_nameinfo_again_allow.result
index a172dff7935..7963bed8213 100644
--- a/mysql-test/suite/perfschema/r/hostcache_ipv4_nameinfo_again_allow.result
+++ b/mysql-test/suite/perfschema/r/hostcache_ipv4_nameinfo_again_allow.result
@@ -118,7 +118,7 @@ Con4 is alive
 Con4 is alive
 select current_user();
 current_user()
-root@192.0.2.4
+root@santa.claus.ipv4.example.com
 disconnect con4;
 connection default;
 "Dumping performance_schema.host_cache"
@@ -155,7 +155,7 @@ Con5 is alive
 Con5 is alive
 select current_user();
 current_user()
-root@192.0.2.4
+root@santa.claus.ipv4.example.com
 disconnect con5;
 connection default;
 "Dumping performance_schema.host_cache"
diff --git a/mysql-test/suite/perfschema/r/hostcache_ipv6_nameinfo_again_allow.result b/mysql-test/suite/perfschema/r/hostcache_ipv6_nameinfo_again_allow.result
index 4fdc6ef1b4c..baddc88b116 100644
--- a/mysql-test/suite/perfschema/r/hostcache_ipv6_nameinfo_again_allow.result
+++ b/mysql-test/suite/perfschema/r/hostcache_ipv6_nameinfo_again_allow.result
@@ -118,7 +118,7 @@ Con4 is alive
 Con4 is alive
 select current_user();
 current_user()
-root@2001:db8::6:6
+root@santa.claus.ipv6.example.com
 disconnect con4;
 connection default;
 "Dumping performance_schema.host_cache"
@@ -155,7 +155,7 @@ Con5 is alive
 Con5 is alive
 select current_user();
 current_user()
-root@2001:db8::6:6
+root@santa.claus.ipv6.example.com
 disconnect con5;
 connection default;
 "Dumping performance_schema.host_cache"
diff --git a/mysql-test/suite/perfschema/r/socket_connect.result b/mysql-test/suite/perfschema/r/socket_connect.result
index 1ac22f6ca34..304521b044f 100644
--- a/mysql-test/suite/perfschema/r/socket_connect.result
+++ b/mysql-test/suite/perfschema/r/socket_connect.result
@@ -167,19 +167,6 @@ connection default;
 
 # 6.1 Verify that there are no TCP/IP connections in the socket instance table
 
-SELECT COUNT(*) = 0 AS 'Expect 1'
-FROM performance_schema.socket_instances
-WHERE EVENT_NAME LIKE '%client_connection%'
-  AND OBJECT_INSTANCE_BEGIN <> @default_object_instance_begin
-AND (IP LIKE '%127.0.0.1' OR IP LIKE '%::1');
-Expect 1
-1
 
 # 6.2 Verify that there are no TCP/IP connections in the summary instance table
 
-SELECT COUNT(*) = 0 AS 'Expect 1'
-FROM performance_schema.socket_summary_by_instance
-WHERE EVENT_NAME LIKE '%client_connection%'
-  AND OBJECT_INSTANCE_BEGIN <> @default_object_instance_begin;
-Expect 1
-1
diff --git a/mysql-test/suite/perfschema/t/socket_connect.test b/mysql-test/suite/perfschema/t/socket_connect.test
index 909840144ef..b4579605eb5 100644
--- a/mysql-test/suite/perfschema/t/socket_connect.test
+++ b/mysql-test/suite/perfschema/t/socket_connect.test
@@ -273,18 +273,21 @@ WHERE EVENT_NAME LIKE '%client_connection%'
 --echo
 --echo # 6.1 Verify that there are no TCP/IP connections in the socket instance table
 --echo
-eval SELECT COUNT(*) = 0 AS 'Expect 1'
+let $wait_condition=
+SELECT COUNT(*) = 0 AS 'Expect 1'
 FROM performance_schema.socket_instances
 WHERE EVENT_NAME LIKE '%client_connection%'
   AND OBJECT_INSTANCE_BEGIN <> @default_object_instance_begin
   AND $ip_localhost;
+--source include/wait_condition.inc
 
 --echo
 --echo # 6.2 Verify that there are no TCP/IP connections in the summary instance table
 --echo
-eval SELECT COUNT(*) = 0 AS 'Expect 1'
+let $wait_condition=
+SELECT COUNT(*) = 0 AS 'Expect 1'
 FROM performance_schema.socket_summary_by_instance
 WHERE EVENT_NAME LIKE '%client_connection%'
   AND OBJECT_INSTANCE_BEGIN <> @default_object_instance_begin;
-
+--source include/wait_condition.inc
 exit;
diff --git a/mysql-test/suite/roles/flush_roles-17898.result b/mysql-test/suite/roles/flush_roles-17898.result
new file mode 100644
index 00000000000..dbe6ea24afd
--- /dev/null
+++ b/mysql-test/suite/roles/flush_roles-17898.result
@@ -0,0 +1,13 @@
+use mysql;
+insert db (db,user,select_priv) values ('foo','dwr_foo','Y'), ('bar','dwr_bar','Y');
+insert roles_mapping (user,role) values ('dwr_qux_dev','dwr_foo'),('dwr_qux_dev','dwr_bar');
+insert ignore user (user,show_db_priv,is_role) values ('dwr_foo','N','Y'), ('dwr_bar','N','Y'), ('dwr_qux_dev','Y','Y');
+Warnings:
+Warning	1364	Field 'ssl_cipher' doesn't have a default value
+Warning	1364	Field 'x509_issuer' doesn't have a default value
+Warning	1364	Field 'x509_subject' doesn't have a default value
+Warning	1364	Field 'authentication_string' doesn't have a default value
+flush privileges;
+drop role dwr_foo;
+drop role dwr_bar;
+drop role dwr_qux_dev;
diff --git a/mysql-test/suite/roles/flush_roles-17898.test b/mysql-test/suite/roles/flush_roles-17898.test
new file mode 100644
index 00000000000..6a3b8d6f345
--- /dev/null
+++ b/mysql-test/suite/roles/flush_roles-17898.test
@@ -0,0 +1,11 @@
+#
+# MDEV-17898 FLUSH PRIVILEGES crashes server with segfault
+#
+use mysql;
+insert db (db,user,select_priv) values ('foo','dwr_foo','Y'), ('bar','dwr_bar','Y');
+insert roles_mapping (user,role) values ('dwr_qux_dev','dwr_foo'),('dwr_qux_dev','dwr_bar');
+insert ignore user (user,show_db_priv,is_role) values ('dwr_foo','N','Y'), ('dwr_bar','N','Y'), ('dwr_qux_dev','Y','Y');
+flush privileges;
+drop role dwr_foo;
+drop role dwr_bar;
+drop role dwr_qux_dev;
diff --git a/mysql-test/suite/rpl/r/rpl_gtid_mdev4484.result b/mysql-test/suite/rpl/r/rpl_gtid_mdev4484.result
index 00307a8c104..5dffdd9809c 100644
--- a/mysql-test/suite/rpl/r/rpl_gtid_mdev4484.result
+++ b/mysql-test/suite/rpl/r/rpl_gtid_mdev4484.result
@@ -16,36 +16,32 @@ INSERT INTO t1 VALUES (1);
 connection slave;
 connection slave;
 include/stop_slave.inc
+SET @old_gtid_cleanup_batch_size= @@GLOBAL.gtid_cleanup_batch_size;
+SET GLOBAL gtid_cleanup_batch_size= 2;
 SET @old_dbug= @@GLOBAL.debug_dbug;
 SET GLOBAL debug_dbug="+d,gtid_slave_pos_simulate_failed_delete";
 SET sql_log_bin= 0;
-CALL mtr.add_suppression("Can't find file");
+CALL mtr.add_suppression("<DEBUG> Error deleting old GTID row");
 SET sql_log_bin= 1;
 include/start_slave.inc
 connection master;
-INSERT INTO t1 VALUES (2);
-connection slave;
-include/wait_for_slave_sql_error.inc [errno=1942]
-STOP SLAVE IO_THREAD;
-SELECT domain_id, server_id, seq_no FROM mysql.gtid_slave_pos
-ORDER BY domain_id, sub_id DESC LIMIT 1;
-domain_id	server_id	seq_no
-0	1	3
+connection slave;
+SELECT COUNT(*), MAX(seq_no) INTO @pre_count, @pre_max_seq_no
+FROM mysql.gtid_slave_pos;
+SELECT IF(@pre_count >= 20, "OK", CONCAT("Error: too few rows seen while errors injected: ", @pre_count));
+IF(@pre_count >= 20, "OK", CONCAT("Error: too few rows seen while errors injected: ", @pre_count))
+OK
 SET GLOBAL debug_dbug= @old_dbug;
-include/start_slave.inc
 connection master;
-INSERT INTO t1 VALUES (3);
-connection slave;
-connection slave;
-SELECT domain_id, server_id, seq_no FROM mysql.gtid_slave_pos
-ORDER BY domain_id, sub_id DESC LIMIT 1;
-domain_id	server_id	seq_no
-0	1	4
-SELECT * FROM t1 ORDER BY i;
-i
-1
-2
-3
+connection slave;
+connection slave;
+SELECT IF(COUNT(*) >= 1, "OK", CONCAT("Error: too few rows seen after errors no longer injected: ", COUNT(*)))
+FROM mysql.gtid_slave_pos
+WHERE seq_no <= @pre_max_seq_no;
+IF(COUNT(*) >= 1, "OK", CONCAT("Error: too few rows seen after errors no longer injected: ", COUNT(*)))
+OK
 connection master;
 DROP TABLE t1;
+connection slave;
+SET GLOBAL gtid_cleanup_batch_size= @old_gtid_cleanup_batch_size;
 include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/r/rpl_gtid_stop_start.result b/mysql-test/suite/rpl/r/rpl_gtid_stop_start.result
index 1647b6c998c..50f24d56e9a 100644
--- a/mysql-test/suite/rpl/r/rpl_gtid_stop_start.result
+++ b/mysql-test/suite/rpl/r/rpl_gtid_stop_start.result
@@ -171,7 +171,7 @@ include/start_slave.inc
 *** MDEV-4692: mysql.gtid_slave_pos accumulates values for a domain ***
 SELECT domain_id, COUNT(*) FROM mysql.gtid_slave_pos GROUP BY domain_id;
 domain_id	COUNT(*)
-0	2
+0	3
 1	2
 connection server_1;
 INSERT INTO t1 VALUES (11);
@@ -179,7 +179,7 @@ connection server_2;
 FLUSH NO_WRITE_TO_BINLOG TABLES;
 SELECT domain_id, COUNT(*) FROM mysql.gtid_slave_pos GROUP BY domain_id;
 domain_id	COUNT(*)
-0	2
+0	4
 1	2
 include/start_slave.inc
 connection server_1;
@@ -189,8 +189,8 @@ connection server_2;
 FLUSH NO_WRITE_TO_BINLOG TABLES;
 SELECT domain_id, COUNT(*) FROM mysql.gtid_slave_pos GROUP BY domain_id;
 domain_id	COUNT(*)
-0	2
-1	2
+0	3
+1	1
 *** MDEV-4650: show variables; ERROR 1946 (HY000): Failed to load replication slave GTID position ***
 connection server_2;
 SET sql_log_bin=0;
diff --git a/mysql-test/suite/rpl/r/rpl_mysql57_stm_temporal_round.result b/mysql-test/suite/rpl/r/rpl_mysql57_stm_temporal_round.result
new file mode 100644
index 00000000000..bedd103c2a0
--- /dev/null
+++ b/mysql-test/suite/rpl/r/rpl_mysql57_stm_temporal_round.result
@@ -0,0 +1,22 @@
+#
+# MDEV-8894 Inserting fractional seconds into MySQL 5.6 master breaks consistency on MariaDB 10 slave
+#
+include/master-slave.inc
+[connection master]
+connection slave;
+CREATE TABLE t1 (id SERIAL, a DATETIME(3));
+include/stop_slave.inc
+connection master;
+include/rpl_stop_server.inc [server_number=1]
+include/rpl_start_server.inc [server_number=1]
+connection slave;
+CHANGE MASTER TO master_host='127.0.0.1', master_port=SERVER_MYPORT_1, master_user='root', master_log_file='master-bin.000001', master_log_pos=4;
+include/start_slave.inc
+connection master;
+connection slave;
+SELECT * FROM t1 ORDER BY id;
+id	a
+1	2001-01-01 00:00:01.000
+include/stop_slave.inc
+DROP TABLE t1;
+include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/r/rpl_mysql80_stm_temporal_round.result b/mysql-test/suite/rpl/r/rpl_mysql80_stm_temporal_round.result
new file mode 100644
index 00000000000..23b3217895a
--- /dev/null
+++ b/mysql-test/suite/rpl/r/rpl_mysql80_stm_temporal_round.result
@@ -0,0 +1,23 @@
+#
+# MDEV-8894 Inserting fractional seconds into MySQL 5.6 master breaks consistency on MariaDB 10 slave
+#
+include/master-slave.inc
+[connection master]
+connection slave;
+CREATE TABLE t1 (id SERIAL, a DATETIME(3));
+include/stop_slave.inc
+connection master;
+include/rpl_stop_server.inc [server_number=1]
+include/rpl_start_server.inc [server_number=1]
+connection slave;
+CHANGE MASTER TO master_host='127.0.0.1', master_port=SERVER_MYPORT_1, master_user='root', master_log_file='master-bin.000001', master_log_pos=4;
+include/start_slave.inc
+connection master;
+connection slave;
+SELECT * FROM t1 ORDER BY id;
+id	a
+1	2001-01-01 00:00:01.000
+2	2001-01-01 00:00:00.999
+include/stop_slave.inc
+DROP TABLE t1;
+include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/r/rpl_parallel_optimistic.result b/mysql-test/suite/rpl/r/rpl_parallel_optimistic.result
index ca202a66b0e..83343e52cab 100644
--- a/mysql-test/suite/rpl/r/rpl_parallel_optimistic.result
+++ b/mysql-test/suite/rpl/r/rpl_parallel_optimistic.result
@@ -12,6 +12,8 @@ SET GLOBAL slave_parallel_threads=10;
 CHANGE MASTER TO master_use_gtid=slave_pos;
 SET @old_parallel_mode=@@GLOBAL.slave_parallel_mode;
 SET GLOBAL slave_parallel_mode='optimistic';
+SET @old_gtid_cleanup_batch_size= @@GLOBAL.gtid_cleanup_batch_size;
+SET GLOBAL gtid_cleanup_batch_size= 1000000;
 connection server_1;
 INSERT INTO t1 VALUES(1,1);
 BEGIN;
@@ -131,6 +133,11 @@ c
 204
 205
 206
+SELECT IF(COUNT(*) >= 30, "OK", CONCAT("Error: too few old rows found: ", COUNT(*)))
+FROM mysql.gtid_slave_pos;
+IF(COUNT(*) >= 30, "OK", CONCAT("Error: too few old rows found: ", COUNT(*)))
+OK
+SET GLOBAL gtid_cleanup_batch_size=1;
 *** Test @@skip_parallel_replication. ***
 connection server_2;
 include/stop_slave.inc
@@ -651,9 +658,10 @@ DROP TABLE t1, t2, t3;
 include/save_master_gtid.inc
 connection server_2;
 include/sync_with_master_gtid.inc
-Check that no more than the expected last four GTIDs are in mysql.gtid_slave_pos
-select count(4) <= 4 from mysql.gtid_slave_pos order by domain_id, sub_id;
-count(4) <= 4
+SELECT COUNT(*) <= 5*@@GLOBAL.gtid_cleanup_batch_size
+FROM mysql.gtid_slave_pos;
+COUNT(*) <= 5*@@GLOBAL.gtid_cleanup_batch_size
 1
+SET GLOBAL gtid_cleanup_batch_size= @old_gtid_cleanup_batch_size;
 connection server_1;
 include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/t/rpl_gtid_mdev4484.test b/mysql-test/suite/rpl/t/rpl_gtid_mdev4484.test
index 1e5d1abbb3b..5c17653da8a 100644
--- a/mysql-test/suite/rpl/t/rpl_gtid_mdev4484.test
+++ b/mysql-test/suite/rpl/t/rpl_gtid_mdev4484.test
@@ -28,37 +28,79 @@ INSERT INTO t1 VALUES (1);
 # Inject an artificial error deleting entries, and check that the error handling code works.
 --connection slave
 --source include/stop_slave.inc
+SET @old_gtid_cleanup_batch_size= @@GLOBAL.gtid_cleanup_batch_size;
+SET GLOBAL gtid_cleanup_batch_size= 2;
 SET @old_dbug= @@GLOBAL.debug_dbug;
 SET GLOBAL debug_dbug="+d,gtid_slave_pos_simulate_failed_delete";
 SET sql_log_bin= 0;
-CALL mtr.add_suppression("Can't find file");
+CALL mtr.add_suppression("<DEBUG> Error deleting old GTID row");
 SET sql_log_bin= 1;
 --source include/start_slave.inc
 
 --connection master
-INSERT INTO t1 VALUES (2);
+--disable_query_log
+let $i = 20;
+while ($i) {
+  eval INSERT INTO t1 VALUES ($i+10);
+  dec $i;
+}
+--enable_query_log
+--save_master_pos
 
 --connection slave
---let $slave_sql_errno= 1942
---source include/wait_for_slave_sql_error.inc
-STOP SLAVE IO_THREAD;
-SELECT domain_id, server_id, seq_no FROM mysql.gtid_slave_pos
- ORDER BY domain_id, sub_id DESC LIMIT 1;
+--sync_with_master
+
+# Now wait for the slave background thread to try to delete old rows and
+# hit the error injection.
+--let _TEST_MYSQLD_ERROR_LOG=$MYSQLTEST_VARDIR/log/mysqld.2.err
+--perl
+  open F, '<', $ENV{'_TEST_MYSQLD_ERROR_LOG'} or die;
+  outer: while (1) {
+    inner: while (<F>) {
+      last outer if /<DEBUG> Error deleting old GTID row/;
+    }
+    # Easy way to do sub-second sleep without extra modules.
+    select(undef, undef, undef, 0.1);
+  }
+EOF
+
+# Since we injected error in the cleanup code, the rows should remain in
+# mysql.gtid_slave_pos. Check that we have at least 20 (more robust against
+# non-deterministic cleanup and future changes than checking for exact number).
+SELECT COUNT(*), MAX(seq_no) INTO @pre_count, @pre_max_seq_no
+  FROM mysql.gtid_slave_pos;
+SELECT IF(@pre_count >= 20, "OK", CONCAT("Error: too few rows seen while errors injected: ", @pre_count));
 SET GLOBAL debug_dbug= @old_dbug;
---source include/start_slave.inc
 
 --connection master
-INSERT INTO t1 VALUES (3);
+--disable_query_log
+let $i = 20;
+while ($i) {
+  eval INSERT INTO t1 VALUES ($i+40);
+  dec $i;
+}
+--enable_query_log
 --sync_slave_with_master
 
 --connection slave
-SELECT domain_id, server_id, seq_no FROM mysql.gtid_slave_pos
- ORDER BY domain_id, sub_id DESC LIMIT 1;
-SELECT * FROM t1 ORDER BY i;
-
+# Now check that 1) rows are being deleted again after removing error
+# injection, and 2) old rows are left that failed their delete while errors
+# where injected (again compensating for non-deterministic deletion).
+# Deletion is async and slightly non-deterministic, so we wait for at
+# least 10 of the 20 new rows to be deleted.
+let $wait_condition=
+  SELECT COUNT(*) <= 20-10
+    FROM mysql.gtid_slave_pos
+   WHERE seq_no > @pre_max_seq_no;
+--source include/wait_condition.inc
+SELECT IF(COUNT(*) >= 1, "OK", CONCAT("Error: too few rows seen after errors no longer injected: ", COUNT(*)))
+  FROM mysql.gtid_slave_pos
+ WHERE seq_no <= @pre_max_seq_no;
 
 # Clean up
 --connection master
 DROP TABLE t1;
+--connection slave
+SET GLOBAL gtid_cleanup_batch_size= @old_gtid_cleanup_batch_size;
 
 --source include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/t/rpl_mysql57_stm_temporal_round.test b/mysql-test/suite/rpl/t/rpl_mysql57_stm_temporal_round.test
new file mode 100644
index 00000000000..675b7db0603
--- /dev/null
+++ b/mysql-test/suite/rpl/t/rpl_mysql57_stm_temporal_round.test
@@ -0,0 +1,58 @@
+--source include/have_binlog_format_statement.inc
+
+--echo #
+--echo # MDEV-8894 Inserting fractional seconds into MySQL 5.6 master breaks consistency on MariaDB 10 slave
+--echo #
+
+--source include/have_innodb.inc
+--source include/master-slave.inc
+
+--connection slave
+CREATE TABLE t1 (id SERIAL, a DATETIME(3));
+--source include/stop_slave.inc
+
+--connection master
+--let $datadir= `SELECT @@datadir`
+
+--let $rpl_server_number= 1
+--source include/rpl_stop_server.inc
+
+--remove_file $datadir/master-bin.000001
+
+#
+# Simulate MySQL 5.7.x master
+#
+# mysql-5.7.11-stm-temporal-round-binlog.000001 was recorded against a
+# table with this structure:
+#CREATE TABLE t1 (id SERIAL, a DATETIME(3));
+# (note, the CREATE statement is not inside the binary log)
+#
+# using this command line:
+#   mysqld --log-bin --binlog-format=statement
+# with the following single SQL statement:
+#
+#INSERT INTO t1 (a) VALUES ('2001-01-01 00:00:00.999999');
+#
+
+--copy_file $MYSQL_TEST_DIR/std_data/rpl/mysql-5.7.11-stm-temporal-round-binlog.000001 $datadir/master-bin.000001
+
+--let $rpl_server_number= 1
+--source include/rpl_start_server.inc
+
+--source include/wait_until_connected_again.inc
+
+--connection slave
+--replace_result $SERVER_MYPORT_1 SERVER_MYPORT_1
+eval CHANGE MASTER TO master_host='127.0.0.1', master_port=$SERVER_MYPORT_1, master_user='root', master_log_file='master-bin.000001', master_log_pos=4;
+
+--source include/start_slave.inc
+
+--connection master
+--sync_slave_with_master
+SELECT * FROM t1 ORDER BY id;
+
+--source include/stop_slave.inc
+DROP TABLE t1;
+
+--let $rpl_only_running_threads= 1
+--source include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/t/rpl_mysql80_stm_temporal_round.test b/mysql-test/suite/rpl/t/rpl_mysql80_stm_temporal_round.test
new file mode 100644
index 00000000000..ad6df9d9993
--- /dev/null
+++ b/mysql-test/suite/rpl/t/rpl_mysql80_stm_temporal_round.test
@@ -0,0 +1,62 @@
+--source include/have_binlog_format_statement.inc
+
+--echo #
+--echo # MDEV-8894 Inserting fractional seconds into MySQL 5.6 master breaks consistency on MariaDB 10 slave
+--echo #
+
+--source include/have_innodb.inc
+--source include/master-slave.inc
+
+--connection slave
+CREATE TABLE t1 (id SERIAL, a DATETIME(3));
+--source include/stop_slave.inc
+
+--connection master
+--let $datadir= `SELECT @@datadir`
+
+--let $rpl_server_number= 1
+--source include/rpl_stop_server.inc
+
+--remove_file $datadir/master-bin.000001
+
+#
+# Simulate MySQL 8.0.x master
+#
+# mysql-8.0.13-stm-temporal-round-binlog.000001 was recorded against a
+# table with this structure:
+#CREATE TABLE t1 (id SERIAL, a DATETIME(3));
+# (note, the CREATE statement is not inside the binary log)
+#
+# using this command line:
+#   mysqld --log-bin --binlog-format=statement --server-id=1 --character-set-server=latin1
+# with the following SQL script:
+#
+#SET NAMES latin1 COLLATE latin1_swedish_ci;
+#SET sql_mode='';
+#INSERT INTO t1 (a) VALUES ('2001-01-01 00:00:00.999999');
+#SET sql_mode=TIME_TRUNCATE_FRACTIONAL;
+#INSERT INTO t1 (a) VALUES ('2001-01-01 00:00:00.999999');
+#
+
+--copy_file $MYSQL_TEST_DIR/std_data/rpl/mysql-8.0.13-stm-temporal-round-binlog.000001 $datadir/master-bin.000001
+
+--let $rpl_server_number= 1
+--source include/rpl_start_server.inc
+
+--source include/wait_until_connected_again.inc
+
+--connection slave
+--replace_result $SERVER_MYPORT_1 SERVER_MYPORT_1
+eval CHANGE MASTER TO master_host='127.0.0.1', master_port=$SERVER_MYPORT_1, master_user='root', master_log_file='master-bin.000001', master_log_pos=4;
+
+--source include/start_slave.inc
+
+--connection master
+--sync_slave_with_master
+SELECT * FROM t1 ORDER BY id;
+
+--source include/stop_slave.inc
+DROP TABLE t1;
+
+--let $rpl_only_running_threads= 1
+--source include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/t/rpl_parallel_optimistic.test b/mysql-test/suite/rpl/t/rpl_parallel_optimistic.test
index e08472d5f51..0060cf4416c 100644
--- a/mysql-test/suite/rpl/t/rpl_parallel_optimistic.test
+++ b/mysql-test/suite/rpl/t/rpl_parallel_optimistic.test
@@ -21,6 +21,10 @@ SET GLOBAL slave_parallel_threads=10;
 CHANGE MASTER TO master_use_gtid=slave_pos;
 SET @old_parallel_mode=@@GLOBAL.slave_parallel_mode;
 SET GLOBAL slave_parallel_mode='optimistic';
+# Run the first part of the test with high batch size and see that
+# old rows remain in the table.
+SET @old_gtid_cleanup_batch_size= @@GLOBAL.gtid_cleanup_batch_size;
+SET GLOBAL gtid_cleanup_batch_size= 1000000;
 
 
 --connection server_1
@@ -108,7 +112,12 @@ SELECT * FROM t3 ORDER BY c;
 SELECT * FROM t1 ORDER BY a;
 SELECT * FROM t2 ORDER BY a;
 SELECT * FROM t3 ORDER BY c;
-#SHOW STATUS LIKE 'Slave_retried_transactions';
+# Check that we have a bunch of old rows left-over - they were not deleted
+# due to high @@gtid_cleanup_batch_size. Then set a low
+# @@gtid_cleanup_batch_size so we can test that rows start being deleted.
+SELECT IF(COUNT(*) >= 30, "OK", CONCAT("Error: too few old rows found: ", COUNT(*)))
+  FROM mysql.gtid_slave_pos;
+SET GLOBAL gtid_cleanup_batch_size=1;
 
 
 --echo *** Test @@skip_parallel_replication. ***
@@ -557,25 +566,18 @@ DROP TABLE t1, t2, t3;
 
 --connection server_2
 --source include/sync_with_master_gtid.inc
-# Check for left-over rows in table mysql.gtid_slave_pos (MDEV-12147).
-#
-# There was a bug when a transaction got a conflict and was rolled back. It
-# might have also handled deletion of some old rows, and these deletions would
-# then also be rolled back. And since the deletes were never re-tried, old no
-# longer needed rows would accumulate in the table without limit.
-# 
-# The earlier part of this test file have plenty of transactions being rolled
-# back. But the last DROP TABLE statement runs on its own and should never
-# conflict, thus at this point the mysql.gtid_slave_pos table should be clean.
-#
-# To support @@gtid_pos_auto_engines, when a row is inserted in the table, it
-# is associated with the engine of the table at insertion time, and it will
-# only be deleted during record_gtid from a table of the same engine. Since we
-# alter the table from MyISAM to InnoDB at the start of this test, we should
-# end up with 4 rows: two left-over from when the table was MyISAM, and two
-# left-over from the InnoDB part.
---echo Check that no more than the expected last four GTIDs are in mysql.gtid_slave_pos
-select count(4) <= 4 from mysql.gtid_slave_pos order by domain_id, sub_id;
+# Check that old rows are deleted from mysql.gtid_slave_pos.
+# Deletion is asynchronous, so use wait_condition.inc.
+# Also, there is a small amount of non-determinism in the deletion of old
+# rows, so it is not guaranteed that there can never be more than
+# @@gtid_cleanup_batch_size rows in the table; so allow a bit of slack
+# here.
+let $wait_condition=
+  SELECT COUNT(*) <= 5*@@GLOBAL.gtid_cleanup_batch_size
+    FROM mysql.gtid_slave_pos;
+--source include/wait_condition.inc
+eval $wait_condition;
+SET GLOBAL gtid_cleanup_batch_size= @old_gtid_cleanup_batch_size;
 
 --connection server_1
 --source include/rpl_end.inc
diff --git a/mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result b/mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result
index 56f7a136983..71761df1ab3 100644
--- a/mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result
+++ b/mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result
@@ -1202,6 +1202,20 @@ NUMERIC_BLOCK_SIZE	NULL
 ENUM_VALUE_LIST	NULL
 READ_ONLY	NO
 COMMAND_LINE_ARGUMENT	NULL
+VARIABLE_NAME	GTID_CLEANUP_BATCH_SIZE
+SESSION_VALUE	NULL
+GLOBAL_VALUE	64
+GLOBAL_VALUE_ORIGIN	COMPILE-TIME
+DEFAULT_VALUE	64
+VARIABLE_SCOPE	GLOBAL
+VARIABLE_TYPE	INT UNSIGNED
+VARIABLE_COMMENT	Normally does not need tuning. How many old rows must accumulate in the mysql.gtid_slave_pos table before a background job will be run to delete them. Can be increased to reduce number of commits if using many different engines with --gtid_pos_auto_engines, or to reduce CPU overhead if using a huge number of different gtid_domain_ids. Can be decreased to reduce number of old rows in the table.
+NUMERIC_MIN_VALUE	0
+NUMERIC_MAX_VALUE	2147483647
+NUMERIC_BLOCK_SIZE	1
+ENUM_VALUE_LIST	NULL
+READ_ONLY	NO
+COMMAND_LINE_ARGUMENT	REQUIRED
 VARIABLE_NAME	GTID_CURRENT_POS
 SESSION_VALUE	NULL
 GLOBAL_VALUE	
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 651fb4ce5b1..3ea6be95dda 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -101,16 +101,11 @@ TYPELIB binlog_checksum_typelib=
    TODO: correct the constant when it has been determined 
    (which main tree to push and when) 
 */
-const uchar checksum_version_split_mysql[3]= {5, 6, 1};
-const ulong checksum_version_product_mysql=
-  (checksum_version_split_mysql[0] * 256 +
-   checksum_version_split_mysql[1]) * 256 +
-  checksum_version_split_mysql[2];
-const uchar checksum_version_split_mariadb[3]= {5, 3, 0};
-const ulong checksum_version_product_mariadb=
-  (checksum_version_split_mariadb[0] * 256 +
-   checksum_version_split_mariadb[1]) * 256 +
-  checksum_version_split_mariadb[2];
+const Version checksum_version_split_mysql(5, 6, 1);
+const Version checksum_version_split_mariadb(5, 3, 0);
+
+// First MySQL version with fraction seconds
+const Version fsp_version_split_mysql(5, 6, 0);
 
 #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
 static int rows_event_stmt_cleanup(rpl_group_info *rgi, THD* thd);
@@ -4541,6 +4536,7 @@ code_name(int code)
 }
 #endif
 
+
 /**
    Macro to check that there is enough space to read from memory.
 
@@ -4764,6 +4760,30 @@ Query_log_event::Query_log_event(const char* buf, uint event_len,
     }
   }
 
+#if !defined(MYSQL_CLIENT)
+  if (description_event->server_version_split.kind ==
+      Format_description_log_event::master_version_split::KIND_MYSQL)
+  {
+    // Handle MariaDB/MySQL incompatible sql_mode bits
+    sql_mode_t mysql_sql_mode= sql_mode;
+    sql_mode&= MODE_MASK_MYSQL_COMPATIBLE; // Unset MySQL specific bits
+
+    /*
+      sql_mode flags related to fraction second rounding/truncation
+      have opposite meaning in MySQL vs MariaDB.
+      MySQL:
+       - rounds fractional seconds by default
+       - truncates if TIME_TRUNCATE_FRACTIONAL is set
+      MariaDB:
+       - truncates fractional seconds by default
+       - rounds if TIME_ROUND_FRACTIONAL is set
+    */
+    if (description_event->server_version_split >= fsp_version_split_mysql &&
+       !(mysql_sql_mode & MODE_MYSQL80_TIME_TRUNCATE_FRACTIONAL))
+      sql_mode|= MODE_TIME_ROUND_FRACTIONAL;
+  }
+#endif
+
   /**
     Layout for the data buffer is as follows
     +--------+-----------+------+------+---------+----+-------+
@@ -5581,7 +5601,7 @@ int Query_log_event::do_apply_event(rpl_group_info *rgi,
           gtid= rgi->current_gtid;
           if (unlikely(rpl_global_gtid_slave_state->record_gtid(thd, &gtid,
                                                                 sub_id,
-                                                                rgi, false,
+                                                                true, false,
                                                                 &hton)))
           {
             int errcode= thd->get_stmt_da()->sql_errno();
@@ -6549,26 +6569,24 @@ bool Format_description_log_event::start_decryption(Start_encryption_log_event*
   return crypto_data.init(sele->crypto_scheme, sele->key_version);
 }
 
-static inline void
-do_server_version_split(char* version,
-                        Format_description_log_event::master_version_split *split_versions)
+
+Version::Version(const char *version, const char **endptr)
 {
-  char *p= version, *r;
+  const char *p= version;
   ulong number;
   for (uint i= 0; i<=2; i++)
   {
+    char *r;
     number= strtoul(p, &r, 10);
     /*
       It is an invalid version if any version number greater than 255 or
       first number is not followed by '.'.
     */
     if (number < 256 && (*r == '.' || i != 0))
-      split_versions->ver[i]= (uchar) number;
+      m_ver[i]= (uchar) number;
     else
     {
-      split_versions->ver[0]= 0;
-      split_versions->ver[1]= 0;
-      split_versions->ver[2]= 0;
+      *this= Version();
       break;
     }
 
@@ -6576,12 +6594,19 @@ do_server_version_split(char* version,
     if (*r == '.')
       p++; // skip the dot
   }
+  endptr[0]= p;
+}
+
+
+Format_description_log_event::
+  master_version_split::master_version_split(const char *version)
+{
+  const char *p;
+  static_cast<Version*>(this)[0]= Version(version, &p);
   if (strstr(p, "MariaDB") != 0 || strstr(p, "-maria-") != 0)
-    split_versions->kind=
-      Format_description_log_event::master_version_split::KIND_MARIADB;
+    kind= KIND_MARIADB;
   else
-    split_versions->kind=
-      Format_description_log_event::master_version_split::KIND_MYSQL;
+    kind= KIND_MYSQL;
 }
 
 
@@ -6595,20 +6620,14 @@ do_server_version_split(char* version,
 */
 void Format_description_log_event::calc_server_version_split()
 {
-  do_server_version_split(server_version, &server_version_split);
+  server_version_split= master_version_split(server_version);
 
   DBUG_PRINT("info",("Format_description_log_event::server_version_split:"
                      " '%s' %d %d %d", server_version,
-                     server_version_split.ver[0],
-                     server_version_split.ver[1], server_version_split.ver[2]));
+                     server_version_split[0],
+                     server_version_split[1], server_version_split[2]));
 }
 
-static inline ulong
-version_product(const Format_description_log_event::master_version_split* version_split)
-{
-  return ((version_split->ver[0] * 256 + version_split->ver[1]) * 256
-          + version_split->ver[2]);
-}
 
 /**
    @return TRUE is the event's version is earlier than one that introduced
@@ -6618,9 +6637,9 @@ bool
 Format_description_log_event::is_version_before_checksum(const master_version_split
                                                          *version_split)
 {
-  return version_product(version_split) <
+  return *version_split <
     (version_split->kind == master_version_split::KIND_MARIADB ?
-     checksum_version_product_mariadb : checksum_version_product_mysql);
+     checksum_version_split_mariadb : checksum_version_split_mysql);
 }
 
 /**
@@ -6636,7 +6655,6 @@ enum enum_binlog_checksum_alg get_checksum_alg(const char* buf, ulong len)
 {
   enum enum_binlog_checksum_alg ret;
   char version[ST_SERVER_VER_LEN];
-  Format_description_log_event::master_version_split version_split;
 
   DBUG_ENTER("get_checksum_alg");
   DBUG_ASSERT(buf[EVENT_TYPE_OFFSET] == FORMAT_DESCRIPTION_EVENT);
@@ -6646,7 +6664,7 @@ enum enum_binlog_checksum_alg get_checksum_alg(const char* buf, ulong len)
          ST_SERVER_VER_LEN);
   version[ST_SERVER_VER_LEN - 1]= 0;
   
-  do_server_version_split(version, &version_split);
+  Format_description_log_event::master_version_split version_split(version);
   ret= Format_description_log_event::is_version_before_checksum(&version_split)
     ? BINLOG_CHECKSUM_ALG_UNDEF
     : (enum_binlog_checksum_alg)buf[len - BINLOG_CHECKSUM_LEN - BINLOG_CHECKSUM_ALG_DESC_LEN];
@@ -8378,7 +8396,7 @@ Gtid_list_log_event::do_apply_event(rpl_group_info *rgi)
     {
       if ((ret= rpl_global_gtid_slave_state->record_gtid(thd, &list[i],
                                                          sub_id_list[i],
-                                                         NULL, false, &hton)))
+                                                         false, false, &hton)))
         return ret;
       rpl_global_gtid_slave_state->update_state_hash(sub_id_list[i], &list[i],
                                                      hton, NULL);
@@ -8915,7 +8933,7 @@ int Xid_log_event::do_apply_event(rpl_group_info *rgi)
     rgi->gtid_pending= false;
 
     gtid= rgi->current_gtid;
-    err= rpl_global_gtid_slave_state->record_gtid(thd, &gtid, sub_id, rgi,
+    err= rpl_global_gtid_slave_state->record_gtid(thd, &gtid, sub_id, true,
                                                   false, &hton);
     if (unlikely(err))
     {
diff --git a/sql/log_event.h b/sql/log_event.h
index 84025554ee9..38a40c90799 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -2728,6 +2728,38 @@ protected:
 };
 
 
+class Version
+{
+protected:
+  uchar m_ver[3];
+  int cmp(const Version &other) const
+  {
+    return memcmp(m_ver, other.m_ver, 3);
+  }
+public:
+  Version()
+  {
+    m_ver[0]= m_ver[1]= m_ver[2]= '\0';
+  }
+  Version(uchar v0, uchar v1, uchar v2)
+  {
+    m_ver[0]= v0;
+    m_ver[1]= v1;
+    m_ver[2]= v2;
+  }
+  Version(const char *version, const char **endptr);
+  const uchar& operator [] (size_t i) const
+  {
+    DBUG_ASSERT(i < 3);
+    return m_ver[i];
+  }
+  bool operator<(const Version &other) const { return cmp(other) < 0; }
+  bool operator>(const Version &other) const { return cmp(other) > 0; }
+  bool operator<=(const Version &other) const { return cmp(other) <= 0; }
+  bool operator>=(const Version &other) const { return cmp(other) >= 0; }
+};
+
+
 /**
   @class Format_description_log_event
 
@@ -2754,10 +2786,17 @@ public:
      by the checksum alg decription byte
   */
   uint8 *post_header_len;
-  struct master_version_split {
+  class master_version_split: public Version {
+  public:
     enum {KIND_MYSQL, KIND_MARIADB};
     int kind;
-    uchar ver[3];
+    master_version_split() :kind(KIND_MARIADB) { }
+    master_version_split(const char *version);
+    bool version_is_valid() const
+    {
+      /* It is invalid only when all version numbers are 0 */
+      return !(m_ver[0] == 0 && m_ver[1] == 0 && m_ver[2] == 0);
+    }
   };
   master_version_split server_version_split;
   const uint8 *event_type_permutation;
@@ -2781,17 +2820,9 @@ public:
             (post_header_len != NULL));
   }
 
-  bool version_is_valid() const
-  {
-    /* It is invalid only when all version numbers are 0 */
-    return !(server_version_split.ver[0] == 0 &&
-             server_version_split.ver[1] == 0 &&
-             server_version_split.ver[2] == 0);
-  }
-
   bool is_valid() const
   {
-    return header_is_valid() && version_is_valid();
+    return header_is_valid() && server_version_split.version_is_valid();
   }
 
   int get_data_size()
diff --git a/sql/mysql_install_db.cc b/sql/mysql_install_db.cc
index fc8bce08276..6d09726481b 100644
--- a/sql/mysql_install_db.cc
+++ b/sql/mysql_install_db.cc
@@ -198,7 +198,7 @@ int main(int argc, char **argv)
     die("database creation failed");
   }
 
-  printf("Creation of the database was successful");
+  printf("Creation of the database was successful\n");
   return 0;
 }
 
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 9ff47dc1ff1..0f116395fe0 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -568,6 +568,7 @@ ulong opt_binlog_commit_wait_count= 0;
 ulong opt_binlog_commit_wait_usec= 0;
 ulong opt_slave_parallel_max_queued= 131072;
 my_bool opt_gtid_ignore_duplicates= FALSE;
+uint opt_gtid_cleanup_batch_size= 64;
 
 const double log_10[] = {
   1e000, 1e001, 1e002, 1e003, 1e004, 1e005, 1e006, 1e007, 1e008, 1e009,
diff --git a/sql/mysqld.h b/sql/mysqld.h
index 75f35a6df24..9c02b9127a2 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -262,6 +262,7 @@ extern ulong opt_slave_parallel_mode;
 extern ulong opt_binlog_commit_wait_count;
 extern ulong opt_binlog_commit_wait_usec;
 extern my_bool opt_gtid_ignore_duplicates;
+extern uint opt_gtid_cleanup_batch_size;
 extern ulong back_log;
 extern ulong executed_events;
 extern char language[FN_REFLEN];
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index d6db365a8a2..231dcecf6f5 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -2869,7 +2869,9 @@ bool create_key_parts_for_pseudo_indexes(RANGE_OPT_PARAM *param,
     if (bitmap_is_set(used_fields, (*field_ptr)->field_index))
     {
       Field *field= *field_ptr;
-      if (field->type() == MYSQL_TYPE_GEOMETRY)
+      Column_statistics* col_stats= field->read_stats;
+      if (field->type() == MYSQL_TYPE_GEOMETRY ||
+           !col_stats || col_stats->no_stat_values_provided())
         continue;
 
       uint16 store_length;
diff --git a/sql/rpl_gtid.cc b/sql/rpl_gtid.cc
index 322b84130f2..17f474c2acf 100644
--- a/sql/rpl_gtid.cc
+++ b/sql/rpl_gtid.cc
@@ -79,7 +79,7 @@ rpl_slave_state::record_and_update_gtid(THD *thd, rpl_group_info *rgi)
     rgi->gtid_pending= false;
     if (rgi->gtid_ignore_duplicate_state!=rpl_group_info::GTID_DUPLICATE_IGNORE)
     {
-      if (record_gtid(thd, &rgi->current_gtid, sub_id, NULL, false, &hton))
+      if (record_gtid(thd, &rgi->current_gtid, sub_id, false, false, &hton))
         DBUG_RETURN(1);
       update_state_hash(sub_id, &rgi->current_gtid, hton, rgi);
     }
@@ -244,7 +244,7 @@ rpl_slave_state_free_element(void *arg)
 
 
 rpl_slave_state::rpl_slave_state()
-  : last_sub_id(0), gtid_pos_tables(0), loaded(false)
+  : pending_gtid_count(0), last_sub_id(0), gtid_pos_tables(0), loaded(false)
 {
   mysql_mutex_init(key_LOCK_slave_state, &LOCK_slave_state,
                    MY_MUTEX_INIT_SLOW);
@@ -331,14 +331,11 @@ rpl_slave_state::update(uint32 domain_id, uint32 server_id, uint64 sub_id,
       }
     }
     rgi->gtid_ignore_duplicate_state= rpl_group_info::GTID_DUPLICATE_NULL;
-
-#ifdef HAVE_REPLICATION
-    rgi->pending_gtid_deletes_clear();
-#endif
   }
 
   if (!(list_elem= (list_element *)my_malloc(sizeof(*list_elem), MYF(MY_WME))))
     return 1;
+  list_elem->domain_id= domain_id;
   list_elem->server_id= server_id;
   list_elem->sub_id= sub_id;
   list_elem->seq_no= seq_no;
@@ -348,6 +345,15 @@ rpl_slave_state::update(uint32 domain_id, uint32 server_id, uint64 sub_id,
   if (last_sub_id < sub_id)
     last_sub_id= sub_id;
 
+#ifdef HAVE_REPLICATION
+  ++pending_gtid_count;
+  if (pending_gtid_count >= opt_gtid_cleanup_batch_size)
+  {
+    pending_gtid_count = 0;
+    slave_background_gtid_pending_delete_request();
+  }
+#endif
+
   return 0;
 }
 
@@ -382,20 +388,22 @@ rpl_slave_state::get_element(uint32 domain_id)
 
 
 int
-rpl_slave_state::put_back_list(uint32 domain_id, list_element *list)
+rpl_slave_state::put_back_list(list_element *list)
 {
-  element *e;
+  element *e= NULL;
   int err= 0;
 
   mysql_mutex_lock(&LOCK_slave_state);
-  if (!(e= (element *)my_hash_search(&hash, (const uchar *)&domain_id, 0)))
-  {
-    err= 1;
-    goto end;
-  }
   while (list)
   {
     list_element *next= list->next;
+
+    if ((!e || e->domain_id != list->domain_id) &&
+        !(e= (element *)my_hash_search(&hash, (const uchar *)&list->domain_id, 0)))
+    {
+      err= 1;
+      goto end;
+    }
     e->add(list);
     list= next;
   }
@@ -572,12 +580,12 @@ rpl_slave_state::select_gtid_pos_table(THD *thd, LEX_CSTRING *out_tablename)
 /*
   Write a gtid to the replication slave state table.
 
+  Do it as part of the transaction, to get slave crash safety, or as a separate
+  transaction if !in_transaction (eg. MyISAM or DDL).
+
     gtid    The global transaction id for this event group.
     sub_id  Value allocated within the sub_id when the event group was
             read (sub_id must be consistent with commit order in master binlog).
-    rgi     rpl_group_info context, if we are recording the gtid transactionally
-            as part of replicating a transactional event. NULL if called from
-            outside of a replicated transaction.
 
   Note that caller must later ensure that the new gtid and sub_id is inserted
   into the appropriate HASH element with rpl_slave_state.add(), so that it can
@@ -585,16 +593,13 @@ rpl_slave_state::select_gtid_pos_table(THD *thd, LEX_CSTRING *out_tablename)
 */
 int
 rpl_slave_state::record_gtid(THD *thd, const rpl_gtid *gtid, uint64 sub_id,
-                             rpl_group_info *rgi, bool in_statement,
+                             bool in_transaction, bool in_statement,
                              void **out_hton)
 {
   TABLE_LIST tlist;
   int err= 0, not_sql_thread;
   bool table_opened= false;
   TABLE *table;
-  list_element *delete_list= 0, *next, *cur, **next_ptr_ptr, **best_ptr_ptr;
-  uint64 best_sub_id;
-  element *elem;
   ulonglong thd_saved_option= thd->variables.option_bits;
   Query_tables_list lex_backup;
   wait_for_commit* suspended_wfc;
@@ -684,7 +689,7 @@ rpl_slave_state::record_gtid(THD *thd, const rpl_gtid *gtid, uint64 sub_id,
   thd->wsrep_ignore_table= true;
 #endif
 
-  if (!rgi)
+  if (!in_transaction)
   {
     DBUG_PRINT("info", ("resetting OPTION_BEGIN"));
     thd->variables.option_bits&=
@@ -716,168 +721,280 @@ rpl_slave_state::record_gtid(THD *thd, const rpl_gtid *gtid, uint64 sub_id,
     my_error(ER_OUT_OF_RESOURCES, MYF(0));
     goto end;
   }
+end:
 
-  mysql_mutex_lock(&LOCK_slave_state);
-  if ((elem= get_element(gtid->domain_id)) == NULL)
+#ifdef WITH_WSREP
+  thd->wsrep_ignore_table= false;
+#endif
+
+  if (table_opened)
   {
-    mysql_mutex_unlock(&LOCK_slave_state);
-    my_error(ER_OUT_OF_RESOURCES, MYF(0));
-    err= 1;
-    goto end;
+    if (err || (err= ha_commit_trans(thd, FALSE)))
+      ha_rollback_trans(thd, FALSE);
+
+    close_thread_tables(thd);
+    if (in_transaction)
+      thd->mdl_context.release_statement_locks();
+    else
+      thd->mdl_context.release_transactional_locks();
   }
+  thd->lex->restore_backup_query_tables_list(&lex_backup);
+  thd->variables.option_bits= thd_saved_option;
+  thd->resume_subsequent_commits(suspended_wfc);
+  DBUG_EXECUTE_IF("inject_record_gtid_serverid_100_sleep",
+    {
+      if (gtid->server_id == 100)
+        my_sleep(500000);
+    });
+  DBUG_RETURN(err);
+}
 
-  /* Now pull out all GTIDs that were recorded in this engine. */
-  delete_list = NULL;
-  next_ptr_ptr= &elem->list;
-  cur= elem->list;
-  best_sub_id= 0;
-  best_ptr_ptr= NULL;
-  while (cur)
+
+/*
+  Return a list of all old GTIDs in any mysql.gtid_slave_pos* table that are
+  no longer needed and can be deleted from the table.
+
+  Within each domain, we need to keep around the latest GTID (the one with the
+  highest sub_id), but any others in that domain can be deleted.
+*/
+rpl_slave_state::list_element *
+rpl_slave_state::gtid_grab_pending_delete_list()
+{
+  uint32 i;
+  list_element *full_list;
+
+  mysql_mutex_lock(&LOCK_slave_state);
+  full_list= NULL;
+  for (i= 0; i < hash.records; ++i)
   {
-    list_element *next= cur->next;
-    if (cur->hton == hton)
-    {
-      /* Belongs to same engine, so move it to the delete list. */
-      cur->next= delete_list;
-      delete_list= cur;
-      if (cur->sub_id > best_sub_id)
+    element *elem= (element *)my_hash_element(&hash, i);
+    list_element *elist= elem->list;
+    list_element *last_elem, **best_ptr_ptr, *cur, *next;
+    uint64 best_sub_id;
+
+    if (!elist)
+      continue;                                 /* Nothing here */
+
+    /* Delete any old stuff, but keep around the most recent one. */
+    cur= elist;
+    best_sub_id= cur->sub_id;
+    best_ptr_ptr= &elist;
+    last_elem= cur;
+    while ((next= cur->next)) {
+      last_elem= next;
+      if (next->sub_id > best_sub_id)
       {
-        best_sub_id= cur->sub_id;
-        best_ptr_ptr= &delete_list;
-      }
-      else if (best_ptr_ptr == &delete_list)
+        best_sub_id= next->sub_id;
         best_ptr_ptr= &cur->next;
-    }
-    else
-    {
-      /* Another engine, leave it in the list. */
-      if (cur->sub_id > best_sub_id)
-      {
-        best_sub_id= cur->sub_id;
-        /* Current best is not on the delete list. */
-        best_ptr_ptr= NULL;
       }
-      *next_ptr_ptr= cur;
-      next_ptr_ptr= &cur->next;
+      cur= next;
     }
-    cur= next;
-  }
-  *next_ptr_ptr= NULL;
-  /*
-    If the highest sub_id element is on the delete list, put it back on the
-    original list, to preserve the highest sub_id element in the table for
-    GTID position recovery.
-  */
-  if (best_ptr_ptr)
-  {
+    /*
+      Append the new elements to the full list. Note the order is important;
+      we do it here so that we do not break the list if best_sub_id is the
+      last of the new elements.
+    */
+    last_elem->next= full_list;
+    /*
+      Delete the highest sub_id element from the old list, and put it back as
+      the single-element new list.
+    */
     cur= *best_ptr_ptr;
     *best_ptr_ptr= cur->next;
-    cur->next= elem->list;
+    cur->next= NULL;
     elem->list= cur;
+
+    /*
+      Collect the full list so far here. Note that elist may have moved if we
+      deleted the first element, so order is again important.
+    */
+    full_list= elist;
   }
   mysql_mutex_unlock(&LOCK_slave_state);
 
-  if (!delete_list)
-    goto end;
+  return full_list;
+}
+
 
-  /* Now delete any already committed GTIDs. */
-  bitmap_set_bit(table->read_set, table->field[0]->field_index);
-  bitmap_set_bit(table->read_set, table->field[1]->field_index);
+/* Find the mysql.gtid_slave_posXXX table associated with a given hton. */
+LEX_CSTRING *
+rpl_slave_state::select_gtid_pos_table(void *hton)
+{
+  struct gtid_pos_table *table_entry;
 
-  if ((err= table->file->ha_index_init(0, 0)))
+  /*
+    See comments on rpl_slave_state::gtid_pos_tables for rules around proper
+    access to the list.
+  */
+  table_entry= (struct gtid_pos_table *)
+    my_atomic_loadptr_explicit(&gtid_pos_tables, MY_MEMORY_ORDER_ACQUIRE);
+
+  while (table_entry)
   {
-    table->file->print_error(err, MYF(0));
-    goto end;
+    if (table_entry->table_hton == hton)
+    {
+      if (likely(table_entry->state == GTID_POS_AVAILABLE))
+        return &table_entry->table_name;
+    }
+    table_entry= table_entry->next;
   }
-  cur = delete_list;
-  while (cur)
-  {
-    uchar key_buffer[4+8];
 
-    DBUG_EXECUTE_IF("gtid_slave_pos_simulate_failed_delete",
-                    { err= ENOENT;
-                      table->file->print_error(err, MYF(0));
-                      /* `break' does not work inside DBUG_EXECUTE_IF */
-                      goto dbug_break; });
+  table_entry= (struct gtid_pos_table *)
+    my_atomic_loadptr_explicit(&default_gtid_pos_table, MY_MEMORY_ORDER_ACQUIRE);
+  return &table_entry->table_name;
+}
 
-    next= cur->next;
 
-    table->field[1]->store(cur->sub_id, true);
-    /* domain_id is already set in table->record[0] from write_row() above. */
-    key_copy(key_buffer, table->record[0], &table->key_info[0], 0, false);
-    if (table->file->ha_index_read_map(table->record[1], key_buffer,
-                                       HA_WHOLE_KEY, HA_READ_KEY_EXACT))
-      /* We cannot find the row, assume it is already deleted. */
-      ;
-    else if ((err= table->file->ha_delete_row(table->record[1])))
-      table->file->print_error(err, MYF(0));
-    /*
-      In case of error, we still discard the element from the list. We do
-      not want to endlessly error on the same element in case of table
-      corruption or such.
-    */
-    cur= next;
-    if (err)
-      break;
-  }
-IF_DBUG(dbug_break:, )
-  table->file->ha_index_end();
+void
+rpl_slave_state::gtid_delete_pending(THD *thd,
+                                     rpl_slave_state::list_element **list_ptr)
+{
+  int err= 0;
+  ulonglong thd_saved_option;
 
-end:
+  if (unlikely(!loaded))
+    return;
 
 #ifdef WITH_WSREP
-  thd->wsrep_ignore_table= false;
+  /*
+    Updates in slave state table should not be appended to galera transaction
+    writeset.
+  */
+  thd->wsrep_ignore_table= true;
 #endif
 
-  if (table_opened)
+  thd_saved_option= thd->variables.option_bits;
+  thd->variables.option_bits&=
+    ~(ulonglong)(OPTION_NOT_AUTOCOMMIT |OPTION_BEGIN |OPTION_BIN_LOG |
+                 OPTION_GTID_BEGIN);
+
+  while (*list_ptr)
   {
-    if (err || (err= ha_commit_trans(thd, FALSE)))
-    {
-      /*
-        If error, we need to put any remaining delete_list back into the HASH
-        so we can do another delete attempt later.
-      */
-      if (delete_list)
-      {
-        put_back_list(gtid->domain_id, delete_list);
-        delete_list = 0;
-      }
+    LEX_CSTRING *gtid_pos_table_name, *tmp_table_name;
+    Query_tables_list lex_backup;
+    TABLE_LIST tlist;
+    TABLE *table;
+    handler::Table_flags direct_pos;
+    list_element *cur, **cur_ptr_ptr;
+    bool table_opened= false;
+    void *hton= (*list_ptr)->hton;
 
-      ha_rollback_trans(thd, FALSE);
+    thd->reset_for_next_command();
+
+    /*
+      Only the SQL thread can call select_gtid_pos_table without a mutex
+      Other threads needs to use a mutex and take into account that the
+      result may change during execution, so we have to make a copy.
+    */
+    mysql_mutex_lock(&LOCK_slave_state);
+    tmp_table_name= select_gtid_pos_table(hton);
+    gtid_pos_table_name= thd->make_clex_string(tmp_table_name->str,
+                                               tmp_table_name->length);
+    mysql_mutex_unlock(&LOCK_slave_state);
+    if (!gtid_pos_table_name)
+    {
+      /* Out of memory - we can try again later. */
+      break;
     }
-    close_thread_tables(thd);
-    if (rgi)
+
+    thd->lex->reset_n_backup_query_tables_list(&lex_backup);
+    tlist.init_one_table(&MYSQL_SCHEMA_NAME, gtid_pos_table_name, NULL, TL_WRITE);
+    if ((err= open_and_lock_tables(thd, &tlist, FALSE, 0)))
+      goto end;
+    table_opened= true;
+    table= tlist.table;
+
+    if ((err= gtid_check_rpl_slave_state_table(table)))
+      goto end;
+
+    direct_pos= table->file->ha_table_flags() & HA_PRIMARY_KEY_REQUIRED_FOR_POSITION;
+    bitmap_set_all(table->write_set);
+    table->rpl_write_set= table->write_set;
+
+    /* Now delete any already committed GTIDs. */
+    bitmap_set_bit(table->read_set, table->field[0]->field_index);
+    bitmap_set_bit(table->read_set, table->field[1]->field_index);
+
+    if (!direct_pos && (err= table->file->ha_index_init(0, 0)))
     {
-      thd->mdl_context.release_statement_locks();
-      /*
-        Save the list of old gtid entries we deleted. If this transaction
-        fails later for some reason and is rolled back, the deletion of those
-        entries will be rolled back as well, and we will need to put them back
-        on the to-be-deleted list so we can re-do the deletion. Otherwise
-        redundant rows in mysql.gtid_slave_pos may accumulate if transactions
-        are rolled back and retried after record_gtid().
-      */
-#ifdef HAVE_REPLICATION
-      rgi->pending_gtid_deletes_save(gtid->domain_id, delete_list);
-#endif
+      table->file->print_error(err, MYF(0));
+      goto end;
     }
-    else
+
+    cur = *list_ptr;
+    cur_ptr_ptr = list_ptr;
+    do
     {
-      thd->mdl_context.release_transactional_locks();
-#ifdef HAVE_REPLICATION
-      rpl_group_info::pending_gtid_deletes_free(delete_list);
-#endif
+      uchar key_buffer[4+8];
+      list_element *next= cur->next;
+
+      if (cur->hton == hton)
+      {
+        int res;
+
+        table->field[0]->store((ulonglong)cur->domain_id, true);
+        table->field[1]->store(cur->sub_id, true);
+        if (direct_pos)
+        {
+          res= table->file->ha_rnd_pos_by_record(table->record[0]);
+        }
+        else
+        {
+          key_copy(key_buffer, table->record[0], &table->key_info[0], 0, false);
+          res= table->file->ha_index_read_map(table->record[0], key_buffer,
+                                              HA_WHOLE_KEY, HA_READ_KEY_EXACT);
+        }
+        DBUG_EXECUTE_IF("gtid_slave_pos_simulate_failed_delete",
+              { res= 1;
+                err= ENOENT;
+                sql_print_error("<DEBUG> Error deleting old GTID row");
+              });
+        if (res)
+          /* We cannot find the row, assume it is already deleted. */
+          ;
+        else if ((err= table->file->ha_delete_row(table->record[0])))
+        {
+          sql_print_error("Error deleting old GTID row: %s",
+                          thd->get_stmt_da()->message());
+          /*
+            In case of error, we still discard the element from the list. We do
+            not want to endlessly error on the same element in case of table
+            corruption or such.
+          */
+        }
+        *cur_ptr_ptr= next;
+        my_free(cur);
+      }
+      else
+      {
+        /* Leave this one in the list until we get to the table for its hton. */
+        cur_ptr_ptr= &cur->next;
+      }
+      cur= next;
+      if (err)
+        break;
+    } while (cur);
+end:
+    if (table_opened)
+    {
+      if (!direct_pos)
+        table->file->ha_index_end();
+
+      if (err || (err= ha_commit_trans(thd, FALSE)))
+        ha_rollback_trans(thd, FALSE);
     }
+    close_thread_tables(thd);
+    thd->mdl_context.release_transactional_locks();
+    thd->lex->restore_backup_query_tables_list(&lex_backup);
+
+    if (err)
+      break;
   }
-  thd->lex->restore_backup_query_tables_list(&lex_backup);
   thd->variables.option_bits= thd_saved_option;
-  thd->resume_subsequent_commits(suspended_wfc);
-  DBUG_EXECUTE_IF("inject_record_gtid_serverid_100_sleep",
-    {
-      if (gtid->server_id == 100)
-        my_sleep(500000);
-    });
-  DBUG_RETURN(err);
+
+#ifdef WITH_WSREP
+  thd->wsrep_ignore_table= false;
+#endif
 }
 
 
@@ -1251,7 +1368,7 @@ rpl_slave_state::load(THD *thd, const char *state_from_master, size_t len,
 
     if (gtid_parser_helper(&state_from_master, end, &gtid) ||
         !(sub_id= next_sub_id(gtid.domain_id)) ||
-        record_gtid(thd, &gtid, sub_id, NULL, in_statement, &hton) ||
+        record_gtid(thd, &gtid, sub_id, false, in_statement, &hton) ||
         update(gtid.domain_id, gtid.server_id, sub_id, gtid.seq_no, hton, NULL))
       return 1;
     if (state_from_master == end)
diff --git a/sql/rpl_gtid.h b/sql/rpl_gtid.h
index 0fc92d5e33c..60d822f7b0d 100644
--- a/sql/rpl_gtid.h
+++ b/sql/rpl_gtid.h
@@ -118,8 +118,9 @@ struct rpl_slave_state
   {
     struct list_element *next;
     uint64 sub_id;
-    uint64 seq_no;
+    uint32 domain_id;
     uint32 server_id;
+    uint64 seq_no;
     /*
       hton of mysql.gtid_slave_pos* table used to record this GTID.
       Can be NULL if the gtid table failed to load (eg. missing
@@ -191,6 +192,8 @@ struct rpl_slave_state
 
   /* Mapping from domain_id to its element. */
   HASH hash;
+  /* GTIDs added since last purge of old mysql.gtid_slave_pos rows. */
+  uint32 pending_gtid_count;
   /* Mutex protecting access to the state. */
   mysql_mutex_t LOCK_slave_state;
   /* Auxiliary buffer to sort gtid list. */
@@ -233,7 +236,10 @@ struct rpl_slave_state
   int truncate_state_table(THD *thd);
   void select_gtid_pos_table(THD *thd, LEX_CSTRING *out_tablename);
   int record_gtid(THD *thd, const rpl_gtid *gtid, uint64 sub_id,
-                  rpl_group_info *rgi, bool in_statement, void **out_hton);
+                  bool in_transaction, bool in_statement, void **out_hton);
+  list_element *gtid_grab_pending_delete_list();
+  LEX_CSTRING *select_gtid_pos_table(void *hton);
+  void gtid_delete_pending(THD *thd, rpl_slave_state::list_element **list_ptr);
   uint64 next_sub_id(uint32 domain_id);
   int iterate(int (*cb)(rpl_gtid *, void *), void *data,
               rpl_gtid *extra_gtids, uint32 num_extra,
@@ -245,7 +251,7 @@ struct rpl_slave_state
   bool is_empty();
 
   element *get_element(uint32 domain_id);
-  int put_back_list(uint32 domain_id, list_element *list);
+  int put_back_list(list_element *list);
 
   void update_state_hash(uint64 sub_id, rpl_gtid *gtid, void *hton,
                          rpl_group_info *rgi);
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index b275ad884bd..2d91620c898 100644
--- a/sql/rpl_rli.cc
+++ b/sql/rpl_rli.cc
@@ -1820,6 +1820,7 @@ rpl_load_gtid_slave_state(THD *thd)
   int err= 0;
   uint32 i;
   load_gtid_state_cb_data cb_data;
+  rpl_slave_state::list_element *old_gtids_list;
   DBUG_ENTER("rpl_load_gtid_slave_state");
 
   mysql_mutex_lock(&rpl_global_gtid_slave_state->LOCK_slave_state);
@@ -1905,6 +1906,13 @@ rpl_load_gtid_slave_state(THD *thd)
   rpl_global_gtid_slave_state->loaded= true;
   mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
 
+  /* Clear out no longer needed elements now. */
+  old_gtids_list=
+    rpl_global_gtid_slave_state->gtid_grab_pending_delete_list();
+  rpl_global_gtid_slave_state->gtid_delete_pending(thd, &old_gtids_list);
+  if (old_gtids_list)
+    rpl_global_gtid_slave_state->put_back_list(old_gtids_list);
+
 end:
   if (array_inited)
     delete_dynamic(&array);
@@ -2086,7 +2094,6 @@ rpl_group_info::reinit(Relay_log_info *rli)
   long_find_row_note_printed= false;
   did_mark_start_commit= false;
   gtid_ev_flags2= 0;
-  pending_gtid_delete_list= NULL;
   last_master_timestamp = 0;
   gtid_ignore_duplicate_state= GTID_DUPLICATE_NULL;
   speculation= SPECULATE_NO;
@@ -2217,12 +2224,6 @@ void rpl_group_info::cleanup_context(THD *thd, bool error)
       erroneously update the GTID position.
     */
     gtid_pending= false;
-
-    /*
-      Rollback will have undone any deletions of old rows we might have made
-      in mysql.gtid_slave_pos. Put those rows back on the list to be deleted.
-    */
-    pending_gtid_deletes_put_back();
   }
   m_table_map.clear_tables();
   slave_close_thread_tables(thd);
@@ -2448,78 +2449,6 @@ rpl_group_info::unmark_start_commit()
 }
 
 
-/*
-  When record_gtid() has deleted any old rows from the table
-  mysql.gtid_slave_pos as part of a replicated transaction, save the list of
-  rows deleted here.
-
-  If later the transaction fails (eg. optimistic parallel replication), the
-  deletes will be undone when the transaction is rolled back. Then we can
-  put back the list of rows into the rpl_global_gtid_slave_state, so that
-  we can re-do the deletes and avoid accumulating old rows in the table.
-*/
-void
-rpl_group_info::pending_gtid_deletes_save(uint32 domain_id,
-                                          rpl_slave_state::list_element *list)
-{
-  /*
-    We should never get to a state where we try to save a new pending list of
-    gtid deletes while we still have an old one. But make sure we handle it
-    anyway just in case, so we avoid leaving stray entries in the
-    mysql.gtid_slave_pos table.
-  */
-  DBUG_ASSERT(!pending_gtid_delete_list);
-  if (unlikely(pending_gtid_delete_list))
-    pending_gtid_deletes_put_back();
-
-  pending_gtid_delete_list= list;
-  pending_gtid_delete_list_domain= domain_id;
-}
-
-
-/*
-  Take the list recorded by pending_gtid_deletes_save() and put it back into
-  rpl_global_gtid_slave_state. This is needed if deletion of the rows was
-  rolled back due to transaction failure.
-*/
-void
-rpl_group_info::pending_gtid_deletes_put_back()
-{
-  if (pending_gtid_delete_list)
-  {
-    rpl_global_gtid_slave_state->put_back_list(pending_gtid_delete_list_domain,
-                                               pending_gtid_delete_list);
-    pending_gtid_delete_list= NULL;
-  }
-}
-
-
-/*
-  Free the list recorded by pending_gtid_deletes_save(). Done when the deletes
-  in the list have been permanently committed.
-*/
-void
-rpl_group_info::pending_gtid_deletes_clear()
-{
-  pending_gtid_deletes_free(pending_gtid_delete_list);
-  pending_gtid_delete_list= NULL;
-}
-
-
-void
-rpl_group_info::pending_gtid_deletes_free(rpl_slave_state::list_element *list)
-{
-  rpl_slave_state::list_element *next;
-
-  while (list)
-  {
-    next= list->next;
-    my_free(list);
-    list= next;
-  }
-}
-
-
 rpl_sql_thread_info::rpl_sql_thread_info(Rpl_filter *filter)
   : rpl_filter(filter)
 {
diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h
index d9f0e0e5d3b..b8b153c34be 100644
--- a/sql/rpl_rli.h
+++ b/sql/rpl_rli.h
@@ -757,11 +757,6 @@ struct rpl_group_info
   /* Needs room for "Gtid D-S-N\x00". */
   char gtid_info_buf[5+10+1+10+1+20+1];
 
-  /* List of not yet committed deletions in mysql.gtid_slave_pos. */
-  rpl_slave_state::list_element *pending_gtid_delete_list;
-  /* Domain associated with pending_gtid_delete_list. */
-  uint32 pending_gtid_delete_list_domain;
-
   /*
     The timestamp, from the master, of the commit event.
     Used to do delayed update of rli->last_master_timestamp, for getting
@@ -903,12 +898,6 @@ struct rpl_group_info
   char *gtid_info();
   void unmark_start_commit();
 
-  static void pending_gtid_deletes_free(rpl_slave_state::list_element *list);
-  void pending_gtid_deletes_save(uint32 domain_id,
-                                 rpl_slave_state::list_element *list);
-  void pending_gtid_deletes_put_back();
-  void pending_gtid_deletes_clear();
-
   longlong get_row_stmt_start_timestamp()
   {
     return row_stmt_start_timestamp;
diff --git a/sql/slave.cc b/sql/slave.cc
index 16fa890d86c..03b730b2b00 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -465,6 +465,8 @@ static struct slave_background_gtid_pos_create_t {
   void *hton;
 } *slave_background_gtid_pos_create_list;
 
+static volatile bool slave_background_gtid_pending_delete_flag;
+
 
 pthread_handler_t
 handle_slave_background(void *arg __attribute__((unused)))
@@ -499,6 +501,7 @@ handle_slave_background(void *arg __attribute__((unused)))
   {
     slave_background_kill_t *kill_list;
     slave_background_gtid_pos_create_t *create_list;
+    bool pending_deletes;
 
     thd->ENTER_COND(&COND_slave_background, &LOCK_slave_background,
                     &stage_slave_background_wait_request,
@@ -508,13 +511,15 @@ handle_slave_background(void *arg __attribute__((unused)))
       stop= abort_loop || thd->killed || slave_background_thread_stop;
       kill_list= slave_background_kill_list;
       create_list= slave_background_gtid_pos_create_list;
-      if (stop || kill_list || create_list)
+      pending_deletes= slave_background_gtid_pending_delete_flag;
+      if (stop || kill_list || create_list || pending_deletes)
         break;
       mysql_cond_wait(&COND_slave_background, &LOCK_slave_background);
     }
 
     slave_background_kill_list= NULL;
     slave_background_gtid_pos_create_list= NULL;
+    slave_background_gtid_pending_delete_flag= false;
     thd->EXIT_COND(&old_stage);
 
     while (kill_list)
@@ -541,6 +546,17 @@ handle_slave_background(void *arg __attribute__((unused)))
       create_list= next;
     }
 
+    if (pending_deletes)
+    {
+      rpl_slave_state::list_element *list;
+
+      slave_background_gtid_pending_delete_flag= false;
+      list= rpl_global_gtid_slave_state->gtid_grab_pending_delete_list();
+      rpl_global_gtid_slave_state->gtid_delete_pending(thd, &list);
+      if (list)
+        rpl_global_gtid_slave_state->put_back_list(list);
+    }
+
     mysql_mutex_lock(&LOCK_slave_background);
   } while (!stop);
 
@@ -615,6 +631,23 @@ slave_background_gtid_pos_create_request(
 
 
 /*
+  Request the slave background thread to delete no longer used rows from the
+  mysql.gtid_slave_pos* tables.
+
+  This is called from time-critical rpl_slave_state::update(), so we avoid
+  taking any locks here. This means we may race with the background thread
+  to occasionally lose a signal. This is not a problem; any pending rows to
+  be deleted will just be deleted a bit later as part of the next batch.
+*/
+void
+slave_background_gtid_pending_delete_request(void)
+{
+  slave_background_gtid_pending_delete_flag= true;
+  mysql_cond_signal(&COND_slave_background);
+}
+
+
+/*
   Start the slave background thread.
 
   This thread is currently used for two purposes:
@@ -7910,8 +7943,8 @@ bool rpl_master_has_bug(const Relay_log_info *rli, uint bug_id, bool report,
 {
   struct st_version_range_for_one_bug {
     uint        bug_id;
-    const uchar introduced_in[3]; // first version with bug
-    const uchar fixed_in[3];      // first version with fix
+    Version introduced_in; // first version with bug
+    Version fixed_in;      // first version with fix
   };
   static struct st_version_range_for_one_bug versions_for_all_bugs[]=
   {
@@ -7921,19 +7954,17 @@ bool rpl_master_has_bug(const Relay_log_info *rli, uint bug_id, bool report,
     {33029, { 5, 1,  0 }, { 5, 1, 12 } },
     {37426, { 5, 1,  0 }, { 5, 1, 26 } },
   };
-  const uchar *master_ver=
-    rli->relay_log.description_event_for_exec->server_version_split.ver;
-
-  DBUG_ASSERT(sizeof(rli->relay_log.description_event_for_exec->server_version_split.ver) == 3);
+  const Version &master_ver=
+    rli->relay_log.description_event_for_exec->server_version_split;
 
   for (uint i= 0;
        i < sizeof(versions_for_all_bugs)/sizeof(*versions_for_all_bugs);i++)
   {
-    const uchar *introduced_in= versions_for_all_bugs[i].introduced_in,
-      *fixed_in= versions_for_all_bugs[i].fixed_in;
+    const Version &introduced_in= versions_for_all_bugs[i].introduced_in;
+    const Version &fixed_in= versions_for_all_bugs[i].fixed_in;
     if ((versions_for_all_bugs[i].bug_id == bug_id) &&
-        (memcmp(introduced_in, master_ver, 3) <= 0) &&
-        (memcmp(fixed_in,      master_ver, 3) >  0) &&
+        introduced_in <= master_ver &&
+        fixed_in > master_ver &&
         (pred == NULL || (*pred)(param)))
     {
       if (!report)
diff --git a/sql/slave.h b/sql/slave.h
index 649d55b45b9..12d569b0333 100644
--- a/sql/slave.h
+++ b/sql/slave.h
@@ -276,6 +276,7 @@ bool net_request_file(NET* net, const char* fname);
 void slave_background_kill_request(THD *to_kill);
 void slave_background_gtid_pos_create_request
         (rpl_slave_state::gtid_pos_table *table_entry);
+void slave_background_gtid_pending_delete_request(void);
 
 extern bool volatile abort_loop;
 extern Master_info *active_mi; /* active_mi for multi-master */
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 052c5ada3a2..954f588b061 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -187,6 +187,8 @@ public:
 
   bool eq(const char *user2, const char *host2) { return !cmp(user2, host2); }
 
+  const char *get_username(){ return user.str; }
+
   bool wild_eq(const char *user2, const char *host2, const char *ip2)
   {
     if (strcmp(user.str, user2))
@@ -228,6 +230,8 @@ public:
   acl_host_and_ip host;
   const char *user,*db;
   ulong initial_access; /* access bits present in the table */
+
+  const char *get_username() { return user; }
 };
 
 #ifndef DBUG_OFF
@@ -592,7 +596,9 @@ bool ROLE_GRANT_PAIR::init(MEM_ROOT *mem, const char *username,
 /* Flag to mark that on_node was already called for this role */
 #define ROLE_OPENED             (1L << 3)
 
-static DYNAMIC_ARRAY acl_hosts, acl_users, acl_dbs, acl_proxy_users;
+static DYNAMIC_ARRAY acl_hosts, acl_users, acl_proxy_users;
+static Dynamic_array<ACL_DB> acl_dbs(0U,50U);
+typedef Dynamic_array<ACL_DB>::CMP_FUNC acl_dbs_cmp;
 static HASH acl_roles;
 /*
   An hash containing mappings user <--> role
@@ -610,7 +616,11 @@ static DYNAMIC_ARRAY acl_wild_hosts;
 static Hash_filo<acl_entry> *acl_cache;
 static uint grant_version=0; /* Version of priv tables. incremented by acl_load */
 static ulong get_access(TABLE *form,uint fieldnr, uint *next_field=0);
-static int acl_compare(ACL_ACCESS *a,ACL_ACCESS *b);
+static int acl_compare(const ACL_ACCESS *a, const ACL_ACCESS *b);
+static int acl_user_compare(const ACL_USER *a, const ACL_USER *b);
+static void rebuild_acl_users();
+static int acl_db_compare(const ACL_DB *a, const ACL_DB *b);
+static void rebuild_acl_dbs();
 static ulong get_sort(uint count,...);
 static void init_check_host(void);
 static void rebuild_check_host(void);
@@ -1973,8 +1983,7 @@ static bool acl_load(THD *thd, const Grant_tables& tables)
     DBUG_PRINT("info", ("Found user %s", user.user.str));
     push_new_user(user);
   }
-  my_qsort((uchar*) dynamic_element(&acl_users,0,ACL_USER*),acl_users.elements,
-	   sizeof(ACL_USER),(qsort_cmp) acl_compare);
+  rebuild_acl_users();
   end_read_record(&read_record_info);
   freeze_size(&acl_users);
 
@@ -2036,12 +2045,11 @@ static bool acl_load(THD *thd, const Grant_tables& tables)
 	db.access|=REFERENCES_ACL | INDEX_ACL | ALTER_ACL;
     }
 #endif
-    (void) push_dynamic(&acl_dbs,(uchar*) &db);
+    acl_dbs.push(db);
   }
-  my_qsort((uchar*) dynamic_element(&acl_dbs,0,ACL_DB*),acl_dbs.elements,
-	   sizeof(ACL_DB),(qsort_cmp) acl_compare);
   end_read_record(&read_record_info);
-  freeze_size(&acl_dbs);
+  rebuild_acl_dbs();
+  acl_dbs.freeze();
 
   const Proxies_priv_table& proxies_priv_table= tables.proxies_priv_table();
   if (proxies_priv_table.table_exists())
@@ -2120,7 +2128,7 @@ void acl_free(bool end)
   free_root(&acl_memroot,MYF(0));
   delete_dynamic(&acl_hosts);
   delete_dynamic_with_callback(&acl_users, (FREE_FUNC) free_acl_user);
-  delete_dynamic(&acl_dbs);
+  acl_dbs.free_memory();
   delete_dynamic(&acl_wild_hosts);
   delete_dynamic(&acl_proxy_users);
   my_hash_free(&acl_check_hosts);
@@ -2158,7 +2166,8 @@ void acl_free(bool end)
 
 bool acl_reload(THD *thd)
 {
-  DYNAMIC_ARRAY old_acl_hosts, old_acl_users, old_acl_dbs, old_acl_proxy_users;
+  DYNAMIC_ARRAY old_acl_hosts, old_acl_users, old_acl_proxy_users;
+  Dynamic_array<ACL_DB> old_acl_dbs(0U,0U);
   HASH old_acl_roles, old_acl_roles_mappings;
   MEM_ROOT old_mem;
   int result;
@@ -2194,7 +2203,7 @@ bool acl_reload(THD *thd)
   old_acl_dbs= acl_dbs;
   my_init_dynamic_array(&acl_hosts, sizeof(ACL_HOST), 20, 50, MYF(0));
   my_init_dynamic_array(&acl_users, sizeof(ACL_USER), 50, 100, MYF(0));
-  my_init_dynamic_array(&acl_dbs, sizeof(ACL_DB), 50, 100, MYF(0));
+  acl_dbs.init(50, 100);
   my_init_dynamic_array(&acl_proxy_users, sizeof(ACL_PROXY_USER), 50, 100, MYF(0));
   my_hash_init2(&acl_roles,50, &my_charset_utf8_bin,
                 0, 0, 0, (my_hash_get_key) acl_role_get_key, 0,
@@ -2215,6 +2224,7 @@ bool acl_reload(THD *thd)
     acl_roles_mappings= old_acl_roles_mappings;
     acl_proxy_users= old_acl_proxy_users;
     acl_dbs= old_acl_dbs;
+    old_acl_dbs.init(0,0);
     acl_memroot= old_mem;
     init_check_host();
   }
@@ -2225,7 +2235,6 @@ bool acl_reload(THD *thd)
     delete_dynamic(&old_acl_hosts);
     delete_dynamic_with_callback(&old_acl_users, (FREE_FUNC) free_acl_user);
     delete_dynamic(&old_acl_proxy_users);
-    delete_dynamic(&old_acl_dbs);
     my_hash_free(&old_acl_roles_mappings);
   }
   mysql_mutex_unlock(&acl_cache->lock);
@@ -2317,7 +2326,7 @@ static ulong get_sort(uint count,...)
 }
 
 
-static int acl_compare(ACL_ACCESS *a,ACL_ACCESS *b)
+static int acl_compare(const ACL_ACCESS *a, const ACL_ACCESS *b)
 {
   if (a->sort > b->sort)
     return -1;
@@ -2326,6 +2335,154 @@ static int acl_compare(ACL_ACCESS *a,ACL_ACCESS *b)
   return 0;
 }
 
+static int acl_user_compare(const ACL_USER *a, const ACL_USER *b)
+{
+  int res= strcmp(a->user.str, b->user.str);
+  if (res)
+    return res;
+
+  res= acl_compare(a, b);
+  if (res)
+    return res;
+
+  /*
+    For more deterministic results, resolve ambiguity between
+    "localhost" and "127.0.0.1"/"::1" by sorting "localhost" before
+    loopback addresses.
+    Test suite (on Windows) expects "root@localhost", even if
+    root@::1 would also match.
+  */
+  return -strcmp(a->host.hostname, b->host.hostname);
+}
+
+static int acl_db_compare(const ACL_DB *a, const ACL_DB *b)
+{
+  int res= strcmp(a->user, b->user);
+  if (res)
+    return res;
+
+  return acl_compare(a, b);
+}
+
+static void rebuild_acl_users()
+{
+   my_qsort((uchar*)dynamic_element(&acl_users, 0, ACL_USER*), acl_users.elements,
+     sizeof(ACL_USER), (qsort_cmp)acl_user_compare);
+}
+
+static void rebuild_acl_dbs()
+{
+  acl_dbs.sort(acl_db_compare);
+}
+
+
+/*
+  Return index of the first entry with given user in the array,
+  or UINT_MAX if not found.
+
+  Assumes the array is sorted by get_username
+*/
+template<typename T> uint find_first_user(T* arr, uint len, const char *user)
+{
+  uint low= 0;
+  uint high= len;
+  uint mid;
+
+  bool found= false;
+  if(!len)
+    return  UINT_MAX;
+
+#ifndef DBUG_OFF
+  for (uint i = 0; i < len - 1; i++)
+    DBUG_ASSERT(strcmp(arr[i].get_username(), arr[i + 1].get_username()) <= 0);
+#endif
+  while (low < high)
+  {
+    mid= low + (high - low) / 2;
+    int cmp= strcmp(arr[mid].get_username(),user);
+    if (cmp == 0)
+      found= true;
+
+    if (cmp >= 0 )
+      high= mid;
+    else
+      low= mid + 1;
+  }
+  return  (!found || low == len || strcmp(arr[low].get_username(), user)!=0 )?UINT_MAX:low;
+}
+
+static uint acl_find_user_by_name(const char *user)
+{
+  return find_first_user<ACL_USER>((ACL_USER *)acl_users.buffer,acl_users.elements,user);
+}
+
+static uint acl_find_db_by_username(const char *user)
+{
+  return find_first_user<ACL_DB>(acl_dbs.front(), acl_dbs.elements(), user);
+}
+
+static bool match_db(ACL_DB *acl_db, const char *db, my_bool db_is_pattern)
+{
+  return !acl_db->db || (db && !wild_compare(db, acl_db->db, db_is_pattern));
+}
+
+
+/*
+  Lookup in the acl_users or acl_dbs for the best matching entry corresponding to
+  given user, host and ip parameters (also db, in case of ACL_DB)
+
+  Historical note:
+
+  In the past, both arrays were sorted just by ACL_ENTRY::sort field and were
+  searched linearly, until the first match of (username,host) pair was found.
+
+  This function uses optimizations (binary search by username), yet preserves the
+  historical behavior, i.e the returns a match with highest ACL_ENTRY::sort.
+*/
+template <typename T> T* find_by_username_or_anon(T* arr, size_t len, const char *user,
+  const char *host, const char *ip,
+  const char *db, my_bool db_is_pattern, bool (*match_db_func)(T*,const char *,my_bool))
+{
+  size_t i;
+  T *ret = NULL;
+
+  // Check  entries matching user name.
+  size_t start = find_first_user(arr, len, user);
+  for (i= start; i < len; i++)
+  {
+    T *entry= &arr[i];
+    if (i > start && strcmp(user, entry->get_username()))
+      break;
+
+    if (compare_hostname(&entry->host, host, ip) && (!match_db_func || match_db_func(entry, db, db_is_pattern)))
+    {
+      ret= entry;
+      break;
+    }
+  }
+
+  // Look also for anonymous user (username is empty string)
+  // Due to sort by name, entries for anonymous user start at the start of array.
+  for (i= 0; i < len; i++)
+  {
+    T *entry = &arr[i];
+    if (*entry->get_username() || (ret && acl_compare(entry, ret) >= 0))
+      break;
+    if (compare_hostname(&entry->host, host, ip) && (!match_db_func || match_db_func(entry, db, db_is_pattern)))
+    {
+      ret= entry;
+      break;
+    }
+  }
+  return ret;
+}
+
+static ACL_DB *acl_db_find(const char *db, const char *user, const char *host, const char *ip, my_bool db_is_pattern)
+{
+  return find_by_username_or_anon(acl_dbs.front(), acl_dbs.elements(),
+                                  user, host, ip, db, db_is_pattern, match_db);
+}
+
 
 /*
   Gets user credentials without authentication and resource limit checks.
@@ -2347,7 +2504,6 @@ bool acl_getroot(Security_context *sctx, const char *user, const char *host,
                  const char *ip, const char *db)
 {
   int res= 1;
-  uint i;
   ACL_USER *acl_user= 0;
   DBUG_ENTER("acl_getroot");
 
@@ -2380,21 +2536,9 @@ bool acl_getroot(Security_context *sctx, const char *user, const char *host,
     if (acl_user)
     {
       res= 0;
-      for (i=0 ; i < acl_dbs.elements ; i++)
-      {
-        ACL_DB *acl_db= dynamic_element(&acl_dbs, i, ACL_DB*);
-        if (!*acl_db->user || !strcmp(user, acl_db->user))
-        {
-          if (compare_hostname(&acl_db->host, host, ip))
-          {
-            if (!acl_db->db || (db && !wild_compare(db, acl_db->db, 0)))
-            {
-              sctx->db_access= acl_db->access;
-              break;
-            }
-          }
-        }
-      }
+      if (ACL_DB *acl_db= acl_db_find(db, user, host, ip, FALSE))
+        sctx->db_access= acl_db->access;
+
       sctx->master_access= acl_user->access;
 
       strmake_buf(sctx->priv_user, user);
@@ -2409,21 +2553,9 @@ bool acl_getroot(Security_context *sctx, const char *user, const char *host,
     if (acl_role)
     {
       res= 0;
-      for (i=0 ; i < acl_dbs.elements ; i++)
-      {
-        ACL_DB *acl_db= dynamic_element(&acl_dbs, i, ACL_DB*);
-        if (!*acl_db->user || !strcmp(user, acl_db->user))
-        {
-          if (compare_hostname(&acl_db->host, "", ""))
-          {
-            if (!acl_db->db || (db && !wild_compare(db, acl_db->db, 0)))
-            {
-              sctx->db_access= acl_db->access;
-              break;
-            }
-          }
-        }
-      }
+      if (ACL_DB *acl_db= acl_db_find(db, user, "", "", FALSE))
+        sctx->db_access = acl_db->access;
+
       sctx->master_access= acl_role->access;
 
       strmake_buf(sctx->priv_role, user);
@@ -2622,10 +2754,10 @@ static bool acl_update_db(const char *user, const char *host, const char *db,
 
   bool updated= false;
 
-  for (uint i=0 ; i < acl_dbs.elements ; i++)
+  for (uint i= acl_find_db_by_username(user); i < acl_dbs.elements(); i++)
   {
-    ACL_DB *acl_db=dynamic_element(&acl_dbs,i,ACL_DB*);
-    if (!strcmp(user, acl_db->user))
+    ACL_DB *acl_db= &acl_dbs.at(i);
+    if (!strcmp(user,acl_db->user))
     {
       if ((!acl_db->host.hostname && !host[0]) ||
           (acl_db->host.hostname && !strcmp(host, acl_db->host.hostname)))
@@ -2640,11 +2772,13 @@ static bool acl_update_db(const char *user, const char *host, const char *db,
             acl_db->initial_access= acl_db->access;
           }
           else
-            delete_dynamic_element(&acl_dbs,i);
+            acl_dbs.del(i);
           updated= true;
         }
       }
     }
+    else
+     break;
   }
 
   return updated;
@@ -2675,9 +2809,8 @@ static void acl_insert_db(const char *user, const char *host, const char *db,
   acl_db.db=strdup_root(&acl_memroot,db);
   acl_db.initial_access= acl_db.access= privileges;
   acl_db.sort=get_sort(3,acl_db.host.hostname,acl_db.db,acl_db.user);
-  (void) push_dynamic(&acl_dbs,(uchar*) &acl_db);
-  my_qsort((uchar*) dynamic_element(&acl_dbs,0,ACL_DB*),acl_dbs.elements,
-	   sizeof(ACL_DB),(qsort_cmp) acl_compare);
+  acl_dbs.push(acl_db);
+  rebuild_acl_dbs();
 }
 
 
@@ -2723,26 +2856,16 @@ ulong acl_get(const char *host, const char *ip,
   /*
     Check if there are some access rights for database and user
   */
-  for (i=0 ; i < acl_dbs.elements ; i++)
+  if (ACL_DB *acl_db= acl_db_find(db,user, host, ip, db_is_pattern))
   {
-    ACL_DB *acl_db=dynamic_element(&acl_dbs,i,ACL_DB*);
-    if (!*acl_db->user || !strcmp(user, acl_db->user))
-    {
-      if (compare_hostname(&acl_db->host,host,ip))
-      {
-        if (!acl_db->db || !wild_compare(db,acl_db->db,db_is_pattern))
-        {
-          db_access=acl_db->access;
-          if (acl_db->host.hostname)
-            goto exit;                          // Fully specified. Take it
-          /* the host table is not used for roles */
-          if ((!host || !host[0]) && !acl_db->host.hostname && find_acl_role(user))
-            goto exit;
-          break; /* purecov: tested */
-	}
-      }
-    }
+    db_access= acl_db->access;
+    if (acl_db->host.hostname)
+      goto exit; // Fully specified. Take it
+    /* the host table is not used for roles */
+    if ((!host || !host[0]) && !acl_db->host.hostname && find_acl_role(user))
+      goto exit;
   }
+
   if (!db_access)
     goto exit;					// Can't be better
 
@@ -3397,20 +3520,9 @@ bool is_acl_user(const char *host, const char *user)
 */
 static ACL_USER *find_user_or_anon(const char *host, const char *user, const char *ip)
 {
-  ACL_USER *result= NULL;
-  mysql_mutex_assert_owner(&acl_cache->lock);
-  for (uint i=0; i < acl_users.elements; i++)
-  {
-    ACL_USER *acl_user_tmp= dynamic_element(&acl_users, i, ACL_USER*);
-    if ((!acl_user_tmp->user.length ||
-         !strcmp(user, acl_user_tmp->user.str)) &&
-         compare_hostname(&acl_user_tmp->host, host, ip))
-    {
-      result= acl_user_tmp;
-      break;
-    }
-  }
-  return result;
+  return find_by_username_or_anon<ACL_USER>
+    (reinterpret_cast<ACL_USER*>(acl_users.buffer), acl_users.elements,
+     user, host, ip, NULL, FALSE, NULL);
 }
 
 
@@ -3420,11 +3532,15 @@ static ACL_USER *find_user_or_anon(const char *host, const char *user, const cha
 static ACL_USER *find_user_exact(const char *host, const char *user)
 {
   mysql_mutex_assert_owner(&acl_cache->lock);
+  uint start= acl_find_user_by_name(user);
 
-  for (uint i=0 ; i < acl_users.elements ; i++)
+  for (uint i= start; i < acl_users.elements; i++)
   {
-    ACL_USER *acl_user=dynamic_element(&acl_users, i, ACL_USER*);
-    if (acl_user->eq(user, host))
+    ACL_USER *acl_user= dynamic_element(&acl_users, i, ACL_USER*);
+    if (i > start && strcmp(acl_user->user.str, user))
+      return 0;
+
+    if (!my_strcasecmp(system_charset_info, acl_user->host.hostname, host))
       return acl_user;
   }
   return 0;
@@ -3437,10 +3553,14 @@ static ACL_USER * find_user_wild(const char *host, const char *user, const char
 {
   mysql_mutex_assert_owner(&acl_cache->lock);
 
-  for (uint i=0 ; i < acl_users.elements ; i++)
+  uint start = acl_find_user_by_name(user);
+
+  for (uint i= start; i < acl_users.elements; i++)
   {
     ACL_USER *acl_user=dynamic_element(&acl_users,i,ACL_USER*);
-    if (acl_user->wild_eq(user, host, ip))
+    if (i > start && strcmp(acl_user->user.str, user))
+      break;
+    if (compare_hostname(&acl_user->host, host, ip ? ip : host))
       return acl_user;
   }
   return 0;
@@ -3970,8 +4090,7 @@ end:
       else
       {
         push_new_user(new_acl_user);
-        my_qsort(dynamic_element(&acl_users, 0, ACL_USER*), acl_users.elements,
-                 sizeof(ACL_USER),(qsort_cmp) acl_compare);
+        rebuild_acl_users();
 
         /* Rebuild 'acl_check_hosts' since 'acl_users' has been modified */
         rebuild_check_host();
@@ -5554,15 +5673,15 @@ static bool merge_role_global_privileges(ACL_ROLE *grantee)
   return old != grantee->access;
 }
 
-static int db_name_sort(ACL_DB * const *db1, ACL_DB * const *db2)
+static int db_name_sort(const int *db1, const int *db2)
 {
-  return strcmp((*db1)->db, (*db2)->db);
+  return strcmp(acl_dbs.at(*db1).db, acl_dbs.at(*db2).db);
 }
 
 /**
   update ACL_DB for given database and a given role with merged privileges
 
-  @param merged ACL_DB of the role in question (or NULL if it wasn't found)
+  @param merged ACL_DB of the role in question (or -1 if it wasn't found)
   @param first  first ACL_DB in an array for the database in question
   @param access new privileges for the given role on the gived database
   @param role   the name of the given role
@@ -5572,15 +5691,15 @@ static int db_name_sort(ACL_DB * const *db1, ACL_DB * const *db2)
           2 - ACL_DB was added
           4 - ACL_DB was deleted
 */
-static int update_role_db(ACL_DB *merged, ACL_DB **first, ulong access,
+static int update_role_db(int merged, int first, ulong access,
                           const char *role)
 {
-  if (!first)
+  if (first < 0)
     return 0;
 
   DBUG_EXECUTE_IF("role_merge_stats", role_db_merges++;);
 
-  if (merged == NULL)
+  if (merged < 0)
   {
     /*
       there's no ACL_DB for this role (all db grants come from granted roles)
@@ -5595,11 +5714,11 @@ static int update_role_db(ACL_DB *merged, ACL_DB **first, ulong access,
     acl_db.user= role;
     acl_db.host.hostname= const_cast<char*>("");
     acl_db.host.ip= acl_db.host.ip_mask= 0;
-    acl_db.db= first[0]->db;
+    acl_db.db= acl_dbs.at(first).db;
     acl_db.access= access;
     acl_db.initial_access= 0;
     acl_db.sort=get_sort(3, "", acl_db.db, role);
-    push_dynamic(&acl_dbs,(uchar*) &acl_db);
+    acl_dbs.push(acl_db);
     return 2;
   }
   else if (access == 0)
@@ -5615,13 +5734,13 @@ static int update_role_db(ACL_DB *merged, ACL_DB **first, ulong access,
       2. it's O(N) operation, and we may need many of them
       so we only mark elements deleted and will delete later.
     */
-    merged->sort= 0; // lower than any valid ACL_DB sort value, will be sorted last
+    acl_dbs.at(merged).sort= 0; // lower than any valid ACL_DB sort value, will be sorted last
     return 4;
   }
-  else if (merged->access != access)
+  else if (acl_dbs.at(merged).access != access)
   {
     /* this is easy */
-    merged->access= access;
+    acl_dbs.at(merged).access= access;
     return 1;
   }
   return 0;
@@ -5636,7 +5755,7 @@ static int update_role_db(ACL_DB *merged, ACL_DB **first, ulong access,
 static bool merge_role_db_privileges(ACL_ROLE *grantee, const char *dbname,
                                      role_hash_t *rhash)
 {
-  Dynamic_array<ACL_DB *> dbs; 
+  Dynamic_array<int> dbs;
 
   /*
     Supposedly acl_dbs can be huge, but only a handful of db grants
@@ -5644,9 +5763,9 @@ static bool merge_role_db_privileges(ACL_ROLE *grantee, const char *dbname,
 
     Collect these applicable db grants.
   */
-  for (uint i=0 ; i < acl_dbs.elements ; i++)
+  for (uint i=0 ; i < acl_dbs.elements() ; i++)
   {
-    ACL_DB *db= dynamic_element(&acl_dbs,i,ACL_DB*);
+    ACL_DB *db= &acl_dbs.at(i);
     if (db->host.hostname[0])
       continue;
     if (dbname && strcmp(db->db, dbname))
@@ -5654,7 +5773,7 @@ static bool merge_role_db_privileges(ACL_ROLE *grantee, const char *dbname,
     ACL_ROLE *r= rhash->find(db->user, strlen(db->user));
     if (!r)
       continue;
-    dbs.append(db);
+    dbs.append(i);
   }
   dbs.sort(db_name_sort);
 
@@ -5663,42 +5782,47 @@ static bool merge_role_db_privileges(ACL_ROLE *grantee, const char *dbname,
     (that should be merged) are sorted together. The grantee's ACL_DB element
     is not necessarily the first and may be not present at all.
   */
-  ACL_DB **first= NULL, *merged= NULL;
+  int first= -1, merged= -1;
   ulong access= 0, update_flags= 0;
-  for (ACL_DB **cur= dbs.front(); cur <= dbs.back(); cur++)
+  for (int *p= dbs.front(); p <= dbs.back(); p++)
   {
-    if (!first || (!dbname && strcmp(cur[0]->db, cur[-1]->db)))
+    if (first<0 || (!dbname && strcmp(acl_dbs.at(*p).db, acl_dbs.at(*p-1).db)))
     { // new db name series
       update_flags|= update_role_db(merged, first, access, grantee->user.str);
-      merged= NULL;
+      merged= -1;
       access= 0;
-      first= cur;
+      first= *p;
     }
-    if (strcmp(cur[0]->user, grantee->user.str) == 0)
-      access|= (merged= cur[0])->initial_access;
+    if (strcmp(acl_dbs.at(*p).user, grantee->user.str) == 0)
+      access|= acl_dbs.at(merged= *p).initial_access;
     else
-      access|= cur[0]->access;
+      access|= acl_dbs.at(*p).access;
   }
   update_flags|= update_role_db(merged, first, access, grantee->user.str);
 
-  /*
-    to make this code a bit simpler, we sort on deletes, to move
-    deleted elements to the end of the array. strictly speaking it's
-    unnecessary, it'd be faster to remove them in one O(N) array scan.
-    
-    on the other hand, qsort on almost sorted array is pretty fast anyway...
-  */
-  if (update_flags & (2|4))
-  { // inserted or deleted, need to sort
-    my_qsort((uchar*) dynamic_element(&acl_dbs,0,ACL_DB*),acl_dbs.elements,
-             sizeof(ACL_DB),(qsort_cmp) acl_compare);
-  }
   if (update_flags & 4)
-  { // deleted, trim the end
-    while (acl_dbs.elements &&
-           dynamic_element(&acl_dbs, acl_dbs.elements-1, ACL_DB*)->sort == 0)
-      acl_dbs.elements--;
+  {
+    // Remove elements marked for deletion.
+    uint count= 0;
+    for(uint i= 0; i < acl_dbs.elements(); i++)
+    {
+      ACL_DB *acl_db= &acl_dbs.at(i);
+      if (acl_db->sort)
+      {
+        if (i > count)
+          acl_dbs.set(count, *acl_db);
+        count++;
+      }
+    }
+    acl_dbs.elements(count);
   }
+
+
+  if (update_flags & 2)
+  { // inserted, need to sort
+    rebuild_acl_dbs();
+  }
+
   return update_flags;
 }
 
@@ -8645,16 +8769,14 @@ static bool show_database_privileges(THD *thd, const char *username,
                                      const char *hostname,
                                      char *buff, size_t buffsize)
 {
-  ACL_DB *acl_db;
   ulong want_access;
-  uint counter;
   Protocol *protocol= thd->protocol;
 
-  for (counter=0 ; counter < acl_dbs.elements ; counter++)
+  for (uint i=0 ; i < acl_dbs.elements() ; i++)
   {
     const char *user, *host;
 
-    acl_db=dynamic_element(&acl_dbs,counter,ACL_DB*);
+    ACL_DB *acl_db= &acl_dbs.at(i);
     user= acl_db->user;
     host=acl_db->host.hostname;
 
@@ -9409,7 +9531,7 @@ static int handle_grant_struct(enum enum_acl_lists struct_no, bool drop,
     elements= acl_users.elements;
     break;
   case DB_ACL:
-    elements= acl_dbs.elements;
+    elements= acl_dbs.elements();
     break;
   case COLUMN_PRIVILEGES_HASH:
     grant_name_hash= &column_priv_hash;
@@ -9461,7 +9583,7 @@ static int handle_grant_struct(enum enum_acl_lists struct_no, bool drop,
     break;
 
     case DB_ACL:
-      acl_db= dynamic_element(&acl_dbs, idx, ACL_DB*);
+      acl_db= &acl_dbs.at(idx);
       user= acl_db->user;
       host= acl_db->host.hostname;
       break;
@@ -9545,7 +9667,7 @@ static int handle_grant_struct(enum enum_acl_lists struct_no, bool drop,
         break;
 
       case DB_ACL:
-        delete_dynamic_element(&acl_dbs, idx);
+        acl_dbs.del(idx);
         break;
 
       case COLUMN_PRIVILEGES_HASH:
@@ -10245,6 +10367,8 @@ bool mysql_rename_user(THD *thd, List <LEX_USER> &list)
     some_users_renamed= TRUE;
   }
 
+  rebuild_acl_users();
+
   /* Rebuild 'acl_check_hosts' since 'acl_users' has been modified */
   rebuild_check_host();
 
@@ -10438,11 +10562,11 @@ bool mysql_revoke_all(THD *thd,  List <LEX_USER> &list)
      */
     do
     {
-      for (counter= 0, revoked= 0 ; counter < acl_dbs.elements ; )
+      for (counter= 0, revoked= 0 ; counter < acl_dbs.elements() ; )
       {
-	const char *user,*host;
+        const char *user, *host;
 
-	acl_db=dynamic_element(&acl_dbs,counter,ACL_DB*);
+        acl_db= &acl_dbs.at(counter);
 
         user= acl_db->user;
         host= safe_str(acl_db->host.hostname);
@@ -11045,6 +11169,14 @@ static int show_column_grants(THD *thd, SHOW_VAR *var, char *buff,
   return 0;
 }
 
+static int show_database_grants(THD *thd, SHOW_VAR *var, char *buff,
+                                enum enum_var_type scope)
+{
+  var->type= SHOW_UINT;
+  var->value= buff;
+  *(uint *)buff= acl_dbs.elements();
+  return 0;
+}
 
 #else
 bool check_grant(THD *, ulong, TABLE_LIST *, bool, uint, bool)
@@ -11056,7 +11188,7 @@ bool check_grant(THD *, ulong, TABLE_LIST *, bool, uint, bool)
 SHOW_VAR acl_statistics[] = {
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
   {"column_grants",    (char*)show_column_grants,          SHOW_SIMPLE_FUNC},
-  {"database_grants",  (char*)&acl_dbs.elements,           SHOW_UINT},
+  {"database_grants",  (char*)show_database_grants,        SHOW_SIMPLE_FUNC},
   {"function_grants",  (char*)&func_priv_hash.records,     SHOW_ULONG},
   {"procedure_grants", (char*)&proc_priv_hash.records,     SHOW_ULONG},
   {"package_spec_grants", (char*)&package_spec_priv_hash.records, SHOW_ULONG},
@@ -11329,11 +11461,11 @@ int fill_schema_schema_privileges(THD *thd, TABLE_LIST *tables, COND *cond)
     DBUG_RETURN(0);
   mysql_mutex_lock(&acl_cache->lock);
 
-  for (counter=0 ; counter < acl_dbs.elements ; counter++)
+  for (counter=0 ; counter < acl_dbs.elements() ; counter++)
   {
     const char *user, *host, *is_grantable="YES";
 
-    acl_db=dynamic_element(&acl_dbs,counter,ACL_DB*);
+    acl_db=&acl_dbs.at(counter);
     user= acl_db->user;
     host= safe_str(acl_db->host.hostname);
 
diff --git a/sql/sql_array.h b/sql/sql_array.h
index 0e5246b7e2a..0f18a89360a 100644
--- a/sql/sql_array.h
+++ b/sql/sql_array.h
@@ -123,8 +123,7 @@ public:
 
   void init(uint prealloc=16, uint increment=16)
   {
-    my_init_dynamic_array(&array, sizeof(Elem), prealloc, increment,
-                          MYF(0));
+    init_dynamic_array2(&array, sizeof(Elem), 0, prealloc, increment, MYF(0));
   }
 
   /**
@@ -218,6 +217,11 @@ public:
     set_dynamic(&array, &el, idx);
   }
 
+  void freeze()
+  {
+    freeze_size(&array);
+  }
+
   bool resize(size_t new_size, Elem default_val)
   {
     size_t old_size= elements();
@@ -252,7 +256,7 @@ public:
     my_qsort(array.buffer, array.elements, sizeof(Elem), (qsort_cmp)cmp_func);
   }
 
-  typedef int (*CMP_FUNC2)(const Elem *el1, const Elem *el2, void *);
+  typedef int (*CMP_FUNC2)(void *, const Elem *el1, const Elem *el2);
   void sort(CMP_FUNC2 cmp_func, void *data)
   {
     my_qsort2(array.buffer, array.elements, sizeof(Elem), (qsort2_cmp)cmp_func, data);
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 1997add96dc..1078eacd035 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -156,9 +156,15 @@ enum enum_binlog_row_image {
 #define MODE_HIGH_NOT_PRECEDENCE        (1ULL << 29)
 #define MODE_NO_ENGINE_SUBSTITUTION     (1ULL << 30)
 #define MODE_PAD_CHAR_TO_FULL_LENGTH    (1ULL << 31)
+/* SQL mode bits defined above are common for MariaDB and MySQL */
+#define MODE_MASK_MYSQL_COMPATIBLE      0xFFFFFFFFULL
+/* The following modes are specific to MariaDB */
 #define MODE_EMPTY_STRING_IS_NULL       (1ULL << 32)
 #define MODE_SIMULTANEOUS_ASSIGNMENT    (1ULL << 33)
 #define MODE_TIME_ROUND_FRACTIONAL      (1ULL << 34)
+/* The following modes are specific to MySQL */
+#define MODE_MYSQL80_TIME_TRUNCATE_FRACTIONAL (1ULL << 32)
+
 
 /* Bits for different old style modes */
 #define OLD_MODE_NO_DUP_KEY_WARNINGS_WITH_IGNORE	(1 << 0)
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 127b4b10eb4..c8b7554605d 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -640,7 +640,7 @@ create_insert_stmt_from_insert_delayed(THD *thd, String *buf)
   if (buf->append(thd->query()) ||
       buf->replace(thd->lex->keyword_delayed_begin_offset,
                    thd->lex->keyword_delayed_end_offset -
-                   thd->lex->keyword_delayed_begin_offset, 0))
+                   thd->lex->keyword_delayed_begin_offset, NULL, 0))
     return 1;
   return 0;
 }
diff --git a/sql/sql_string.cc b/sql/sql_string.cc
index 39d9438d5bf..b84585114b7 100644
--- a/sql/sql_string.cc
+++ b/sql/sql_string.cc
@@ -31,7 +31,7 @@
 ** String functions
 *****************************************************************************/
 
-bool String::real_alloc(size_t length)
+bool Binary_string::real_alloc(size_t length)
 {
   size_t arg_length= ALIGN_SIZE(length + 1);
   DBUG_ASSERT(arg_length > length);
@@ -81,7 +81,7 @@ bool String::real_alloc(size_t length)
 
    @retval true An error occurred when attempting to allocate memory.
 */
-bool String::realloc_raw(size_t alloc_length)
+bool Binary_string::realloc_raw(size_t alloc_length)
 {
   if (Alloced_length <= alloc_length)
   {
@@ -126,19 +126,18 @@ bool String::set_int(longlong num, bool unsigned_flag, CHARSET_INFO *cs)
   if (alloc(l))
     return TRUE;
   str_length=(uint32) (cs->cset->longlong10_to_str)(cs,Ptr,l,base,num);
-  str_charset=cs;
+  set_charset(cs);
   return FALSE;
 }
 
 
 // Convert a number into its HEX representation
-bool String::set_hex(ulonglong num)
+bool Binary_string::set_hex(ulonglong num)
 {
   char *n_end;
   if (alloc(65) || !(n_end= longlong2str(num, Ptr, 16)))
     return true;
   length((uint32) (n_end - Ptr));
-  set_charset(&my_charset_latin1);
   return false;
 }
 
@@ -156,7 +155,7 @@ static inline void APPEND_HEX(char *&to, uchar value)
 }
 
 
-void String::qs_append_hex(const char *str, uint32 len)
+void Static_binary_string::qs_append_hex(const char *str, uint32 len)
 {
   const char *str_end= str + len;
   for (char *to= Ptr + str_length ; str < str_end; str++)
@@ -166,7 +165,7 @@ void String::qs_append_hex(const char *str, uint32 len)
 
 
 // Convert a string to its HEX representation
-bool String::set_hex(const char *str, uint32 len)
+bool Binary_string::set_hex(const char *str, uint32 len)
 {
   /*
     Safety: cut the source string if "len" is too large.
@@ -180,7 +179,6 @@ bool String::set_hex(const char *str, uint32 len)
     return true;
   length(0);
   qs_append_hex(str, len);
-  set_charset(&my_charset_latin1);
   return false;
 }
 
@@ -191,7 +189,7 @@ bool String::set_real(double num,uint decimals, CHARSET_INFO *cs)
   uint dummy_errors;
   size_t len;
 
-  str_charset=cs;
+  set_charset(cs);
   if (decimals >= FLOATING_POINT_DECIMALS)
   {
     len= my_gcvt(num, MY_GCVT_ARG_DOUBLE, sizeof(buff) - 1, buff, NULL);
@@ -203,7 +201,7 @@ bool String::set_real(double num,uint decimals, CHARSET_INFO *cs)
 }
 
 
-bool String::copy()
+bool Binary_string::copy()
 {
   if (!alloced)
   {
@@ -224,18 +222,17 @@ bool String::copy()
    @retval false Success.
    @retval true Memory allocation failed.
 */
-bool String::copy(const String &str)
+bool Binary_string::copy(const Binary_string &str)
 {
   if (alloc(str.str_length))
     return TRUE;
   str_length=str.str_length;
   bmove(Ptr,str.Ptr,str_length);		// May be overlapping
   Ptr[str_length]=0;
-  str_charset=str.str_charset;
   return FALSE;
 }
 
-bool String::copy(const char *str,size_t arg_length, CHARSET_INFO *cs)
+bool Binary_string::copy(const char *str, size_t arg_length)
 {
   DBUG_ASSERT(arg_length < UINT_MAX32);
   if (alloc(arg_length))
@@ -252,7 +249,6 @@ bool String::copy(const char *str,size_t arg_length, CHARSET_INFO *cs)
   else if ((str_length=uint32(arg_length)))
     memcpy(Ptr,str,arg_length);
   Ptr[arg_length]=0;
-  str_charset=cs;
   return FALSE;
 }
 
@@ -262,7 +258,7 @@ bool String::copy(const char *str,size_t arg_length, CHARSET_INFO *cs)
   from valgrind
 */
 
-bool String::copy_or_move(const char *str,size_t arg_length, CHARSET_INFO *cs)
+bool Binary_string::copy_or_move(const char *str, size_t arg_length)
 {
   DBUG_ASSERT(arg_length < UINT_MAX32);
   if (alloc(arg_length))
@@ -270,7 +266,6 @@ bool String::copy_or_move(const char *str,size_t arg_length, CHARSET_INFO *cs)
   if ((str_length=uint32(arg_length)))
     memmove(Ptr,str,arg_length);
   Ptr[arg_length]=0;
-  str_charset=cs;
   return FALSE;
 }
 
@@ -396,7 +391,7 @@ bool String::copy_aligned(const char *str, size_t arg_length, size_t offset,
   Ptr[aligned_length]=0;
   /* str_length is always >= 0 as arg_length is != 0 */
   str_length= (uint32)aligned_length;
-  str_charset= cs;
+  set_charset(cs);
   return FALSE;
 }
 
@@ -449,7 +444,7 @@ bool String::copy(const char *str, size_t arg_length,
     return TRUE;
   str_length=copy_and_convert((char*) Ptr, new_length, to_cs,
                               str, arg_length, from_cs, errors);
-  str_charset=to_cs;
+  set_charset(to_cs);
   return FALSE;
 }
 
@@ -475,19 +470,20 @@ bool String::copy(const char *str, size_t arg_length,
 
 bool String::set_ascii(const char *str, size_t arg_length)
 {
-  if (str_charset->mbminlen == 1)
+  if (mbminlen() == 1)
   {
-    set(str, arg_length, str_charset);
+    set(str, arg_length, charset());
     return 0;
   }
   uint dummy_errors;
-  return copy(str, (uint32)arg_length, &my_charset_latin1, str_charset, &dummy_errors);
+  return copy(str, (uint32) arg_length, &my_charset_latin1,
+              charset(), &dummy_errors);
 }
 
 
 /* This is used by mysql.cc */
 
-bool String::fill(uint32 max_length,char fill_char)
+bool Binary_string::fill(uint32 max_length,char fill_char)
 {
   if (str_length > max_length)
     Ptr[str_length=max_length]=0;
@@ -503,22 +499,10 @@ bool String::fill(uint32 max_length,char fill_char)
 
 void String::strip_sp()
 {
-   while (str_length && my_isspace(str_charset,Ptr[str_length-1]))
+   while (str_length && my_isspace(charset(), Ptr[str_length-1]))
     str_length--;
 }
 
-bool String::append(const String &s)
-{
-  if (s.length())
-  {
-    if (realloc_with_extra_if_needed(str_length+s.length()))
-      return TRUE;
-    memcpy(Ptr+str_length,s.ptr(),s.length());
-    str_length+=s.length();
-  }
-  return FALSE;
-}
-
 
 /*
   Append an ASCII string to the a string of the current character set
@@ -534,13 +518,13 @@ bool String::append(const char *s,size_t size)
   /*
     For an ASCII incompatible string, e.g. UCS-2, we need to convert
   */
-  if (str_charset->mbminlen > 1)
+  if (mbminlen() > 1)
   {
-    uint32 add_length=arg_length * str_charset->mbmaxlen;
+    uint32 add_length= arg_length * mbmaxlen();
     uint dummy_errors;
     if (realloc_with_extra_if_needed(str_length+ add_length))
       return TRUE;
-    str_length+= copy_and_convert(Ptr+str_length, add_length, str_charset,
+    str_length+= copy_and_convert(Ptr + str_length, add_length, charset(),
 				  s, arg_length, &my_charset_latin1,
                                   &dummy_errors);
     return FALSE;
@@ -549,24 +533,11 @@ bool String::append(const char *s,size_t size)
   /*
     For an ASCII compatinble string we can just append.
   */
-  if (realloc_with_extra_if_needed(str_length+arg_length))
-    return TRUE;
-  memcpy(Ptr+str_length,s,arg_length);
-  str_length+=arg_length;
-  return FALSE;
+  return Binary_string::append(s, arg_length);
 }
 
 
-/*
-  Append a 0-terminated ASCII string
-*/
-
-bool String::append(const char *s)
-{
-  return append(s, (uint) strlen(s));
-}
-
-bool String::append_longlong(longlong val)
+bool Binary_string::append_longlong(longlong val)
 {
   if (realloc(str_length+MAX_BIGINT_WIDTH+2))
     return TRUE;
@@ -576,7 +547,7 @@ bool String::append_longlong(longlong val)
 }
 
 
-bool String::append_ulonglong(ulonglong val)
+bool Binary_string::append_ulonglong(ulonglong val)
 {
   if (realloc(str_length+MAX_BIGINT_WIDTH+2))
     return TRUE;
@@ -594,13 +565,13 @@ bool String::append(const char *s, size_t arg_length, CHARSET_INFO *cs)
 {
   uint32 offset;
   
-  if (needs_conversion((uint32)arg_length, cs, str_charset, &offset))
+  if (needs_conversion((uint32)arg_length, cs, charset(), &offset))
   {
     size_t add_length;
     if ((cs == &my_charset_bin) && offset)
     {
-      DBUG_ASSERT(str_charset->mbminlen > offset);
-      offset= str_charset->mbminlen - offset; // How many characters to pad
+      DBUG_ASSERT(mbminlen() > offset);
+      offset= mbminlen() - offset; // How many characters to pad
       add_length= arg_length + offset;
       if (realloc(str_length + add_length))
         return TRUE;
@@ -610,24 +581,19 @@ bool String::append(const char *s, size_t arg_length, CHARSET_INFO *cs)
       return FALSE;
     }
 
-    add_length= arg_length / cs->mbminlen * str_charset->mbmaxlen;
+    add_length= arg_length / cs->mbminlen * mbmaxlen();
     uint dummy_errors;
     if (realloc_with_extra_if_needed(str_length + add_length)) 
       return TRUE;
-    str_length+= copy_and_convert(Ptr+str_length, (uint32)add_length, str_charset,
-				  s, (uint32)arg_length, cs, &dummy_errors);
+    str_length+= copy_and_convert(Ptr + str_length, (uint32)add_length, charset(),
+                                  s, (uint32)arg_length, cs, &dummy_errors);
+    return false;
   }
-  else
-  {
-    if (realloc_with_extra_if_needed(str_length + arg_length)) 
-      return TRUE;
-    memcpy(Ptr + str_length, s, arg_length);
-    str_length+= (uint32)arg_length;
-  }
-  return FALSE;
+  return Binary_string::append(s, arg_length);
 }
 
-bool String::append(IO_CACHE* file, uint32 arg_length)
+
+bool Binary_string::append(IO_CACHE* file, uint32 arg_length)
 {
   if (realloc_with_extra_if_needed(str_length+arg_length))
     return TRUE;
@@ -675,19 +641,8 @@ bool String::append_with_prefill(const char *s,uint32 arg_length,
   return FALSE;
 }
 
-uint32 String::numchars() const
-{
-  return (uint32) str_charset->cset->numchars(str_charset, Ptr, Ptr+str_length);
-}
-
-int String::charpos(longlong i,uint32 offset)
-{
-  if (i <= 0)
-    return (int)i;
-  return (int)str_charset->cset->charpos(str_charset,Ptr+offset,Ptr+str_length,(size_t)i);
-}
 
-int String::strstr(const String &s,uint32 offset)
+int Static_binary_string::strstr(const Static_binary_string &s, uint32 offset)
 {
   if (s.length()+offset <= str_length)
   {
@@ -718,7 +673,7 @@ skip:
 ** Search string from end. Offset is offset to the end of string
 */
 
-int String::strrstr(const String &s,uint32 offset)
+int Static_binary_string::strrstr(const Static_binary_string &s, uint32 offset)
 {
   if (s.length() <= offset && offset <= str_length)
   {
@@ -745,18 +700,9 @@ skip:
   return -1;
 }
 
-/*
-  Replace substring with string
-  If wrong parameter or not enough memory, do nothing
-*/
-
-bool String::replace(uint32 offset,uint32 arg_length,const String &to)
-{
-  return replace(offset,arg_length,to.ptr(),to.length());
-}
 
-bool String::replace(uint32 offset,uint32 arg_length,
-                     const char *to, uint32 to_length)
+bool Binary_string::replace(uint32 offset, uint32 arg_length,
+                            const char *to, uint32 to_length)
 {
   long diff = (long) to_length-(long) arg_length;
   if (offset+arg_length <= str_length)
@@ -787,7 +733,7 @@ bool String::replace(uint32 offset,uint32 arg_length,
 
 
 // added by Holyfoot for "geometry" needs
-int String::reserve(size_t space_needed, size_t grow_by)
+int Binary_string::reserve(size_t space_needed, size_t grow_by)
 {
   if (Alloced_length < str_length + space_needed)
   {
@@ -797,34 +743,34 @@ int String::reserve(size_t space_needed, size_t grow_by)
   return FALSE;
 }
 
-void String::qs_append(const char *str, size_t len)
+void Static_binary_string::qs_append(const char *str, size_t len)
 {
   memcpy(Ptr + str_length, str, len + 1);
   str_length += (uint32)len;
 }
 
-void String::qs_append(double d)
+void Static_binary_string::qs_append(double d)
 {
   char *buff = Ptr + str_length;
   str_length+= (uint32) my_gcvt(d, MY_GCVT_ARG_DOUBLE, FLOATING_POINT_BUFFER - 1, buff,
                        NULL);
 }
 
-void String::qs_append(double *d)
+void Static_binary_string::qs_append(double *d)
 {
   double ld;
   float8get(ld, (char*) d);
   qs_append(ld);
 }
 
-void String::qs_append(int i)
+void Static_binary_string::qs_append(int i)
 {
   char *buff= Ptr + str_length;
   char *end= int10_to_str(i, buff, -10);
   str_length+= (int) (end-buff);
 }
 
-void String::qs_append(ulonglong i)
+void Static_binary_string::qs_append(ulonglong i)
 {
   char *buff= Ptr + str_length;
   char *end= longlong10_to_str(i, buff, 10);
@@ -946,12 +892,12 @@ String *copy_if_not_alloced(String *to,String *from,uint32 from_length)
        of a constant string.
        Not safe to reuse.
   */
-  if (from->Alloced_length > 0) // "from" is  #c or #d (not a constant)
+  if (from->alloced_length() > 0) // "from" is  #c or #d (not a constant)
   {
-    if (from->Alloced_length >= from_length)
+    if (from->alloced_length() >= from_length)
       return from; // #c or #d (large enough to store from_length bytes)
 
-    if (from->alloced)
+    if (from->is_alloced())
     {
       (void) from->realloc(from_length);
       return from; // #d (reallocated to fit from_length bytes)
@@ -990,7 +936,7 @@ String *copy_if_not_alloced(String *to,String *from,uint32 from_length)
       Note, as we can't distinguish between #a and #b for sure,
       so we can't assert "not #a", but we can at least assert "not #e".
     */
-    DBUG_ASSERT(!from->alloced || from->Alloced_length > 0); // Not #e
+    DBUG_ASSERT(!from->is_alloced() || from->alloced_length() > 0); // Not #e
 
     (void) from->realloc(from_length);
     return from;
@@ -999,7 +945,7 @@ String *copy_if_not_alloced(String *to,String *from,uint32 from_length)
     return from;				// Actually an error
   if ((to->str_length=MY_MIN(from->str_length,from_length)))
     memcpy(to->Ptr,from->Ptr,to->str_length);
-  to->str_charset=from->str_charset;
+  to->set_charset(*from);
   return to; // "from" was of types #a, #b, #e, or small #c.
 }
 
@@ -1159,26 +1105,6 @@ void String::print_with_conversion(String *print, CHARSET_INFO *cs) const
 }
 
 
-/*
-  Exchange state of this object and argument.
-
-  SYNOPSIS
-    String::swap()
-
-  RETURN
-    Target string will contain state of this object and vice versa.
-*/
-
-void String::swap(String &s)
-{
-  swap_variables(char *, Ptr, s.Ptr);
-  swap_variables(uint32, str_length, s.str_length);
-  swap_variables(uint32, Alloced_length, s.Alloced_length);
-  swap_variables(bool, alloced, s.alloced);
-  swap_variables(CHARSET_INFO*, str_charset, s.str_charset);
-}
-
-
 /**
   Convert string to printable ASCII string
 
diff --git a/sql/sql_string.h b/sql/sql_string.h
index 0ae68cb3796..a4574c2ef2d 100644
--- a/sql/sql_string.h
+++ b/sql/sql_string.h
@@ -126,56 +126,62 @@ uint convert_to_printable(char *to, size_t to_len,
                           const char *from, size_t from_len,
                           CHARSET_INFO *from_cs, size_t nbytes= 0);
 
-class String
+
+class Charset
 {
-  char *Ptr;
-  uint32 str_length,Alloced_length, extra_alloc;
-  bool alloced,thread_specific;
-  CHARSET_INFO *str_charset;
+  CHARSET_INFO *m_charset;
 public:
-  String()
-  { 
-    Ptr=0; str_length=Alloced_length=extra_alloc=0;
-    alloced= thread_specific= 0; 
-    str_charset= &my_charset_bin; 
+  Charset() :m_charset(&my_charset_bin) { }
+  Charset(CHARSET_INFO *cs) :m_charset(cs) { }
+
+  CHARSET_INFO *charset() const { return m_charset; }
+  uint mbminlen() const { return m_charset->mbminlen; }
+  uint mbmaxlen() const { return m_charset->mbmaxlen; }
+
+  size_t numchars(const char *str, const char *end) const
+  {
+    return m_charset->cset->numchars(m_charset, str, end);
   }
-  String(size_t length_arg)
-  { 
-    alloced= thread_specific= 0;
-    Alloced_length= extra_alloc= 0; (void) real_alloc(length_arg); 
-    str_charset= &my_charset_bin;
+  size_t charpos(const char *str, const char *end, size_t pos) const
+  {
+    return m_charset->cset->charpos(m_charset, str, end, pos);
   }
-  String(const char *str, CHARSET_INFO *cs)
-  { 
-    Ptr=(char*) str; str_length= (uint32) strlen(str);
-    Alloced_length= extra_alloc= 0;
-    alloced= thread_specific= 0;
-    str_charset=cs;
+  void set_charset(CHARSET_INFO *charset_arg)
+  {
+    m_charset= charset_arg;
   }
-  /*
-    NOTE: If one intend to use the c_ptr() method, the following two
-    contructors need the size of memory for STR to be at least LEN+1 (to make
-    room for zero termination).
-  */
-  String(const char *str,size_t len, CHARSET_INFO *cs)
-  { 
-    Ptr=(char*) str; str_length=(uint32)len; Alloced_length= extra_alloc=0;
-    alloced= thread_specific= 0;
-    str_charset=cs;
+  void set_charset(const Charset &other)
+  {
+    m_charset= other.m_charset;
   }
-  String(char *str,size_t len, CHARSET_INFO *cs)
-  { 
-    Ptr=(char*) str; Alloced_length=str_length=(uint32)len; extra_alloc= 0;
-    alloced= thread_specific= 0;
-    str_charset=cs;
+  void swap(Charset &other)
+  {
+    swap_variables(CHARSET_INFO*, m_charset, other.m_charset);
   }
-  String(const String &str)
-  { 
-    Ptr=str.Ptr ; str_length=str.str_length ;
-    Alloced_length=str.Alloced_length; extra_alloc= 0;
-    alloced= thread_specific= 0;
-    str_charset=str.str_charset;
+};
+
+
+/*
+  A storage for String.
+  Should be eventually derived from LEX_STRING.
+*/
+class Static_binary_string
+{
+protected:
+  char *Ptr;
+  uint32 str_length;
+public:
+  Static_binary_string()
+   :Ptr(NULL),
+    str_length(0)
+  { }
+  Static_binary_string(char *str, size_t length_arg)
+   :Ptr(str),
+    str_length((uint32) length_arg)
+  {
+    DBUG_ASSERT(length_arg < UINT_MAX32);
   }
+
   static void *operator new(size_t size, MEM_ROOT *mem_root) throw ()
   { return (void*) alloc_root(mem_root, size); }
   static void *operator new[](size_t size, MEM_ROOT *mem_root) throw ()
@@ -193,50 +199,13 @@ public:
   static void operator delete[](void *, MEM_ROOT *)
   { /* never called */ }
 
-  ~String() { free(); }
-
-  /* Mark variable thread specific it it's not allocated already */
-  inline void set_thread_specific()
-  {
-    if (!alloced)
-      thread_specific= 1;
-  }
-  inline void set_charset(CHARSET_INFO *charset_arg)
-  { str_charset= charset_arg; }
-  inline CHARSET_INFO *charset() const { return str_charset; }
   inline uint32 length() const { return str_length;}
-  inline uint32 alloced_length() const { return Alloced_length;}
-  inline uint32 extra_allocation() const { return extra_alloc;}
   inline char& operator [] (size_t i) const { return Ptr[i]; }
   inline void length(size_t len) { str_length=(uint32)len ; }
-  inline void extra_allocation(size_t len) { extra_alloc= (uint32)len; }
   inline bool is_empty() const { return (str_length == 0); }
-  inline void mark_as_const() { Alloced_length= 0;}
   inline const char *ptr() const { return Ptr; }
   inline const char *end() const { return Ptr + str_length; }
-  inline char *c_ptr()
-  {
-    DBUG_ASSERT(!alloced || !Ptr || !Alloced_length || 
-                (Alloced_length >= (str_length + 1)));
 
-    if (!Ptr || Ptr[str_length])		/* Should be safe */
-      (void) realloc(str_length);
-    return Ptr;
-  }
-  inline char *c_ptr_quick()
-  {
-    if (Ptr && str_length < Alloced_length)
-      Ptr[str_length]=0;
-    return Ptr;
-  }
-  inline char *c_ptr_safe()
-  {
-    if (Ptr && str_length < Alloced_length)
-      Ptr[str_length]=0;
-    else
-      (void) realloc(str_length);
-    return Ptr;
-  }
   LEX_STRING lex_string() const
   {
     LEX_STRING str = { (char*) ptr(), length() };
@@ -248,77 +217,32 @@ public:
     return skr;
   }
 
-  void set(String &str,size_t offset,size_t arg_length)
-  {
-    DBUG_ASSERT(&str != this);
-    free();
-    Ptr=(char*) str.ptr()+offset; str_length=(uint32)arg_length;
-    if (str.Alloced_length)
-      Alloced_length=(uint32)(str.Alloced_length-offset);
-    str_charset=str.str_charset;
-  }
-
-
-  /**
-     Points the internal buffer to the supplied one. The old buffer is freed.
-     @param str Pointer to the new buffer.
-     @param arg_length Length of the new buffer in characters, excluding any 
-            null character.
-     @param cs Character set to use for interpreting string data.
-     @note The new buffer will not be null terminated.
-  */
-  inline void set(char *str,size_t arg_length, CHARSET_INFO *cs)
-  {
-    free();
-    Ptr=(char*) str; str_length=Alloced_length=(uint32)arg_length;
-    str_charset=cs;
-  }
-  inline void set(const char *str,size_t arg_length, CHARSET_INFO *cs)
+  bool has_8bit_bytes() const
   {
-    free();
-    Ptr=(char*) str; str_length=(uint32)arg_length;
-    str_charset=cs;
-  }
-  bool set_ascii(const char *str, size_t arg_length);
-  inline void set_quick(char *str,size_t arg_length, CHARSET_INFO *cs)
-  {
-    if (!alloced)
+    for (const char *c= ptr(), *c_end= end(); c < c_end; c++)
     {
-      Ptr=(char*) str; str_length=Alloced_length=(uint32)arg_length;
+      if (!my_isascii(*c))
+        return true;
     }
-    str_charset=cs;
+    return false;
   }
-  bool set_int(longlong num, bool unsigned_flag, CHARSET_INFO *cs);
-  bool set(int num, CHARSET_INFO *cs) { return set_int(num, false, cs); }
-  bool set(uint num, CHARSET_INFO *cs) { return set_int(num, true, cs); }
-  bool set(long num, CHARSET_INFO *cs) { return set_int(num, false, cs); }
-  bool set(ulong num, CHARSET_INFO *cs) { return set_int(num, true, cs); }
-  bool set(longlong num, CHARSET_INFO *cs) { return set_int(num, false, cs); }
-  bool set(ulonglong num, CHARSET_INFO *cs) { return set_int((longlong)num, true, cs); }
-  bool set_real(double num,uint decimals, CHARSET_INFO *cs);
 
-  bool set_hex(ulonglong num);
-  bool set_hex(const char *str, uint32 len);
+  bool bin_eq(const Static_binary_string *other) const
+  {
+    return length() == other->length() &&
+           !memcmp(ptr(), other->ptr(), length());
+  }
 
-  /* Take over handling of buffer from some other object */
-  void reset(char *ptr_arg, size_t length_arg, size_t alloced_length_arg,
-             CHARSET_INFO *cs)
-  { 
-    free();
-    Ptr= ptr_arg;
-    str_length= (uint32)length_arg;
-    Alloced_length= (uint32)alloced_length_arg;
-    str_charset= cs;
-    alloced= ptr_arg != 0;
+  void set(char *str, size_t len)
+  {
+    Ptr= str;
+    str_length= (uint32) len;
   }
 
-  /* Forget about the buffer, let some other object handle it */
-  char *release()
+  void swap(Static_binary_string &s)
   {
-    char *old= Ptr;
-    Ptr=0; str_length= Alloced_length= extra_alloc= 0;
-    alloced= thread_specific= 0;
-    return old;
+    swap_variables(char *, Ptr, s.Ptr);
+    swap_variables(uint32, str_length, s.str_length);
   }
 
   /*
@@ -330,8 +254,8 @@ public:
     statement to be run on the remote server, and have a comma after each.
     When the list is complete, I "chop" off the trailing comma
 
-    ex. 
-      String stringobj; 
+    ex.
+      String stringobj;
       stringobj.append("VALUES ('foo', 'fi', 'fo',");
       stringobj.chop();
       stringobj.append(")");
@@ -341,7 +265,6 @@ public:
     VALUES ('foo', 'fi', 'fo',
     VALUES ('foo', 'fi', 'fo'
     VALUES ('foo', 'fi', 'fo')
-      
   */
   inline void chop()
   {
@@ -350,6 +273,329 @@ public:
     DBUG_ASSERT(strlen(Ptr) == str_length);
   }
 
+  // Returns offset to substring or -1
+  int strstr(const Static_binary_string &search, uint32 offset=0);
+  // Returns offset to substring or -1
+  int strrstr(const Static_binary_string &search, uint32 offset=0);
+
+  /*
+    The following append operations do NOT check alloced memory
+    q_*** methods writes values of parameters itself
+    qs_*** methods writes string representation of value
+  */
+  void q_append(const char c)
+  {
+    Ptr[str_length++] = c;
+  }
+  void q_append2b(const uint32 n)
+  {
+    int2store(Ptr + str_length, n);
+    str_length += 2;
+  }
+  void q_append(const uint32 n)
+  {
+    int4store(Ptr + str_length, n);
+    str_length += 4;
+  }
+  void q_append(double d)
+  {
+    float8store(Ptr + str_length, d);
+    str_length += 8;
+  }
+  void q_append(double *d)
+  {
+    float8store(Ptr + str_length, *d);
+    str_length += 8;
+  }
+  void q_append(const char *data, size_t data_len)
+  {
+    memcpy(Ptr + str_length, data, data_len);
+    DBUG_ASSERT(str_length <= UINT_MAX32 - data_len);
+    str_length += (uint)data_len;
+  }
+  void q_append(const LEX_CSTRING *ls)
+  {
+    DBUG_ASSERT(ls->length < UINT_MAX32 &&
+                ((ls->length == 0 && !ls->str) ||
+                 ls->length == strlen(ls->str)));
+    q_append(ls->str, (uint32) ls->length);
+  }
+
+  void write_at_position(int position, uint32 value)
+  {
+    int4store(Ptr + position,value);
+  }
+
+  void qs_append(const char *str)
+  {
+    qs_append(str, (uint32)strlen(str));
+  }
+  void qs_append(const LEX_CSTRING *ls)
+  {
+    DBUG_ASSERT(ls->length < UINT_MAX32 &&
+                ((ls->length == 0 && !ls->str) ||
+                 ls->length == strlen(ls->str)));
+    qs_append(ls->str, (uint32)ls->length);
+  }
+  void qs_append(const char *str, size_t len);
+  void qs_append_hex(const char *str, uint32 len);
+  void qs_append(double d);
+  void qs_append(double *d);
+  inline void qs_append(const char c)
+  {
+     Ptr[str_length]= c;
+     str_length++;
+  }
+  void qs_append(int i);
+  void qs_append(uint i)
+  {
+    qs_append((ulonglong)i);
+  }
+  void qs_append(ulong i)
+  {
+    qs_append((ulonglong)i);
+  }
+  void qs_append(ulonglong i);
+  void qs_append(longlong i, int radix)
+  {
+    char *buff= Ptr + str_length;
+    char *end= ll2str(i, buff, radix, 0);
+    str_length+= uint32(end-buff);
+  }
+};
+
+
+class Binary_string: public Static_binary_string
+{
+  uint32 Alloced_length, extra_alloc;
+  bool alloced, thread_specific;
+  void init_private_data()
+  {
+    Alloced_length= extra_alloc= 0;
+    alloced= thread_specific= false;
+  }
+public:
+  Binary_string()
+  {
+    init_private_data();
+  }
+  explicit Binary_string(size_t length_arg)
+  {
+    init_private_data();
+    (void) real_alloc(length_arg);
+  }
+  explicit Binary_string(const char *str)
+   :Binary_string(str, strlen(str))
+  { }
+  /*
+    NOTE: If one intend to use the c_ptr() method, the following two
+    contructors need the size of memory for STR to be at least LEN+1 (to make
+    room for zero termination).
+  */
+  Binary_string(const char *str, size_t len)
+   :Static_binary_string((char *) str, len)
+  {
+    init_private_data();
+  }
+  Binary_string(char *str, size_t len)
+   :Static_binary_string(str, len)
+  {
+    Alloced_length= (uint32) len;
+    extra_alloc= 0;
+    alloced= thread_specific= 0;
+  }
+  explicit Binary_string(const Binary_string &str)
+   :Static_binary_string(str)
+  {
+    Alloced_length= str.Alloced_length;
+    extra_alloc= 0;
+    alloced= thread_specific= 0;
+  }
+
+  ~Binary_string() { free(); }
+
+  /* Mark variable thread specific it it's not allocated already */
+  inline void set_thread_specific()
+  {
+    if (!alloced)
+      thread_specific= 1;
+  }
+  bool is_alloced() const { return alloced; }
+  inline uint32 alloced_length() const { return Alloced_length;}
+  inline uint32 extra_allocation() const { return extra_alloc;}
+  inline void extra_allocation(size_t len) { extra_alloc= (uint32)len; }
+  inline void mark_as_const() { Alloced_length= 0;}
+
+  inline bool uses_buffer_owned_by(const Binary_string *s) const
+  {
+    return (s->alloced && Ptr >= s->Ptr && Ptr < s->Ptr + s->str_length);
+  }
+
+  /* Swap two string objects. Efficient way to exchange data without memcpy. */
+  void swap(Binary_string &s)
+  {
+    Static_binary_string::swap(s);
+    swap_variables(uint32, Alloced_length, s.Alloced_length);
+    swap_variables(bool, alloced, s.alloced);
+  }
+
+  /**
+     Points the internal buffer to the supplied one. The old buffer is freed.
+     @param str Pointer to the new buffer.
+     @param arg_length Length of the new buffer in characters, excluding any
+            null character.
+     @note The new buffer will not be null terminated.
+  */
+  void set_alloced(char *str, size_t length_arg, size_t alloced_length_arg)
+  {
+    free();
+    Static_binary_string::set(str, length_arg);
+    DBUG_ASSERT(alloced_length_arg < UINT_MAX32);
+    Alloced_length= (uint32) alloced_length_arg;
+  }
+  inline void set(char *str, size_t arg_length)
+  {
+    set_alloced(str, arg_length, arg_length);
+  }
+  inline void set(const char *str, size_t arg_length)
+  {
+    free();
+    Static_binary_string::set((char *) str, arg_length);
+  }
+
+  void set(Binary_string &str, size_t offset, size_t arg_length)
+  {
+    DBUG_ASSERT(&str != this);
+    free();
+    Static_binary_string::set((char*) str.ptr() + offset, arg_length);
+    if (str.Alloced_length)
+      Alloced_length= (uint32) (str.Alloced_length - offset);
+  }
+
+  /* Take over handling of buffer from some other object */
+  void reset(char *ptr_arg, size_t length_arg, size_t alloced_length_arg)
+  {
+    set_alloced(ptr_arg, length_arg, alloced_length_arg);
+    alloced= ptr_arg != 0;
+  }
+
+  /* Forget about the buffer, let some other object handle it */
+  char *release()
+  {
+    char *old= Ptr;
+    Static_binary_string::set(NULL, 0);
+    init_private_data();
+    return old;
+  }
+
+  inline void set_quick(char *str, size_t arg_length)
+  {
+    if (!alloced)
+    {
+      Static_binary_string::set(str, arg_length);
+      Alloced_length= (uint32) arg_length;
+    }
+  }
+
+  inline Binary_string& operator=(const Binary_string &s)
+  {
+    if (&s != this)
+    {
+      /*
+        It is forbidden to do assignments like
+        some_string = substring_of_that_string
+      */
+      DBUG_ASSERT(!s.uses_buffer_owned_by(this));
+      set_alloced((char *) s.Ptr, s.str_length, s.Alloced_length);
+    }
+    return *this;
+  }
+
+  bool set_hex(ulonglong num);
+  bool set_hex(const char *str, uint32 len);
+
+  bool copy();                                  // Alloc string if not alloced
+  bool copy(const Binary_string &s);            // Allocate new string
+  bool copy(const char *s, size_t arg_length);	// Allocate new string
+  bool copy_or_move(const char *s,size_t arg_length);
+
+  bool append_ulonglong(ulonglong val);
+  bool append_longlong(longlong val);
+
+  bool append(const char *s, size_t size)
+  {
+    if (!size)
+      return false;
+    if (realloc_with_extra_if_needed(str_length + size))
+      return true;
+    q_append(s, size);
+    return false;
+  }
+  bool append(const Binary_string &s)
+  {
+    return append(s.ptr(), s.length());
+  }
+  bool append(IO_CACHE* file, uint32 arg_length);
+
+  inline bool append_char(char chr)
+  {
+    if (str_length < Alloced_length)
+    {
+      Ptr[str_length++]= chr;
+    }
+    else
+    {
+      if (unlikely(realloc_with_extra(str_length + 1)))
+	return true;
+      Ptr[str_length++]= chr;
+    }
+    return false;
+  }
+  bool append_hex(const char *src, uint32 srclen)
+  {
+    for (const char *src_end= src + srclen ; src != src_end ; src++)
+    {
+      if (unlikely(append_char(_dig_vec_lower[((uchar) *src) >> 4])) ||
+          unlikely(append_char(_dig_vec_lower[((uchar) *src) & 0x0F])))
+        return true;
+    }
+    return false;
+  }
+
+  bool append_with_step(const char *s, uint32 arg_length, uint32 step_alloc)
+  {
+    uint32 new_length= arg_length + str_length;
+    if (new_length > Alloced_length &&
+        unlikely(realloc(new_length + step_alloc)))
+      return true;
+    q_append(s, arg_length);
+    return false;
+  }
+
+  inline char *c_ptr()
+  {
+    DBUG_ASSERT(!alloced || !Ptr || !Alloced_length ||
+                (Alloced_length >= (str_length + 1)));
+
+    if (!Ptr || Ptr[str_length])              // Should be safe
+      (void) realloc(str_length);
+    return Ptr;
+  }
+  inline char *c_ptr_quick()
+  {
+    if (Ptr && str_length < Alloced_length)
+      Ptr[str_length]=0;
+    return Ptr;
+  }
+  inline char *c_ptr_safe()
+  {
+    if (Ptr && str_length < Alloced_length)
+      Ptr[str_length]=0;
+    else
+      (void) realloc(str_length);
+    return Ptr;
+  }
+
   inline void free()
   {
     if (alloced)
@@ -358,8 +604,7 @@ public:
       my_free(Ptr);
     }
     Alloced_length= extra_alloc= 0;
-    Ptr=0;
-    str_length=0;				/* Safety */
+    Static_binary_string::set(NULL, 0); // Safety
   }
   inline bool alloc(size_t arg_length)
   {
@@ -367,13 +612,13 @@ public:
       return 0;
     return real_alloc(arg_length);
   }
-  bool real_alloc(size_t arg_length);			// Empties old string
+  bool real_alloc(size_t arg_length);  // Empties old string
   bool realloc_raw(size_t arg_length);
   bool realloc(size_t arg_length)
   {
     if (realloc_raw(arg_length))
       return TRUE;
-    Ptr[arg_length]=0;        // This make other funcs shorter
+    Ptr[arg_length]= 0; // This make other funcs shorter
     return FALSE;
   }
   bool realloc_with_extra(size_t arg_length)
@@ -407,37 +652,179 @@ public:
                                 arg_length,MYF((thread_specific ?
                                                 MY_THREAD_SPECIFIC : 0))))))
       {
-	Alloced_length = 0;
-	real_alloc(arg_length);
+        Alloced_length= 0;
+        real_alloc(arg_length);
       }
       else
       {
-	Ptr=new_ptr;
-	Alloced_length=(uint32)arg_length;
+        Ptr= new_ptr;
+        Alloced_length= (uint32) arg_length;
       }
     }
   }
-  bool is_alloced() const { return alloced; }
+  void move(Binary_string &s)
+  {
+    set_alloced(s.Ptr, s.str_length, s.Alloced_length);
+    extra_alloc= s.extra_alloc;
+    alloced= s.alloced;
+    thread_specific= s.thread_specific;
+    s.alloced= 0;
+  }
+  bool fill(uint32 max_length,char fill);
+  /*
+    Replace substring with string
+    If wrong parameter or not enough memory, do nothing
+  */
+  bool replace(uint32 offset,uint32 arg_length, const char *to, uint32 length);
+  bool replace(uint32 offset,uint32 arg_length, const Static_binary_string &to)
+  {
+    return replace(offset,arg_length,to.ptr(),to.length());
+  }
+
+  int reserve(size_t space_needed)
+  {
+    return realloc(str_length + space_needed);
+  }
+  int reserve(size_t space_needed, size_t grow_by);
+
+  inline char *prep_append(uint32 arg_length, uint32 step_alloc)
+  {
+    uint32 new_length= arg_length + str_length;
+    if (new_length > Alloced_length)
+    {
+      if (unlikely(realloc(new_length + step_alloc)))
+        return 0;
+    }
+    uint32 old_length= str_length;
+    str_length+= arg_length;
+    return Ptr + old_length;                  // Area to use
+  }
+
+
+  void q_net_store_length(ulonglong length)
+  {
+    DBUG_ASSERT(Alloced_length >= (str_length + net_length_size(length)));
+    char *pos= (char *) net_store_length((uchar *)(Ptr + str_length), length);
+    str_length= uint32(pos - Ptr);
+  }
+  void q_net_store_data(const uchar *from, size_t length)
+  {
+    DBUG_ASSERT(length < UINT_MAX32);
+    DBUG_ASSERT(Alloced_length >= (str_length + length +
+                                   net_length_size(length)));
+    q_net_store_length(length);
+    q_append((const char *)from, (uint32) length);
+  }
+};
+
+
+class String: public Charset, public Binary_string
+{
+public:
+  String() { }
+  String(size_t length_arg)
+   :Binary_string(length_arg)
+  { }
+  String(const char *str, CHARSET_INFO *cs)
+   :Charset(cs),
+    Binary_string(str)
+  { }
+  /*
+    NOTE: If one intend to use the c_ptr() method, the following two
+    contructors need the size of memory for STR to be at least LEN+1 (to make
+    room for zero termination).
+  */
+  String(const char *str, size_t len, CHARSET_INFO *cs)
+   :Charset(cs),
+    Binary_string((char *) str, len)
+  { }
+  String(char *str, size_t len, CHARSET_INFO *cs)
+   :Charset(cs),
+    Binary_string(str, len)
+  { }
+  String(const String &str)
+   :Charset(str),
+    Binary_string(str)
+  { }
+
+  void set(String &str,size_t offset,size_t arg_length)
+  {
+    Binary_string::set(str, offset, arg_length);
+    set_charset(str);
+  }
+  inline void set(char *str,size_t arg_length, CHARSET_INFO *cs)
+  {
+    Binary_string::set(str, arg_length);
+    set_charset(cs);
+  }
+  inline void set(const char *str,size_t arg_length, CHARSET_INFO *cs)
+  {
+    Binary_string::set(str, arg_length);
+    set_charset(cs);
+  }
+  bool set_ascii(const char *str, size_t arg_length);
+  inline void set_quick(char *str,size_t arg_length, CHARSET_INFO *cs)
+  {
+    Binary_string::set_quick(str, arg_length);
+    set_charset(cs);
+  }
+  bool set_int(longlong num, bool unsigned_flag, CHARSET_INFO *cs);
+  bool set(int num, CHARSET_INFO *cs) { return set_int(num, false, cs); }
+  bool set(uint num, CHARSET_INFO *cs) { return set_int(num, true, cs); }
+  bool set(long num, CHARSET_INFO *cs) { return set_int(num, false, cs); }
+  bool set(ulong num, CHARSET_INFO *cs) { return set_int(num, true, cs); }
+  bool set(longlong num, CHARSET_INFO *cs) { return set_int(num, false, cs); }
+  bool set(ulonglong num, CHARSET_INFO *cs) { return set_int((longlong)num, true, cs); }
+  bool set_real(double num,uint decimals, CHARSET_INFO *cs);
+
+  bool set_hex(ulonglong num)
+  {
+    set_charset(&my_charset_latin1);
+    return Binary_string::set_hex(num);
+  }
+  bool set_hex(const char *str, uint32 len)
+  {
+    set_charset(&my_charset_latin1);
+    return Binary_string::set_hex(str, len);
+  }
+
+  /* Take over handling of buffer from some other object */
+  void reset(char *ptr_arg, size_t length_arg, size_t alloced_length_arg,
+             CHARSET_INFO *cs)
+  {
+    Binary_string::reset(ptr_arg, length_arg, alloced_length_arg);
+    set_charset(cs);
+  }
+
   inline String& operator = (const String &s)
   {
     if (&s != this)
     {
-      /*
-        It is forbidden to do assignments like 
-        some_string = substring_of_that_string
-       */
-      DBUG_ASSERT(!s.uses_buffer_owned_by(this));
-      free();
-      Ptr=s.Ptr ; str_length=s.str_length ; Alloced_length=s.Alloced_length;
-      str_charset=s.str_charset;
+      set_charset(s);
+      Binary_string::operator=(s);
     }
     return *this;
   }
 
-  bool copy();					// Alloc string if not alloced
-  bool copy(const String &s);			// Allocate new string
-  bool copy(const char *s,size_t arg_length, CHARSET_INFO *cs);	// Allocate new string
-  bool copy_or_move(const char *s,size_t arg_length, CHARSET_INFO *cs);
+  bool copy()
+  {
+    return Binary_string::copy();
+  }
+  bool copy(const String &s)
+  {
+    set_charset(s);
+    return Binary_string::copy(s);
+  }
+  bool copy(const char *s, size_t arg_length, CHARSET_INFO *cs)
+  {
+    set_charset(cs);
+    return Binary_string::copy(s, arg_length);
+  }
+  bool copy_or_move(const char *s, size_t arg_length, CHARSET_INFO *cs)
+  {
+    set_charset(cs);
+    return Binary_string::copy_or_move(s, arg_length);
+  }
   static bool needs_conversion(size_t arg_length,
   			       CHARSET_INFO *cs_from, CHARSET_INFO *cs_to,
 			       uint32 *offset);
@@ -459,206 +846,84 @@ public:
   {
     if (unlikely(alloc(tocs->mbmaxlen * src_length)))
       return true;
-    str_length= copier->well_formed_copy(tocs, Ptr, Alloced_length,
+    str_length= copier->well_formed_copy(tocs, Ptr, alloced_length(),
                                          fromcs, src, (uint)src_length, (uint)nchars);
-    str_charset= tocs;
+    set_charset(tocs);
     return false;
   }
-  void move(String &s)
-  {
-    free();
-    Ptr=s.Ptr ; str_length=s.str_length ; Alloced_length=s.Alloced_length;
-    extra_alloc= s.extra_alloc;
-    alloced= s.alloced;
-    thread_specific= s.thread_specific;
-    s.alloced= 0;
-  }
-  bool append(const String &s);
-  bool append(const char *s);
-  bool append(const LEX_STRING *ls)
-  {
-    DBUG_ASSERT(ls->length < UINT_MAX32 &&
-                ((ls->length == 0 && !ls->str) ||
-                 ls->length == strlen(ls->str)));
-    return append(ls->str, (uint32) ls->length);
-  }
-  bool append(const LEX_CSTRING *ls)
-  {
-    DBUG_ASSERT(ls->length < UINT_MAX32 &&
-                ((ls->length == 0 && !ls->str) ||
-                 ls->length == strlen(ls->str)));
-    return append(ls->str, (uint32) ls->length);
-  }
-  bool append(const LEX_CSTRING &ls)
+  // Append without character set conversion
+  bool append(const String &s)
   {
-    return append(&ls);
+    return Binary_string::append(s);
   }
-  bool append(const char *s, size_t size);
-  bool append(const char *s, size_t arg_length, CHARSET_INFO *cs);
-  bool append_ulonglong(ulonglong val);
-  bool append_longlong(longlong val);
-  bool append(IO_CACHE* file, uint32 arg_length);
-  bool append_with_prefill(const char *s, uint32 arg_length, 
-			   uint32 full_length, char fill_char);
-  bool append_parenthesized(long nr, int radix= 10);
-  int strstr(const String &search,uint32 offset=0); // Returns offset to substring or -1
-  int strrstr(const String &search,uint32 offset=0); // Returns offset to substring or -1
-  bool replace(uint32 offset,uint32 arg_length,const char *to,uint32 length);
-  bool replace(uint32 offset,uint32 arg_length,const String &to);
   inline bool append(char chr)
   {
-    if (str_length < Alloced_length)
-    {
-      Ptr[str_length++]=chr;
-    }
-    else
-    {
-      if (unlikely(realloc_with_extra(str_length + 1)))
-	return 1;
-      Ptr[str_length++]=chr;
-    }
-    return 0;
+    return Binary_string::append_char(chr);
   }
   bool append_hex(const char *src, uint32 srclen)
   {
-    for (const char *src_end= src + srclen ; src != src_end ; src++)
-    {
-      if (unlikely(append(_dig_vec_lower[((uchar) *src) >> 4])) ||
-          unlikely(append(_dig_vec_lower[((uchar) *src) & 0x0F])))
-        return true;
-    }
-    return false;
+    return Binary_string::append_hex(src, srclen);
   }
   bool append_hex(const uchar *src, uint32 srclen)
   {
-    return append_hex((const char*)src, srclen);
-  }
-  bool fill(uint32 max_length,char fill);
-  void strip_sp();
-  friend int sortcmp(const String *a,const String *b, CHARSET_INFO *cs);
-  friend int stringcmp(const String *a,const String *b);
-  friend String *copy_if_not_alloced(String *a,String *b,uint32 arg_length);
-  friend class Field;
-  uint32 numchars() const;
-  int charpos(longlong i,uint32 offset=0);
-
-  int reserve(size_t space_needed)
-  {
-    return realloc(str_length + space_needed);
-  }
-  int reserve(size_t space_needed, size_t grow_by);
-
-  /*
-    The following append operations do NOT check alloced memory
-    q_*** methods writes values of parameters itself
-    qs_*** methods writes string representation of value
-  */
-  void q_append(const char c)
-  {
-    Ptr[str_length++] = c;
+    return Binary_string::append_hex((const char*)src, srclen);
   }
-  void q_append2b(const uint32 n)
+  bool append(IO_CACHE* file, uint32 arg_length)
   {
-    int2store(Ptr + str_length, n);
-    str_length += 2;
+    return Binary_string::append(file, arg_length);
   }
-  void q_append(const uint32 n)
-  {
-    int4store(Ptr + str_length, n);
-    str_length += 4;
-  }
-  void q_append(double d)
+  inline bool append(const char *s, uint32 arg_length, uint32 step_alloc)
   {
-    float8store(Ptr + str_length, d);
-    str_length += 8;
+    return append_with_step(s, arg_length, step_alloc);
   }
-  void q_append(double *d)
-  {
-    float8store(Ptr + str_length, *d);
-    str_length += 8;
-  }
-  void q_append(const char *data, size_t data_len)
+
+  // Append with optional character set conversion from ASCII (e.g. to UCS2)
+  bool append(const char *s)
   {
-    memcpy(Ptr + str_length, data, data_len);
-    DBUG_ASSERT(str_length <= UINT_MAX32 - data_len);
-    str_length += (uint)data_len;
+    return append(s, strlen(s));
   }
-  void q_append(const LEX_CSTRING *ls)
+  bool append(const LEX_STRING *ls)
   {
     DBUG_ASSERT(ls->length < UINT_MAX32 &&
                 ((ls->length == 0 && !ls->str) ||
                  ls->length == strlen(ls->str)));
-    q_append(ls->str, (uint32) ls->length);
-  }
-
-  void write_at_position(int position, uint32 value)
-  {
-    int4store(Ptr + position,value);
-  }
-
-  void qs_append(const char *str)
-  {
-    qs_append(str, (uint32)strlen(str));
+    return append(ls->str, (uint32) ls->length);
   }
-  void qs_append(const LEX_CSTRING *ls)
+  bool append(const LEX_CSTRING *ls)
   {
     DBUG_ASSERT(ls->length < UINT_MAX32 &&
                 ((ls->length == 0 && !ls->str) ||
                  ls->length == strlen(ls->str)));
-    qs_append(ls->str, (uint32)ls->length);
-  }
-  void qs_append(const char *str, size_t len);
-  void qs_append_hex(const char *str, uint32 len);
-  void qs_append(double d);
-  void qs_append(double *d);
-  inline void qs_append(const char c)
-  {
-     Ptr[str_length]= c;
-     str_length++;
-  }
-  void qs_append(int i);
-  void qs_append(uint i)
-  {
-    qs_append((ulonglong)i);
-  }
-  void qs_append(ulong i)
-  {
-    qs_append((ulonglong)i);
+    return append(ls->str, (uint32) ls->length);
   }
-  void qs_append(ulonglong i);
-  void qs_append(longlong i, int radix)
+  bool append(const LEX_CSTRING &ls)
   {
-    char *buff= Ptr + str_length;
-    char *end= ll2str(i, buff, radix, 0);
-    str_length+= uint32(end-buff);
+    return append(&ls);
   }
+  bool append(const char *s, size_t size);
+  bool append_with_prefill(const char *s, uint32 arg_length,
+			   uint32 full_length, char fill_char);
+  bool append_parenthesized(long nr, int radix= 10);
 
-  /* Inline (general) functions used by the protocol functions */
+  // Append with optional character set conversion from cs to charset()
+  bool append(const char *s, size_t arg_length, CHARSET_INFO *cs);
 
-  inline char *prep_append(uint32 arg_length, uint32 step_alloc)
+  void strip_sp();
+  friend int sortcmp(const String *a,const String *b, CHARSET_INFO *cs);
+  friend int stringcmp(const String *a,const String *b);
+  friend String *copy_if_not_alloced(String *a,String *b,uint32 arg_length);
+  friend class Field;
+  uint32 numchars() const
   {
-    uint32 new_length= arg_length + str_length;
-    if (new_length > Alloced_length)
-    {
-      if (unlikely(realloc(new_length + step_alloc)))
-        return 0;
-    }
-    uint32 old_length= str_length;
-    str_length+= arg_length;
-    return Ptr+ old_length;			/* Area to use */
+    return (uint32) Charset::numchars(ptr(), end());
   }
-
-
-  inline bool append(const char *s, uint32 arg_length, uint32 step_alloc)
+  int charpos(longlong i, uint32 offset=0)
   {
-    uint32 new_length= arg_length + str_length;
-    if (new_length > Alloced_length &&
-        unlikely(realloc(new_length + step_alloc)))
-      return TRUE;
-    memcpy(Ptr+str_length, s, arg_length);
-    str_length+= arg_length;
-    return FALSE;
+    if (i <= 0)
+      return (int) i;
+    return (int) Charset::charpos(ptr() + offset, end(), (size_t) i);
   }
+
   void print(String *to) const;
   void print_with_conversion(String *to, CHARSET_INFO *cs) const;
   void print(String *to, CHARSET_INFO *cs) const
@@ -681,13 +946,12 @@ public:
     return append_for_single_quote(st, (uint32) len);
   }
 
-  /* Swap two string objects. Efficient way to exchange data without memcpy. */
-  void swap(String &s);
-
-  inline bool uses_buffer_owned_by(const String *s) const
+  void swap(String &s)
   {
-    return (s->alloced && Ptr >= s->Ptr && Ptr < s->Ptr + s->str_length);
+    Charset::swap(s);
+    Binary_string::swap(s);
   }
+
   uint well_formed_length() const
   {
     return (uint) Well_formed_prefix(charset(), ptr(), length()).length();
@@ -698,36 +962,12 @@ public:
       return TRUE;
     if (charset()->mbminlen > 1)
       return FALSE;
-    for (const char *c= ptr(), *c_end= c + length(); c < c_end; c++)
-    {
-      if (!my_isascii(*c))
-        return FALSE;
-    }
-    return TRUE;
-  }
-  bool bin_eq(const String *other) const
-  {
-    return length() == other->length() &&
-           !memcmp(ptr(), other->ptr(), length());
+    return !has_8bit_bytes();
   }
   bool eq(const String *other, CHARSET_INFO *cs) const
   {
     return !sortcmp(this, other, cs);
   }
-  void q_net_store_length(ulonglong length)
-  {
-    DBUG_ASSERT(Alloced_length >= (str_length + net_length_size(length)));
-    char *pos= (char *) net_store_length((uchar *)(Ptr + str_length), length);
-    str_length= uint32(pos - Ptr);
-  }
-  void q_net_store_data(const uchar *from, size_t length)
-  {
-    DBUG_ASSERT(length < UINT_MAX32);
-    DBUG_ASSERT(Alloced_length >= (str_length + length +
-                                   net_length_size(length)));
-    q_net_store_length(length);
-    q_append((const char *)from, (uint32) length);
-  }
 };
 
 
@@ -774,7 +1014,7 @@ public:
 
 
 static inline bool check_if_only_end_space(CHARSET_INFO *cs,
-                                           const char *str, 
+                                           const char *str,
                                            const char *end)
 {
   return str+ cs->cset->scan(cs, str, end, MY_SEQ_SPACES) == end;
diff --git a/sql/sql_type.h b/sql/sql_type.h
index a5d8f4ae6df..0e023a3ef65 100644
--- a/sql/sql_type.h
+++ b/sql/sql_type.h
@@ -1940,10 +1940,6 @@ public:
   { }
 
   Datetime(THD *thd, int *warn, const MYSQL_TIME *from, date_conv_mode_t flags);
-  Datetime()
-  {
-    set_zero_time(this, MYSQL_TIMESTAMP_DATETIME);
-  }
   Datetime(THD *thd, MYSQL_TIME_STATUS *status,
            const char *str, size_t len, CHARSET_INFO *cs,
            const date_mode_t fuzzydate)
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index 3a4871875e5..22d017fe345 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -1942,6 +1942,19 @@ Sys_var_last_gtid::session_value_ptr(THD *thd, const LEX_CSTRING *base)
 }
 
 
+static Sys_var_uint Sys_gtid_cleanup_batch_size(
+       "gtid_cleanup_batch_size",
+       "Normally does not need tuning. How many old rows must accumulate in "
+       "the mysql.gtid_slave_pos table before a background job will be run to "
+       "delete them. Can be increased to reduce number of commits if "
+       "using many different engines with --gtid_pos_auto_engines, or to "
+       "reduce CPU overhead if using a huge number of different "
+       "gtid_domain_ids. Can be decreased to reduce number of old rows in the "
+       "table.",
+       GLOBAL_VAR(opt_gtid_cleanup_batch_size), CMD_LINE(REQUIRED_ARG),
+       VALID_RANGE(0,2147483647), DEFAULT(64), BLOCK_SIZE(1));
+
+
 static bool
 check_slave_parallel_threads(sys_var *self, THD *thd, set_var *var)
 {
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 6031fcf3f98..e0a066a595a 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -11076,9 +11076,6 @@ err_col:
 	case DB_SUCCESS:
 		ut_ad(table);
 		m_table = table;
-		if (m_flags2 & DICT_TF2_FTS) {
-			fts_optimize_add_table(table);
-		}
 		DBUG_RETURN(0);
 	default:
 		break;
@@ -11097,7 +11094,8 @@ err_col:
 			 : ER_TABLESPACE_EXISTS, MYF(0), display_name);
 	}
 
-	DBUG_RETURN(convert_error_code_to_mysql(err, m_flags, m_thd));}
+	DBUG_RETURN(convert_error_code_to_mysql(err, m_flags, m_thd));
+}
 
 /*****************************************************************//**
 Creates an index in an InnoDB database. */
@@ -12485,6 +12483,10 @@ create_table_info_t::create_table_update_dict()
 			trx_free(m_trx);
 			DBUG_RETURN(-1);
 		}
+
+		mutex_enter(&dict_sys->mutex);
+		fts_optimize_add_table(innobase_table);
+		mutex_exit(&dict_sys->mutex);
 	}
 
 	if (const Field* ai = m_form->found_next_number_field) {
diff --git a/storage/innobase/row/row0ftsort.cc b/storage/innobase/row/row0ftsort.cc
index f46c381fbf7..941b7798c39 100644
--- a/storage/innobase/row/row0ftsort.cc
+++ b/storage/innobase/row/row0ftsort.cc
@@ -1579,9 +1579,6 @@ row_fts_merge_insert(
 	dict_table_t*		aux_table;
 	dict_index_t*		aux_index;
 	trx_t*			trx;
-	byte			trx_id_buf[6];
-	roll_ptr_t		roll_ptr = 0;
-	dfield_t*		field;
 
 	ut_ad(index);
 	ut_ad(table);
@@ -1692,16 +1689,13 @@ row_fts_merge_insert(
 			      dict_index_get_n_fields(aux_index));
 
 	/* Set TRX_ID and ROLL_PTR */
-	trx_write_trx_id(trx_id_buf, trx->id);
-	field = dtuple_get_nth_field(ins_ctx.tuple, 2);
-	dfield_set_data(field, &trx_id_buf, 6);
+	dfield_set_data(dtuple_get_nth_field(ins_ctx.tuple, 2),
+			&reset_trx_id, DATA_TRX_ID_LEN);
+	dfield_set_data(dtuple_get_nth_field(ins_ctx.tuple, 3),
+			&reset_trx_id[DATA_TRX_ID_LEN], DATA_ROLL_PTR_LEN);
 
-	field = dtuple_get_nth_field(ins_ctx.tuple, 3);
-	dfield_set_data(field, &roll_ptr, 7);
+	ut_d(ins_ctx.aux_index_id = id);
 
-#ifdef UNIV_DEBUG
-	ins_ctx.aux_index_id = id;
-#endif
 	const ulint space = table->space_id;
 
 	for (i = 0; i < fts_sort_pll_degree; i++) {
diff --git a/storage/rocksdb/mysql-test/rocksdb_rpl/r/mdev12179.result b/storage/rocksdb/mysql-test/rocksdb_rpl/r/mdev12179.result
index 9c20fea97ae..a1e501f78f4 100644
--- a/storage/rocksdb/mysql-test/rocksdb_rpl/r/mdev12179.result
+++ b/storage/rocksdb/mysql-test/rocksdb_rpl/r/mdev12179.result
@@ -2,6 +2,7 @@ include/master-slave.inc
 [connection master]
 connection server_2;
 include/stop_slave.inc
+SET GLOBAL gtid_cleanup_batch_size = 999999999;
 CHANGE MASTER TO master_use_gtid=slave_pos;
 SET sql_log_bin=0;
 CREATE TABLE mysql.gtid_slave_pos_innodb LIKE mysql.gtid_slave_pos;
@@ -41,6 +42,8 @@ a
 1
 SELECT * FROM mysql.gtid_slave_pos ORDER BY sub_id;
 domain_id	sub_id	server_id	seq_no
+0	1	1	1
+0	2	1	2
 0	3	1	3
 0	4	1	4
 SELECT * FROM ( SELECT * FROM mysql.gtid_slave_pos_innodb
@@ -121,6 +124,21 @@ Transactions_multi_engine	6
 DELETE FROM t1 WHERE a >= 100;
 DELETE FROM t2 WHERE a >= 100;
 DELETE FROM t3 WHERE a >= 100;
+connection server_1;
+include/save_master_gtid.inc
+connection server_2;
+include/sync_with_master_gtid.inc
+SELECT COUNT(*)>=10 FROM mysql.gtid_slave_pos;
+COUNT(*)>=10
+1
+SELECT COUNT(*)>=10 FROM ( SELECT * FROM mysql.gtid_slave_pos_innodb
+UNION ALL SELECT * FROM mysql.gtid_slave_pos_innodb_redundant) inner_select;
+COUNT(*)>=10
+1
+SELECT COUNT(*)>=10 FROM mysql.gtid_slave_pos_rocksdb;
+COUNT(*)>=10
+1
+SET GLOBAL gtid_cleanup_batch_size = 3;
 connection server_2;
 include/stop_slave.inc
 SET sql_log_bin=0;
diff --git a/storage/rocksdb/mysql-test/rocksdb_rpl/t/mdev12179.test b/storage/rocksdb/mysql-test/rocksdb_rpl/t/mdev12179.test
index e0d16e7f242..631d9ca533f 100644
--- a/storage/rocksdb/mysql-test/rocksdb_rpl/t/mdev12179.test
+++ b/storage/rocksdb/mysql-test/rocksdb_rpl/t/mdev12179.test
@@ -4,6 +4,12 @@
 
 --connection server_2
 --source include/stop_slave.inc
+
+# Set GTID cleanup limit high enough that cleanup will not run and we
+# can rely on consistent table output in .result.
+--let $old_gtid_cleanup_batch_size=`SELECT @@GLOBAL.gtid_cleanup_batch_size`
+SET GLOBAL gtid_cleanup_batch_size = 999999999;
+
 CHANGE MASTER TO master_use_gtid=slave_pos;
 SET sql_log_bin=0;
 CREATE TABLE mysql.gtid_slave_pos_innodb LIKE mysql.gtid_slave_pos;
@@ -89,6 +95,82 @@ DELETE FROM t2 WHERE a >= 100;
 DELETE FROM t3 WHERE a >= 100;
 
 
+# Create a bunch more GTIDs in mysql.gtid_slave_pos* tables to test with.
+--connection server_1
+--disable_query_log
+let $i=10;
+while ($i) {
+  eval INSERT INTO t1 VALUES (300+$i);
+  eval INSERT INTO t2 VALUES (300+$i);
+  eval INSERT INTO t3 VALUES (300+$i);
+  dec $i;
+}
+--enable_query_log
+--source include/save_master_gtid.inc
+
+--connection server_2
+--source include/sync_with_master_gtid.inc
+
+# Check that we have many rows in mysql.gtid_slave_pos now (since
+# @@gtid_cleanup_batch_size was set to a huge value). No need to check
+# for an exact number, since that will require changing .result if
+# anything changes prior to this point, and we just need to know that
+# we have still have some data in the tables to make the following
+# test effective.
+SELECT COUNT(*)>=10 FROM mysql.gtid_slave_pos;
+SELECT COUNT(*)>=10 FROM ( SELECT * FROM mysql.gtid_slave_pos_innodb
+      UNION ALL SELECT * FROM mysql.gtid_slave_pos_innodb_redundant) inner_select;
+SELECT COUNT(*)>=10 FROM mysql.gtid_slave_pos_rocksdb;
+
+# Check that old GTID rows will be deleted when batch delete size is
+# set reasonably. Old row deletion is not 100% deterministic (by design), so
+# we must wait for it to occur, but it should occur eventually.
+SET GLOBAL gtid_cleanup_batch_size = 3;
+let $i=40;
+--disable_query_log
+--let $keep_include_silent=1
+while ($i) {
+  let N=`SELECT 1+($i MOD 3)`;
+  --connection server_1
+  eval UPDATE t$N SET a=a+1 WHERE a=(SELECT MAX(a) FROM t$N);
+  --source include/save_master_gtid.inc
+  --connection server_2
+  --source include/sync_with_master_gtid.inc
+  let $j=50;
+  while ($j) {
+    let $is_done=`SELECT SUM(a)=1 FROM (
+	 SELECT COUNT(*) AS a FROM mysql.gtid_slave_pos
+	 UNION ALL
+	 SELECT COUNT(*) AS a FROM ( SELECT * FROM mysql.gtid_slave_pos_innodb
+	       UNION ALL SELECT * FROM mysql.gtid_slave_pos_innodb_redundant) inner_select
+	 UNION ALL
+	 SELECT COUNT(*) AS a FROM mysql.gtid_slave_pos_rocksdb) outer_select`;
+    if ($is_done) {
+      let $j=0;
+    }
+    if (!$is_done) {
+      real_sleep 0.1;
+      dec $j;
+    }
+  }
+  dec $i;
+  if ($is_done) {
+    let $i=0;
+  }
+}
+--enable_query_log
+--let $keep_include_silent=0
+if (!$is_done) {
+  --echo Timed out waiting for mysql.gtid_slave_pos* tables to be cleaned up
+}
+
+--disable_query_log
+DELETE FROM t1 WHERE a >= 100;
+DELETE FROM t2 WHERE a >= 100;
+DELETE FROM t3 WHERE a >= 100;
+--enable_query_log
+
+
 # Test status variables Rpl_transactions_multi_engine and Transactions_gtid_foreign_engine.
 # Have mysql.gtid_slave_pos* for myisam and innodb but not rocksdb.
 --connection server_2
@@ -223,6 +305,9 @@ SHOW STATUS LIKE "%transactions%engine";
 SET sql_log_bin=0;
 DROP TABLE mysql.gtid_slave_pos_innodb;
 SET sql_log_bin=1;
+--disable_query_log
+eval SET GLOBAL gtid_cleanup_batch_size = $old_gtid_cleanup_batch_size;
+--enable_query_log
 
 --connection server_1
 DROP TABLE t1;
diff --git a/storage/tokudb/mysql-test/tokudb_rpl/r/mdev12179.result b/storage/tokudb/mysql-test/tokudb_rpl/r/mdev12179.result
index d4532eec4e2..d79e7e59aa4 100644
--- a/storage/tokudb/mysql-test/tokudb_rpl/r/mdev12179.result
+++ b/storage/tokudb/mysql-test/tokudb_rpl/r/mdev12179.result
@@ -2,6 +2,7 @@ include/master-slave.inc
 [connection master]
 connection server_2;
 include/stop_slave.inc
+SET GLOBAL gtid_cleanup_batch_size = 999999999;
 CHANGE MASTER TO master_use_gtid=slave_pos;
 SET sql_log_bin=0;
 CREATE TABLE mysql.gtid_slave_pos_innodb LIKE mysql.gtid_slave_pos;
@@ -41,6 +42,8 @@ a
 1
 SELECT * FROM mysql.gtid_slave_pos ORDER BY sub_id;
 domain_id	sub_id	server_id	seq_no
+0	1	1	1
+0	2	1	2
 0	3	1	3
 0	4	1	4
 SELECT * FROM ( SELECT * FROM mysql.gtid_slave_pos_innodb
@@ -121,6 +124,21 @@ Transactions_multi_engine	6
 DELETE FROM t1 WHERE a >= 100;
 DELETE FROM t2 WHERE a >= 100;
 DELETE FROM t3 WHERE a >= 100;
+connection server_1;
+include/save_master_gtid.inc
+connection server_2;
+include/sync_with_master_gtid.inc
+SELECT COUNT(*)>=10 FROM mysql.gtid_slave_pos;
+COUNT(*)>=10
+1
+SELECT COUNT(*)>=10 FROM ( SELECT * FROM mysql.gtid_slave_pos_innodb
+UNION ALL SELECT * FROM mysql.gtid_slave_pos_innodb_redundant) inner_select;
+COUNT(*)>=10
+1
+SELECT COUNT(*)>=10 FROM mysql.gtid_slave_pos_tokudb;
+COUNT(*)>=10
+1
+SET GLOBAL gtid_cleanup_batch_size = 3;
 connection server_2;
 include/stop_slave.inc
 SET sql_log_bin=0;
diff --git a/storage/tokudb/mysql-test/tokudb_rpl/t/mdev12179.test b/storage/tokudb/mysql-test/tokudb_rpl/t/mdev12179.test
index ceb119cd0dc..1d19a25889e 100644
--- a/storage/tokudb/mysql-test/tokudb_rpl/t/mdev12179.test
+++ b/storage/tokudb/mysql-test/tokudb_rpl/t/mdev12179.test
@@ -4,6 +4,12 @@
 
 --connection server_2
 --source include/stop_slave.inc
+
+# Set GTID cleanup limit high enough that cleanup will not run and we
+# can rely on consistent table output in .result.
+--let $old_gtid_cleanup_batch_size=`SELECT @@GLOBAL.gtid_cleanup_batch_size`
+SET GLOBAL gtid_cleanup_batch_size = 999999999;
+
 CHANGE MASTER TO master_use_gtid=slave_pos;
 SET sql_log_bin=0;
 CREATE TABLE mysql.gtid_slave_pos_innodb LIKE mysql.gtid_slave_pos;
@@ -89,6 +95,82 @@ DELETE FROM t2 WHERE a >= 100;
 DELETE FROM t3 WHERE a >= 100;
 
 
+# Create a bunch more GTIDs in mysql.gtid_slave_pos* tables to test with.
+--connection server_1
+--disable_query_log
+let $i=10;
+while ($i) {
+  eval INSERT INTO t1 VALUES (300+$i);
+  eval INSERT INTO t2 VALUES (300+$i);
+  eval INSERT INTO t3 VALUES (300+$i);
+  dec $i;
+}
+--enable_query_log
+--source include/save_master_gtid.inc
+
+--connection server_2
+--source include/sync_with_master_gtid.inc
+
+# Check that we have many rows in mysql.gtid_slave_pos now (since
+# @@gtid_cleanup_batch_size was set to a huge value). No need to check
+# for an exact number, since that will require changing .result if
+# anything changes prior to this point, and we just need to know that
+# we have still have some data in the tables to make the following
+# test effective.
+SELECT COUNT(*)>=10 FROM mysql.gtid_slave_pos;
+SELECT COUNT(*)>=10 FROM ( SELECT * FROM mysql.gtid_slave_pos_innodb
+      UNION ALL SELECT * FROM mysql.gtid_slave_pos_innodb_redundant) inner_select;
+SELECT COUNT(*)>=10 FROM mysql.gtid_slave_pos_tokudb;
+
+# Check that old GTID rows will be deleted when batch delete size is
+# set reasonably. Old row deletion is not 100% deterministic (by design), so
+# we must wait for it to occur, but it should occur eventually.
+SET GLOBAL gtid_cleanup_batch_size = 3;
+let $i=40;
+--disable_query_log
+--let $keep_include_silent=1
+while ($i) {
+  let N=`SELECT 1+($i MOD 3)`;
+  --connection server_1
+  eval UPDATE t$N SET a=a+1 WHERE a=(SELECT MAX(a) FROM t$N);
+  --source include/save_master_gtid.inc
+  --connection server_2
+  --source include/sync_with_master_gtid.inc
+  let $j=50;
+  while ($j) {
+    let $is_done=`SELECT SUM(a)=1 FROM (
+	 SELECT COUNT(*) AS a FROM mysql.gtid_slave_pos
+	 UNION ALL
+	 SELECT COUNT(*) AS a FROM ( SELECT * FROM mysql.gtid_slave_pos_innodb
+	       UNION ALL SELECT * FROM mysql.gtid_slave_pos_innodb_redundant) inner_select
+	 UNION ALL
+	 SELECT COUNT(*) AS a FROM mysql.gtid_slave_pos_tokudb) outer_select`;
+    if ($is_done) {
+      let $j=0;
+    }
+    if (!$is_done) {
+      real_sleep 0.1;
+      dec $j;
+    }
+  }
+  dec $i;
+  if ($is_done) {
+    let $i=0;
+  }
+}
+--enable_query_log
+--let $keep_include_silent=0
+if (!$is_done) {
+  --echo Timed out waiting for mysql.gtid_slave_pos* tables to be cleaned up
+}
+
+--disable_query_log
+DELETE FROM t1 WHERE a >= 100;
+DELETE FROM t2 WHERE a >= 100;
+DELETE FROM t3 WHERE a >= 100;
+--enable_query_log
+
+
 # Test status variables Rpl_transactions_multi_engine and Transactions_gtid_foreign_engine.
 # Have mysql.gtid_slave_pos* for myisam and innodb but not tokudb.
 --connection server_2
@@ -223,6 +305,9 @@ SHOW STATUS LIKE "%transactions%engine";
 SET sql_log_bin=0;
 DROP TABLE mysql.gtid_slave_pos_innodb;
 SET sql_log_bin=1;
+--disable_query_log
+eval SET GLOBAL gtid_cleanup_batch_size = $old_gtid_cleanup_batch_size;
+--enable_query_log
 
 --connection server_1
 DROP TABLE t1;