List:Commits« Previous MessageNext Message »
From:Gopal Shankar Date:November 12 2012 6:51pm
Subject:bzr push into mysql-trunk branch (gopal.shankar:4969 to 4970)
View as plain text  
 4970 Gopal Shankar	2012-11-13
      WL6379 - Schema definitions for new DD
      
      Prototype 1
      - A bit of code reorg

    added:
      sql/datadict0.h
    modified:
      sql/datadict.cc
      sql/datadict.h
      sql/sql_table.cc
 4969 Gopal Shankar	2012-11-12
      WL6379 - Schema definitions for new DD
      
      Prototype 1 - patch 1
      
      Aim: Introduce 2 basic new DD tables,
           mysql.tables - To store metadata of user tables.
           mysql.fields - To store metadata of user table fields.
      
      This patch stores most of user table metadata into mysql.tables
      and mysql.fields. A test demonstrates the same.
      
      Notes:
      * Not all FRM contents are stored in new DD in current patch.
        keys, partitioning and triggers are pending.
      
      * Only CREATE TABLE causes the new DD to be updated,
        DROP and ALTER not addressed for now.
      
      * The patch 2 will be focusing on reading metadata and avoid
        creating .frm for user tables.

    modified:
      mysql-test/r/1st.result
      mysql-test/r/connect.result
      mysql-test/r/ctype_cp932_binlog_stm.result
      mysql-test/r/information_schema.result
      mysql-test/r/log_tables_upgrade.result
      mysql-test/r/mysql_upgrade.result
      mysql-test/r/mysql_upgrade_ssl.result
      mysql-test/r/mysqlbinlog.result
      mysql-test/r/mysqlcheck.result
      mysql-test/r/partition_explicit_prune.result
      mysql-test/r/partition_locking.result
      mysql-test/r/plugin_auth.result
      mysql-test/r/show_check.result
      mysql-test/r/system_mysql_db.result
      mysql-test/r/table_definition_cache_functionality.result
      mysql-test/r/wl6219-upgrade.result
      mysql-test/r/wl6443_deprecation.result
      mysql-test/suite/binlog/r/binlog_checksum.result
      mysql-test/suite/binlog/r/binlog_database.result
      mysql-test/suite/binlog/r/binlog_delete_and_flush_index.result
      mysql-test/suite/binlog/r/binlog_drop_if_exists.result
      mysql-test/suite/binlog/r/binlog_innodb.result
      mysql-test/suite/binlog/r/binlog_mysqlbinlog_filter.result
      mysql-test/suite/binlog/r/binlog_server_id.result
      mysql-test/suite/binlog/r/binlog_stm_binlog.result
      mysql-test/suite/binlog/r/binlog_stm_blackhole.result
      mysql-test/suite/binlog/r/binlog_stm_do_db.result
      mysql-test/suite/binlog/r/binlog_stm_drop_tbl.result
      mysql-test/suite/binlog/r/binlog_stm_drop_tmp_tbl.result
      mysql-test/suite/binlog/r/binlog_stm_ps.result
      mysql-test/suite/binlog/r/binlog_stm_row.result
      mysql-test/suite/binlog/r/binlog_stm_user_variables.result
      mysql-test/suite/funcs_1/r/is_columns_mysql.result
      mysql-test/suite/funcs_1/r/is_key_column_usage.result
      mysql-test/suite/funcs_1/r/is_statistics.result
      mysql-test/suite/funcs_1/r/is_statistics_mysql.result
      mysql-test/suite/funcs_1/r/is_table_constraints.result
      mysql-test/suite/funcs_1/r/is_table_constraints_mysql.result
      mysql-test/suite/funcs_1/r/is_tables_mysql.result
      mysql-test/suite/perfschema/r/binlog_edge_mix.result
      mysql-test/suite/perfschema/r/binlog_edge_stmt.result
      mysql-test/suite/perfschema/r/binlog_ok_mix.result
      mysql-test/suite/perfschema/r/binlog_ok_stmt.result
      mysql-test/suite/perfschema/r/myisam_file_io.result
      mysql-test/suite/perfschema/r/no_threads.result
      mysql-test/suite/perfschema/r/one_thread_per_con.result
      mysql-test/suite/perfschema/r/pfs_upgrade_event.result
      mysql-test/suite/perfschema/r/pfs_upgrade_func.result
      mysql-test/suite/perfschema/r/pfs_upgrade_proc.result
      mysql-test/suite/perfschema/r/pfs_upgrade_table.result
      mysql-test/suite/perfschema/r/pfs_upgrade_view.result
      mysql-test/suite/perfschema/r/relaylog.result
      mysql-test/suite/perfschema/r/sizing_default.result
      mysql-test/suite/perfschema/r/sizing_high.result
      mysql-test/suite/perfschema/r/sizing_low.result
      mysql-test/suite/perfschema/r/sizing_med.result
      mysql-test/suite/perfschema/r/stage_mdl_table.result
      mysql-test/suite/rpl/r/rpl_conditional_comments.result
      mysql-test/suite/rpl/r/rpl_do_db_filter.result
      mysql-test/suite/rpl/r/rpl_do_table_filter_insensitive.result
      mysql-test/suite/rpl/r/rpl_do_table_filter_sensitive.result
      mysql-test/suite/rpl/r/rpl_drop_temp.result
      mysql-test/suite/rpl/r/rpl_ignore_db_filter.result
      mysql-test/suite/rpl/r/rpl_ignore_table_filter_insensitive.result
      mysql-test/suite/rpl/r/rpl_ignore_table_filter_sensitive.result
      mysql-test/suite/rpl/r/rpl_innodb_mixed_ddl.result
      mysql-test/suite/rpl/r/rpl_innodb_mixed_dml.result
      mysql-test/suite/rpl/r/rpl_loaddata_map.result
      mysql-test/suite/rpl/r/rpl_rbr_to_sbr.result
      mysql-test/suite/rpl/r/rpl_rewrite_db_filter.result
      mysql-test/suite/rpl/r/rpl_sp.result
      mysql-test/suite/rpl/r/rpl_stm_loaddata_concurrent.result
      mysql-test/suite/rpl/r/rpl_stm_log.result
      mysql-test/suite/rpl/r/rpl_stm_mix_show_relaylog_events.result
      mysql-test/suite/rpl/r/rpl_stm_multi_query.result
      mysql-test/suite/rpl/r/rpl_stm_user_variables.result
      mysql-test/suite/rpl/r/rpl_temp_table_mix_row.result
      scripts/mysql_system_tables.sql
      sql/datadict.cc
      sql/datadict.h
      sql/sql_table.cc
=== modified file 'sql/datadict.cc'
--- a/sql/datadict.cc	2012-11-12 08:00:23 +0000
+++ b/sql/datadict.cc	2012-11-12 18:48:23 +0000
@@ -22,6 +22,7 @@
 #include "sql_base.h"
 #include "key.h"
 #include "lock.h"
+#include "datadict0.h"
 
 /**
   Check type of .frm if we are not going to parse it.
@@ -197,9 +198,8 @@ bool dd_recreate_table(THD *thd, const c
 /*
   New Data Dictionary Prototype
 */
-
-// DD tables
-const char *mysql_dd_catalog[][2]= {
+// DD tables names
+static const char *mysql_dd_catalog[][2]= {
   {"mysql", "tables"},
   {"mysql", "fields"},
   {"mysql", "table_keys"},
@@ -207,8 +207,7 @@ const char *mysql_dd_catalog[][2]= {
   {(const char *)NULL, (const char *)NULL} /* This must be at the end */
 };
 
-// Check if db.name is one of DD table names
-bool is_dd_table(const char *db, const char *name)
+bool Data_dictionary::is_dd_table(const char *db, const char *name)
 {
   int i=0;
   while (mysql_dd_catalog[i][0])
@@ -222,322 +221,237 @@ bool is_dd_table(const char *db, const c
   return false;
 }
 
-bool dd_create_fields(THD *thd, TABLE *table, long long id,
-                     List<Create_field> &create_fields);
-
-/**
-  Write new table into mysql.tables.
-
-  @param thd  Thread context.
-  @param sp   Stored routine object to store.
-
-  @retval false success
-  @retval true  error
-*/
-bool dd_create_table(THD *thd, const char *db, const char *name,
-                     HA_CREATE_INFO *create_info,
-                     List<Create_field> &create_fields,
-                     uint keys, KEY *key_info, handler *file)
+int Data_dictionary::add_table(THD *thd, const char *db, const char *name,
+                               HA_CREATE_INFO *create_info,
+                               List<Create_field> &create_fields, uint keys)
 {
-  bool error= true;
-  TABLE_LIST table_list[2];
-  TABLE *tbl_tables;
-  TABLE *tbl_fields;
-  Open_tables_backup open_tables_state_backup;
-  Dummy_error_handler error_handler;
-  Query_tables_list query_table_list_backup;
+  bool error= false;
   bool store_failed= FALSE;
-  long long new_table_id=0;
-  bool save_binlog_row_based;
-
-  DBUG_ENTER("dd_create_table");
-
-  sql_mode_t saved_mode= thd->variables.sql_mode;
-  enum_check_fields saved_count_cuted_fields;
+  int null_fields= 0;
+  TABLE *table;
+  table= dd_tables[MYSQL_DD_TABLES].table;
 
-  thd->push_internal_handler(&error_handler);
-  MDL_savepoint mdl_savepoint= thd->mdl_context.mdl_savepoint();
-  thd->lex->reset_n_backup_query_tables_list(&query_table_list_backup);
-  thd->reset_n_backup_open_tables_state(&open_tables_state_backup);
-
-  /*
-    This statement will be replicated as a statement, even when using
-    row-based replication.  The flag will be reset at the end of the
-    statement.
-  */
-  if ((save_binlog_row_based= thd->is_current_stmt_binlog_format_row()))
-    thd->clear_current_stmt_binlog_format_row();
+  // Get default values for fields
+  restore_record(table, s->default_values);
 
-  // Grab an exclusive MDL lock. 
-  if (lock_object_name(thd, MDL_key::TABLE, db, name))
-  {
-    my_error(ER_CANT_CREATE_TABLE, MYF(0), name, 1);
-    goto done;
-  }
+  table->next_number_field= table->found_next_number_field;
+  table->auto_increment_field_not_null= TRUE;
 
-  table_list[0].init_one_table("mysql", 5, "tables", 6, "tables", TL_WRITE);
-  table_list[1].init_one_table("mysql", 5, "fields", 6, "fields", TL_WRITE);
-  table_list[0].next_local= table_list[0].next_global= table_list + 1;
+  bitmap_set_all(table->write_set);
+  bitmap_set_all(table->read_set);
 
-  /* Reset sql_mode during data dictionary operations. */
-  thd->variables.sql_mode= 0;
+  table->field[MYSQL_DD_TABLES_FIELD_DB]->set_notnull();
+  store_failed= table->field[MYSQL_DD_TABLES_FIELD_DB]->
+    store(db, strlen(db), system_charset_info);
 
-  saved_count_cuted_fields= thd->count_cuted_fields;
-  thd->count_cuted_fields= CHECK_FIELD_WARN;
+  table->field[MYSQL_DD_TABLES_FIELD_NAME]->set_notnull();
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_NAME]->
+      store(name, strlen(name), system_charset_info);
 
-  if (open_and_lock_tables(thd, table_list, FALSE, MYSQL_LOCK_IGNORE_TIMEOUT))
+  if (thd->auto_inc_intervals_in_cur_stmt_for_binlog.nb_elements() > 0)
   {
-    my_error(ER_CANT_CREATE_TABLE, MYF(0), name, 1);
-    goto done;
-  }
-  else
-  {
-    int null_fields= 0;
-    tbl_tables= table_list[0].table;
-    restore_record(tbl_tables, s->default_values); // Get default values for fields
-    tbl_tables->next_number_field= tbl_tables->found_next_number_field;
-    tbl_tables->auto_increment_field_not_null= TRUE;
-
-    bitmap_set_all(tbl_tables->write_set);
-    bitmap_set_all(tbl_tables->read_set);
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_DB]->set_notnull();
-    store_failed= tbl_tables->field[MYSQL_DD_TABLES_FIELD_DB]->
-        store(db, strlen(db), system_charset_info);
-
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_NAME]->set_notnull();
-    store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_NAME]->
-        store(name, strlen(name), system_charset_info);
-
-    if (thd->auto_inc_intervals_in_cur_stmt_for_binlog.nb_elements() > 0)
-    {
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_ID]->set_notnull();
-      if (tbl_tables->
-          field[MYSQL_DD_TABLES_FIELD_ID]->store(
-            thd->auto_inc_intervals_in_cur_stmt_for_binlog.minimum(), TRUE))
-        goto done;
-    }
-
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_LEGACY_DB_TYPE]->set_notnull();
-    store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_LEGACY_DB_TYPE]->
-        store(ha_legacy_type(create_info->db_type), TRUE);
-
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_MAX_ROWS]->set_notnull();
+    table->field[MYSQL_DD_TABLES_FIELD_ID]->set_notnull();
     store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_MAX_ROWS]->
-        store(create_info->max_rows, TRUE);
-
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_MIN_ROWS]->set_notnull();
-    store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_MIN_ROWS]->
-        store(create_info->min_rows, TRUE);
-
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_SYSTEM]->set_notnull();
-    uchar system= test((create_info->max_rows == 1) &&
-                       (create_info->min_rows == 1) && (keys == 0));
-    store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_SYSTEM]->
-        store(system, TRUE);
-
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_DD_VERSION]->set_notnull();
-    store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_DD_VERSION]->
-        store(FRM_VER + 3 + test(create_info->varchar), TRUE);
-
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_DD_CREATE_OPTIONS]->set_notnull();
-    store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_DD_CREATE_OPTIONS]->
-        store(create_info->table_options|HA_OPTION_LONG_BLOB_PTR, TRUE);
-
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_AVG_ROW_LENGTH]->set_notnull();
-    store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_AVG_ROW_LENGTH]->
-        store(create_info->avg_row_length, TRUE);
-
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_CSID]->set_notnull();
-    uint csid= (create_info->default_table_charset ?
-                create_info->default_table_charset->number : 0);
-    store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_CSID]->
-        store(csid, TRUE);
-
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_ROW_TYPE]->set_notnull();
-    store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_ROW_TYPE]->
-        store(create_info->row_type, TRUE);
-
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_STATS_SAMPLE_PAGES]->set_notnull();
-    store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_STATS_SAMPLE_PAGES]->
-        store(create_info->stats_sample_pages & 0xffff, TRUE);
-
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_STATS_AUTO_RECALC]->set_notnull();
-    store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_STATS_AUTO_RECALC]->
-        store(create_info->stats_auto_recalc, TRUE);
-
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_MYSQL_VERSION]->set_notnull();
-    store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_MYSQL_VERSION]->
-        store(MYSQL_VERSION_ID, TRUE);
-
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_DEFAULT_PART_DB_TYPE]->set_notnull();
-    store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_DEFAULT_PART_DB_TYPE]->
-        store(0, TRUE); //TODO - Handle when dealing with partitions
-
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_KEY_BLOCK_SIZE]->set_notnull();
-    store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_KEY_BLOCK_SIZE]->
-        store(create_info->key_block_size, TRUE); 
-
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_COMMENT]->set_notnull();
-    store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_COMMENT]->
-        store(create_info->comment.str, 
-              create_info->comment.length, system_charset_info);
+      table->field[MYSQL_DD_TABLES_FIELD_ID]->store(
+        thd->auto_inc_intervals_in_cur_stmt_for_binlog.minimum(), TRUE);
+  }
 
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_FIELD_COUNT]->set_notnull();
-    store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_FIELD_COUNT]->
-        store(create_fields.elements, TRUE);
+  table->field[MYSQL_DD_TABLES_FIELD_LEGACY_DB_TYPE]->set_notnull();
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_LEGACY_DB_TYPE]->
+      store(ha_legacy_type(create_info->db_type), TRUE);
+
+  table->field[MYSQL_DD_TABLES_FIELD_MAX_ROWS]->set_notnull();
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_MAX_ROWS]->
+      store(create_info->max_rows, TRUE);
+
+  table->field[MYSQL_DD_TABLES_FIELD_MIN_ROWS]->set_notnull();
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_MIN_ROWS]->
+      store(create_info->min_rows, TRUE);
+
+  table->field[MYSQL_DD_TABLES_FIELD_SYSTEM]->set_notnull();
+  uchar system= test((create_info->max_rows == 1) &&
+                     (create_info->min_rows == 1) && (keys == 0));
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_SYSTEM]->
+      store(system, TRUE);
+
+  table->field[MYSQL_DD_TABLES_FIELD_DD_VERSION]->set_notnull();
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_DD_VERSION]->
+      store(FRM_VER + 3 + test(create_info->varchar), TRUE);
+
+  table->field[MYSQL_DD_TABLES_FIELD_DD_CREATE_OPTIONS]->set_notnull();
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_DD_CREATE_OPTIONS]->
+      store(create_info->table_options|HA_OPTION_LONG_BLOB_PTR, TRUE);
+
+  table->field[MYSQL_DD_TABLES_FIELD_AVG_ROW_LENGTH]->set_notnull();
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_AVG_ROW_LENGTH]->
+      store(create_info->avg_row_length, TRUE);
+
+  table->field[MYSQL_DD_TABLES_FIELD_CSID]->set_notnull();
+  uint csid= (create_info->default_table_charset ?
+              create_info->default_table_charset->number : 0);
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_CSID]->
+      store(csid, TRUE);
+
+  table->field[MYSQL_DD_TABLES_FIELD_ROW_TYPE]->set_notnull();
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_ROW_TYPE]->
+      store(create_info->row_type, TRUE);
+
+  table->field[MYSQL_DD_TABLES_FIELD_STATS_SAMPLE_PAGES]->set_notnull();
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_STATS_SAMPLE_PAGES]->
+      store(create_info->stats_sample_pages & 0xffff, TRUE);
+
+  table->field[MYSQL_DD_TABLES_FIELD_STATS_AUTO_RECALC]->set_notnull();
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_STATS_AUTO_RECALC]->
+      store(create_info->stats_auto_recalc, TRUE);
+
+  table->field[MYSQL_DD_TABLES_FIELD_MYSQL_VERSION]->set_notnull();
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_MYSQL_VERSION]->
+      store(MYSQL_VERSION_ID, TRUE);
+
+  table->field[MYSQL_DD_TABLES_FIELD_DEFAULT_PART_DB_TYPE]->set_notnull();
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_DEFAULT_PART_DB_TYPE]->
+      store(0, TRUE); //TODO - Handle when dealing with partitions
+
+  table->field[MYSQL_DD_TABLES_FIELD_KEY_BLOCK_SIZE]->set_notnull();
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_KEY_BLOCK_SIZE]->
+      store(create_info->key_block_size, TRUE); 
+
+  table->field[MYSQL_DD_TABLES_FIELD_COMMENT]->set_notnull();
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_COMMENT]->
+      store(create_info->comment.str, 
+            create_info->comment.length, system_charset_info);
+
+  table->field[MYSQL_DD_TABLES_FIELD_FIELD_COUNT]->set_notnull();
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_FIELD_COUNT]->
+      store(create_fields.elements, TRUE);
 
-    List_iterator<Create_field> it(create_fields);
-    Create_field *field;
-    while ((field=it++)) 
-      if (f_maybe_null(field->pack_flag))
-          null_fields++;
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_NULL_FIELDS]->set_notnull();
-    store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_NULL_FIELDS]->
-        store(null_fields, TRUE);
+  List_iterator<Create_field> it(create_fields);
+  Create_field *field;
+  while ((field=it++)) 
+    if (f_maybe_null(field->pack_flag))
+        null_fields++;
+  table->field[MYSQL_DD_TABLES_FIELD_NULL_FIELDS]->set_notnull();
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_NULL_FIELDS]->
+      store(null_fields, TRUE);
+
+  table->field[MYSQL_DD_TABLES_FIELD_KEYS]->set_notnull();
+  store_failed= store_failed ||
+    table->field[MYSQL_DD_TABLES_FIELD_KEYS]->
+      store(keys, TRUE);
 
-    tbl_tables->field[MYSQL_DD_TABLES_FIELD_KEYS]->set_notnull();
+  if (create_info->tablespace)
+  {
+    table->field[MYSQL_DD_TABLES_FIELD_TABLESPACE]->set_notnull();
     store_failed= store_failed ||
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_KEYS]->
-        store(keys, TRUE);
-
-    if (create_info->tablespace)
-    {
-      tbl_tables->field[MYSQL_DD_TABLES_FIELD_TABLESPACE]->set_notnull();
-      store_failed= store_failed ||
-        tbl_tables->field[MYSQL_DD_TABLES_FIELD_TABLESPACE]->
-          store(create_info->tablespace, 
-                strlen(create_info->tablespace), system_charset_info);
-    }
-
-    if (store_failed)
-    {
-      my_error(ER_CANT_CREATE_TABLE, MYF(0), name, store_failed);
-      goto done;
-    }
-
-    // Store table definition in DD
-    if (tbl_tables->file->ha_write_row(tbl_tables->record[0]))
-    {
-      my_error(ER_TABLE_EXISTS_ERROR, MYF(0), name);
-      goto done;
-    }
-
-    // Store field definition in DD
-    tbl_fields= table_list[1].table;
-    bitmap_set_all(tbl_fields->write_set);
-    bitmap_set_all(tbl_fields->read_set);
-    new_table_id= tbl_tables->file->insert_id_for_cur_row + new_table_id;
-    dd_create_fields(thd, tbl_fields, new_table_id, create_fields);
-
-    error= false;
+      table->field[MYSQL_DD_TABLES_FIELD_TABLESPACE]->
+        store(create_info->tablespace, 
+              strlen(create_info->tablespace), system_charset_info);
   }
 
-done:
-  thd->count_cuted_fields= saved_count_cuted_fields;
-  thd->variables.sql_mode= saved_mode;
-  if (tbl_tables)
+  if (store_failed)
   {
-    tbl_tables->auto_increment_field_not_null= FALSE;
-    tbl_tables->file->ha_release_auto_increment();
-    tbl_tables->next_number_field= NULL;
+    my_error(ER_CANT_CREATE_TABLE, MYF(0), name, store_failed);
+    error= ER_CANT_CREATE_TABLE;
+    goto done;
   }
 
-  if (save_binlog_row_based)
-    thd->set_current_stmt_binlog_format_row();
-
-  thd->pop_internal_handler();
-  close_thread_tables(thd);
-  thd->mdl_context.rollback_to_savepoint(mdl_savepoint);
-  thd->lex->restore_backup_query_tables_list(&query_table_list_backup);
-  thd->restore_backup_open_tables_state(&open_tables_state_backup);
-  DBUG_RETURN(error);
-}
+  // Store table definition in DD
+  if (table->file->ha_write_row(table->record[0]))
+  {
+    my_error(ER_TABLE_EXISTS_ERROR, MYF(0), name);
+    error= ER_TABLE_EXISTS_ERROR;
+    goto done;
+  }
 
-/**
-  Write fields into mysql.fields.
+  // Store field definition in DD
+  latest_auto_incr_id= table->file->insert_id_for_cur_row;
+  error= add_fields(thd, create_fields);
 
-  @param thd  Thread context.
-  @param sp   Stored routine object to store.
+done:
+  table->auto_increment_field_not_null= FALSE;
+  table->file->ha_release_auto_increment();
+  table->next_number_field= NULL;
 
-  @retval false success
-  @retval true  error
-*/
+  return (error);
+}
 
-bool dd_create_fields(THD *thd, TABLE *table, long long id,
-                     List<Create_field> &create_fields)
+int Data_dictionary::add_fields(THD *thd, List<Create_field> &create_fields)
 {
-  bool error= true;
+  bool error= false;
   bool store_failed= FALSE;
+  TABLE *table= dd_tables[MYSQL_DD_TABLE_FIELDS].table;
 
-  DBUG_ENTER("dd_create_fields");
+  bitmap_set_all(table->write_set);
+  bitmap_set_all(table->read_set);
 
   Create_field *field;
   List_iterator<Create_field> it(create_fields);
   while ((field=it++))
   {
-    restore_record(table, s->default_values); // Get default values for fields
+    // Get default values for fields
+    restore_record(table, s->default_values);
 
     table->field[MYSQL_DD_FIELDS_FIELD_TABLE_ID]->set_notnull();
     store_failed= table->field[MYSQL_DD_FIELDS_FIELD_TABLE_ID]->
-        store(id, TRUE);
+      store(latest_auto_incr_id, TRUE);
 
     table->field[MYSQL_DD_FIELDS_FIELD_NAME]->set_notnull();
     store_failed= store_failed ||
-        table->field[MYSQL_DD_FIELDS_FIELD_NAME]->
+      table->field[MYSQL_DD_FIELDS_FIELD_NAME]->
         store(field->field_name, strlen(field->field_name),
               system_charset_info);
 
     table->field[MYSQL_DD_FIELDS_FIELD_TYPE]->set_notnull();
     store_failed= store_failed ||
-        table->field[MYSQL_DD_FIELDS_FIELD_TYPE]->
+      table->field[MYSQL_DD_FIELDS_FIELD_TYPE]->
         store(field->sql_type, TRUE);
 
     table->field[MYSQL_DD_FIELDS_FIELD_FIELD_LENGTH]->set_notnull();
     store_failed= store_failed ||
-        table->field[MYSQL_DD_FIELDS_FIELD_FIELD_LENGTH]->
+      table->field[MYSQL_DD_FIELDS_FIELD_FIELD_LENGTH]->
         store(field->length, TRUE);
 
     table->field[MYSQL_DD_FIELDS_FIELD_UNIREG_TYPE]->set_notnull();
     store_failed= store_failed ||
-        table->field[MYSQL_DD_FIELDS_FIELD_UNIREG_TYPE]->
+      table->field[MYSQL_DD_FIELDS_FIELD_UNIREG_TYPE]->
         store(field->unireg_check, TRUE);
 
     table->field[MYSQL_DD_FIELDS_FIELD_RECORD_OFFSET]->set_notnull();
     store_failed= store_failed ||
-        table->field[MYSQL_DD_FIELDS_FIELD_RECORD_OFFSET]->
+      table->field[MYSQL_DD_FIELDS_FIELD_RECORD_OFFSET]->
         store(field->offset, TRUE);
 
     table->field[MYSQL_DD_FIELDS_FIELD_PACK_FLAG]->set_notnull();
     store_failed= store_failed ||
-        table->field[MYSQL_DD_FIELDS_FIELD_PACK_FLAG]->
+      table->field[MYSQL_DD_FIELDS_FIELD_PACK_FLAG]->
         store(field->pack_flag, TRUE);
 
     table->field[MYSQL_DD_FIELDS_FIELD_CSID]->set_notnull();
     store_failed= store_failed ||
-        table->field[MYSQL_DD_FIELDS_FIELD_CSID]->
+      table->field[MYSQL_DD_FIELDS_FIELD_CSID]->
         store(field->charset->number, TRUE);
 
     table->field[MYSQL_DD_FIELDS_FIELD_COMMENT]->set_notnull();
     store_failed= store_failed ||
-        table->field[MYSQL_DD_FIELDS_FIELD_COMMENT]->
+      table->field[MYSQL_DD_FIELDS_FIELD_COMMENT]->
         store(field->comment.str, field->comment.length,
               system_charset_info);
 
@@ -546,7 +460,7 @@ bool dd_create_fields(THD *thd, TABLE *t
     {
       uchar buff[MAX_FIELD_WIDTH];
       unsigned char  sep= ',';
-      uint           i;
+      uint           i= 0;
       String tmp((char*) buff,sizeof(buff), &my_charset_bin);
       tmp.length(0);
       /*unsigned char  occ[256]; // NewDD - Why do we need this occ ??
@@ -609,18 +523,61 @@ bool dd_create_fields(THD *thd, TABLE *t
     {
         my_error(ER_CANT_CREATE_TABLE, MYF(0), field->field_name,
                  store_failed);
-        goto err;
+        error= ER_CANT_CREATE_TABLE;
+        goto done;
     }
 
     if (table->file->ha_write_row(table->record[0]))
     {
         my_error(ER_TABLE_EXISTS_ERROR, MYF(0), field->field_name);
-        goto err;
+        error= ER_TABLE_EXISTS_ERROR;
+        goto done;
     }
   }
-  error= false;
+done:
+  return (error);
+}
 
-err:
-  DBUG_RETURN(error);
+int Data_dictionary::open_tables(THD *thd)
+{
+  if (open_and_lock_tables(thd, dd_tables, FALSE,
+                           MYSQL_LOCK_IGNORE_TIMEOUT))
+  {
+    my_error(ER_CANT_CREATE_TABLE, MYF(0), 
+             "todo", 1);
+    return ER_CANT_CREATE_TABLE;
+  }
+  return 0;
+}
+
+void Data_dictionary::save_state(THD *thd)
+{
+  saved_count_cuted_fields= thd->count_cuted_fields;
+  thd->count_cuted_fields= CHECK_FIELD_WARN; // TODO Required ?
+
+  saved_mode= thd->variables.sql_mode;
+  thd->variables.sql_mode= 0; // Reset during DD operations
+
+  if ((save_binlog_row_based= thd->is_current_stmt_binlog_format_row()))
+    thd->clear_current_stmt_binlog_format_row();
+
+  thd->push_internal_handler(&error_handler);
+  thd->lex->reset_n_backup_query_tables_list(&query_table_list_backup);
+  thd->reset_n_backup_open_tables_state(&open_tables_state_backup);
 }
 
+void Data_dictionary::restore_state(THD *thd)
+{
+  thd->count_cuted_fields= saved_count_cuted_fields;
+  thd->variables.sql_mode= saved_mode;
+
+  if (save_binlog_row_based)
+    thd->set_current_stmt_binlog_format_row();
+  save_binlog_row_based= 0;
+
+  // We are ignoring errors for now should propagate this UP TODO
+  thd->pop_internal_handler();
+
+  thd->lex->restore_backup_query_tables_list(&query_table_list_backup);
+  thd->restore_backup_open_tables_state(&open_tables_state_backup);
+}

=== modified file 'sql/datadict.h'
--- a/sql/datadict.h	2012-11-12 08:00:23 +0000
+++ b/sql/datadict.h	2012-11-12 18:48:23 +0000
@@ -40,104 +40,4 @@ bool dd_check_storage_engine_flag(THD *t
                                   bool *yes_no);
 bool dd_recreate_table(THD *thd, const char *db, const char *table_name);
 
-/*
- New DD Prototype
-*/
-
-// Return codes from dd_create_*, dp_drop_*
-#define DD_OK                 0
-#define DD_KEY_NOT_FOUND     -1
-#define DD_OPEN_TABLE_FAILED -2
-#define DD_WRITE_ROW_FAILED  -3
-#define DD_DELETE_ROW_FAILED -4
-#define DD_GET_FIELD_FAILED  -5
-#define DD_PARSE_ERROR       -6
-#define DD_INTERNAL_ERROR    -7
-#define DD_NO_DB_ERROR       -8
-#define DD_BAD_IDENTIFIER    -9
-#define DD_BODY_TOO_LONG    -10
-#define DD_FLD_STORE_FAILED -11
-
-/* Fields in data dictionary table 'TABLES' */
-enum
-{
-  MYSQL_DD_TABLES_FIELD_DB = 0,
-  MYSQL_DD_TABLES_FIELD_NAME,
-  MYSQL_DD_TABLES_FIELD_ID,
-  MYSQL_DD_TABLES_FIELD_LEGACY_DB_TYPE,
-  MYSQL_DD_TABLES_FIELD_MAX_ROWS,
-  MYSQL_DD_TABLES_FIELD_MIN_ROWS,
-  MYSQL_DD_TABLES_FIELD_SYSTEM,
-  MYSQL_DD_TABLES_FIELD_DD_VERSION,
-  MYSQL_DD_TABLES_FIELD_DD_CREATE_OPTIONS,
-  MYSQL_DD_TABLES_FIELD_AVG_ROW_LENGTH,
-  MYSQL_DD_TABLES_FIELD_CSID,
-  MYSQL_DD_TABLES_FIELD_ROW_TYPE,
-  MYSQL_DD_TABLES_FIELD_STATS_SAMPLE_PAGES,
-  MYSQL_DD_TABLES_FIELD_STATS_AUTO_RECALC,
-  MYSQL_DD_TABLES_FIELD_MYSQL_VERSION,
-  MYSQL_DD_TABLES_FIELD_DEFAULT_PART_DB_TYPE,
-  MYSQL_DD_TABLES_FIELD_KEY_BLOCK_SIZE,
-  MYSQL_DD_TABLES_FIELD_COMMENT,
-  MYSQL_DD_TABLES_FIELD_FIELD_COUNT,
-  MYSQL_DD_TABLES_FIELD_NULL_FIELDS,
-  MYSQL_DD_TABLES_FIELD_KEYS,
-  MYSQL_DD_TABLES_FIELD_TABLESPACE,
-  MYSQL_DD_TABLES_FIELD_COUNT  // Total number of fields
-};
-
-/* Fields in data dictionary table 'FIELDS' */
-enum
-{
-  MYSQL_DD_FIELDS_FIELD_TABLE_ID= 0,
-  MYSQL_DD_FIELDS_FIELD_NAME,
-  MYSQL_DD_FIELDS_FIELD_TYPE,
-  MYSQL_DD_FIELDS_FIELD_FIELD_LENGTH,
-  MYSQL_DD_FIELDS_FIELD_UNIREG_TYPE,
-  MYSQL_DD_FIELDS_FIELD_RECORD_OFFSET,
-  MYSQL_DD_FIELDS_FIELD_PACK_FLAG,
-  MYSQL_DD_FIELDS_FIELD_CSID,
-  MYSQL_DD_FIELDS_FIELD_COMMENT,
-  MYSQL_DD_FIELDS_FIELD_INTERVAL,
-  MYSQL_DD_FIELDS_FIELD_INTERVAL_PARTS,
-  MYSQL_DD_FIELDS_FIELD_FLAGS,
-  MYSQL_DD_FIELDS_FIELD_COUNT  // Total number of fields
-};
-
-/* Fields in data dictionary table 'TABLE_KEYS' */
-enum
-{
-  MYSQL_DD_TABLE_KEYS_FIELD_ID= 0,
-  MYSQL_DD_TABLE_KEYS_FIELD_TABLE_ID,
-  MYSQL_DD_TABLE_KEYS_FIELD_NAME,
-  MYSQL_DD_TABLE_KEYS_FIELD_FLAGS,
-  MYSQL_DD_TABLE_KEYS_FIELD_KEY_PARTS,
-  MYSQL_DD_TABLE_KEYS_FIELD_KEY_LENGTH,
-  MYSQL_DD_TABLE_KEYS_FIELD_ALGORITHM,
-  MYSQL_DD_TABLE_KEYS_FIELD_BLOCK_SIZE,
-  MYSQL_DD_TABLE_KEYS_FIELD_USER_DEFINED_KEY_PARTS,
-  MYSQL_DD_TABLE_KEYS_FIELD_PARSER,
-  MYSQL_DD_TABLE_KEYS_FIELD_COMMENT,
-  MYSQL_DD_TABLE_KEYS_FIELD_COUNT  // Total number of fields
-};
-
-/* Fields in data dictionary table 'TABLE_KEY_PARTS' */
-enum
-{
-  MYSQL_DD_TABLE_KEY_PARTS_FIELD_KEY_ID= 0,
-  MYSQL_DD_TABLE_KEY_PARTS_FIELD_FIELD_NR,
-  MYSQL_DD_TABLE_KEY_PARTS_FIELD_LENGTH,
-  MYSQL_DD_TABLE_KEY_PARTS_FIELD_RECORD_OFFSET,
-  MYSQL_DD_TABLE_KEY_PARTS_FIELD_KEY_TYPE,
-  MYSQL_DD_TABLE_KEY_PARTS_FIELD_KEY_PART_FLAG,
-  MYSQL_DD_TABLE_KEY_PARTS_FIELD_COUNT  // Total number of fields
-};
-
-
-bool is_dd_table(const char *db, const char *name);
-bool dd_create_table(THD *thd, const char *db, const char *name,
-                     HA_CREATE_INFO *create_info,
-                     List<Create_field> &create_fields,
-                     uint keys, KEY *key_info, handler *file);
-
 #endif // DATADICT_INCLUDED

=== added file 'sql/datadict0.h'
--- a/sql/datadict0.h	1970-01-01 00:00:00 +0000
+++ b/sql/datadict0.h	2012-11-12 18:48:23 +0000
@@ -0,0 +1,194 @@
+#ifndef DATADICT0_INCLUDED
+#define DATADICT0_INCLUDED
+/* Copyright (c) 2010, Oracle and/or its affiliates. All rights reserved.
+
+   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., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
+
+#include "handler.h"
+#include "sql_priv.h"
+#include "sql_class.h"
+#include "sql_table.h"
+
+#include "unireg.h"
+#include "sql_base.h"
+#include "key.h"
+#include "lock.h"
+
+/*
+ New DD Prototype
+*/
+
+// Return codes from dd_create_*, dp_drop_*
+#define DD_OK                 0
+#define DD_KEY_NOT_FOUND     -1
+#define DD_OPEN_TABLE_FAILED -2
+#define DD_WRITE_ROW_FAILED  -3
+#define DD_DELETE_ROW_FAILED -4
+#define DD_GET_FIELD_FAILED  -5
+#define DD_PARSE_ERROR       -6
+#define DD_INTERNAL_ERROR    -7
+#define DD_NO_DB_ERROR       -8
+#define DD_BAD_IDENTIFIER    -9
+#define DD_BODY_TOO_LONG    -10
+#define DD_FLD_STORE_FAILED -11
+
+/* Data dictionary tables */
+enum enum_dd_tables
+{
+  MYSQL_DD_TABLES,
+  MYSQL_DD_TABLE_FIELDS,
+  MYSQL_DD_TABLE_KEYS,
+  MYSQL_DD_TABLE_KEY_PARTS,
+  MYSQL_DD_TABLE_COUNT
+};
+
+/* Fields in data dictionary table 'TABLES' */
+enum enum_dd_tables_fields
+{
+  MYSQL_DD_TABLES_FIELD_DB = 0,
+  MYSQL_DD_TABLES_FIELD_NAME,
+  MYSQL_DD_TABLES_FIELD_ID,
+  MYSQL_DD_TABLES_FIELD_LEGACY_DB_TYPE,
+  MYSQL_DD_TABLES_FIELD_MAX_ROWS,
+  MYSQL_DD_TABLES_FIELD_MIN_ROWS,
+  MYSQL_DD_TABLES_FIELD_SYSTEM,
+  MYSQL_DD_TABLES_FIELD_DD_VERSION,
+  MYSQL_DD_TABLES_FIELD_DD_CREATE_OPTIONS,
+  MYSQL_DD_TABLES_FIELD_AVG_ROW_LENGTH,
+  MYSQL_DD_TABLES_FIELD_CSID,
+  MYSQL_DD_TABLES_FIELD_ROW_TYPE,
+  MYSQL_DD_TABLES_FIELD_STATS_SAMPLE_PAGES,
+  MYSQL_DD_TABLES_FIELD_STATS_AUTO_RECALC,
+  MYSQL_DD_TABLES_FIELD_MYSQL_VERSION,
+  MYSQL_DD_TABLES_FIELD_DEFAULT_PART_DB_TYPE,
+  MYSQL_DD_TABLES_FIELD_KEY_BLOCK_SIZE,
+  MYSQL_DD_TABLES_FIELD_COMMENT,
+  MYSQL_DD_TABLES_FIELD_FIELD_COUNT,
+  MYSQL_DD_TABLES_FIELD_NULL_FIELDS,
+  MYSQL_DD_TABLES_FIELD_KEYS,
+  MYSQL_DD_TABLES_FIELD_TABLESPACE,
+  MYSQL_DD_TABLES_FIELD_COUNT  // Total number of fields
+};
+
+/* Fields in data dictionary table 'FIELDS' */
+enum enum_dd_table_keys_fields
+{
+  MYSQL_DD_FIELDS_FIELD_TABLE_ID= 0,
+  MYSQL_DD_FIELDS_FIELD_NAME,
+  MYSQL_DD_FIELDS_FIELD_TYPE,
+  MYSQL_DD_FIELDS_FIELD_FIELD_LENGTH,
+  MYSQL_DD_FIELDS_FIELD_UNIREG_TYPE,
+  MYSQL_DD_FIELDS_FIELD_RECORD_OFFSET,
+  MYSQL_DD_FIELDS_FIELD_PACK_FLAG,
+  MYSQL_DD_FIELDS_FIELD_CSID,
+  MYSQL_DD_FIELDS_FIELD_COMMENT,
+  MYSQL_DD_FIELDS_FIELD_INTERVAL,
+  MYSQL_DD_FIELDS_FIELD_INTERVAL_PARTS,
+  MYSQL_DD_FIELDS_FIELD_FLAGS,
+  MYSQL_DD_FIELDS_FIELD_COUNT  // Total number of fields
+};
+
+/* Fields in data dictionary table 'TABLE_KEYS' */
+enum enum_dd_table_key_parts_fields
+{
+  MYSQL_DD_TABLE_KEYS_FIELD_ID= 0,
+  MYSQL_DD_TABLE_KEYS_FIELD_TABLE_ID,
+  MYSQL_DD_TABLE_KEYS_FIELD_NAME,
+  MYSQL_DD_TABLE_KEYS_FIELD_FLAGS,
+  MYSQL_DD_TABLE_KEYS_FIELD_KEY_PARTS,
+  MYSQL_DD_TABLE_KEYS_FIELD_KEY_LENGTH,
+  MYSQL_DD_TABLE_KEYS_FIELD_ALGORITHM,
+  MYSQL_DD_TABLE_KEYS_FIELD_BLOCK_SIZE,
+  MYSQL_DD_TABLE_KEYS_FIELD_USER_DEFINED_KEY_PARTS,
+  MYSQL_DD_TABLE_KEYS_FIELD_PARSER,
+  MYSQL_DD_TABLE_KEYS_FIELD_COMMENT,
+  MYSQL_DD_TABLE_KEYS_FIELD_COUNT  // Total number of fields
+};
+
+/* Fields in data dictionary table 'TABLE_KEY_PARTS' */
+enum
+{
+  MYSQL_DD_TABLE_KEY_PARTS_FIELD_KEY_ID= 0,
+  MYSQL_DD_TABLE_KEY_PARTS_FIELD_FIELD_NR,
+  MYSQL_DD_TABLE_KEY_PARTS_FIELD_LENGTH,
+  MYSQL_DD_TABLE_KEY_PARTS_FIELD_RECORD_OFFSET,
+  MYSQL_DD_TABLE_KEY_PARTS_FIELD_KEY_TYPE,
+  MYSQL_DD_TABLE_KEY_PARTS_FIELD_KEY_PART_FLAG,
+  MYSQL_DD_TABLE_KEY_PARTS_FIELD_COUNT  // Total number of fields
+};
+
+class Data_dictionary
+{
+  // DD Tables
+  TABLE_LIST dd_tables[MYSQL_DD_TABLE_COUNT];
+
+  // State before DD operations
+  Dummy_error_handler error_handler;
+  Query_tables_list query_table_list_backup;
+  Open_tables_backup open_tables_state_backup;
+  MDL_savepoint mdl_savepoint;
+  bool save_binlog_row_based;
+  sql_mode_t saved_mode;
+  long long latest_auto_incr_id;
+  enum_check_fields saved_count_cuted_fields;
+
+public:
+  Data_dictionary()
+  {
+    latest_auto_incr_id= -1;
+    dd_tables[MYSQL_DD_TABLES].init_one_table("mysql", 5, "tables", 6, "tables", TL_WRITE);
+    dd_tables[MYSQL_DD_TABLE_FIELDS].init_one_table("mysql", 5, "fields", 6, "fields", TL_WRITE);
+    dd_tables[MYSQL_DD_TABLE_KEYS].init_one_table("mysql", 5, "table_keys", 10, "table_keys", TL_WRITE);
+    dd_tables[MYSQL_DD_TABLE_KEY_PARTS].init_one_table("mysql", 5, "table_key_parts", 15, "table_key_parts", TL_WRITE);
+    dd_tables[MYSQL_DD_TABLES].next_local= dd_tables[MYSQL_DD_TABLES].next_global= dd_tables + 1;
+    dd_tables[MYSQL_DD_TABLE_FIELDS].next_local= dd_tables[MYSQL_DD_TABLE_FIELDS].next_global= dd_tables + 2;
+    dd_tables[MYSQL_DD_TABLE_KEYS].next_local= dd_tables[MYSQL_DD_TABLE_KEYS].next_global= dd_tables + 3;
+  }
+
+  // Check if db.name is one of DD table names
+  static bool is_dd_table(const char *db, const char *name);
+
+  void save_state(THD *thd);
+
+  void restore_state(THD *thd);
+
+  int open_tables(THD *thd);
+
+  void close_tables(THD *thd)
+  {
+    close_thread_tables(thd);
+  }
+
+  int add_table(THD *thd, const char *db, const char *name,
+                HA_CREATE_INFO *create_info,
+                List<Create_field> &create_fields, uint keys);
+
+  int add_fields(THD *thd, List<Create_field> &create_fields);
+
+  /*
+  add_key();
+  add_key_part();
+
+  update_table();
+  update_field();
+  update_key();
+  update_key_part();
+
+  delete_table();
+  delete_field();
+  delete_key();
+  delete_key_part(); */
+};
+
+#endif // DATADICT0_INCLUDED

=== modified file 'sql/sql_table.cc'
--- a/sql/sql_table.cc	2012-11-12 08:00:23 +0000
+++ b/sql/sql_table.cc	2012-11-12 18:48:23 +0000
@@ -56,6 +56,7 @@
 #include "sql_resolver.h"              // setup_order, fix_inner_refs
 #include "table_cache.h"
 #include <mysql/psi/mysql_table.h>
+#include "datadict0.h"
 
 #ifdef __WIN__
 #include <io.h>
@@ -4749,12 +4750,24 @@ bool create_table_impl(THD *thd,
 
   // Add table details into new DD
   if (!(create_info->options & HA_LEX_CREATE_TMP_TABLE) &&
-      !thd->is_current_stmt_binlog_format_row() && //TODO We get core without
-                                                    // this condition for now
-      !is_dd_table(db, table_name))
-    dd_create_table(thd, db, table_name, create_info,
-                    alter_info->create_list, *key_count,
-                    *key_info, file);
+      !Data_dictionary::is_dd_table(db, table_name))
+  {
+    int error= 0;
+    Data_dictionary datadict;
+
+    datadict.save_state(thd);
+    datadict.open_tables(thd);
+
+    error= datadict.add_table(thd, db, table_name, create_info,
+                       alter_info->create_list, *key_count);
+    error= 0; // Intentionally ignoring errors for now
+
+    datadict.close_tables(thd);
+    datadict.restore_state(thd);
+
+    if (error)
+      goto err;
+  }
 
   if (!no_ha_table && create_info->options & HA_LEX_CREATE_TMP_TABLE)
   {

No bundle (reason: useless for push emails).
Thread
bzr push into mysql-trunk branch (gopal.shankar:4969 to 4970) Gopal Shankar13 Nov