1 files changed, 233 insertions, 54 deletions

diff --git a/sql/sql_alter.cc b/sql/sql_alter.cc
index 8716dde5727..59498d71bb1 100644
--- a/sql/sql_alter.cc
+++ b/sql/sql_alter.cc
@@ -1,5 +1,5 @@
 /* Copyright (c) 2010, Oracle and/or its affiliates. All rights reserved.
-   Copyright (c) 2016, MariaDB Corporation
+   Copyright (c) 2016, 2020, MariaDB Corporation
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -14,6 +14,7 @@
    along with this program; if not, write to the Free Software
    Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1335  USA */
 
+#include "mariadb.h"
 #include "sql_parse.h"                       // check_access
 #include "sql_table.h"                       // mysql_alter_table,
                                              // mysql_exchange_partition
@@ -26,7 +27,7 @@ Alter_info::Alter_info(const Alter_info &rhs, MEM_ROOT *mem_root)
   key_list(rhs.key_list, mem_root),
   create_list(rhs.create_list, mem_root),
   check_constraint_list(rhs.check_constraint_list, mem_root),
-  flags(rhs.flags),
+  flags(rhs.flags), partition_flags(rhs.partition_flags),
   keys_onoff(rhs.keys_onoff),
   partition_names(rhs.partition_names, mem_root),
   num_parts(rhs.num_parts),
@@ -50,61 +51,233 @@ Alter_info::Alter_info(const Alter_info &rhs, MEM_ROOT *mem_root)
 }
 
 
-bool Alter_info::set_requested_algorithm(const LEX_STRING *str)
+bool Alter_info::set_requested_algorithm(const LEX_CSTRING *str)
 {
   // To avoid adding new keywords to the grammar, we match strings here.
-  if (!my_strcasecmp(system_charset_info, str->str, "INPLACE"))
+  if (lex_string_eq(str, STRING_WITH_LEN("INPLACE")))
     requested_algorithm= ALTER_TABLE_ALGORITHM_INPLACE;
-  else if (!my_strcasecmp(system_charset_info, str->str, "COPY"))
+  else if (lex_string_eq(str, STRING_WITH_LEN("COPY")))
     requested_algorithm= ALTER_TABLE_ALGORITHM_COPY;
-  else if (!my_strcasecmp(system_charset_info, str->str, "DEFAULT"))
+  else if (lex_string_eq(str, STRING_WITH_LEN("DEFAULT")))
     requested_algorithm= ALTER_TABLE_ALGORITHM_DEFAULT;
+  else if (lex_string_eq(str, STRING_WITH_LEN("NOCOPY")))
+    requested_algorithm= ALTER_TABLE_ALGORITHM_NOCOPY;
+  else if (lex_string_eq(str, STRING_WITH_LEN("INSTANT")))
+    requested_algorithm= ALTER_TABLE_ALGORITHM_INSTANT;
   else
     return true;
   return false;
 }
 
+void Alter_info::set_requested_algorithm(enum_alter_table_algorithm algo_val)
+{
+  requested_algorithm= algo_val;
+}
 
-bool Alter_info::set_requested_lock(const LEX_STRING *str)
+bool Alter_info::set_requested_lock(const LEX_CSTRING *str)
 {
   // To avoid adding new keywords to the grammar, we match strings here.
-  if (!my_strcasecmp(system_charset_info, str->str, "NONE"))
+  if (lex_string_eq(str, STRING_WITH_LEN("NONE")))
     requested_lock= ALTER_TABLE_LOCK_NONE;
-  else if (!my_strcasecmp(system_charset_info, str->str, "SHARED"))
+  else if (lex_string_eq(str, STRING_WITH_LEN("SHARED")))
     requested_lock= ALTER_TABLE_LOCK_SHARED;
-  else if (!my_strcasecmp(system_charset_info, str->str, "EXCLUSIVE"))
+  else if (lex_string_eq(str, STRING_WITH_LEN("EXCLUSIVE")))
     requested_lock= ALTER_TABLE_LOCK_EXCLUSIVE;
-  else if (!my_strcasecmp(system_charset_info, str->str, "DEFAULT"))
+  else if (lex_string_eq(str, STRING_WITH_LEN("DEFAULT")))
     requested_lock= ALTER_TABLE_LOCK_DEFAULT;
   else
     return true;
   return false;
 }
 
+const char* Alter_info::algorithm_clause(THD *thd) const
+{
+  switch (algorithm(thd)) {
+  case ALTER_TABLE_ALGORITHM_INPLACE:
+    return "ALGORITHM=INPLACE";
+  case ALTER_TABLE_ALGORITHM_COPY:
+    return "ALGORITHM=COPY";
+  case ALTER_TABLE_ALGORITHM_NONE:
+    DBUG_ASSERT(0);
+    /* Fall through */
+  case ALTER_TABLE_ALGORITHM_DEFAULT:
+    return "ALGORITHM=DEFAULT";
+  case ALTER_TABLE_ALGORITHM_NOCOPY:
+    return "ALGORITHM=NOCOPY";
+  case ALTER_TABLE_ALGORITHM_INSTANT:
+    return "ALGORITHM=INSTANT";
+  }
+
+  return NULL; /* purecov: begin deadcode */
+}
+
+const char* Alter_info::lock() const
+{
+  switch (requested_lock) {
+  case ALTER_TABLE_LOCK_SHARED:
+    return "LOCK=SHARED";
+  case ALTER_TABLE_LOCK_NONE:
+    return "LOCK=NONE";
+  case ALTER_TABLE_LOCK_DEFAULT:
+    return "LOCK=DEFAULT";
+  case ALTER_TABLE_LOCK_EXCLUSIVE:
+    return "LOCK=EXCLUSIVE";
+  }
+  return NULL; /* purecov: begin deadcode */
+}
+
+
+bool Alter_info::supports_algorithm(THD *thd,
+                                    const Alter_inplace_info *ha_alter_info)
+{
+  switch (ha_alter_info->inplace_supported) {
+  case HA_ALTER_INPLACE_EXCLUSIVE_LOCK:
+  case HA_ALTER_INPLACE_SHARED_LOCK:
+  case HA_ALTER_INPLACE_NO_LOCK:
+  case HA_ALTER_INPLACE_INSTANT:
+     return false;
+  case HA_ALTER_INPLACE_COPY_NO_LOCK:
+  case HA_ALTER_INPLACE_COPY_LOCK:
+    if (algorithm(thd) >= Alter_info::ALTER_TABLE_ALGORITHM_NOCOPY)
+    {
+      ha_alter_info->report_unsupported_error(algorithm_clause(thd),
+                                              "ALGORITHM=INPLACE");
+      return true;
+    }
+    return false;
+  case HA_ALTER_INPLACE_NOCOPY_NO_LOCK:
+  case HA_ALTER_INPLACE_NOCOPY_LOCK:
+    if (algorithm(thd) == Alter_info::ALTER_TABLE_ALGORITHM_INSTANT)
+    {
+      ha_alter_info->report_unsupported_error("ALGORITHM=INSTANT",
+                                              "ALGORITHM=NOCOPY");
+      return true;
+    }
+    return false;
+  case HA_ALTER_INPLACE_NOT_SUPPORTED:
+    if (algorithm(thd) >= Alter_info::ALTER_TABLE_ALGORITHM_INPLACE)
+    {
+      ha_alter_info->report_unsupported_error(algorithm_clause(thd),
+					      "ALGORITHM=COPY");
+      return true;
+    }
+    return false;
+  case HA_ALTER_ERROR:
+    return true;
+  }
+  /* purecov: begin deadcode */
+  DBUG_ASSERT(0);
+  return false;
+}
+
+
+bool Alter_info::supports_lock(THD *thd,
+                               const Alter_inplace_info *ha_alter_info)
+{
+  switch (ha_alter_info->inplace_supported) {
+  case HA_ALTER_INPLACE_EXCLUSIVE_LOCK:
+    // If SHARED lock and no particular algorithm was requested, use COPY.
+    if (requested_lock == Alter_info::ALTER_TABLE_LOCK_SHARED &&
+        algorithm(thd) == Alter_info::ALTER_TABLE_ALGORITHM_DEFAULT &&
+        thd->variables.alter_algorithm ==
+                Alter_info::ALTER_TABLE_ALGORITHM_DEFAULT)
+         return false;
+
+    if (requested_lock == Alter_info::ALTER_TABLE_LOCK_SHARED ||
+        requested_lock == Alter_info::ALTER_TABLE_LOCK_NONE)
+    {
+      ha_alter_info->report_unsupported_error(lock(), "LOCK=EXCLUSIVE");
+      return true;
+    }
+    return false;
+  case HA_ALTER_INPLACE_NO_LOCK:
+  case HA_ALTER_INPLACE_INSTANT:
+  case HA_ALTER_INPLACE_COPY_NO_LOCK:
+  case HA_ALTER_INPLACE_NOCOPY_NO_LOCK:
+    return false;
+  case HA_ALTER_INPLACE_COPY_LOCK:
+  case HA_ALTER_INPLACE_NOCOPY_LOCK:
+  case HA_ALTER_INPLACE_NOT_SUPPORTED:
+  case HA_ALTER_INPLACE_SHARED_LOCK:
+    if (requested_lock == Alter_info::ALTER_TABLE_LOCK_NONE)
+    {
+      ha_alter_info->report_unsupported_error("LOCK=NONE", "LOCK=SHARED");
+      return true;
+    }
+    return false;
+  case HA_ALTER_ERROR:
+    return true;
+  }
+  /* purecov: begin deadcode */
+  DBUG_ASSERT(0);
+  return false;
+}
+
+bool Alter_info::vers_prohibited(THD *thd) const
+{
+  if (thd->slave_thread ||
+      thd->variables.vers_alter_history != VERS_ALTER_HISTORY_ERROR)
+  {
+    return false;
+  }
+  if (flags & (
+    ALTER_PARSER_ADD_COLUMN |
+    ALTER_PARSER_DROP_COLUMN |
+    ALTER_CHANGE_COLUMN |
+    ALTER_COLUMN_ORDER))
+  {
+    return true;
+  }
+  if (flags & ALTER_ADD_INDEX)
+  {
+    List_iterator_fast<Key> key_it(const_cast<List<Key> &>(key_list));
+    Key *key;
+    while ((key= key_it++))
+    {
+      if (key->type == Key::PRIMARY || key->type == Key::UNIQUE)
+        return true;
+    }
+  }
+  return false;
+}
+
+Alter_info::enum_alter_table_algorithm
+Alter_info::algorithm(const THD *thd) const
+{
+  if (requested_algorithm == ALTER_TABLE_ALGORITHM_NONE)
+   return (Alter_info::enum_alter_table_algorithm) thd->variables.alter_algorithm;
+  return requested_algorithm;
+}
+
 
 Alter_table_ctx::Alter_table_ctx()
   : datetime_field(NULL), error_if_not_empty(false),
     tables_opened(0),
-    db(NULL), table_name(NULL), alias(NULL),
-    new_db(NULL), new_name(NULL), new_alias(NULL),
+    db(null_clex_str), table_name(null_clex_str), alias(null_clex_str),
+    new_db(null_clex_str), new_name(null_clex_str), new_alias(null_clex_str),
     fk_error_if_delete_row(false), fk_error_id(NULL),
-    fk_error_table(NULL)
-#ifndef DBUG_OFF
+    fk_error_table(NULL), modified_primary_key(false)
+#ifdef DBUG_ASSERT_EXISTS
     , tmp_table(false)
 #endif
 {
 }
 
+/*
+  TODO: new_name_arg changes if lower case table names.
+  Should be copied or converted before call
+*/
 
 Alter_table_ctx::Alter_table_ctx(THD *thd, TABLE_LIST *table_list,
                                  uint tables_opened_arg,
-                                 char *new_db_arg, char *new_name_arg)
+                                 const LEX_CSTRING *new_db_arg,
+                                 const LEX_CSTRING *new_name_arg)
   : datetime_field(NULL), error_if_not_empty(false),
     tables_opened(tables_opened_arg),
-    new_db(new_db_arg), new_name(new_name_arg),
+    new_db(*new_db_arg), new_name(*new_name_arg),
     fk_error_if_delete_row(false), fk_error_id(NULL),
-    fk_error_table(NULL)
-#ifndef DBUG_OFF
+    fk_error_table(NULL), modified_primary_key(false)
+#ifdef DBUG_ASSERT_EXISTS
     , tmp_table(false)
 #endif
 {
@@ -117,28 +290,31 @@ Alter_table_ctx::Alter_table_ctx(THD *thd, TABLE_LIST *table_list,
   table_name= table_list->table_name;
   alias= (lower_case_table_names == 2) ? table_list->alias : table_name;
 
-  if (!new_db || !my_strcasecmp(table_alias_charset, new_db, db))
+  if (!new_db.str || !my_strcasecmp(table_alias_charset, new_db.str, db.str))
     new_db= db;
 
-  if (new_name)
+  if (new_name.str)
   {
-    DBUG_PRINT("info", ("new_db.new_name: '%s'.'%s'", new_db, new_name));
+    DBUG_PRINT("info", ("new_db.new_name: '%s'.'%s'", new_db.str, new_name.str));
 
-    if (lower_case_table_names == 1) // Convert new_name/new_alias to lower case
+    if (lower_case_table_names == 1) // Convert new_name/new_alias to lower
     {
-      my_casedn_str(files_charset_info, new_name);
+      new_name.length= my_casedn_str(files_charset_info, (char*) new_name.str);
       new_alias= new_name;
     }
     else if (lower_case_table_names == 2) // Convert new_name to lower case
     {
-      strmov(new_alias= new_alias_buff, new_name);
-      my_casedn_str(files_charset_info, new_name);
+      new_alias.str=    new_alias_buff;
+      new_alias.length= new_name.length;
+      strmov(new_alias_buff, new_name.str);
+      new_name.length= my_casedn_str(files_charset_info, (char*) new_name.str);
+
     }
     else
       new_alias= new_name; // LCTN=0 => case sensitive + case preserving
 
     if (!is_database_changed() &&
-        !my_strcasecmp(table_alias_charset, new_name, table_name))
+        !my_strcasecmp(table_alias_charset, new_name.str, table_name.str))
     {
       /*
         Source and destination table names are equal:
@@ -154,22 +330,23 @@ Alter_table_ctx::Alter_table_ctx(THD *thd, TABLE_LIST *table_list,
     new_name= table_name;
   }
 
-  my_snprintf(tmp_name, sizeof(tmp_name), "%s-%lx_%lx", tmp_file_prefix,
-              current_pid, thd->thread_id);
+  tmp_name.str= tmp_name_buff;
+  tmp_name.length= my_snprintf(tmp_name_buff, sizeof(tmp_name_buff), "%s-%lx_%llx",
+                               tmp_file_prefix, current_pid, thd->thread_id);
   /* Safety fix for InnoDB */
   if (lower_case_table_names)
-    my_casedn_str(files_charset_info, tmp_name);
+    tmp_name.length= my_casedn_str(files_charset_info, tmp_name_buff);
 
   if (table_list->table->s->tmp_table == NO_TMP_TABLE)
   {
-    build_table_filename(path, sizeof(path) - 1, db, table_name, "", 0);
+    build_table_filename(path, sizeof(path) - 1, db.str, table_name.str, "", 0);
 
-    build_table_filename(new_path, sizeof(new_path) - 1, new_db, new_name, "", 0);
+    build_table_filename(new_path, sizeof(new_path) - 1, new_db.str, new_name.str, "", 0);
 
     build_table_filename(new_filename, sizeof(new_filename) - 1,
-                         new_db, new_name, reg_ext, 0);
+                         new_db.str, new_name.str, reg_ext, 0);
 
-    build_table_filename(tmp_path, sizeof(tmp_path) - 1, new_db, tmp_name, "",
+    build_table_filename(tmp_path, sizeof(tmp_path) - 1, new_db.str, tmp_name.str, "",
                          FN_IS_TMP);
   }
   else
@@ -180,7 +357,7 @@ Alter_table_ctx::Alter_table_ctx(THD *thd, TABLE_LIST *table_list,
       this case. This fact is enforced with assert.
     */
     build_tmptable_filename(thd, tmp_path, sizeof(tmp_path));
-#ifndef DBUG_OFF
+#ifdef DBUG_ASSERT_EXISTS
     tmp_table= true;
 #endif
   }
@@ -221,25 +398,28 @@ bool Sql_cmd_alter_table::execute(THD *thd)
 
   DBUG_ENTER("Sql_cmd_alter_table::execute");
 
-  if (thd->is_fatal_error) /* out of memory creating a copy of alter_info */
+  if (unlikely(thd->is_fatal_error))
+  {
+    /* out of memory creating a copy of alter_info */
     DBUG_RETURN(TRUE);
+  }
   /*
     We also require DROP priv for ALTER TABLE ... DROP PARTITION, as well
     as for RENAME TO, as being done by SQLCOM_RENAME_TABLE
   */
-  if (alter_info.flags & (Alter_info::ALTER_DROP_PARTITION |
-                          Alter_info::ALTER_RENAME))
+  if ((alter_info.partition_flags & ALTER_PARTITION_DROP) ||
+      (alter_info.flags & ALTER_RENAME))
     priv_needed|= DROP_ACL;
 
   /* Must be set in the parser */
-  DBUG_ASSERT(select_lex->db);
-  DBUG_ASSERT(!(alter_info.flags & Alter_info::ALTER_EXCHANGE_PARTITION));
-  DBUG_ASSERT(!(alter_info.flags & Alter_info::ALTER_ADMIN_PARTITION));
-  if (check_access(thd, priv_needed, first_table->db,
+  DBUG_ASSERT(select_lex->db.str);
+  DBUG_ASSERT(!(alter_info.partition_flags & ALTER_PARTITION_EXCHANGE));
+  DBUG_ASSERT(!(alter_info.partition_flags & ALTER_PARTITION_ADMIN));
+  if (check_access(thd, priv_needed, first_table->db.str,
                    &first_table->grant.privilege,
                    &first_table->grant.m_internal,
                    0, 0) ||
-      check_access(thd, INSERT_ACL | CREATE_ACL, select_lex->db,
+      check_access(thd, INSERT_ACL | CREATE_ACL, select_lex->db.str,
                    &priv,
                    NULL, /* Don't use first_tab->grant with sel_lex->db */
                    0, 0))
@@ -295,9 +475,7 @@ bool Sql_cmd_alter_table::execute(THD *thd)
   {
     // Rename of table
     TABLE_LIST tmp_table;
-    tmp_table.init_one_table(select_lex->db, strlen(select_lex->db),
-                             lex->name.str, lex->name.length,
-                             lex->name.str, TL_IGNORE);
+    tmp_table.init_one_table(&select_lex->db, &lex->name, 0, TL_IGNORE);
     tmp_table.grant.privilege= priv;
     if (check_grant(thd, INSERT_ACL | CREATE_ACL, &tmp_table, FALSE,
                     UINT_MAX, FALSE))
@@ -315,22 +493,23 @@ bool Sql_cmd_alter_table::execute(THD *thd)
                         "INDEX DIRECTORY");
   create_info.data_file_name= create_info.index_file_name= NULL;
 
-#ifdef WITH_WSREP
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+  thd->work_part_info= 0;
+#endif
+
   if (WSREP(thd) &&
       (!thd->is_current_stmt_binlog_format_row() ||
        !thd->find_temporary_table(first_table)))
   {
-    WSREP_TO_ISOLATION_BEGIN_ALTER(((lex->name.str) ? select_lex->db : NULL),
-                                    ((lex->name.str) ? lex->name.str : NULL),
-                                    first_table,
-                                    &alter_info);
+    WSREP_TO_ISOLATION_BEGIN_ALTER((lex->name.str ? select_lex->db.str : NULL),
+                                   (lex->name.str ? lex->name.str : NULL),
+                                   first_table, &alter_info);
 
     thd->variables.auto_increment_offset = 1;
     thd->variables.auto_increment_increment = 1;
   }
-#endif /* WITH_WSREP */
 
-  result= mysql_alter_table(thd, select_lex->db, lex->name.str,
+  result= mysql_alter_table(thd, &select_lex->db, &lex->name,
                             &create_info,
                             first_table,
                             &alter_info,
@@ -352,7 +531,7 @@ bool Sql_cmd_discard_import_tablespace::execute(THD *thd)
   /* first table of first SELECT_LEX */
   TABLE_LIST *table_list= (TABLE_LIST*) select_lex->table_list.first;
 
-  if (check_access(thd, ALTER_ACL, table_list->db,
+  if (check_access(thd, ALTER_ACL, table_list->db.str,
                    &table_list->grant.privilege,
                    &table_list->grant.m_internal,
                    0, 0))