diff --git a/mysql-test/r/dd_is_compatibility_ci.result b/mysql-test/r/dd_is_compatibility_ci.result
index 2988ceb92df4..52a054632068 100644
--- a/mysql-test/r/dd_is_compatibility_ci.result
+++ b/mysql-test/r/dd_is_compatibility_ci.result
@@ -164,6 +164,7 @@ COLUMNS
 COLUMNS_EXTENSIONS
 COLUMN_PRIVILEGES
 COLUMN_STATISTICS
+DATABASE_APPLIED_HLC
 ENABLED_ROLES
 ENGINES
 EVENTS
diff --git a/mysql-test/r/dd_is_compatibility_cs.result b/mysql-test/r/dd_is_compatibility_cs.result
index f62a852a1011..1a08e0639a2f 100644
--- a/mysql-test/r/dd_is_compatibility_cs.result
+++ b/mysql-test/r/dd_is_compatibility_cs.result
@@ -164,6 +164,7 @@ COLUMNS
 COLUMNS_EXTENSIONS
 COLUMN_PRIVILEGES
 COLUMN_STATISTICS
+DATABASE_APPLIED_HLC
 ENABLED_ROLES
 ENGINES
 EVENTS
diff --git a/mysql-test/r/information_schema_ci.result b/mysql-test/r/information_schema_ci.result
index 006d165df0ea..2b14ae742b0d 100644
--- a/mysql-test/r/information_schema_ci.result
+++ b/mysql-test/r/information_schema_ci.result
@@ -72,6 +72,7 @@ COLUMNS
 COLUMNS_EXTENSIONS
 COLUMN_PRIVILEGES
 COLUMN_STATISTICS
+DATABASE_APPLIED_HLC
 ENABLED_ROLES
 ENGINES
 EVENTS
@@ -904,7 +905,7 @@ AND table_name COLLATE utf8_general_ci not like 'innodb_%'
 AND table_name COLLATE utf8_general_ci not like 'rocksdb_%'
 GROUP BY TABLE_SCHEMA;
 TABLE_SCHEMA	count(*)
-information_schema	51
+information_schema	52
 mysql	35
 create table t1 (i int, j int);
 create trigger trg1 before insert on t1 for each row
@@ -2510,6 +2511,7 @@ COLUMNS	TABLE_SCHEMA
 COLUMNS_EXTENSIONS	TABLE_SCHEMA
 COLUMN_PRIVILEGES	TABLE_SCHEMA
 COLUMN_STATISTICS	SCHEMA_NAME
+DATABASE_APPLIED_HLC	DATABASE_NAME
 ENABLED_ROLES	ROLE_NAME
 ENGINES	ENGINE
 EVENTS	EVENT_SCHEMA
@@ -2581,6 +2583,7 @@ COLUMNS	TABLE_SCHEMA
 COLUMNS_EXTENSIONS	TABLE_SCHEMA
 COLUMN_PRIVILEGES	TABLE_SCHEMA
 COLUMN_STATISTICS	SCHEMA_NAME
+DATABASE_APPLIED_HLC	DATABASE_NAME
 ENABLED_ROLES	ROLE_NAME
 ENGINES	ENGINE
 EVENTS	EVENT_SCHEMA
diff --git a/mysql-test/r/information_schema_cs.result b/mysql-test/r/information_schema_cs.result
index 6934ecbb2adc..b37bb0afa487 100644
--- a/mysql-test/r/information_schema_cs.result
+++ b/mysql-test/r/information_schema_cs.result
@@ -72,6 +72,7 @@ COLUMNS
 COLUMNS_EXTENSIONS
 COLUMN_PRIVILEGES
 COLUMN_STATISTICS
+DATABASE_APPLIED_HLC
 ENABLED_ROLES
 ENGINES
 EVENTS
@@ -904,7 +905,7 @@ AND table_name COLLATE utf8_general_ci not like 'innodb_%'
 AND table_name COLLATE utf8_general_ci not like 'rocksdb_%'
 GROUP BY TABLE_SCHEMA;
 TABLE_SCHEMA	count(*)
-information_schema	51
+information_schema	52
 mysql	35
 create table t1 (i int, j int);
 create trigger trg1 before insert on t1 for each row
@@ -2510,6 +2511,7 @@ COLUMNS	TABLE_SCHEMA
 COLUMNS_EXTENSIONS	TABLE_SCHEMA
 COLUMN_PRIVILEGES	TABLE_SCHEMA
 COLUMN_STATISTICS	SCHEMA_NAME
+DATABASE_APPLIED_HLC	DATABASE_NAME
 ENABLED_ROLES	ROLE_NAME
 ENGINES	ENGINE
 EVENTS	EVENT_SCHEMA
@@ -2581,6 +2583,7 @@ COLUMNS	TABLE_SCHEMA
 COLUMNS_EXTENSIONS	TABLE_SCHEMA
 COLUMN_PRIVILEGES	TABLE_SCHEMA
 COLUMN_STATISTICS	SCHEMA_NAME
+DATABASE_APPLIED_HLC	DATABASE_NAME
 ENABLED_ROLES	ROLE_NAME
 ENGINES	ENGINE
 EVENTS	EVENT_SCHEMA
diff --git a/mysql-test/r/mysqld--help-notwin.result b/mysql-test/r/mysqld--help-notwin.result
index 27381b01ea5f..46cdd915241b 100644
--- a/mysql-test/r/mysqld--help-notwin.result
+++ b/mysql-test/r/mysqld--help-notwin.result
@@ -726,6 +726,8 @@ The following options may be given as the first argument:
  If set to 1 table names are stored in lowercase on disk
  and table names will be case-insensitive.  Should be set
  to 2 if you are using a case insensitive file system
+ --maintain-database-hlc 
+ Enable maintaining of max HLC applied per database
  --mandatory-roles=name 
  All the specified roles are always considered granted to
  every user and they can't be revoked. Mandatory roles
@@ -2603,6 +2605,7 @@ log-timestamps UTC
 long-query-time 10
 low-priority-updates FALSE
 lower-case-table-names 1
+maintain-database-hlc FALSE
 mandatory-roles 
 master-info-file master.info
 master-info-repository TABLE
diff --git a/mysql-test/r/mysqld--help-win.result b/mysql-test/r/mysqld--help-win.result
index 5c737e2b0a23..b675748137df 100644
--- a/mysql-test/r/mysqld--help-win.result
+++ b/mysql-test/r/mysqld--help-win.result
@@ -644,6 +644,8 @@ The following options may be given as the first argument:
  If set to 1 table names are stored in lowercase on disk
  and table names will be case-insensitive.  Should be set
  to 2 if you are using a case insensitive file system
+ --maintain-database-hlc 
+ Enable maintaining of max HLC applied per database
  --mandatory-roles=name 
  All the specified roles are always considered granted to
  every user and they can't be revoked. Mandatory roles
@@ -1820,6 +1822,7 @@ log-timestamps UTC
 long-query-time 10
 low-priority-updates FALSE
 lower-case-table-names 1
+maintain-database-hlc FALSE
 mandatory-roles 
 master-info-file master.info
 master-info-repository TABLE
diff --git a/mysql-test/r/mysqlshow_ci.result b/mysql-test/r/mysqlshow_ci.result
index 463dd51ea1d1..7c80df080c62 100644
--- a/mysql-test/r/mysqlshow_ci.result
+++ b/mysql-test/r/mysqlshow_ci.result
@@ -90,6 +90,7 @@ Database: information_schema
 | COLUMN_STATISTICS                     |
 | COLUMNS                               |
 | COLUMNS_EXTENSIONS                    |
+| DATABASE_APPLIED_HLC                  |
 | ENABLED_ROLES                         |
 | ENGINES                               |
 | EVENTS                                |
@@ -192,6 +193,7 @@ Database: INFORMATION_SCHEMA
 | COLUMN_STATISTICS                     |
 | COLUMNS                               |
 | COLUMNS_EXTENSIONS                    |
+| DATABASE_APPLIED_HLC                  |
 | ENABLED_ROLES                         |
 | ENGINES                               |
 | EVENTS                                |
diff --git a/mysql-test/r/mysqlshow_cs.result b/mysql-test/r/mysqlshow_cs.result
index 4c12fd6f853e..a4605c8906a6 100644
--- a/mysql-test/r/mysqlshow_cs.result
+++ b/mysql-test/r/mysqlshow_cs.result
@@ -90,6 +90,7 @@ Database: information_schema
 | COLUMNS_EXTENSIONS                    |
 | COLUMN_PRIVILEGES                     |
 | COLUMN_STATISTICS                     |
+| DATABASE_APPLIED_HLC                  |
 | ENABLED_ROLES                         |
 | ENGINES                               |
 | EVENTS                                |
@@ -193,6 +194,7 @@ Database: INFORMATION_SCHEMA
 | COLUMNS_EXTENSIONS                    |
 | COLUMN_PRIVILEGES                     |
 | COLUMN_STATISTICS                     |
+| DATABASE_APPLIED_HLC                  |
 | ENABLED_ROLES                         |
 | ENGINES                               |
 | EVENTS                                |
diff --git a/mysql-test/suite/funcs_1/r/is_columns_is_ci.result b/mysql-test/suite/funcs_1/r/is_columns_is_ci.result
index d2806bb2f07b..f3d1862d4be9 100644
--- a/mysql-test/suite/funcs_1/r/is_columns_is_ci.result
+++ b/mysql-test/suite/funcs_1/r/is_columns_is_ci.result
@@ -80,6 +80,8 @@ def	information_schema	COLUMN_STATISTICS	SCHEMA_NAME	1	NULL	NO	varchar	64	192	NU
 def	information_schema	COLUMN_STATISTICS	TABLE_NAME	2	NULL	NO	varchar	64	192	NULL	NULL	NULL	utf8mb3	utf8_bin	varchar(64)			select			NULL
 def	information_schema	COLUMN_STATISTICS	COLUMN_NAME	3	NULL	NO	varchar	64	192	NULL	NULL	NULL	utf8mb3	utf8_bin	varchar(64)			select			NULL
 def	information_schema	COLUMN_STATISTICS	HISTOGRAM	4	NULL	NO	json	NULL	NULL	NULL	NULL	NULL	NULL	NULL	json			select			NULL
+def	information_schema	DATABASE_APPLIED_HLC	DATABASE_NAME	1		NO	varchar	64	192	NULL	NULL	NULL	utf8	utf8_general_ci	varchar(192)			select			NULL
+def	information_schema	DATABASE_APPLIED_HLC	APPLIED_HLC	2		NO	bigint	NULL	NULL	NULL	NULL	NULL	NULL	NULL	bigint unsigned			select			NULL
 def	information_schema	ENABLED_ROLES	ROLE_NAME	1	NULL	YES	varchar	255	1020	NULL	NULL	NULL	utf8mb4	utf8mb4_0900_ai_ci	varchar(255)			select			NULL
 def	information_schema	ENABLED_ROLES	ROLE_HOST	2	NULL	YES	varchar	255	1020	NULL	NULL	NULL	utf8mb4	utf8mb4_0900_ai_ci	varchar(255)			select			NULL
 def	information_schema	ENABLED_ROLES	IS_DEFAULT	3	NULL	YES	varchar	3	9	NULL	NULL	NULL	utf8mb3	utf8mb3_general_ci	varchar(3)			select			NULL
@@ -736,6 +738,8 @@ NULL	information_schema	COLUMNS_EXTENSIONS	SECONDARY_ENGINE_ATTRIBUTE	json	NULL
 3.0000	information_schema	COLUMN_STATISTICS	TABLE_NAME	varchar	64	192	utf8mb3	utf8_bin	varchar(64)
 3.0000	information_schema	COLUMN_STATISTICS	COLUMN_NAME	varchar	64	192	utf8mb3	utf8_bin	varchar(64)
 NULL	information_schema	COLUMN_STATISTICS	HISTOGRAM	json	NULL	NULL	NULL	NULL	json
+3.0000	information_schema	DATABASE_APPLIED_HLC	DATABASE_NAME	varchar	64	192	utf8	utf8_general_ci	varchar(192)
+NULL	information_schema	DATABASE_APPLIED_HLC	APPLIED_HLC	bigint	NULL	NULL	NULL	NULL	bigint unsigned
 4.0000	information_schema	ENABLED_ROLES	ROLE_NAME	varchar	255	1020	utf8mb4	utf8mb4_0900_ai_ci	varchar(255)
 4.0000	information_schema	ENABLED_ROLES	ROLE_HOST	varchar	255	1020	utf8mb4	utf8mb4_0900_ai_ci	varchar(255)
 3.0000	information_schema	ENABLED_ROLES	IS_DEFAULT	varchar	3	9	utf8mb3	utf8mb3_general_ci	varchar(3)
diff --git a/mysql-test/suite/funcs_1/r/is_columns_is_cs.result b/mysql-test/suite/funcs_1/r/is_columns_is_cs.result
index c1cd60f6e550..794f5b853682 100644
--- a/mysql-test/suite/funcs_1/r/is_columns_is_cs.result
+++ b/mysql-test/suite/funcs_1/r/is_columns_is_cs.result
@@ -85,6 +85,8 @@ def	information_schema	COLUMN_STATISTICS	SCHEMA_NAME	1	NULL	NO	varchar	64	192	NU
 def	information_schema	COLUMN_STATISTICS	TABLE_NAME	2	NULL	NO	varchar	64	192	NULL	NULL	NULL	utf8mb3	utf8mb3_bin	varchar(64)			select			NULL
 def	information_schema	COLUMN_STATISTICS	COLUMN_NAME	3	NULL	NO	varchar	64	192	NULL	NULL	NULL	utf8mb3	utf8_bin	varchar(64)			select			NULL
 def	information_schema	COLUMN_STATISTICS	HISTOGRAM	4	NULL	NO	json	NULL	NULL	NULL	NULL	NULL	NULL	NULL	json			select			NULL
+def	information_schema	DATABASE_APPLIED_HLC	DATABASE_NAME	1		NO	varchar	64	192	NULL	NULL	NULL	utf8mb3	utf8mb3_general_ci	varchar(192)			select			NULL
+def	information_schema	DATABASE_APPLIED_HLC	APPLIED_HLC	2		NO	bigint	NULL	NULL	NULL	NULL	NULL	NULL	NULL	bigint unsigned			select			NULL
 def	information_schema	ENABLED_ROLES	ROLE_NAME	1	NULL	YES	varchar	255	1020	NULL	NULL	NULL	utf8mb4	utf8mb4_0900_ai_ci	varchar(255)			select			NULL
 def	information_schema	ENABLED_ROLES	ROLE_HOST	2	NULL	YES	varchar	255	1020	NULL	NULL	NULL	utf8mb4	utf8mb4_0900_ai_ci	varchar(255)			select			NULL
 def	information_schema	ENABLED_ROLES	IS_DEFAULT	3	NULL	YES	varchar	3	9	NULL	NULL	NULL	utf8mb3	utf8mb3_general_ci	varchar(3)			select			NULL
@@ -759,6 +761,8 @@ NULL	information_schema	COLUMNS_EXTENSIONS	SECONDARY_ENGINE_ATTRIBUTE	json	NULL
 3.0000	information_schema	COLUMN_STATISTICS	TABLE_NAME	varchar	64	192	utf8mb3	utf8mb3_bin	varchar(64)
 3.0000	information_schema	COLUMN_STATISTICS	COLUMN_NAME	varchar	64	192	utf8mb3	utf8_bin	varchar(64)
 NULL	information_schema	COLUMN_STATISTICS	HISTOGRAM	json	NULL	NULL	NULL	NULL	json
+3.0000	information_schema	DATABASE_APPLIED_HLC	DATABASE_NAME	varchar	64	192	utf8mb3	utf8mb3_general_ci	varchar(192)
+NULL	information_schema	DATABASE_APPLIED_HLC	APPLIED_HLC	bigint	NULL	NULL	NULL	NULL	bigint unsigned
 4.0000	information_schema	ENABLED_ROLES	ROLE_NAME	varchar	255	1020	utf8mb4	utf8mb4_0900_ai_ci	varchar(255)
 4.0000	information_schema	ENABLED_ROLES	ROLE_HOST	varchar	255	1020	utf8mb4	utf8mb4_0900_ai_ci	varchar(255)
 3.0000	information_schema	ENABLED_ROLES	IS_DEFAULT	varchar	3	9	utf8mb3	utf8mb3_general_ci	varchar(3)
diff --git a/mysql-test/suite/funcs_1/r/is_tables_is.result b/mysql-test/suite/funcs_1/r/is_tables_is.result
index 0b44e52e3c3e..9b1c0b4e389c 100644
--- a/mysql-test/suite/funcs_1/r/is_tables_is.result
+++ b/mysql-test/suite/funcs_1/r/is_tables_is.result
@@ -271,6 +271,29 @@ user_comment
 Separator	-----------------------------------------------------
 TABLE_CATALOG	def
 TABLE_SCHEMA	information_schema
+TABLE_NAME	DATABASE_APPLIED_HLC
+TABLE_TYPE	SYSTEM VIEW
+ENGINE	NULL
+VERSION	10
+ROW_FORMAT	NULL
+TABLE_ROWS	#TBLR#
+AVG_ROW_LENGTH	#ARL#
+DATA_LENGTH	#DL#
+MAX_DATA_LENGTH	#MDL#
+INDEX_LENGTH	#IL#
+DATA_FREE	#DF#
+AUTO_INCREMENT	#AI#
+CREATE_TIME	#CRT#
+UPDATE_TIME	#UT#
+CHECK_TIME	#CT#
+TABLE_COLLATION	NULL
+CHECKSUM	NULL
+CREATE_OPTIONS	#CO#
+TABLE_COMMENT	#TC#
+user_comment	
+Separator	-----------------------------------------------------
+TABLE_CATALOG	def
+TABLE_SCHEMA	information_schema
 TABLE_NAME	ENABLED_ROLES
 TABLE_TYPE	SYSTEM VIEW
 ENGINE	NULL
@@ -1441,6 +1464,29 @@ user_comment
 Separator	-----------------------------------------------------
 TABLE_CATALOG	def
 TABLE_SCHEMA	information_schema
+TABLE_NAME	DATABASE_APPLIED_HLC
+TABLE_TYPE	SYSTEM VIEW
+ENGINE	NULL
+VERSION	10
+ROW_FORMAT	NULL
+TABLE_ROWS	#TBLR#
+AVG_ROW_LENGTH	#ARL#
+DATA_LENGTH	#DL#
+MAX_DATA_LENGTH	#MDL#
+INDEX_LENGTH	#IL#
+DATA_FREE	#DF#
+AUTO_INCREMENT	#AI#
+CREATE_TIME	#CRT#
+UPDATE_TIME	#UT#
+CHECK_TIME	#CT#
+TABLE_COLLATION	NULL
+CHECKSUM	NULL
+CREATE_OPTIONS	#CO#
+TABLE_COMMENT	#TC#
+user_comment	
+Separator	-----------------------------------------------------
+TABLE_CATALOG	def
+TABLE_SCHEMA	information_schema
 TABLE_NAME	ENABLED_ROLES
 TABLE_TYPE	SYSTEM VIEW
 ENGINE	NULL
diff --git a/mysql-test/suite/information_schema/r/information_schema_db.result b/mysql-test/suite/information_schema/r/information_schema_db.result
index afdc4bdb5e3a..8bbc791580e1 100644
--- a/mysql-test/suite/information_schema/r/information_schema_db.result
+++ b/mysql-test/suite/information_schema/r/information_schema_db.result
@@ -17,6 +17,7 @@ COLUMNS
 COLUMNS_EXTENSIONS
 COLUMN_PRIVILEGES
 COLUMN_STATISTICS
+DATABASE_APPLIED_HLC
 ENABLED_ROLES
 ENGINES
 EVENTS
diff --git a/mysql-test/suite/rpl_gtid/r/rpl_maintain_database_hlc_rbr.result b/mysql-test/suite/rpl_gtid/r/rpl_maintain_database_hlc_rbr.result
new file mode 100644
index 000000000000..001b79bb464f
--- /dev/null
+++ b/mysql-test/suite/rpl_gtid/r/rpl_maintain_database_hlc_rbr.result
@@ -0,0 +1,431 @@
+include/master-slave.inc
+Warnings:
+Note	####	Sending passwords in plain text without SSL/TLS is extremely insecure.
+Note	####	Storing MySQL user name or password information in the connection metadata repository is not secure and is therefore not recommended. Please consider using the USER and PASSWORD connection options for START REPLICA; see the 'START REPLICA Syntax' in the MySQL Manual for more information.
+[connection master]
+[connection master]
+FLUSH LOGS;
+purge binary logs to 'binlog';
+[connection slave]
+FLUSH LOGS;
+purge binary logs to 'binlog';
+[connection master]
+call mtr.add_suppression("Databases were empty for this trx");
+CREATE DATABASE test1;
+CREATE DATABASE test2;
+CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB;
+CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB;
+include/sync_slave_sql_with_master.inc
+[connection master]
+SET SESSION DEBUG = "+d,allow_long_hlc_drift_for_tests";
+SET @@global.minimum_hlc_ns = 2538630000000000000;
+[connection master]
+SET @@global.enable_binlog_hlc = TRUE;
+SET @@global.maintain_database_hlc = TRUE;
+[connection slave]
+SET @@global.enable_binlog_hlc = TRUE;
+SET @@global.maintain_database_hlc = TRUE;
+Case 1: Single statement txn
+[connection master]
+USE test1;
+INSERT INTO t1 VALUES(1, 'a');
+INSERT INTO t1 VALUES(2, 'b');
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (mtr.test_suppressions)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Query	#	#	CREATE DATABASE test1
+master-bin.000002	#	Query	#	#	CREATE DATABASE test2
+master-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+master-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000001
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000002
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+SELECT * FROM test1.t1;
+a	b
+1	a
+2	b
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000002
+include/sync_slave_sql_with_master.inc
+[connection slave]
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (mtr.test_suppressions)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Query	#	#	CREATE DATABASE test1
+slave-bin.000002	#	Query	#	#	CREATE DATABASE test2
+slave-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+slave-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000001
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000002
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+SELECT * FROM test1.t1;
+a	b
+1	a
+2	b
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000002
+Case 2: Multi statement txn
+[connection master]
+SET autocommit = OFF;
+USE test1;
+INSERT INTO t1 VALUES(3, 'c');
+INSERT INTO t1 VALUES(4, 'd');
+COMMIT;
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (mtr.test_suppressions)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Query	#	#	CREATE DATABASE test1
+master-bin.000002	#	Query	#	#	CREATE DATABASE test2
+master-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+master-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000001
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000002
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000003
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+SELECT * FROM test1.t1;
+a	b
+1	a
+2	b
+3	c
+4	d
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000003
+include/sync_slave_sql_with_master.inc
+[connection slave]
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (mtr.test_suppressions)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Query	#	#	CREATE DATABASE test1
+slave-bin.000002	#	Query	#	#	CREATE DATABASE test2
+slave-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+slave-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000001
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000002
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000003
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+SELECT * FROM test1.t1;
+a	b
+1	a
+2	b
+3	c
+4	d
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000003
+Case 3: Multi statement txn spanning multiple database
+[connection master]
+SET autocommit = OFF;
+INSERT INTO test1.t1 VALUES(5, 'e');
+INSERT INTO test2.t1 VALUES(5, 'e');
+COMMIT;
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (mtr.test_suppressions)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Query	#	#	CREATE DATABASE test1
+master-bin.000002	#	Query	#	#	CREATE DATABASE test2
+master-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+master-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000001
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000002
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000003
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000004
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Table_map	#	#	table_id: # (test2.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+SELECT * FROM test1.t1;
+a	b
+1	a
+2	b
+3	c
+4	d
+5	e
+SELECT * FROM test2.t1;
+a	b
+5	e
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000004
+test2	2538630000000000004
+include/sync_slave_sql_with_master.inc
+[connection slave]
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (mtr.test_suppressions)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Query	#	#	CREATE DATABASE test1
+slave-bin.000002	#	Query	#	#	CREATE DATABASE test2
+slave-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+slave-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000001
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000002
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000003
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000004
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Table_map	#	#	table_id: # (test2.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+SELECT * FROM test1.t1;
+a	b
+1	a
+2	b
+3	c
+4	d
+5	e
+SELECT * FROM test2.t1;
+a	b
+5	e
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000004
+test2	2538630000000000004
+Case 4: Multiple single stmt txns txn spanning multiple database
+[connection master]
+SET autocommit = OFF;
+INSERT INTO test1.t1 VALUES(6, 'f');
+COMMIT;
+INSERT INTO test2.t1 VALUES(8, 'g');
+COMMIT;
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (mtr.test_suppressions)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Query	#	#	CREATE DATABASE test1
+master-bin.000002	#	Query	#	#	CREATE DATABASE test2
+master-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+master-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000001
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000002
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000003
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000004
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Table_map	#	#	table_id: # (test2.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000005
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000006
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Table_map	#	#	table_id: # (test2.t1)
+master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+SELECT * FROM test1.t1;
+a	b
+1	a
+2	b
+3	c
+4	d
+5	e
+6	f
+SELECT * FROM test2.t1;
+a	b
+5	e
+8	g
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000005
+test2	2538630000000000006
+include/sync_slave_sql_with_master.inc
+[connection slave]
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (mtr.test_suppressions)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Query	#	#	CREATE DATABASE test1
+slave-bin.000002	#	Query	#	#	CREATE DATABASE test2
+slave-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+slave-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000001
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000002
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000003
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000004
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Table_map	#	#	table_id: # (test2.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000005
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (test1.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000006
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Table_map	#	#	table_id: # (test2.t1)
+slave-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+SELECT * FROM test1.t1;
+a	b
+1	a
+2	b
+3	c
+4	d
+5	e
+6	f
+SELECT * FROM test2.t1;
+a	b
+5	e
+8	g
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000005
+test2	2538630000000000006
+Case 5: Pure DDL should also update and maintain per database HLC
+[connection master]
+SET autocommit = OFF;
+CREATE TABLE test1.t2(a INT);
+CREATE TABLE test2.t2(a INT);
+CREATE DATABASE test3;
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000007
+test2	2538630000000000008
+test3	2538630000000000009
+include/sync_slave_sql_with_master.inc
+[connection slave]
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000007
+test2	2538630000000000008
+test3	2538630000000000009
+[connection master]
+DROP TABLE test1.t1;
+DROP TABLE test2.t1;
+DROP TABLE test1.t2;
+DROP TABLE test2.t2;
+DROP DATABASE test1;
+DROP DATABASE test2;
+DROP DATABASE test3;
+include/sync_slave_sql_with_master.inc
+[connection slave]
+include/rpl_end.inc
diff --git a/mysql-test/suite/rpl_gtid/r/rpl_maintain_database_hlc_sbr.result b/mysql-test/suite/rpl_gtid/r/rpl_maintain_database_hlc_sbr.result
new file mode 100644
index 000000000000..4eda35ed58a2
--- /dev/null
+++ b/mysql-test/suite/rpl_gtid/r/rpl_maintain_database_hlc_sbr.result
@@ -0,0 +1,383 @@
+include/master-slave.inc
+Warnings:
+Note	####	Sending passwords in plain text without SSL/TLS is extremely insecure.
+Note	####	Storing MySQL user name or password information in the connection metadata repository is not secure and is therefore not recommended. Please consider using the USER and PASSWORD connection options for START REPLICA; see the 'START REPLICA Syntax' in the MySQL Manual for more information.
+[connection master]
+[connection master]
+FLUSH LOGS;
+purge binary logs to 'binlog';
+[connection slave]
+FLUSH LOGS;
+purge binary logs to 'binlog';
+[connection master]
+call mtr.add_suppression("Databases were empty for this trx");
+CREATE DATABASE test1;
+CREATE DATABASE test2;
+CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB;
+CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB;
+include/sync_slave_sql_with_master.inc
+[connection master]
+SET SESSION DEBUG = "+d,allow_long_hlc_drift_for_tests";
+SET @@global.minimum_hlc_ns = 2538630000000000000;
+[connection master]
+SET @@global.enable_binlog_hlc = TRUE;
+SET @@global.maintain_database_hlc = TRUE;
+[connection slave]
+SET @@global.enable_binlog_hlc = TRUE;
+SET @@global.maintain_database_hlc = TRUE;
+Case 1: Single statement txn
+[connection master]
+USE test1;
+INSERT INTO t1 VALUES(1, 'a');
+INSERT INTO t1 VALUES(2, 'b');
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `mtr`; INSERT INTO test_suppressions (pattern) VALUES ( NAME_CONST('pattern',_utf8mb4'Databases were empty for this trx' COLLATE 'utf8mb4_0900_ai_ci'))
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Query	#	#	CREATE DATABASE test1
+master-bin.000002	#	Query	#	#	CREATE DATABASE test2
+master-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+master-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000001
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(1, 'a')
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000002
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(2, 'b')
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+SELECT * FROM test1.t1;
+a	b
+1	a
+2	b
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000002
+include/sync_slave_sql_with_master.inc
+[connection slave]
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `mtr`; INSERT INTO test_suppressions (pattern) VALUES ( NAME_CONST('pattern',_utf8mb4'Databases were empty for this trx' COLLATE 'utf8mb4_0900_ai_ci'))
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Query	#	#	CREATE DATABASE test1
+slave-bin.000002	#	Query	#	#	CREATE DATABASE test2
+slave-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+slave-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000001
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(1, 'a')
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000002
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(2, 'b')
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+SELECT * FROM test1.t1;
+a	b
+1	a
+2	b
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000002
+Case 2: Multi statement txn
+[connection master]
+SET autocommit = OFF;
+USE test1;
+INSERT INTO t1 VALUES(3, 'c');
+INSERT INTO t1 VALUES(4, 'd');
+COMMIT;
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `mtr`; INSERT INTO test_suppressions (pattern) VALUES ( NAME_CONST('pattern',_utf8mb4'Databases were empty for this trx' COLLATE 'utf8mb4_0900_ai_ci'))
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Query	#	#	CREATE DATABASE test1
+master-bin.000002	#	Query	#	#	CREATE DATABASE test2
+master-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+master-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000001
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(1, 'a')
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000002
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(2, 'b')
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000003
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(3, 'c')
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(4, 'd')
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+SELECT * FROM test1.t1;
+a	b
+1	a
+2	b
+3	c
+4	d
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000003
+include/sync_slave_sql_with_master.inc
+[connection slave]
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `mtr`; INSERT INTO test_suppressions (pattern) VALUES ( NAME_CONST('pattern',_utf8mb4'Databases were empty for this trx' COLLATE 'utf8mb4_0900_ai_ci'))
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Query	#	#	CREATE DATABASE test1
+slave-bin.000002	#	Query	#	#	CREATE DATABASE test2
+slave-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+slave-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000001
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(1, 'a')
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000002
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(2, 'b')
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000003
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(3, 'c')
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(4, 'd')
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+SELECT * FROM test1.t1;
+a	b
+1	a
+2	b
+3	c
+4	d
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000003
+Case 3: Multi statement txn spanning multiple database
+[connection master]
+SET autocommit = OFF;
+INSERT INTO test1.t1 VALUES(5, 'e');
+INSERT INTO test2.t1 VALUES(5, 'e');
+COMMIT;
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `mtr`; INSERT INTO test_suppressions (pattern) VALUES ( NAME_CONST('pattern',_utf8mb4'Databases were empty for this trx' COLLATE 'utf8mb4_0900_ai_ci'))
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Query	#	#	CREATE DATABASE test1
+master-bin.000002	#	Query	#	#	CREATE DATABASE test2
+master-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+master-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000001
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(1, 'a')
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000002
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(2, 'b')
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000003
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(3, 'c')
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(4, 'd')
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000004
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO test1.t1 VALUES(5, 'e')
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO test2.t1 VALUES(5, 'e')
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+SELECT * FROM test1.t1;
+a	b
+1	a
+2	b
+3	c
+4	d
+5	e
+SELECT * FROM test2.t1;
+a	b
+5	e
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000004
+test2	2538630000000000004
+include/sync_slave_sql_with_master.inc
+[connection slave]
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `mtr`; INSERT INTO test_suppressions (pattern) VALUES ( NAME_CONST('pattern',_utf8mb4'Databases were empty for this trx' COLLATE 'utf8mb4_0900_ai_ci'))
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Query	#	#	CREATE DATABASE test1
+slave-bin.000002	#	Query	#	#	CREATE DATABASE test2
+slave-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+slave-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000001
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(1, 'a')
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000002
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(2, 'b')
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000003
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(3, 'c')
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(4, 'd')
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000004
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO test1.t1 VALUES(5, 'e')
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO test2.t1 VALUES(5, 'e')
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+SELECT * FROM test1.t1;
+a	b
+1	a
+2	b
+3	c
+4	d
+5	e
+SELECT * FROM test2.t1;
+a	b
+5	e
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000004
+test2	2538630000000000004
+Case 4: Multiple single stmt txns txn spanning multiple database
+[connection master]
+SET autocommit = OFF;
+INSERT INTO test1.t1 VALUES(6, 'f');
+COMMIT;
+INSERT INTO test2.t1 VALUES(8, 'g');
+COMMIT;
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `mtr`; INSERT INTO test_suppressions (pattern) VALUES ( NAME_CONST('pattern',_utf8mb4'Databases were empty for this trx' COLLATE 'utf8mb4_0900_ai_ci'))
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Query	#	#	CREATE DATABASE test1
+master-bin.000002	#	Query	#	#	CREATE DATABASE test2
+master-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+master-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000001
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(1, 'a')
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000002
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(2, 'b')
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000003
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(3, 'c')
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(4, 'd')
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000004
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO test1.t1 VALUES(5, 'e')
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO test2.t1 VALUES(5, 'e')
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000005
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO test1.t1 VALUES(6, 'f')
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+master-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000006
+master-bin.000002	#	Query	#	#	BEGIN
+master-bin.000002	#	Query	#	#	use `test1`; INSERT INTO test2.t1 VALUES(8, 'g')
+master-bin.000002	#	Xid	#	#	COMMIT /* XID */
+SELECT * FROM test1.t1;
+a	b
+1	a
+2	b
+3	c
+4	d
+5	e
+6	f
+SELECT * FROM test2.t1;
+a	b
+5	e
+8	g
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000005
+test2	2538630000000000006
+include/sync_slave_sql_with_master.inc
+[connection slave]
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `mtr`; INSERT INTO test_suppressions (pattern) VALUES ( NAME_CONST('pattern',_utf8mb4'Databases were empty for this trx' COLLATE 'utf8mb4_0900_ai_ci'))
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Query	#	#	CREATE DATABASE test1
+slave-bin.000002	#	Query	#	#	CREATE DATABASE test2
+slave-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+slave-bin.000002	#	Query	#	#	use `test`; CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000001
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(1, 'a')
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000002
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(2, 'b')
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000003
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(3, 'c')
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO t1 VALUES(4, 'd')
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000004
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO test1.t1 VALUES(5, 'e')
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO test2.t1 VALUES(5, 'e')
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000005
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO test1.t1 VALUES(6, 'f')
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+slave-bin.000002	#	Metadata	#	#	HLC time: 2538630000000000006
+slave-bin.000002	#	Query	#	#	BEGIN
+slave-bin.000002	#	Query	#	#	use `test1`; INSERT INTO test2.t1 VALUES(8, 'g')
+slave-bin.000002	#	Xid	#	#	COMMIT /* XID */
+SELECT * FROM test1.t1;
+a	b
+1	a
+2	b
+3	c
+4	d
+5	e
+6	f
+SELECT * FROM test2.t1;
+a	b
+5	e
+8	g
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000005
+test2	2538630000000000006
+Case 5: Pure DDL should also update and maintain per database HLC
+[connection master]
+SET autocommit = OFF;
+CREATE TABLE test1.t2(a INT);
+CREATE TABLE test2.t2(a INT);
+CREATE DATABASE test3;
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000007
+test2	2538630000000000008
+test3	2538630000000000009
+include/sync_slave_sql_with_master.inc
+[connection slave]
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+DATABASE_NAME	APPLIED_HLC
+test1	2538630000000000007
+test2	2538630000000000008
+test3	2538630000000000009
+[connection master]
+DROP TABLE test1.t1;
+DROP TABLE test2.t1;
+DROP TABLE test1.t2;
+DROP TABLE test2.t2;
+DROP DATABASE test1;
+DROP DATABASE test2;
+DROP DATABASE test3;
+include/sync_slave_sql_with_master.inc
+[connection slave]
+include/rpl_end.inc
diff --git a/mysql-test/suite/rpl_gtid/r/rpl_stress_hlc.result b/mysql-test/suite/rpl_gtid/r/rpl_stress_hlc.result
index 13894d6e38fc..1b2d551361b3 100644
--- a/mysql-test/suite/rpl_gtid/r/rpl_stress_hlc.result
+++ b/mysql-test/suite/rpl_gtid/r/rpl_stress_hlc.result
@@ -7,4 +7,7 @@ include/sync_slave_sql_with_master.inc
 # Build connections to master server
 # Stress test that execute massive queries on every connection
 include/sync_slave_sql_with_master.inc
+[connection master]
+[connection slave]
+include/assert.inc [Master and Slave should have the same applied hlc on the test database]
 include/rpl_end.inc
diff --git a/mysql-test/suite/rpl_gtid/t/rpl_maintain_database_hlc.inc b/mysql-test/suite/rpl_gtid/t/rpl_maintain_database_hlc.inc
new file mode 100644
index 000000000000..33f4f1ae9e2a
--- /dev/null
+++ b/mysql-test/suite/rpl_gtid/t/rpl_maintain_database_hlc.inc
@@ -0,0 +1,142 @@
+# Setup
+--source include/rpl_connection_master.inc
+call mtr.add_suppression("Databases were empty for this trx");
+CREATE DATABASE test1;
+CREATE DATABASE test2;
+
+CREATE TABLE test1.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB;
+CREATE TABLE test2.t1(a INT PRIMARY KEY, b CHAR(8)) ENGINE=InnoDB;
+--source include/sync_slave_sql_with_master.inc
+
+# Set minimum_hlc_ns to a high value. Subsequent txn's should see monotonically
+# increasing timestamp from this point
+--source include/rpl_connection_master.inc
+SET SESSION DEBUG = "+d,allow_long_hlc_drift_for_tests";
+--let $saved_minimum_hlc_ns = `SELECT @@global.minimum_hlc_ns`
+SET @@global.minimum_hlc_ns = 2538630000000000000; # ~2050 AD
+
+# Enable binlog_hlc and maintain_database_hlc in both master and slave
+--source include/rpl_connection_master.inc
+--let $saved_master_enable_binlog_hlc = `SELECT @@global.enable_binlog_hlc`
+--let $saved_master_maintain_database_hlc = `SELECT @@global.maintain_database_hlc`
+SET @@global.enable_binlog_hlc = TRUE;
+SET @@global.maintain_database_hlc = TRUE;
+--source include/rpl_connection_slave.inc
+--let $saved_slave_enable_binlog_hlc= `SELECT @@global.enable_binlog_hlc`
+--let $saved_slave_maintain_database_hlc = `SELECT @@global.maintain_database_hlc`
+SET @@global.enable_binlog_hlc = TRUE;
+SET @@global.maintain_database_hlc = TRUE;
+
+--echo Case 1: Single statement txn
+--source include/rpl_connection_master.inc
+USE test1;
+INSERT INTO t1 VALUES(1, 'a');
+INSERT INTO t1 VALUES(2, 'b');
+--source include/show_binlog_events.inc
+SELECT * FROM test1.t1;
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+
+--source include/sync_slave_sql_with_master.inc
+--source include/rpl_connection_slave.inc
+--source include/show_binlog_events.inc
+SELECT * FROM test1.t1;
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+
+--echo Case 2: Multi statement txn
+--source include/rpl_connection_master.inc
+SET autocommit = OFF;
+
+USE test1;
+INSERT INTO t1 VALUES(3, 'c');
+INSERT INTO t1 VALUES(4, 'd');
+COMMIT;
+--source include/show_binlog_events.inc
+SELECT * FROM test1.t1;
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+
+--source include/sync_slave_sql_with_master.inc
+--source include/rpl_connection_slave.inc
+--source include/show_binlog_events.inc
+SELECT * FROM test1.t1;
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+
+--echo Case 3: Multi statement txn spanning multiple database
+--source include/rpl_connection_master.inc
+SET autocommit = OFF;
+
+INSERT INTO test1.t1 VALUES(5, 'e');
+INSERT INTO test2.t1 VALUES(5, 'e');
+COMMIT;
+--source include/show_binlog_events.inc
+SELECT * FROM test1.t1;
+SELECT * FROM test2.t1;
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+
+--source include/sync_slave_sql_with_master.inc
+--source include/rpl_connection_slave.inc
+--source include/show_binlog_events.inc
+SELECT * FROM test1.t1;
+SELECT * FROM test2.t1;
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+
+--echo Case 4: Multiple single stmt txns txn spanning multiple database
+--source include/rpl_connection_master.inc
+SET autocommit = OFF;
+
+INSERT INTO test1.t1 VALUES(6, 'f');
+COMMIT;
+
+INSERT INTO test2.t1 VALUES(8, 'g');
+COMMIT;
+
+--source include/show_binlog_events.inc
+SELECT * FROM test1.t1;
+SELECT * FROM test2.t1;
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+
+--source include/sync_slave_sql_with_master.inc
+--source include/rpl_connection_slave.inc
+--source include/show_binlog_events.inc
+SELECT * FROM test1.t1;
+SELECT * FROM test2.t1;
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+
+
+--echo Case 5: Pure DDL should also update and maintain per database HLC
+--source include/rpl_connection_master.inc
+SET autocommit = OFF;
+
+CREATE TABLE test1.t2(a INT);
+CREATE TABLE test2.t2(a INT);
+CREATE DATABASE test3;
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+
+--source include/sync_slave_sql_with_master.inc
+--source include/rpl_connection_slave.inc
+SELECT * FROM information_schema.database_applied_hlc ORDER BY 2, 1;
+
+# Cleanup
+--source include/rpl_connection_master.inc
+--disable_query_log
+--eval SET SESSION DEBUG = "-d,allow_long_hlc_drift_for_tests"
+--eval SET GLOBAL minimum_hlc_ns = $saved_minimum_hlc_ns
+--eval SET GLOBAL enable_binlog_hlc = $saved_master_enable_binlog_hlc
+--eval SET GLOBAL maintain_database_hlc = $saved_master_maintain_database_hlc
+--enable_query_log
+
+DROP TABLE test1.t1;
+DROP TABLE test2.t1;
+DROP TABLE test1.t2;
+DROP TABLE test2.t2;
+DROP DATABASE test1;
+DROP DATABASE test2;
+DROP DATABASE test3;
+
+--source include/sync_slave_sql_with_master.inc
+--source include/rpl_connection_slave.inc
+--disable_query_log
+--eval SET GLOBAL enable_binlog_hlc = $saved_slave_enable_binlog_hlc
+--eval SET GLOBAL maintain_database_hlc = $saved_slave_maintain_database_hlc
+--enable_query_log
+
+--source include/rpl_end.inc
diff --git a/mysql-test/suite/rpl_gtid/t/rpl_maintain_database_hlc_rbr-master.opt b/mysql-test/suite/rpl_gtid/t/rpl_maintain_database_hlc_rbr-master.opt
new file mode 100644
index 000000000000..cef79bc8585a
--- /dev/null
+++ b/mysql-test/suite/rpl_gtid/t/rpl_maintain_database_hlc_rbr-master.opt
@@ -0,0 +1 @@
+--force-restart
diff --git a/mysql-test/suite/rpl_gtid/t/rpl_maintain_database_hlc_rbr.test b/mysql-test/suite/rpl_gtid/t/rpl_maintain_database_hlc_rbr.test
new file mode 100644
index 000000000000..8f71ab8df5d1
--- /dev/null
+++ b/mysql-test/suite/rpl_gtid/t/rpl_maintain_database_hlc_rbr.test
@@ -0,0 +1,18 @@
+--source include/have_binlog_format_row.inc
+--source include/have_debug.inc
+--source include/master-slave.inc
+
+# Cleanup old binlog
+--source include/rpl_connection_master.inc
+FLUSH LOGS;
+--let $binlog= query_get_value(SHOW MASTER STATUS, File, 1)
+--replace_result $binlog binlog
+--eval purge binary logs to '$binlog'
+
+--source include/rpl_connection_slave.inc
+FLUSH LOGS;
+--let $binlog= query_get_value(SHOW MASTER STATUS, File, 1)
+--replace_result $binlog binlog
+--eval purge binary logs to '$binlog'
+
+--source suite/rpl_gtid/t/rpl_maintain_database_hlc.inc
diff --git a/mysql-test/suite/rpl_gtid/t/rpl_maintain_database_hlc_sbr-master.opt b/mysql-test/suite/rpl_gtid/t/rpl_maintain_database_hlc_sbr-master.opt
new file mode 100644
index 000000000000..5fdca8857481
--- /dev/null
+++ b/mysql-test/suite/rpl_gtid/t/rpl_maintain_database_hlc_sbr-master.opt
@@ -0,0 +1 @@
+--force-restart
\ No newline at end of file
diff --git a/mysql-test/suite/rpl_gtid/t/rpl_maintain_database_hlc_sbr.test b/mysql-test/suite/rpl_gtid/t/rpl_maintain_database_hlc_sbr.test
new file mode 100644
index 000000000000..b25d135d2381
--- /dev/null
+++ b/mysql-test/suite/rpl_gtid/t/rpl_maintain_database_hlc_sbr.test
@@ -0,0 +1,18 @@
+--source include/have_binlog_format_mixed_or_statement.inc
+--source include/have_debug.inc
+--source include/master-slave.inc
+
+# Cleanup old binlog
+--source include/rpl_connection_master.inc
+FLUSH LOGS;
+--let $binlog= query_get_value(SHOW MASTER STATUS, File, 1)
+--replace_result $binlog binlog
+--eval purge binary logs to '$binlog'
+
+--source include/rpl_connection_slave.inc
+FLUSH LOGS;
+--let $binlog= query_get_value(SHOW MASTER STATUS, File, 1)
+--replace_result $binlog binlog
+--eval purge binary logs to '$binlog'
+
+--source suite/rpl_gtid/t/rpl_maintain_database_hlc.inc
diff --git a/mysql-test/suite/rpl_gtid/t/rpl_stress_hlc-master.opt b/mysql-test/suite/rpl_gtid/t/rpl_stress_hlc-master.opt
index 56786c52377a..b5012d184bd7 100644
--- a/mysql-test/suite/rpl_gtid/t/rpl_stress_hlc-master.opt
+++ b/mysql-test/suite/rpl_gtid/t/rpl_stress_hlc-master.opt
@@ -1 +1 @@
---enable_binlog_hlc=ON
+--enable_binlog_hlc=ON --maintain_database_hlc=ON
diff --git a/mysql-test/suite/rpl_gtid/t/rpl_stress_hlc-slave.opt b/mysql-test/suite/rpl_gtid/t/rpl_stress_hlc-slave.opt
index 0d0107947308..bd6fdfe2cc47 100644
--- a/mysql-test/suite/rpl_gtid/t/rpl_stress_hlc-slave.opt
+++ b/mysql-test/suite/rpl_gtid/t/rpl_stress_hlc-slave.opt
@@ -1 +1 @@
---enable_binlog_hlc=ON --slave_parallel_workers=30
+--enable_binlog_hlc=ON --slave_parallel_workers=30 --maintain_database_hlc=ON
diff --git a/mysql-test/suite/rpl_gtid/t/rpl_stress_hlc.test b/mysql-test/suite/rpl_gtid/t/rpl_stress_hlc.test
index 2bd4b87d8ceb..6620e82b0337 100644
--- a/mysql-test/suite/rpl_gtid/t/rpl_stress_hlc.test
+++ b/mysql-test/suite/rpl_gtid/t/rpl_stress_hlc.test
@@ -48,4 +48,14 @@ exec rm $MYSQLTEST_VARDIR/tmp/unique-sorted-hlc.dat;
 exec rm $MYSQLTEST_VARDIR/tmp/master-unsorted-hlc.dat;
 exec rm $MYSQLTEST_VARDIR/tmp/slave-sorted-hlc.dat;
 
+--source include/rpl_connection_master.inc
+--let $master_applied_hlc = `SELECT applied_hlc FROM information_schema.database_applied_hlc WHERE database_name LIKE 'test'`
+
+--source include/rpl_connection_slave.inc
+--let $slave_applied_hlc = `SELECT applied_hlc FROM information_schema.database_applied_hlc WHERE database_name LIKE 'test'`
+
+--let $assert_cond= $master_applied_hlc = $slave_applied_hlc
+--let $assert_text= Master and Slave should have the same applied hlc on the test database
+--source include/assert.inc
+
 --source include/rpl_end.inc
diff --git a/mysql-test/suite/sys_vars/r/maintain_database_hlc_basic.result b/mysql-test/suite/sys_vars/r/maintain_database_hlc_basic.result
new file mode 100644
index 000000000000..7371342937a9
--- /dev/null
+++ b/mysql-test/suite/sys_vars/r/maintain_database_hlc_basic.result
@@ -0,0 +1,54 @@
+SET @@global.maintain_database_hlc = true;
+ERROR 42000: Variable 'maintain_database_hlc' can't be set to the value of '1'
+SELECT @@global.maintain_database_hlc;
+@@global.maintain_database_hlc
+0
+SET @@global.enable_binlog_hlc = true;
+SELECT COUNT(@@GLOBAL.maintain_database_hlc);
+COUNT(@@GLOBAL.maintain_database_hlc)
+1
+SELECT COUNT(@@SESSION.maintain_database_hlc);
+ERROR HY000: Variable 'maintain_database_hlc' is a GLOBAL variable
+SELECT VARIABLE_NAME FROM performance_schema.global_variables WHERE VARIABLE_NAME='maintain_database_hlc';
+VARIABLE_NAME
+maintain_database_hlc
+SELECT VARIABLE_NAME FROM performance_schema.session_variables WHERE VARIABLE_NAME='maintain_database_hlc';
+VARIABLE_NAME
+maintain_database_hlc
+SET GLOBAL maintain_database_hlc= ON;
+include/assert.inc ['maintain_database_hlc is a dynamic variable']
+SET GLOBAL maintain_database_hlc= OFF;
+include/assert.inc ['maintain_database_hlc should be OFF']
+SET GLOBAL maintain_database_hlc= ON;
+include/assert.inc ['maintain_database_hlc should be ON']
+SET GLOBAL maintain_database_hlc= 0;
+include/assert.inc ['maintain_database_hlc should be OFF']
+SET GLOBAL maintain_database_hlc= 1;
+include/assert.inc ['maintain_database_hlc should be ON']
+SET GLOBAL maintain_database_hlc= DEFAULT;
+include/assert.inc ['maintain_database_hlc should be OFF']
+SET GLOBAL maintain_database_hlc= NULL;
+ERROR 42000: Variable 'maintain_database_hlc' can't be set to the value of 'NULL'
+SET GLOBAL maintain_database_hlc= '';
+ERROR 42000: Variable 'maintain_database_hlc' can't be set to the value of ''
+SET GLOBAL maintain_database_hlc= -1;
+ERROR 42000: Variable 'maintain_database_hlc' can't be set to the value of '-1'
+SET GLOBAL maintain_database_hlc= 1.0;
+ERROR 42000: Incorrect argument type to variable 'maintain_database_hlc'
+SET GLOBAL maintain_database_hlc= 'GARBAGE';
+ERROR 42000: Variable 'maintain_database_hlc' can't be set to the value of 'GARBAGE'
+SET GLOBAL maintain_database_hlc= 2;
+ERROR 42000: Variable 'maintain_database_hlc' can't be set to the value of '2'
+Expect value still set to "OFF"
+SELECT @@global.maintain_database_hlc;
+@@global.maintain_database_hlc
+0
+CREATE USER user1;
+SET GLOBAL maintain_database_hlc=ON;
+ERROR 42000: Access denied; you need (at least one of) the SUPER or SYSTEM_VARIABLES_ADMIN privilege(s) for this operation
+GRANT SYSTEM_VARIABLES_ADMIN ON *.* TO user1@'%';
+SET GLOBAL maintain_database_hlc=ON;
+REVOKE SYSTEM_VARIABLES_ADMIN ON *.* FROM user1@'%';
+SET GLOBAL maintain_database_hlc=OFF;
+ERROR 42000: Access denied; you need (at least one of) the SUPER or SYSTEM_VARIABLES_ADMIN privilege(s) for this operation
+DROP USER user1;
diff --git a/mysql-test/suite/sys_vars/t/maintain_database_hlc_basic-master.opt b/mysql-test/suite/sys_vars/t/maintain_database_hlc_basic-master.opt
new file mode 100644
index 000000000000..e7d9f84929ca
--- /dev/null
+++ b/mysql-test/suite/sys_vars/t/maintain_database_hlc_basic-master.opt
@@ -0,0 +1 @@
+--gtid_mode=ON --enforce_gtid_consistency
diff --git a/mysql-test/suite/sys_vars/t/maintain_database_hlc_basic.test b/mysql-test/suite/sys_vars/t/maintain_database_hlc_basic.test
new file mode 100644
index 000000000000..49c8b3ecaefd
--- /dev/null
+++ b/mysql-test/suite/sys_vars/t/maintain_database_hlc_basic.test
@@ -0,0 +1,134 @@
+###############################################################################
+# Variable Name: maintain_database_hlc
+# Scope: global
+# Access Type: dynamic
+# Data Type: boolean
+#
+# Description: Test case for checking the behavior of dynamic system variable
+#              "maintain_database_hlc", specifically regarding:
+#              - Scope & access type
+#              - Valid & default value
+#              - Invalid values
+#              - Required privileges
+#
+###############################################################################
+
+--source include/count_sessions.inc
+
+# Save initial value
+--let $saved_maintain_database_hlc= `SELECT @@global.maintain_database_hlc`
+--let $saved_enable_binlog_hlc= `SELECT @@global.enable_binlog_hlc`
+
+# Cannot be set unless enable_binlog_hlc is also set
+--Error ER_WRONG_VALUE_FOR_VAR
+SET @@global.maintain_database_hlc = true;
+SELECT @@global.maintain_database_hlc;
+
+SET @@global.enable_binlog_hlc = true;
+
+#
+# Scope: Global only
+#
+SELECT COUNT(@@GLOBAL.maintain_database_hlc);
+
+--error ER_INCORRECT_GLOBAL_LOCAL_VAR
+SELECT COUNT(@@SESSION.maintain_database_hlc);
+
+--disable_warnings
+SELECT VARIABLE_NAME FROM performance_schema.global_variables WHERE VARIABLE_NAME='maintain_database_hlc';
+SELECT VARIABLE_NAME FROM performance_schema.session_variables WHERE VARIABLE_NAME='maintain_database_hlc';
+--enable_warnings
+
+#
+# Access Type: Dynamic
+#
+SET GLOBAL maintain_database_hlc= ON;
+--let $assert_text= 'maintain_database_hlc is a dynamic variable'
+--let $assert_cond= "[SELECT @@GLOBAL.maintain_database_hlc]" = "1"
+--source include/assert.inc
+
+#
+# Valid values and Default value
+#
+SET GLOBAL maintain_database_hlc= OFF;
+--let $assert_text= 'maintain_database_hlc should be OFF'
+--let $assert_cond= "[SELECT @@GLOBAL.maintain_database_hlc]" = "0"
+--source include/assert.inc
+
+SET GLOBAL maintain_database_hlc= ON;
+--let $assert_text= 'maintain_database_hlc should be ON'
+--let $assert_cond= "[SELECT @@GLOBAL.maintain_database_hlc]" = "1"
+--source include/assert.inc
+
+SET GLOBAL maintain_database_hlc= 0;
+--let $assert_text= 'maintain_database_hlc should be OFF'
+--let $assert_cond= "[SELECT @@GLOBAL.maintain_database_hlc]" = "0"
+--source include/assert.inc
+
+SET GLOBAL maintain_database_hlc= 1;
+--let $assert_text= 'maintain_database_hlc should be ON'
+--let $assert_cond= "[SELECT @@GLOBAL.maintain_database_hlc]" = "1"
+--source include/assert.inc
+
+SET GLOBAL maintain_database_hlc= DEFAULT;
+--let $assert_text= 'maintain_database_hlc should be OFF'
+--let $assert_cond= "[SELECT @@GLOBAL.maintain_database_hlc]" = "0"
+--source include/assert.inc
+
+#
+# Invalid values
+#
+--error ER_WRONG_VALUE_FOR_VAR
+SET GLOBAL maintain_database_hlc= NULL;
+
+--error ER_WRONG_VALUE_FOR_VAR
+SET GLOBAL maintain_database_hlc= '';
+
+--error ER_WRONG_VALUE_FOR_VAR
+SET GLOBAL maintain_database_hlc= -1;
+
+--error ER_WRONG_TYPE_FOR_VAR
+SET GLOBAL maintain_database_hlc= 1.0;
+
+--error ER_WRONG_VALUE_FOR_VAR
+SET GLOBAL maintain_database_hlc= 'GARBAGE';
+
+--error ER_WRONG_VALUE_FOR_VAR
+SET GLOBAL maintain_database_hlc= 2;
+
+--echo Expect value still set to "OFF"
+SELECT @@global.maintain_database_hlc;
+
+#
+# Privileges
+#
+
+CREATE USER user1;
+--connect(conn_user1,localhost,user1,,)
+
+--Error ER_SPECIFIC_ACCESS_DENIED_ERROR
+SET GLOBAL maintain_database_hlc=ON;
+
+--connection default
+GRANT SYSTEM_VARIABLES_ADMIN ON *.* TO user1@'%';
+--connection conn_user1
+SET GLOBAL maintain_database_hlc=ON;
+
+--connection default
+REVOKE SYSTEM_VARIABLES_ADMIN ON *.* FROM user1@'%';
+--connection conn_user1
+# FAST_INTEGER_TO_STRING_ADMIN is not enough
+--Error ER_SPECIFIC_ACCESS_DENIED_ERROR
+SET GLOBAL maintain_database_hlc=OFF;
+
+--connection default
+--disconnect conn_user1
+DROP USER user1;
+
+# Clean up
+--disable_query_log
+--eval SET GLOBAL enable_binlog_hlc = $saved_enable_binlog_hlc
+--eval SET GLOBAL maintain_database_hlc= $saved_maintain_database_hlc
+--enable_query_log
+
+--source include/wait_until_count_sessions.inc
diff --git a/sql/binlog.cc b/sql/binlog.cc
index 5f4a86e6d826..5922059aad10 100644
--- a/sql/binlog.cc
+++ b/sql/binlog.cc
@@ -127,7 +127,6 @@
 #include "sql/sql_backup_lock.h"  // is_instance_backup_locked et al.
 #include "sql/sql_base.h"         // find_temporary_table
 #include "sql/sql_bitmap.h"
-#include "sql/sql_class.h"  // THD
 #include "sql/sql_const.h"
 #include "sql/sql_data_change.h"
 #include "sql/sql_error.h"
@@ -2737,6 +2736,27 @@ uint64_t HybridLogicalClock::update(uint64_t minimum_hlc) {
   return new_min_hlc;
 }
 
+void HybridLogicalClock::update_database_hlc(
+    const database_container &databases, uint64_t applied_hlc) {
+  std::unique_lock<std::mutex> lock(database_applied_hlc_lock_);
+
+  for (const auto &database : databases) {
+    auto database_hlc = database_applied_hlc_.find(database);
+    if (database_hlc == database_applied_hlc_.end()) {
+      // Seeing this database for the first time. Create a new entry
+      database_applied_hlc_.emplace(database, applied_hlc);
+    } else if (database_hlc->second < applied_hlc) {
+      database_hlc->second = applied_hlc;
+    }
+  }
+}
+
+database_hlc_container HybridLogicalClock::get_database_hlc() const {
+  std::unique_lock<std::mutex> lock(database_applied_hlc_lock_);
+
+  return database_applied_hlc_;
+}
+
 /**
   Write a rollback record of the transaction to the binary log.
 
@@ -9145,6 +9165,20 @@ void MYSQL_BIN_LOG::process_commit_stage_queue(THD *thd, THD *first) {
 
 void MYSQL_BIN_LOG::process_after_commit_stage_queue(THD *thd, THD *first) {
   for (THD *head = first; head; head = head->next_to_commit) {
+    if (enable_binlog_hlc && maintain_database_hlc && head->hlc_time_ns_next) {
+      if (likely(!head->databases.empty())) {
+        // Successfully committed the trx to engine. Update applied hlc for
+        // all databases that this trx touches
+        hlc.update_database_hlc(head->databases, head->hlc_time_ns_next);
+      } else {
+        // Log a error line if databases are empty. This could happen in SBR
+        // NO_LINT_DEBUG
+        sql_print_error("Databases were empty for this trx. HLC= %lu",
+                        head->hlc_time_ns_next);
+      }
+    }
+    head->databases.clear();
+
     if (head->get_transaction()->m_flags.run_hooks &&
         head->commit_error != THD::CE_COMMIT_ERROR) {
       /*
diff --git a/sql/binlog.h b/sql/binlog.h
index 8a810ad79e09..aa409b838d56 100644
--- a/sql/binlog.h
+++ b/sql/binlog.h
@@ -29,6 +29,8 @@
 #include <atomic>
 #include <limits>
 #include <list>
+#include <mutex>
+#include <unordered_map>
 #include <utility>
 
 #include "libbinlogevents/include/binlog_event.h"  // enum_binlog_checksum_alg
@@ -51,6 +53,7 @@
 #include "sql/binlog_reader.h"  // Binlog_file_reader
 #include "sql/rpl_commit_stage_manager.h"
 #include "sql/rpl_trx_tracking.h"
+#include "sql/sql_class.h"         // THD
 #include "sql/tc_log.h"            // TC_LOG
 #include "sql/transaction_info.h"  // Transaction_ctx
 #include "thr_mutex.h"
@@ -77,6 +80,7 @@ struct Gtid;
 typedef int64 query_id_t;
 
 using log_file_name_container = std::list<std::string>;
+using database_hlc_container = std::unordered_map<std::string, uint64_t>;
 
 /*
   Maximum unique log filename extension.
@@ -186,9 +190,34 @@ class HybridLogicalClock {
    */
   uint64_t update(uint64_t minimum_hlc);
 
+  /**
+   * Update the applied HLC for specified databases
+   *
+   * @param  databases - the databases for which hlc needs to be updated
+   * @param  applied_hlc - Applied HLC
+   */
+  void update_database_hlc(const database_container &databases,
+                           uint64_t applied_hlc);
+
+  /**
+   * Get the applied hlc for all the database that is being tracked by this
+   * clock
+   *
+   * @param  [out] A map of database->applied_hlc
+   */
+  database_hlc_container get_database_hlc() const;
+
  private:
   // nanosecond precision internal clock
   std::atomic<uint64_t> current_;
+
+  /**
+   * A map of applied HLC for each database. The key is the name of the database
+   * and the value is the applied_hlc for that database. Applied HLC is the HLC
+   * of the last known trx that was applied (committed) to the engine
+   */
+  database_hlc_container database_applied_hlc_;
+  mutable std::mutex database_applied_hlc_lock_;
 };
 
 struct st_filenum_pos {
@@ -672,6 +701,11 @@ class MYSQL_BIN_LOG : public TC_LOG {
      be used to synchronize HLC across different communicating instances */
   uint64_t update_hlc(uint64_t minimum_hlc) { return hlc.update(minimum_hlc); }
 
+  /* get the applied HLC for all known databases in this instance */
+  database_hlc_container get_database_hlc() const {
+    return hlc.get_database_hlc();
+  }
+
  private:
   std::atomic<enum_log_state> atomic_log_state{LOG_CLOSED};
 
diff --git a/sql/handler.h b/sql/handler.h
index a975ae0c8534..0bca6f1432f2 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -943,7 +943,8 @@ enum enum_schema_tables : int {
   SCH_TMP_TABLE_KEYS,
   SCH_AUTHINFO,
   SCH_SLAVE_DB_LOAD,
-  SCH_LAST = SCH_SLAVE_DB_LOAD
+  SCH_DATABASE_APPLIED_HLC,
+  SCH_LAST = SCH_DATABASE_APPLIED_HLC
 };
 
 enum ha_stat_type {
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 556db151fd8f..7e61aaf79ee0 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -3656,8 +3656,11 @@ bool Query_log_event::write(Basic_ostream *ostream) {
       if (dbs == 1 && !strcmp(db_name, "")) dbs = OVER_MAX_DBS_IN_EVENT_MTS;
       *start++ = dbs;
       if (dbs != OVER_MAX_DBS_IN_EVENT_MTS) do {
+          size_t db_name_len = strlen(db_name);
           strcpy((char *)start, db_name);
-          start += strlen(db_name) + 1;
+          start += db_name_len + 1;
+          if (enable_binlog_hlc && maintain_database_hlc)
+            thd->databases.emplace(db_name, db_name_len);
         } while ((db_name = it++));
     } else {
       *start++ = dbs;
@@ -11169,6 +11172,9 @@ Table_map_log_event::Table_map_log_event(THD *thd_arg, TABLE *tbl,
   } else {
     m_data_size += m_metadata_buf.length();
   }
+
+  if (enable_binlog_hlc && maintain_database_hlc && !m_dbnam.empty())
+    thd->databases.insert(m_dbnam);
 }
 #endif /* defined(MYSQL_SERVER) */
 
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 69b194ce6c37..4bb43e973956 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -1221,6 +1221,7 @@ ulong log_error_verbosity = 3;  // have a non-zero value during early start-up
 bool opt_keyring_migration_to_component = false;
 bool opt_persist_sensitive_variables_in_plaintext{true};
 bool enable_binlog_hlc = 0;
+bool maintain_database_hlc = false;
 
 #if defined(_WIN32)
 /*
diff --git a/sql/mysqld.h b/sql/mysqld.h
index 96248fa92587..b0e6ce2f8963 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -285,6 +285,7 @@ extern std::atomic<bool> offline_mode;
 extern uint test_flags, select_errors, ha_open_options;
 extern uint protocol_version, mysqld_port;
 extern bool enable_binlog_hlc;
+extern bool maintain_database_hlc;
 
 enum enum_delay_key_write {
   DELAY_KEY_WRITE_NONE,
diff --git a/sql/sql_class.h b/sql/sql_class.h
index d7a57c3536b5..e6b3253df8cf 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -172,6 +172,7 @@ typedef struct user_conn USER_CONN;
 struct MYSQL_LOCK;
 struct st_ac_node;
 using st_ac_node_ptr = std::shared_ptr<st_ac_node>;
+using database_container = std::unordered_set<std::string>;
 
 enum enum_slave_use_idempotent_for_recovery {
   SLAVE_USE_IDEMPOTENT_FOR_RECOVERY_NO,
@@ -1083,6 +1084,7 @@ class THD : public MDL_context_owner,
   /* Next HLC value */
   uint64_t hlc_time_ns_next = 0;
   bool should_write_hlc = false;
+  database_container databases;
 
   /**
     The function checks whether the thread is processing queries from binlog,
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 98ac6b3183b2..898f886c62ac 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -5261,6 +5261,19 @@ static int hton_fill_schema_table(THD *thd, Table_ref *tables, Item *cond) {
   return 0;
 }
 
+int fill_db_applied_hlc(THD *thd, Table_ref *tables, Item *) {
+  auto table = tables->table;
+
+  for (const auto &hlc : mysql_bin_log.get_database_hlc()) {
+    table->field[0]->store(hlc.first.c_str(), hlc.first.length(),
+                           system_charset_info);
+    table->field[1]->store(static_cast<ulonglong>(hlc.second), true);
+    schema_table_store_record(thd, table);
+  }
+
+  return 0;
+}
+
 ST_FIELD_INFO engines_fields_info[] = {
     {"ENGINE", 64, MYSQL_TYPE_STRING, 0, 0, "Engine", 0},
     {"SUPPORT", 8, MYSQL_TYPE_STRING, 0, 0, "Support", 0},
@@ -5415,6 +5428,11 @@ ST_FIELD_INFO tmp_table_columns_fields_info[] = {
      MYSQL_TYPE_STRING, 0, 0, "Generation expression", 0},
     {nullptr, 0, MYSQL_TYPE_STRING, 0, 0, nullptr, 0}};
 
+ST_FIELD_INFO db_applied_hlc_fields_info[] = {
+    {"DATABASE_NAME", NAME_LEN, MYSQL_TYPE_STRING, 0, 0, 0, 0},
+    {"APPLIED_HLC", 21, MYSQL_TYPE_LONGLONG, 0, MY_I_S_UNSIGNED, 0, 0},
+    {0, 0, MYSQL_TYPE_STRING, 0, 0, 0, 0}};
+
 /** For creating fields of information_schema.OPTIMIZER_TRACE */
 extern ST_FIELD_INFO optimizer_trace_info[];
 
@@ -5459,6 +5477,8 @@ ST_SCHEMA_TABLE schema_tables[] = {
      false},
     {"SLAVE_DB_LOAD", slave_db_load_fields_info, fill_slave_db_load, 0, 0,
      false},
+    {"DATABASE_APPLIED_HLC", db_applied_hlc_fields_info, fill_db_applied_hlc,
+     nullptr, nullptr, false},
     {nullptr, nullptr, nullptr, nullptr, nullptr, false}};
 
 int initialize_schema_table(st_plugin_int *plugin) {
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index 9dfbeed05d0a..5c09c3117dc5 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -8476,3 +8476,17 @@ static Sys_var_bool Sys_enable_binlog_hlc(
     "Enable logging HLC timestamp as part of Metadata log event",
     GLOBAL_VAR(enable_binlog_hlc), CMD_LINE(OPT_ARG), DEFAULT(false),
     NO_MUTEX_GUARD, NOT_IN_BINLOG, ON_CHECK(check_enable_binlog_hlc));
+
+static bool check_maintain_database_hlc(sys_var *, THD *, set_var *var) {
+  uint64_t new_maintain_db_hlc = var->save_result.ulonglong_value;
+  if (!enable_binlog_hlc && new_maintain_db_hlc)
+    return true;  // Needs enable_binlog_hlc
+
+  return false;
+}
+
+static Sys_var_bool Sys_maintain_database_hlc(
+    "maintain_database_hlc",
+    "Enable maintaining of max HLC applied per database",
+    GLOBAL_VAR(maintain_database_hlc), CMD_LINE(OPT_ARG), DEFAULT(false),
+    NO_MUTEX_GUARD, NOT_IN_BINLOG, ON_CHECK(check_maintain_database_hlc));