MDEV-5271 Support engine-defined attributes per partitionbb-10.8-mdev-5271

Make it possible to specify engine-defined attributes on partitions
as well as tables.

If an engine-defined attribute is only specified at the table level,
it applies to all the partitions in the table.
This is a backward-compatible behavior.

If the same attribute is specified both at the table level and the
partition level, the per-partition one takes precedence.
So, we can consider per-table attributes as default values.

One cannot specify engine-defined attributes on subpartitions.

Implementation details:

* We store per-partition attributes in the partition_element class
  because we already have the part_comment field, which is for
  per-partition comments.

* In the case of ALTER TABLE statements, the partition_elements in
  table->part_info is set up by mysql_unpack_partition().
  So, we parse per-partition attributes after the call of the function.

11 files changed, 588 insertions, 124 deletions

diff --git a/mysql-test/main/partition_error.result b/mysql-test/main/partition_error.result
index bff08c0d447..cf057fe5b47 100644
--- a/mysql-test/main/partition_error.result
+++ b/mysql-test/main/partition_error.result
@@ -896,7 +896,7 @@ partitions 3
 (partition tablespace ts1,
 partition x2 tablespace ts2,
 partition x3 tablespace ts3);
-ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'ts1,
+ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near '
 partition x2 tablespace ts2,
 partition x3 tablespace ts3)' at line 8
 CREATE TABLE t1 (
diff --git a/mysql-test/suite/parts/r/engine_defined_part_attributes.result b/mysql-test/suite/parts/r/engine_defined_part_attributes.result
new file mode 100644
index 00000000000..81063d82c6f
--- /dev/null
+++ b/mysql-test/suite/parts/r/engine_defined_part_attributes.result
@@ -0,0 +1,218 @@
+#
+# MDEV-5271 Support engine-defined attributes per partition
+#
+# partitioned tables
+CREATE TABLE `t1` (
+`id` INT
+) ENGINE=InnoDB ENCRYPTED="YES" PARTITION BY RANGE(id) (
+PARTITION pt1 VALUES LESS THAN (100) ENCRYPTED="NO" ENCRYPTION_KEY_ID=1,
+PARTITION pt2 VALUES LESS THAN MAXVALUE ENCRYPTED="DEFAULT" ENCRYPTION_KEY_ID=1
+);
+SHOW CREATE TABLE `t1`;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `id` int(11) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 `ENCRYPTED`='YES'
+ PARTITION BY RANGE (`id`)
+(PARTITION `pt1` VALUES LESS THAN (100) ENGINE = InnoDB ENCRYPTED = 'NO' ENCRYPTION_KEY_ID = 1,
+ PARTITION `pt2` VALUES LESS THAN MAXVALUE ENGINE = InnoDB ENCRYPTED = 'DEFAULT' ENCRYPTION_KEY_ID = 1)
+INSERT INTO t1 VALUES (1), (2), (3);
+DELETE FROM t1 WHERE id = 1;
+UPDATE t1 SET id = 4 WHERE id = 3;
+SELECT * FROM t1 WHERE id IN (2, 3);
+id
+2
+DROP TABLE `t1`;
+CREATE TABLE `t2` (
+`id` INT
+) ENGINE=InnoDB ENCRYPTED="YES" ENCRYPTION_KEY_ID=2 PARTITION BY RANGE(id) (
+PARTITION pt1 VALUES LESS THAN (100),
+PARTITION pt2 VALUES LESS THAN MAXVALUE
+);
+ERROR HY000: Can't create table `test`.`t2` (errno: 140 "Wrong create options")
+CREATE TABLE `t3` (
+`id` INT
+) ENGINE=InnoDB ENCRYPTED="NO" PARTITION BY RANGE(id) (
+PARTITION pt1 VALUES LESS THAN (100) ENCRYPTED="YES" ENCRYPTION_KEY_ID=2,
+PARTITION pt2 VALUES LESS THAN MAXVALUE
+);
+ERROR HY000: Can't create table `test`.`t3` (errno: 140 "Wrong create options")
+CREATE TABLE `t4` (
+`id` INT
+) ENGINE=InnoDB ENCRYPTED="NO";
+SHOW CREATE TABLE `t4`;
+Table	Create Table
+t4	CREATE TABLE `t4` (
+  `id` int(11) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 `ENCRYPTED`='NO'
+ALTER TABLE `t4` PARTITION BY RANGE(id) (
+PARTITION pt1 VALUES LESS THAN (100) ENCRYPTED="NO",
+PARTITION pt2 VALUES LESS THAN MAXVALUE ENCRYPTED="DEFAULT"
+);
+SHOW CREATE TABLE `t4`;
+Table	Create Table
+t4	CREATE TABLE `t4` (
+  `id` int(11) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 `ENCRYPTED`='NO'
+ PARTITION BY RANGE (`id`)
+(PARTITION `pt1` VALUES LESS THAN (100) ENGINE = InnoDB ENCRYPTED = 'NO',
+ PARTITION `pt2` VALUES LESS THAN MAXVALUE ENGINE = InnoDB ENCRYPTED = 'DEFAULT')
+ALTER TABLE `t4` PARTITION BY RANGE(id) (
+PARTITION pt1 VALUES LESS THAN (100),
+PARTITION pt2 VALUES LESS THAN MAXVALUE
+);
+SHOW CREATE TABLE `t4`;
+Table	Create Table
+t4	CREATE TABLE `t4` (
+  `id` int(11) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 `ENCRYPTED`='NO'
+ PARTITION BY RANGE (`id`)
+(PARTITION `pt1` VALUES LESS THAN (100) ENGINE = InnoDB,
+ PARTITION `pt2` VALUES LESS THAN MAXVALUE ENGINE = InnoDB)
+ALTER TABLE `t4` PARTITION BY RANGE(id) (
+PARTITION pt1 VALUES LESS THAN (100) ENCRYPTED="YES" ENCRYPTION_KEY_ID=2,
+PARTITION pt2 VALUES LESS THAN MAXVALUE ENCRYPTED="DEFAULT"
+);
+ERROR HY000: Can't create table `test`.`t4` (errno: 140 "Wrong create options")
+DROP TABLE `t4`;
+# subpartitioned tables
+CREATE TABLE `t5` (
+`id` INT
+) ENGINE=InnoDB ENCRYPTED="NO" PARTITION BY RANGE(id)
+SUBPARTITION BY HASH(id)
+SUBPARTITIONS 2 (
+PARTITION pt1 VALUES LESS THAN (100),
+PARTITION pt2 VALUES LESS THAN MAXVALUE
+);
+SHOW CREATE TABLE `t5`;
+Table	Create Table
+t5	CREATE TABLE `t5` (
+  `id` int(11) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 `ENCRYPTED`='NO'
+ PARTITION BY RANGE (`id`)
+SUBPARTITION BY HASH (`id`)
+SUBPARTITIONS 2
+(PARTITION `pt1` VALUES LESS THAN (100) ENGINE = InnoDB,
+ PARTITION `pt2` VALUES LESS THAN MAXVALUE ENGINE = InnoDB)
+DROP TABLE `t5`;
+CREATE TABLE `t6` (
+`id` INT
+) ENGINE=InnoDB PARTITION BY RANGE(id)
+SUBPARTITION BY HASH(id)
+SUBPARTITIONS 2 (
+PARTITION pt1 VALUES LESS THAN (100) ENCRYPTED="YES",
+PARTITION pt2 VALUES LESS THAN MAXVALUE
+);
+ERROR HY000: Can't create table `test`.`t6` (errno: 140 "Wrong create options")
+CREATE TABLE `t7` (
+id INT
+) ENGINE=InnoDB PARTITION BY RANGE(id)
+SUBPARTITION BY HASH(id) (
+PARTITION pt1 VALUES LESS THAN (100)(
+SUBPARTITION spt1 ENCRYPTED="NO",
+SUBPARTITION spt2
+),
+PARTITION pt2 VALUES LESS THAN MAXVALUE (
+SUBPARTITION spt3,
+SUBPARTITION spt4
+)
+);
+ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'ENCRYPTED="NO",
+SUBPARTITION spt2
+),
+PARTITION pt2 VALUES LESS THAN MAXVALUE ...' at line 6
+CREATE TABLE `t8` (
+id INT
+) ENGINE=InnoDB ENCRYPTED="NO" PARTITION BY RANGE(id)
+SUBPARTITION BY HASH(id) (
+PARTITION pt1 VALUES LESS THAN (100) (
+SUBPARTITION spt1,
+SUBPARTITION spt2
+),
+PARTITION pt2 VALUES LESS THAN MAXVALUE (
+SUBPARTITION spt3,
+SUBPARTITION spt4
+)
+);
+SHOW CREATE TABLE `t8`;
+Table	Create Table
+t8	CREATE TABLE `t8` (
+  `id` int(11) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 `ENCRYPTED`='NO'
+ PARTITION BY RANGE (`id`)
+SUBPARTITION BY HASH (`id`)
+(PARTITION `pt1` VALUES LESS THAN (100)
+ (SUBPARTITION `spt1` ENGINE = InnoDB,
+  SUBPARTITION `spt2` ENGINE = InnoDB),
+ PARTITION `pt2` VALUES LESS THAN MAXVALUE
+ (SUBPARTITION `spt3` ENGINE = InnoDB,
+  SUBPARTITION `spt4` ENGINE = InnoDB))
+DROP TABLE `t8`;
+CREATE TABLE `t9` (
+id INT
+) ENGINE=InnoDB PARTITION BY RANGE(id)
+SUBPARTITION BY HASH(id) (
+PARTITION pt1 VALUES LESS THAN (100) ENCRYPTED="NO" (
+SUBPARTITION spt1,
+SUBPARTITION spt2
+),
+PARTITION pt2 VALUES LESS THAN MAXVALUE (
+SUBPARTITION spt3,
+SUBPARTITION spt4
+)
+);
+SHOW CREATE TABLE `t9`;
+Table	Create Table
+t9	CREATE TABLE `t9` (
+  `id` int(11) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+ PARTITION BY RANGE (`id`)
+SUBPARTITION BY HASH (`id`)
+(PARTITION `pt1` VALUES LESS THAN (100)
+ (SUBPARTITION `spt1` ENGINE = InnoDB,
+  SUBPARTITION `spt2` ENGINE = InnoDB),
+ PARTITION `pt2` VALUES LESS THAN MAXVALUE
+ (SUBPARTITION `spt3` ENGINE = InnoDB,
+  SUBPARTITION `spt4` ENGINE = InnoDB))
+DROP TABLE `t9`;
+CREATE TABLE `t10` (
+id INT
+) ENGINE=InnoDB PARTITION BY RANGE(id)
+SUBPARTITION BY HASH(id) (
+PARTITION pt1 VALUES LESS THAN (100) ENCRYPTED="YES" (
+SUBPARTITION spt1,
+SUBPARTITION spt2
+),
+PARTITION pt2 VALUES LESS THAN MAXVALUE (
+SUBPARTITION spt3,
+SUBPARTITION spt4
+)
+);
+ERROR HY000: Can't create table `test`.`t10` (errno: 140 "Wrong create options")
+CREATE TABLE `t11` (
+id INT
+) ENGINE=InnoDB ENCRYPTED="YES" PARTITION BY RANGE(id)
+SUBPARTITION BY HASH(id) (
+PARTITION pt1 VALUES LESS THAN (100) ENCRYPTED="NO" (
+SUBPARTITION spt1,
+SUBPARTITION spt2
+),
+PARTITION pt2 VALUES LESS THAN MAXVALUE ENCRYPTED="NO" (
+SUBPARTITION spt3,
+SUBPARTITION spt4
+)
+);
+SHOW CREATE TABLE `t11`;
+Table	Create Table
+t11	CREATE TABLE `t11` (
+  `id` int(11) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 `ENCRYPTED`='YES'
+ PARTITION BY RANGE (`id`)
+SUBPARTITION BY HASH (`id`)
+(PARTITION `pt1` VALUES LESS THAN (100)
+ (SUBPARTITION `spt1` ENGINE = InnoDB,
+  SUBPARTITION `spt2` ENGINE = InnoDB),
+ PARTITION `pt2` VALUES LESS THAN MAXVALUE
+ (SUBPARTITION `spt3` ENGINE = InnoDB,
+  SUBPARTITION `spt4` ENGINE = InnoDB))
+DROP TABLE `t11`;
diff --git a/mysql-test/suite/parts/t/engine_defined_part_attributes.test b/mysql-test/suite/parts/t/engine_defined_part_attributes.test
new file mode 100644
index 00000000000..8d08f4654cc
--- /dev/null
+++ b/mysql-test/suite/parts/t/engine_defined_part_attributes.test
@@ -0,0 +1,169 @@
+--echo #
+--echo # MDEV-5271 Support engine-defined attributes per partition
+--echo #
+
+--source include/have_partition.inc
+--source include/have_innodb.inc
+
+--echo # partitioned tables
+
+CREATE TABLE `t1` (
+  `id` INT
+) ENGINE=InnoDB ENCRYPTED="YES" PARTITION BY RANGE(id) (
+    PARTITION pt1 VALUES LESS THAN (100) ENCRYPTED="NO" ENCRYPTION_KEY_ID=1,
+    PARTITION pt2 VALUES LESS THAN MAXVALUE ENCRYPTED="DEFAULT" ENCRYPTION_KEY_ID=1
+);
+SHOW CREATE TABLE `t1`;
+
+INSERT INTO t1 VALUES (1), (2), (3);
+DELETE FROM t1 WHERE id = 1;
+UPDATE t1 SET id = 4 WHERE id = 3;
+SELECT * FROM t1 WHERE id IN (2, 3);
+
+DROP TABLE `t1`;
+
+--error ER_CANT_CREATE_TABLE
+CREATE TABLE `t2` (
+  `id` INT
+) ENGINE=InnoDB ENCRYPTED="YES" ENCRYPTION_KEY_ID=2 PARTITION BY RANGE(id) (
+    PARTITION pt1 VALUES LESS THAN (100),
+    PARTITION pt2 VALUES LESS THAN MAXVALUE
+);
+
+--error ER_CANT_CREATE_TABLE
+CREATE TABLE `t3` (
+  `id` INT
+) ENGINE=InnoDB ENCRYPTED="NO" PARTITION BY RANGE(id) (
+    PARTITION pt1 VALUES LESS THAN (100) ENCRYPTED="YES" ENCRYPTION_KEY_ID=2,
+    PARTITION pt2 VALUES LESS THAN MAXVALUE
+);
+
+CREATE TABLE `t4` (
+  `id` INT
+) ENGINE=InnoDB ENCRYPTED="NO";
+SHOW CREATE TABLE `t4`;
+
+ALTER TABLE `t4` PARTITION BY RANGE(id) (
+    PARTITION pt1 VALUES LESS THAN (100) ENCRYPTED="NO",
+    PARTITION pt2 VALUES LESS THAN MAXVALUE ENCRYPTED="DEFAULT"
+);
+SHOW CREATE TABLE `t4`;
+
+ALTER TABLE `t4` PARTITION BY RANGE(id) (
+    PARTITION pt1 VALUES LESS THAN (100),
+    PARTITION pt2 VALUES LESS THAN MAXVALUE
+);
+SHOW CREATE TABLE `t4`;
+
+--error ER_CANT_CREATE_TABLE
+ALTER TABLE `t4` PARTITION BY RANGE(id) (
+    PARTITION pt1 VALUES LESS THAN (100) ENCRYPTED="YES" ENCRYPTION_KEY_ID=2,
+    PARTITION pt2 VALUES LESS THAN MAXVALUE ENCRYPTED="DEFAULT"
+);
+
+DROP TABLE `t4`;
+
+--echo # subpartitioned tables
+
+CREATE TABLE `t5` (
+  `id` INT
+) ENGINE=InnoDB ENCRYPTED="NO" PARTITION BY RANGE(id)
+SUBPARTITION BY HASH(id)
+SUBPARTITIONS 2 (
+  PARTITION pt1 VALUES LESS THAN (100),
+  PARTITION pt2 VALUES LESS THAN MAXVALUE
+);
+SHOW CREATE TABLE `t5`;
+
+DROP TABLE `t5`;
+
+--error ER_CANT_CREATE_TABLE
+CREATE TABLE `t6` (
+  `id` INT
+) ENGINE=InnoDB PARTITION BY RANGE(id)
+SUBPARTITION BY HASH(id)
+SUBPARTITIONS 2 (
+  PARTITION pt1 VALUES LESS THAN (100) ENCRYPTED="YES",
+  PARTITION pt2 VALUES LESS THAN MAXVALUE
+);
+
+--error ER_PARSE_ERROR
+CREATE TABLE `t7` (
+  id INT
+) ENGINE=InnoDB PARTITION BY RANGE(id)
+SUBPARTITION BY HASH(id) (
+  PARTITION pt1 VALUES LESS THAN (100)(
+    SUBPARTITION spt1 ENCRYPTED="NO",
+    SUBPARTITION spt2
+  ),
+  PARTITION pt2 VALUES LESS THAN MAXVALUE (
+    SUBPARTITION spt3,
+    SUBPARTITION spt4
+  )
+);
+
+CREATE TABLE `t8` (
+  id INT
+) ENGINE=InnoDB ENCRYPTED="NO" PARTITION BY RANGE(id)
+SUBPARTITION BY HASH(id) (
+  PARTITION pt1 VALUES LESS THAN (100) (
+    SUBPARTITION spt1,
+    SUBPARTITION spt2
+  ),
+  PARTITION pt2 VALUES LESS THAN MAXVALUE (
+    SUBPARTITION spt3,
+    SUBPARTITION spt4
+  )
+);
+SHOW CREATE TABLE `t8`;
+
+DROP TABLE `t8`;
+
+CREATE TABLE `t9` (
+  id INT
+) ENGINE=InnoDB PARTITION BY RANGE(id)
+SUBPARTITION BY HASH(id) (
+  PARTITION pt1 VALUES LESS THAN (100) ENCRYPTED="NO" (
+    SUBPARTITION spt1,
+    SUBPARTITION spt2
+  ),
+  PARTITION pt2 VALUES LESS THAN MAXVALUE (
+    SUBPARTITION spt3,
+    SUBPARTITION spt4
+  )
+);
+SHOW CREATE TABLE `t9`;
+
+DROP TABLE `t9`;
+
+--error ER_CANT_CREATE_TABLE
+CREATE TABLE `t10` (
+  id INT
+) ENGINE=InnoDB PARTITION BY RANGE(id)
+SUBPARTITION BY HASH(id) (
+  PARTITION pt1 VALUES LESS THAN (100) ENCRYPTED="YES" (
+    SUBPARTITION spt1,
+    SUBPARTITION spt2
+  ),
+  PARTITION pt2 VALUES LESS THAN MAXVALUE (
+    SUBPARTITION spt3,
+    SUBPARTITION spt4
+  )
+);
+
+CREATE TABLE `t11` (
+  id INT
+) ENGINE=InnoDB ENCRYPTED="YES" PARTITION BY RANGE(id)
+SUBPARTITION BY HASH(id) (
+  PARTITION pt1 VALUES LESS THAN (100) ENCRYPTED="NO" (
+    SUBPARTITION spt1,
+    SUBPARTITION spt2
+  ),
+  PARTITION pt2 VALUES LESS THAN MAXVALUE ENCRYPTED="NO" (
+    SUBPARTITION spt3,
+    SUBPARTITION spt4
+  )
+);
+SHOW CREATE TABLE `t11`;
+
+DROP TABLE `t11`;
diff --git a/sql/create_options.cc b/sql/create_options.cc
index 5437de0f0c3..cea5d7af950 100644
--- a/sql/create_options.cc
+++ b/sql/create_options.cc
@@ -1,4 +1,4 @@
-/* Copyright (C) 2010, 2020, MariaDB Corporation.
+/* Copyright (C) 2010, 2020, 2021, MariaDB Corporation.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -21,6 +21,7 @@
 
 #include "mariadb.h"
 #include "create_options.h"
+#include "partition_info.h"
 #include <my_getopt.h>
 #include "set_var.h"
 
@@ -339,8 +340,11 @@ bool parse_option_list(THD* thd, handlerton *hton, void *option_struct_arg,
             LEX_CSTRING name= { opt->name, opt->name_length };
             default_val.str= strmake_root(root, str->ptr(), str->length());
             default_val.length= str->length();
-            val= new (root) engine_option_value(name, default_val,
-                         opt->type != HA_OPTION_TYPE_ULL, option_list, &last);
+            val= new (root) engine_option_value(
+                name, default_val, opt->type != HA_OPTION_TYPE_ULL);
+            if (!val)
+              DBUG_RETURN(TRUE);
+            val->link(option_list, &last);
             val->parsed= true;
           }
         }
@@ -497,6 +501,61 @@ bool parse_engine_table_options(THD *thd, handlerton *ht, TABLE_SHARE *share)
   DBUG_RETURN(FALSE);
 }
 
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+/**
+  Parses engine-defined partition options
+
+  @param [in] thd   thread handler
+  @parem [in] table table with part_info
+
+  @retval TRUE  Error
+  @retval FALSE OK
+
+  In the case of ALTER TABLE statements, table->part_info is set up
+  by mysql_unpack_partition(). So, one should not call the present
+  function before the call of mysql_unpack_partition().
+*/
+bool parse_engine_part_options(THD *thd, TABLE *table)
+{
+  MEM_ROOT *root= &table->mem_root;
+  TABLE_SHARE *share= table->s;
+  partition_info *part_info= table->part_info;
+  engine_option_value *tmp_option_list;
+  handlerton *ht;
+  DBUG_ENTER("parse_engine_part_options");
+
+  if (!part_info)
+    DBUG_RETURN(FALSE);
+
+  List_iterator<partition_element> it(part_info->partitions);
+  while (partition_element *part_elem= it++)
+  {
+    if (merge_engine_options(share->option_list, part_elem->option_list,
+                             &tmp_option_list, root))
+      DBUG_RETURN(TRUE);
+
+    if (!part_info->is_sub_partitioned())
+    {
+      ht= part_elem->engine_type;
+      if (parse_option_list(thd, ht, &part_elem->option_struct,
+                            &tmp_option_list, ht->table_options, TRUE, root))
+        DBUG_RETURN(TRUE);
+    }
+    else
+    {
+      List_iterator<partition_element> sub_it(part_elem->subpartitions);
+      while (partition_element *sub_part_elem= sub_it++)
+      {
+        ht= sub_part_elem->engine_type;
+        if (parse_option_list(thd, ht, &sub_part_elem->option_struct,
+                              &tmp_option_list, ht->table_options, TRUE, root))
+          DBUG_RETURN(TRUE);
+      }
+    }
+  }
+  DBUG_RETURN(FALSE);
+}
+#endif
 
 bool engine_options_differ(void *old_struct, void *new_struct,
                            ha_create_table_option *rules)
@@ -694,10 +753,11 @@ uchar *engine_option_value::frm_read(const uchar *buff, const uchar *buff_end,
     return NULL;
   buff+= value.length;
 
-  engine_option_value *ptr=new (root)
-    engine_option_value(name, value, len & FRM_QUOTED_VALUE, start, end);
+  engine_option_value *ptr=
+      new (root) engine_option_value(name, value, len & FRM_QUOTED_VALUE);
   if (!ptr)
     return NULL;
+  ptr->link(start, end);
 
   return (uchar *)buff;
 }
@@ -766,23 +826,40 @@ bool engine_table_options_frm_read(const uchar *buff, size_t length,
 
 /**
   Merges two lists of engine_option_value's with duplicate removal.
-*/
 
-engine_option_value *merge_engine_table_options(engine_option_value *first,
-                                                engine_option_value *second,
-                                                MEM_ROOT *root)
+  @param [in] source  option list
+  @param [in] changes option list whose options overwrite source's
+  @param [out] out    new option list created by merging given two
+  @param [in] root    MEM_ROOT for allocating memory
+
+  @retval TRUE  Error
+  @retval FALSE OK
+*/
+bool merge_engine_options(engine_option_value *source,
+                          engine_option_value *changes,
+                          engine_option_value **out, MEM_ROOT *root)
 {
-  engine_option_value *UNINIT_VAR(end), *opt;
-  DBUG_ENTER("merge_engine_table_options");
+  engine_option_value *UNINIT_VAR(end), *opt, *opt_copy;
+  *out= 0;
+  DBUG_ENTER("merge_engine_options");
 
-  /* Create copy of first list */
-  for (opt= first, first= 0; opt; opt= opt->next)
-    new (root) engine_option_value(opt, &first, &end);
+  /* Create copy of source list */
+  for (opt= source; opt; opt= opt->next)
+  {
+    opt_copy= new (root) engine_option_value(opt);
+    if (!opt_copy)
+      DBUG_RETURN(TRUE);
+    opt_copy->link(out, &end);
+  }
 
-  for (opt= second; opt; opt= opt->next)
-    new (root) engine_option_value(opt->name, opt->value, opt->quoted_value,
-                                   &first, &end);
-  DBUG_RETURN(first);
+  for (opt= changes; opt; opt= opt->next)
+  {
+    opt_copy= new (root) engine_option_value(opt);
+    if (!opt_copy)
+      DBUG_RETURN(TRUE);
+    opt_copy->link(out, &end);
+  }
+  DBUG_RETURN(FALSE);
 }
 
 bool is_engine_option_known(engine_option_value *opt,
diff --git a/sql/create_options.h b/sql/create_options.h
index ce64516794b..4961231820f 100644
--- a/sql/create_options.h
+++ b/sql/create_options.h
@@ -1,4 +1,4 @@
-/* Copyright (C) 2010 Monty Program Ab
+/* Copyright (C) 2010, 2021, MariaDB Corporation.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -35,30 +35,23 @@ class engine_option_value: public Sql_alloc
   bool parsed;                  ///< to detect unrecognized options
   bool quoted_value;            ///< option=VAL vs. option='VAL'
 
-  engine_option_value(engine_option_value *src,
-                      engine_option_value **start, engine_option_value **end) :
+  engine_option_value(engine_option_value *src) :
     name(src->name), value(src->value),
     next(NULL), parsed(src->parsed), quoted_value(src->quoted_value)
   {
-    link(start, end);
   }
   engine_option_value(LEX_CSTRING &name_arg, LEX_CSTRING &value_arg,
-                      bool quoted,
-                      engine_option_value **start, engine_option_value **end) :
+                      bool quoted) :
     name(name_arg), value(value_arg),
     next(NULL), parsed(false), quoted_value(quoted)
   {
-    link(start, end);
   }
-  engine_option_value(LEX_CSTRING &name_arg,
-                      engine_option_value **start, engine_option_value **end) :
+  engine_option_value(LEX_CSTRING &name_arg):
     name(name_arg), value(null_clex_str),
     next(NULL), parsed(false), quoted_value(false)
   {
-    link(start, end);
   }
   engine_option_value(LEX_CSTRING &name_arg, ulonglong value_arg,
-                      engine_option_value **start, engine_option_value **end,
                       MEM_ROOT *root) :
     name(name_arg), next(NULL), parsed(false), quoted_value(false)
   {
@@ -66,7 +59,6 @@ class engine_option_value: public Sql_alloc
     if (likely((value.str= str= (char *)alloc_root(root, 22))))
     {
       value.length= longlong10_to_str(value_arg, str, 10) - str;
-      link(start, end);
     }
   }
   static uchar *frm_read(const uchar *buff, const uchar *buff_end,
@@ -83,15 +75,18 @@ class Create_field;
 bool resolve_sysvar_table_options(handlerton *hton);
 void free_sysvar_table_options(handlerton *hton);
 bool parse_engine_table_options(THD *thd, handlerton *ht, TABLE_SHARE *share);
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+bool parse_engine_part_options(THD *thd, TABLE *table);
+#endif
 bool parse_option_list(THD* thd, handlerton *hton, void *option_struct,
                        engine_option_value **option_list,
                        ha_create_table_option *rules,
                        bool suppress_warning, MEM_ROOT *root);
 bool engine_table_options_frm_read(const uchar *buff, size_t length,
                                    TABLE_SHARE *share);
-engine_option_value *merge_engine_table_options(engine_option_value *source,
-                                                engine_option_value *changes,
-                                                MEM_ROOT *root);
+bool merge_engine_options(engine_option_value *source,
+                          engine_option_value *changes,
+                          engine_option_value **out, MEM_ROOT *root);
 
 uint engine_table_options_frm_length(engine_option_value *table_option_list,
                                      List<Create_field> &create_fields,
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index f17abed82ff..85f2e5ac78e 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -2716,6 +2716,7 @@ register_query_cache_dependant_tables(THD *thd,
     2) MAX_ROWS, MIN_ROWS on partition
     3) Index file name on partition
     4) Data file name on partition
+    5) Engine-defined attributes on partition
 */
 
 int ha_partition::set_up_table_before_create(TABLE *tbl,
@@ -2753,6 +2754,10 @@ int ha_partition::set_up_table_before_create(TABLE *tbl,
   if (info->connect_string.length)
     info->used_fields|= HA_CREATE_USED_CONNECTION;
   tbl->s->connect_string= part_elem->connect_string;
+  if (part_elem->option_list)
+    tbl->s->option_list= part_elem->option_list;
+  if (part_elem->option_struct)
+    tbl->s->option_struct= part_elem->option_struct;
   DBUG_RETURN(0);
 }
 
diff --git a/sql/partition_element.h b/sql/partition_element.h
index c551baa3092..756cab2b7f2 100644
--- a/sql/partition_element.h
+++ b/sql/partition_element.h
@@ -2,6 +2,7 @@
 #define PARTITION_ELEMENT_INCLUDED
 
 /* Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
+   Copyright (c) 2021, MariaDB Corporation.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -126,6 +127,9 @@ public:
   bool empty;
   elem_type_enum type;
 
+  engine_option_value *option_list;      // create options for partition
+  ha_table_option_struct *option_struct; // structure with parsed options
+
   partition_element()
   : part_max_rows(0), part_min_rows(0), range_value(0),
     partition_name(NULL),
@@ -136,7 +140,8 @@ public:
     signed_flag(FALSE), max_value(FALSE),
     id(UINT_MAX32),
     empty(true),
-    type(CONVENTIONAL)
+    type(CONVENTIONAL),
+    option_list(NULL), option_struct(NULL)
   {}
   partition_element(partition_element *part_elem)
   : part_max_rows(part_elem->part_max_rows),
@@ -155,7 +160,9 @@ public:
     max_value(part_elem->max_value),
     id(part_elem->id),
     empty(part_elem->empty),
-    type(CONVENTIONAL)
+    type(CONVENTIONAL),
+    option_list(part_elem->option_list),
+    option_struct(part_elem->option_struct)
   {}
   ~partition_element() {}
 
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index b75a318ab65..3b7d10aa8a8 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -1,5 +1,5 @@
 /* Copyright (c) 2005, 2017, Oracle and/or its affiliates.
-   Copyright (c) 2009, 2020, MariaDB
+   Copyright (c) 2009, 2020, 2021, MariaDB
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -69,6 +69,7 @@
 #include "sql_select.h"
 #include "ddl_log.h"
 #include "tztime.h"                     // my_tz_OFFSET0
+#include "create_options.h"             // engine_option_value
 
 #include <algorithm>
 using std::max;
@@ -2205,7 +2206,7 @@ static int add_keyword_int(String *str, const char *keyword, longlong num)
   return err + str->append_longlong(num);
 }
 
-static int add_partition_options(String *str, partition_element *p_elem)
+static int add_server_part_options(String *str, partition_element *p_elem)
 {
   int err= 0;
 
@@ -2232,6 +2233,20 @@ static int add_partition_options(String *str, partition_element *p_elem)
   return err;
 }
 
+static int add_engine_part_options(String *str, partition_element *p_elem)
+{
+  engine_option_value *opt= p_elem->option_list;
+
+  for (; opt; opt= opt->next)
+  {
+    if (!opt->value.str)
+      continue;
+    if ((add_keyword_string(str, opt->name.str, opt->quoted_value,
+                                 opt->value.str)))
+      return 1;
+  }
+  return 0;
+}
 
 /*
   Find the given field's Create_field object using name of field
@@ -2655,7 +2670,10 @@ char *generate_partition_syntax(THD *thd, partition_info *part_info,
             part_info->use_default_subpartitions)
         {
           if (show_partition_options)
-            err+= add_partition_options(&str, part_elem);
+          {
+            err+= add_server_part_options(&str, part_elem);
+            err+= add_engine_part_options(&str, part_elem);
+          }
         }
         else
         {
@@ -2669,7 +2687,7 @@ char *generate_partition_syntax(THD *thd, partition_info *part_info,
             err+= append_identifier(thd, &str, part_elem->partition_name,
                                                strlen(part_elem->partition_name));
             if (show_partition_options)
-              err+= add_partition_options(&str, part_elem);
+              err+= add_server_part_options(&str, part_elem);
             if (j != (num_subparts-1))
               err+= str.append(STRING_WITH_LEN(",\n  "));
             else
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index b07efb29bba..8c2c6aab3d6 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -7859,8 +7859,9 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
          thd->calloc(sizeof(void*) * table->s->keys)) == NULL)
     DBUG_RETURN(1);
 
-  create_info->option_list= merge_engine_table_options(table->s->option_list,
-                                        create_info->option_list, thd->mem_root);
+  if (merge_engine_options(table->s->option_list, create_info->option_list,
+                           &create_info->option_list, thd->mem_root))
+    DBUG_RETURN(1);
 
   /*
     First collect all fields from table which isn't in drop_list
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 442644eddd6..ace14285a4b 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -1,6 +1,6 @@
 /*
    Copyright (c) 2000, 2015, Oracle and/or its affiliates.
-   Copyright (c) 2010, 2021, MariaDB
+   Copyright (c) 2010, 2020, 2021, MariaDB
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -301,6 +301,7 @@ void _CONCAT_UNDERSCORED(turn_parser_debug_on,yyparse)()
   class With_element_head *with_element_head;
   class With_clause *with_clause;
   class Virtual_column_info *virtual_column;
+  engine_option_value *engine_option_value_ptr;
 
   handlerton *db_type;
   st_select_lex *select_lex;
@@ -1802,6 +1803,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %type <vers_range_unit> opt_history_unit
 %type <vers_history_point> history_point
 %type <vers_column_versioning> with_or_without_system
+%type <engine_option_value_ptr> engine_defined_option;
 
 %ifdef MARIADB
 %type <NONE> sp_tail_standalone
@@ -5040,7 +5042,7 @@ sub_part_definition:
             part_info->use_default_num_subpartitions= FALSE;
             part_info->count_curr_subparts++;
           }
-          sub_name opt_part_options {}
+          sub_name opt_subpart_options {}
         ;
 
 sub_name:
@@ -5054,15 +5056,34 @@ sub_name:
 
 opt_part_options:
          /* empty */ {}
-       | opt_part_option_list {}
+       | part_option_list {}
        ;
 
-opt_part_option_list:
-         opt_part_option_list opt_part_option {}
-       | opt_part_option {}
+part_option_list:
+         part_option_list part_option {}
+       | part_option {}
        ;
 
-opt_part_option:
+part_option:
+          server_part_option {}
+        | engine_defined_option
+          {
+            $1->link(&Lex->part_info->curr_part_elem->option_list,
+                     &Lex->option_list_last);
+          }
+        ;
+
+opt_subpart_options:
+         /* empty */ {}
+       | subpart_option_list {}
+       ;
+
+subpart_option_list:
+         subpart_option_list server_part_option {}
+       | server_part_option {}
+       ;
+
+server_part_option:
           TABLESPACE opt_equal ident_or_text
           { /* Compatibility with MySQL */ }
         | opt_storage ENGINE_SYM opt_equal storage_engines
@@ -5424,42 +5445,43 @@ create_table_option:
 	    Lex->create_info.used_fields|= HA_CREATE_USED_TRANSACTIONAL;
             Lex->create_info.transactional= $3;
           }
-        | IDENT_sys equal TEXT_STRING_sys
+        | engine_defined_option
+          {
+            $1->link(&Lex->create_info.option_list, &Lex->option_list_last);
+          }
+        | SEQUENCE_SYM opt_equal choice
+          {
+            Lex->create_info.used_fields|= HA_CREATE_USED_SEQUENCE;
+            Lex->create_info.sequence= ($3 == HA_CHOICE_YES);
+          }
+        | versioning_option
+        ;
+
+engine_defined_option:
+          IDENT_sys equal TEXT_STRING_sys
           {
             if (unlikely($3.length > ENGINE_OPTION_MAX_LENGTH))
               my_yyabort_error((ER_VALUE_TOO_LONG, MYF(0), $1.str));
-            (void) new (thd->mem_root)
-                   engine_option_value($1, $3, true,
-                                       &Lex->create_info.option_list,
-                                       &Lex->option_list_last);
+            $$= new (thd->mem_root) engine_option_value($1, $3, true);
+            MYSQL_YYABORT_UNLESS($$);
           }
         | IDENT_sys equal ident
           {
             if (unlikely($3.length > ENGINE_OPTION_MAX_LENGTH))
               my_yyabort_error((ER_VALUE_TOO_LONG, MYF(0), $1.str));
-            (void) new (thd->mem_root)
-                   engine_option_value($1, $3, false,
-                                       &Lex->create_info.option_list,
-                                       &Lex->option_list_last);
+            $$= new (thd->mem_root) engine_option_value($1, $3, false);
+            MYSQL_YYABORT_UNLESS($$);
           }
         | IDENT_sys equal real_ulonglong_num
           {
-            (void) new (thd->mem_root)
-                   engine_option_value($1, $3, &Lex->create_info.option_list,
-                                       &Lex->option_list_last, thd->mem_root);
+            $$= new (thd->mem_root) engine_option_value($1, $3, thd->mem_root);
+            MYSQL_YYABORT_UNLESS($$);
           }
         | IDENT_sys equal DEFAULT
           {
-            (void) new (thd->mem_root)
-                   engine_option_value($1, &Lex->create_info.option_list,
-                                       &Lex->option_list_last);
+            $$= new (thd->mem_root) engine_option_value($1);
+            MYSQL_YYABORT_UNLESS($$);
           }
-        | SEQUENCE_SYM opt_equal choice
-          {
-	    Lex->create_info.used_fields|= HA_CREATE_USED_SEQUENCE;
-            Lex->create_info.sequence= ($3 == HA_CHOICE_YES);
-	  }
-        | versioning_option
         ;
 
 opt_versioning_option:
@@ -6349,35 +6371,9 @@ asrow_attribute:
 
 serial_attribute:
           asrow_attribute
-        | IDENT_sys equal TEXT_STRING_sys
-          {
-            if (unlikely($3.length > ENGINE_OPTION_MAX_LENGTH))
-              my_yyabort_error((ER_VALUE_TOO_LONG, MYF(0), $1.str));
-            (void) new (thd->mem_root)
-                   engine_option_value($1, $3, true,
-                                       &Lex->last_field->option_list,
-                                       &Lex->option_list_last);
-          }
-        | IDENT_sys equal ident
+        | engine_defined_option
           {
-            if (unlikely($3.length > ENGINE_OPTION_MAX_LENGTH))
-              my_yyabort_error((ER_VALUE_TOO_LONG, MYF(0), $1.str));
-            (void) new (thd->mem_root)
-                   engine_option_value($1, $3, false,
-                                       &Lex->last_field->option_list,
-                                       &Lex->option_list_last);
-          }
-        | IDENT_sys equal real_ulonglong_num
-          {
-            (void) new (thd->mem_root)
-                   engine_option_value($1, $3, &Lex->last_field->option_list,
-                                       &Lex->option_list_last, thd->mem_root);
-          }
-        | IDENT_sys equal DEFAULT
-          {
-            (void) new (thd->mem_root)
-                   engine_option_value($1, &Lex->last_field->option_list,
-                                       &Lex->option_list_last);
+            $1->link(&Lex->last_field->option_list, &Lex->option_list_last);
           }
         | with_or_without_system VERSIONING_SYM
           {
@@ -6780,33 +6776,9 @@ all_key_opt:
           {
             Lex->last_key->key_create_info.is_ignored= $1;
           }
-        | IDENT_sys equal TEXT_STRING_sys
-          {
-            if (unlikely($3.length > ENGINE_OPTION_MAX_LENGTH))
-              my_yyabort_error((ER_VALUE_TOO_LONG, MYF(0), $1.str));
-            (void) new (thd->mem_root)
-                   engine_option_value($1, $3, true, &Lex->option_list,
-                                       &Lex->option_list_last);
-          }
-        | IDENT_sys equal ident
-          {
-            if (unlikely($3.length > ENGINE_OPTION_MAX_LENGTH))
-              my_yyabort_error((ER_VALUE_TOO_LONG, MYF(0), $1.str));
-            (void) new (thd->mem_root)
-                   engine_option_value($1, $3, false, &Lex->option_list,
-                                       &Lex->option_list_last);
-          }
-        | IDENT_sys equal real_ulonglong_num
-          {
-            (void) new (thd->mem_root)
-                  engine_option_value($1, $3, &Lex->option_list,
-                                      &Lex->option_list_last, thd->mem_root);
-          }
-        | IDENT_sys equal DEFAULT
+        | engine_defined_option
           {
-            (void) new (thd->mem_root)
-                   engine_option_value($1, &Lex->option_list,
-                                       &Lex->option_list_last);
+            $1->link(&Lex->option_list, &Lex->option_list_last);
           }
         ;
 
diff --git a/sql/table.cc b/sql/table.cc
index b1a7b6bfe2b..6349db84f61 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -4244,6 +4244,8 @@ enum open_frm_error open_table_from_share(THD *thd, TABLE_SHARE *share,
       thd->restore_active_arena(&part_func_arena, &backup_arena);
       goto partititon_err;
     }
+    if (parse_engine_part_options(thd, outparam))
+      goto err;
     outparam->part_info->is_auto_partitioned= share->auto_partitioned;
     DBUG_PRINT("info", ("autopartitioned: %u", share->auto_partitioned));
     /*