MDEV-34716 Allow arbitrary options in CREATE SERVER

The existing syntax for CREATE SERVER

CREATE [OR REPLACE] SERVER [IF NOT EXISTS] server_name
    FOREIGN DATA WRAPPER wrapper_name
    OPTIONS (option [, option] ...)

option:
  { HOST character-literal
  | DATABASE character-literal
  | USER character-literal
  | PASSWORD character-literal
  | SOCKET character-literal
  | OWNER character-literal
  | PORT numeric-literal }

With this change we have:

option:
  { HOST character-literal
  | DATABASE character-literal
  | USER character-literal
  | PASSWORD character-literal
  | SOCKET character-literal
  | OWNER character-literal
  | PORT numeric-literal
  | PORT quoted-numerical-literal
  | identifier character-literal}

We store these options as a JSON field in the mysql.servers system
table. We retain the restriction that PORT needs to be a number, but
also allow it to be a quoted number, so that SHOW CREATE SERVER can be
used for dumping. Without an accompanied implementation of SHOW CREATE
SERVER, some mysqldump tests will fail. Therefore this commit should
be immediately followed by the one implementating SHOW CREATE SERVER,
with testing covering both.

mysql-test/main/information_schema.result

@@ -914,6 +914,7 @@ sys	processlist	current_statement
 sys	processlist	last_statement
 sys	schema_auto_increment_columns	column_type
 sys	schema_table_lock_waits	waiting_query
+mysql	servers	Options
 sys	session	current_statement
 sys	session	last_statement
 sys	statement_analysis	query
@@ -2440,6 +2441,7 @@ using (CONSTRAINT_CATALOG, CONSTRAINT_SCHEMA, TABLE_NAME, CONSTRAINT_NAME)
 ;
 TABLE_SCHEMA	TABLE_NAME	CONSTRAINT_NAME	CHECK_CLAUSE
 mysql	global_priv	Priv	json_valid(`Priv`)
+mysql	servers	Options	json_valid(`Options`)
 test	t	a	`i` > 0
 select
 tc.TABLE_SCHEMA,
@@ -2452,6 +2454,7 @@ using (CONSTRAINT_CATALOG, CONSTRAINT_SCHEMA, TABLE_NAME, CONSTRAINT_NAME)
 ;
 TABLE_SCHEMA	TABLE_NAME	CONSTRAINT_NAME	CHECK_CLAUSE
 mysql	global_priv	Priv	json_valid(`Priv`)
+mysql	servers	Options	json_valid(`Options`)
 test	t	a	`i` > 0
 select
 tc.TABLE_SCHEMA,
@@ -2463,6 +2466,7 @@ NATURAL join information_schema.CHECK_CONSTRAINTS cc
 ;
 TABLE_SCHEMA	TABLE_NAME	CONSTRAINT_NAME	CHECK_CLAUSE
 mysql	global_priv	Priv	json_valid(`Priv`)
+mysql	servers	Options	json_valid(`Options`)
 test	t	a	`i` > 0
 select
 tc.TABLE_SCHEMA,
@@ -2474,6 +2478,7 @@ NATURAL join information_schema.TABLE_CONSTRAINTS tc
 ;
 TABLE_SCHEMA	TABLE_NAME	CONSTRAINT_NAME	CHECK_CLAUSE
 mysql	global_priv	Priv	json_valid(`Priv`)
+mysql	servers	Options	json_valid(`Options`)
 test	t	a	`i` > 0
 select
 tc.TABLE_SCHEMA,
@@ -2488,6 +2493,7 @@ using (CONSTRAINT_CATALOG, CONSTRAINT_SCHEMA, TABLE_NAME, CONSTRAINT_NAME)
 ;
 TABLE_SCHEMA	TABLE_NAME	CONSTRAINT_NAME	CHECK_CLAUSE	CONSTRAINT_CATALOG	CONSTRAINT_SCHEMA
 mysql	global_priv	Priv	json_valid(`Priv`)	def	mysql
+mysql	servers	Options	json_valid(`Options`)	def	mysql
 test	t	a	`i` > 0	def	test
 drop table t;
 #

mysql-test/main/servers.result

@@ -4,25 +4,25 @@ set sql_mode="";
 #
 CREATE SERVER s1 FOREIGN DATA WRAPPER mysql OPTIONS(HOST 'localhost');
 SELECT * FROM mysql.servers;
-Server_name	Host	Db	Username	Password	Port	Socket	Wrapper	Owner
-s1	localhost				3306		mysql	
+Server_name	Host	Db	Username	Password	Port	Socket	Wrapper	Owner	Options
+s1	localhost				3306		mysql		{"HOST": "localhost"}
 DROP SERVER s1;
 CREATE SERVER s1 FOREIGN DATA WRAPPER foo OPTIONS(USER 'bar');
 SELECT * FROM mysql.servers;
-Server_name	Host	Db	Username	Password	Port	Socket	Wrapper	Owner
-s1			bar		0		foo	
+Server_name	Host	Db	Username	Password	Port	Socket	Wrapper	Owner	Options
+s1			bar		0		foo		{"USER": "bar"}
 DROP SERVER s1;
 CREATE SERVER s1 FOREIGN DATA WRAPPER mysql OPTIONS(USER 'bar');
 ERROR HY000: Can't create federated table. Foreign data src error:  either HOST or SOCKET must be set
 CREATE SERVER s1 FOREIGN DATA WRAPPER mysql OPTIONS(HOST 'bar');
 SELECT * FROM mysql.servers;
-Server_name	Host	Db	Username	Password	Port	Socket	Wrapper	Owner
-s1	bar				3306		mysql	
+Server_name	Host	Db	Username	Password	Port	Socket	Wrapper	Owner	Options
+s1	bar				3306		mysql		{"HOST": "bar"}
 DROP SERVER s1;
 CREATE SERVER s1 FOREIGN DATA WRAPPER mysql OPTIONS(SOCKET 'bar');
 SELECT * FROM mysql.servers;
-Server_name	Host	Db	Username	Password	Port	Socket	Wrapper	Owner
-s1					3306	bar	mysql	
+Server_name	Host	Db	Username	Password	Port	Socket	Wrapper	Owner	Options
+s1					3306	bar	mysql		{"SOCKET": "bar"}
 DROP SERVER s1;
 CREATE SERVER s1 FOREIGN DATA WRAPPER mysql OPTIONS(SOCKET '/tmp/1234567890_1234567890_1234567890_1234567890_1234567890_1234567890.sock');
 SELECT Socket FROM mysql.servers where Server_name = 's1';

mysql-test/main/system_mysql_db.result

@@ -133,6 +133,7 @@ servers	CREATE TABLE `servers` (
   `Socket` char(108) NOT NULL DEFAULT '',
   `Wrapper` char(64) NOT NULL DEFAULT '',
   `Owner` varchar(512) NOT NULL DEFAULT '',
+  `Options` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL DEFAULT '{}' CHECK (json_valid(`Options`)),
   PRIMARY KEY (`Server_name`)
 ) ENGINE=Aria DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_general_ci PAGE_CHECKSUM=1 TRANSACTIONAL=1 COMMENT='MySQL Foreign Servers table'
 show create table proc;

mysql-test/main/system_mysql_db_fix40123.result

@@ -171,6 +171,7 @@ servers	CREATE TABLE `servers` (
   `Socket` char(108) NOT NULL DEFAULT '',
   `Wrapper` char(64) NOT NULL DEFAULT '',
   `Owner` varchar(512) NOT NULL DEFAULT '',
+  `Options` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL DEFAULT '{}' CHECK (json_valid(`Options`)),
   PRIMARY KEY (`Server_name`)
 ) ENGINE=Aria DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_general_ci PAGE_CHECKSUM=1 TRANSACTIONAL=1 COMMENT='MySQL Foreign Servers table'
 show create table proc;

mysql-test/main/system_mysql_db_fix50030.result

@@ -175,6 +175,7 @@ servers	CREATE TABLE `servers` (
   `Socket` char(108) NOT NULL DEFAULT '',
   `Wrapper` char(64) NOT NULL DEFAULT '',
   `Owner` varchar(512) NOT NULL DEFAULT '',
+  `Options` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL DEFAULT '{}' CHECK (json_valid(`Options`)),
   PRIMARY KEY (`Server_name`)
 ) ENGINE=Aria DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_general_ci PAGE_CHECKSUM=1 TRANSACTIONAL=1 COMMENT='MySQL Foreign Servers table'
 show create table proc;

mysql-test/main/system_mysql_db_fix50117.result

@@ -155,6 +155,7 @@ servers	CREATE TABLE `servers` (
   `Socket` char(108) NOT NULL DEFAULT '',
   `Wrapper` char(64) NOT NULL DEFAULT '',
   `Owner` varchar(512) NOT NULL DEFAULT '',
+  `Options` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL DEFAULT '{}' CHECK (json_valid(`Options`)),
   PRIMARY KEY (`Server_name`)
 ) ENGINE=Aria DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_general_ci PAGE_CHECKSUM=1 TRANSACTIONAL=1 COMMENT='MySQL Foreign Servers table'
 show create table proc;

mysql-test/main/system_mysql_db_fix50568.result

@@ -176,6 +176,7 @@ servers	CREATE TABLE `servers` (
   `Socket` char(108) NOT NULL DEFAULT '',
   `Wrapper` char(64) NOT NULL DEFAULT '',
   `Owner` varchar(512) NOT NULL DEFAULT '',
+  `Options` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL DEFAULT '{}' CHECK (json_valid(`Options`)),
   PRIMARY KEY (`Server_name`)
 ) ENGINE=Aria DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_general_ci PAGE_CHECKSUM=1 TRANSACTIONAL=1 COMMENT='MySQL Foreign Servers table'
 show create table proc;

mysql-test/suite/federated/federated_server.result

@@ -61,9 +61,9 @@ PORT SLAVE_PORT,
 SOCKET '',
 OWNER 'root');
 select * from mysql.servers order by db;
-Server_name	Host	Db	Username	Password	Port	Socket	Wrapper	Owner
-server_one	127.0.0.1	first_db	root		SLAVE_PORT		mysql	root
-server_two	127.0.0.1	second_db	root		SLAVE_PORT		mysql	root
+Server_name	Host	Db	Username	Password	Port	Socket	Wrapper	Owner	Options
+server_one	127.0.0.1	first_db	root		SLAVE_PORT		mysql	root	{"HOST": "127.0.0.1", "DATABASE": "first_db", "USER": "root", "PASSWORD": "", "PORT": "SLAVE_PORT", "SOCKET": "", "OWNER": "root"}
+server_two	127.0.0.1	second_db	root		SLAVE_PORT		mysql	root	{"HOST": "127.0.0.1", "DATABASE": "second_db", "USER": "root", "PASSWORD": "", "PORT": "SLAVE_PORT", "SOCKET": "", "OWNER": "root"}
 DROP TABLE IF EXISTS federated.old;
 Warnings:
 Note	1051	Unknown table 'federated.old'
@@ -161,7 +161,7 @@ drop table federated.t1;
 drop server 'server_one';
 drop server 'server_two';
 select * from mysql.servers order by db;
-Server_name	Host	Db	Username	Password	Port	Socket	Wrapper	Owner
+Server_name	Host	Db	Username	Password	Port	Socket	Wrapper	Owner	Options
 connection slave;
 drop table first_db.t1;
 drop table second_db.t1;

mysql-test/suite/funcs_1/r/is_check_constraints.result

@@ -139,6 +139,7 @@ SELECT * FROM information_schema.check_constraints;
 CONSTRAINT_CATALOG	CONSTRAINT_SCHEMA	TABLE_NAME	CONSTRAINT_NAME	LEVEL	CHECK_CLAUSE
 def	db	t1	CONSTRAINT_1	Table	`b` > 0
 def	mysql	global_priv	Priv	Column	json_valid(`Priv`)
+def	mysql	servers	Options	Column	json_valid(`Options`)
 CONNECT con1,localhost, foo,, db;
 SELECT a FROM t1;
 a
@@ -172,6 +173,7 @@ t	CREATE TABLE `t` (
 ) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_uca1400_ai_ci
 select * from information_schema.table_constraints where CONSTRAINT_TYPE='CHECK';
 CONSTRAINT_CATALOG	CONSTRAINT_SCHEMA	CONSTRAINT_NAME	TABLE_SCHEMA	TABLE_NAME	CONSTRAINT_TYPE
+def	mysql	Options	mysql	servers	CHECK
 def	mysql	Priv	mysql	global_priv	CHECK
 def	test	CONSTRAINT_1	test	t	CHECK
 def	test	t1	test	t	CHECK
@@ -181,6 +183,7 @@ def	test	tc_1	test	t	CHECK
 select * from information_schema.check_constraints;
 CONSTRAINT_CATALOG	CONSTRAINT_SCHEMA	TABLE_NAME	CONSTRAINT_NAME	LEVEL	CHECK_CLAUSE
 def	mysql	global_priv	Priv	Column	json_valid(`Priv`)
+def	mysql	servers	Options	Column	json_valid(`Options`)
 def	test	t	CONSTRAINT_1	Table	`t0` > 0
 def	test	t	t1	Column	`t1` < 0
 def	test	t	t2	Column	`t2` < -1

mysql-test/suite/funcs_1/r/is_columns_mysql.result

@@ -160,6 +160,7 @@ def	mysql	roles_mapping	Role	3	''	NO	char	128	384	NULL	NULL	NULL	utf8mb3	utf8mb3
 def	mysql	roles_mapping	User	2	''	NO	char	128	384	NULL	NULL	NULL	utf8mb3	utf8mb3_bin	char(128)	PRI		select,insert,update,references		NEVER	NULL	NO	NO
 def	mysql	servers	Db	3	''	NO	char	64	192	NULL	NULL	NULL	utf8mb3	utf8mb3_general_ci	char(64)			select,insert,update,references		NEVER	NULL	NO	NO
 def	mysql	servers	Host	2	''	NO	varchar	2048	6144	NULL	NULL	NULL	utf8mb3	utf8mb3_general_ci	varchar(2048)			select,insert,update,references		NEVER	NULL	NO	NO
+def	mysql	servers	Options	10	'{}'	NO	longtext	4294967295	4294967295	NULL	NULL	NULL	utf8mb4	utf8mb4_bin	longtext			select,insert,update,references		NEVER	NULL	NO	NO
 def	mysql	servers	Owner	9	''	NO	varchar	512	1536	NULL	NULL	NULL	utf8mb3	utf8mb3_general_ci	varchar(512)			select,insert,update,references		NEVER	NULL	NO	NO
 def	mysql	servers	Password	5	''	NO	char	64	192	NULL	NULL	NULL	utf8mb3	utf8mb3_general_ci	char(64)			select,insert,update,references		NEVER	NULL	NO	NO
 def	mysql	servers	Port	6	0	NO	int	NULL	NULL	10	0	NULL	NULL	NULL	int(4)			select,insert,update,references		NEVER	NULL	NO	NO
@@ -495,6 +496,7 @@ NULL	mysql	servers	Port	int	NULL	NULL	NULL	NULL	int(4)
 3.0000	mysql	servers	Socket	char	108	324	utf8mb3	utf8mb3_general_ci	char(108)
 3.0000	mysql	servers	Wrapper	char	64	192	utf8mb3	utf8mb3_general_ci	char(64)
 3.0000	mysql	servers	Owner	varchar	512	1536	utf8mb3	utf8mb3_general_ci	varchar(512)
+1.0000	mysql	servers	Options	longtext	4294967295	4294967295	utf8mb4	utf8mb4_bin	longtext
 NULL	mysql	slow_log	start_time	timestamp	NULL	NULL	NULL	NULL	timestamp(6)
 1.0000	mysql	slow_log	user_host	mediumtext	16777215	16777215	utf8mb3	utf8mb3_general_ci	mediumtext
 NULL	mysql	slow_log	query_time	time	NULL	NULL	NULL	NULL	time(6)

mysql-test/suite/funcs_1/r/is_table_constraints.result

@@ -81,6 +81,7 @@ def	mysql	PRIMARY	mysql	proc
 def	mysql	PRIMARY	mysql	procs_priv
 def	mysql	PRIMARY	mysql	proxies_priv
 def	mysql	Host	mysql	roles_mapping
+def	mysql	Options	mysql	servers
 def	mysql	PRIMARY	mysql	servers
 def	mysql	PRIMARY	mysql	tables_priv
 def	mysql	PRIMARY	mysql	table_stats

mysql-test/suite/funcs_1/r/is_table_constraints_mysql.result

@@ -30,6 +30,7 @@ def	mysql	PRIMARY	mysql	proc	PRIMARY KEY
 def	mysql	PRIMARY	mysql	procs_priv	PRIMARY KEY
 def	mysql	PRIMARY	mysql	proxies_priv	PRIMARY KEY
 def	mysql	Host	mysql	roles_mapping	UNIQUE
+def	mysql	Options	mysql	servers	CHECK
 def	mysql	PRIMARY	mysql	servers	PRIMARY KEY
 def	mysql	PRIMARY	mysql	tables_priv	PRIMARY KEY
 def	mysql	PRIMARY	mysql	table_stats	PRIMARY KEY

mysql-test/suite/funcs_1/r/is_tables_mysql.result

@@ -515,7 +515,7 @@ TABLE_SCHEMA	mysql
 TABLE_NAME	servers
 TABLE_TYPE	BASE TABLE
 ENGINE	MYISAM_OR_MARIA
-VERSION	10
+VERSION	11
 ROW_FORMAT	DYNAMIC_OR_PAGE
 TABLE_ROWS	#TBLR#
 AVG_ROW_LENGTH	#ARL#

mysql-test/suite/galera/r/galera_server.result

@@ -8,15 +8,15 @@ OPTIONS (HOST 'foo');
 connection node_2;
 # On node_2
 SELECT * FROM mysql.servers;
-Server_name	Host	Db	Username	Password	Port	Socket	Wrapper	Owner
-s1	foo				3306		mysql	
+Server_name	Host	Db	Username	Password	Port	Socket	Wrapper	Owner	Options
+s1	foo				3306		mysql		{"HOST": "foo"}
 ALTER SERVER s1
 OPTIONS (HOST 'bar');
 connection node_1;
 # On node_1
 SELECT * FROM mysql.servers;
-Server_name	Host	Db	Username	Password	Port	Socket	Wrapper	Owner
-s1	bar				3306		mysql	
+Server_name	Host	Db	Username	Password	Port	Socket	Wrapper	Owner	Options
+s1	bar				3306		mysql		{"HOST": "bar"}
 DROP SERVER s1;
 connection node_2;
 # On node_2

scripts/mariadb_system_tables.sql

@@ -109,7 +109,7 @@ CREATE TABLE IF NOT EXISTS func (  name char(64) binary DEFAULT '' NOT NULL, ret
 CREATE TABLE IF NOT EXISTS plugin ( name varchar(64) DEFAULT '' NOT NULL, dl varchar(128) DEFAULT '' NOT NULL, PRIMARY KEY (name) ) engine=Aria transactional=1 CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci comment='MySQL plugins';
 
 
-CREATE TABLE IF NOT EXISTS servers ( Server_name char(64) NOT NULL DEFAULT '', Host varchar(2048) NOT NULL DEFAULT '', Db char(64) NOT NULL DEFAULT '', Username char(128) NOT NULL DEFAULT '', Password char(64) NOT NULL DEFAULT '', Port INT(4) NOT NULL DEFAULT '0', Socket char(108) NOT NULL DEFAULT '', Wrapper char(64) NOT NULL DEFAULT '', Owner varchar(512) NOT NULL DEFAULT '', PRIMARY KEY (Server_name)) engine=Aria transactional=1 CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci comment='MySQL Foreign Servers table';
+CREATE TABLE IF NOT EXISTS servers ( Server_name char(64) NOT NULL DEFAULT '', Host varchar(2048) NOT NULL DEFAULT '', Db char(64) NOT NULL DEFAULT '', Username char(128) NOT NULL DEFAULT '', Password char(64) NOT NULL DEFAULT '', Port INT(4) NOT NULL DEFAULT '0', Socket char(108) NOT NULL DEFAULT '', Wrapper char(64) NOT NULL DEFAULT '', Owner varchar(512) NOT NULL DEFAULT '', PRIMARY KEY (Server_name), Options JSON NOT NULL DEFAULT '{}' CHECK(JSON_VALID(Options))) engine=Aria transactional=1 CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci comment='MySQL Foreign Servers table';
 
 
 CREATE TABLE IF NOT EXISTS tables_priv ( Host char(255) binary DEFAULT '' NOT NULL, Db char(64) binary DEFAULT '' NOT NULL, User char(128) binary DEFAULT '' NOT NULL, Table_name char(64) binary DEFAULT '' NOT NULL, Grantor varchar(384) DEFAULT '' NOT NULL, Timestamp timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, Table_priv set('Select','Insert','Update','Delete','Create','Drop','Grant','References','Index','Alter','Create View','Show view','Trigger','Delete versioning rows') COLLATE utf8mb3_general_ci DEFAULT '' NOT NULL, Column_priv set('Select','Insert','Update','References') COLLATE utf8mb3_general_ci DEFAULT '' NOT NULL, PRIMARY KEY (Host,Db,User,Table_name), KEY Grantor (Grantor) ) engine=Aria transactional=1 CHARACTER SET utf8mb3 COLLATE utf8mb3_bin   comment='Table privileges';

scripts/mariadb_system_tables_fix.sql

@@ -901,3 +901,10 @@ ALTER TABLE servers
 # MDEV-34716 Fix mysql.servers socket max length too short
 ALTER TABLE servers
   MODIFY Socket char(108) NOT NULL DEFAULT '';
+
+# MDEV-34716 Allow arbitrary options in CREATE SERVER
+ALTER TABLE servers
+  ADD Options JSON NOT NULL DEFAULT '{}' CHECK(JSON_VALID(Options));
+# Ensure the collation is utf8mb4_bin (default for JSON)
+ALTER TABLE servers
+  MODIFY Options JSON  NOT NULL DEFAULT '{}' CHECK(JSON_VALID(Options));

sql/create_options.h

@@ -26,6 +26,10 @@
 
 enum { ENGINE_OPTION_MAX_LENGTH=32767 };
 
+/*
+  Key-value list. Used for engine-defined options in CREATE TABLE
+  and OPTIONS in CREATE SERVER.
+*/
 class engine_option_value: public Sql_alloc
 {
 public:

sql/sql_lex.h

@@ -317,11 +317,13 @@ typedef struct st_lex_server_options
 {
   long port;
   LEX_CSTRING server_name, host, db, username, password, scheme, socket, owner;
+  engine_option_value *option_list;
   void reset(LEX_CSTRING name)
   {
     server_name= name;
     host= db= username= password= scheme= socket= owner= null_clex_str;
     port= -1;
+    option_list= NULL;
   }
 } LEX_SERVER_OPTIONS;
 

sql/sql_servers.cc

@@ -45,6 +45,7 @@
 #include "sp.h"
 #include "transaction.h"
 #include "lock.h"                               // MYSQL_LOCK_IGNORE_TIMEOUT
+#include "create_options.h"
 
 /*
   We only use 1 mutex to guard the data structures - THR_LOCK_servers.
@@ -434,6 +435,42 @@ get_server_from_table_to_cache(TABLE *table)
   server->scheme= ptr ? ptr : blank;
   ptr= get_field(&mem, table->field[8]);
   server->owner= ptr ? ptr : blank;
+  ptr= get_field(&mem, table->field[9]);
+  enum json_types vt;
+  const char *keyname, *keyname_end, *v;
+  int v_len, nkey= 0;
+  engine_option_value *option_list_last;
+  server->option_list= NULL;
+  while ((vt= json_get_object_nkey(ptr, ptr+strlen(ptr), nkey++,
+                    &keyname, &keyname_end, &v, &v_len)) != JSV_NOTHING)
+  {
+    if (vt != JSV_STRING)
+      DBUG_RETURN(TRUE);
+    /*
+      We have to make copies here to create "clean" strings and
+      avoid mutating ptr.
+    */
+    Lex_cstring name= {keyname, keyname_end}, value= {v, v + v_len},
+      name_copy= safe_lexcstrdup_root(&mem, name),
+      value_copy= safe_lexcstrdup_root(&mem, value);
+    engine_option_value *option= new (&mem) engine_option_value(
+      engine_option_value::Name(name_copy),
+      engine_option_value::Value(value_copy), true);
+    option->link(&server->option_list, &option_list_last);
+    if (option->value.length)
+    {
+      LEX_CSTRING *optval= &option->value;
+      char *unescaped= (char *) alloca(optval->length);
+      int len= json_unescape_json(optval->str, optval->str + optval->length,
+                                  unescaped, unescaped + optval->length);
+      if (len < 0)
+        DBUG_RETURN(TRUE);
+      DBUG_ASSERT(len <= (int) optval->length);
+      if (len < (int) optval->length)
+        strncpy((char *) optval->str, unescaped, len);
+      optval->length= len;
+    }
+  }
   DBUG_PRINT("info", ("server->server_name %s", server->server_name));
   DBUG_PRINT("info", ("server->host %s", server->host));
   DBUG_PRINT("info", ("server->db %s", server->db));
@@ -591,6 +628,30 @@ store_server_fields(TABLE *table, FOREIGN_SERVER *server)
   if (server->owner)
     table->field[8]->store(server->owner,
                            (uint) strlen(server->owner), system_charset_info);
+  engine_option_value *option= server->option_list;
+  StringBuffer<1024> json(table->field[9]->charset());
+  json.append('{');
+  while (option)
+  {
+    if (option->value.str)
+    {
+      json.append('"');
+      json.append(option->name.str, option->name.length);
+      json.append('"');
+      json.append({STRING_WITH_LEN(": \"")});
+      int len= json_escape_string(
+        option->value.str, option->value.str + option->value.length,
+        json.c_ptr() + json.length(), json.c_ptr() + json.alloced_length());
+      json.length(json.length() + len);
+      json.append('"');
+      json.append({STRING_WITH_LEN(", ")});
+    }
+    option= option->next;
+  }
+  if (server->option_list)
+    json.length(json.length() - 2);
+  json.append('}');
+  table->field[9]->store(json.ptr(), json.length(), system_charset_info);
 }
 
 /*
@@ -1185,6 +1246,23 @@ int alter_server(THD *thd, LEX_SERVER_OPTIONS *server_options)
 }
 
 
+static void copy_option_list(MEM_ROOT *mem, FOREIGN_SERVER *server,
+                             engine_option_value *option_list)
+{
+  engine_option_value *option_list_last;
+  server->option_list= NULL;
+  for (engine_option_value *option= option_list; option;
+       option= option->next)
+  {
+    engine_option_value *new_option= new (mem) engine_option_value(option);
+    new_option->name= engine_option_value::Name(
+      safe_lexcstrdup_root(mem, option->name));
+    new_option->value= engine_option_value::Value(
+      safe_lexcstrdup_root(mem, option->value));
+    new_option->link(&server->option_list, &option_list_last);
+  }
+}
+
 /*
 
   SYNOPSIS
@@ -1240,6 +1318,7 @@ prepare_server_struct_for_insert(LEX_SERVER_OPTIONS *server_options)
   SET_SERVER_OR_RETURN(password, "");
   SET_SERVER_OR_RETURN(socket, "");
   SET_SERVER_OR_RETURN(owner, "");
+  copy_option_list(&mem, server, server_options->option_list);
 
   server->server_name_length= server_options->server_name.length;
 
@@ -1294,6 +1373,8 @@ prepare_server_struct_for_update(LEX_SERVER_OPTIONS *server_options,
   SET_ALTERED(socket);
   SET_ALTERED(scheme);
   SET_ALTERED(owner);
+  merge_engine_options(existing->option_list, server_options->option_list,
+                       &altered->option_list, &mem);
 
   /*
     port is initialised to -1, so if unset, it will be -1
@@ -1355,7 +1436,7 @@ void servers_free(bool end)
    FOREIGN_SEVER pointer (copy of one supplied FOREIGN_SERVER)
 */
 
-static FOREIGN_SERVER *clone_server(MEM_ROOT *mem, const FOREIGN_SERVER *server,
+static FOREIGN_SERVER *clone_server(MEM_ROOT *mem, FOREIGN_SERVER *server,
                                     FOREIGN_SERVER *buffer)
 {
   DBUG_ENTER("sql_server.cc:clone_server");
@@ -1376,6 +1457,7 @@ static FOREIGN_SERVER *clone_server(MEM_ROOT *mem, const FOREIGN_SERVER *server,
   buffer->socket= safe_strdup_root(mem, server->socket);
   buffer->owner= safe_strdup_root(mem, server->owner);
   buffer->host= safe_strdup_root(mem, server->host);
+  copy_option_list(mem, buffer, server->option_list);
 
  DBUG_RETURN(buffer);
 }

sql/sql_servers.h

@@ -17,6 +17,7 @@
    Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1335  USA */
 
 #include "slave.h" // for tables_ok(), rpl_filter
+#include "create_options.h"
 
 class THD;
 typedef struct st_lex_server_options LEX_SERVER_OPTIONS;
@@ -29,6 +30,7 @@ typedef struct st_federated_server
   long port;
   size_t server_name_length;
   const char *db, *scheme, *username, *password, *socket, *owner, *host, *sport;
+  engine_option_value *option_list;
 } FOREIGN_SERVER;
 
 /* cache handlers */

sql/sql_yacc.yy

@@ -2907,35 +2907,123 @@ server_option:
           {
             MYSQL_YYABORT_UNLESS(Lex->server_options.username.str == 0);
             Lex->server_options.username= $2;
+            engine_option_value *new_option=
+              new (thd->mem_root) engine_option_value(
+                engine_option_value::Name(
+                  safe_lexcstrdup_root(thd->mem_root, $1)),
+                engine_option_value::Value(
+                  safe_lexcstrdup_root(thd->mem_root, $2)), true);
+            new_option->link(&Lex->server_options.option_list,
+                             &Lex->option_list_last);
           }
         | HOST_SYM TEXT_STRING_sys
           {
             MYSQL_YYABORT_UNLESS(Lex->server_options.host.str == 0);
             Lex->server_options.host= $2;
+            engine_option_value *new_option=
+              new (thd->mem_root) engine_option_value(
+                engine_option_value::Name(
+                  safe_lexcstrdup_root(thd->mem_root, $1)),
+                engine_option_value::Value(
+                  safe_lexcstrdup_root(thd->mem_root, $2)), true);
+            new_option->link(&Lex->server_options.option_list,
+                             &Lex->option_list_last);
           }
         | DATABASE TEXT_STRING_sys
           {
             MYSQL_YYABORT_UNLESS(Lex->server_options.db.str == 0);
             Lex->server_options.db= $2;
+            engine_option_value *new_option=
+              new (thd->mem_root) engine_option_value(
+                engine_option_value::Name(
+                  safe_lexcstrdup_root(thd->mem_root, $1)),
+                engine_option_value::Value(
+                  safe_lexcstrdup_root(thd->mem_root, $2)), true);
+            new_option->link(&Lex->server_options.option_list,
+                             &Lex->option_list_last);
           }
         | OWNER_SYM TEXT_STRING_sys
           {
             MYSQL_YYABORT_UNLESS(Lex->server_options.owner.str == 0);
             Lex->server_options.owner= $2;
+            engine_option_value *new_option=
+              new (thd->mem_root) engine_option_value(
+                engine_option_value::Name(
+                  safe_lexcstrdup_root(thd->mem_root, $1)),
+                engine_option_value::Value(
+                  safe_lexcstrdup_root(thd->mem_root, $2)), true);
+            new_option->link(&Lex->server_options.option_list,
+                             &Lex->option_list_last);
           }
         | PASSWORD_SYM TEXT_STRING_sys
           {
             MYSQL_YYABORT_UNLESS(Lex->server_options.password.str == 0);
             Lex->server_options.password= $2;
+            engine_option_value *new_option=
+              new (thd->mem_root) engine_option_value(
+                engine_option_value::Name(
+                  safe_lexcstrdup_root(thd->mem_root, $1)),
+                engine_option_value::Value(
+                  safe_lexcstrdup_root(thd->mem_root, $2)), true);
+            new_option->link(&Lex->server_options.option_list,
+                             &Lex->option_list_last);
           }
         | SOCKET_SYM TEXT_STRING_sys
           {
             MYSQL_YYABORT_UNLESS(Lex->server_options.socket.str == 0);
             Lex->server_options.socket= $2;
+            engine_option_value *new_option=
+              new (thd->mem_root) engine_option_value(
+                engine_option_value::Name(
+                  safe_lexcstrdup_root(thd->mem_root, $1)),
+                engine_option_value::Value(
+                  safe_lexcstrdup_root(thd->mem_root, $2)), true);
+            new_option->link(&Lex->server_options.option_list,
+                             &Lex->option_list_last);
           }
         | PORT_SYM ulong_num
           {
             Lex->server_options.port= $2;
+            engine_option_value *new_option=
+              new (thd->mem_root) engine_option_value(
+                engine_option_value::Name(
+                  safe_lexcstrdup_root(thd->mem_root, $1)),
+                $2, thd->mem_root);
+            new_option->link(&Lex->server_options.option_list,
+                             &Lex->option_list_last);
+          }
+        /* port can be a quoted number */
+        | PORT_SYM TEXT_STRING_sys
+          {
+            int error;
+            char *end= (char *) $2.str + $2.length;
+            longlong p= my_strtoll10($2.str, &end, &error);
+            if (error > 0 || end != (char *) $2.str + $2.length ||
+                p > LONG_MAX || p < LONG_MIN)
+            {
+              thd->parse_error();
+              MYSQL_YYABORT;
+            }
+            Lex->server_options.port= (long) p;
+            engine_option_value *new_option=
+              new (thd->mem_root) engine_option_value(
+                engine_option_value::Name(
+                  safe_lexcstrdup_root(thd->mem_root, $1)),
+                engine_option_value::Value(
+                  safe_lexcstrdup_root(thd->mem_root, $2)), true);
+            new_option->link(&Lex->server_options.option_list,
+                             &Lex->option_list_last);
+          }
+        | IDENT_sys TEXT_STRING_sys
+          {
+            engine_option_value *new_option=
+              new (thd->mem_root) engine_option_value(
+                engine_option_value::Name(
+                  safe_lexcstrdup_root(thd->mem_root, $1)),
+                engine_option_value::Value(
+                  safe_lexcstrdup_root(thd->mem_root, $2)), true);
+            new_option->link(&Lex->server_options.option_list,
+                             &Lex->option_list_last);
           }
         ;