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/db_read_only.result b/mysql-test/r/db_read_only.result
new file mode 100644
index 000000000000..9edd3dce9122
--- /dev/null
+++ b/mysql-test/r/db_read_only.result
@@ -0,0 +1,767 @@
+drop table if exists t1, t2;
+drop database if exists test2;
+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 latin1 */
+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(11)	YES	MUL	NULL	
+drop index a on t1;
+alter table t2 add key (a);
+describe t2;
+Field	Type	Null	Key	Default	Extra
+a	int(11)	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 latin1 */
+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(11)	YES	MUL	NULL	
+drop index a on t1;
+alter table t2 add key (a);
+describe t2;
+Field	Type	Null	Key	Default	Extra
+a	int(11)	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 latin1 */
+connection con1;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET latin1 */
+# 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)
+alter database test read_only = true;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET latin1 READ_ONLY */
+connection con1;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET latin1 READ_ONLY */
+# 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
+t1	6
+t1	7
+t1	8
+select 't2', a from t2;
+t2	a
+t2	2
+t2	4
+t2	6
+t2	8
+t2	10
+t2	12
+# 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. Implicit commit failed.
+#
+# 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 latin1 */
+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;
+# super user can still write in read_only mode
+connection root2;
+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(11)	YES	MUL	NULL	
+drop index a on t1;
+alter table t2 add key (a);
+describe t2;
+Field	Type	Null	Key	Default	Extra
+a	int(11)	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;
+#
+# 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 latin1 */
+connection root2;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET latin1 */
+# 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)
+alter database test super_read_only = true;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET latin1 SUPER_READ_ONLY */
+connection root2;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET latin1 SUPER_READ_ONLY */
+# 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
+t1	6
+t1	7
+t1	8
+select 't2', a from t2;
+t2	a
+t2	2
+t2	4
+t2	6
+t2	8
+t2	10
+t2	12
+# 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. Implicit commit failed.
+#
+# 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 latin1 */
+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 latin1 READ_ONLY */
+# 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(11)	YES	MUL	NULL	
+drop index a on t1;
+alter table t2 add key (a);
+describe t2;
+Field	Type	Null	Key	Default	Extra
+a	int(11)	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 latin1 SUPER_READ_ONLY */
+alter database test read_only = false;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET latin1 */
+alter database test read_only = true;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET latin1 READ_ONLY */
+connection default;
+#
+# on non-user databases
+#
+alter database mysql read_only = true;
+show create database mysql;
+Database	Create Database
+mysql	CREATE DATABASE `mysql` /*!40100 DEFAULT CHARACTER SET latin1 READ_ONLY */
+alter database mysql read_only = false;
+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 utf8 */
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET latin1 READ_ONLY */
+# restarting mysqld
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET latin1 READ_ONLY */
+connection default;
+alter database test read_only = false;
+show create database test;
+Database	Create Database
+test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET latin1 */
+drop table t1, t2;
+drop user test@localhost;
diff --git a/mysql-test/suite/perfschema/r/sizing_default.result b/mysql-test/suite/perfschema/r/sizing_default.result
index 1ca202e821f5..8206389f067b 100644
--- a/mysql-test/suite/perfschema/r/sizing_default.result
+++ b/mysql-test/suite/perfschema/r/sizing_default.result
@@ -28,7 +28,7 @@ performance_schema_max_file_classes	50
 performance_schema_max_file_handles	32768
 performance_schema_max_file_instances	7693
 performance_schema_max_mutex_classes	200
-performance_schema_max_mutex_instances	15906
+performance_schema_max_mutex_instances	16208
 performance_schema_max_rwlock_classes	40
 performance_schema_max_rwlock_instances	9102
 performance_schema_max_socket_classes	10
diff --git a/mysql-test/suite/perfschema/r/sizing_growth.result b/mysql-test/suite/perfschema/r/sizing_growth.result
index 2190e904fd4e..67e1e52770e8 100644
--- a/mysql-test/suite/perfschema/r/sizing_growth.result
+++ b/mysql-test/suite/perfschema/r/sizing_growth.result
@@ -214,7 +214,7 @@ select @file_per_share <= 3;
 1
 select @mutex_per_con;
 @mutex_per_con
-3
+4
 select @rwlock_per_con;
 @rwlock_per_con
 1
@@ -240,7 +240,7 @@ select if(
 (@rwlock_per_share <= 3)
 AND (@cond_per_share = 0)
 AND (@file_per_share <= 3)
-AND (@mutex_per_con = 3)
+AND (@mutex_per_con = 4)
 AND (@rwlock_per_con = 1)
 AND (@cond_per_con = 2)
 AND (@file_per_con = 0)
diff --git a/mysql-test/suite/perfschema/r/sizing_high.result b/mysql-test/suite/perfschema/r/sizing_high.result
index b633d5fce8db..d298d36866c8 100644
--- a/mysql-test/suite/perfschema/r/sizing_high.result
+++ b/mysql-test/suite/perfschema/r/sizing_high.result
@@ -28,7 +28,7 @@ performance_schema_max_file_classes	50
 performance_schema_max_file_handles	32768
 performance_schema_max_file_instances	23385
 performance_schema_max_mutex_classes	200
-performance_schema_max_mutex_instances	52200
+performance_schema_max_mutex_instances	52600
 performance_schema_max_rwlock_classes	40
 performance_schema_max_rwlock_instances	30800
 performance_schema_max_socket_classes	10
diff --git a/mysql-test/suite/perfschema/r/sizing_low.result b/mysql-test/suite/perfschema/r/sizing_low.result
index dce5a994099d..11e19160360d 100644
--- a/mysql-test/suite/perfschema/r/sizing_low.result
+++ b/mysql-test/suite/perfschema/r/sizing_low.result
@@ -28,7 +28,7 @@ performance_schema_max_file_classes	50
 performance_schema_max_file_handles	32768
 performance_schema_max_file_instances	1556
 performance_schema_max_mutex_classes	200
-performance_schema_max_mutex_instances	2945
+performance_schema_max_mutex_instances	3000
 performance_schema_max_rwlock_classes	40
 performance_schema_max_rwlock_instances	1612
 performance_schema_max_socket_classes	10
diff --git a/mysql-test/suite/perfschema/r/sizing_med.result b/mysql-test/suite/perfschema/r/sizing_med.result
index 2eda017467ab..2056517fc10a 100644
--- a/mysql-test/suite/perfschema/r/sizing_med.result
+++ b/mysql-test/suite/perfschema/r/sizing_med.result
@@ -28,7 +28,7 @@ performance_schema_max_file_classes	50
 performance_schema_max_file_handles	32768
 performance_schema_max_file_instances	1754
 performance_schema_max_mutex_classes	200
-performance_schema_max_mutex_instances	4230
+performance_schema_max_mutex_instances	4448
 performance_schema_max_rwlock_classes	40
 performance_schema_max_rwlock_instances	2222
 performance_schema_max_socket_classes	10
diff --git a/mysql-test/suite/perfschema/t/sizing_growth.test b/mysql-test/suite/perfschema/t/sizing_growth.test
index d6944d2aee0d..ad81ccadd1ee 100644
--- a/mysql-test/suite/perfschema/t/sizing_growth.test
+++ b/mysql-test/suite/perfschema/t/sizing_growth.test
@@ -464,7 +464,7 @@ select if(
       (@rwlock_per_share <= 3)
   AND (@cond_per_share = 0)
   AND (@file_per_share <= 3)
-  AND (@mutex_per_con = 3)
+  AND (@mutex_per_con = 4)
   AND (@rwlock_per_con = 1)
   AND (@cond_per_con = 2)
   AND (@file_per_con = 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..d24177b5c174
--- /dev/null
+++ b/mysql-test/t/db_read_only.test
@@ -0,0 +1,202 @@
+# Test per-database read-only attribute
+--source include/have_innodb.inc
+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;
+
+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)
+alter database test read_only = true;
+show create database test;
+
+# con1
+--echo connection con1;
+connection con1;
+show create database test;
+
+--source include/db_read_only_on.inc
+
+--echo # super user can still write in read_only mode
+--echo connection root2;
+connection root2;
+
+--source include/db_read_only_off.inc
+
+--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)
+alter database test super_read_only = true;
+show create database test;
+
+# 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 #
+alter database mysql read_only = true;
+show create database mysql;
+alter database mysql read_only = false;
+
+--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;
+
+#
+# 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;
+
+# cleaning up db.opt files
+file_exists $MYSQLD_DATADIR/mysql/db.opt;
+file_exists $MYSQLD_DATADIR/test/db.opt;
+
+remove_file $MYSQLD_DATADIR/mysql/db.opt;
+remove_file $MYSQLD_DATADIR/test/db.opt;
+
+--error 1
+file_exists $MYSQLD_DATADIR/mysql/db.opt;
+--error 1
+file_exists $MYSQLD_DATADIR/test/db.opt;
diff --git a/sql/handler.cc b/sql/handler.cc
index ebef0349e850..147b165c80af 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -41,6 +41,8 @@
 #include <my_bit.h>
 #include <list>
 #include "sql_readonly.h"       // check_ro
+#include "sql_db.h"      // init_thd_db_read_only
+                         // is_thd_db_read_only_by_name
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
 #include "ha_partition.h"
@@ -1422,10 +1424,10 @@ int ha_commit_trans(THD *thd, bool all, bool async,
   MDL_request mdl_request;
   bool release_mdl= false;
 
+  bool rw_trans = false;
   if (ha_info)
   {
     uint rw_ha_count;
-    bool rw_trans;
 
     DBUG_EXECUTE_IF("crash_commit_before", DBUG_SUICIDE(););
 
@@ -1438,6 +1440,17 @@ int ha_commit_trans(THD *thd, bool all, bool async,
     /* rw_trans is TRUE when we in a transaction changing data */
     rw_trans= is_real_trans && (rw_ha_count > 0);
 
+    // initialize thread's db_read_only_hash on the first time
+    if (rw_trans && !my_hash_inited(&thd->db_read_only_hash))
+      init_thd_db_read_only(thd);
+
+    /* If this is a write transaction, we need to lock the
+     * (local) db_read_only hash map during the whole transaction,
+     * so the read_only option is not changed once a commit starts.
+     */
+    if (rw_trans)
+      mysql_mutex_lock(&thd->LOCK_thd_db_read_only_hash);
+
     DBUG_EXECUTE_IF("dbug.enabled_commit",
                     {
                       const char act[]= "now signal Reached wait_for signal.commit_continue";
@@ -1488,6 +1501,30 @@ int ha_commit_trans(THD *thd, bool all, bool async,
       goto end;
     }
 
+    if (rw_trans && thd->db_read_only_hash.records)
+    {
+      // 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.");
+          ha_rollback_trans(thd, all);
+          error= 1;
+          goto end;
+        }
+      }
+    }
+
     if (!trans->no_2pc && (rw_ha_count > 1))
       error= tc_log->prepare(thd, all, async);
   }
@@ -1514,6 +1551,9 @@ int ha_commit_trans(THD *thd, bool all, bool async,
   /* Free resources and perform other cleanup even for 'empty' transactions. */
   if (is_real_trans)
     thd->transaction.cleanup();
+  /* Unlock db_read_only hash after a write-transaction */
+  if (rw_trans)
+    mysql_mutex_unlock(&thd->LOCK_thd_db_read_only_hash);
   DBUG_RETURN(error);
 }
 
diff --git a/sql/handler.h b/sql/handler.h
index bada3b43cf17..4c9ca73e16e1 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -1071,6 +1071,7 @@ enum enum_stats_auto_recalc { HA_STATS_AUTO_RECALC_DEFAULT= 0,
 typedef struct st_ha_create_information
 {
   const CHARSET_INFO *table_charset, *default_table_charset;
+  uchar db_read_only;
   LEX_STRING connect_string;
   const char *password, *tablespace;
   LEX_STRING comment;
@@ -1105,6 +1106,9 @@ typedef struct st_ha_create_information
   enum ha_storage_media storage_media;  /* DEFAULT, DISK or MEMORY */
   bool rbr_column_names; /* If true, column names for this table are logged
                             in Table_map_log_events */
+
+  /* initialize db_read_only parameter */
+  st_ha_create_information() : db_read_only(0) {}
 } HA_CREATE_INFO;
 
 /**
diff --git a/sql/lex.h b/sql/lex.h
index e44af15e90ff..eaf1d75a45b9 100644
--- a/sql/lex.h
+++ b/sql/lex.h
@@ -585,6 +585,7 @@ static SYMBOL symbols[] = {
   { "SUBPARTITION",     SYM(SUBPARTITION_SYM)},
   { "SUBPARTITIONS",    SYM(SUBPARTITIONS_SYM)},
   { "SUPER",		SYM(SUPER_SYM)},
+  { "SUPER_READ_ONLY",		SYM(SUPER_READ_ONLY_SYM)},
   { "SUSPEND",          SYM(SUSPEND_SYM)},
   { "SWAPS",      SYM(SWAPS_SYM)},
   { "SWITCHES",   SYM(SWITCHES_SYM)},
diff --git a/sql/mdl.cc b/sql/mdl.cc
index bf15fb0cb768..dcb2e9db40ac 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -2708,6 +2708,29 @@ void MDL_context::find_deadlock()
   }
 }
 
+/*
+ * Get the list of database names of all lock objects in a transaction.
+ * This is used to check per-database read-only.
+ */
+void MDL_context::get_locked_object_db_names(MDL_DB_Name_List &list)
+{
+  MDL_ticket *ticket;
+  Ticket_iterator it(m_tickets[MDL_TRANSACTION]);
+  DBUG_ENTER("MDL_context::get_locked_object_db_names");
+
+  if (m_tickets[MDL_TRANSACTION].is_empty())
+    DBUG_VOID_RETURN;
+
+  while ((ticket= it++) && ticket)
+  {
+    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;
+}
 
 /**
   Release lock.
diff --git a/sql/mdl.h b/sql/mdl.h
index 2677bf848e42..83d96eded718 100644
--- a/sql/mdl.h
+++ b/sql/mdl.h
@@ -30,6 +30,8 @@
 #include <mysql_com.h>
 
 #include <algorithm>
+#include <unordered_set>
+#include <string>
 
 class THD;
 
@@ -719,6 +721,8 @@ typedef I_P_List<MDL_request, I_P_List_adapter<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.
@@ -808,6 +812,8 @@ class MDL_context
   {
     return m_needs_thr_lock_abort;
   }
+
+  void get_locked_object_db_names(MDL_DB_Name_List &list);
 public:
   /**
     If our request for a lock is scheduled, or aborted by the deadlock
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 5d50c78519b1..a4b703bdffaa 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -10504,6 +10504,7 @@ PSI_mutex_key
   key_LOCK_slave_net_timeout,
   key_LOCK_server_started, key_LOCK_status,
   key_LOCK_system_variables_hash, key_LOCK_table_share, key_LOCK_thd_data,
+  key_LOCK_thd_db_read_only_hash,
   key_LOCK_user_conn, key_LOCK_uuid_generator, key_LOG_LOCK_log,
   key_master_info_data_lock, key_master_info_run_lock,
   key_master_info_sleep_lock, key_master_info_thd_lock,
@@ -10598,6 +10599,7 @@ static PSI_mutex_info all_server_mutexes[]=
   { &key_LOCK_system_variables_hash, "LOCK_system_variables_hash", PSI_FLAG_GLOBAL},
   { &key_LOCK_table_share, "LOCK_table_share", PSI_FLAG_GLOBAL},
   { &key_LOCK_thd_data, "THD::LOCK_thd_data", 0},
+  { &key_LOCK_thd_db_read_only_hash, "THD::LOCK_thd_db_read_only_hash", 0},
   { &key_LOCK_user_conn, "LOCK_user_conn", PSI_FLAG_GLOBAL},
   { &key_LOCK_uuid_generator, "LOCK_uuid_generator", PSI_FLAG_GLOBAL},
   { &key_LOCK_sql_rand, "LOCK_sql_rand", PSI_FLAG_GLOBAL},
diff --git a/sql/mysqld.h b/sql/mysqld.h
index 421c6534c902..1a5963363bf5 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -905,6 +905,7 @@ extern PSI_mutex_key
   key_LOCK_slave_net_timeout,
   key_LOCK_server_started, key_LOCK_status,
   key_LOCK_table_share, key_LOCK_thd_data,
+  key_LOCK_thd_db_read_only_hash,
   key_LOCK_user_conn, key_LOCK_uuid_generator, key_LOG_LOCK_log,
   key_master_info_data_lock, key_master_info_run_lock,
   key_master_info_sleep_lock, key_master_info_thd_lock,
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 595a5aaceb31..3700fd8c92b4 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -7197,6 +7197,8 @@ ER_ORDERBY_AS_TYPE_NOT_SUPPORTED
 ER_TOO_LONG_DOCUMENT_PATH_INDEX
   eng "Specified document path key was too long; max path length is %d bytes"
 
+ER_DB_READ_ONLY
+  eng "Database '%s' is in read-only mode. %s"
 #
 #  End of 5.6 error messages.
 #
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index d4825d042f79..f530a57d9da6 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -1016,6 +1016,7 @@ THD::THD(bool enable_plugins)
   query_id= 0;
   query_name_consts= 0;
   db_charset= global_system_variables.collation_database;
+  my_hash_clear(&db_read_only_hash);
   memset(ha_data, 0, sizeof(ha_data));
   mysys_var=0;
   binlog_evt_union.do_union= FALSE;
@@ -1042,6 +1043,8 @@ THD::THD(bool enable_plugins)
   active_vio = 0;
 #endif
   mysql_mutex_init(key_LOCK_thd_data, &LOCK_thd_data, MY_MUTEX_INIT_FAST);
+  mysql_mutex_init(key_LOCK_thd_db_read_only_hash, &LOCK_thd_db_read_only_hash,
+                   MY_MUTEX_INIT_FAST);
 
   /* Variables with default values */
   proc_info="login";
@@ -1686,6 +1689,19 @@ THD::~THD()
   db= NULL;
   free_root(&transaction.mem_root,MYF(0));
   mysql_mutex_destroy(&LOCK_thd_data);
+
+  /* The session may still holding the lock in an ongoing transaction, so we
+   * trylock and then unlock before destroying it. Note: we will not have
+   * another thread holding the lock to update the read_only hash table while
+   * this thread is being deleted.  LOCK_thd_remove is locked before updating
+   * local read_only hash map, which prevents any thread being deleted. See
+   * sql_db.cc:update_thd_db_read_only() for details. */
+  mysql_mutex_trylock(&LOCK_thd_db_read_only_hash);
+  mysql_mutex_unlock(&LOCK_thd_db_read_only_hash);
+
+  my_hash_free(&db_read_only_hash);
+  mysql_mutex_destroy(&LOCK_thd_db_read_only_hash);
+
 #ifndef DBUG_OFF
   dbug_sentry= THD_SENTRY_GONE;
 #endif  
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 7c4f2209c1b9..1c72261e0a08 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -2325,6 +2325,7 @@ class THD :public MDL_context_owner,
     Is locked when THD is deleted.
   */
   mysql_mutex_t LOCK_thd_data;
+  mysql_mutex_t LOCK_thd_db_read_only_hash;
 
   /* all prepared statements and cursors of this connection */
   Statement_map stmt_map;
@@ -3089,6 +3090,8 @@ class THD :public MDL_context_owner,
 
   void capture_system_thread_id();
 
+  /* local hash map of db opt */
+  HASH db_read_only_hash;
   const CHARSET_INFO *db_charset;
 #if defined(ENABLED_PROFILING)
   PROFILING  profiling;
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index af9aced59367..42e7b3901c77 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -31,6 +31,7 @@
 #include "log_event.h"                   // Query_log_event
 #include "sql_base.h"                    // lock_table_names, tdc_remove_table
 #include "sql_handler.h"                 // mysql_ha_rm_tables
+#include "global_threads.h"              // LOCK_thd_remove
 #include <mysys_err.h>
 #include "sp.h"
 #include "events.h"
@@ -43,6 +44,7 @@
 #include <direct.h>
 #endif
 #include "debug_sync.h"
+#include "sql_show.h"
 
 #define MAX_DROP_TABLE_Q_LEN      1024
 
@@ -75,6 +77,7 @@ typedef struct my_dbopt_st
   char *name;			/* Database name                  */
   uint name_length;		/* Database length name           */
   const CHARSET_INFO *charset;	/* Database default character set */
+  uchar db_read_only;
 } my_dbopt_t;
 
 
@@ -224,6 +227,7 @@ static my_bool get_dbopt(const char *dbname, HA_CREATE_INFO *create)
   if ((opt= (my_dbopt_t*) my_hash_search(&dboptions, (uchar*) dbname, length)))
   {
     create->default_table_charset= opt->charset;
+    create->db_read_only= opt->db_read_only;
     error= 0;
   }
   mysql_rwlock_unlock(&LOCK_dboptions);
@@ -279,12 +283,41 @@ static my_bool put_dbopt(const char *dbname, HA_CREATE_INFO *create)
 
   /* Update / write options in hash */
   opt->charset= create->default_table_charset;
+  opt->db_read_only= create->db_read_only;
 
 end:
   mysql_rwlock_unlock(&LOCK_dboptions);
   DBUG_RETURN(error);
 }
 
+/* Delete db opt from all thread's local hash maps */
+static void del_thd_db_read_only(my_dbopt_t *opt)
+{
+  DBUG_ENTER("del_thd_db_read_only");
+  DBUG_ASSERT(opt);
+
+  // We only need to update the existing threads (and block removing threads)
+  // For new threads, they will initialize the local hash maps propoerly
+  std::set<THD*> global_thread_list_copy;
+  mysql_mutex_lock(&LOCK_thd_remove);
+  copy_global_thread_list(&global_thread_list_copy);
+
+  Thread_iterator it= global_thread_list_copy.begin();
+  Thread_iterator end= global_thread_list_copy.end();
+  for (; it != end; ++it)
+  {
+    THD *tmp= *it;
+    mysql_mutex_lock(&tmp->LOCK_thd_db_read_only_hash);
+    // update if the thread's db_read_only_hash is inited
+    if (my_hash_inited(&tmp->db_read_only_hash))
+      my_hash_delete(&tmp->db_read_only_hash, (uchar*) opt);
+    mysql_mutex_unlock(&tmp->LOCK_thd_db_read_only_hash);
+  }
+
+  mysql_mutex_unlock(&LOCK_thd_remove);
+
+  DBUG_VOID_RETURN;
+}
 
 /*
   Deletes database options from the hash.
@@ -296,10 +329,155 @@ static void del_dbopt(const char *path)
   mysql_rwlock_wrlock(&LOCK_dboptions);
   if ((opt= (my_dbopt_t *)my_hash_search(&dboptions, (const uchar*) path,
                                          strlen(path))))
+  {
+    if (opt->db_read_only)
+      del_thd_db_read_only(opt); // removing db_opt from threads
     my_hash_delete(&dboptions, (uchar*) opt);
+  }
   mysql_rwlock_unlock(&LOCK_dboptions);
 }
 
+/*
+ * 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");
+  mysql_mutex_assert_owner(&thd->LOCK_thd_db_read_only_hash);
+
+  bool ret = false;
+  char path[FN_REFLEN + 1];
+  uint path_len= build_table_filename(path, sizeof(path) - 1,
+                                      db, "", MY_DB_OPT_FILE, 0);
+
+  bool super = (thd->security_ctx->master_access & SUPER_ACL);
+  my_dbopt_t *opt= (my_dbopt_t *)my_hash_search
+    (&thd->db_read_only_hash, (const uchar*)path, path_len);
+
+  if (opt)
+  {
+    DBUG_ASSERT(opt->db_read_only);
+    if (opt->db_read_only > 1 || !super)
+      ret = true;
+  }
+
+  DBUG_RETURN(ret);
+}
+
+/*
+ * Initialize thread's local db opt hash map
+ */
+void init_thd_db_read_only(THD *thd)
+{
+  DBUG_ENTER("init_thd_db_read_only");
+  DBUG_ASSERT(!my_hash_inited(&thd->db_read_only_hash));
+  mysql_mutex_assert_not_owner(&thd->LOCK_thd_db_read_only_hash);
+
+  bool empty = (dboptions.records == 0);
+
+  // load dboption cache if the cache is empty
+  // ok if we turned out to load cache twice
+  if (empty)
+    fetch_schema_schemata(thd);
+
+  mysql_rwlock_rdlock(&LOCK_dboptions);
+  mysql_mutex_lock(&thd->LOCK_thd_db_read_only_hash);
+
+  /* check again whether it is initialized */
+  if (!my_hash_inited(&thd->db_read_only_hash))
+  {
+    /*
+     * Note: we do not create new my_dbopt_t objects, but stores objects in the
+     * shared map into the local hash map. So the local hash map doesn't need a
+     * free_element function.
+     */
+    if (!my_hash_init(&thd->db_read_only_hash, lower_case_table_names ?
+                      system_charset_info : &my_charset_bin,
+                      32, 0, 0, (my_hash_get_key) dboptions_get_key,
+                      NULL /* free_element */, 0))
+    {
+      for (uint idx= 0; idx < dboptions.records; ++idx)
+      {
+        my_dbopt_t *opt= (my_dbopt_t*) my_hash_element(&dboptions, idx);
+        // only insert the db opt if db_read_only is turned on
+        if (opt->db_read_only)
+          my_hash_insert(&thd->db_read_only_hash, (uchar*) opt);
+      }
+    }
+    else
+      // initialization failed. clear the hash map
+      my_hash_clear(&thd->db_read_only_hash);
+  }
+
+  mysql_mutex_unlock(&thd->LOCK_thd_db_read_only_hash);
+  mysql_rwlock_unlock(&LOCK_dboptions);
+  DBUG_VOID_RETURN;
+}
+
+/**
+ * Return db_read_only value in the dbopt
+ *
+ * Similar logic here as in get_default_db_collation() -
+ * In case load_db_opt_by_name() fails (e.g. db.opt file
+ * doesn't exist), db_read_only will be false (0) by default.
+ */
+static uchar get_db_read_only(THD *thd, const char *path)
+{
+  HA_CREATE_INFO db_info;
+
+  load_db_opt(thd, path, &db_info);
+  return db_info.db_read_only;
+}
+
+/* Update db_ops in all threads' local hash map */
+static void update_thd_db_read_only(const char *path, uchar db_read_only)
+{
+  DBUG_ENTER("update_thd_db_read_only");
+
+  mysql_rwlock_rdlock(&LOCK_dboptions);
+
+  // We only need to update the existing threads (and block removing threads)
+  // For new threads, they will initialize the local hash maps propoerly
+  std::set<THD*> global_thread_list_copy;
+  mysql_mutex_lock(&LOCK_thd_remove);
+  copy_global_thread_list(&global_thread_list_copy);
+
+  Thread_iterator it= global_thread_list_copy.begin();
+  Thread_iterator end= global_thread_list_copy.end();
+  for (; it != end; ++it)
+  {
+    THD *tmp= *it;
+    mysql_mutex_lock(&tmp->LOCK_thd_db_read_only_hash);
+
+    // update if the thread's db_read_only_hash is inited
+    if (my_hash_inited(&tmp->db_read_only_hash))
+    {
+      my_dbopt_t *opt= (my_dbopt_t *)my_hash_search
+        (&tmp->db_read_only_hash, (const uchar*) path, strlen(path));
+
+      // db_opt is in the hash map only if db_read_only is turned on
+      if (!db_read_only && opt)
+        my_hash_delete(&tmp->db_read_only_hash, (uchar*) opt);
+      else if (db_read_only && !opt)
+      {
+        // get the db_opt from shared hash map
+        opt= (my_dbopt_t *)my_hash_search
+          (&dboptions, (const uchar*) path, strlen(path));
+
+        // check if the db_opt actually exists
+        if (opt)
+          my_hash_insert(&tmp->db_read_only_hash, (uchar*) opt);
+      }
+    }
+
+    mysql_mutex_unlock(&tmp->LOCK_thd_db_read_only_hash);
+  }
+
+  mysql_mutex_unlock(&LOCK_thd_remove);
+  mysql_rwlock_unlock(&LOCK_dboptions);
+
+  DBUG_VOID_RETURN;
+}
 
 /*
   Create database options file:
@@ -315,12 +493,14 @@ static void del_dbopt(const char *path)
 static bool write_db_opt(THD *thd, const char *path, HA_CREATE_INFO *create)
 {
   register File file;
-  char buf[256]; // Should be enough for one option
+  char buf[512]; // Should be enough for options
   bool error=1;
 
   if (!create->default_table_charset)
     create->default_table_charset= thd->variables.collation_server;
 
+  uchar prev_db_read_only= get_db_read_only(thd, path);
+
   if (put_dbopt(path, create))
     return 1;
 
@@ -332,6 +512,9 @@ static bool write_db_opt(THD *thd, const char *path, HA_CREATE_INFO *create)
                               create->default_table_charset->csname,
                               "\ndefault-collation=",
                               create->default_table_charset->name,
+                              "\ndb-read-only=",
+                              create->db_read_only==0? "0":
+                              create->db_read_only==1? "1":"2",
                               "\n", NullS) - buf);
 
     /* Error is written by mysql_file_write */
@@ -339,6 +522,12 @@ static bool write_db_opt(THD *thd, const char *path, HA_CREATE_INFO *create)
       error=0;
     mysql_file_close(file, MYF(0));
   }
+
+  /* If the read_only option is not changed,
+   * don't need to update it across all sessions */
+  if (prev_db_read_only != create->db_read_only)
+    update_thd_db_read_only(path, create->db_read_only);
+
   return error;
 }
 
@@ -361,7 +550,7 @@ static bool write_db_opt(THD *thd, const char *path, HA_CREATE_INFO *create)
 bool load_db_opt(THD *thd, const char *path, HA_CREATE_INFO *create)
 {
   File file;
-  char buf[256];
+  char buf[512];
   DBUG_ENTER("load_db_opt");
   bool error=1;
   uint nbytes;
@@ -420,6 +609,10 @@ bool load_db_opt(THD *thd, const char *path, HA_CREATE_INFO *create)
           create->default_table_charset= default_charset_info;
         }
       }
+      else if (!strncmp(buf,"db-read-only", (pos-buf)))
+      {
+        create->db_read_only = (*(pos+1) - '0');
+      }
     }
   }
   /*
diff --git a/sql/sql_db.h b/sql/sql_db.h
index 09d6b210d557..01397aa6a3bd 100644
--- a/sql/sql_db.h
+++ b/sql/sql_db.h
@@ -44,6 +44,8 @@ bool load_db_opt_by_name(THD *thd, const char *db_name,
 const CHARSET_INFO *get_default_db_collation(THD *thd, const char *db_name);
 bool my_dbopt_init(void);
 void my_dbopt_cleanup(void);
+void init_thd_db_read_only(THD *thd);
+bool is_thd_db_read_only_by_name(THD *thd, const char *db);
 
 #define MY_DB_OPT_FILE "db.opt"
 
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index d84bdd1be528..7caee92f9e37 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -108,6 +108,9 @@
 
 #include "sql_digest.h"
 
+#include "sql_db.h"      // init_thd_db_read_only
+                         // is_thd_db_read_only_by_name
+
 #ifdef HAVE_JEMALLOC
 #ifndef EMBEDDED_LIBRARY
 #include "jemalloc/jemalloc.h"
@@ -1113,11 +1116,11 @@ static my_bool deny_updates_if_read_only_option(THD *thd,
   if (lex->sql_command == SQLCOM_UPDATE_MULTI)
     DBUG_RETURN(FALSE);
 
-  const my_bool create_temp_tables= 
+  const my_bool create_temp_tables=
     (lex->sql_command == SQLCOM_CREATE_TABLE) &&
     (lex->create_info.options & HA_LEX_CREATE_TMP_TABLE);
 
-  const my_bool drop_temp_tables= 
+  const my_bool drop_temp_tables=
     (lex->sql_command == SQLCOM_DROP_TABLE) &&
     lex->drop_temporary;
 
@@ -1143,6 +1146,75 @@ static my_bool deny_updates_if_read_only_option(THD *thd,
   DBUG_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_LIST *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);
+
+  if (!my_hash_inited(&thd->db_read_only_hash))
+    init_thd_db_read_only(thd);
+
+  mysql_mutex_lock(&thd->LOCK_thd_db_read_only_hash);
+
+  if (thd->db_read_only_hash.records)
+  {
+    /* Check if the database to be dropped is read_only */
+    if (lex->sql_command == SQLCOM_DROP_DB &&
+        is_thd_db_read_only_by_name(thd, lex->name.str))
+      db_name = lex->name.str;
+
+    for (TABLE_LIST *table = all_tables; !db_name && table;
+         table = table->next_global)
+    {
+      DBUG_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;
+    }
+  }
+
+  mysql_mutex_unlock(&thd->LOCK_thd_db_read_only_hash);
+
+  DBUG_RETURN(db_name);
+}
+
 
 #ifdef HAVE_MY_TIMER
 /**
@@ -3021,6 +3093,19 @@ mysql_execute_command(THD *thd,
     /* Statement transaction still should not be started. */
     DBUG_ASSERT(thd->transaction.stmt.is_empty());
 
+    /*
+     * 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.");
+      DBUG_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_show.cc b/sql/sql_show.cc
index 9d04e547269a..dfc2880f89e7 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -1022,6 +1022,7 @@ 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));
 
+  bool enclose_comment = false;
   if (create.default_table_charset)
   {
     buffer.append(STRING_WITH_LEN(" /*!40100"));
@@ -1032,8 +1033,26 @@ bool mysqld_show_create_db(THD *thd, char *dbname,
       buffer.append(STRING_WITH_LEN(" COLLATE "));
       buffer.append(create.default_table_charset->name);
     }
-    buffer.append(STRING_WITH_LEN(" */"));
+    enclose_comment = true;
+  }
+
+  if (create.db_read_only)
+  {
+    if (!enclose_comment)
+    {
+      buffer.append(STRING_WITH_LEN(" /*"));
+      enclose_comment = true;
+    }
+
+    if (create.db_read_only > 1)
+      buffer.append(STRING_WITH_LEN(" SUPER_READ_ONLY"));
+    else
+      buffer.append(STRING_WITH_LEN(" READ_ONLY"));
   }
+
+  if (enclose_comment)
+    buffer.append(STRING_WITH_LEN(" */"));
+
   protocol->store(buffer.ptr(), buffer.length(), buffer.charset());
 
   if (protocol->write())
@@ -4516,11 +4535,73 @@ bool store_schema_shemata(THD* thd, TABLE *table, LEX_STRING *db_name,
   return schema_table_store_record(thd, table);
 }
 
+// This is called by per database read_only code path.  We fetch all database
+// directories and load db options.  This function is similar to
+// fill_schema_schemata() but without storing into schema table.
+int fetch_schema_schemata(THD *thd)
+{
+  /*
+    TODO: fill_schema_schemata() is called when new client is connected.
+    Returning error status in this case leads to client hangup.
+  */
+
+  /*
+   * A temporary class is created to free tmp_mem_root when we return from
+   * this function, since we have 'return' from this function from many
+   * places. This is just to avoid goto.
+   */
+  class free_tmp_mem_root
+  {
+  public:
+    free_tmp_mem_root()
+    {
+      init_sql_alloc(&tmp_mem_root, TABLE_ALLOC_BLOCK_SIZE, 0);
+    }
+    ~free_tmp_mem_root()
+    {
+      free_root(&tmp_mem_root, MYF(0));
+    }
+    MEM_ROOT tmp_mem_root;
+  };
+
+  free_tmp_mem_root dummy_member;
+
+  LOOKUP_FIELD_VALUES lookup_field_vals;
+  List<LEX_STRING> db_names;
+  LEX_STRING *db_name;
+  bool with_i_schema;
+  HA_CREATE_INFO create;
+
+#ifndef NO_EMBEDDED_ACCESS_CHECKS
+  Security_context *sctx= thd->security_ctx;
+#endif
+  DBUG_ENTER("fetch_schema_schemata");
+
+  // Set lookup_field_vals to empty since we will fetch all db opts
+  memset(&lookup_field_vals, 0, sizeof(LOOKUP_FIELD_VALUES));
+  if (make_db_list(thd, &db_names, &lookup_field_vals,
+                   &with_i_schema, &dummy_member.tmp_mem_root))
+    DBUG_RETURN(1);
+
+  List_iterator_fast<LEX_STRING> it(db_names);
+  while ((db_name=it++))
+  {
+#ifndef NO_EMBEDDED_ACCESS_CHECKS
+    if (sctx->master_access & (DB_ACLS | SHOW_DB_ACL) ||
+        acl_get(sctx->get_host()->ptr(), sctx->get_ip()->ptr(),
+          sctx->priv_user, db_name->str, 0) ||
+        !check_grant_db(thd, db_name->str))
+#endif
+      load_db_opt_by_name(thd, db_name->str, &create);
+  }
+  DBUG_RETURN(0);
+}
+
 
 int fill_schema_schemata(THD *thd, TABLE_LIST *tables, Item *cond)
 {
   /*
-    TODO: fill_schema_shemata() is called when new client is connected.
+    TODO: fill_schema_schemata() is called when new client is connected.
     Returning error status in this case leads to client hangup.
   */
 
@@ -4554,7 +4635,7 @@ int fill_schema_schemata(THD *thd, TABLE_LIST *tables, Item *cond)
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
   Security_context *sctx= thd->security_ctx;
 #endif
-  DBUG_ENTER("fill_schema_shemata");
+  DBUG_ENTER("fill_schema_schemata");
 
   if (get_lookup_field_values(thd, cond, tables, &lookup_field_vals))
     DBUG_RETURN(0);
diff --git a/sql/sql_show.h b/sql/sql_show.h
index 4e7ff911f874..f6ce82dcf079 100644
--- a/sql/sql_show.h
+++ b/sql/sql_show.h
@@ -218,4 +218,6 @@ bool ignore_db_dirs_process_additions();
 bool push_ignored_db_dir(char *path);
 extern char *opt_ignore_db_dirs;
 
+int fill_schema_schemata(THD *thd, TABLE_LIST *tables, Item *cond);
+int fetch_schema_schemata(THD *thd);
 #endif /* SQL_SHOW_H */
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index efc094a80b43..86d16cb2d640 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -1221,10 +1221,10 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 %lex-param { class THD *YYTHD }
 %pure-parser                                    /* We have threads */
 /*
-  Currently there are 180 shift/reduce conflicts.
+  Currently there are 182 shift/reduce conflicts.
   We should not introduce new conflicts any more.
 */
-%expect 180
+%expect 182
 
 /*
    Comments for TOKENS.
@@ -1792,6 +1792,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 %token  SUBSTRING                     /* SQL-2003-N */
 %token  SUM_SYM                       /* SQL-2003-N */
 %token  SUPER_SYM
+%token  SUPER_READ_ONLY_SYM
 %token  SUSPEND_SYM
 %token  SWAPS_SYM
 %token  SWITCHES_SYM
@@ -2162,6 +2163,7 @@ END_OF_INPUT
 
 %type <is_not_empty> opt_union_order_or_limit
 
+%type <num> read_only_opt boolean_val
 %%
 
 /*
@@ -2723,6 +2725,7 @@ create:
           {
             Lex->create_info.default_table_charset= NULL;
             Lex->create_info.used_fields= 0;
+            Lex->create_info.db_read_only= 0;
           }
           opt_create_database_options
           {
@@ -6188,6 +6191,7 @@ create_database_options:
 create_database_option:
           default_collation {}
         | default_charset {}
+        | db_read_only {}
         ;
 
 opt_table_options:
@@ -6469,6 +6473,32 @@ default_collation:
           }
         ;
 
+db_read_only:
+          read_only_opt equal boolean_val
+          {
+            Lex->create_info.db_read_only= 0; /* read_only = false */
+            if (($1 == 0 && $3 == 1) || ($1 == 1 && $3 == 0))
+            {
+              Lex->create_info.db_read_only= 1; /* read_only = true */
+                                                /* super_read_only = false */
+            }
+            else if ($1 == 1 && $3 == 1)
+            {
+              Lex->create_info.db_read_only= 2; /* super_read_only = true */
+            }
+          }
+        ;
+
+read_only_opt:
+          READ_ONLY_SYM { $$ = 0; }
+        | SUPER_READ_ONLY_SYM { $$ = 1; }
+        ;
+
+boolean_val:
+          FALSE_SYM { $$ = 0; }
+        | TRUE_SYM { $$ = 1; }
+        ;
+
 storage_engines:
           ident_or_text
           {
@@ -7668,6 +7698,7 @@ alter:
           {
             Lex->create_info.default_table_charset= NULL;
             Lex->create_info.used_fields= 0;
+            Lex->create_info.db_read_only= 0;
           }
           create_database_options
           {
@@ -15076,6 +15107,7 @@ keyword_sp:
         | SUBPARTITION_SYM         {}
         | SUBPARTITIONS_SYM        {}
         | SUPER_SYM                {}
+        | SUPER_READ_ONLY_SYM      {}
         | SUSPEND_SYM              {}
         | SWAPS_SYM                {}
         | SWITCHES_SYM             {}
diff --git a/storage/perfschema/pfs_autosize.cc b/storage/perfschema/pfs_autosize.cc
index 38bd36d8321a..a8f6204f5f2d 100644
--- a/storage/perfschema/pfs_autosize.cc
+++ b/storage/perfschema/pfs_autosize.cc
@@ -33,7 +33,7 @@ static const ulong fixed_file_instances= 200;
 static const ulong fixed_socket_instances= 10;
 static const ulong fixed_thread_instances= 50;
 
-static const ulong mutex_per_connection= 3;
+static const ulong mutex_per_connection= 4;
 static const ulong rwlock_per_connection= 1;
 static const ulong cond_per_connection= 2;
 static const ulong file_per_connection= 0;