Implementation of MDEV-5491: CREATE OR REPLACE TABLE

Using CREATE OR REPLACE TABLE is be identical to

DROP TABLE IF EXISTS table_name;
CREATE TABLE ...;

Except that:

* CREATE OR REPLACE is be atomic (now one can create the same table between drop and create).
* Temporary tables will not shadow the table name for the DROP as the CREATE TABLE tells us already if we are using a temporary table or not.
* If the table was locked with LOCK TABLES, the new table will be locked with the same lock after it's created.

Implementation details:
- We don't anymore open the to-be-created table during CREATE TABLE, which the original code did.
  - There is no need to open a table we are planning to create. It's enough to check if the table exists or not.
- Removed some of duplicated code for CREATE IF NOT EXISTS.
- Give an error when using CREATE OR REPLACE with IF NOT EXISTS (conflicting options).
- As a side effect of the code changes, we don't anymore have to internally re-prepare prepared statements with CREATE TABLE if the table exists.
- Made one code path for all testing if log table are in use.
- Better error message if one tries to create/drop/alter a log table in use
- Added back disabled rpl_row_create_table test as it now seams to work and includes a lot of interesting tests.
- Added HA_LEX_CREATE_REPLACE to mark if we are using CREATE OR REPLACE
- Aligned CREATE OR REPLACE parsing code in sql_yacc.yy for TABLE and VIEW
- Changed interface for drop_temporary_table() to make it more reusable
- Changed Locked_tables_list::init_locked_tables() to work on the table object instead of the table list object. Before this it used a mix of both, which was not good.
- Locked_tables_list::unlock_locked_tables(THD *thd) now requires a valid thd argument. Old usage of calling this with 0 i changed to instead call Locked_tables_list::reset()
- Added functions Locked_tables_list:restore_lock() and Locked_tables_list::add_back_last_deleted_lock() to be able to easily add back a locked table to the lock list.
- Added restart_trans_for_tables() to be able to restart a transaction.
- DROP_ACL is required if one uses CREATE TABLE OR REPLACE.
- Added drop of normal and temporary tables in create_table_imp() if CREATE OR REPLACE was used.
- Added reacquiring of table locks in mysql_create_table() and mysql_create_like_table()




mysql-test/include/commit.inc:
  With new code we get fewer status increments
mysql-test/r/commit_1innodb.result:
  With new code we get fewer status increments
mysql-test/r/create.result:
  Added testing of create or replace with timeout
mysql-test/r/create_or_replace.result:
  Basic testing of CREATE OR REPLACE TABLE
mysql-test/r/partition_exchange.result:
  New error message
mysql-test/r/ps_ddl.result:
  Fewer reprepares with new code
mysql-test/suite/archive/discover.result:
  Don't rediscover archive tables if the .frm file exists
  (Sergei will look at this if there is a better way...)
mysql-test/suite/archive/discover.test:
  Don't rediscover archive tables if the .frm file exists
  (Sergei will look at this if there is a better way...)
mysql-test/suite/funcs_1/r/innodb_views.result:
  New error message
mysql-test/suite/funcs_1/r/memory_views.result:
  New error message
mysql-test/suite/rpl/disabled.def:
  rpl_row_create_table should now be safe to use
mysql-test/suite/rpl/r/rpl_row_create_table.result:
  Updated results after adding back disabled test
mysql-test/suite/rpl/t/rpl_create_if_not_exists.test:
  Added comment
mysql-test/suite/rpl/t/rpl_row_create_table.test:
  Added CREATE OR REPLACE TABLE test
mysql-test/t/create.test:
  Added CREATE OR REPLACE TABLE test
mysql-test/t/create_or_replace-master.opt:
  Create logs
mysql-test/t/create_or_replace.test:
  Basic testing of CREATE OR REPLACE TABLE
mysql-test/t/partition_exchange.test:
  Error number changed as we are now using same code for all log table change issues
mysql-test/t/ps_ddl.test:
  Fewer reprepares with new code
sql/handler.h:
  Moved things around a bit in a structure to get better alignment.
  Added HA_LEX_CREATE_REPLACE to mark if we are using CREATE OR REPLACE
  Added 3 elements to end of HA_CREATE_INFO to be able to store state to add backs locks in case of LOCK TABLES.
sql/log.cc:
  Reimplemented check_if_log_table():
  - Simpler and faster usage
  - Can give error messages
  
  This gives us one code path for allmost all error messages if log tables are in use
sql/log.h:
  New interface for check_if_log_table()
sql/slave.cc:
  More logging
sql/sql_alter.cc:
  New interface for check_if_log_table()
sql/sql_base.cc:
  More documentation
  Changed interface for drop_temporary_table() to make it more reusable
  Changed Locked_tables_list::init_locked_tables() to work on the table object instead of the table list object. Before this it used a mix of both, which was not good.
  Locked_tables_list::unlock_locked_tables(THD *thd) now requires a valid thd argument.  Old usage of calling this with 0 i changed to instead call Locked_tables_list::reset()
  Added functions Locked_tables_list:restore_lock() and Locked_tables_list::add_back_last_deleted_lock() to be able to easily add back a locked table to the lock list.
  Check for command number instead of open_strategy of CREATE TABLE was used.
  Added restart_trans_for_tables() to be able to restart a transaction.  This was needed in "create or replace ... select" between the drop table and the select.
sql/sql_base.h:
  Added and updated function prototypes
sql/sql_class.h:
  Added new prototypes to Locked_tables_list class
  Added extra argument to select_create to avoid double call to eof() or send_error()
  - I needed this in some edge case where the table was not created against expections.
sql/sql_db.cc:
  New interface for check_if_log_table()
sql/sql_insert.cc:
  Remember position to lock information so that we can reaquire table lock for LOCK TABLES + CREATE OR REPLACE TABLE SELECT. Later add back the lock by calling restore_lock().
  Removed one not needed indentation level in create_table_from_items()
  Ensure we don't call send_eof() or abort_result_set() twice.
sql/sql_lex.h:
  Removed variable that I temporarly added in an earlier changeset
sql/sql_parse.cc:
  Removed old test code (marked with QQ)
  Ensure that we have open_strategy set as TABLE_LIST::OPEN_STUB in CREATE TABLE
  Removed some IF NOT EXISTS code as this is now handled in create_table_table_impl().
  Set OPTION_KEEP_LOGS later. This code had to be moved as the test for IF EXISTS has changed place.
  DROP_ACL is required if one uses CREATE TABLE OR REPLACE.
sql/sql_partition_admin.cc:
  New interface for check_if_log_table()
sql/sql_rename.cc:
  New interface for check_if_log_table()
sql/sql_table.cc:
  New interface for check_if_log_table()
  Moved some code in mysql_rm_table() under a common test.
  - Safe as temporary tables doesn't have statistics.
  - !is_temporary_table(table) test was moved out from drop_temporary_table() and merged with upper level code.
  - Added drop of normal and temporary tables in create_table_imp() if CREATE OR REPLACE was used.
  - Added reacquiring of table locks in mysql_create_table() and mysql_create_like_table()
  - In mysql_create_like_table(), restore table->open_strategy() if it was changed.
  - Re-test if table was a view after opening it.
sql/sql_table.h:
  New prototype for mysql_create_table_no_lock()
sql/sql_yacc.yy:
  Added syntax for CREATE OR REPLACE TABLE
  Reuse new code for CREATE OR REPLACE VIEW
sql/table.h:
  Added name for enum type
sql/table_cache.cc:
  More DBUG

1 files changed, 163 insertions, 51 deletions

diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 45d9c5dc091..32e4fdc18a1 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -2031,33 +2031,29 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
   bool error;
   Drop_table_error_handler err_handler;
   TABLE_LIST *table;
-
   DBUG_ENTER("mysql_rm_table");
 
   /* Disable drop of enabled log tables, must be done before name locking */
   for (table= tables; table; table= table->next_local)
   {
-    if (check_if_log_table(table->db_length, table->db,
-                           table->table_name_length, table->table_name, true))
-    {
-      my_error(ER_BAD_LOG_STATEMENT, MYF(0), "DROP");
+    if (check_if_log_table(table, TRUE, "DROP"))
       DBUG_RETURN(true);
-    }
   }
 
-  if (!in_bootstrap)
+  if (!drop_temporary)
   {
-    for (table= tables; table; table= table->next_local)
+    if (!in_bootstrap)
     {
-      LEX_STRING db_name= { table->db, table->db_length };
-      LEX_STRING table_name= { table->table_name, table->table_name_length };
-      if (table->open_type == OT_BASE_ONLY || !find_temporary_table(thd, table))
-        (void) delete_statistics_for_table(thd, &db_name, &table_name);
+      for (table= tables; table; table= table->next_local)
+      {
+        LEX_STRING db_name= { table->db, table->db_length };
+        LEX_STRING table_name= { table->table_name, table->table_name_length };
+        if (table->open_type == OT_BASE_ONLY ||
+            !find_temporary_table(thd, table))
+          (void) delete_statistics_for_table(thd, &db_name, &table_name);
+      }
     }
-  }
 
-  if (!drop_temporary)
-  {
     if (!thd->locked_tables_mode)
     {
       if (lock_table_names(thd, tables, NULL,
@@ -2286,7 +2282,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
 
   for (table= tables; table; table= table->next_local)
   {
-    bool is_trans;
+    bool is_trans= 0;
     char *db=table->db;
     size_t db_length= table->db_length;
     handlerton *table_type= 0;
@@ -2311,12 +2307,16 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       .  1 - a temporary table was not found.
       . -1 - a temporary table is used by an outer statement.
     */
-    if (table->open_type == OT_BASE_ONLY)
+    if (table->open_type == OT_BASE_ONLY || !is_temporary_table(table))
       error= 1;
-    else if ((error= drop_temporary_table(thd, table, &is_trans)) == -1)
+    else
     {
-      DBUG_ASSERT(thd->in_sub_stmt);
-      goto err;
+      if ((error= drop_temporary_table(thd, table->table, &is_trans)) == -1)
+      {
+        DBUG_ASSERT(thd->in_sub_stmt);
+        goto err;
+      }
+      table->table= 0;
     }
 
     if ((drop_temporary && if_exists) || !error)
@@ -4517,12 +4517,13 @@ err:
   way to ensure that concurrent operations won't intervene.
   mysql_create_table() is a wrapper that can be used for this.
 
-  @retval false OK
-  @retval true  error
+  @retval 0 OK
+  @retval 1 error
+  @retval -1 table existed but IF EXISTS was used
 */
 
 static
-bool create_table_impl(THD *thd,
+int create_table_impl(THD *thd,
                        const char *db, const char *table_name,
                        const char *path,
                        HA_CREATE_INFO *create_info,
@@ -4535,7 +4536,7 @@ bool create_table_impl(THD *thd,
 {
   const char	*alias;
   handler	*file= 0;
-  bool		error= TRUE;
+  int		error= 1;
   bool          frm_only= create_table_mode == C_ALTER_TABLE_FRM_ONLY;
   bool          internal_tmp_table= create_table_mode == C_ALTER_TABLE || frm_only;
   DBUG_ENTER("mysql_create_table_no_lock");
@@ -4565,22 +4566,66 @@ bool create_table_impl(THD *thd,
   /* Check if table exists */
   if (create_info->tmp_table())
   {
-    if (find_temporary_table(thd, db, table_name))
+    TABLE *tmp_table;
+    if ((tmp_table= find_temporary_table(thd, db, table_name)))
     {
-      if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
+      if (create_info->options & HA_LEX_CREATE_REPLACE)
+      {
+        bool is_trans;
+        /*
+          We are using CREATE OR REPLACE on an existing temporary table
+          Remove the old table so that we can re-create it.
+        */
+        if (drop_temporary_table(thd, tmp_table, &is_trans))
+          goto err;
+      }
+      else if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
         goto warn;
-      my_error(ER_TABLE_EXISTS_ERROR, MYF(0), alias);
-      goto err;
+      else
+      {
+        my_error(ER_TABLE_EXISTS_ERROR, MYF(0), alias);
+        goto err;
+      }
     }
   }
-  else  
+  else
   {
     if (!internal_tmp_table && ha_table_exists(thd, db, table_name))
     {
-      if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
+      if (create_info->options & HA_LEX_CREATE_REPLACE)
+      {
+        TABLE_LIST table_list;
+        table_list.init_one_table(db, strlen(db), table_name,
+                                  strlen(table_name), table_name,
+                                  TL_WRITE_ALLOW_WRITE);
+        table_list.table= create_info->table;
+
+        if (check_if_log_table(&table_list, TRUE, "CREATE OR REPLACE"))
+          goto err;
+        
+        /*
+          Rollback the empty transaction started in mysql_create_table()
+          call to open_and_lock_tables() when we are using LOCK TABLES.
+        */
+        (void) trans_rollback_stmt(thd);
+        /* Remove normal table without logging */
+        if (mysql_rm_table_no_locks(thd, &table_list, 0, 0, 0, 1))
+          goto err;
+        /*
+          The test of query_tables is to ensure we have any tables in the
+          select part
+        */
+        if (thd->lex->query_tables &&
+            restart_trans_for_tables(thd, thd->lex->query_tables->next_global))
+          goto err;
+      }
+      else if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
         goto warn;
-      my_error(ER_TABLE_EXISTS_ERROR, MYF(0), table_name);
-      goto err;
+      else
+      {
+        my_error(ER_TABLE_EXISTS_ERROR, MYF(0), table_name);
+        goto err;
+      }
     }
   }
 
@@ -4702,14 +4747,14 @@ bool create_table_impl(THD *thd,
   }
 #endif
 
-  error= FALSE;
+  error= 0;
 err:
   THD_STAGE_INFO(thd, stage_after_create);
   delete file;
   DBUG_RETURN(error);
 
 warn:
-  error= FALSE;
+  error= -1;
   push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE,
                       ER_TABLE_EXISTS_ERROR, ER(ER_TABLE_EXISTS_ERROR),
                       alias);
@@ -4720,7 +4765,8 @@ warn:
   Simple wrapper around create_table_impl() to be used
   in various version of CREATE TABLE statement.
 */
-bool mysql_create_table_no_lock(THD *thd,
+
+int mysql_create_table_no_lock(THD *thd,
                                 const char *db, const char *table_name,
                                 HA_CREATE_INFO *create_info,
                                 Alter_info *alter_info, bool *is_trans,
@@ -4728,6 +4774,7 @@ bool mysql_create_table_no_lock(THD *thd,
 {
   KEY *not_used_1;
   uint not_used_2;
+  int res;
   char path[FN_REFLEN + 1];
   LEX_CUSTRING frm= {0,0};
 
@@ -4747,9 +4794,9 @@ bool mysql_create_table_no_lock(THD *thd,
     }
   }
 
-  bool res= create_table_impl(thd, db, table_name, path, create_info,
-                              alter_info, create_table_mode, is_trans,
-                              &not_used_1, &not_used_2, &frm);
+  res= create_table_impl(thd, db, table_name, path, create_info,
+                         alter_info, create_table_mode, is_trans,
+                         &not_used_1, &not_used_2, &frm);
   my_free(const_cast<uchar*>(frm.str));
   return res;
 }
@@ -4772,15 +4819,21 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
   const char *table_name= create_table->table_name;
   bool is_trans= FALSE;
   int create_table_mode;
+  TABLE_LIST *pos_in_locked_tables= 0;
   DBUG_ENTER("mysql_create_table");
 
+  DBUG_ASSERT(create_table == thd->lex->query_tables);
+
   /* Open or obtain an exclusive metadata lock on table being created  */
   if (open_and_lock_tables(thd, thd->lex->query_tables, FALSE, 0))
   {
     /* is_error() may be 0 if table existed and we generated a warning */
     DBUG_RETURN(thd->is_error());
   }
-
+  /* The following is needed only in case of lock tables */
+  if ((create_info->table= thd->lex->query_tables->table))
+    pos_in_locked_tables= create_info->table->pos_in_locked_tables;
+  
   /* Got lock. */
   DEBUG_SYNC(thd, "locked_table_name");
 
@@ -4791,9 +4844,25 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
 
   promote_first_timestamp_column(&alter_info->create_list);
   if (mysql_create_table_no_lock(thd, db, table_name, create_info, alter_info,
-                                 &is_trans, create_table_mode))
+                                 &is_trans, create_table_mode) > 0)
     DBUG_RETURN(1);
 
+  /*
+    Check if we are doing CREATE OR REPLACE TABLE under LOCK TABLES
+    on a non temporary table
+  */
+  if (thd->locked_tables_mode && pos_in_locked_tables &&
+      (create_info->options & HA_LEX_CREATE_REPLACE))
+  {
+    /*
+      Add back the deleted table and re-created table as a locked table
+      This should always work as we have a meta lock on the table.
+     */
+    thd->locked_tables_list.add_back_last_deleted_lock(pos_in_locked_tables);
+    if (thd->locked_tables_list.reopen_tables(thd))
+      thd->locked_tables_list.unlink_all_closed_tables(thd, NULL, 0);
+  }
+
   /* In RBR we don't need to log CREATE TEMPORARY TABLE */
   if (thd->is_current_stmt_binlog_format_row() && create_info->tmp_table())
     DBUG_RETURN(0);
@@ -4986,10 +5055,12 @@ mysql_rename_table(handlerton *base, const char *old_db,
     TRUE  error
 */
 
-bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
+bool mysql_create_like_table(THD* thd, TABLE_LIST* table,
+                             TABLE_LIST* src_table,
                              HA_CREATE_INFO *create_info)
 {
   HA_CREATE_INFO local_create_info;
+  TABLE_LIST *pos_in_locked_tables= 0;
   Alter_info local_alter_info;
   Alter_table_ctx local_alter_ctx; // Not used
   bool res= TRUE;
@@ -4997,7 +5068,6 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
   uint not_used;
   DBUG_ENTER("mysql_create_like_table");
 
-
   /*
     We the open source table to get its description in HA_CREATE_INFO
     and Alter_info objects. This also acquires a shared metadata lock
@@ -5041,7 +5111,9 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
   if (src_table->schema_table)
     local_create_info.max_rows= 0;
   /* Set IF NOT EXISTS option as in the CREATE TABLE LIKE statement. */
-  local_create_info.options|= create_info->options&HA_LEX_CREATE_IF_NOT_EXISTS;
+  local_create_info.options|= (create_info->options &
+                               (HA_LEX_CREATE_IF_NOT_EXISTS | 
+                                HA_LEX_CREATE_REPLACE));
   /* Replace type of source table with one specified in the statement. */
   local_create_info.options&= ~HA_LEX_CREATE_TMP_TABLE;
   local_create_info.options|= create_info->tmp_table();
@@ -5053,12 +5125,40 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
   */
   local_create_info.data_file_name= local_create_info.index_file_name= NULL;
 
-  if ((res= mysql_create_table_no_lock(thd, table->db, table->table_name,
-                                       &local_create_info, &local_alter_info,
-                                       &is_trans, C_ORDINARY_CREATE)))
+  /* The following is needed only in case of lock tables */
+  if ((local_create_info.table= thd->lex->query_tables->table))
+    pos_in_locked_tables= local_create_info.table->pos_in_locked_tables;    
+
+  if ((res= (mysql_create_table_no_lock(thd, table->db, table->table_name,
+                                        &local_create_info, &local_alter_info,
+                                        &is_trans, C_ORDINARY_CREATE) > 0)))
     goto err;
 
   /*
+    Check if we are doing CREATE OR REPLACE TABLE under LOCK TABLES
+    on a non temporary table
+  */
+  if (thd->locked_tables_mode && pos_in_locked_tables &&
+      (create_info->options & HA_LEX_CREATE_REPLACE))
+  {
+    /*
+      Add back the deleted table and re-created table as a locked table
+      This should always work as we have a meta lock on the table.
+     */
+    thd->locked_tables_list.add_back_last_deleted_lock(pos_in_locked_tables);
+    if (thd->locked_tables_list.reopen_tables(thd))
+      thd->locked_tables_list.unlink_all_closed_tables(thd, NULL, 0);
+    else
+    {
+      /*
+        Get pointer to the newly opened table. We need this to ensure we
+        don't reopen the table when doing statment logging below.
+      */
+      table->table= pos_in_locked_tables->table;
+    }
+  }
+
+  /*
     Ensure that we have an exclusive lock on target table if we are creating
     non-temporary table.
   */
@@ -5108,6 +5208,11 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
         {
           if (!table->table)
           {
+            TABLE_LIST::enum_open_strategy save_open_strategy;
+            int open_res;
+            /* Force the newly created table to be opened */
+            save_open_strategy= table->open_strategy;
+            table->open_strategy= TABLE_LIST::OPEN_NORMAL;
 
             /*
               In order for store_create_info() to work we need to open
@@ -5117,11 +5222,20 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
               lock on this table. The table will be closed by
               close_thread_table() at the end of this branch.
             */
-            if (open_table(thd, table, thd->mem_root, &ot_ctx))
+            open_res= open_table(thd, table, thd->mem_root, &ot_ctx);
+            /* Restore */
+            table->open_strategy= save_open_strategy;
+            if (open_res)
               goto err;
             new_table= TRUE;
           }
-
+        }
+        /*
+          We have to re-test if the table was a view as the view may not
+          have been opened until just above.
+        */
+        if (!table->view)
+        {
           int result __attribute__((unused))=
             store_create_info(thd, table, &query,
                               create_info, FALSE /* show_database */);
@@ -7726,9 +7840,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
     it is the case.
     TODO: this design is obsolete and will be removed.
   */
-  int table_kind= check_if_log_table(table_list->db_length, table_list->db,
-                                     table_list->table_name_length,
-                                     table_list->table_name, false);
+  int table_kind= check_if_log_table(table_list, FALSE, NullS);
 
   if (table_kind)
   {