diff --git a/mysql-test/include/commit.inc b/mysql-test/include/commit.inc
index 43a4af38e3ad..260a67583878 100644
--- a/mysql-test/include/commit.inc
+++ b/mysql-test/include/commit.inc
@@ -673,7 +673,7 @@ begin
 end|
 delimiter ;|
 --disable_warnings
-call p_verify_status_increment(6, 4, 2, 0);
+call p_verify_status_increment(7, 4, 2, 0);
 --enable_warnings
 
 --echo # 16. A function changes non-trans-table.
@@ -731,14 +731,14 @@ create temporary table t2 (a int);
 let $have_debug = `SELECT VERSION() LIKE '%debug%'`;
 if (!$have_debug)
 {
-  --replace_regex /8/X/
-  call p_verify_status_increment(8, 0, 0, 0);
+  --replace_regex /9/X/
+  call p_verify_status_increment(9, 0, 0, 0);
 }
 
 if ($have_debug)
 {
-  --replace_regex /9/X/
-  call p_verify_status_increment(9, 0, 0, 0);
+  --replace_regex /10/X/
+  call p_verify_status_increment(10, 0, 0, 0);
 }
 --enable_warnings
 set sql_mode=default;
@@ -878,13 +878,13 @@ create table t2 (a int);
 # COM_PREPARE.
 if (`SELECT $PS_PROTOCOL = 0`)
 {
-  --replace_regex /11/<commit_count>/
-  call p_verify_status_increment(11, 4, 6, 0);
+  --replace_regex /12/<commit_count>/
+  call p_verify_status_increment(12, 4, 6, 0);
 }
 if (`SELECT $PS_PROTOCOL > 0`)
 {
-  --replace_regex /12/<commit_count>/
-  call p_verify_status_increment(12, 4, 6, 0);
+  --replace_regex /13/<commit_count>/
+  call p_verify_status_increment(13, 4, 6, 0);
 }
 --echo # One extra transaction to load table 't2' into data-dictionary cache.
 do (select f1() from t1 where a=2);
@@ -938,48 +938,48 @@ create table t3 select a from t2;
 # COM_PREPARE.
 if (`SELECT $PS_PROTOCOL = 0`)
 {
- --replace_regex /12/<commit_count>/
-  call p_verify_status_increment(12, 4, 4, 4);
+ --replace_regex /13/<commit_count>/
+  call p_verify_status_increment(13, 4, 4, 4);
 }
 if (`SELECT $PS_PROTOCOL > 0`)
 {
-  --replace_regex /13/<commit_count>/
-  call p_verify_status_increment(13, 4, 4, 4);
+  --replace_regex /14/<commit_count>/
+  call p_verify_status_increment(14, 4, 4, 4);
 }
 alter table t3 add column (b int);
 
 if (!$have_debug)
 {
-  --replace_regex /20/X/
-  call p_verify_status_increment(20, 4, 2, 0);
+  --replace_regex /21/X/
+  call p_verify_status_increment(21, 4, 2, 0);
 }
 
 if ($have_debug)
 {
-  --replace_regex /21/X/
-  call p_verify_status_increment(21, 4, 2, 0);
+  --replace_regex /22/X/
+  call p_verify_status_increment(22, 4, 2, 0);
 }
 
 alter table t3 rename t4;
-call p_verify_status_increment(20, 4, 3, 0);
+call p_verify_status_increment(21, 4, 3, 0);
 rename table t4 to t3;
-call p_verify_status_increment(24, 4, 5, 0);
+call p_verify_status_increment(25, 4, 5, 0);
 truncate table t3;
 
 if (!$have_debug)
 {
-  --replace_regex /10/X/
-  call p_verify_status_increment(10, 4, 4, 0);
+  --replace_regex /12/X/
+  call p_verify_status_increment(12, 4, 4, 0);
 }
 
 if ($have_debug)
 {
-  --replace_regex /11/X/
-  call p_verify_status_increment(11, 4, 4, 0);
+  --replace_regex /13/X/
+  call p_verify_status_increment(13, 4, 4, 0);
 }
 
 create view v1 as select * from t2;
-call p_verify_status_increment(7, 4, 3, 0);
+call p_verify_status_increment(8, 4, 3, 0);
 check table t1;
 call p_verify_status_increment(2, 0, 2, 0);
 --echo # Sic: after this bug is fixed, CHECK leaves no pending transaction
@@ -995,14 +995,14 @@ drop view v1;
 --echo # One extra transaction to load v1 into data-dictionary cache.
 if (!$have_debug)
 {
-  --replace_regex /\(6/(X/
-  call p_verify_status_increment(6, 4, 3, 0);
+  --replace_regex /\(7/(X/
+  call p_verify_status_increment(7, 4, 3, 0);
 }
 
 if ($have_debug)
 {
-  --replace_regex /7/X/
-  call p_verify_status_increment(7, 4, 3, 0);
+  --replace_regex /8/X/
+  call p_verify_status_increment(8, 4, 3, 0);
 }
 
 --enable_warnings
diff --git a/mysql-test/include/db_read_only_off.inc b/mysql-test/include/db_read_only_off.inc
new file mode 100644
index 000000000000..987cb462dcbf
--- /dev/null
+++ b/mysql-test/include/db_read_only_off.inc
@@ -0,0 +1,31 @@
+truncate t1;
+truncate t2;
+
+begin;
+insert into t1 values (0), (1), (2);
+insert into t2 values (0), (1), (2);
+update t1 set a = a + 1;
+# multiple table updates
+update t1, t2 set t1.a=t1.a+1, t2.a=t2.a*2;
+select 't1', a from t1;
+select 't2', a from t2;
+commit;
+
+alter table t1 add key (a);
+describe t1;
+drop index a on t1;
+
+alter table t2 add key (a);
+describe t2;
+drop index a on t2;
+
+truncate t1;
+truncate t2;
+
+--echo # populate some data
+begin;
+insert into t1 values (1), (2), (3);
+insert into t2 values (1), (2), (3);
+select 't1', a from t1;
+select 't2', a from t2;
+commit;
diff --git a/mysql-test/include/db_read_only_on.inc b/mysql-test/include/db_read_only_on.inc
new file mode 100644
index 000000000000..88488b776fe6
--- /dev/null
+++ b/mysql-test/include/db_read_only_on.inc
@@ -0,0 +1,120 @@
+--echo # continue previous transaction
+--echo # read_only was turned on in the middle of a transaction
+--echo # new update/insert statement will be blocked immediately
+update t1 set a = a + 1;
+# multiple table updates
+update t1, t2 set t1.a=t1.a+1, t2.a=t2.a*2;
+select 't1', a from t1;
+select 't2', a from t2;
+
+--echo # write transaction was rolled back at the end
+--error ER_DB_READ_ONLY
+commit;
+
+# insert/update rolled back
+select 't1', a from t1;
+select 't2', a from t2;
+
+--echo # write transaction with 'begin'
+begin;
+insert into t1 values (4), (5), (6);
+update t1 set a = a + 1;
+# creating a table DDL is failed immdiately
+--error ER_DB_READ_ONLY
+create table t3 (a int) engine=innodb;
+select a from t1;
+--error ER_DB_READ_ONLY
+commit;
+
+--echo # read-only transactions are ok
+begin;
+select count(*) from t1;
+select count(*) from t2;
+commit;
+
+--echo # transaction without 'begin'
+insert into t1 values (4), (5), (6);
+insert into t1 values (7), (8), (9);
+select a from t1;
+--error ER_DB_READ_ONLY
+commit;
+select a from t1;
+
+--echo # rolled-back transaction
+insert into t1 values (4), (5), (6);
+insert into t1 values (7), (8), (9);
+select a from t1;
+rollback;
+select a from t1;
+
+set autocommit = 1;
+--echo # multiple table updates (autocommit)
+--error ER_DB_READ_ONLY
+update t1, t2 set t1.a=t1.a+1, t2.a=t2.a*2;
+select 't1', a from t1;
+select 't2', a from t2;
+set autocommit = 0;
+
+# table update DDL is blocked
+--error ER_DB_READ_ONLY
+alter table t1 add key (a);
+
+# table update DDL is blocked
+--error ER_DB_READ_ONLY
+create index a on t1 (a);
+
+# drop table is not allowed
+--error ER_DB_READ_ONLY
+drop table t1;
+
+# drop database is not allowed
+--error ER_DB_READ_ONLY
+drop database test;
+
+--echo #
+--echo # OK to create temporary table
+--echo #
+create temporary table temp1 (a int);
+insert into temp1 select * from t1;
+update temp1 set a = a + 1;
+select * from temp1;
+drop temporary table temp1;
+
+--echo #
+--echo # OK to switch and write another database
+--echo # read_only scope is per database
+--echo #
+create database test2;
+use test2;
+show create database test2;
+create table t1 (a int) engine = innodb;
+insert into t1 values (0), (1), (2);
+update t1 set a = a + 1;
+select a from t1;
+
+--echo #
+--echo # cross-db/noncurrent-db transaction
+--echo # Transaction writing to test db from session of test2 db
+--echo #
+begin;
+insert into test.t1 values (4), (5), (6);
+update test.t1 set a = a + 1;
+select a from test.t1;
+--error ER_DB_READ_ONLY
+commit;
+select a from test.t1;
+select a from test2.t1;
+
+begin;
+insert into test.t1 values (4), (5), (6);
+update test.t1 set a = a + 1;
+select a from test.t1;
+update test2.t1 set a = a + 1;
+select a from test2.t1;
+--error ER_DB_READ_ONLY
+commit;
+select a from test.t1;
+select a from test2.t1;
+
+use test;
+drop database test2;
diff --git a/mysql-test/r/commit_1innodb.result b/mysql-test/r/commit_1innodb.result
index c3fa2664706a..fbeea82b64c2 100644
--- a/mysql-test/r/commit_1innodb.result
+++ b/mysql-test/r/commit_1innodb.result
@@ -576,7 +576,7 @@ begin
 insert t2 set a=2;
 return 2;
 end|
-call p_verify_status_increment(6, 4, 2, 0);
+call p_verify_status_increment(7, 4, 2, 0);
 SUCCESS
 
 # 16. A function changes non-trans-table.
@@ -853,11 +853,11 @@ call p_verify_status_increment(X, 4, 2, 0);
 SUCCESS
 
 alter table t3 rename t4;
-call p_verify_status_increment(20, 4, 3, 0);
+call p_verify_status_increment(21, 4, 3, 0);
 SUCCESS
 
 rename table t4 to t3;
-call p_verify_status_increment(24, 4, 5, 0);
+call p_verify_status_increment(25, 4, 5, 0);
 SUCCESS
 
 truncate table t3;
@@ -865,7 +865,7 @@ call p_verify_status_increment(X, 4, 4, 0);
 SUCCESS
 
 create view v1 as select * from t2;
-call p_verify_status_increment(7, 4, 3, 0);
+call p_verify_status_increment(8, 4, 3, 0);
 SUCCESS
 
 check table t1;
diff --git a/mysql-test/r/create.result b/mysql-test/r/create.result
index 607b7741d771..f9c2ca7413d1 100644
--- a/mysql-test/r/create.result
+++ b/mysql-test/r/create.result
@@ -956,7 +956,7 @@ create table t2 select sql_big_result f1,count(f2) from t1 group by f1;
 show status like 'handler_read%';
 Variable_name	Value
 Handler_read_first	1
-Handler_read_key	23
+Handler_read_key	24
 Handler_read_last	0
 Handler_read_next	0
 Handler_read_prev	0
diff --git a/mysql-test/r/db_read_only.result b/mysql-test/r/db_read_only.result
new file mode 100644
index 000000000000..a6107883ce70
--- /dev/null
+++ b/mysql-test/r/db_read_only.result
@@ -0,0 +1,551 @@
+drop table if exists t1, t2;
+drop database if exists test2;
+CREATE USER test@localhost;
+grant CREATE, SELECT, INSERT, UPDATE, DROP on *.* to test@localhost;
+connect (con1,localhost,test,,test);
+connect (root2,localhost,root,,test);
+# turn off autocommit in con1
+connection con1;
+set autocommit = 0;
+select @@autocommit;
+@@autocommit
+0
+# turn off autocommit in root2
+connection root2;
+set autocommit = 0;
+select @@autocommit;
+@@autocommit
+0
+create table t1 (a int) engine = innodb;
+create table t2 (a int) engine = innodb;
+#
+# read_only = false
+# both regular and super users can write
+#
+connection con1;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */
+truncate t1;
+truncate t2;
+begin;
+insert into t1 values (0), (1), (2);
+insert into t2 values (0), (1), (2);
+update t1 set a = a + 1;
+update t1, t2 set t1.a=t1.a+1, t2.a=t2.a*2;
+select 't1', a from t1;
+t1	a
+t1	2
+t1	3
+t1	4
+select 't2', a from t2;
+t2	a
+t2	0
+t2	2
+t2	4
+commit;
+alter table t1 add key (a);
+describe t1;
+Field	Type	Null	Key	Default	Extra
+a	int	YES	MUL	NULL	
+drop index a on t1;
+alter table t2 add key (a);
+describe t2;
+Field	Type	Null	Key	Default	Extra
+a	int	YES	MUL	NULL	
+drop index a on t2;
+truncate t1;
+truncate t2;
+# populate some data
+begin;
+insert into t1 values (1), (2), (3);
+insert into t2 values (1), (2), (3);
+select 't1', a from t1;
+t1	a
+t1	1
+t1	2
+t1	3
+select 't2', a from t2;
+t2	a
+t2	1
+t2	2
+t2	3
+commit;
+connection root2;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */
+truncate t1;
+truncate t2;
+begin;
+insert into t1 values (0), (1), (2);
+insert into t2 values (0), (1), (2);
+update t1 set a = a + 1;
+update t1, t2 set t1.a=t1.a+1, t2.a=t2.a*2;
+select 't1', a from t1;
+t1	a
+t1	2
+t1	3
+t1	4
+select 't2', a from t2;
+t2	a
+t2	0
+t2	2
+t2	4
+commit;
+alter table t1 add key (a);
+describe t1;
+Field	Type	Null	Key	Default	Extra
+a	int	YES	MUL	NULL	
+drop index a on t1;
+alter table t2 add key (a);
+describe t2;
+Field	Type	Null	Key	Default	Extra
+a	int	YES	MUL	NULL	
+drop index a on t2;
+truncate t1;
+truncate t2;
+# populate some data
+begin;
+insert into t1 values (1), (2), (3);
+insert into t2 values (1), (2), (3);
+select 't1', a from t1;
+t1	a
+t1	1
+t1	2
+t1	3
+select 't2', a from t2;
+t2	a
+t2	1
+t2	2
+t2	3
+commit;
+#
+# read_only
+#
+connection default;
+# turn off read_only initially
+alter database test read_only = false;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */
+connection con1;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */
+# simulate long running transactions
+begin;
+insert into t1 values (4), (5), (6);
+insert into t2 values (4), (5), (6);
+select 't1', a from t1;
+t1	a
+t1	1
+t1	2
+t1	3
+t1	4
+t1	5
+t1	6
+select 't2', a from t2;
+t2	a
+t2	1
+t2	2
+t2	3
+t2	4
+t2	5
+t2	6
+connection default;
+# turn on read_only in the middle of transaction (con1)
+set lock_wait_timeout = 1;
+alter database test read_only = true;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction: Timeout on table metadata: test.t1
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */
+connection con1;
+rollback;
+#
+# super_read_only
+#
+connection default;
+# turn off read_only initially
+alter database test read_only = false;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */
+connection root2;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */
+# simulate long running transactions
+begin;
+insert into t1 values (4), (5), (6);
+insert into t2 values (4), (5), (6);
+select 't1', a from t1;
+t1	a
+t1	1
+t1	2
+t1	3
+t1	4
+t1	5
+t1	6
+select 't2', a from t2;
+t2	a
+t2	1
+t2	2
+t2	3
+t2	4
+t2	5
+t2	6
+connection default;
+# now turn on super read_only in the middle of transaction (root2)
+set lock_wait_timeout = 1;
+alter database test super_read_only = true;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction: Timeout on table metadata: test.t1
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */
+connection root2;
+rollback;
+alter database test super_read_only = true;
+connection root2;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci SUPER_READ_ONLY */ /*!80016 DEFAULT ENCRYPTION='N' */
+# continue previous transaction
+# read_only was turned on in the middle of a transaction
+# new update/insert statement will be blocked immediately
+update t1 set a = a + 1;
+update t1, t2 set t1.a=t1.a+1, t2.a=t2.a*2;
+select 't1', a from t1;
+t1	a
+t1	3
+t1	4
+t1	5
+select 't2', a from t2;
+t2	a
+t2	2
+t2	4
+t2	6
+# write transaction was rolled back at the end
+commit;
+ERROR HY000: Database 'test' is in read-only mode. Transaction was rolled back.
+select 't1', a from t1;
+t1	a
+t1	1
+t1	2
+t1	3
+select 't2', a from t2;
+t2	a
+t2	1
+t2	2
+t2	3
+# write transaction with 'begin'
+begin;
+insert into t1 values (4), (5), (6);
+update t1 set a = a + 1;
+create table t3 (a int) engine=innodb;
+ERROR HY000: Database 'test' is in read-only mode. Implicit commit failed.
+select a from t1;
+a
+2
+3
+4
+5
+6
+7
+commit;
+ERROR HY000: Database 'test' is in read-only mode. Transaction was rolled back.
+# read-only transactions are ok
+begin;
+select count(*) from t1;
+count(*)
+3
+select count(*) from t2;
+count(*)
+3
+commit;
+# transaction without 'begin'
+insert into t1 values (4), (5), (6);
+insert into t1 values (7), (8), (9);
+select a from t1;
+a
+1
+2
+3
+4
+5
+6
+7
+8
+9
+commit;
+ERROR HY000: Database 'test' is in read-only mode. Transaction was rolled back.
+select a from t1;
+a
+1
+2
+3
+# rolled-back transaction
+insert into t1 values (4), (5), (6);
+insert into t1 values (7), (8), (9);
+select a from t1;
+a
+1
+2
+3
+4
+5
+6
+7
+8
+9
+rollback;
+select a from t1;
+a
+1
+2
+3
+set autocommit = 1;
+# multiple table updates (autocommit)
+update t1, t2 set t1.a=t1.a+1, t2.a=t2.a*2;
+ERROR HY000: Database 'test' is in read-only mode. Transaction was rolled back.
+select 't1', a from t1;
+t1	a
+t1	1
+t1	2
+t1	3
+select 't2', a from t2;
+t2	a
+t2	1
+t2	2
+t2	3
+set autocommit = 0;
+alter table t1 add key (a);
+ERROR HY000: Database 'test' is in read-only mode. Implicit commit failed.
+create index a on t1 (a);
+ERROR HY000: Database 'test' is in read-only mode. Implicit commit failed.
+drop table t1;
+ERROR HY000: Database 'test' is in read-only mode. Implicit commit failed.
+drop database test;
+ERROR HY000: Database 'test' is in read-only mode. Cannot drop read only DB.
+#
+# OK to create temporary table
+#
+create temporary table temp1 (a int);
+insert into temp1 select * from t1;
+update temp1 set a = a + 1;
+select * from temp1;
+a
+2
+3
+4
+drop temporary table temp1;
+#
+# OK to switch and write another database
+# read_only scope is per database
+#
+create database test2;
+use test2;
+show create database test2;
+Database	Create Database
+test2	CREATE DATABASE `test2` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */
+create table t1 (a int) engine = innodb;
+insert into t1 values (0), (1), (2);
+update t1 set a = a + 1;
+select a from t1;
+a
+1
+2
+3
+#
+# cross-db/noncurrent-db transaction
+# Transaction writing to test db from session of test2 db
+#
+begin;
+insert into test.t1 values (4), (5), (6);
+update test.t1 set a = a + 1;
+select a from test.t1;
+a
+2
+3
+4
+5
+6
+7
+commit;
+ERROR HY000: Database 'test' is in read-only mode. Transaction was rolled back.
+select a from test.t1;
+a
+1
+2
+3
+select a from test2.t1;
+a
+1
+2
+3
+begin;
+insert into test.t1 values (4), (5), (6);
+update test.t1 set a = a + 1;
+select a from test.t1;
+a
+2
+3
+4
+5
+6
+7
+update test2.t1 set a = a + 1;
+select a from test2.t1;
+a
+2
+3
+4
+commit;
+ERROR HY000: Database 'test' is in read-only mode. Transaction was rolled back.
+select a from test.t1;
+a
+1
+2
+3
+select a from test2.t1;
+a
+1
+2
+3
+use test;
+drop database test2;
+#
+# changing db_read_only in root2 session
+#
+alter database test super_read_only = false;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci READ_ONLY */ /*!80016 DEFAULT ENCRYPTION='N' */
+# super_read_only turned into read_only
+truncate t1;
+truncate t2;
+begin;
+insert into t1 values (0), (1), (2);
+insert into t2 values (0), (1), (2);
+update t1 set a = a + 1;
+update t1, t2 set t1.a=t1.a+1, t2.a=t2.a*2;
+select 't1', a from t1;
+t1	a
+t1	2
+t1	3
+t1	4
+select 't2', a from t2;
+t2	a
+t2	0
+t2	2
+t2	4
+commit;
+alter table t1 add key (a);
+describe t1;
+Field	Type	Null	Key	Default	Extra
+a	int	YES	MUL	NULL	
+drop index a on t1;
+alter table t2 add key (a);
+describe t2;
+Field	Type	Null	Key	Default	Extra
+a	int	YES	MUL	NULL	
+drop index a on t2;
+truncate t1;
+truncate t2;
+# populate some data
+begin;
+insert into t1 values (1), (2), (3);
+insert into t2 values (1), (2), (3);
+select 't1', a from t1;
+t1	a
+t1	1
+t1	2
+t1	3
+select 't2', a from t2;
+t2	a
+t2	1
+t2	2
+t2	3
+commit;
+alter database test super_read_only = true;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci SUPER_READ_ONLY */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test read_only = false;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test read_only = true;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci READ_ONLY */ /*!80016 DEFAULT ENCRYPTION='N' */
+connection default;
+#
+# on non-user databases
+#
+alter database mysql read_only = true;
+ERROR HY000: Access to system schema 'mysql' is rejected.
+alter database information_schema read_only = true;
+ERROR 42000: Access denied for user 'root'@'localhost' to database 'information_schema'
+show create database information_schema;
+Database	Create Database
+information_schema	CREATE DATABASE `information_schema` /*!40100 DEFAULT CHARACTER SET utf8mb3 */ /*!80016 DEFAULT ENCRYPTION='N' */
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci READ_ONLY */ /*!80016 DEFAULT ENCRYPTION='N' */
+# restarting mysqld
+# restart
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci READ_ONLY */ /*!80016 DEFAULT ENCRYPTION='N' */
+create database test_db_opt;
+show create database test_db_opt;
+Database	Create Database
+test_db_opt	CREATE DATABASE `test_db_opt` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test_db_opt read_only = true;
+show create database test_db_opt;
+Database	Create Database
+test_db_opt	CREATE DATABASE `test_db_opt` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci READ_ONLY */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test_db_opt default charset = latin1;
+show create database test_db_opt;
+Database	Create Database
+test_db_opt	CREATE DATABASE `test_db_opt` /*!40100 DEFAULT CHARACTER SET latin1 READ_ONLY */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test_db_opt default collate = latin1_bin;
+show create database test_db_opt;
+Database	Create Database
+test_db_opt	CREATE DATABASE `test_db_opt` /*!40100 DEFAULT CHARACTER SET latin1 COLLATE latin1_bin READ_ONLY */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test_db_opt super_read_only  = true;
+show create database test_db_opt;
+Database	Create Database
+test_db_opt	CREATE DATABASE `test_db_opt` /*!40100 DEFAULT CHARACTER SET latin1 COLLATE latin1_bin SUPER_READ_ONLY */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test_db_opt super_read_only  = false;
+show create database test_db_opt;
+Database	Create Database
+test_db_opt	CREATE DATABASE `test_db_opt` /*!40100 DEFAULT CHARACTER SET latin1 COLLATE latin1_bin READ_ONLY */ /*!80016 DEFAULT ENCRYPTION='N' */
+drop database test_db_opt;
+create database test_db_opt /*!40100 DEFAULT CHARACTER SET latin1 READ_ONLY*/;
+show create database test_db_opt;
+Database	Create Database
+test_db_opt	CREATE DATABASE `test_db_opt` /*!40100 DEFAULT CHARACTER SET latin1 */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test_db_opt read_only;
+show create database test_db_opt;
+Database	Create Database
+test_db_opt	CREATE DATABASE `test_db_opt` /*!40100 DEFAULT CHARACTER SET latin1 */ /*!80016 DEFAULT ENCRYPTION='N' */
+drop database test_db_opt;
+create database test_db_opt /*!40100 DEFAULT CHARACTER SET latin1 SUPER_READ_ONLY*/;
+show create database test_db_opt;
+Database	Create Database
+test_db_opt	CREATE DATABASE `test_db_opt` /*!40100 DEFAULT CHARACTER SET latin1 */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test_db_opt super_read_only;
+show create database test_db_opt;
+Database	Create Database
+test_db_opt	CREATE DATABASE `test_db_opt` /*!40100 DEFAULT CHARACTER SET latin1 */ /*!80016 DEFAULT ENCRYPTION='N' */
+drop database test_db_opt;
+connection default;
+alter database test read_only = false;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */
+drop table t1, t2;
+drop user test@localhost;
diff --git a/mysql-test/r/db_read_only_debug.result b/mysql-test/r/db_read_only_debug.result
new file mode 100644
index 000000000000..1726fa2cb618
--- /dev/null
+++ b/mysql-test/r/db_read_only_debug.result
@@ -0,0 +1,22 @@
+create database test2;
+use test2;
+create table t (i int);
+select * from t;
+i
+show create database test2;
+Database	Create Database
+test2	CREATE DATABASE `test2` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */
+use test2;
+SET DEBUG_SYNC='ha_commit_trans_after_check_db_readonly SIGNAL parked WAIT_FOR go';
+insert into t values (1);;
+SET DEBUG_SYNC='now WAIT_FOR parked';
+alter database test2 super_read_only = true;;
+SET DEBUG_SYNC='now SIGNAL go';
+select * from t;
+i
+1
+show create database test2;
+Database	Create Database
+test2	CREATE DATABASE `test2` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci SUPER_READ_ONLY */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test2 read_only = false;
+drop database test2;
diff --git a/mysql-test/r/group_min_max.result b/mysql-test/r/group_min_max.result
index ce8db7079c06..1d22d780b3f2 100644
--- a/mysql-test/r/group_min_max.result
+++ b/mysql-test/r/group_min_max.result
@@ -2529,7 +2529,7 @@ FLUSH STATUS;
 CREATE TABLE t2 SELECT max(b), a FROM t1 GROUP BY a;
 SHOW STATUS LIKE 'handler_read__e%';
 Variable_name	Value
-Handler_read_key	24
+Handler_read_key	26
 Handler_read_next	2
 FLUSH STATUS;
 SELECT * FROM (SELECT max(b), a FROM t1 GROUP BY a) b;
diff --git a/mysql-test/r/information_schema_keywords.result b/mysql-test/r/information_schema_keywords.result
index 98222b50e251..95606caf965a 100644
--- a/mysql-test/r/information_schema_keywords.result
+++ b/mysql-test/r/information_schema_keywords.result
@@ -662,6 +662,7 @@ SUBJECT	0
 SUBPARTITION	0
 SUBPARTITIONS	0
 SUPER	0
+SUPER_READ_ONLY	0
 SUSPEND	0
 SWAPS	0
 SWITCHES	0
diff --git a/mysql-test/r/partition_locking.result b/mysql-test/r/partition_locking.result
index f25120f5dd85..83c784cad374 100644
--- a/mysql-test/r/partition_locking.result
+++ b/mysql-test/r/partition_locking.result
@@ -2739,46 +2739,46 @@ Handler_write	1
 FLUSH STATUS;
 CREATE VIEW v1_25 AS SELECT a, b FROM t1 PARTITION (p2, p5);
 VARIABLE_NAME	VARIABLE_VALUE
-Handler_commit	7
-Handler_external_lock	72
+Handler_commit	8
+Handler_external_lock	74
 Handler_prepare	4
-Handler_read_key	3
+Handler_read_key	4
 Handler_write	4
 # 34 locks (dictionary related)
 FLUSH STATUS;
 CREATE VIEW v1_25_check AS SELECT a, b FROM t1 PARTITION (p2, p5) t1_alias WITH CHECK OPTION;
 VARIABLE_NAME	VARIABLE_VALUE
-Handler_commit	7
-Handler_external_lock	72
+Handler_commit	8
+Handler_external_lock	74
 Handler_prepare	4
-Handler_read_key	3
+Handler_read_key	4
 Handler_write	4
 # 34 locks (dictionary related)
 FLUSH STATUS;
 CREATE VIEW v1_9 AS SELECT a, b FROM t1 WHERE a = 9;
 VARIABLE_NAME	VARIABLE_VALUE
-Handler_commit	7
-Handler_external_lock	72
+Handler_commit	8
+Handler_external_lock	74
 Handler_prepare	4
-Handler_read_key	3
+Handler_read_key	4
 Handler_write	4
 # 34 locks (dictionary related)
 FLUSH STATUS;
 CREATE VIEW v1_9_check AS SELECT a, b FROM t1 WHERE a = 9 WITH CHECK OPTION;
 VARIABLE_NAME	VARIABLE_VALUE
-Handler_commit	7
-Handler_external_lock	72
+Handler_commit	8
+Handler_external_lock	74
 Handler_prepare	4
-Handler_read_key	3
+Handler_read_key	4
 Handler_write	4
 # 34 locks (dictionary related)
 FLUSH STATUS;
 CREATE VIEW v1_all AS SELECT a, b FROM t1;
 VARIABLE_NAME	VARIABLE_VALUE
-Handler_commit	7
-Handler_external_lock	72
+Handler_commit	8
+Handler_external_lock	74
 Handler_prepare	4
-Handler_read_key	3
+Handler_read_key	4
 Handler_write	4
 # 34 locks (dictionary related)
 SELECT TABLE_NAME, CHECK_OPTION, IS_UPDATABLE, VIEW_DEFINITION
@@ -2966,11 +2966,11 @@ DROP VIEW v1_25, v1_25_check;
 FLUSH STATUS;
 CREATE TABLE t3 SELECT a, b FROM t1 WHERE a IN (0, 1, 13, 113, 26);
 VARIABLE_NAME	VARIABLE_VALUE
-Handler_commit	13
-Handler_external_lock	164
+Handler_commit	14
+Handler_external_lock	166
 Handler_prepare	4
 Handler_read_first	3
-Handler_read_key	25
+Handler_read_key	26
 Handler_read_next	12
 Handler_update	12
 Handler_write	18
@@ -2984,10 +2984,10 @@ DROP TABLE t3;
 FLUSH STATUS;
 CREATE TABLE t3 SELECT a, b FROM t1 WHERE b LIKE 'First%';
 VARIABLE_NAME	VARIABLE_VALUE
-Handler_commit	13
-Handler_external_lock	164
+Handler_commit	14
+Handler_external_lock	166
 Handler_prepare	4
-Handler_read_key	35
+Handler_read_key	36
 Handler_read_next	5
 Handler_update	12
 Handler_write	19
@@ -4630,11 +4630,11 @@ INSERT INTO t2 VALUES (0, "Zero"), (1, "One"), (2, "Two"),
 FLUSH STATUS;
 ALTER TABLE t2 EXCHANGE PARTITION p1 WITH TABLE t1;
 VARIABLE_NAME	VARIABLE_VALUE
-Handler_commit	23
-Handler_external_lock	144
+Handler_commit	24
+Handler_external_lock	146
 Handler_prepare	4
 Handler_read_first	1
-Handler_read_key	72
+Handler_read_key	73
 Handler_read_next	16
 Handler_read_rnd_next	4
 Handler_update	36
@@ -4847,11 +4847,11 @@ INSERT INTO t1 VALUES (1), (3), (9), (2), (8), (7);
 FLUSH STATUS;
 CREATE TABLE t2 SELECT * FROM t1 PARTITION (p1, p2);
 VARIABLE_NAME	VARIABLE_VALUE
-Handler_commit	12
-Handler_external_lock	162
+Handler_commit	13
+Handler_external_lock	164
 Handler_prepare	4
 Handler_read_first	2
-Handler_read_key	21
+Handler_read_key	22
 Handler_read_rnd_next	4
 Handler_update	10
 Handler_write	16
@@ -4865,11 +4865,11 @@ DROP TABLE t2;
 FLUSH STATUS;
 CREATE TABLE t2 SELECT * FROM t1 WHERE a IN (1, 3, 9);
 VARIABLE_NAME	VARIABLE_VALUE
-Handler_commit	12
-Handler_external_lock	162
+Handler_commit	13
+Handler_external_lock	164
 Handler_prepare	4
 Handler_read_first	2
-Handler_read_key	21
+Handler_read_key	22
 Handler_read_rnd_next	4
 Handler_update	10
 Handler_write	15
diff --git a/mysql-test/r/slow_log_extra.result b/mysql-test/r/slow_log_extra.result
index 1577017041c8..dd726de39e9c 100644
--- a/mysql-test/r/slow_log_extra.result
+++ b/mysql-test/r/slow_log_extra.result
@@ -68,7 +68,7 @@ DROP TABLE islow;
 # verify the rest are correct.
 #
 Rows_sent: 0  Rows_examined: 0 Thread_id: 0 Errno: 0 Killed: 0 Bytes_received: 33 Bytes_sent: 11 Read_first: 0 Read_last: 0 Read_key: 5 Read_next: 0 Read_prev: 0 Read_rnd: 0 Read_rnd_next: 0 RocksDB_key_skipped: 0 RocksDB_del_skipped: 0 Sort_merge_passes: 0 Sort_range_count: 0 Sort_rows: 0 Sort_scan_count: 0 Created_tmp_disk_tables: 0 Created_tmp_tables: 0
-Rows_sent: 0  Rows_examined: 0 Thread_id: 0 Errno: 0 Killed: 0 Bytes_received: 46 Bytes_sent: 11 Read_first: 0 Read_last: 0 Read_key: 18 Read_next: 0 Read_prev: 0 Read_rnd: 0 Read_rnd_next: 0 RocksDB_key_skipped: 0 RocksDB_del_skipped: 0 Sort_merge_passes: 0 Sort_range_count: 0 Sort_rows: 0 Sort_scan_count: 0 Created_tmp_disk_tables: 0 Created_tmp_tables: 0
+Rows_sent: 0  Rows_examined: 0 Thread_id: 0 Errno: 0 Killed: 0 Bytes_received: 46 Bytes_sent: 11 Read_first: 0 Read_last: 0 Read_key: 19 Read_next: 0 Read_prev: 0 Read_rnd: 0 Read_rnd_next: 0 RocksDB_key_skipped: 0 RocksDB_del_skipped: 0 Sort_merge_passes: 0 Sort_range_count: 0 Sort_rows: 0 Sort_scan_count: 0 Created_tmp_disk_tables: 0 Created_tmp_tables: 0
 Rows_sent: 0  Rows_examined: 0 Thread_id: 0 Errno: 0 Killed: 0 Bytes_received: 70 Bytes_sent: 50 Read_first: 0 Read_last: 0 Read_key: 11 Read_next: 9 Read_prev: 0 Read_rnd: 0 Read_rnd_next: 0 RocksDB_key_skipped: 0 RocksDB_del_skipped: 0 Sort_merge_passes: 0 Sort_range_count: 0 Sort_rows: 0 Sort_scan_count: 0 Created_tmp_disk_tables: 0 Created_tmp_tables: 0
 Rows_sent: 8  Rows_examined: 8 Thread_id: 0 Errno: 0 Killed: 0 Bytes_received: 26 Bytes_sent: 106 Read_first: 1 Read_last: 0 Read_key: 1 Read_next: 0 Read_prev: 0 Read_rnd: 0 Read_rnd_next: 9 RocksDB_key_skipped: 0 RocksDB_del_skipped: 0 Sort_merge_passes: 0 Sort_range_count: 0 Sort_rows: 0 Sort_scan_count: 0 Created_tmp_disk_tables: 0 Created_tmp_tables: 0
 Rows_sent: 8  Rows_examined: 8 Thread_id: 0 Errno: 0 Killed: 0 Bytes_received: 26 Bytes_sent: 106 Read_first: 1 Read_last: 0 Read_key: 1 Read_next: 0 Read_prev: 0 Read_rnd: 0 Read_rnd_next: 9 RocksDB_key_skipped: 0 RocksDB_del_skipped: 0 Sort_merge_passes: 0 Sort_range_count: 0 Sort_rows: 0 Sort_scan_count: 0 Created_tmp_disk_tables: 0 Created_tmp_tables: 0
diff --git a/mysql-test/r/slow_log_extra_myisam.result b/mysql-test/r/slow_log_extra_myisam.result
index 241a3a8d7979..76e0c7fabf12 100644
--- a/mysql-test/r/slow_log_extra_myisam.result
+++ b/mysql-test/r/slow_log_extra_myisam.result
@@ -61,9 +61,9 @@ DROP TABLE mslow;
 #
 Rows_sent: 0  Rows_examined: 0 Thread_id: 0 Errno: 0 Killed: 0 Bytes_received: 33 Bytes_sent: 11 Read_first: 0 Read_last: 0 Read_key: 5 Read_next: 0 Read_prev: 0 Read_rnd: 0 Read_rnd_next: 0 RocksDB_key_skipped: 0 RocksDB_del_skipped: 0 Sort_merge_passes: 0 Sort_range_count: 0 Sort_rows: 0 Sort_scan_count: 0 Created_tmp_disk_tables: 0 Created_tmp_tables: 0
 Rows_sent: 0  Rows_examined: 0 Thread_id: 0 Errno: 0 Killed: 0 Bytes_received: 33 Bytes_sent: 11 Read_first: 0 Read_last: 0 Read_key: 5 Read_next: 0 Read_prev: 0 Read_rnd: 0 Read_rnd_next: 0 RocksDB_key_skipped: 0 RocksDB_del_skipped: 0 Sort_merge_passes: 0 Sort_range_count: 0 Sort_rows: 0 Sort_scan_count: 0 Created_tmp_disk_tables: 0 Created_tmp_tables: 0
-Rows_sent: 0  Rows_examined: 0 Thread_id: 0 Errno: 0 Killed: 0 Bytes_received: 46 Bytes_sent: 11 Read_first: 0 Read_last: 0 Read_key: 18 Read_next: 0 Read_prev: 0 Read_rnd: 0 Read_rnd_next: 0 RocksDB_key_skipped: 0 RocksDB_del_skipped: 0 Sort_merge_passes: 0 Sort_range_count: 0 Sort_rows: 0 Sort_scan_count: 0 Created_tmp_disk_tables: 0 Created_tmp_tables: 0
+Rows_sent: 0  Rows_examined: 0 Thread_id: 0 Errno: 0 Killed: 0 Bytes_received: 46 Bytes_sent: 11 Read_first: 0 Read_last: 0 Read_key: 19 Read_next: 0 Read_prev: 0 Read_rnd: 0 Read_rnd_next: 0 RocksDB_key_skipped: 0 RocksDB_del_skipped: 0 Sort_merge_passes: 0 Sort_range_count: 0 Sort_rows: 0 Sort_scan_count: 0 Created_tmp_disk_tables: 0 Created_tmp_tables: 0
 Rows_sent: 0  Rows_examined: 0 Thread_id: 0 Errno: 0 Killed: 0 Bytes_received: 70 Bytes_sent: 50 Read_first: 0 Read_last: 0 Read_key: 11 Read_next: 9 Read_prev: 0 Read_rnd: 0 Read_rnd_next: 0 RocksDB_key_skipped: 0 RocksDB_del_skipped: 0 Sort_merge_passes: 0 Sort_range_count: 0 Sort_rows: 0 Sort_scan_count: 0 Created_tmp_disk_tables: 0 Created_tmp_tables: 0
-Rows_sent: 0  Rows_examined: 0 Thread_id: 0 Errno: 0 Killed: 0 Bytes_received: 46 Bytes_sent: 11 Read_first: 0 Read_last: 0 Read_key: 12 Read_next: 1 Read_prev: 0 Read_rnd: 0 Read_rnd_next: 0 RocksDB_key_skipped: 0 RocksDB_del_skipped: 0 Sort_merge_passes: 0 Sort_range_count: 0 Sort_rows: 0 Sort_scan_count: 0 Created_tmp_disk_tables: 0 Created_tmp_tables: 0
+Rows_sent: 0  Rows_examined: 0 Thread_id: 0 Errno: 0 Killed: 0 Bytes_received: 46 Bytes_sent: 11 Read_first: 0 Read_last: 0 Read_key: 13 Read_next: 1 Read_prev: 0 Read_rnd: 0 Read_rnd_next: 0 RocksDB_key_skipped: 0 RocksDB_del_skipped: 0 Sort_merge_passes: 0 Sort_range_count: 0 Sort_rows: 0 Sort_scan_count: 0 Created_tmp_disk_tables: 0 Created_tmp_tables: 0
 Rows_sent: 0  Rows_examined: 0 Thread_id: 0 Errno: 0 Killed: 0 Bytes_received: 70 Bytes_sent: 50 Read_first: 0 Read_last: 0 Read_key: 1 Read_next: 0 Read_prev: 0 Read_rnd: 0 Read_rnd_next: 0 RocksDB_key_skipped: 0 RocksDB_del_skipped: 0 Sort_merge_passes: 0 Sort_range_count: 0 Sort_rows: 0 Sort_scan_count: 0 Created_tmp_disk_tables: 0 Created_tmp_tables: 0
 Rows_sent: 8  Rows_examined: 8 Thread_id: 0 Errno: 0 Killed: 0 Bytes_received: 26 Bytes_sent: 106 Read_first: 1 Read_last: 0 Read_key: 1 Read_next: 0 Read_prev: 0 Read_rnd: 0 Read_rnd_next: 9 RocksDB_key_skipped: 0 RocksDB_del_skipped: 0 Sort_merge_passes: 0 Sort_range_count: 0 Sort_rows: 0 Sort_scan_count: 0 Created_tmp_disk_tables: 0 Created_tmp_tables: 0
 Rows_sent: 8  Rows_examined: 8 Thread_id: 0 Errno: 0 Killed: 0 Bytes_received: 26 Bytes_sent: 106 Read_first: 1 Read_last: 0 Read_key: 1 Read_next: 0 Read_prev: 0 Read_rnd: 0 Read_rnd_next: 9 RocksDB_key_skipped: 0 RocksDB_del_skipped: 0 Sort_merge_passes: 0 Sort_range_count: 0 Sort_rows: 0 Sort_scan_count: 0 Created_tmp_disk_tables: 0 Created_tmp_tables: 0
diff --git a/mysql-test/r/subquery_all.result b/mysql-test/r/subquery_all.result
index 34ab03b8784e..d7e0ca9e6df8 100644
--- a/mysql-test/r/subquery_all.result
+++ b/mysql-test/r/subquery_all.result
@@ -4678,7 +4678,7 @@ oref	a	Z
 show status like '%Handler_read%';
 Variable_name	Value
 Handler_read_first	5
-Handler_read_key	8
+Handler_read_key	10
 Handler_read_last	0
 Handler_read_next	0
 Handler_read_prev	0
diff --git a/mysql-test/r/subquery_all_bka.result b/mysql-test/r/subquery_all_bka.result
index e349e6ae9f47..599490198be2 100644
--- a/mysql-test/r/subquery_all_bka.result
+++ b/mysql-test/r/subquery_all_bka.result
@@ -4679,7 +4679,7 @@ oref	a	Z
 show status like '%Handler_read%';
 Variable_name	Value
 Handler_read_first	5
-Handler_read_key	8
+Handler_read_key	10
 Handler_read_last	0
 Handler_read_next	0
 Handler_read_prev	0
diff --git a/mysql-test/r/subquery_all_bka_nobnl.result b/mysql-test/r/subquery_all_bka_nobnl.result
index e19be189da6d..dbeb99ea3c7a 100644
--- a/mysql-test/r/subquery_all_bka_nobnl.result
+++ b/mysql-test/r/subquery_all_bka_nobnl.result
@@ -4679,7 +4679,7 @@ oref	a	Z
 show status like '%Handler_read%';
 Variable_name	Value
 Handler_read_first	5
-Handler_read_key	8
+Handler_read_key	10
 Handler_read_last	0
 Handler_read_next	0
 Handler_read_prev	0
diff --git a/mysql-test/r/subquery_nomat_nosj.result b/mysql-test/r/subquery_nomat_nosj.result
index e5f9ecd46781..d2d67e7dd78b 100644
--- a/mysql-test/r/subquery_nomat_nosj.result
+++ b/mysql-test/r/subquery_nomat_nosj.result
@@ -4678,7 +4678,7 @@ oref	a	Z
 show status like '%Handler_read%';
 Variable_name	Value
 Handler_read_first	5
-Handler_read_key	8
+Handler_read_key	10
 Handler_read_last	0
 Handler_read_next	0
 Handler_read_prev	0
diff --git a/mysql-test/r/subquery_nomat_nosj_bka.result b/mysql-test/r/subquery_nomat_nosj_bka.result
index 7eac2da34b1a..c302f604c21d 100644
--- a/mysql-test/r/subquery_nomat_nosj_bka.result
+++ b/mysql-test/r/subquery_nomat_nosj_bka.result
@@ -4679,7 +4679,7 @@ oref	a	Z
 show status like '%Handler_read%';
 Variable_name	Value
 Handler_read_first	5
-Handler_read_key	8
+Handler_read_key	10
 Handler_read_last	0
 Handler_read_next	0
 Handler_read_prev	0
diff --git a/mysql-test/r/subquery_nomat_nosj_bka_nobnl.result b/mysql-test/r/subquery_nomat_nosj_bka_nobnl.result
index 854756c80d2d..395945a8a9c3 100644
--- a/mysql-test/r/subquery_nomat_nosj_bka_nobnl.result
+++ b/mysql-test/r/subquery_nomat_nosj_bka_nobnl.result
@@ -4679,7 +4679,7 @@ oref	a	Z
 show status like '%Handler_read%';
 Variable_name	Value
 Handler_read_first	5
-Handler_read_key	8
+Handler_read_key	10
 Handler_read_last	0
 Handler_read_next	0
 Handler_read_prev	0
diff --git a/mysql-test/r/subquery_none.result b/mysql-test/r/subquery_none.result
index 0a5d11780444..1c0bf315b769 100644
--- a/mysql-test/r/subquery_none.result
+++ b/mysql-test/r/subquery_none.result
@@ -4677,7 +4677,7 @@ oref	a	Z
 show status like '%Handler_read%';
 Variable_name	Value
 Handler_read_first	5
-Handler_read_key	8
+Handler_read_key	10
 Handler_read_last	0
 Handler_read_next	0
 Handler_read_prev	0
diff --git a/mysql-test/r/subquery_none_bka.result b/mysql-test/r/subquery_none_bka.result
index 4b464ff8481b..ea037db5d8de 100644
--- a/mysql-test/r/subquery_none_bka.result
+++ b/mysql-test/r/subquery_none_bka.result
@@ -4678,7 +4678,7 @@ oref	a	Z
 show status like '%Handler_read%';
 Variable_name	Value
 Handler_read_first	5
-Handler_read_key	8
+Handler_read_key	10
 Handler_read_last	0
 Handler_read_next	0
 Handler_read_prev	0
diff --git a/mysql-test/r/subquery_none_bka_nobnl.result b/mysql-test/r/subquery_none_bka_nobnl.result
index 4e29caf163c4..833e5f598fa3 100644
--- a/mysql-test/r/subquery_none_bka_nobnl.result
+++ b/mysql-test/r/subquery_none_bka_nobnl.result
@@ -4678,7 +4678,7 @@ oref	a	Z
 show status like '%Handler_read%';
 Variable_name	Value
 Handler_read_first	5
-Handler_read_key	8
+Handler_read_key	10
 Handler_read_last	0
 Handler_read_next	0
 Handler_read_prev	0
diff --git a/mysql-test/t/db_read_only.test b/mysql-test/t/db_read_only.test
new file mode 100644
index 000000000000..1dac23582e0b
--- /dev/null
+++ b/mysql-test/t/db_read_only.test
@@ -0,0 +1,222 @@
+# Test per-database read-only attribute
+connection default;
+
+--disable_warnings
+drop table if exists t1, t2;
+drop database if exists test2;
+--enable_warnings
+
+--disable_query_log
+let $MYSQLD_DATADIR= `select @@datadir`;
+--enable_query_log
+
+--error 1
+file_exists $MYSQLD_DATADIR/mysql/db.opt;
+--error 1
+file_exists $MYSQLD_DATADIR/test/db.opt;
+
+CREATE USER test@localhost;
+grant CREATE, SELECT, INSERT, UPDATE, DROP on *.* to test@localhost;
+
+--echo connect (con1,localhost,test,,test);
+connect (con1,localhost,test,,test);
+--echo connect (root2,localhost,root,,test);
+connect (root2,localhost,root,,test);
+
+--echo # turn off autocommit in con1
+--echo connection con1;
+connection con1;
+set autocommit = 0;
+select @@autocommit;
+
+--echo # turn off autocommit in root2
+--echo connection root2;
+connection root2;
+set autocommit = 0;
+select @@autocommit;
+
+create table t1 (a int) engine = innodb;
+create table t2 (a int) engine = innodb;
+
+--echo #
+--echo # read_only = false
+--echo # both regular and super users can write
+--echo #
+--echo connection con1;
+connection con1;
+show create database test;
+
+--source include/db_read_only_off.inc
+
+--echo connection root2;
+connection root2;
+show create database test;
+
+--source include/db_read_only_off.inc
+
+--echo #
+--echo # read_only
+--echo #
+# default
+--echo connection default;
+connection default;
+
+--echo # turn off read_only initially
+alter database test read_only = false;
+show create database test;
+
+# con1
+--echo connection con1;
+connection con1;
+show create database test;
+
+--echo # simulate long running transactions
+begin;
+insert into t1 values (4), (5), (6);
+insert into t2 values (4), (5), (6);
+select 't1', a from t1;
+select 't2', a from t2;
+
+# default
+--echo connection default;
+connection default;
+
+--echo # turn on read_only in the middle of transaction (con1)
+set lock_wait_timeout = 1;
+--error ER_LOCK_WAIT_TIMEOUT
+alter database test read_only = true;
+show create database test;
+--echo connection con1;
+connection con1;
+# Rollback the transaction.
+rollback;
+
+--echo #
+--echo # super_read_only
+--echo #
+# default
+--echo connection default;
+connection default;
+
+--echo # turn off read_only initially
+alter database test read_only = false;
+show create database test;
+
+# root2
+--echo connection root2;
+connection root2;
+show create database test;
+
+--echo # simulate long running transactions
+begin;
+insert into t1 values (4), (5), (6);
+insert into t2 values (4), (5), (6);
+select 't1', a from t1;
+select 't2', a from t2;
+
+# default
+--echo connection default;
+connection default;
+
+--echo # now turn on super read_only in the middle of transaction (root2)
+set lock_wait_timeout = 1;
+--error ER_LOCK_WAIT_TIMEOUT
+alter database test super_read_only = true;
+show create database test;
+--echo connection root2;
+connection root2;
+# Rollback the transaction.
+rollback;
+alter database test super_read_only = true;
+
+# root2
+--echo connection root2;
+connection root2;
+show create database test;
+
+--source include/db_read_only_on.inc
+
+--echo #
+--echo # changing db_read_only in root2 session
+--echo #
+alter database test super_read_only = false;
+show create database test;
+
+--echo # super_read_only turned into read_only
+--source include/db_read_only_off.inc
+
+alter database test super_read_only = true;
+show create database test;
+
+alter database test read_only = false;
+show create database test;
+
+alter database test read_only = true;
+show create database test;
+
+--echo connection default;
+connection default;
+
+--echo #
+--echo # on non-user databases
+--echo #
+--error ER_NO_SYSTEM_SCHEMA_ACCESS
+alter database mysql read_only = true;
+
+--error ER_DBACCESS_DENIED_ERROR
+alter database information_schema read_only = true;
+show create database information_schema;
+
+#
+# read_only survives restart mysqld
+#
+show create database test;
+--echo # restarting mysqld
+--source include/restart_mysqld.inc
+show create database test;
+
+#
+# Alter database options will not overwrite un-altered optinos
+#
+create database test_db_opt;
+show create database test_db_opt;
+alter database test_db_opt read_only = true;
+show create database test_db_opt;
+alter database test_db_opt default charset = latin1;
+show create database test_db_opt;
+alter database test_db_opt default collate = latin1_bin;
+show create database test_db_opt;
+alter database test_db_opt super_read_only  = true;
+show create database test_db_opt;
+alter database test_db_opt super_read_only  = false;
+show create database test_db_opt;
+drop database test_db_opt;
+
+#
+# [super_]read_only is ignored in create database comment
+# and is ignored in alter without assignment
+#
+create database test_db_opt /*!40100 DEFAULT CHARACTER SET latin1 READ_ONLY*/;
+show create database test_db_opt;
+alter database test_db_opt read_only;
+show create database test_db_opt;
+drop database test_db_opt;
+
+create database test_db_opt /*!40100 DEFAULT CHARACTER SET latin1 SUPER_READ_ONLY*/;
+show create database test_db_opt;
+alter database test_db_opt super_read_only;
+show create database test_db_opt;
+drop database test_db_opt;
+
+#
+# cleanup
+#
+--echo connection default;
+connection default;
+alter database test read_only = false;
+show create database test;
+
+disconnect con1;
+disconnect root2;
+drop table t1, t2;
+drop user test@localhost;
diff --git a/mysql-test/t/db_read_only_debug.test b/mysql-test/t/db_read_only_debug.test
new file mode 100644
index 000000000000..f81d76fc4b6f
--- /dev/null
+++ b/mysql-test/t/db_read_only_debug.test
@@ -0,0 +1,42 @@
+--source include/have_debug.inc
+--source include/have_debug_sync.inc
+
+#
+# Check that locking is correct to ensure that transactions cannot finish
+# committing after DB read only was set.
+#
+
+create database test2;
+use test2;
+
+create table t (i int);
+select * from t;
+
+connect (con1,localhost,root,,test);
+show create database test2;
+use test2;
+SET DEBUG_SYNC='ha_commit_trans_after_check_db_readonly SIGNAL parked WAIT_FOR go';
+--send insert into t values (1);
+
+connect (con2,localhost,root,,test);
+SET DEBUG_SYNC='now WAIT_FOR parked';
+--send alter database test2 super_read_only = true;
+
+connection default;
+let $wait_condition= SELECT count(*) = 1 FROM INFORMATION_SCHEMA.PROCESSLIST where STATE = 'Waiting for table metadata lock' AND ID != CONNECTION_ID();
+--source include/wait_condition.inc
+SET DEBUG_SYNC='now SIGNAL go';
+
+connection con1;
+# The insert must succeed
+--reap
+select * from t;
+
+connection con2;
+# The alter also succeeds, after the insert finishes
+--reap
+
+show create database test2;
+
+alter database test2 read_only = false;
+drop database test2;
diff --git a/share/messages_to_clients.txt b/share/messages_to_clients.txt
index 6c50937fe2ee..9756893e1cb2 100644
--- a/share/messages_to_clients.txt
+++ b/share/messages_to_clients.txt
@@ -10081,8 +10081,8 @@ ER_PLACEHOLDER_50025
 ER_PLACEHOLDER_50026
   eng "Placeholder"
 
-ER_PLACEHOLDER_50027
-  eng "Placeholder"
+ER_DB_READ_ONLY
+  eng "Database '%s' is in read-only mode. %s"
 
 ER_PLACEHOLDER_50028
   eng "Placeholder"
@@ -10102,8 +10102,8 @@ ER_PLACEHOLDER_50032
 ER_PLACEHOLDER_50033
   eng "Placeholder"
 
-ER_PLACEHOLDER_50034
-  eng "Placeholder"
+ER_UNKNOWN_DB_READ_ONLY
+  eng "Unknown db read only option: '%-.64s'"
 
 ER_PLACEHOLDER_50035
   eng "Placeholder"
diff --git a/sql/dd/impl/types/schema_impl.cc b/sql/dd/impl/types/schema_impl.cc
index 2c399b2ad70e..f8854ebe6ef5 100644
--- a/sql/dd/impl/types/schema_impl.cc
+++ b/sql/dd/impl/types/schema_impl.cc
@@ -71,7 +71,10 @@ using dd::tables::Tables;
 
 namespace dd {
 
-static const std::set<String_type> default_valid_option_keys = {"read_only"};
+static constexpr auto fb_read_only_options_key = "read_only";
+static constexpr auto mysql_db_read_only_options_key = "mysql_db_read_only";
+static const std::set<String_type> default_valid_option_keys = {
+    mysql_db_read_only_options_key, fb_read_only_options_key};
 
 ///////////////////////////////////////////////////////////////////////////
 // Schema_impl implementation.
@@ -96,7 +99,8 @@ bool Schema_impl::validate() const {
 
 bool Schema_impl::read_only() const {
   bool state = false;
-  if (options().exists("read_only") && options().get("read_only", &state)) {
+  if (options().exists(mysql_db_read_only_options_key) &&
+      options().get(mysql_db_read_only_options_key, &state)) {
     return false;
   }
   return state;
@@ -105,11 +109,33 @@ bool Schema_impl::read_only() const {
 /////////////////////////////////////////////////////////////////////////
 
 void Schema_impl::set_read_only(bool state) {
-  options().set("read_only", state);
+  options().set(mysql_db_read_only_options_key, state);
 }
 
 ///////////////////////////////////////////////////////////////////////////
 
+void Schema_impl::set_db_read_only(int state) {
+  options().set(fb_read_only_options_key, state);
+}
+
+int Schema_impl::get_db_read_only() const {
+  DBUG_TRACE;
+  int val = 0;
+
+  if (!options().exists(fb_read_only_options_key)) {
+    // No option set. Assume not read only.
+    return DB_READ_ONLY_NO;
+  }
+
+  if (options().get(fb_read_only_options_key, &val) || val < DB_READ_ONLY_NO ||
+      val > DB_READ_ONLY_SUPER) {
+    my_error(ER_UNKNOWN_DB_READ_ONLY, MYF(0), std::to_string(val).c_str());
+    return DB_READ_ONLY_NO;
+  }
+
+  return val;
+}
+
 bool Schema_impl::restore_attributes(const Raw_record &r) {
   restore_id(r, Schemata::FIELD_ID);
   restore_name(r, Schemata::FIELD_NAME);
diff --git a/sql/dd/impl/types/schema_impl.h b/sql/dd/impl/types/schema_impl.h
index 26e42890bd54..59ad8ae1e059 100644
--- a/sql/dd/impl/types/schema_impl.h
+++ b/sql/dd/impl/types/schema_impl.h
@@ -105,6 +105,8 @@ class Schema_impl : public Entity_object_impl, public Schema {
   /////////////////////////////////////////////////////////////////////////
   bool read_only() const override;
   void set_read_only(bool state) override;
+  void set_db_read_only(int state) override;
+  int get_db_read_only() const override;
 
   /////////////////////////////////////////////////////////////////////////
   // created
diff --git a/sql/dd/types/schema.h b/sql/dd/types/schema.h
index cd03b621022f..3c697da390f9 100644
--- a/sql/dd/types/schema.h
+++ b/sql/dd/types/schema.h
@@ -112,6 +112,8 @@ class Schema : virtual public Entity_object {
   /////////////////////////////////////////////////////////////////////////
   virtual bool read_only() const = 0;
   virtual void set_read_only(bool state) = 0;
+  virtual void set_db_read_only(int state) = 0;
+  virtual int get_db_read_only() const = 0;
 
   /////////////////////////////////////////////////////////////////////////
   // created
diff --git a/sql/handler.cc b/sql/handler.cc
index 12ae59360cc9..0bca7e9e0ac1 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -111,6 +111,7 @@
 #include "sql/sql_base.h"  // free_io_cache
 #include "sql/sql_bitmap.h"
 #include "sql/sql_class.h"
+#include "sql/sql_db.h"  // is_thd_db_read_only_by_name
 #include "sql/sql_error.h"
 #include "sql/sql_lex.h"
 #include "sql/sql_parse.h"   // check_stack_overrun
@@ -1527,6 +1528,28 @@ std::pair<int, bool> commit_owned_gtids(THD *thd, bool all) {
   return std::make_pair(error, need_clear_owned_gtid);
 }
 
+static bool check_db_readonly(THD *thd) {
+  // This is a write transaction and we have db_read_only on some database.
+  // Since this transaction may be an explicit "commit" which doesn't have any
+  // "current" table in the statement, we need to check the list of MDL lock
+  // object to get the list of databases we obtained metadata locks on. There
+  // may be false positive cases that the read-only DB in the list is not
+  // actually modified in the transaction.  We do not differentiate such false
+  // positives.
+  //
+  MDL_DB_Name_List db_names;
+  thd->mdl_context.get_locked_object_db_names(db_names);
+  for (auto db_name : db_names) {
+    if (is_thd_db_read_only_by_name(thd, db_name.c_str())) {
+      my_error(ER_DB_READ_ONLY, MYF(0), db_name.c_str(),
+               "Transaction was rolled back.");
+      return true;
+    }
+  }
+
+  return false;
+}
+
 /**
   @param[in] thd                       Thread handle.
   @param[in] all                       Session transaction if true, statement
@@ -1700,13 +1723,16 @@ int ha_commit_trans(THD *thd, bool all, bool ignore_global_read_lock) {
       DEBUG_SYNC(thd, "ha_commit_trans_after_acquire_commit_lock");
     }
 
-    if (rw_trans && stmt_has_updated_trans_table(ha_info) &&
-        check_readonly(thd, true)) {
+    if (rw_trans &&
+        ((stmt_has_updated_trans_table(ha_info) && check_readonly(thd, true)) ||
+         check_db_readonly(thd))) {
       ha_rollback_trans(thd, all);
       error = 1;
       goto end;
     }
 
+    DEBUG_SYNC(thd, "ha_commit_trans_after_check_db_readonly");
+
     if (!trn_ctx->no_2pc(trx_scope) && (trn_ctx->rw_ha_count(trx_scope) > 1))
       error = tc_log->prepare(thd, all);
   }
diff --git a/sql/handler.h b/sql/handler.h
index 9755711bef4b..beddba8d5a01 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -3066,6 +3066,13 @@ enum enum_stats_auto_recalc : int {
   HA_STATS_AUTO_RECALC_OFF
 };
 
+enum enum_db_read_only : int {
+  DB_READ_ONLY_NULL = 0,
+  DB_READ_ONLY_NO = 1,
+  DB_READ_ONLY_YES = 2,
+  DB_READ_ONLY_SUPER = 3
+};
+
 /**
   Struct to hold information about the table that should be created.
  */
@@ -3073,6 +3080,7 @@ struct HA_CREATE_INFO {
   const CHARSET_INFO *table_charset{nullptr};
   const CHARSET_INFO *default_table_charset{nullptr};
   bool schema_read_only{false};
+  enum enum_db_read_only db_read_only { DB_READ_ONLY_NULL };
   LEX_STRING connect_string{nullptr, 0};
   const char *password{nullptr};
   const char *tablespace{nullptr};
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index c9096e4b3a0c..12bc3aa8b2e9 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -4653,9 +4653,9 @@ String *Item_func_get_dd_schema_options::val_str(String *str) {
   ptr = option_buff;
 
   // Print READ ONLY clause if set.
-  if (p->exists("read_only")) {
+  if (p->exists("mysql_db_read_only")) {
     dd::String_type schema_read_only;
-    p->get("read_only", &schema_read_only);
+    p->get("mysql_db_read_only", &schema_read_only);
     assert(schema_read_only == "0" || schema_read_only == "1");
     if (schema_read_only == "1") ptr = my_stpcpy(ptr, " READ ONLY=1");
   }
diff --git a/sql/lex.h b/sql/lex.h
index d2c35a9c6690..f0fb2d2eaa3f 100644
--- a/sql/lex.h
+++ b/sql/lex.h
@@ -730,6 +730,7 @@ static const SYMBOL symbols[] = {
     {SYM("SUBPARTITION", SUBPARTITION_SYM)},
     {SYM("SUBPARTITIONS", SUBPARTITIONS_SYM)},
     {SYM("SUPER", SUPER_SYM)},
+    {SYM("SUPER_READ_ONLY", SUPER_READ_ONLY_SYM)},
     {SYM("SUSPEND", SUSPEND_SYM)},
     {SYM("SWAPS", SWAPS_SYM)},
     {SYM("SWITCHES", SWITCHES_SYM)},
diff --git a/sql/mdl.cc b/sql/mdl.cc
index 2ec8c9e3918c..4f8e3552fe4c 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -4663,6 +4663,21 @@ void MDL_context::set_transaction_duration_for_all_locks() {
   m_ticket_store.move_explicit_to_transaction_duration();
 }
 
+void MDL_context::get_locked_object_db_names(MDL_DB_Name_List &list) {
+  DBUG_ENTER("MDL_context::get_locked_object_db_names");
+  MDL_ticket_store::List_iterator it =
+      m_ticket_store.list_iterator(MDL_TRANSACTION);
+
+  for (MDL_ticket *ticket = it++; ticket != nullptr; ticket = it++) {
+    MDL_lock *lock = ticket->m_lock;
+    DBUG_PRINT("enter",
+               ("db=%s name=%s", lock->key.db_name(), lock->key.name()));
+    list.insert(std::string(lock->key.db_name()));
+  }
+
+  DBUG_VOID_RETURN;
+}
+
 size_t MDL_ticket_store::Hash::operator()(const MDL_key *k) const {
   return static_cast<size_t>(murmur3_32(k->ptr(), k->length(), 0));
 }
diff --git a/sql/mdl.h b/sql/mdl.h
index 1c5af58249af..fd8db4764962 100644
--- a/sql/mdl.h
+++ b/sql/mdl.h
@@ -27,7 +27,9 @@
 #include <sys/types.h>
 #include <algorithm>
 #include <new>
+#include <string>
 #include <unordered_map>
+#include <unordered_set>
 
 #include "m_string.h"
 #include "my_alloc.h"
@@ -1411,6 +1413,8 @@ typedef I_P_List<MDL_request,
                  I_P_List_counter>
     MDL_request_list;
 
+typedef std::unordered_set<std::string> MDL_DB_Name_List;
+
 /**
   Context of the owner of metadata locks. I.e. each server
   connection has such a context.
@@ -1526,6 +1530,8 @@ class MDL_context {
   }
   bool get_needs_thr_lock_abort() const { return m_needs_thr_lock_abort; }
 
+  void get_locked_object_db_names(MDL_DB_Name_List &list);
+
   void set_force_dml_deadlock_weight(bool force_dml_deadlock_weight) {
     m_force_dml_deadlock_weight = force_dml_deadlock_weight;
   }
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 3030ca49f905..37439c5113d0 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -12181,6 +12181,7 @@ PSI_mutex_key key_LOCK_thd_data;
 PSI_mutex_key key_LOCK_thd_sysvar;
 PSI_mutex_key key_LOCK_thd_protocol;
 PSI_mutex_key key_LOCK_thd_security_ctx;
+PSI_mutex_key key_LOCK_thd_db_read_only_hash;
 PSI_mutex_key key_LOG_LOCK_log;
 PSI_mutex_key key_source_info_data_lock;
 PSI_mutex_key key_source_info_run_lock;
@@ -12268,6 +12269,7 @@ static PSI_mutex_info all_server_mutexes[]=
   { &key_LOCK_thd_sysvar, "THD::LOCK_thd_sysvar", 0, PSI_VOLATILITY_SESSION, PSI_DOCUMENT_ME},
   { &key_LOCK_thd_protocol, "THD::LOCK_thd_protocol", 0, PSI_VOLATILITY_SESSION, PSI_DOCUMENT_ME},
   { &key_LOCK_thd_security_ctx, "THD::LOCK_thd_security_ctx", 0, PSI_VOLATILITY_SESSION, "A lock to control access to a THD's security context"},
+  { &key_LOCK_thd_db_read_only_hash, "THD::LOCK_thd_db_read_only_hash", 0, PSI_VOLATILITY_SESSION, PSI_DOCUMENT_ME},
   { &key_LOCK_user_conn, "LOCK_user_conn", PSI_FLAG_SINGLETON, 0, PSI_DOCUMENT_ME},
   { &key_LOCK_uuid_generator, "LOCK_uuid_generator", PSI_FLAG_SINGLETON, 0, PSI_DOCUMENT_ME},
   { &key_LOCK_sql_rand, "LOCK_sql_rand", PSI_FLAG_SINGLETON, 0, PSI_DOCUMENT_ME},
diff --git a/sql/mysqld.h b/sql/mysqld.h
index 39b4fa832f99..0f2f570cd739 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -513,6 +513,7 @@ extern PSI_mutex_key key_LOCK_thd_data;
 extern PSI_mutex_key key_LOCK_thd_sysvar;
 extern PSI_mutex_key key_LOCK_thd_protocol;
 extern PSI_mutex_key key_LOCK_thd_security_ctx;
+extern PSI_mutex_key key_LOCK_thd_db_read_only_hash;
 extern PSI_mutex_key key_LOG_LOCK_log;
 extern PSI_mutex_key key_source_info_data_lock;
 extern PSI_mutex_key key_source_info_run_lock;
diff --git a/sql/parse_tree_nodes.cc b/sql/parse_tree_nodes.cc
index 80133f5d14e4..e07a4ae74d2a 100644
--- a/sql/parse_tree_nodes.cc
+++ b/sql/parse_tree_nodes.cc
@@ -2109,6 +2109,26 @@ bool set_default_collation(HA_CREATE_INFO *create_info,
   return false;
 }
 
+bool set_db_read_only(HA_CREATE_INFO *create_info, int super_read_only,
+                      int on) {
+  assert(super_read_only >= 0 && super_read_only <= 1);
+  assert(on >= 0 && on <= 1);
+
+  if (on) {
+    create_info->db_read_only =
+        super_read_only ? DB_READ_ONLY_SUPER : DB_READ_ONLY_YES;
+  } else {
+    /*
+      For SUPER_READ_ONLY = FALSE, we assume user meant to downgrade to
+      READ_ONLY = TRUE.
+    */
+    create_info->db_read_only =
+        super_read_only ? DB_READ_ONLY_YES : DB_READ_ONLY_NO;
+  }
+
+  return false;
+}
+
 bool PT_create_table_default_collation::contextualize(
     Table_ddl_parse_context *pc) {
   return (super::contextualize(pc) ||
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 00a9d0e231e6..604141da417e 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -778,6 +778,8 @@ THD::THD(bool enable_plugins)
   m_resource_group_ctx.m_warn = 0;
   m_safe_to_display.store(false);
 
+  m_db_read_only_hash.clear();
+
   mysql_mutex_init(key_LOCK_thd_data, &LOCK_thd_data, MY_MUTEX_INIT_FAST);
   mysql_mutex_init(key_LOCK_thd_query, &LOCK_thd_query, MY_MUTEX_INIT_FAST);
   mysql_mutex_init(key_LOCK_thd_sysvar, &LOCK_thd_sysvar, MY_MUTEX_INIT_FAST);
@@ -788,6 +790,8 @@ THD::THD(bool enable_plugins)
   mysql_mutex_init(key_LOCK_query_plan, &LOCK_query_plan, MY_MUTEX_INIT_FAST);
   mysql_mutex_init(key_LOCK_current_cond, &LOCK_current_cond,
                    MY_MUTEX_INIT_FAST);
+  mysql_mutex_init(key_LOCK_thd_db_read_only_hash, &LOCK_thd_db_read_only_hash,
+                   MY_MUTEX_INIT_FAST);
   mysql_cond_init(key_COND_thr_lock, &COND_thr_lock);
 
   /*Initialize connection delegation mutex and cond*/
@@ -1448,6 +1452,9 @@ THD::~THD() {
   mysql_mutex_unlock(&LOCK_thd_data);
   mysql_mutex_lock(&LOCK_thd_query);
   mysql_mutex_unlock(&LOCK_thd_query);
+  mysql_mutex_lock(&LOCK_thd_db_read_only_hash);
+  m_db_read_only_hash.clear();
+  mysql_mutex_unlock(&LOCK_thd_db_read_only_hash);
 
   assert(!m_attachable_trx);
 
@@ -1462,6 +1469,7 @@ THD::~THD() {
   mysql_mutex_destroy(&LOCK_thd_security_ctx);
   mysql_mutex_destroy(&LOCK_current_cond);
   mysql_mutex_destroy(&LOCK_group_replication_connection_mutex);
+  mysql_mutex_destroy(&LOCK_thd_db_read_only_hash);
 
   mysql_cond_destroy(&COND_thr_lock);
   mysql_cond_destroy(&COND_group_replication_connection_cond_var);
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 02c9222cba10..26c134b689ce 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -114,6 +114,7 @@
 enum enum_check_fields : int;
 enum enum_tx_isolation : int;
 enum ha_notification_type : int;
+enum enum_db_read_only : int;
 class Item;
 class Parser_state;
 class PROFILING;
@@ -1218,6 +1219,11 @@ class THD : public MDL_context_owner,
   */
   mysql_mutex_t LOCK_thd_security_ctx;
 
+  /**
+    Protects THD::db_read_only_hash.
+  */
+  mysql_mutex_t LOCK_thd_db_read_only_hash;
+
   /**
     Protects query plan (SELECT/UPDATE/DELETE's) from being freed/changed
     while another thread explains it. Following structures are protected by
@@ -2385,6 +2391,7 @@ class THD : public MDL_context_owner,
   void set_status_no_index_used();
   void set_status_no_good_index_used();
 
+  std::unordered_map<std::string, enum_db_read_only> m_db_read_only_hash;
   const CHARSET_INFO *db_charset;
 #if defined(ENABLED_PROFILING)
   std::unique_ptr<PROFILING> profiling;
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index b673264290eb..f739f244b2dc 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -81,7 +81,8 @@
 #include "sql/log.h"        // log_*()
 #include "sql/log_event.h"  // Query_log_event
 #include "sql/mdl.h"
-#include "sql/mysqld.h"          // key_file_misc
+#include "sql/mysqld.h"  // key_file_misc
+#include "sql/mysqld_thd_manager.h"
 #include "sql/psi_memory_key.h"  // key_memory_THD_db
 #include "sql/rpl_gtid.h"
 #include "sql/rpl_replica_commit_order_manager.h"  // Commit_order_manager
@@ -159,6 +160,123 @@ bool get_default_db_collation(THD *thd, const char *db_name,
   return false;
 }
 
+/*
+ * Reads from the data dictionary to determine read only setting, and cache
+ * this on the THD, and optionally return in out parameter.
+ *
+ * Returns false on error, true on success.
+ */
+static bool fill_db_read_only_from_dd(THD *thd, const char *db,
+                                      enum enum_db_read_only *out) {
+  DBUG_ENTER("fill_db_read_only_from_dd");
+  dd::Schema_MDL_locker mdl_handler(thd);
+  dd::cache::Dictionary_client::Auto_releaser releaser(thd->dd_client());
+  const dd::Schema *sch_obj = NULL;
+  enum enum_db_read_only flag = DB_READ_ONLY_NULL;
+
+  if (mdl_handler.ensure_locked(db) ||
+      thd->dd_client()->acquire(db, &sch_obj)) {
+    assert(thd->is_error() || thd->killed);
+    DBUG_RETURN(true);
+  }
+
+  if (sch_obj) {
+    flag = static_cast<enum_db_read_only>(sch_obj->get_db_read_only());
+  } else {
+    // Database not found in DD. This happens during CREATE DATABASE, so
+    // assume not read only.
+    flag = DB_READ_ONLY_NO;
+    goto end;
+  }
+
+  // Cache value in THD.
+  mysql_mutex_lock(&thd->LOCK_thd_db_read_only_hash);
+  thd->m_db_read_only_hash[db] = flag;
+  mysql_mutex_unlock(&thd->LOCK_thd_db_read_only_hash);
+
+end:
+  if (out) {
+    *out = flag;
+  }
+  DBUG_RETURN(false);
+}
+
+/*
+ * Check if the database is read-only from thread's local map.
+ */
+bool is_thd_db_read_only_by_name(THD *thd, const char *db) {
+  DBUG_ENTER("is_thd_db_read_only_by_name");
+  bool super = thd->m_main_security_ctx.check_access(SUPER_ACL);
+  enum enum_db_read_only flag = DB_READ_ONLY_NULL;
+
+  // Check cached info in THD first.
+  mysql_mutex_lock(&thd->LOCK_thd_db_read_only_hash);
+  auto it = thd->m_db_read_only_hash.find(std::string(db));
+  if (it != thd->m_db_read_only_hash.end()) {
+    flag = it->second;
+  }
+  mysql_mutex_unlock(&thd->LOCK_thd_db_read_only_hash);
+
+  // Info was not found in THD. Check data dictionary.
+  if (flag == DB_READ_ONLY_NULL) {
+    if (fill_db_read_only_from_dd(thd, db, &flag)) {
+      // Assume read only if we fail to read from DD.
+      DBUG_RETURN(true);
+    }
+  }
+
+  assert(flag >= DB_READ_ONLY_NO && flag <= DB_READ_ONLY_SUPER);
+
+  if (flag == DB_READ_ONLY_SUPER || (flag == DB_READ_ONLY_YES && !super)) {
+    DBUG_RETURN(true);
+  }
+
+  DBUG_RETURN(false);
+}
+
+/* Update db read only flag in all threads' local hash map */
+static void update_thd_db_read_only(const char *db,
+                                    enum enum_db_read_only db_read_only) {
+  DBUG_ENTER("update_thd_db_read_only");
+
+  struct Update_THD_Read_Only : public Do_THD_Impl {
+    const char *m_db;
+    enum enum_db_read_only m_db_read_only;
+    Update_THD_Read_Only(const char *db, enum enum_db_read_only db_read_only)
+        : m_db(db), m_db_read_only(db_read_only) {}
+    virtual void operator()(THD *thd) override {
+      mysql_mutex_lock(&thd->LOCK_thd_db_read_only_hash);
+      thd->m_db_read_only_hash[m_db] = m_db_read_only;
+      mysql_mutex_unlock(&thd->LOCK_thd_db_read_only_hash);
+    }
+  } update_thd(db, db_read_only);
+
+  Global_THD_manager *thd_manager = Global_THD_manager::get_instance();
+  thd_manager->do_for_all_thd(&update_thd);
+
+  DBUG_VOID_RETURN;
+}
+
+/* Delete db read only flag in all threads' local hash map */
+static void del_thd_db_read_only(const char *db) {
+  DBUG_ENTER("del_thd_db_read_only");
+
+  struct Delete_THD_Read_Only : public Do_THD_Impl {
+    const char *m_db;
+    Delete_THD_Read_Only(const char *db) : m_db(db) {}
+    virtual void operator()(THD *thd) override {
+      mysql_mutex_lock(&thd->LOCK_thd_db_read_only_hash);
+      thd->m_db_read_only_hash.erase(m_db);
+      mysql_mutex_unlock(&thd->LOCK_thd_db_read_only_hash);
+    }
+  } update_thd(db);
+
+  Global_THD_manager *thd_manager = Global_THD_manager::get_instance();
+  thd_manager->do_for_all_thd(&update_thd);
+
+  DBUG_VOID_RETURN;
+}
+
 /**
   Check if the thread type is allowed to ignore the schema read
   only option.
@@ -583,6 +701,23 @@ bool mysql_alter_db(THD *thd, const char *db, HA_CREATE_INFO *create_info) {
         create_info->default_table_charset->number);
   }
 
+  if (create_info->db_read_only != DB_READ_ONLY_NULL) {
+    schema->set_db_read_only(create_info->db_read_only);
+
+    // We need to update the value on the current thd early, so that the
+    // statement commit does not error with read only.
+    if (create_info->db_read_only != DB_READ_ONLY_SUPER) {
+      mysql_mutex_lock(&thd->LOCK_thd_db_read_only_hash);
+      thd->m_db_read_only_hash[db] = create_info->db_read_only;
+      mysql_mutex_unlock(&thd->LOCK_thd_db_read_only_hash);
+    } else {
+      // If the current ALTER statement is for super read only, we also need
+      // to populate the THD cache before the data dictionary is updated, so
+      // that statement commit does not error with read only.
+      fill_db_read_only_from_dd(thd, db, nullptr);
+    }
+  }
+
   // Set encryption type.
   if (create_info->used_fields & HA_CREATE_USED_DEFAULT_ENCRYPTION) {
     assert(create_info->encrypt_type.length > 0);
@@ -646,6 +781,9 @@ bool mysql_alter_db(THD *thd, const char *db, HA_CREATE_INFO *create_info) {
                        false);
     }
   }
+  if (create_info->db_read_only != DB_READ_ONLY_NULL) {
+    update_thd_db_read_only(db, create_info->db_read_only);
+  }
 
   my_ok(thd, 1);
   return false;
@@ -725,6 +863,11 @@ bool mysql_rm_db(THD *thd, const LEX_CSTRING &db, bool if_exists) {
 
   if (lock_schema_name(thd, db.str)) return true;
 
+  if (is_thd_db_read_only_by_name(thd, db.str)) {
+    my_error(ER_DB_READ_ONLY, MYF(0), db.str, "Cannot drop read only DB.");
+    return true;
+  }
+
   build_table_filename(path, sizeof(path) - 1, db.str, "", "", 0);
 
   DEBUG_SYNC(thd, "before_acquire_in_drop_schema");
@@ -916,6 +1059,8 @@ bool mysql_rm_db(THD *thd, const LEX_CSTRING &db, bool if_exists) {
       }
       return true;
     }
+
+    del_thd_db_read_only(db.str);
   }
 
   /*
diff --git a/sql/sql_db.h b/sql/sql_db.h
index 7bbc6f24fa08..98c0e895a8c7 100644
--- a/sql/sql_db.h
+++ b/sql/sql_db.h
@@ -31,6 +31,7 @@ namespace dd {
 class Schema;
 }
 
+enum enum_db_read_only : int;
 struct CHARSET_INFO;
 struct HA_CREATE_INFO;
 
@@ -49,4 +50,6 @@ bool get_default_db_collation(THD *thd, const char *db_name,
                               const CHARSET_INFO **collation);
 bool check_schema_readonly(THD *thd, const char *schema_name,
                            TABLE_SHARE *share = nullptr);
+bool is_thd_db_read_only_by_name(THD *thd, const char *db);
+enum_db_read_only get_db_read_only(const dd::Schema &schema);
 #endif /* SQL_DB_INCLUDED */
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 889d5661380d..a479d78fbf08 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -1524,6 +1524,52 @@ static bool deny_updates_if_read_only_option(THD *thd, Table_ref *all_tables) {
   return false;
 }
 
+/**
+ * @brief Determine if an attempt to change a non-temporary table while
+ * the database is read-only.
+ *
+ * Note: similar to deny_updates_if_read_only_option(), the db_read_only
+ * does not block creating/droping/updating temporary tables.
+ *
+ * This is a helper function to mysql_execute_command, and is called
+ * only on DDL statements (CF_AUTO_COMMIT_TRANS).
+ *
+ * @see mysql_execute_command
+ *
+ * @param  thd   Thread (session) context.
+ * @param  all_tables all the tables in the DDL
+ *
+ * @returns Status code
+ *   @retval NULL if updates are OK
+ *   @retval name of the db whose read_only is turned on
+ */
+static const char *deny_implicit_commit_if_db_read_only(THD *thd,
+                                                        Table_ref *all_tables) {
+  DBUG_ENTER("deny_implicit_commit_if_db_read_only");
+  LEX *lex = thd->lex;
+  const char *db_name = nullptr;
+  /* If the DDL does not change data, db_read_only doesn't apply
+   * eihter */
+  if (!(sql_command_flags[lex->sql_command] & CF_CHANGES_DATA))
+    DBUG_RETURN(nullptr);
+  if (!(sql_command_flags[lex->sql_command] & CF_CHANGES_DATA))
+    DBUG_RETURN(nullptr);
+  if ((lex->sql_command == SQLCOM_CREATE_TABLE) &&
+      (lex->create_info->options & HA_LEX_CREATE_TMP_TABLE))
+    DBUG_RETURN(nullptr);
+  if ((lex->sql_command == SQLCOM_DROP_TABLE) && lex->drop_temporary)
+    DBUG_RETURN(nullptr);
+
+  for (Table_ref *table = all_tables; !db_name && table;
+       table = table->next_global) {
+    assert(table->db && table->table_name);
+    if (table->updating && !find_temporary_table(thd, table) &&
+        is_thd_db_read_only_by_name(thd, table->db))
+      db_name = table->db;
+  }
+  DBUG_RETURN(db_name);
+}
+
 /*
   This is the function to perform the check for variable
   "allow_noncurrent_db_rw". It will assume the command is
@@ -3339,6 +3385,18 @@ int mysql_execute_command(THD *thd, bool first_level) {
     /* Statement transaction still should not be started. */
     assert(thd->get_transaction()->is_empty(Transaction_ctx::STMT));
 
+    /*
+     * Implicit commits that change permanent tables (DDLs)
+     * are not allowed if per-db read_only is set.
+     * We simply return here (i.e. abort this DDL)
+     * without rolling back any explicit transaction.
+     */
+    const char *db = deny_implicit_commit_if_db_read_only(thd, all_tables);
+    if (db) {
+      my_error(ER_DB_READ_ONLY, MYF(0), db, "Implicit commit failed.");
+      return -1;
+    }
+
     /*
       Implicit commit is not allowed with an active XA transaction.
       In this case we should not release metadata locks as the XA transaction
diff --git a/sql/sql_parse.h b/sql/sql_parse.h
index 4f0bee05a362..afb0404a429d 100644
--- a/sql/sql_parse.h
+++ b/sql/sql_parse.h
@@ -294,6 +294,7 @@ bool set_default_charset(HA_CREATE_INFO *create_info,
 // TODO: remove after refactoring of ALTER DATABASE:
 bool set_default_collation(HA_CREATE_INFO *create_info,
                            const CHARSET_INFO *value);
+bool set_db_read_only(HA_CREATE_INFO *create_info, int super_read_only, int on);
 
 /* Bits in sql_command_flags */
 
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 003ee8843903..d27628b28b9f 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -1412,6 +1412,9 @@ bool mysqld_show_create_db(THD *thd, char *dbname,
 
     schema_read_only = schema->read_only();
 
+    create.db_read_only =
+        static_cast<enum_db_read_only>(schema->get_db_read_only());
+
     if (create.default_table_charset == nullptr)
       create.default_table_charset = thd->collation();
 
@@ -1433,15 +1436,23 @@ bool mysqld_show_create_db(THD *thd, char *dbname,
     buffer.append(STRING_WITH_LEN("/*!32312 IF NOT EXISTS*/ "));
   append_identifier(thd, &buffer, orig_dbname, strlen(orig_dbname));
 
-  if (create.default_table_charset) {
+  if (create.default_table_charset || create.db_read_only > DB_READ_ONLY_NO) {
     buffer.append(STRING_WITH_LEN(" /*!40100"));
-    buffer.append(STRING_WITH_LEN(" DEFAULT CHARACTER SET "));
-    buffer.append(create.default_table_charset->csname);
-    if (!(create.default_table_charset->state & MY_CS_PRIMARY) ||
-        create.default_table_charset == &my_charset_utf8mb4_0900_ai_ci) {
-      buffer.append(STRING_WITH_LEN(" COLLATE "));
-      buffer.append(create.default_table_charset->m_coll_name);
+    if (create.default_table_charset) {
+      buffer.append(STRING_WITH_LEN(" DEFAULT CHARACTER SET "));
+      buffer.append(create.default_table_charset->csname);
+      if (!(create.default_table_charset->state & MY_CS_PRIMARY) ||
+          create.default_table_charset == &my_charset_utf8mb4_0900_ai_ci) {
+        buffer.append(STRING_WITH_LEN(" COLLATE "));
+        buffer.append(create.default_table_charset->m_coll_name);
+      }
     }
+    if (create.db_read_only == DB_READ_ONLY_YES) {
+      buffer.append(STRING_WITH_LEN(" READ_ONLY"));
+    } else if (create.db_read_only == DB_READ_ONLY_SUPER) {
+      buffer.append(STRING_WITH_LEN(" SUPER_READ_ONLY"));
+    }
+
     buffer.append(STRING_WITH_LEN(" */"));
   }
   buffer.append(STRING_WITH_LEN(" /*!80016"));
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 2c25fd11c679..9380657c8e3e 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -523,7 +523,7 @@ void warn_on_deprecated_user_defined_collation(
   2. We should not introduce new shift/reduce conflicts any more.
 */
 
-%expect 63
+%expect 65
 
 /*
    MAINTAINER:
@@ -1427,6 +1427,7 @@ void warn_on_deprecated_user_defined_collation(
 %token  FIND 1250
 %token<lexer.keyword> GTID_SYM 1251
 %token<lexer.keyword> GTID_EXECUTED 1252           /* MYSQL */
+%token<lexer.keyword> SUPER_READ_ONLY_SYM 1253
 
 /*
   Resolve column attribute ambiguity -- force precedence of "UNIQUE KEY" against
@@ -2211,6 +2212,7 @@ void warn_on_deprecated_user_defined_collation(
 
 %type <query_id> opt_for_query
 
+%type <num> read_only_opt boolean_val
 %%
 
 /*
@@ -6692,6 +6694,9 @@ create_database_option:
             Lex->create_info->encrypt_type= $1;
             Lex->create_info->used_fields |= HA_CREATE_USED_DEFAULT_ENCRYPTION;
           }
+        | db_read_only
+          {
+          }
         ;
 
 opt_if_not_exists:
@@ -6940,6 +6945,25 @@ default_encryption:
           opt_default ENCRYPTION_SYM opt_equal TEXT_STRING_sys { $$ = $4;}
         ;
 
+db_read_only:
+          read_only_opt { /* Ignored */ }
+        | read_only_opt opt_equal boolean_val
+          {
+            if (set_db_read_only(Lex->create_info, $1, $3))
+              MYSQL_YYABORT;
+          }
+        ;
+
+read_only_opt:
+          READ_ONLY_SYM { $$ = 0; }
+        | SUPER_READ_ONLY_SYM { $$ = 1; }
+        ;
+
+boolean_val:
+          FALSE_SYM { $$ = 0; }
+        | TRUE_SYM { $$ = 1; }
+        ;
+
 row_types:
           DEFAULT_SYM    { $$= ROW_TYPE_DEFAULT; }
         | FIXED_SYM      { $$= ROW_TYPE_FIXED; }
@@ -15701,6 +15725,7 @@ ident_keywords_unambiguous:
         | SUBJECT_SYM
         | SUBPARTITIONS_SYM
         | SUBPARTITION_SYM
+        | SUPER_READ_ONLY_SYM
         | SUSPEND_SYM
         | SWAPS_SYM
         | SWITCHES_SYM