MWL#43 CREATE TABLE options (by Sanja)

Docs/sp-imp-spec.txt:
  New sql_mode added.
include/my_base.h:
  Flag in frm of create options.
libmysqld/CMakeLists.txt:
  New files added.
libmysqld/Makefile.am:
  New files added.
mysql-test/r/events_bugs.result:
  New sql_mode added.
mysql-test/r/information_schema.result:
  New sql_mode added.
mysql-test/r/sp.result:
  New sql_mode added.
mysql-test/r/system_mysql_db.result:
  New sql_mode added.
mysql-test/suite/funcs_1/r/is_columns_mysql.result:
  New sql_mode added.
mysql-test/suite/funcs_1/r/is_columns_mysql_embedded.result:
  New sql_mode added.
mysql-test/t/events_bugs.test:
  New sql_mode added.
mysql-test/t/sp.test:
  New sql_mode added.
scripts/mysql_system_tables.sql:
  New sql_mode added.
scripts/mysql_system_tables_fix.sql:
  New sql_mode added.
sql/CMakeLists.txt:
  New files added.
sql/Makefile.am:
  New files added.
sql/event_db_repository.cc:
  New sql_mode added.
sql/field.cc:
  Create options support added.
sql/field.h:
  Create options support added.
sql/ha_partition.cc:
  Create options support added.
sql/handler.cc:
  Create options support added.
sql/handler.h:
  Create options support added.
sql/log_event.h:
  New sql_mode added.
sql/mysql_priv.h:
  New sql_mode added.
sql/mysqld.cc:
  New sql_mode added.
sql/share/errmsg.txt:
  New error messages added.
sql/sp.cc:
  New sql_mode added.
sql/sp_head.cc:
  Create options support added.
sql/sql_class.cc:
  Create options support added.
  Debug added.
sql/sql_class.h:
  Create options support added.
sql/sql_insert.cc:
  my_safe_a* moved to mysqld_priv.h
sql/sql_lex.h:
  Create options support added.
sql/sql_parse.cc:
  Create options support added.
sql/sql_show.cc:
  Create options support added.
sql/sql_table.cc:
  Create options support added.
sql/sql_view.cc:
  New sql_mode added.
sql/sql_yacc.yy:
  Create options support added.
sql/structs.h:
  Create options support added.
sql/table.cc:
  Create options support added.
sql/table.h:
  Create options support added.
sql/unireg.cc:
  Create options support added.
storage/example/ha_example.cc:
  Create options example.
storage/example/ha_example.h:
  Create options example.
storage/pbxt/src/discover_xt.cc:
  Create options support added.

29 files changed, 1182 insertions, 116 deletions

diff --git a/sql/CMakeLists.txt b/sql/CMakeLists.txt
index 69fbaa0d587..a13e726fe99 100755
--- a/sql/CMakeLists.txt
+++ b/sql/CMakeLists.txt
@@ -77,6 +77,7 @@ SET (SQL_SOURCE
                rpl_rli.cc rpl_mi.cc sql_servers.cc
                sql_connect.cc scheduler.cc 
                sql_profile.cc event_parse_data.cc opt_table_elimination.cc
+               create_options.cc
                ${PROJECT_SOURCE_DIR}/sql/sql_yacc.cc
                ${PROJECT_SOURCE_DIR}/sql/sql_yacc.h
                ${PROJECT_SOURCE_DIR}/include/mysqld_error.h
diff --git a/sql/Makefile.am b/sql/Makefile.am
index a3393c6bfc2..b738565f818 100644
--- a/sql/Makefile.am
+++ b/sql/Makefile.am
@@ -78,7 +78,8 @@ noinst_HEADERS =	item.h item_func.h item_sum.h item_cmpfunc.h \
 			sql_plugin.h authors.h event_parse_data.h \
 			event_data_objects.h event_scheduler.h \
 			sql_partition.h partition_info.h partition_element.h \
-			contributors.h sql_servers.h
+			contributors.h sql_servers.h \
+			create_options.h
 
 mysqld_SOURCES =	sql_lex.cc sql_handler.cc sql_partition.cc \
 			item.cc item_sum.cc item_buff.cc item_func.cc \
@@ -124,7 +125,7 @@ mysqld_SOURCES =	sql_lex.cc sql_handler.cc sql_partition.cc \
 			sql_plugin.cc sql_binlog.cc \
 			sql_builtin.cc sql_tablespace.cc partition_info.cc \
 			sql_servers.cc event_parse_data.cc \
-                        opt_table_elimination.cc
+                        opt_table_elimination.cc create_options.cc
 
 nodist_mysqld_SOURCES =	mini_client_errors.c pack.c client.c my_time.c my_user.c client_plugin.c
 
diff --git a/sql/create_options.cc b/sql/create_options.cc
new file mode 100644
index 00000000000..f4868f95330
--- /dev/null
+++ b/sql/create_options.cc
@@ -0,0 +1,612 @@
+/* Copyright (C) 2010 Monty Program Ab
+
+   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
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
+
+/**
+  @file
+
+  Engine defined options of tables/fields/keys in CREATE/ALTER TABLE.
+*/
+
+#include "mysql_priv.h"
+#include "create_options.h"
+#include <my_getopt.h>
+
+#define FRM_QUOTED_VALUE 0x8000
+
+/**
+  Links this item to the given list end
+
+  @param start           The list beginning or NULL
+  @param end             The list last element or does not matter
+*/
+
+void engine_option_value::link(engine_option_value **start,
+                               engine_option_value **end)
+{
+  DBUG_ENTER("engine_option_value::link");
+  DBUG_PRINT("enter", ("name: '%s' (%u)  value: '%s' (%u)",
+                       name.str, (uint) name.length,
+                       value.str, (uint) value.length));
+  engine_option_value *opt;
+  /* check duplicates to avoid writing them to frm*/
+  for(opt= *start;
+      opt && ((opt->parsed && !opt->value.str) ||
+              my_strnncoll(system_charset_info,
+                           (uchar *)name.str, name.length,
+                           (uchar*)opt->name.str, opt->name.length));
+      opt= opt->next) /* no-op */;
+  if (opt)
+  {
+    opt->value.str= NULL;       /* remove previous value */
+    opt->parsed= TRUE;          /* and don't issue warnings for it anymore */
+  }
+  /*
+    Add this option to the end of the list
+
+    @note: We add even if it is opt->value.str == NULL because it can be
+    ALTER TABLE to remove the option.
+  */
+  if (*start)
+  {
+    (*end)->next= this;
+    *end= this;
+  }
+  else
+  {
+    /*
+      note that is *start == 0, the value of *end does not matter,
+      it can be uninitialized.
+    */
+    *start= *end= this;
+  }
+  DBUG_VOID_RETURN;
+}
+
+static bool report_wrong_value(THD *thd, const char *name, const char *val,
+                               my_bool suppress_warning)
+{
+  if (!(thd->variables.sql_mode & MODE_IGNORE_BAD_TABLE_OPTIONS))
+  {
+    my_error(ER_BAD_OPTION_VALUE, MYF(0), val, name);
+    return 1;
+  }
+
+  /*
+    We may need to suppress warnings to avoid duplicate messages
+    about the same option (option list is parsed more than once during
+    CREATE/ALTER table).
+    Errors are not suppressed, as they abort the execution on the first parsing.
+  */
+  if (!suppress_warning)
+    push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                        ER_BAD_OPTION_VALUE,
+                        ER(ER_BAD_OPTION_VALUE), val, name);
+  return 0;
+}
+
+static bool report_unknown_option(THD *thd, engine_option_value *val,
+                                  my_bool suppress_warning)
+{
+  DBUG_ENTER("report_unknown_option");
+  if (val->parsed)
+  {
+    DBUG_PRINT("info", ("parsed => exiting"));
+    DBUG_RETURN(FALSE);
+  }
+
+  if (!(thd->variables.sql_mode & MODE_IGNORE_BAD_TABLE_OPTIONS))
+  {
+    my_error(ER_UNKNOWN_OPTION, MYF(0), val->name.str);
+    DBUG_RETURN(TRUE);
+  }
+
+  if (!suppress_warning)
+    push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                        ER_UNKNOWN_OPTION,
+                        ER(ER_UNKNOWN_OPTION),
+                        val->name.str);
+  DBUG_RETURN(FALSE);
+}
+
+static bool set_one_value(ha_create_table_option *opt,
+                          THD *thd, LEX_STRING *value, void *base,
+                          my_bool suppress_warning,
+                          MEM_ROOT *root)
+{
+  DBUG_ENTER("set_one_value");
+  DBUG_PRINT("enter", ("opt: 0x%lx type: %u name '%s' value: '%s'",
+                       (ulong) opt,
+                       opt->type, opt->name,
+                       (value->str ? value->str : "<DEFAULT>")));
+  switch (opt->type)
+  {
+  case HA_OPTION_TYPE_ULL:
+    {
+      ulonglong *val= (ulonglong*)((char*)base + opt->offset);
+      if (!value->str)
+      {
+        *val= opt->def_value;
+        DBUG_RETURN(0);
+      }
+
+      my_option optp= { opt->name, 1, 0, (uchar **)val, 0, 0, GET_ULL,
+        REQUIRED_ARG, opt->def_value, opt->min_value, opt->max_value,
+        0, opt->block_size, 0};
+
+      ulonglong orig_val= strtoull(value->str, NULL, 10);
+      my_bool unused;
+      *val= orig_val;
+      *val= getopt_ull_limit_value(*val, &optp, &unused);
+      if (*val == orig_val)
+        DBUG_RETURN(0);
+
+      DBUG_RETURN(report_wrong_value(thd, opt->name, value->str,
+                                     suppress_warning));
+    }
+  case HA_OPTION_TYPE_STRING:
+    {
+      char **val= (char **)((char *)base + opt->offset);
+      if (!value->str)
+      {
+        *val= 0;
+        DBUG_RETURN(0);
+      }
+
+      if (!(*val= strmake_root(root, value->str, value->length)))
+        DBUG_RETURN(1);
+      DBUG_RETURN(0);
+    }
+  case HA_OPTION_TYPE_ENUM:
+    {
+      uint *val= (uint *)((char *)base + opt->offset), num;
+
+      *val= opt->def_value;
+      if (!value->str)
+        DBUG_RETURN(0);
+
+      const char *start= opt->values, *end;
+
+      num= 0;
+      while (*start)
+      {
+        for (end=start;
+             *end && *end != ',';
+             end+= my_mbcharlen(system_charset_info, *end)) /* no-op */;
+        if (!my_strnncoll(system_charset_info,
+                          (uchar*)start, end-start,
+                          (uchar*)value->str, value->length))
+        {
+          *val= num;
+          DBUG_RETURN(0);
+        }
+        if (*end) *end++;
+        start= end;
+        num++;
+      }
+
+      DBUG_RETURN(report_wrong_value(thd, opt->name, value->str,
+                                     suppress_warning));
+    }
+  case HA_OPTION_TYPE_BOOL:
+    {
+      bool *val= (bool *)((char *)base + opt->offset);
+      *val= opt->def_value;
+
+      if (!value->str)
+        DBUG_RETURN(0);
+
+      if (!my_strnncoll(system_charset_info,
+                        (const uchar*)"NO", 2,
+                        (uchar *)value->str, value->length) ||
+          !my_strnncoll(system_charset_info,
+                        (const uchar*)"OFF", 3,
+                        (uchar *)value->str, value->length) ||
+          !my_strnncoll(system_charset_info,
+                        (const uchar*)"0", 1,
+                        (uchar *)value->str, value->length))
+      {
+        *val= FALSE;
+        DBUG_RETURN(FALSE);
+      }
+
+      if (!my_strnncoll(system_charset_info,
+                        (const uchar*)"YES", 3,
+                        (uchar *)value->str, value->length) ||
+          !my_strnncoll(system_charset_info,
+                        (const uchar*)"ON", 2,
+                        (uchar *)value->str, value->length) ||
+          !my_strnncoll(system_charset_info,
+                        (const uchar*)"1", 1,
+                        (uchar *)value->str, value->length))
+      {
+        *val= TRUE;
+        DBUG_RETURN(FALSE);
+      }
+
+      DBUG_RETURN(report_wrong_value(thd, opt->name, value->str,
+                                     suppress_warning));
+    }
+  }
+  DBUG_ASSERT(0);
+  my_error(ER_UNKNOWN_ERROR, MYF(0));
+  DBUG_RETURN(1);
+}
+
+static const size_t ha_option_type_sizeof[]=
+{ sizeof(ulonglong), sizeof(char *), sizeof(uint), sizeof(bool)};
+
+/**
+  Creates option structure and parses list of options in it
+
+  @param thd              thread handler
+  @param option_struct    where to store pointer on the option struct
+  @param option_list      list of options given by user
+  @param rules            list of option description by engine
+  @param suppress_warning second parse so we do not need warnings
+  @param root             MEM_ROOT where allocate memory
+
+  @retval TRUE  Error
+  @retval FALSE OK
+*/
+
+my_bool parse_option_list(THD* thd, void **option_struct,
+                          engine_option_value *option_list,
+                          ha_create_table_option *rules,
+                          my_bool suppress_warning,
+                          MEM_ROOT *root)
+{
+  ha_create_table_option *opt;
+  size_t option_struct_size= 0;
+  engine_option_value *val= option_list;
+  DBUG_ENTER("parse_option_list");
+  DBUG_PRINT("enter",
+             ("struct: 0x%lx list: 0x%lx rules: 0x%lx suppres %u root 0x%lx",
+              (ulong) *option_struct, (ulong)option_list, (ulong)rules,
+              (uint) suppress_warning, (ulong) root));
+
+  if (rules)
+  {
+    LEX_STRING default_val= {NULL, 0};
+    for (opt= rules; opt->name; opt++)
+      set_if_bigger(option_struct_size, opt->offset +
+                    ha_option_type_sizeof[opt->type]);
+
+    *option_struct= alloc_root(root, option_struct_size);
+
+    /* set all values to default */
+    for (opt= rules; opt->name; opt++)
+      set_one_value(opt, thd, &default_val, *option_struct,
+                    suppress_warning, root);
+  }
+
+  for (; val; val= val->next)
+  {
+    for (opt= rules; opt && opt->name; opt++)
+    {
+      if (my_strnncoll(system_charset_info,
+                       (uchar*)opt->name, opt->name_length,
+                       (uchar*)val->name.str, val->name.length))
+        continue;
+
+      if (set_one_value(opt, thd, &val->value,
+                        *option_struct, suppress_warning || val->parsed, root))
+        DBUG_RETURN(TRUE);
+      val->parsed= true;
+      break;
+    }
+    if (report_unknown_option(thd, val, suppress_warning))
+      DBUG_RETURN(TRUE);
+    val->parsed= true;
+  }
+
+  DBUG_RETURN(FALSE);
+}
+
+
+/**
+  Parses all table/fields/keys options
+
+  @param thd             thread handler
+  @param file            handler of the table
+  @parem share           descriptor of the table
+
+  @retval TRUE  Error
+  @retval FALSE OK
+*/
+
+my_bool parse_engine_table_options(THD *thd, handlerton *ht,
+                                   TABLE_SHARE *share)
+{
+  MEM_ROOT *root= &share->mem_root;
+  DBUG_ENTER("parse_engine_table_options");
+
+  if (parse_option_list(thd, &share->option_struct, share->option_list,
+                        ht->table_options, TRUE, root))
+    DBUG_RETURN(TRUE);
+
+  for (Field **field= share->field; *field; field++)
+  {
+    if (parse_option_list(thd, &(*field)->option_struct, (*field)->option_list,
+                          ht->field_options, TRUE, root))
+      DBUG_RETURN(TRUE);
+  }
+
+  for (uint index= 0; index < share->keys; index ++)
+  {
+    if (parse_option_list(thd, &share->key_info[index].option_struct,
+                          share->key_info[index].option_list,
+                          ht->index_options, TRUE, root))
+      DBUG_RETURN(TRUE);
+  }
+
+  DBUG_RETURN(FALSE);
+}
+
+
+/**
+  Returns representation length of key and value in the frm file
+*/
+
+uint engine_option_value::frm_length()
+{
+  /*
+    1 byte  - name length
+    2 bytes - value length
+
+    if value.str is NULL, this option is not written to frm (=DEFAULT)
+  */
+  return value.str ? 1 + name.length + 2 + value.length : 0;
+}
+
+
+/**
+  Returns length of representation of option list in the frm file
+*/
+
+static uint option_list_frm_length(engine_option_value *opt)
+{
+  uint res= 0;
+
+  for (; opt; opt= opt->next)
+    res+= opt->frm_length();
+
+  return res;
+}
+
+
+/**
+  Calculates length of options image in the .frm
+
+  @param table_option_list list of table options
+  @param create_fields     field descriptors list
+  @param keys              number of keys
+  @param key_info          array of key descriptors
+
+  @returns length of image in frm
+*/
+
+uint engine_table_options_frm_length(engine_option_value *table_option_list,
+                                     List<Create_field> &create_fields,
+                                     uint keys, KEY *key_info)
+{
+  List_iterator<Create_field> it(create_fields);
+  Create_field *field;
+  uint res, index;
+  DBUG_ENTER("engine_table_options_frm_length");
+
+  res= option_list_frm_length(table_option_list);
+
+  while ((field= it++))
+    res+= option_list_frm_length(field->option_list);
+
+  for (index= 0; index < keys; index++, key_info++)
+    res+= option_list_frm_length(key_info->option_list);
+
+  /*
+    if there's at least one option somewhere (res > 0)
+    we write option lists for all fields and keys, zero-terminated.
+    If there're no options we write nothing at all (backward compatibility)
+  */
+  DBUG_RETURN(res ? res + 1 + create_fields.elements + keys : 0);
+}
+
+
+/**
+  Writes image of the key and value to the frm image buffer
+
+  @param buff            pointer to the buffer free space beginning
+
+  @returns pointer to byte after last recorded in the buffer
+*/
+
+uchar *engine_option_value::frm_image(uchar *buff)
+{
+  if (value.str)
+  {
+    *buff++= name.length;
+    memcpy(buff, name.str, name.length);
+    buff+= name.length;
+    int2store(buff, value.length | (quoted_value ? FRM_QUOTED_VALUE : 0));
+    buff+= 2;
+    memcpy(buff, (const uchar *) value.str, value.length);
+    buff+= value.length;
+  }
+  return buff;
+}
+
+/**
+  Writes image of the key and value to the frm image buffer
+
+  @param buff            pointer to the buffer to store the options in
+  @param opt             list of options;
+
+  @returns pointer to the end of the stored data in the buffer
+*/
+static uchar *option_list_frm_image(uchar *buff, engine_option_value *opt)
+{
+  for (; opt; opt= opt->next)
+    buff= opt->frm_image(buff);
+
+  *buff++= 0;
+  return buff;
+}
+
+
+/**
+  Writes options image in the .frm buffer
+
+  @param buff              pointer to the buffer
+  @param table_option_list list of table options
+  @param create_fields     field descriptors list
+  @param keys              number of keys
+  @param key_info          array of key descriptors
+
+  @returns pointer to byte after last recorded in the buffer
+*/
+
+uchar *engine_table_options_frm_image(uchar *buff,
+                                      engine_option_value *table_option_list,
+                                      List<Create_field> &create_fields,
+                                      uint keys, KEY *key_info)
+{
+  List_iterator<Create_field> it(create_fields);
+  Create_field *field;
+  KEY *key_info_end= key_info + keys;
+  DBUG_ENTER("engine_table_options_frm_image");
+
+  buff= option_list_frm_image(buff, table_option_list);
+
+  while ((field= it++))
+    buff= option_list_frm_image(buff, field->option_list);
+
+  while (key_info < key_info_end)
+    buff= option_list_frm_image(buff, (key_info++)->option_list);
+
+  DBUG_RETURN(buff);
+}
+
+/**
+  Reads name and value from buffer, then link it in the list
+
+  @param buff            the buffer to read from
+  @param start           The list beginning or NULL
+  @param end             The list last element or does not matter
+  @param root            MEM_ROOT for allocating
+
+  @returns pointer to byte after last recorded in the buffer
+*/
+uchar *engine_option_value::frm_read(const uchar *buff, engine_option_value **start,
+                                     engine_option_value **end, MEM_ROOT *root)
+{
+  LEX_STRING name, value;
+  uint len;
+
+  name.length= buff[0];
+  buff++;
+  if (!(name.str= strmake_root(root, (const char*)buff, name.length)))
+    return NULL;
+  buff+= name.length;
+  len= uint2korr(buff);
+  value.length= len & ~FRM_QUOTED_VALUE;
+  buff+= 2;
+  if (!(value.str= strmake_root(root, (const char*)buff, value.length)))
+    return NULL;
+  buff+= value.length;
+
+  engine_option_value *ptr=new (root)
+    engine_option_value(name, value, len & FRM_QUOTED_VALUE, start, end);
+  if (!ptr)
+    return NULL;
+
+  return (uchar *)buff;
+}
+
+
+/**
+  Reads options from this buffer
+
+  @param buff            the buffer to read from
+  @param length          buffer length
+  @param share           table descriptor
+  @param root            MEM_ROOT for allocating
+
+  @retval TRUE  Error
+  @retval FALSE OK
+*/
+
+my_bool engine_table_options_frm_read(const uchar *buff, uint length,
+                                      TABLE_SHARE *share)
+{
+  const uchar *buff_end= buff + length;
+  engine_option_value *end;
+  MEM_ROOT *root= &share->mem_root;
+  uint count;
+  DBUG_ENTER("engine_table_options_frm_read");
+
+  while (buff < buff_end && *buff)
+  {
+    if (!(buff= engine_option_value::frm_read(buff, &share->option_list, &end,
+                                              root)))
+      DBUG_RETURN(TRUE);
+  }
+  buff++;
+
+  for (count=0; count < share->fields; count++)
+  {
+    while (buff < buff_end && *buff)
+    {
+      if (!(buff= engine_option_value::frm_read(buff,
+                                                &share->field[count]->option_list,
+                                                &end, root)))
+        DBUG_RETURN(TRUE);
+    }
+    buff++;
+  }
+
+  for (count=0; count < share->keys; count++)
+  {
+    while (buff < buff_end && *buff)
+    {
+      if (!(buff= engine_option_value::frm_read(buff,
+                                                &share->key_info[count].option_list,
+                                                &end, root)))
+        DBUG_RETURN(TRUE);
+    }
+    buff++;
+  }
+
+  DBUG_RETURN(buff != buff_end);
+}
+
+/**
+  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)
+{
+  engine_option_value *end, *opt;
+  DBUG_ENTER("merge_engine_table_options");
+
+  /* find last element */
+  if (first && second)
+    for (end= first; end->next; end= end->next) /* no-op */;
+
+  for (opt= second; opt; opt= opt->next)
+    new (root) engine_option_value(opt->name, opt->value, opt->quoted_value,
+                                   &first, &end);
+  DBUG_RETURN(first);
+}
diff --git a/sql/create_options.h b/sql/create_options.h
new file mode 100644
index 00000000000..b66bbf43570
--- /dev/null
+++ b/sql/create_options.h
@@ -0,0 +1,92 @@
+/* Copyright (C) 2010 Monty Program Ab
+
+   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
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
+
+/**
+  @file
+
+  Engine defined options of tables/fields/keys in CREATE/ALTER TABLE.
+*/
+
+#ifndef SQL_CREATE_OPTIONS_INCLUDED
+#define SQL_CREATE_OPTIONS_INCLUDED
+
+#include "handler.h"
+
+class engine_option_value: public Sql_alloc
+{
+ public:
+  LEX_STRING name;
+  LEX_STRING value;
+  engine_option_value *next;    ///< parser puts them in a FIFO linked list
+  bool parsed;                  ///< to detect unrecognized options
+  bool quoted_value;            ///< option=VAL vs. option='VAL'
+
+  engine_option_value(LEX_STRING &name_arg, LEX_STRING &value_arg, bool quoted,
+                      engine_option_value **start, engine_option_value **end) :
+    name(name_arg), value(value_arg),
+    next(NULL), parsed(false), quoted_value(quoted)
+  {
+    link(start, end);
+  }
+  engine_option_value(LEX_STRING &name_arg,
+                      engine_option_value **start, engine_option_value **end) :
+    name(name_arg), value(null_lex_str),
+    next(NULL), parsed(false), quoted_value(false)
+  {
+    link(start, end);
+  }
+  engine_option_value(LEX_STRING &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)
+  {
+    if ((value.str= (char *)alloc_root(root, 22)))
+    {
+      value.length= longlong10_to_str(value_arg, value.str, 10) - value.str;
+      link(start, end);
+    }
+  }
+  static uchar *frm_read(const uchar *buff, engine_option_value **start,
+                         engine_option_value **end, MEM_ROOT *root);
+  void link(engine_option_value **start, engine_option_value **end);
+  uint frm_length();
+  uchar *frm_image(uchar *buff);
+};
+
+typedef struct st_key KEY;
+class Create_field;
+
+my_bool parse_engine_table_options(THD *thd, handlerton *ht,
+                                   TABLE_SHARE *share);
+my_bool parse_option_list(THD* thd, void **option_struct,
+                          engine_option_value *option_list,
+                          ha_create_table_option *rules,
+                          my_bool suppress_warning,
+                          MEM_ROOT *root);
+my_bool engine_table_options_frm_read(const uchar *buff,
+                                      uint length,
+                                      TABLE_SHARE *share);
+engine_option_value *merge_engine_table_options(engine_option_value *source,
+                                                engine_option_value *changes,
+                                                MEM_ROOT *root);
+
+uint engine_table_options_frm_length(engine_option_value *table_option_list,
+                                     List<Create_field> &create_fields,
+                                     uint keys, KEY *key_info);
+uchar *engine_table_options_frm_image(uchar *buff,
+                                      engine_option_value *table_option_list,
+                                      List<Create_field> &create_fields,
+                                      uint keys, KEY *key_info);
+#endif
diff --git a/sql/event_db_repository.cc b/sql/event_db_repository.cc
index c6595e952cc..82f9d354888 100644
--- a/sql/event_db_repository.cc
+++ b/sql/event_db_repository.cc
@@ -105,7 +105,8 @@ const TABLE_FIELD_TYPE event_table_fields[ET_FIELD_COUNT] =
   {
     { C_STRING_WITH_LEN("sql_mode") },
     { C_STRING_WITH_LEN("set('REAL_AS_FLOAT','PIPES_AS_CONCAT','ANSI_QUOTES',"
-    "'IGNORE_SPACE','NOT_USED','ONLY_FULL_GROUP_BY','NO_UNSIGNED_SUBTRACTION',"
+    "'IGNORE_SPACE','IGNORE_BAD_TABLE_OPTIONS','ONLY_FULL_GROUP_BY',"
+    "'NO_UNSIGNED_SUBTRACTION',"
     "'NO_DIR_IN_CREATE','POSTGRESQL','ORACLE','MSSQL','DB2','MAXDB',"
     "'NO_KEY_OPTIONS','NO_TABLE_OPTIONS','NO_FIELD_OPTIONS','MYSQL323','MYSQL40',"
     "'ANSI','NO_AUTO_VALUE_ON_ZERO','NO_BACKSLASH_ESCAPES','STRICT_TRANS_TABLES',"
diff --git a/sql/field.cc b/sql/field.cc
index a6751151beb..99848bb7fbd 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -1306,14 +1306,13 @@ String *Field::val_int_as_str(String *val_buffer, my_bool unsigned_val)
 Field::Field(uchar *ptr_arg,uint32 length_arg,uchar *null_ptr_arg,
 	     uchar null_bit_arg,
 	     utype unireg_check_arg, const char *field_name_arg)
-  :ptr(ptr_arg), null_ptr(null_ptr_arg),
-   table(0), orig_table(0), table_name(0),
-   field_name(field_name_arg),
-   key_start(0), part_of_key(0), part_of_key_not_clustered(0),
-   part_of_sortkey(0), unireg_check(unireg_check_arg),
-   field_length(length_arg), null_bit(null_bit_arg), 
-   is_created_from_null_item(FALSE),
-   vcol_info(0), stored_in_db(TRUE)
+  :ptr(ptr_arg), null_ptr(null_ptr_arg), table(0), orig_table(0),
+  table_name(0), field_name(field_name_arg), option_list(0),
+  option_struct(0), key_start(0), part_of_key(0),
+  part_of_key_not_clustered(0), part_of_sortkey(0),
+  unireg_check(unireg_check_arg), field_length(length_arg),
+  null_bit(null_bit_arg), is_created_from_null_item(FALSE), vcol_info(0),
+  stored_in_db(TRUE)
 {
   flags=null_ptr ? 0: NOT_NULL_FLAG;
   comment.str= (char*) "";
@@ -9567,7 +9566,8 @@ bool Create_field::init(THD *thd, char *fld_name, enum_field_types fld_type,
                         Item *fld_on_update_value, LEX_STRING *fld_comment,
                         char *fld_change, List<String> *fld_interval_list,
                         CHARSET_INFO *fld_charset, uint fld_geom_type,
-			Virtual_column_info *fld_vcol_info)
+			Virtual_column_info *fld_vcol_info,
+                        engine_option_value *create_opt)
 {
   uint sign_len, allowed_type_modifier= 0;
   ulong max_field_charlength= MAX_FIELD_CHARLENGTH;
@@ -9578,6 +9578,7 @@ bool Create_field::init(THD *thd, char *fld_name, enum_field_types fld_type,
   field_name= fld_name;
   def= fld_default_value;
   flags= fld_type_modifier;
+  option_list= create_opt;
   unireg_check= (fld_type_modifier & AUTO_INCREMENT_FLAG ?
                  Field::NEXT_NUMBER : Field::NONE);
   decimals= fld_decimals ? (uint)atoi(fld_decimals) : 0;
@@ -10217,6 +10218,8 @@ Create_field::Create_field(Field *old_field,Field *orig_field)
   decimals=   old_field->decimals();
   vcol_info=  old_field->vcol_info;
   stored_in_db= old_field->stored_in_db;
+  option_list= old_field->option_list;
+  option_struct= old_field->option_struct;
 
   /* Fix if the original table had 4 byte pointer blobs */
   if (flags & BLOB_FLAG)
@@ -10291,6 +10294,19 @@ Create_field::Create_field(Field *old_field,Field *orig_field)
 
 
 /**
+  Makes a clone of this object for ALTER/CREATE TABLE
+
+  @param mem_root        MEM_ROOT where to clone the field
+*/
+
+Create_field *Create_field::clone(MEM_ROOT *mem_root) const
+{
+  Create_field *res= new (mem_root) Create_field(*this);
+  return res;
+}
+
+
+/**
   maximum possible display length for blob.
 
   @return
diff --git a/sql/field.h b/sql/field.h
index 042e4610b51..d74e2d163ee 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -137,6 +137,9 @@ public:
   struct st_table *table;		// Pointer for table
   struct st_table *orig_table;		// Pointer to original table
   const char	**table_name, *field_name;
+  /** reference to the list of options or NULL */
+  engine_option_value *option_list;
+  void *option_struct;                  /* structure with parsed options */
   LEX_STRING	comment;
   /* Field is part of the following keys */
   key_map	key_start, part_of_key, part_of_key_not_clustered;
@@ -2145,6 +2148,9 @@ public:
   CHARSET_INFO *charset;
   Field::geometry_type geom_type;
   Field *field;				// For alter table
+  engine_option_value *option_list;
+  /** structure with parsed options (for comparing fields in ALTER TABLE) */
+  void *option_struct;
 
   uint8 row,col,sc_length,interval_id;	// For rea_create_table
   uint	offset,pack_flag;
@@ -2162,11 +2168,11 @@ public:
   */
   bool stored_in_db;
 
-  Create_field() :after(0) {}
+  Create_field() :after(0), option_list(NULL), option_struct(NULL)
+  {}
   Create_field(Field *field, Field *orig_field);
   /* Used to make a clone of this object for ALTER/CREATE TABLE */
-  Create_field *clone(MEM_ROOT *mem_root) const
-    { return new (mem_root) Create_field(*this); }
+  Create_field *clone(MEM_ROOT *mem_root) const;
   void create_length_to_internal_length(void);
 
   /* Init for a tmp table field. To be extended if need be. */
@@ -2178,8 +2184,8 @@ public:
             char *decimals, uint type_modifier, Item *default_value,
             Item *on_update_value, LEX_STRING *comment, char *change,
             List<String> *interval_list, CHARSET_INFO *cs,
-            uint uint_geom_type,
-	    Virtual_column_info *vcol_info);
+            uint uint_geom_type, Virtual_column_info *vcol_info,
+            engine_option_value *option_list);
 
   bool field_flags_are_binary()
   {
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index a88723db7c5..c065a35b8e3 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -52,6 +52,7 @@
 #endif
 
 #include "mysql_priv.h"
+#include "create_options.h"
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
 #include "ha_partition.h"
@@ -1218,7 +1219,9 @@ int ha_partition::prepare_new_partition(TABLE *tbl,
   DBUG_ENTER("prepare_new_partition");
 
   if ((error= set_up_table_before_create(tbl, part_name, create_info,
-                                         0, p_elem)))
+                                         0, p_elem)) ||
+      parse_engine_table_options(ha_thd(), file->ht,
+                                 file->table_share))
     goto error_create;
   if ((error= file->ha_create(part_name, tbl, create_info)))
   {
@@ -1869,6 +1872,8 @@ uint ha_partition::del_ren_cre_table(const char *from,
     {
       if ((error= set_up_table_before_create(table_arg, from_buff,
                                              create_info, i, NULL)) ||
+          parse_engine_table_options(ha_thd(), (*file)->ht,
+                                     (*file)->table_share) ||
           ((error= (*file)->ha_create(from_buff, table_arg, create_info))))
         goto create_error;
     }
diff --git a/sql/handler.cc b/sql/handler.cc
index 28d898e1614..44f9809626d 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -24,6 +24,7 @@
 #endif
 
 #include "mysql_priv.h"
+#include "create_options.h"
 #include "rpl_filter.h"
 #include <myisampack.h>
 #include "myisam.h"
@@ -3719,7 +3720,11 @@ int ha_create_table(THD *thd, const char *path,
 
   name= get_canonical_filename(table.file, share.path.str, name_buff);
 
+  if (parse_engine_table_options(thd, table.file->ht, &share))
+    goto err;
+
   error= table.file->ha_create(name, &table, create_info);
+
   VOID(closefrm(&table, 0));
   if (error)
   {
diff --git a/sql/handler.h b/sql/handler.h
index 0df20a2b769..ef69c18e846 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -16,6 +16,9 @@
 
 /* Definitions for parameters to do with handler-routines */
 
+#ifndef SQL_HANDLER_INCLUDED
+#define SQL_HANDLER_INCLUDED
+
 #ifdef USE_PRAGMA_INTERFACE
 #pragma interface			/* gcc class implementation */
 #endif
@@ -549,6 +552,103 @@ struct handler_log_file_data {
   enum log_status status;
 };
 
+/*
+  Definitions for engine-specific table/field/index options in the CREATE TABLE.
+
+  Options are declared with HA_*OPTION_* macros (HA_TOPTION_ULL, HA_FOPTION_ENUM,
+  HA_KOPTION_STRING, etc).
+
+  Every macros takes the option name, and the name of the underlying field of
+  the appropriate C structure. The "appropriate C structure" is
+  ha_table_option_struct for table level options,
+  ha_field_option_struct for field level options,
+  ha_key_option_struct for key level options. The engine either
+  defines a structure of this name, or uses #define's to map
+  these "appropriate" names to the actual structure type name.
+
+  ULL options use a ulonglong as the backing store.
+  HA_*OPTION_ULL() takes the option name, the structure field name,
+  the default value for the option, min, max, and blk_siz values.
+
+  STRING options use a char* as a backing store.
+  HA_*OPTION_STRING takes the option name and the structure field name.
+  The default value will be 0.
+
+  ENUM options use a uint as a backing store (not enum!!!).
+  HA_*OPTION_ENUM takes the option name, the structure field name,
+  the default value for the option as a number, and a string with the
+  permitted values for this enum - one string with comma separated values,
+  for example: "gzip,bzip2,lzma"
+
+  BOOL options use a bool as a backing store.
+  HA_*OPTION_BOOL takes the option name, the structure field name,
+  and the default value for the option.
+  From the SQL, BOOL options accept YES/NO, ON/OFF, and 1/0.
+
+  The name of the option is limited to 255 bytes,
+  the value (for string options) - to the 32767 bytes.
+
+  See ha_example.cc for an example.
+*/
+enum ha_option_type { HA_OPTION_TYPE_ULL,    /* unsigned long long */
+                      HA_OPTION_TYPE_STRING, /* char * */
+                      HA_OPTION_TYPE_ENUM,   /* uint */
+                      HA_OPTION_TYPE_BOOL};  /* bool */
+
+#define HA_xOPTION_ULL(name, struc, field, def, min, max, blk_siz)   \
+  { HA_OPTION_TYPE_ULL, name, sizeof(name)-1,                        \
+    offsetof(struc, field), def, min, max, blk_siz, 0 }
+#define HA_xOPTION_STRING(name, struc, field)                        \
+  { HA_OPTION_TYPE_STRING, name, sizeof(name)-1,                     \
+    offsetof(struc, field), 0, 0, 0, 0, 0 }
+#define HA_xOPTION_ENUM(name, struc, field, values, def)             \
+  { HA_OPTION_TYPE_ENUM, name, sizeof(name)-1,                       \
+    offsetof(struc, field), def, 0,                                  \
+    sizeof(values)-1, 0, values }
+#define HA_xOPTION_BOOL(name, struc, field, def)                     \
+  { HA_OPTION_TYPE_BOOL, name, sizeof(name)-1,                       \
+    offsetof(struc, field), def, 0, 1, 0, 0 }
+#define HA_xOPTION_END { HA_OPTION_TYPE_ULL, 0, 0, 0, 0, 0, 0, 0, 0 }
+
+#define HA_TOPTION_ULL(name, field, def, min, max, blk_siz)          \
+  HA_xOPTION_ULL(name, ha_table_option_struct, field, def, min, max, blk_siz)
+#define HA_TOPTION_STRING(name, field)                               \
+  HA_xOPTION_STRING(name, ha_table_option_struct, field)
+#define HA_TOPTION_ENUM(name, field, values, def)                    \
+  HA_xOPTION_ENUM(name, ha_table_option_struct, field, values, def)
+#define HA_TOPTION_BOOL(name, field, def)                            \
+  HA_xOPTION_BOOL(name, ha_table_option_struct, field, def)
+#define HA_TOPTION_END HA_xOPTION_END
+
+#define HA_FOPTION_ULL(name, field, def, min, max, blk_siz)          \
+  HA_xOPTION_ULL(name, ha_field_option_struct, field, def, min, max, blk_siz)
+#define HA_FOPTION_STRING(name, field)                               \
+  HA_xOPTION_STRING(name, ha_field_option_struct, field)
+#define HA_FOPTION_ENUM(name, field, values, def)                    \
+  HA_xOPTION_ENUM(name, ha_field_option_struct, field, values, def)
+#define HA_FOPTION_BOOL(name, field, def)                            \
+  HA_xOPTION_BOOL(name, ha_field_option_struct, field, def)
+#define HA_FOPTION_END HA_xOPTION_END
+
+#define HA_KOPTION_ULL(name, field, def, min, max, blk_siz)          \
+  HA_xOPTION_ULL(name, ha_key_option_struct, field, def, min, max, blk_siz)
+#define HA_KOPTION_STRING(name, field)                               \
+  HA_xOPTION_STRING(name, ha_key_option_struct, field)
+#define HA_KOPTION_ENUM(name, field, values, def)                    \
+  HA_xOPTION_ENUM(name, ha_key_option_struct, field, values, def)
+#define HA_KOPTION_BOOL(name, field, values, def)                    \
+  HA_xOPTION_BOOL(name, ha_key_option_struct, field, values, def)
+#define HA_KOPTION_END HA_xOPTION_END
+
+typedef struct st_ha_create_table_option {
+  enum ha_option_type type;
+  const char *name;
+  size_t name_length;
+  ptrdiff_t offset;
+  ulonglong def_value;
+  ulonglong min_value, max_value, block_size;
+  const char *values;
+} ha_create_table_option;
 
 enum handler_iterator_type
 {
@@ -721,7 +821,13 @@ struct handlerton
    int (*table_exists_in_engine)(handlerton *hton, THD* thd, const char *db,
                                  const char *name);
    uint32 license; /* Flag for Engine License */
-   void *data; /* Location for engines to keep personal structures */
+   /*
+     Optional clauses in the CREATE/ALTER TABLE
+   */
+   ha_create_table_option *table_options; // table level options
+   ha_create_table_option *field_options; // these are specified per field
+   ha_create_table_option *index_options; // these are specified per index
+
 };
 
 
@@ -945,11 +1051,16 @@ typedef struct st_ha_create_information
   uint extra_size;                      /* length of extra data segment */
   /** Transactional or not. Unused; reserved for future versions. */
   enum ha_choice transactional;
-  bool table_existed;			/* 1 in create if table existed */
-  bool frm_only;                        /* 1 if no ha_create_table() */
-  bool varchar;                         /* 1 if table has a VARCHAR */
-  enum ha_storage_media storage_media;  /* DEFAULT, DISK or MEMORY */
-  enum ha_choice page_checksum;         /* If we have page_checksums */
+  bool table_existed;			///< 1 in create if table existed
+  bool frm_only;                        ///< 1 if no ha_create_table()
+  bool varchar;                         ///< 1 if table has a VARCHAR
+  enum ha_storage_media storage_media;  ///< DEFAULT, DISK or MEMORY
+  enum ha_choice page_checksum;         ///< If we have page_checksums
+  engine_option_value *option_list;     ///< list of table create options
+  /* the following three are only for ALTER TABLE, check_if_incompatible_data() */
+  void *option_struct;           ///< structure with parsed table options
+  void **fileds_option_struct;   ///< array of field option structures
+  void **keys_option_struct;     ///< array of key option structures
 } HA_CREATE_INFO;
 
 
@@ -2255,3 +2366,5 @@ int ha_binlog_end(THD *thd);
 #define ha_binlog_wait(a) do {} while (0)
 #define ha_binlog_end(a)  do {} while (0)
 #endif
+
+#endif
diff --git a/sql/log_event.h b/sql/log_event.h
index 9aac52eebbd..d88f53fc864 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -1371,7 +1371,7 @@ protected:
     MODE_PIPES_AS_CONCAT==0x2
     MODE_ANSI_QUOTES==0x4
     MODE_IGNORE_SPACE==0x8
-    MODE_NOT_USED==0x10
+    MODE_IGNORE_BAD_TABLE_OPTIONS==0x10
     MODE_ONLY_FULL_GROUP_BY==0x20
     MODE_NO_UNSIGNED_SUBTRACTION==0x40
     MODE_NO_DIR_IN_CREATE==0x80
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index c9afe10a97d..c178f348206 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -520,7 +520,7 @@ protected:
 #define MODE_PIPES_AS_CONCAT            2
 #define MODE_ANSI_QUOTES                4
 #define MODE_IGNORE_SPACE		8
-#define MODE_NOT_USED			16
+#define MODE_IGNORE_BAD_TABLE_OPTIONS   16
 #define MODE_ONLY_FULL_GROUP_BY		32
 #define MODE_NO_UNSIGNED_SUBTRACTION	64
 #define MODE_NO_DIR_IN_CREATE		128
@@ -1508,7 +1508,8 @@ bool add_field_to_list(THD *thd, LEX_STRING *field_name, enum enum_field_types t
 		       char *change, List<String> *interval_list,
 		       CHARSET_INFO *cs,
 		       uint uint_geom_type,
-                       Virtual_column_info *vcol_info);
+                       Virtual_column_info *vcol_info,
+                       engine_option_value *create_options);
 Create_field * new_create_field(THD *thd, char *field_name, enum_field_types type,
 				char *length, char *decimals,
 				uint type_modifier, 
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index f6bad3cf337..66248461ff8 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -244,7 +244,7 @@ const char *show_comp_option_name[]= {"YES", "NO", "DISABLED"};
 static const char *sql_mode_names[]=
 {
   "REAL_AS_FLOAT", "PIPES_AS_CONCAT", "ANSI_QUOTES", "IGNORE_SPACE",
-  "?", "ONLY_FULL_GROUP_BY", "NO_UNSIGNED_SUBTRACTION",
+  "IGNORE_BAD_TABLE_OPTIONS", "ONLY_FULL_GROUP_BY", "NO_UNSIGNED_SUBTRACTION",
   "NO_DIR_IN_CREATE",
   "POSTGRESQL", "ORACLE", "MSSQL", "DB2", "MAXDB", "NO_KEY_OPTIONS",
   "NO_TABLE_OPTIONS", "NO_FIELD_OPTIONS", "MYSQL323", "MYSQL40", "ANSI",
@@ -264,7 +264,7 @@ static const unsigned int sql_mode_names_len[]=
   /*PIPES_AS_CONCAT*/             15,
   /*ANSI_QUOTES*/                 11,
   /*IGNORE_SPACE*/                12,
-  /*?*/                           1,
+  /*IGNORE_BAD_TABLE_OPTIONS*/    24,
   /*ONLY_FULL_GROUP_BY*/          18,
   /*NO_UNSIGNED_SUBTRACTION*/     23,
   /*NO_DIR_IN_CREATE*/            16,
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index c6c26444c82..9e8360487a4 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -6240,3 +6240,8 @@ ER_DEBUG_SYNC_TIMEOUT
 ER_DEBUG_SYNC_HIT_LIMIT
   eng "debug sync point hit limit reached"
   ger "Debug Sync Point Hit Limit erreicht"
+
+ER_UNKNOWN_OPTION
+        eng "Unknown option '%-.64s'"
+ER_BAD_OPTION_VALUE
+        eng "Incorrect value '%-.64s' for option '%-.64s'"
diff --git a/sql/sp.cc b/sql/sp.cc
index f81e52f40ab..66a2f703df3 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -147,7 +147,8 @@ TABLE_FIELD_TYPE proc_table_fields[MYSQL_PROC_FIELD_COUNT] =
   {
     { C_STRING_WITH_LEN("sql_mode") },
     { C_STRING_WITH_LEN("set('REAL_AS_FLOAT','PIPES_AS_CONCAT','ANSI_QUOTES',"
-    "'IGNORE_SPACE','NOT_USED','ONLY_FULL_GROUP_BY','NO_UNSIGNED_SUBTRACTION',"
+    "'IGNORE_SPACE','IGNORE_BAD_TABLE_OPTIONS','ONLY_FULL_GROUP_BY',"
+    "'NO_UNSIGNED_SUBTRACTION',"
     "'NO_DIR_IN_CREATE','POSTGRESQL','ORACLE','MSSQL','DB2','MAXDB',"
     "'NO_KEY_OPTIONS','NO_TABLE_OPTIONS','NO_FIELD_OPTIONS','MYSQL323','MYSQL40',"
     "'ANSI','NO_AUTO_VALUE_ON_ZERO','NO_BACKSLASH_ESCAPES','STRICT_TRANS_TABLES',"
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index dd253a3910e..84689b2efb2 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -2218,7 +2218,7 @@ sp_head::fill_field_definition(THD *thd, LEX *lex,
                       lex->charset ? lex->charset :
                                      thd->variables.collation_database,
                       lex->uint_geom_type,
-		      lex->vcol_info))
+		      lex->vcol_info, NULL))
     return TRUE;
 
   if (field_def->interval_list.elements)
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 61554e58592..8afcb4bc4a6 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -106,6 +106,7 @@ Key::Key(const Key &rhs, MEM_ROOT *mem_root)
   key_create_info(rhs.key_create_info),
   columns(rhs.columns, mem_root),
   name(rhs.name),
+  option_list(rhs.option_list),
   generated(rhs.generated)
 {
   list_copy_and_replace_each_value(columns, mem_root);
@@ -775,6 +776,7 @@ THD::THD()
 
 void THD::push_internal_handler(Internal_error_handler *handler)
 {
+  DBUG_ENTER("THD::push_internal_handler");
   if (m_internal_handler)
   {
     handler->m_prev_internal_handler= m_internal_handler;
@@ -784,6 +786,7 @@ void THD::push_internal_handler(Internal_error_handler *handler)
   {
     m_internal_handler= handler;
   }
+  DBUG_VOID_RETURN;
 }
 
 
@@ -803,8 +806,10 @@ bool THD::handle_error(uint sql_errno, const char *message,
 
 void THD::pop_internal_handler()
 {
+  DBUG_ENTER("THD::pop_internal_handler");
   DBUG_ASSERT(m_internal_handler != NULL);
   m_internal_handler= m_internal_handler->m_prev_internal_handler;
+  DBUG_VOID_RETURN;
 }
 
 extern "C"
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 33bf1013269..a46fa6cec98 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -205,13 +205,15 @@ public:
   KEY_CREATE_INFO key_create_info;
   List<Key_part_spec> columns;
   const char *name;
+  engine_option_value *option_list;
   bool generated;
 
   Key(enum Keytype type_par, const char *name_arg,
       KEY_CREATE_INFO *key_info_arg,
-      bool generated_arg, List<Key_part_spec> &cols)
+      bool generated_arg, List<Key_part_spec> &cols,
+      engine_option_value *create_opt)
     :type(type_par), key_create_info(*key_info_arg), columns(cols),
-    name(name_arg), generated(generated_arg)
+    name(name_arg), option_list(create_opt), generated(generated_arg)
   {}
   Key(const Key &rhs, MEM_ROOT *mem_root);
   virtual ~Key() {}
@@ -240,7 +242,7 @@ public:
   Foreign_key(const char *name_arg, List<Key_part_spec> &cols,
 	      Table_ident *table,   List<Key_part_spec> &ref_cols,
 	      uint delete_opt_arg, uint update_opt_arg, uint match_opt_arg)
-    :Key(FOREIGN_KEY, name_arg, &default_key_create_info, 0, cols),
+    :Key(FOREIGN_KEY, name_arg, &default_key_create_info, 0, cols, NULL),
     ref_table(table), ref_columns(ref_cols),
     delete_opt(delete_opt_arg), update_opt(update_opt_arg),
     match_opt(match_opt_arg)
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 78cc103793d..eec271d6e13 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -72,15 +72,6 @@ static void unlink_blobs(register TABLE *table);
 #endif
 static bool check_view_insertability(THD *thd, TABLE_LIST *view);
 
-/* Define to force use of my_malloc() if the allocated memory block is big */
-
-#ifndef HAVE_ALLOCA
-#define my_safe_alloca(size, min_length) my_alloca(size)
-#define my_safe_afree(ptr, size, min_length) my_afree(ptr)
-#else
-#define my_safe_alloca(size, min_length) ((size <= min_length) ? my_alloca(size) : my_malloc(size,MYF(0)))
-#define my_safe_afree(ptr, size, min_length) if (size > min_length) my_free(ptr,MYF(0))
-#endif
 
 /*
   Check that insert/update fields are from the same single table of a view.
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index f555784ce4b..22934ad684e 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -1749,6 +1749,11 @@ typedef struct st_lex : public Query_tables_list
   const char *stmt_definition_end;
 
   /**
+    Collects create options for Field and KEY
+  */
+  engine_option_value *option_list, *option_list_last;
+
+  /**
     During name resolution search only in the table list given by 
     Name_resolution_context::first_name_resolution_table and
     Name_resolution_context::last_name_resolution_table
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index d520542700e..ffc1157719b 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -6086,7 +6086,8 @@ bool add_field_to_list(THD *thd, LEX_STRING *field_name, enum_field_types type,
 		       char *change,
                        List<String> *interval_list, CHARSET_INFO *cs,
 		       uint uint_geom_type,
-		       Virtual_column_info *vcol_info)
+		       Virtual_column_info *vcol_info,
+                       engine_option_value *create_options)
 {
   register Create_field *new_field;
   LEX  *lex= thd->lex;
@@ -6104,7 +6105,7 @@ bool add_field_to_list(THD *thd, LEX_STRING *field_name, enum_field_types type,
     lex->col_list.push_back(new Key_part_spec(field_name->str, 0));
     key= new Key(Key::PRIMARY, NullS,
                       &default_key_create_info,
-                      0, lex->col_list);
+                      0, lex->col_list, NULL);
     lex->alter_info.key_list.push_back(key);
     lex->col_list.empty();
   }
@@ -6114,7 +6115,7 @@ bool add_field_to_list(THD *thd, LEX_STRING *field_name, enum_field_types type,
     lex->col_list.push_back(new Key_part_spec(field_name->str, 0));
     key= new Key(Key::UNIQUE, NullS,
                  &default_key_create_info, 0,
-                 lex->col_list);
+                 lex->col_list, NULL);
     lex->alter_info.key_list.push_back(key);
     lex->col_list.empty();
   }
@@ -6172,7 +6173,8 @@ bool add_field_to_list(THD *thd, LEX_STRING *field_name, enum_field_types type,
   if (!(new_field= new Create_field()) ||
       new_field->init(thd, field_name->str, type, length, decimals, type_modifier,
                       default_value, on_update_value, comment, change,
-                      interval_list, cs, uint_geom_type, vcol_info))
+                      interval_list, cs, uint_geom_type, vcol_info,
+                      create_options))
     DBUG_RETURN(1);
 
   lex->alter_info.create_list.push_back(new_field);
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 49e9362c05b..e35cbe17dc5 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -18,6 +18,7 @@
 
 #include "mysql_priv.h"
 #include "sql_select.h"                         // For select_describe
+#include "create_options.h"
 #include "sql_show.h"
 #include "repl_failsafe.h"
 #include "sp.h"
@@ -1043,7 +1044,7 @@ append_identifier(THD *thd, String *packet, const char *name, uint length)
 
   /*
     The identifier must be quoted as it includes a quote character or
-   it's a keyword
+    it's a keyword
   */
 
   VOID(packet->reserve(length*2 + 2));
@@ -1203,6 +1204,31 @@ static bool get_field_default_value(THD *thd, TABLE *table,
   return has_default;
 }
 
+
+/**
+  Appends list of options to string
+
+  @param thd             thread handler
+  @param packet          string to append
+  @param opt             list of options
+*/
+
+static void append_create_options(THD *thd, String *packet,
+				  engine_option_value *opt)
+{
+  for(; opt; opt= opt->next)
+  {
+    DBUG_ASSERT(opt->value.str);
+    packet->append(' ');
+    append_identifier(thd, packet, opt->name.str, opt->name.length);
+    packet->append('=');
+    if (opt->quoted_value)
+      append_unescaped(packet, opt->value.str, opt->value.length);
+    else
+      packet->append(opt->value.str, opt->value.length);
+  }
+}
+
 /*
   Build a CREATE TABLE statement for a table.
 
@@ -1385,6 +1411,7 @@ int store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
       packet->append(STRING_WITH_LEN(" COMMENT "));
       append_unescaped(packet, field->comment.str, field->comment.length);
     }
+    append_create_options(thd, packet, field->option_list);
   }
 
   key_info= table->key_info;
@@ -1456,6 +1483,7 @@ int store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
       append_identifier(thd, packet, parser_name->str, parser_name->length);
       packet->append(STRING_WITH_LEN(" */ "));
     }
+    append_create_options(thd, packet, key_info->option_list);
   }
 
   /*
@@ -1615,6 +1643,7 @@ int store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
       packet->append(STRING_WITH_LEN(" CONNECTION="));
       append_unescaped(packet, share->connect_string.str, share->connect_string.length);
     }
+    append_create_options(thd, packet, share->option_list);
     append_directory(thd, packet, "DATA",  create_info.data_file_name);
     append_directory(thd, packet, "INDEX", create_info.index_file_name);
   }
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 65cdb4ccf7b..fb71cc82efe 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -19,6 +19,7 @@
 #include <hash.h>
 #include <myisam.h>
 #include <my_dir.h>
+#include "create_options.h"
 #include "sp_head.h"
 #include "sql_trigger.h"
 #include "sql_show.h"
@@ -42,17 +43,10 @@ static int copy_data_between_tables(TABLE *from,TABLE *to,
 
 static bool prepare_blob_field(THD *thd, Create_field *sql_field);
 static bool check_engine(THD *, const char *, HA_CREATE_INFO *);
-static int
-mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
-                           Alter_info *alter_info,
-                           bool tmp_table,
-                           uint *db_options,
-                           handler *file, KEY **key_info_buffer,
-                           uint *key_count, int select_field_count);
-static bool
-mysql_prepare_alter_table(THD *thd, TABLE *table,
-                          HA_CREATE_INFO *create_info,
-                          Alter_info *alter_info);
+static int mysql_prepare_create_table(THD *, HA_CREATE_INFO *, Alter_info *,
+                              bool, uint *, handler *, KEY **, uint *, int);
+static bool mysql_prepare_alter_table(THD *, TABLE *, HA_CREATE_INFO *,
+                                      Alter_info *);
 
 #ifndef DBUG_OFF
 
@@ -2863,6 +2857,11 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
     sql_field->offset= record_offset;
     if (MTYP_TYPENR(sql_field->unireg_check) == Field::NEXT_NUMBER)
       auto_increment++;
+    if (parse_option_list(thd, &sql_field->option_struct,
+                          sql_field->option_list,
+                          create_info->db_type->field_options, FALSE,
+                          thd->mem_root))
+      DBUG_RETURN(TRUE);
     /*
       For now skip fields that are not physically stored in the database
       (virtual fields) and update their offset later 
@@ -3061,6 +3060,12 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
     key_info->key_part=key_part_info;
     key_info->usable_key_parts= key_number;
     key_info->algorithm= key->key_create_info.algorithm;
+    key_info->option_list= key->option_list;
+    if (parse_option_list(thd, &key_info->option_struct,
+                          key_info->option_list,
+                          create_info->db_type->index_options, FALSE,
+                          thd->mem_root))
+      DBUG_RETURN(TRUE);
 
     if (key->type == Key::FULLTEXT)
     {
@@ -3438,6 +3443,12 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
     }
   }
 
+  if (parse_option_list(thd, &create_info->option_struct,
+                          create_info->option_list,
+                          create_info->db_type->table_options, FALSE,
+                          thd->mem_root))
+      DBUG_RETURN(TRUE);
+
   DBUG_RETURN(FALSE);
 }
 
@@ -5679,6 +5690,7 @@ compare_tables(TABLE *table,
   KEY_PART_INFO *key_part;
   KEY_PART_INFO *end;
   THD *thd= table->in_use;
+  uint i;
   /*
     Remember if the new definition has new VARCHAR column;
     create_info->varchar will be reset in mysql_prepare_create_table.
@@ -5769,6 +5781,12 @@ compare_tables(TABLE *table,
     DBUG_RETURN(0);
   }
 
+  if ((create_info->fileds_option_struct=
+       (void**)thd->calloc(sizeof(void*) * table->s->fields)) == NULL ||
+      (create_info->keys_option_struct=
+       (void**)thd->calloc(sizeof(void*) * table->s->keys)) == NULL)
+    DBUG_RETURN(1);
+
   /*
     Use transformed info to evaluate possibility of fast ALTER TABLE
     but use the preserved field to persist modifications.
@@ -5776,15 +5794,19 @@ compare_tables(TABLE *table,
   new_field_it.init(alter_info->create_list);
   tmp_new_field_it.init(tmp_alter_info.create_list);
 
-  /*   Go through fields and check if the original ones are compatible
+  /*
+    Go through fields and check if the original ones are compatible
     with new table.
   */
-  for (f_ptr= table->field, new_field= new_field_it++,
+  for (i= 0, f_ptr= table->field, new_field= new_field_it++,
        tmp_new_field= tmp_new_field_it++;
        (field= *f_ptr);
-       f_ptr++, new_field= new_field_it++,
+       i++, f_ptr++, new_field= new_field_it++,
        tmp_new_field= tmp_new_field_it++)
   {
+    DBUG_ASSERT(i < table->s->fields);
+    create_info->fileds_option_struct[i]= tmp_new_field->option_struct;
+
     /* Make sure we have at least the default charset in use. */
     if (!new_field->charset)
       new_field->charset= create_info->default_table_charset;
@@ -5938,7 +5960,9 @@ compare_tables(TABLE *table,
   for (new_key= *key_info_buffer; new_key < new_key_end; new_key++)
   {
     /* Search an old key with the same name. */
-    for (table_key= table->key_info; table_key < table_key_end; table_key++)
+    for (i= 0, table_key= table->key_info;
+         table_key < table_key_end;
+         i++, table_key++)
     {
       if (! strcmp(table_key->name, new_key->name))
         break;
@@ -5957,6 +5981,11 @@ compare_tables(TABLE *table,
       }
       DBUG_PRINT("info", ("index added: '%s'", new_key->name));
     }
+    else
+    {
+      DBUG_ASSERT(i < table->s->keys);
+      create_info->keys_option_struct[i]= new_key->option_struct;
+    }
   }
 
   /* Check if changes are compatible with current handler without a copy */
@@ -6132,6 +6161,8 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
   }
   restore_record(table, s->default_values);     // Empty record for DEFAULT
 
+  create_info->option_list= merge_engine_table_options(table->s->option_list,
+                                        create_info->option_list, thd->mem_root);
   /*
     First collect all fields from table which isn't in drop_list
   */
@@ -6384,7 +6415,7 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
       key= new Key(key_type, key_name,
                    &key_create_info,
                    test(key_info->flags & HA_GENERATED_KEY),
-                   key_parts);
+                   key_parts, key_info->option_list);
       new_key_list.push_back(key);
     }
   }
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index e1e0b3ff6c4..8ef23cfd078 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -1185,7 +1185,7 @@ bool mysql_make_view(THD *thd, File_parser *parser, TABLE_LIST *table,
       + MODE_PIPES_AS_CONCAT          affect expression parsing
       + MODE_ANSI_QUOTES              affect expression parsing
       + MODE_IGNORE_SPACE             affect expression parsing
-      - MODE_NOT_USED                 not used :)
+      - MODE_IGNORE_BAD_TABLE_OPTIONS affect only CREATE/ALTER TABLE parsing
       * MODE_ONLY_FULL_GROUP_BY       affect execution
       * MODE_NO_UNSIGNED_SUBTRACTION  affect execution
       - MODE_NO_DIR_IN_CREATE         affect table creation only
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index ec807738c66..cae37ddf2e6 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -44,6 +44,7 @@
 #include "sp_rcontext.h"
 #include "sp.h"
 #include "event_parse_data.h"
+#include "create_options.h"
 #include <myisam.h>
 #include <myisammrg.h>
 
@@ -608,6 +609,7 @@ static bool add_create_index_prepare (LEX *lex, Table_ident *table)
   lex->alter_info.flags= ALTER_ADD_INDEX;
   lex->col_list.empty();
   lex->change= NullS;
+  lex->option_list= NULL;
   return FALSE;
 }
 
@@ -617,7 +619,7 @@ static bool add_create_index (LEX *lex, Key::Keytype type, const char *name,
 {
   Key *key;
   key= new Key(type, name, info ? info : &lex->key_create_info, generated, 
-               lex->col_list);
+               lex->col_list, lex->option_list);
   if (key == NULL)
     return TRUE;
 
@@ -3898,7 +3900,11 @@ create2:
         ;
 
 create2a:
-          field_list ')' opt_create_table_options
+          field_list ')'
+          {
+            Lex->create_info.option_list= NULL;
+          }
+          opt_create_table_options
           opt_partitioning
           create3 {}
         |  opt_partitioning
@@ -4751,6 +4757,30 @@ create_table_option:
 	    Lex->create_info.used_fields|= HA_CREATE_USED_TRANSACTIONAL;
             Lex->create_info.transactional= $3;
           }
+        | IDENT_sys equal TEXT_STRING_sys
+          {
+            new (YYTHD->mem_root)
+              engine_option_value($1, $3, true, &Lex->create_info.option_list,
+                                  &Lex->option_list_last);
+          }
+        | IDENT_sys equal ident
+          {
+            new (YYTHD->mem_root)
+              engine_option_value($1, $3, false, &Lex->create_info.option_list,
+                                  &Lex->option_list_last);
+          }
+        | IDENT_sys equal ulonglong_num
+          {
+            new (YYTHD->mem_root)
+              engine_option_value($1, $3, &Lex->create_info.option_list,
+                                  &Lex->option_list_last, YYTHD->mem_root);
+          }
+        | IDENT_sys equal DEFAULT
+          {
+            new (YYTHD->mem_root)
+              engine_option_value($1, &Lex->create_info.option_list,
+                                  &Lex->option_list_last);
+          }
         ;
 
 default_charset:
@@ -4872,25 +4902,33 @@ column_def:
         ;
 
 key_def:
-          normal_key_type opt_ident key_alg '(' key_list ')' normal_key_options
+          normal_key_type opt_ident key_alg '(' key_list ')'
+          { Lex->option_list= NULL; }
+          normal_key_options
           {
             if (add_create_index (Lex, $1, $2))
               MYSQL_YYABORT;
           }
         | fulltext opt_key_or_index opt_ident init_key_options 
-            '(' key_list ')' fulltext_key_options
+            '(' key_list ')'
+          { Lex->option_list= NULL; }
+            fulltext_key_options
           {
             if (add_create_index (Lex, $1, $3))
               MYSQL_YYABORT;
           }
         | spatial opt_key_or_index opt_ident init_key_options 
-            '(' key_list ')' spatial_key_options
+            '(' key_list ')'
+          { Lex->option_list= NULL; }
+            spatial_key_options
           {
             if (add_create_index (Lex, $1, $3))
               MYSQL_YYABORT;
           }
         | opt_constraint constraint_key_type opt_ident key_alg
-          '(' key_list ')' normal_key_options
+          '(' key_list ')'
+          { Lex->option_list= NULL; }
+          normal_key_options
           {
             if (add_create_index (Lex, $2, $3 ? $3 : $1))
               MYSQL_YYABORT;
@@ -4953,6 +4991,7 @@ field_spec:
             lex->comment=null_lex_str;
             lex->charset=NULL;
 	    lex->vcol_info= 0;
+            lex->option_list= NULL;
           }
           field_def
           {
@@ -4963,7 +5002,7 @@ field_spec:
                                   &lex->comment,
                                   lex->change,&lex->interval_list,lex->charset,
                                   lex->uint_geom_type,
-                                  lex->vcol_info))
+                                  lex->vcol_info, lex->option_list))
               MYSQL_YYABORT;
           }
         ;
@@ -5383,6 +5422,29 @@ attribute:
               Lex->charset=$2;
             }
           }
+        | IDENT_sys equal TEXT_STRING_sys
+          {
+            new (YYTHD->mem_root)
+              engine_option_value($1, $3, true, &Lex->option_list,
+                                  &Lex->option_list_last);
+          }
+        | IDENT_sys equal ident
+          {
+            new (YYTHD->mem_root)
+              engine_option_value($1, $3, false, &Lex->option_list,
+                                  &Lex->option_list_last);
+          }
+        | IDENT_sys equal ulonglong_num
+          {
+            new (YYTHD->mem_root)
+              engine_option_value($1, $3, &Lex->option_list,
+                                  &Lex->option_list_last, YYTHD->mem_root);
+          }
+        | IDENT_sys equal DEFAULT
+          {
+            new (YYTHD->mem_root)
+              engine_option_value($1, &Lex->option_list, &Lex->option_list_last);
+          }
         ;
 
 now_or_signed_literal:
@@ -5672,6 +5734,29 @@ key_using_alg:
 all_key_opt:
           KEY_BLOCK_SIZE opt_equal ulong_num
           { Lex->key_create_info.block_size= $3; }
+        | IDENT_sys equal TEXT_STRING_sys
+          {
+            new (YYTHD->mem_root)
+              engine_option_value($1, $3, true, &Lex->option_list,
+                                  &Lex->option_list_last);
+          }
+        | IDENT_sys equal ident
+          {
+            new (YYTHD->mem_root)
+              engine_option_value($1, $3, false, &Lex->option_list,
+                                  &Lex->option_list_last);
+          }
+        | IDENT_sys equal ulonglong_num
+          {
+            new (YYTHD->mem_root)
+              engine_option_value($1, $3, &Lex->option_list,
+                                  &Lex->option_list_last, YYTHD->mem_root);
+          }
+        | IDENT_sys equal DEFAULT
+          {
+            new (YYTHD->mem_root)
+              engine_option_value($1, &Lex->option_list, &Lex->option_list_last);
+          }
         ;
 
 normal_key_opt:
@@ -6161,6 +6246,7 @@ alter_list_item:
             LEX *lex=Lex;
             lex->change= $3.str;
             lex->alter_info.flags|= ALTER_CHANGE_COLUMN;
+            lex->option_list= NULL;
           }
           field_spec opt_place
         | MODIFY_SYM opt_column field_ident
@@ -6172,6 +6258,7 @@ alter_list_item:
             lex->charset= NULL;
             lex->alter_info.flags|= ALTER_CHANGE_COLUMN;
 	    lex->vcol_info= 0;
+            lex->option_list= NULL;
           }
           field_def
           {
@@ -6183,7 +6270,7 @@ alter_list_item:
                                   &lex->comment,
                                   $3.str, &lex->interval_list, lex->charset,
                                   lex->uint_geom_type,
-                                  lex->vcol_info))
+                                  lex->vcol_info, lex->option_list))
               MYSQL_YYABORT;
           }
           opt_place
@@ -6290,8 +6377,7 @@ alter_list_item:
           }
         | create_table_options_space_separated
           {
-            LEX *lex=Lex;
-            lex->alter_info.flags|= ALTER_OPTIONS;
+            Lex->alter_info.flags|= ALTER_OPTIONS;
           }
         | FORCE_SYM
           {
diff --git a/sql/structs.h b/sql/structs.h
index cb81e3194fa..64e69fca0d0 100644
--- a/sql/structs.h
+++ b/sql/structs.h
@@ -68,6 +68,7 @@ typedef struct st_key_part_info {	/* Info about a key part */
   uint8 null_bit;			/* Position to null_bit */
 } KEY_PART_INFO ;
 
+class engine_option_value;
 
 typedef struct st_key {
   uint	key_length;			/* Tot length of key */
@@ -101,6 +102,9 @@ typedef struct st_key {
     int  bdb_return_if_eq;
   } handler;
   struct st_table *table;
+  /** reference to the list of options or NULL */
+  engine_option_value *option_list;
+  void *option_struct;                  /* structure with parsed options */
 } KEY;
 
 
diff --git a/sql/table.cc b/sql/table.cc
index b044186f916..cf079ed212a 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -18,6 +18,7 @@
 
 #include "mysql_priv.h"
 #include "sql_trigger.h"
+#include "create_options.h"
 #include <m_ctype.h>
 #include "my_md5.h"
 
@@ -667,12 +668,13 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
   uint db_create_options, keys, key_parts, n_length;
   uint key_info_length, com_length, null_bit_pos;
   uint vcol_screen_length;
-  uint extra_rec_buf_length;
+  uint extra_rec_buf_length, options_len;
   uint i,j;
   bool use_hash;
   char *keynames, *names, *comment_pos, *vcol_screen_pos;
   uchar *record;
-  uchar *disk_buff, *strpos, *null_flags, *null_pos;
+  uchar *disk_buff, *strpos, *null_flags, *null_pos, *options;
+  uchar *buff= 0;
   ulong pos, record_offset, *rec_per_key, rec_buff_length;
   handler *handler_file= 0;
   KEY	*keyinfo;
@@ -788,7 +790,6 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
 
   for (i=0 ; i < keys ; i++, keyinfo++)
   {
-    keyinfo->table= 0;                           // Updated in open_frm
     if (new_frm_ver >= 3)
     {
       keyinfo->flags=	   (uint) uint2korr(strpos) ^ HA_NOSAME;
@@ -858,15 +859,14 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
   if ((n_length= uint4korr(head+55)))
   {
     /* Read extra data segment */
-    uchar *buff, *next_chunk, *buff_end;
+    uchar *next_chunk, *buff_end;
     DBUG_PRINT("info", ("extra segment size is %u bytes", n_length));
     if (!(next_chunk= buff= (uchar*) my_malloc(n_length, MYF(MY_WME))))
       goto err;
     if (my_pread(file, buff, n_length, record_offset + share->reclength,
                  MYF(MY_NABP)))
     {
-      my_free(buff, MYF(0));
-      goto err;
+      goto free_and_err;
     }
     share->connect_string.length= uint2korr(buff);
     if (!(share->connect_string.str= strmake_root(&share->mem_root,
@@ -874,8 +874,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
                                                   share->connect_string.
                                                   length)))
     {
-      my_free(buff, MYF(0));
-      goto err;
+      goto free_and_err;
     }
     next_chunk+= share->connect_string.length + 2;
     buff_end= buff + n_length;
@@ -895,8 +894,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
                 plugin_data(tmp_plugin, handlerton *)))
         {
           /* bad file, legacy_db_type did not match the name */
-          my_free(buff, MYF(0));
-          goto err;
+          goto free_and_err;
         }
         /*
           tmp_plugin is locked with a local lock.
@@ -925,8 +923,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
           error= 8;
           my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0),
                    "--skip-partition");
-          my_free(buff, MYF(0));
-          goto err;
+          goto free_and_err;
         }
         plugin_unlock(NULL, share->db_plugin);
         share->db_plugin= ha_lock_engine(NULL, partition_hton);
@@ -940,8 +937,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
         /* purecov: begin inspected */
         error= 8;
         my_error(ER_UNKNOWN_STORAGE_ENGINE, MYF(0), name.str);
-        my_free(buff, MYF(0));
-        goto err;
+        goto free_and_err;
         /* purecov: end */
       }
       next_chunk+= str_db_type_length + 2;
@@ -957,16 +953,14 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
               memdup_root(&share->mem_root, next_chunk + 4,
                           partition_info_len + 1)))
         {
-          my_free(buff, MYF(0));
-          goto err;
+          goto free_and_err;
         }
       }
 #else
       if (partition_info_len)
       {
         DBUG_PRINT("info", ("WITH_PARTITION_STORAGE_ENGINE is not defined"));
-        my_free(buff, MYF(0));
-        goto err;
+        goto free_and_err;
       }
 #endif
       next_chunk+= 5 + partition_info_len;
@@ -1002,8 +996,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
         {
           DBUG_PRINT("error",
                      ("fulltext key uses parser that is not defined in .frm"));
-          my_free(buff, MYF(0));
-          goto err;
+          goto free_and_err;
         }
         parser_name.str= (char*) next_chunk;
         parser_name.length= strlen((char*) next_chunk);
@@ -1013,12 +1006,20 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
         if (! keyinfo->parser)
         {
           my_error(ER_PLUGIN_IS_NOT_LOADED, MYF(0), parser_name.str);
-          my_free(buff, MYF(0));
-          goto err;
+          goto free_and_err;
         }
       }
     }
-    my_free(buff, MYF(0));
+    if (share->db_create_options & HA_OPTION_TEXT_CREATE_OPTIONS)
+    {
+      /*
+        store options position, but skip till the time we will
+        know number of fields
+      */
+      options_len= uint4korr(next_chunk);
+      options= next_chunk + 4;
+      next_chunk+= options_len + 4;
+    }
   }
   share->key_block_size= uint2korr(head+62);
 
@@ -1028,21 +1029,21 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
   share->rec_buff_length= rec_buff_length;
   if (!(record= (uchar *) alloc_root(&share->mem_root,
                                      rec_buff_length)))
-    goto err;                                   /* purecov: inspected */
+    goto free_and_err;                          /* purecov: inspected */
   share->default_values= record;
   if (my_pread(file, record, (size_t) share->reclength,
                record_offset, MYF(MY_NABP)))
-    goto err;                                   /* purecov: inspected */
+    goto free_and_err;                          /* purecov: inspected */
 
   VOID(my_seek(file,pos,MY_SEEK_SET,MYF(0)));
   if (my_read(file, head,288,MYF(MY_NABP)))
-    goto err;
+    goto free_and_err;
 #ifdef HAVE_CRYPTED_FRM
   if (crypted)
   {
     crypted->decode((char*) head+256,288-256);
     if (sint2korr(head+284) != 0)		// Should be 0
-      goto err;                                 // Wrong password
+      goto free_and_err;                        // Wrong password
   }
 #endif
 
@@ -1062,6 +1063,8 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
                                    share->comment.length);
 
   DBUG_PRINT("info",("i_count: %d  i_parts: %d  index: %d  n_length: %d  int_length: %d  com_length: %d  vcol_screen_length: %d", interval_count,interval_parts, share->keys,n_length,int_length, com_length, vcol_screen_length));
+
+
   if (!(field_ptr = (Field **)
 	alloc_root(&share->mem_root,
 		   (uint) ((share->fields+1)*sizeof(Field*)+
@@ -1070,14 +1073,14 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
 			    keys+3)*sizeof(char *)+
 			   (n_length+int_length+com_length+
 			       vcol_screen_length)))))
-    goto err;                                   /* purecov: inspected */
+    goto free_and_err;                           /* purecov: inspected */
 
   share->field= field_ptr;
   read_length=(uint) (share->fields * field_pack_length +
 		      pos+ (uint) (n_length+int_length+com_length+
 		                   vcol_screen_length));
   if (read_string(file,(uchar**) &disk_buff,read_length))
-    goto err;                                   /* purecov: inspected */
+    goto free_and_err;                           /* purecov: inspected */
 #ifdef HAVE_CRYPTED_FRM
   if (crypted)
   {
@@ -1104,7 +1107,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
 
   fix_type_pointers(&interval_array, &share->fieldnames, 1, &names);
   if (share->fieldnames.count != share->fields)
-    goto err;
+    goto free_and_err;
   fix_type_pointers(&interval_array, share->intervals, interval_count,
 		    &names);
 
@@ -1118,7 +1121,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
       uint count= (uint) (interval->count + 1) * sizeof(uint);
       if (!(interval->type_lengths= (uint *) alloc_root(&share->mem_root,
                                                         count)))
-        goto err;
+        goto free_and_err;
       for (count= 0; count < interval->count; count++)
       {
         char *val= (char*) interval->type_names[count];
@@ -1134,7 +1137,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
  /* Allocate handler */
   if (!(handler_file= get_new_handler(share, thd->mem_root,
                                       share->db_type())))
-    goto err;
+    goto free_and_err;
 
   record= share->default_values-1;              /* Fieldstart = 1 */
   if (share->null_field_first)
@@ -1196,7 +1199,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
 	charset= &my_charset_bin;
 #else
 	error= 4;  // unsupported field type
-	goto err;
+	goto free_and_err;
 #endif
       }
       else
@@ -1207,7 +1210,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
         {
           error= 5; // Unknown or unavailable charset
           errarg= (int) strpos[14];
-          goto err;
+          goto free_and_err;
         }
       }
 
@@ -1247,7 +1250,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
         if ((uint)vcol_screen_pos[0] != 1)
         {
           error= 4;
-          goto err;
+          goto free_and_err;
         }
         field_type= (enum_field_types) (uchar) vcol_screen_pos[1];
         fld_stored_in_db= (bool) (uint) vcol_screen_pos[2];
@@ -1256,7 +1259,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
               (char *)memdup_root(&share->mem_root,
                                   vcol_screen_pos+(uint)FRM_VCOL_HEADER_SIZE,
                                   vcol_expr_length)))
-          goto err;        
+          goto free_and_err;
         vcol_info->expr_str.length= vcol_expr_length;
         vcol_screen_pos+= vcol_info_length;
         share->vfields++;
@@ -1346,7 +1349,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
     if (!reg_field)				// Not supported field type
     {
       error= 4;
-      goto err;			/* purecov: inspected */
+      goto free_and_err;			/* purecov: inspected */
     }
 
     reg_field->field_index= i;
@@ -1385,7 +1388,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
           sent (OOM).
         */
         error= 8; 
-        goto err;
+        goto free_and_err;
       }
     }
     if (!reg_field->stored_in_db)
@@ -1462,7 +1465,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
 	if (!key_part->fieldnr)
         {
           error= 4;                             // Wrong file
-          goto err;
+          goto free_and_err;
         }
         field= key_part->field= share->field[key_part->fieldnr-1];
         key_part->type= field->key_type();
@@ -1627,6 +1630,15 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
           null_length, 255);
   }
 
+  if (share->db_create_options & HA_OPTION_TEXT_CREATE_OPTIONS)
+  {
+    DBUG_ASSERT(options_len);
+    if (engine_table_options_frm_read(options, options_len, share) ||
+        parse_engine_table_options(thd, handler_file->ht, share))
+      goto free_and_err;
+  }
+  my_free(buff, MYF(MY_ALLOW_ZERO_PTR));
+
   if (share->found_next_number_field)
   {
     reg_field= *share->found_next_number_field;
@@ -1685,6 +1697,8 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
 #endif
   DBUG_RETURN (0);
 
+ free_and_err:
+  my_free(buff, MYF(MY_ALLOW_ZERO_PTR));
  err:
   share->error= error;
   share->open_errno= my_errno;
@@ -2883,6 +2897,7 @@ File create_frm(THD *thd, const char *name, const char *db,
   ulong length;
   uchar fill[IO_SIZE];
   int create_flags= O_RDWR | O_TRUNC;
+  DBUG_ENTER("create_frm");
 
   if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
     create_flags|= O_EXCL | O_NOFOLLOW;
@@ -2964,7 +2979,7 @@ File create_frm(THD *thd, const char *name, const char *db,
       {
 	VOID(my_close(file,MYF(0)));
 	VOID(my_delete(name,MYF(0)));
-	return(-1);
+	DBUG_RETURN(-1);
       }
     }
   }
@@ -2975,7 +2990,7 @@ File create_frm(THD *thd, const char *name, const char *db,
     else
       my_error(ER_CANT_CREATE_TABLE,MYF(0),table,my_errno);
   }
-  return (file);
+  DBUG_RETURN(file);
 } /* create_frm */
 
 
@@ -2994,6 +3009,7 @@ void update_create_info_from_table(HA_CREATE_INFO *create_info, TABLE *table)
   create_info->comment= share->comment;
   create_info->transactional= share->transactional;
   create_info->page_checksum= share->page_checksum;
+  create_info->option_list= share->option_list;
 
   DBUG_VOID_RETURN;
 }
diff --git a/sql/table.h b/sql/table.h
index 3a82ce4531a..57932701b93 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -340,6 +340,8 @@ typedef struct st_table_share
 #ifdef NOT_YET
   struct st_table *open_tables;		/* link to open tables */
 #endif
+  engine_option_value *option_list;     /* text options for table */
+  void *option_struct;                  /* structure with parsed options */
 
   /* The following is copied to each TABLE on OPEN */
   Field **field;
diff --git a/sql/unireg.cc b/sql/unireg.cc
index 78d375e663a..3a153277337 100644
--- a/sql/unireg.cc
+++ b/sql/unireg.cc
@@ -24,11 +24,15 @@
 */
 
 #include "mysql_priv.h"
+#include "create_options.h"
 #include <m_ctype.h>
 #include <assert.h>
 
 #define FCOMP			17		/* Bytes for a packed field */
 
+/* threshold for safe_alloca */
+#define ALLOCA_THRESHOLD       2048
+
 static uchar * pack_screens(List<Create_field> &create_fields,
 			    uint *info_length, uint *screens, bool small_file);
 static uint pack_keys(uchar *keybuff,uint key_count, KEY *key_info,
@@ -107,6 +111,7 @@ bool mysql_create_frm(THD *thd, const char *file_name,
   ulong key_buff_length;
   File file;
   ulong filepos, data_offset;
+  uint options_len;
   uchar fileinfo[64],forminfo[288],*keybuff;
   TYPELIB formnames;
   uchar *screen_buff;
@@ -183,6 +188,16 @@ bool mysql_create_frm(THD *thd, const char *file_name,
       create_info->extra_size+= key_info[i].parser_name->length + 1;
   }
 
+  options_len= engine_table_options_frm_length(create_info->option_list,
+                                               create_fields,
+                                               keys, key_info);
+  DBUG_PRINT("info", ("Options length: %u", options_len));
+  if (options_len)
+  {
+    create_info->table_options|= HA_OPTION_TEXT_CREATE_OPTIONS;
+    create_info->extra_size+= (options_len + 4);
+  }
+
   if ((file=create_frm(thd, file_name, db, table, reclength, fileinfo,
 		       create_info, keys)) < 0)
   {
@@ -294,6 +309,7 @@ bool mysql_create_frm(THD *thd, const char *file_name,
     if (my_write(file, (uchar*) buff, 6, MYF_RW))
       goto err;
   }
+
   for (i= 0; i < keys; i++)
   {
     if (key_info[i].parser_name)
@@ -304,6 +320,24 @@ bool mysql_create_frm(THD *thd, const char *file_name,
     }
   }
 
+  if (options_len)
+  {
+    uchar *optbuff= (uchar *)my_safe_alloca(options_len + 4, ALLOCA_THRESHOLD);
+    my_bool error;
+    DBUG_PRINT("info", ("Create options length: %u", options_len));
+    if (!optbuff)
+      goto err;
+    int4store(optbuff, options_len);
+    engine_table_options_frm_image(optbuff + 4,
+                                   create_info->option_list,
+                                   create_fields,
+                                   keys, key_info);
+    error= my_write(file, optbuff, options_len + 4, MYF_RW);
+    my_safe_afree(optbuff, options_len + 4, ALLOCA_THRESHOLD);
+    if (error)
+      goto err;
+  }
+
   VOID(my_seek(file,filepos,MY_SEEK_SET,MYF(0)));
   if (my_write(file, forminfo, 288, MYF_RW) ||
       my_write(file, screen_buff, info_length, MYF_RW) ||