cherry-pick MDEV-1649010.4-nikita-merge

8 files changed, 168 insertions, 71 deletions

diff --git a/mysql-test/suite/versioning/r/alter.result b/mysql-test/suite/versioning/r/alter.result
index e380c207555..d7ebb4dd096 100644
--- a/mysql-test/suite/versioning/r/alter.result
+++ b/mysql-test/suite/versioning/r/alter.result
@@ -642,3 +642,45 @@ create or replace table t1 (f1 int) with system versioning;
 alter table t1 drop system versioning, add f2 int with system versioning;
 ERROR HY000: Table `t1` is not system-versioned
 drop table t1;
+# MDEV-16490 It's possible to make a system versioned table without any versioning field
+set @@system_versioning_alter_history=keep;
+create or replace table t (a int) with system versioning engine=innodb;
+alter table t change column a a int without system versioning;
+ERROR HY000: Table `t` must have at least one versioned column
+alter table t
+change column a a int without system versioning,
+add column b int with system versioning;
+show create table t;
+Table	Create Table
+t	CREATE TABLE `t` (
+  `a` int(11) DEFAULT NULL WITHOUT SYSTEM VERSIONING,
+  `b` int(11) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+alter table t
+change column a new_a int,
+drop system versioning;
+show create table t;
+Table	Create Table
+t	CREATE TABLE `t` (
+  `new_a` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+alter table t add system versioning;
+alter table t change column new_a a int without system versioning;
+show create table t;
+Table	Create Table
+t	CREATE TABLE `t` (
+  `a` int(11) DEFAULT NULL WITHOUT SYSTEM VERSIONING,
+  `b` int(11) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+alter table t
+add column c int,
+change column c c int without system versioning,
+change column b b int without system versioning;
+ERROR HY000: Table `t` must have at least one versioned column
+alter table t
+add column c int without system versioning,
+change column c c int,
+change column b b int without system versioning;
+drop database test;
+create database test;
diff --git a/mysql-test/suite/versioning/r/create.result b/mysql-test/suite/versioning/r/create.result
index 47d069fc4ef..747da5483ec 100644
--- a/mysql-test/suite/versioning/r/create.result
+++ b/mysql-test/suite/versioning/r/create.result
@@ -527,3 +527,13 @@ row_end datetime(6) generated always as row end,
 period for system_time(row_start, row_end)
 ) with system versioning;
 ERROR HY000: `row_start` must be of type TIMESTAMP(6) for system-versioned table `t`
+# MDEV-16490 It's possible to make a system versioned table without any versioning field
+create or replace table t1 (x int without system versioning)
+with system versioning
+select 1 as y;
+create or replace table t1 (x int without system versioning)
+with system versioning
+select 1 as x;
+ERROR HY000: Table `t1` must have at least one versioned column
+drop database test;
+create database test;
diff --git a/mysql-test/suite/versioning/t/alter.test b/mysql-test/suite/versioning/t/alter.test
index b7e623b2897..4cab4798777 100644
--- a/mysql-test/suite/versioning/t/alter.test
+++ b/mysql-test/suite/versioning/t/alter.test
@@ -543,3 +543,37 @@ alter table t1 drop system versioning, add f2 int with system versioning;
 
 drop table t1;
 --source suite/versioning/common_finish.inc
+--echo # MDEV-16490 It's possible to make a system versioned table without any versioning field
+
+set @@system_versioning_alter_history=keep;
+create or replace table t (a int) with system versioning engine=innodb;
+--error ER_VERS_TABLE_MUST_HAVE_COLUMNS
+alter table t change column a a int without system versioning;
+
+alter table t
+  change column a a int without system versioning,
+  add column b int with system versioning;
+show create table t;
+
+alter table t
+  change column a new_a int,
+  drop system versioning;
+show create table t;
+
+alter table t add system versioning;
+alter table t change column new_a a int without system versioning;
+show create table t;
+
+--error ER_VERS_TABLE_MUST_HAVE_COLUMNS
+alter table t
+  add column c int,
+  change column c c int without system versioning,
+  change column b b int without system versioning;
+
+alter table t
+  add column c int without system versioning,
+  change column c c int,
+  change column b b int without system versioning;
+
+drop database test;
+create database test;
diff --git a/mysql-test/suite/versioning/t/create.test b/mysql-test/suite/versioning/t/create.test
index 64858bbe0ec..c98c23f4c05 100644
--- a/mysql-test/suite/versioning/t/create.test
+++ b/mysql-test/suite/versioning/t/create.test
@@ -396,3 +396,14 @@ create table t (
 ) with system versioning;
 
 --source suite/versioning/common_finish.inc
+--echo # MDEV-16490 It's possible to make a system versioned table without any versioning field
+create or replace table t1 (x int without system versioning)
+with system versioning
+select 1 as y;
+--error ER_VERS_TABLE_MUST_HAVE_COLUMNS
+create or replace table t1 (x int without system versioning)
+with system versioning
+select 1 as x;
+
+drop database test;
+create database test;
diff --git a/sql/handler.cc b/sql/handler.cc
index 647848f7702..6abcb8a77ed 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -7375,8 +7375,7 @@ bool Vers_parse_info::fix_implicit(THD *thd, Alter_info *alter_info)
 
 
 bool Table_scope_and_contents_source_st::vers_fix_system_fields(
-  THD *thd, Alter_info *alter_info, const TABLE_LIST &create_table,
-  bool create_select)
+  THD *thd, Alter_info *alter_info, const TABLE_LIST &create_table)
 {
   DBUG_ASSERT(!(alter_info->flags & ALTER_DROP_SYSTEM_VERSIONING));
 
@@ -7416,40 +7415,55 @@ bool Table_scope_and_contents_source_st::vers_fix_system_fields(
   if (vers_info.fix_implicit(thd, alter_info))
     return true;
 
-  int plain_cols= 0; // columns don't have WITH or WITHOUT SYSTEM VERSIONING
-  int vers_cols= 0; // columns have WITH SYSTEM VERSIONING
-  it.rewind();
-  while (const Create_field *f= it++)
-  {
-    if (vers_info.is_start(*f) || vers_info.is_end(*f))
-      continue;
-
-    if (f->versioning == Column_definition::VERSIONING_NOT_SET)
-      plain_cols++;
-    else if (f->versioning == Column_definition::WITH_VERSIONING)
-      vers_cols++;
-  }
-
-  if (!thd->lex->tmp_table() &&
-    // CREATE from SELECT (Create_fields are not yet added)
-    !create_select && vers_cols == 0 && (plain_cols == 0 || !vers_info))
-  {
-    my_error(ER_VERS_TABLE_MUST_HAVE_COLUMNS, MYF(0),
-             create_table.table_name.str);
-    return true;
-  }
-
   return false;
 }
 
 
 bool Table_scope_and_contents_source_st::vers_check_system_fields(
-       THD *thd, Alter_info *alter_info, const TABLE_LIST &create_table)
+        THD *thd, Alter_info *alter_info, const Lex_table_name &table_name,
+        const Lex_table_name &db, int select_count)
 {
   if (!(options & HA_VERSIONED_TABLE))
     return false;
-  return vers_info.check_sys_fields(
-      create_table.table_name, create_table.db, alter_info,
+
+  if (!(alter_info->flags & ALTER_DROP_SYSTEM_VERSIONING))
+  {
+    uint versioned_fields= 0;
+    uint fieldnr= 0;
+    List_iterator<Create_field> field_it(alter_info->create_list);
+    while (Create_field *f= field_it++)
+    {
+      /*
+         The field from the CREATE part can be duplicated in the SELECT part of
+         CREATE...SELECT. In that case double counts should be avoided.
+         select_create::create_table_from_items just pushes the fields back into
+         the create_list, without additional manipulations, so the fields from
+         SELECT go last there.
+       */
+      bool is_dup= false;
+      if (fieldnr >= alter_info->create_list.elements - select_count)
+      {
+        List_iterator<Create_field> dup_it(alter_info->create_list);
+        for (Create_field *dup= dup_it++; !is_dup && dup != f; dup= dup_it++)
+          is_dup= my_strcasecmp(default_charset_info,
+                                dup->field_name.str, f->field_name.str) == 0;
+      }
+
+      if (!(f->flags & VERS_UPDATE_UNVERSIONED_FLAG) && !is_dup)
+        versioned_fields++;
+      fieldnr++;
+    }
+    if (versioned_fields == VERSIONING_FIELDS)
+    {
+      my_error(ER_VERS_TABLE_MUST_HAVE_COLUMNS, MYF(0), table_name.str);
+      return true;
+    }
+  }
+
+  if (!(alter_info->flags & ALTER_ADD_SYSTEM_VERSIONING))
+    return false;
+
+  return vers_info.check_sys_fields(table_name, db, alter_info,
       ha_check_storage_engine_flag(db_type, HTON_NATIVE_SYS_VERSIONING));
 }
 
@@ -7554,20 +7568,7 @@ bool Vers_parse_info::fix_alter_info(THD *thd, Alter_info *alter_info,
     return false;
   }
 
-  if (fix_implicit(thd, alter_info))
-    return true;
-
-  if (alter_info->flags & ALTER_ADD_SYSTEM_VERSIONING)
-  {
-    const bool can_native=
-        ha_check_storage_engine_flag(create_info->db_type,
-                                     HTON_NATIVE_SYS_VERSIONING) ||
-        create_info->db_type->db_type == DB_TYPE_PARTITION_DB;
-    if (check_sys_fields(table_name, share->db, alter_info, can_native))
-      return true;
-  }
-
-  return false;
+  return fix_implicit(thd, alter_info);
 }
 
 bool
@@ -7774,9 +7775,11 @@ bool Table_period_info::check_field(const Create_field* f,
 }
 
 bool Table_scope_and_contents_source_st::check_fields(
-  THD *thd, Alter_info *alter_info, TABLE_LIST &create_table)
+  THD *thd, Alter_info *alter_info, const TABLE_LIST &create_table,
+  int select_count)
 {
-  return vers_check_system_fields(thd, alter_info, create_table)
+  return vers_check_system_fields(thd, alter_info, create_table.table_name,
+                                  create_table.db, select_count)
          || check_period_fields(thd, alter_info);
 }
 
@@ -7826,10 +7829,9 @@ bool Table_scope_and_contents_source_st::check_period_fields(
 bool
 Table_scope_and_contents_source_st::fix_create_fields(THD *thd,
                                                       Alter_info *alter_info,
-                                                      const TABLE_LIST &create_table,
-                                                      bool create_select)
+                                                      const TABLE_LIST &create_table)
 {
-  return vers_fix_system_fields(thd, alter_info, create_table, create_select)
+  return vers_fix_system_fields(thd, alter_info, create_table)
          || fix_period_fields(thd, alter_info);
 }
 
diff --git a/sql/handler.h b/sql/handler.h
index 5e5c0ffc001..f0d3b51a93c 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -1551,7 +1551,7 @@ struct handlerton
     the select statement 'select, otherwise return NULL
   */
   select_handler *(*create_select) (THD *thd, SELECT_LEX *select);
-   
+
    /*********************************************************************
      Table discovery API.
      It allows the server to "discover" tables that exist in the storage
@@ -2163,18 +2163,20 @@ struct Table_scope_and_contents_source_st:
   }
 
   bool fix_create_fields(THD *thd, Alter_info *alter_info,
-                         const TABLE_LIST &create_table,
-                         bool create_select= false);
+                         const TABLE_LIST &create_table);
   bool fix_period_fields(THD *thd, Alter_info *alter_info);
-  bool check_fields(THD *thd, Alter_info *alter_info, TABLE_LIST &create_table);
+  bool check_fields(THD *thd, Alter_info *alter_info, const TABLE_LIST &create_table,
+                    int select_count= 0);
   bool check_period_fields(THD *thd, Alter_info *alter_info);
 
   bool vers_fix_system_fields(THD *thd, Alter_info *alter_info,
-                              const TABLE_LIST &create_table,
-                              bool create_select= false);
+                         const TABLE_LIST &create_table);
 
   bool vers_check_system_fields(THD *thd, Alter_info *alter_info,
-                                const TABLE_LIST &create_table);
+                                const Lex_table_name &table_name,
+                                const Lex_table_name &db,
+                                int select_count= 0);
+
 };
 
 
@@ -2661,7 +2663,7 @@ public:
   {
     return IO_COEFF*io_count*avg_io_cost +
            IO_COEFF*idx_io_count*idx_avg_io_cost +
-           CPU_COEFF*cpu_cost + 
+           CPU_COEFF*cpu_cost +
            MEM_COEFF*mem_cost + IMPORT_COEFF*import_cost;
   }
 
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 8404d777334..89c43938e8a 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -4202,7 +4202,7 @@ TABLE *select_create::create_table_from_items(THD *thd, List<Item> *items,
   if (!opt_explicit_defaults_for_timestamp)
     promote_first_timestamp_column(&alter_info->create_list);
 
-  if (create_info->fix_create_fields(thd, alter_info, *create_table, true))
+  if (create_info->fix_create_fields(thd, alter_info, *create_table))
     DBUG_RETURN(NULL);
 
   while ((item=it++))
@@ -4241,7 +4241,8 @@ TABLE *select_create::create_table_from_items(THD *thd, List<Item> *items,
     alter_info->create_list.push_back(cr_field, thd->mem_root);
   }
 
-  if (create_info->check_fields(thd, alter_info, *create_table))
+  if (create_info->check_fields(thd, alter_info, *create_table,
+                                select_field_count))
     DBUG_RETURN(NULL);
 
   DEBUG_SYNC(thd,"create_table_select_before_create");
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index f736de24bed..38b7f0523fa 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -8697,13 +8697,6 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
     }
   }
 
-  if (table->versioned() && !(alter_info->flags & ALTER_DROP_SYSTEM_VERSIONING) &&
-      new_create_list.elements == VERSIONING_FIELDS)
-  {
-    my_error(ER_VERS_TABLE_MUST_HAVE_COLUMNS, MYF(0), table->s->table_name.str);
-    goto err;
-  }
-
   if (!create_info->comment.str)
   {
     create_info->comment.str= table->s->comment.str;
@@ -9785,11 +9778,13 @@ do_continue:;
     DBUG_RETURN(true);
   }
 
-  set_table_default_charset(thd, create_info, alter_ctx.db);
-
-  if (create_info->check_period_fields(thd, alter_info)
-      || create_info->fix_period_fields(thd, alter_info))
+  if (create_info->check_fields(thd, alter_info, *table_list) ||
+      create_info->fix_period_fields(thd, alter_info))
+  {
     DBUG_RETURN(true);
+  }
+
+  set_table_default_charset(thd, create_info, alter_ctx.db);
 
   if (!opt_explicit_defaults_for_timestamp)
     promote_first_timestamp_column(&alter_info->create_list);
@@ -10287,7 +10282,7 @@ do_continue:;
 
   close_all_tables_for_name(thd, table->s,
                             alter_ctx.is_table_renamed() ?
-                            HA_EXTRA_PREPARE_FOR_RENAME: 
+                            HA_EXTRA_PREPARE_FOR_RENAME:
                             HA_EXTRA_NOT_USED,
                             NULL);
   table_list->table= table= NULL;                  /* Safety */
@@ -10482,7 +10477,7 @@ bool mysql_trans_prepare_alter_copy_data(THD *thd)
   /*
     Turn off recovery logging since rollback of an alter table is to
     delete the new table so there is no need to log the changes to it.
-    
+
     This needs to be done before external_lock.
   */
   DBUG_RETURN(ha_enable_transaction(thd, FALSE) != 0);
@@ -10640,7 +10635,7 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
       char warn_buff[MYSQL_ERRMSG_SIZE];
       bool save_abort_on_warning= thd->abort_on_warning;
       thd->abort_on_warning= false;
-      my_snprintf(warn_buff, sizeof(warn_buff), 
+      my_snprintf(warn_buff, sizeof(warn_buff),
                   "ORDER BY ignored as there is a user-defined clustered index"
                   " in the table '%-.192s'", from->s->table_name.str);
       push_warning(thd, Sql_condition::WARN_LEVEL_WARN, ER_UNKNOWN_ERROR,