MDEV-27293 Allow converting a versioned table from implicit

           to explicit row_start/row_end columns

In case of adding both system fields of same type (length, unsigned
flag) as old implicit system fields do the rename of implicit system
fields to the ones specified in ALTER, remove SYSTEM_INVISIBLE flag in
that case. Correct PERIOD clause must be specified in ALTER as well.

mysql-test/suite/versioning/r/alter.result

@@ -88,9 +88,9 @@ t	CREATE TABLE `t` (
   `a` int(11) DEFAULT NULL
 ) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_uca1400_ai_ci WITH SYSTEM VERSIONING
 alter table t add column trx_start timestamp(6) as row start;
-ERROR HY000: Duplicate ROW START column `trx_start`
+ERROR HY000: Wrong parameters for `t`: missing 'AS ROW END'
 alter table t modify a int as row start;
-ERROR HY000: Duplicate ROW START column `a`
+ERROR 42000: Incorrect column specifier for column 'a'
 alter table t add column b int;
 show create table t;
 Table	Create Table
@@ -879,3 +879,48 @@ t1	CREATE TABLE `t1` (
 ) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_uca1400_ai_ci WITH SYSTEM VERSIONING
 drop table t1;
 # End of 10.5 tests
+#
+# MDEV-27293 Allow converting a versioned table from implicit
+#            to explicit row_start/row_end columns
+#
+create table t1 (x int) with system versioning;
+insert t1 values (0), (1);
+update t1 set x= x + 10;
+set @old_system_versioning_alter_history= @@system_versioning_alter_history;
+set @@system_versioning_alter_history= keep;
+alter table t1 add column y int;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL,
+  `y` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_uca1400_ai_ci WITH SYSTEM VERSIONING
+alter table t1 add column rs timestamp(6) as row start;
+ERROR HY000: Wrong parameters for `t1`: missing 'AS ROW END'
+alter table t1 add column re timestamp(6) as row end, add period for system_time (row_start, re);
+ERROR HY000: Wrong parameters for `t1`: missing 'AS ROW START'
+alter table t1 add column rs timestamp(6) as row start, add column re timestamp(6) as row end;
+ERROR HY000: Wrong parameters for `t1`: missing 'PERIOD FOR SYSTEM_TIME'
+alter table t1 rename column row_start to re;
+ERROR 42S22: Unknown column 'row_start' in 't1'
+alter table t1 add column rs timestamp(5) as row start, add column re timestamp(6) as row end, add period for system_time (rs,re);
+ERROR 42000: Incorrect column specifier for column 'rs'
+alter table t1 add column rs timestamp(6) as row start, add column re timestamp(6) as row end, add period for system_time (rs,re);
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL,
+  `y` int(11) DEFAULT NULL,
+  `rs` timestamp(6) GENERATED ALWAYS AS ROW START,
+  `re` timestamp(6) GENERATED ALWAYS AS ROW END,
+  PERIOD FOR SYSTEM_TIME (`rs`, `re`)
+) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_uca1400_ai_ci WITH SYSTEM VERSIONING
+set @@system_versioning_alter_history= @old_system_versioning_alter_history;
+select x, check_row_ts(rs, re) from t1 for system_time all order by x;
+x	check_row_ts(rs, re)
+0	HISTORICAL ROW
+1	HISTORICAL ROW
+10	CURRENT ROW
+11	CURRENT ROW
+drop table t1;
+# End of 11.6 tests

mysql-test/suite/versioning/t/alter.test

@@ -74,9 +74,9 @@ alter table t add column trx_start timestamp(6) as row start;
 alter table t add system versioning;
 show create table t;
 
---error ER_VERS_DUPLICATE_ROW_START_END
+--error ER_MISSING
 alter table t add column trx_start timestamp(6) as row start;
---error ER_VERS_DUPLICATE_ROW_START_END
+--error ER_WRONG_FIELD_SPEC
 alter table t modify a int as row start;
 
 alter table t add column b int;
@@ -551,7 +551,6 @@ create or replace table t1 (f1 int) with system versioning;
 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;
@@ -779,3 +778,33 @@ show create table t1;
 drop table t1;
 
 --echo # End of 10.5 tests
+
+--echo #
+--echo # MDEV-27293 Allow converting a versioned table from implicit
+--echo #            to explicit row_start/row_end columns
+--echo #
+create table t1 (x int) with system versioning;
+insert t1 values (0), (1);
+update t1 set x= x + 10;
+set @old_system_versioning_alter_history= @@system_versioning_alter_history;
+set @@system_versioning_alter_history= keep;
+alter table t1 add column y int;
+show create table t1;
+--error ER_MISSING
+alter table t1 add column rs timestamp(6) as row start;
+--error ER_MISSING
+alter table t1 add column re timestamp(6) as row end, add period for system_time (row_start, re);
+--error ER_MISSING
+alter table t1 add column rs timestamp(6) as row start, add column re timestamp(6) as row end;
+--error ER_BAD_FIELD_ERROR
+alter table t1 rename column row_start to re;
+--error ER_WRONG_FIELD_SPEC
+alter table t1 add column rs timestamp(5) as row start, add column re timestamp(6) as row end, add period for system_time (rs,re);
+alter table t1 add column rs timestamp(6) as row start, add column re timestamp(6) as row end, add period for system_time (rs,re);
+show create table t1;
+set @@system_versioning_alter_history= @old_system_versioning_alter_history;
+select x, check_row_ts(rs, re) from t1 for system_time all order by x;
+drop table t1;
+
+--echo # End of 11.6 tests
+--source suite/versioning/common_finish.inc

sql/field.h

@@ -5995,6 +5995,12 @@ ulonglong TABLE::vers_start_id() const
   return static_cast<ulonglong>(vers_start_field()->val_int());
 }
 
+inline
+bool TABLE::vers_implicit() const
+{
+  return vers_end_field()->invisible == INVISIBLE_SYSTEM;
+}
+
 double pos_in_interval_for_string(CHARSET_INFO *cset,
                                   const uchar *midp_val, uint32 midp_len,
                                   const uchar *min_val,  uint32 min_len,

sql/handler.cc

@@ -8776,7 +8776,7 @@ bool Vers_parse_info::fix_alter_info(THD *thd, Alter_info *alter_info,
 
   if (!(alter_info->flags & ALTER_ADD_SYSTEM_VERSIONING))
   {
-    List_iterator_fast<Create_field> it(alter_info->create_list);
+    List_iterator<Create_field> it(alter_info->create_list);
     while (Create_field *f= it++)
     {
       if (f->flags & VERS_SYSTEM_FIELD)
@@ -8786,12 +8786,30 @@ bool Vers_parse_info::fix_alter_info(THD *thd, Alter_info *alter_info,
           my_error(ER_VERS_NOT_VERSIONED, MYF(0), table->s->table_name.str);
           return true;
         }
+        if (!table->vers_implicit())
+        {
           my_error(ER_VERS_DUPLICATE_ROW_START_END, MYF(0),
                   f->flags & VERS_ROW_START ? "START" : "END", f->field_name.str);
+
+          return true;
+        }
+        Field *old= f->flags & VERS_ROW_START ? table->vers_start_field() : table->vers_end_field();
+        if (old->type_handler() == f->type_handler() &&
+            old->field_length == f->length &&
+            (old->flags & UNSIGNED_FLAG) == (f->flags & UNSIGNED_FLAG))
+        {
+          alter_info->flags|= ALTER_VERS_EXPLICIT;
+          alter_info->add_alter_list(thd, old->field_name, f->field_name, false);
+          it.remove();
+        }
+        else
+        {
+          my_error(ER_WRONG_FIELD_SPEC, MYF(0), f->field_name.str);
           return true;
         }
       }
     }
+  }
 
   if ((alter_info->flags & ALTER_DROP_PERIOD ||
        versioned_fields || unversioned_fields) && !share->versioned)
@@ -8802,7 +8820,8 @@ bool Vers_parse_info::fix_alter_info(THD *thd, Alter_info *alter_info,
 
   if (share->versioned)
   {
-    if (alter_info->flags & ALTER_ADD_PERIOD)
+    if (!table->vers_implicit() &&
+        (alter_info->flags & ALTER_ADD_PERIOD))
     {
       my_error(ER_VERS_ALREADY_VERSIONED, MYF(0), table_name.str);
       return true;
@@ -8813,6 +8832,9 @@ bool Vers_parse_info::fix_alter_info(THD *thd, Alter_info *alter_info,
 
     DBUG_ASSERT(share->vers_start_field());
     DBUG_ASSERT(share->vers_end_field());
+
+    if (!(alter_info->flags & ALTER_VERS_EXPLICIT))
+    {
       Lex_ident_column start(share->vers_start_field()->field_name);
       Lex_ident_column end(share->vers_end_field()->field_name);
       DBUG_ASSERT(start.str);
@@ -8836,19 +8858,15 @@ bool Vers_parse_info::fix_alter_info(THD *thd, Alter_info *alter_info,
           }
         }
       }
+    } /* if (!convert_explicit) */
+    return check_conditions(table_name, share->db);
+  } /* if (share->versioned) */
 
-    return false;
-  }
-
-  if (fix_implicit(thd, alter_info))
+  if ((alter_info->flags & ALTER_ADD_SYSTEM_VERSIONING) &&
+      (fix_implicit(thd, alter_info) ||
+        check_sys_fields(table_name, share->db, alter_info)))
     return true;
 
-  if (alter_info->flags & ALTER_ADD_SYSTEM_VERSIONING)
-  {
-    if (check_sys_fields(table_name, share->db, alter_info))
-      return true;
-  }
-
   return false;
 }
 

sql/handler.h

@@ -730,6 +730,7 @@ typedef bool Log_func(THD*, TABLE*, Event_log *, binlog_cache_data *, bool,
 #define ALTER_DROP_SYSTEM_VERSIONING (1ULL << 32)
 #define ALTER_ADD_PERIOD             (1ULL << 33)
 #define ALTER_DROP_PERIOD            (1ULL << 34)
+#define ALTER_VERS_EXPLICIT          (1ULL << 35)
 
 /*
   Following defines are used by ALTER_INPLACE_TABLE

sql/sql_alter.h

@@ -325,6 +325,9 @@ class Alter_info
   }
   uint check_vcol_field(Item_field *f) const;
 
+  bool add_alter_list(THD *thd, LEX_CSTRING name, LEX_CSTRING new_name,
+                      bool exists);
+
 private:
   Alter_info &operator=(const Alter_info &rhs); // not implemented
   Alter_info(const Alter_info &rhs);            // not implemented

sql/sql_lex.cc

@@ -562,14 +562,14 @@ bool LEX::add_alter_list(LEX_CSTRING name, Virtual_column_info *expr,
   return false;
 }
 
-
-bool LEX::add_alter_list(LEX_CSTRING name, LEX_CSTRING new_name, bool exists)
+bool Alter_info::add_alter_list(THD *thd, LEX_CSTRING name,
+                                LEX_CSTRING new_name, bool exists)
 {
   Alter_column *ac= new (thd->mem_root) Alter_column(name, new_name, exists);
   if (unlikely(ac == NULL))
     return true;
-  alter_info.alter_list.push_back(ac, thd->mem_root);
-  alter_info.flags|= ALTER_RENAME_COLUMN;
+  alter_list.push_back(ac, thd->mem_root);
+  flags|= ALTER_RENAME_COLUMN;
   return false;
 }
 

sql/sql_lex.h

@@ -4415,7 +4415,11 @@ struct LEX: public Query_tables_list
   }
   bool add_alter_list(LEX_CSTRING par_name, Virtual_column_info *expr,
                       bool par_exists);
-  bool add_alter_list(LEX_CSTRING name, LEX_CSTRING new_name, bool exists);
+  bool add_alter_list(LEX_CSTRING name, LEX_CSTRING new_name, bool exists)
+  {
+    return alter_info.add_alter_list(thd, name, new_name, exists);
+  }
+
   bool add_alter_list_item_convert_to_charset(Sql_used *used,
                                               const Charset_collation_map_st &map,
                                               CHARSET_INFO *cs)

sql/sql_table.cc

@@ -8398,7 +8398,8 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
       dropped_sys_vers_fields|= field->flags;
       drop_it.remove();
     }
-    else if (field->invisible < INVISIBLE_SYSTEM)
+    else if (field->invisible < INVISIBLE_SYSTEM ||
+             (alter_info->flags & ALTER_VERS_EXPLICIT))
     {
       /*
         This field was not dropped and not changed, add it to the list
@@ -8413,12 +8414,14 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
         if (field->field_name.streq(alter->name))
 	  break;
       }
-      if (alter && field->invisible < INVISIBLE_SYSTEM)
+      if (alter)
       {
         if (alter->is_rename())
         {
           def->change= Lex_ident_column(alter->name);
           def->field_name= Lex_ident_column(alter->new_name);
+          if (vers_system_invisible)
+            def->invisible= VISIBLE;
           column_rename_param.fields.push_back(def);
           if (field->flags & VERS_ROW_START)
           {

sql/table.h

@@ -1936,6 +1936,7 @@ struct TABLE
   bool vers_switch_partition(THD *thd, TABLE_LIST *table_list,
                              Open_table_context *ot_ctx);
 #endif
+  bool vers_implicit() const;
 
   int update_generated_fields();
   void period_prepare_autoinc();