diff --git a/libbinlogevents/include/binlog_event.h b/libbinlogevents/include/binlog_event.h
index e034c6b3d4fa..6a55e7928baa 100644
--- a/libbinlogevents/include/binlog_event.h
+++ b/libbinlogevents/include/binlog_event.h
@@ -117,6 +117,9 @@
 */
 #define MAX_TIME_ZONE_NAME_LENGTH (NAME_LEN + 1)
 
+/* Max length for the db-metadata option */
+#define DB_METADATA_MAX_LENGTH 1024
+
 /**
   Max number of possible extra bytes in a replication event compared to a
   packet (i.e. a query) sent from client to master;
diff --git a/mysql-test/r/db_metadata.result b/mysql-test/r/db_metadata.result
new file mode 100644
index 000000000000..73888706f3f9
--- /dev/null
+++ b/mysql-test/r/db_metadata.result
@@ -0,0 +1,157 @@
+create database 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 database test3 character set utf8;
+Warnings:
+Warning	3719	'utf8' is currently an alias for the character set UTF8MB3, but will be an alias for UTF8MB4 in a future release. Please consider using UTF8MB4 in order to be unambiguous.
+show create database test3;
+Database	Create Database
+test3	CREATE DATABASE `test3` /*!40100 DEFAULT CHARACTER SET utf8mb3 */ /*!80016 DEFAULT ENCRYPTION='N' */
+create database test4 read_only = true;
+show create database test4;
+Database	Create Database
+test4	CREATE DATABASE `test4` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */
+create database test5 db_metadata = "{\"shard\":\"test5_shard\"}";
+show create database test5;
+Database	Create Database
+test5	CREATE DATABASE `test5` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DB_METADATA '{"shard":"test5_shard"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+drop database test5;
+CREATE DATABASE `test5` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DB_METADATA '{"shard":"test5_shard"}' */ /*!80016 DEFAULT ENCRYPTION='N' */;
+show create database test5;
+Database	Create Database
+test5	CREATE DATABASE `test5` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DB_METADATA '{"shard":"test5_shard"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+create database test6 character set utf8 db_metadata = "{\"shard\":\"test6_shard\"}";
+Warnings:
+Warning	3719	'utf8' is currently an alias for the character set UTF8MB3, but will be an alias for UTF8MB4 in a future release. Please consider using UTF8MB4 in order to be unambiguous.
+show create database test6;
+Database	Create Database
+test6	CREATE DATABASE `test6` /*!40100 DEFAULT CHARACTER SET utf8mb3 DB_METADATA '{"shard":"test6_shard"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+create database test7 read_only = true db_metadata = "{\"shard\":\"test7_shard\"}";
+show create database test7;
+Database	Create Database
+test7	CREATE DATABASE `test7` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DB_METADATA '{"shard":"test7_shard"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+create database test8 character set utf8 read_only = true;
+Warnings:
+Warning	3719	'utf8' is currently an alias for the character set UTF8MB3, but will be an alias for UTF8MB4 in a future release. Please consider using UTF8MB4 in order to be unambiguous.
+show create database test8;
+Database	Create Database
+test8	CREATE DATABASE `test8` /*!40100 DEFAULT CHARACTER SET utf8mb3 */ /*!80016 DEFAULT ENCRYPTION='N' */
+create database test9 character set utf8 read_only = true db_metadata = "{\"shard\":\"test9_shard\"}";
+Warnings:
+Warning	3719	'utf8' is currently an alias for the character set UTF8MB3, but will be an alias for UTF8MB4 in a future release. Please consider using UTF8MB4 in order to be unambiguous.
+show create database test9;
+Database	Create Database
+test9	CREATE DATABASE `test9` /*!40100 DEFAULT CHARACTER SET utf8mb3 DB_METADATA '{"shard":"test9_shard"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test3 character set ascii;
+show create database test3;
+Database	Create Database
+test3	CREATE DATABASE `test3` /*!40100 DEFAULT CHARACTER SET ascii */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test4 read_only = true;
+show create database test4;
+Database	Create Database
+test4	CREATE DATABASE `test4` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci READ_ONLY */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test5 db_metadata = "{\"shard\":\"test5_shard_altered\"}";
+show create database test5;
+Database	Create Database
+test5	CREATE DATABASE `test5` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DB_METADATA '{"shard":"test5_shard_altered"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test5 character set ascii;
+show create database test5;
+Database	Create Database
+test5	CREATE DATABASE `test5` /*!40100 DEFAULT CHARACTER SET ascii DB_METADATA '{"shard":"test5_shard_altered"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test5 read_only = true;
+show create database test5;
+Database	Create Database
+test5	CREATE DATABASE `test5` /*!40100 DEFAULT CHARACTER SET ascii READ_ONLY DB_METADATA '{"shard":"test5_shard_altered"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test5 character set utf8 read_only = false;
+Warnings:
+Warning	3719	'utf8' is currently an alias for the character set UTF8MB3, but will be an alias for UTF8MB4 in a future release. Please consider using UTF8MB4 in order to be unambiguous.
+show create database test5;
+Database	Create Database
+test5	CREATE DATABASE `test5` /*!40100 DEFAULT CHARACTER SET utf8mb3 DB_METADATA '{"shard":"test5_shard_altered"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test5 db_metadata "";
+show create database test5;
+Database	Create Database
+test5	CREATE DATABASE `test5` /*!40100 DEFAULT CHARACTER SET utf8mb3 */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test5 db_metadata "{\"shard\":\"Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Fin\"}";
+show create database test5;
+Database	Create Database
+test5	CREATE DATABASE `test5` /*!40100 DEFAULT CHARACTER SET utf8mb3 DB_METADATA '{"shard":"Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Fin"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test5 db_metadata "{\"shard\":\"Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Fin\"}";;
+ERROR HY000: Metadata for the database is too long. Max length is 1024 bytes
+alter database test6 character set ascii db_metadata = "{\"shard\":\"test6_shard_altered\"}";
+show create database test6;
+Database	Create Database
+test6	CREATE DATABASE `test6` /*!40100 DEFAULT CHARACTER SET ascii DB_METADATA '{"shard":"test6_shard_altered"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test7 read_only = true db_metadata = "{\"shard\":\"test7_shard_altered\"}";
+show create database test7;
+Database	Create Database
+test7	CREATE DATABASE `test7` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci READ_ONLY DB_METADATA '{"shard":"test7_shard_altered"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test8 character set ascii read_only = true;
+show create database test8;
+Database	Create Database
+test8	CREATE DATABASE `test8` /*!40100 DEFAULT CHARACTER SET ascii READ_ONLY */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test9 character set ascii read_only = true db_metadata = "{\"shard\":\"test9_shard_altered\"}";
+show create database test9;
+Database	Create Database
+test9	CREATE DATABASE `test9` /*!40100 DEFAULT CHARACTER SET ascii READ_ONLY DB_METADATA '{"shard":"test9_shard_altered"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+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 mysql;
+Database	Create Database
+mysql	CREATE DATABASE `mysql` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database information_schema db_metadata "{\"shard\":\"is_shard\"}";
+ERROR 42000: Access denied for user 'root'@'localhost' to database 'information_schema'
+alter database test9 db_metadata = "invalid_json";
+ERROR HY000: Invalid JSON for DB_METADATA attribute: invalid_json.
+show create database test9;
+Database	Create Database
+test9	CREATE DATABASE `test9` /*!40100 DEFAULT CHARACTER SET ascii READ_ONLY DB_METADATA '{"shard":"test9_shard_altered"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+create database test10;
+show create database test10;
+Database	Create Database
+test10	CREATE DATABASE `test10` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test10 read_only = true;
+show create database test10;
+Database	Create Database
+test10	CREATE DATABASE `test10` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci READ_ONLY */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test10 db_metadata = "{\"shard\":\"test10_shard_altered\"}";
+show create database test10;
+Database	Create Database
+test10	CREATE DATABASE `test10` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci READ_ONLY DB_METADATA '{"shard":"test10_shard_altered"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+alter database test10 read_only = false;
+show create database test10;
+Database	Create Database
+test10	CREATE DATABASE `test10` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DB_METADATA '{"shard":"test10_shard_altered"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+create database test11 db_metadata = "{\'shard\':\'test11_shard\'}";
+ERROR HY000: Invalid JSON for DB_METADATA attribute: {'shard':'test11_shard'}.
+create database test12 db_metadata = "{\"sha'rd\":\"test12\\\"_shard\"}";
+show create database test12;
+Database	Create Database
+test12	CREATE DATABASE `test12` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DB_METADATA '{"sha''rd":"test12\\"_shard"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+drop database test12;
+CREATE DATABASE `test12` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DB_METADATA '{"sha''rd":"test12\\"_shard"}' */ /*!80016 DEFAULT ENCRYPTION='N' */;
+show create database test12;
+Database	Create Database
+test12	CREATE DATABASE `test12` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DB_METADATA '{"sha''rd":"test12\\"_shard"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+create database test13 db_metadata = '{"sha\'\\"rd":"test13\'_sh\\"ard"}';
+show create database test13;
+Database	Create Database
+test13	CREATE DATABASE `test13` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DB_METADATA '{"sha''\\"rd":"test13''_sh\\"ard"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+drop database test13;
+CREATE DATABASE `test13` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DB_METADATA '{"sha''\\"rd":"test13''_sh\\"ard"}' */ /*!80016 DEFAULT ENCRYPTION='N' */;
+show create database test13;
+Database	Create Database
+test13	CREATE DATABASE `test13` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DB_METADATA '{"sha''\\"rd":"test13''_sh\\"ard"}' */ /*!80016 DEFAULT ENCRYPTION='N' */
+drop database test2;
+drop database test3;
+drop database test4;
+drop database test5;
+drop database test6;
+drop database test7;
+drop database test8;
+drop database test9;
+drop database test10;
+drop database test12;
+drop database test13;
diff --git a/mysql-test/r/information_schema_keywords.result b/mysql-test/r/information_schema_keywords.result
index 95606caf965a..3e7f7f93700f 100644
--- a/mysql-test/r/information_schema_keywords.result
+++ b/mysql-test/r/information_schema_keywords.result
@@ -122,6 +122,7 @@ DAY_HOUR	1
 DAY_MICROSECOND	1
 DAY_MINUTE	1
 DAY_SECOND	1
+DB_METADATA	0
 DEALLOCATE	0
 DEC	1
 DECIMAL	1
diff --git a/mysql-test/t/db_metadata.test b/mysql-test/t/db_metadata.test
new file mode 100644
index 000000000000..ae1850b4adea
--- /dev/null
+++ b/mysql-test/t/db_metadata.test
@@ -0,0 +1,159 @@
+# Test per-database database-metadata attribute
+
+let $MYSQLD_DATADIR= `select @@datadir`;
+
+# create database without any options
+create database test2;
+show create database test2;
+
+# create database with character set
+create database test3 character set utf8;
+show create database test3;
+
+# create database with read only
+create database test4 read_only = true;
+show create database test4;
+
+# create database with db metadata
+create database test5 db_metadata = "{\"shard\":\"test5_shard\"}";
+show create database test5;
+
+# verify show create is valid
+--let $db_create=query_get_value(show create database test5, 'Create Database', 1)
+drop database test5;
+eval $db_create;
+show create database test5;
+
+# create database with character set and db metadata
+create database test6 character set utf8 db_metadata = "{\"shard\":\"test6_shard\"}";
+show create database test6;
+
+# create database with read only and db metadata
+create database test7 read_only = true db_metadata = "{\"shard\":\"test7_shard\"}";
+show create database test7;
+
+# create database with character set and read only
+create database test8 character set utf8 read_only = true;
+show create database test8;
+
+# create database with character set, read only and db metadata
+create database test9 character set utf8 read_only = true db_metadata = "{\"shard\":\"test9_shard\"}";
+show create database test9;
+
+# alter database tests
+
+# alter database character set
+alter database test3 character set ascii;
+show create database test3;
+
+# alter database read only
+alter database test4 read_only = true;
+show create database test4;
+
+# alter database db metadata
+alter database test5 db_metadata = "{\"shard\":\"test5_shard_altered\"}";
+show create database test5;
+
+# alter database character set but keep db metadata intact
+alter database test5 character set ascii;
+show create database test5;
+
+# alter database read only but keep db metadata intact
+alter database test5 read_only = true;
+show create database test5;
+
+# alter database character set and read only but keep db metadata intact
+alter database test5 character set utf8 read_only = false;
+show create database test5;
+
+# alter database reset db metadata
+alter database test5 db_metadata "";
+show create database test5;
+
+# alter database set db metadata to max length string
+alter database test5 db_metadata "{\"shard\":\"Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Fin\"}";
+show create database test5;
+
+# alter database exceed db metadata max length
+--error ER_DB_METADATA_TOO_LONG
+--eval alter database test5 db_metadata \"{\\\"shard\\\":\\\"Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Really long shard name. Fin\\\"}\";
+
+# alter database character set and db metadata
+alter database test6 character set ascii db_metadata = "{\"shard\":\"test6_shard_altered\"}";
+show create database test6;
+
+# alter database read only and db metadata
+alter database test7 read_only = true db_metadata = "{\"shard\":\"test7_shard_altered\"}";
+show create database test7;
+
+# alter database character set and read only
+alter database test8 character set ascii read_only = true;
+show create database test8;
+
+# alter database character set, read only and db metadata
+alter database test9 character set ascii read_only = true db_metadata = "{\"shard\":\"test9_shard_altered\"}";
+show create database test9;
+
+# ensure information_schema and mysql tables are intact
+show create database information_schema;
+show create database mysql;
+
+--error ER_DBACCESS_DENIED_ERROR
+alter database information_schema db_metadata "{\"shard\":\"is_shard\"}";
+
+# alter database with invalid JSON string for db_metadata
+ --error ER_DB_METADATA_INVALID_JSON
+alter database test9 db_metadata = "invalid_json";
+show create database test9;
+
+
+
+# create database without any options
+create database test10;
+show create database test10;
+
+# alter database read only
+alter database test10 read_only = true;
+show create database test10;
+
+# alter database db metadata
+alter database test10 db_metadata = "{\"shard\":\"test10_shard_altered\"}";
+show create database test10;
+
+# alter database read only
+alter database test10 read_only = false;
+show create database test10;
+
+# create database with single quotation
+--error ER_DB_METADATA_INVALID_JSON
+create database test11 db_metadata = "{\'shard\':\'test11_shard\'}";
+
+# create database with double quote around metadata with single and double quotes in value
+create database test12 db_metadata = "{\"sha'rd\":\"test12\\\"_shard\"}";
+show create database test12;
+--let $db_create=query_get_value(show create database test12, 'Create Database', 1)
+drop database test12;
+eval $db_create;
+show create database test12;
+
+# create database with single quote around metadata with single and double quotes in value
+create database test13 db_metadata = '{"sha\'\\"rd":"test13\'_sh\\"ard"}';
+show create database test13;
+--let $db_create=query_get_value(show create database test13, 'Create Database', 1)
+drop database test13;
+eval $db_create;
+show create database test13;
+
+
+# cleanup
+drop database test2;
+drop database test3;
+drop database test4;
+drop database test5;
+drop database test6;
+drop database test7;
+drop database test8;
+drop database test9;
+drop database test10;
+drop database test12;
+drop database test13;
diff --git a/share/messages_to_clients.txt b/share/messages_to_clients.txt
index 312c62f69c7c..dc72c9099290 100644
--- a/share/messages_to_clients.txt
+++ b/share/messages_to_clients.txt
@@ -10168,14 +10168,14 @@ ER_NP_FAILED
 ER_CANT_SET_DEPENDENCY_REPLICATION_WITHOUT_IDEMPOTENT_RECOVERY
   eng "@@GLOBAL.MTS_DEPENDENCY_REPLICATION cannot be set when @@GLOBAL.SLAVE_USE_IDEMPOTENT_FOR_RECOVERY = NO"
 
-ER_PLACEHOLDER_50063
-  eng "Placeholder"
+ER_DB_METADATA_TOO_LONG
+  eng "Metadata for the database is too long. Max length is %d bytes"
 
-ER_PLACEHOLDER_50064
-  eng "Placeholder"
+ER_DB_METADATA_INVALID_JSON
+  eng "Invalid JSON for DB_METADATA attribute: %s."
 
-ER_PLACEHOLDER_50065
-  eng "Placeholder"
+ER_DB_METADATA_READ_ERROR
+  eng "Error reading db metadata option: '%-.64s'"
 
 ER_UPDATES_WITH_EXPLICIT_SNAPSHOT
   eng "Can't execute updates when an explicit snapshot is associated with the connection using CREATE|ATTACH EXPLICIT [ENGINE] SNAPSHOT"
diff --git a/sql/dd/dd_schema.cc b/sql/dd/dd_schema.cc
index 4042607506c0..3a803c3d2f78 100644
--- a/sql/dd/dd_schema.cc
+++ b/sql/dd/dd_schema.cc
@@ -59,7 +59,8 @@ bool schema_exists(THD *thd, const char *schema_name, bool *exists) {
 }
 
 bool create_schema(THD *thd, const char *schema_name,
-                   const CHARSET_INFO *charset_info, bool default_encryption) {
+                   const CHARSET_INFO *charset_info, bool default_encryption,
+                   const char *db_metadata) {
   // Create dd::Schema object.
   std::unique_ptr<dd::Schema> schema(dd::create_object<dd::Schema>());
 
@@ -68,6 +69,7 @@ bool create_schema(THD *thd, const char *schema_name,
   assert(charset_info);
   schema->set_default_collation_id(charset_info->number);
   schema->set_default_encryption(default_encryption);
+  schema->set_db_metadata(db_metadata != nullptr ? db_metadata : "");
 
   // Get statement start time.
   ulonglong ull_curtime = my_time_t_to_ull_datetime(thd->query_start_in_secs());
diff --git a/sql/dd/dd_schema.h b/sql/dd/dd_schema.h
index 8614ac58987c..1a1ab6656a1a 100644
--- a/sql/dd/dd_schema.h
+++ b/sql/dd/dd_schema.h
@@ -47,7 +47,7 @@ bool schema_exists(THD *thd, const char *schema_name, bool *exists);
 /** Create a schema record into dd.schemata. */
 bool create_schema(THD *thd, const char *schema_name,
                    const CHARSET_INFO *charset_info,
-                   const bool default_encryption);
+                   const bool default_encryption, const char *db_metadata);
 
 /**
   Acquire MDL on schema name.
diff --git a/sql/dd/impl/types/schema_impl.cc b/sql/dd/impl/types/schema_impl.cc
index d945ce269ee9..dba3ab93e8d0 100644
--- a/sql/dd/impl/types/schema_impl.cc
+++ b/sql/dd/impl/types/schema_impl.cc
@@ -74,7 +74,7 @@ namespace dd {
 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};
+    "db_metadata", mysql_db_read_only_options_key, fb_read_only_options_key};
 
 ///////////////////////////////////////////////////////////////////////////
 // Schema_impl implementation.
@@ -112,6 +112,29 @@ void Schema_impl::set_read_only(bool state) {
   options().set(mysql_db_read_only_options_key, state);
 }
 
+static constexpr auto db_metadata_key = "db_metadata";
+
+bool Schema_impl::set_db_metadata(const String_type &metadata) {
+  if (metadata.empty()) {
+    if (options().exists(db_metadata_key)) {
+      options().remove(db_metadata_key);
+    }
+  } else {
+    options().set(db_metadata_key, metadata);
+  }
+  return true;
+}
+
+String_type Schema_impl::get_db_metadata() const noexcept {
+  if (options().exists(db_metadata_key)) {
+    bool ret;
+    String_type value;
+    ret = options().get(db_metadata_key, &value);
+    if (!ret) return value;
+  }
+  return "";
+}
+
 ///////////////////////////////////////////////////////////////////////////
 
 void Schema_impl::set_db_read_only(int state) {
diff --git a/sql/dd/impl/types/schema_impl.h b/sql/dd/impl/types/schema_impl.h
index 48637f25d78c..e7a13755f2cb 100644
--- a/sql/dd/impl/types/schema_impl.h
+++ b/sql/dd/impl/types/schema_impl.h
@@ -153,6 +153,9 @@ class Schema_impl : public Entity_object_impl, public Schema {
   // options.
   /////////////////////////////////////////////////////////////////////////
 
+  virtual bool set_db_metadata(const String_type &metadata) override;
+  virtual String_type get_db_metadata() const noexcept override;
+
   const Properties &options() const override { return m_options; }
 
   Properties &options() override { return m_options; }
@@ -202,16 +205,16 @@ class Schema_impl : public Entity_object_impl, public Schema {
  public:
   void debug_print(String_type &outb) const override {
     char outbuf[1024];
-    sprintf(outbuf,
-            "SCHEMA OBJECT: id= {OID: %lld}, name= %s, "
-            "collation_id={OID: %lld},"
-            "m_created= %llu, m_last_altered= %llu,"
-            "m_default_encryption= %d, "
-            "se_private_data= %s, options= %s",
-            id(), name().c_str(), m_default_collation_id, m_created,
-            m_last_altered, static_cast<int>(m_default_encryption),
-            m_se_private_data.raw_string().c_str(),
-            m_options.raw_string().c_str());
+    snprintf(outbuf, sizeof(outbuf) / sizeof(char),
+             "SCHEMA OBJECT: id= {OID: %lld}, name= %s, "
+             "collation_id={OID: %lld},"
+             "m_created= %llu, m_last_altered= %llu,"
+             "m_default_encryption= %d, "
+             "se_private_data= %s, options= %s",
+             id(), name().c_str(), m_default_collation_id, m_created,
+             m_last_altered, static_cast<int>(m_default_encryption),
+             m_se_private_data.raw_string().c_str(),
+             m_options.raw_string().c_str());
     outb = String_type(outbuf);
   }
 
diff --git a/sql/dd/types/schema.h b/sql/dd/types/schema.h
index 5a7ddea538e4..c7f8455132a0 100644
--- a/sql/dd/types/schema.h
+++ b/sql/dd/types/schema.h
@@ -139,6 +139,9 @@ class Schema : virtual public Entity_object {
   virtual bool set_se_private_data(const String_type &se_private_data_raw) = 0;
   virtual bool set_se_private_data(const Properties &se_private_data) = 0;
 
+  virtual bool set_db_metadata(const String_type &metadata) = 0;
+  virtual String_type get_db_metadata() const noexcept = 0;
+
  public:
   virtual Event *create_event(THD *thd) const = 0;
 
diff --git a/sql/dd/upgrade_57/schema.cc b/sql/dd/upgrade_57/schema.cc
index 4f21072c5d7f..386dd6f775c8 100644
--- a/sql/dd/upgrade_57/schema.cc
+++ b/sql/dd/upgrade_57/schema.cc
@@ -178,7 +178,7 @@ bool migrate_schema_to_dd(THD *thd, const char *dbname) {
   // Disable autocommit option
   Disable_autocommit_guard autocommit_guard(thd);
 
-  if (dd::create_schema(thd, schema_name, schema_charset, false)) {
+  if (dd::create_schema(thd, schema_name, schema_charset, false, nullptr)) {
     trans_rollback_stmt(thd);
     // Full rollback in case we have THD::transaction_rollback_request.
     trans_rollback(thd);
diff --git a/sql/handler.h b/sql/handler.h
index 1c4baa3c42bf..28ae15fd98b1 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -72,7 +72,8 @@
 #include "sql/sql_const.h"       // SHOW_COMP_OPTION
 #include "sql/sql_list.h"        // SQL_I_List
 #include "sql/sql_plugin_ref.h"  // plugin_ref
-#include "thr_lock.h"            // thr_lock_type
+#include "sql_string.h"
+#include "thr_lock.h"  // thr_lock_type
 #include "typelib.h"
 
 class Alter_info;
@@ -3025,6 +3026,7 @@ struct HA_CREATE_INFO {
   const CHARSET_INFO *default_table_charset{nullptr};
   bool schema_read_only{false};
   enum enum_db_read_only db_read_only { DB_READ_ONLY_NULL };
+  String db_metadata;
   LEX_STRING connect_string{nullptr, 0};
   const char *password{nullptr};
   const char *tablespace{nullptr};
diff --git a/sql/lex.h b/sql/lex.h
index ea16b69dd230..7a822ed77517 100644
--- a/sql/lex.h
+++ b/sql/lex.h
@@ -191,6 +191,7 @@ static const SYMBOL symbols[] = {
     {SYM("DAY_MICROSECOND", DAY_MICROSECOND_SYM)},
     {SYM("DAY_MINUTE", DAY_MINUTE_SYM)},
     {SYM("DAY_SECOND", DAY_SECOND_SYM)},
+    {SYM("DB_METADATA", DB_METADATA_SYM)},
     {SYM("DEALLOCATE", DEALLOCATE_SYM)},
     {SYM("DEC", DECIMAL_SYM)},
     {SYM("DECIMAL", DECIMAL_SYM)},
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index 6b1c1a6a84ac..32951af5da1e 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -577,7 +577,7 @@ bool mysql_create_db(THD *thd, const char *db, HA_CREATE_INFO *create_info) {
     set_db_default_charset(thd, create_info);
 
     if (dd::create_schema(thd, db, create_info->default_table_charset,
-                          encrypt_schema)) {
+                          encrypt_schema, create_info->db_metadata.ptr())) {
       /*
         We could be here due an deadlock or some error reported
         by DD API framework. We remove the database directory
@@ -718,6 +718,10 @@ bool mysql_alter_db(THD *thd, const char *db, HA_CREATE_INFO *create_info) {
     }
   }
 
+  if (create_info->db_metadata.ptr() != nullptr) {
+    schema->set_db_metadata(create_info->db_metadata.ptr());
+  }
+
   // Set encryption type.
   if (create_info->used_fields & HA_CREATE_USED_DEFAULT_ENCRYPTION) {
     assert(create_info->encrypt_type.length > 0);
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index b6832e44cfd9..d5f4bc644c7c 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -1384,6 +1384,8 @@ bool mysqld_show_create_db(THD *thd, char *dbname,
   }
 
   bool is_encrypted_schema = false;
+  std::string metadata;
+
   if (is_infoschema_db(dbname)) {
     create.default_table_charset = system_charset_info;
   } else {
@@ -1405,6 +1407,7 @@ bool mysqld_show_create_db(THD *thd, char *dbname,
     }
 
     schema_read_only = schema->read_only();
+    metadata = schema->get_db_metadata().c_str();
 
     create.db_read_only =
         static_cast<enum_db_read_only>(schema->get_db_read_only());
@@ -1447,6 +1450,11 @@ bool mysqld_show_create_db(THD *thd, char *dbname,
       buffer.append(STRING_WITH_LEN(" SUPER_READ_ONLY"));
     }
 
+    if (!metadata.empty()) {
+      buffer.append(STRING_WITH_LEN(" DB_METADATA "));
+      append_unescaped(&buffer, metadata.c_str(), metadata.size());
+    }
+
     buffer.append(STRING_WITH_LEN(" */"));
   }
   buffer.append(STRING_WITH_LEN(" /*!80016"));
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index c21205b7d72e..f91e071632d2 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -56,6 +56,10 @@ Note: YYTHD is passed as an argument to yyparse(), and subsequently to yylex().
 #include <string>
 #include <type_traits>
 #include <utility>
+#include <sstream>
+#define BOOST_BIND_GLOBAL_PLACEHOLDERS
+#include <boost/property_tree/ptree.hpp>
+#include <boost/property_tree/json_parser.hpp>
 
 #include "field_types.h"
 #include "ft_global.h"
@@ -510,7 +514,7 @@ void warn_about_deprecated_binary(THD *thd)
   2. We should not introduce new shift/reduce conflicts any more.
 */
 
-%expect 65
+%expect 66
 
 /*
    MAINTAINER:
@@ -1415,6 +1419,7 @@ void warn_about_deprecated_binary(THD *thd)
 %token<lexer.keyword> GTID_SYM 1251
 %token<lexer.keyword> GTID_EXECUTED 1252           /* MYSQL */
 %token<lexer.keyword> SUPER_READ_ONLY_SYM 1253
+%token<lexer.keyword> DB_METADATA_SYM 1254
 
 /*
   Resolve column attribute ambiguity -- force precedence of "UNIQUE KEY" against
@@ -6681,9 +6686,8 @@ create_database_option:
             Lex->create_info->encrypt_type= $1;
             Lex->create_info->used_fields |= HA_CREATE_USED_DEFAULT_ENCRYPTION;
           }
-        | db_read_only
-          {
-          }
+        | db_read_only {}
+        | db_metadata_str {}
         ;
 
 opt_if_not_exists:
@@ -6951,6 +6955,35 @@ boolean_val:
         | TRUE_SYM { $$ = 1; }
         ;
 
+db_metadata_str:
+          DB_METADATA_SYM opt_equal TEXT_STRING_sys
+          {
+            if ($3.length > DB_METADATA_MAX_LENGTH)
+            {
+              my_error(ER_DB_METADATA_TOO_LONG, MYF(0), DB_METADATA_MAX_LENGTH);
+              MYSQL_YYABORT;
+            }
+            /* Verify that a valid JSON is provided */
+            if ($3.length > 0)
+            {
+              boost::property_tree::ptree db_metadata_root;
+              std::istringstream is($3.str);
+              try
+              {
+                boost::property_tree::json_parser::read_json(is,
+                                                             db_metadata_root);
+              }
+              catch (const std::exception&)
+              {
+                my_error(ER_DB_METADATA_INVALID_JSON, MYF(0), $3.str);
+                MYSQL_YYABORT;
+              }
+            }
+            Lex->create_info->db_metadata= String($3.str, $3.length,
+              &my_charset_bin);
+          }
+        ;
+
 row_types:
           DEFAULT_SYM    { $$= ROW_TYPE_DEFAULT; }
         | FIXED_SYM      { $$= ROW_TYPE_FIXED; }
@@ -15270,6 +15303,7 @@ ident_keywords_ambiguous_2_labels:
         | COMMENT_SYM
         | COMMIT_SYM
         | CONTAINS_SYM
+        | DB_METADATA_SYM
         | DEALLOCATE_SYM
         | DO_SYM
         | END