MDEV-17082 Application-time periods: CREATE

* add syntax `CREATE TABLE ... PERIOD FOR <apptime>`
* add table period entity

mysql-test/suite/period/r/create.result

@@ -0,0 +1,92 @@
+create or replace table t (id int primary key, s date, e date,
+period for mytime(s,e));
+show create table t;
+Table	Create Table
+t	CREATE TABLE `t` (
+  `id` int(11) NOT NULL,
+  `s` date NOT NULL,
+  `e` date NOT NULL,
+  PRIMARY KEY (`id`),
+  PERIOD FOR `mytime` (`s`, `e`),
+  CONSTRAINT `CONSTRAINT_1` CHECK (`s` < `e`)
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+create or replace table t (id int primary key, s timestamp(6), e timestamp(6),
+period for mytime(s,e));
+show create table t;
+Table	Create Table
+t	CREATE TABLE `t` (
+  `id` int(11) NOT NULL,
+  `s` timestamp(6) NOT NULL DEFAULT '0000-00-00 00:00:00.000000',
+  `e` timestamp(6) NOT NULL DEFAULT '0000-00-00 00:00:00.000000',
+  PRIMARY KEY (`id`),
+  PERIOD FOR `mytime` (`s`, `e`),
+  CONSTRAINT `CONSTRAINT_1` CHECK (`s` < `e`)
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+# SQL16, Part 2, 11.3  <table definition>, Syntax Rules, 2)a)
+# 2) If a <table period definition> TPD is specified, then:
+# a) <table scope> shall not be specified.
+create or replace temporary table t (s date, e date, period for mytime(s,e));
+ERROR HY000: Application-time period table cannot be temporary
+# SQL16, Part 2, 11.3  <table definition>, Syntax Rules, 2)e)iii)
+# The <data type or domain name> contained in CD1 is either DATE or a
+# timestamp type and it is equivalent to the <data type or domain name>
+# contained in CD2.
+create or replace table t (id int primary key, s datetime, e date,
+period for mytime(s,e));
+ERROR HY000: Fields of PERIOD FOR `mytime` have different types
+create or replace table t (s timestamp(2), e timestamp(6),
+period for mytime(s,e));
+ERROR HY000: Fields of PERIOD FOR `mytime` have different types
+create or replace table t (id int primary key, s int, e date,
+period for mytime(s,e));
+ERROR 42000: Incorrect column specifier for column 's'
+create or replace table t (id int primary key, s date, e date,
+period for mytime(s,x));
+ERROR 42S22: Unknown column 'x' in 'mytime'
+create or replace table t (id int primary key, s date, e date,
+period for mytime(s,e),
+period for mytime2(s,e));
+ERROR HY000: Cannot specify more than one application-time period
+# SQL16, Part 2, 11.3  <table definition>, Syntax Rules, 2)d)
+# No <column name> in any <column definition> shall be equivalent to PN.
+create or replace table t (mytime int, s date, e date,
+period for mytime(s,e));
+ERROR 42S21: Duplicate column name 'mytime'
+# SQL16, Part 2, 11.3  <table definition>, Syntax Rules, 2)e)v)2)A)
+# Neither CD1 nor CD2 shall contain an <identity column specification>, a
+# <generation clause>, a <system time period start column specification>,
+#  or a <system time period end column specification>.
+create or replace table t (id int primary key,
+s date,
+e date generated always as (s+1),
+period for mytime(s,e));
+ERROR HY000: Period field `e` cannot be GENERATED ALWAYS AS
+create or replace table t (id int primary key,
+s date,
+e date as (s+1) VIRTUAL,
+period for mytime(s,e));
+ERROR HY000: Period field `e` cannot be GENERATED ALWAYS AS
+create or replace table t (id int primary key, s timestamp(6), e timestamp(6),
+st timestamp(6) as row start,
+en timestamp(6) as row end,
+period for system_time (st, en),
+period for mytime(st,e)) with system versioning;
+ERROR HY000: Period field `st` cannot be GENERATED ALWAYS AS
+# SQL16, Part 2, 11.3  <table definition>, Syntax Rules, 2)
+# Let IDCN be an implementation-dependent <constraint name> that is not
+# equivalent to the <constraint name> of any table constraint descriptor
+# included in S.
+create or replace table t (x int, s date, e date,
+period for mytime(s, e),
+constraint mytime check (x > 1));
+show create table t;
+Table	Create Table
+t	CREATE TABLE `t` (
+  `x` int(11) DEFAULT NULL,
+  `s` date NOT NULL,
+  `e` date NOT NULL,
+  PERIOD FOR `mytime` (`s`, `e`),
+  CONSTRAINT `CONSTRAINT_1` CHECK (`s` < `e`),
+  CONSTRAINT `mytime` CHECK (`x` > 1)
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+create or replace database test;

mysql-test/suite/period/t/create.test

@@ -0,0 +1,73 @@
+create or replace table t (id int primary key, s date, e date,
+                           period for mytime(s,e));
+show create table t;
+create or replace table t (id int primary key, s timestamp(6), e timestamp(6),
+                           period for mytime(s,e));
+show create table t;
+
+--echo # SQL16, Part 2, 11.3  <table definition>, Syntax Rules, 2)a)
+--echo # 2) If a <table period definition> TPD is specified, then:
+--echo # a) <table scope> shall not be specified.
+--error ER_PERIOD_TEMPORARY_NOT_ALLOWED
+create or replace temporary table t (s date, e date, period for mytime(s,e));
+
+--echo # SQL16, Part 2, 11.3  <table definition>, Syntax Rules, 2)e)iii)
+--echo # The <data type or domain name> contained in CD1 is either DATE or a
+--echo # timestamp type and it is equivalent to the <data type or domain name>
+--echo # contained in CD2.
+--error ER_PERIOD_TYPES_MISMATCH
+create or replace table t (id int primary key, s datetime, e date,
+                           period for mytime(s,e));
+--error ER_PERIOD_TYPES_MISMATCH
+create or replace table t (s timestamp(2), e timestamp(6),
+                           period for mytime(s,e));
+--error ER_WRONG_FIELD_SPEC
+create or replace table t (id int primary key, s int, e date,
+                           period for mytime(s,e));
+--error ER_BAD_FIELD_ERROR
+create or replace table t (id int primary key, s date, e date,
+                           period for mytime(s,x));
+--error ER_MORE_THAN_ONE_PERIOD
+create or replace table t (id int primary key, s date, e date,
+                           period for mytime(s,e),
+                           period for mytime2(s,e));
+
+--echo # SQL16, Part 2, 11.3  <table definition>, Syntax Rules, 2)d)
+--echo # No <column name> in any <column definition> shall be equivalent to PN.
+--error ER_DUP_FIELDNAME
+create or replace table t (mytime int, s date, e date,
+    period for mytime(s,e));
+
+--echo # SQL16, Part 2, 11.3  <table definition>, Syntax Rules, 2)e)v)2)A)
+--echo # Neither CD1 nor CD2 shall contain an <identity column specification>, a
+--echo # <generation clause>, a <system time period start column specification>,
+--echo #  or a <system time period end column specification>.
+--error ER_PERIOD_FIELD_WRONG_ATTRIBUTES
+create or replace table t (id int primary key,
+                           s date,
+                           e date generated always as (s+1),
+                           period for mytime(s,e));
+
+--error ER_PERIOD_FIELD_WRONG_ATTRIBUTES
+create or replace table t (id int primary key,
+                           s date,
+                           e date as (s+1) VIRTUAL,
+                           period for mytime(s,e));
+
+--error ER_PERIOD_FIELD_WRONG_ATTRIBUTES
+create or replace table t (id int primary key, s timestamp(6), e timestamp(6),
+                           st timestamp(6) as row start,
+                           en timestamp(6) as row end,
+                           period for system_time (st, en),
+                           period for mytime(st,e)) with system versioning;
+
+--echo # SQL16, Part 2, 11.3  <table definition>, Syntax Rules, 2)
+--echo # Let IDCN be an implementation-dependent <constraint name> that is not
+--echo # equivalent to the <constraint name> of any table constraint descriptor
+--echo # included in S.
+create or replace table t (x int, s date, e date,
+                           period for mytime(s, e),
+                           constraint mytime check (x > 1));
+show create table t;
+
+create or replace database test;

sql/field.h

@@ -4560,6 +4560,8 @@ public:
 
   enum_column_versioning versioning;
 
+  Table_period_info *period;
+
   Column_definition()
    :Type_handler_hybrid_field_type(&type_handler_null),
     compression_method_ptr(0),
@@ -4568,7 +4570,7 @@ public:
     flags(0), pack_length(0), key_length(0),
     option_list(NULL),
     vcol_info(0), default_value(0), check_constraint(0),
-    versioning(VERSIONING_NOT_SET)
+    versioning(VERSIONING_NOT_SET), period(NULL)
   {
     interval_list.empty();
   }

sql/handler.cc

@@ -7183,8 +7183,8 @@ bool Vers_parse_info::fix_implicit(THD *thd, Alter_info *alter_info)
 
   alter_info->flags|= ALTER_PARSER_ADD_COLUMN;
 
-  system_time= start_end_t(default_start, default_end);
+  period= start_end_t(default_start, default_end);
-  as_row= system_time;
+  as_row= period;
 
   if (vers_create_sys_field(thd, default_start, alter_info, VERS_SYS_START_FLAG) ||
       vers_create_sys_field(thd, default_end, alter_info, VERS_SYS_END_FLAG))
@@ -7375,7 +7375,7 @@ bool Vers_parse_info::fix_alter_info(THD *thd, Alter_info *alter_info,
     DBUG_ASSERT(end.str);
 
     as_row= start_end_t(start, end);
-    system_time= as_row;
+    period= as_row;
 
     if (alter_info->create_list.elements)
     {
@@ -7461,7 +7461,7 @@ Vers_parse_info::fix_create_like(Alter_info &alter_info, HA_CREATE_INFO &create_
   }
 
   as_row= start_end_t(f_start->field_name, f_end->field_name);
-  system_time= as_row;
+  period= as_row;
 
   create_info.options|= HA_VERSIONED_TABLE;
   return false;
@@ -7486,14 +7486,14 @@ bool Vers_parse_info::check_conditions(const Lex_table_name &table_name,
     return true;
   }
 
-  if (!system_time.start || !system_time.end)
+  if (!period.start || !period.end)
   {
     my_error(ER_MISSING, MYF(0), table_name.str, "PERIOD FOR SYSTEM_TIME");
     return true;
   }
 
-  if (!as_row.start.streq(system_time.start) ||
+  if (!as_row.start.streq(period.start) ||
-      !as_row.end.streq(system_time.end))
+      !as_row.end.streq(period.end))
   {
     my_error(ER_VERS_PERIOD_COLUMNS, MYF(0), as_row.start.str, as_row.end.str);
     return true;
@@ -7583,3 +7583,98 @@ bool Vers_parse_info::check_sys_fields(const Lex_table_name &table_name,
            "ROW END" : found_flag ? "ROW START" : "ROW START/END");
   return true;
 }
+
+static bool check_period_field(const Create_field* f, const char* name,
+                               const char* period_name)
+{
+  bool res= false;
+  if (!f)
+  {
+    my_error(ER_BAD_FIELD_ERROR, MYF(0), name, period_name);
+    res= true;
+  }
+  else if (f->type_handler()->mysql_timestamp_type() == MYSQL_TIMESTAMP_ERROR)
+  {
+    my_error(ER_WRONG_FIELD_SPEC, MYF(0), name);
+    res= true;
+  }
+  else if (f->vcol_info || f->flags & VERS_SYSTEM_FIELD)
+  {
+    my_error(ER_PERIOD_FIELD_WRONG_ATTRIBUTES, MYF(0),
+             f->field_name.str, "GENERATED ALWAYS AS");
+  }
+
+  return res;
+}
+
+bool Table_scope_and_contents_source_st::check_fields(
+  THD *thd, Alter_info *alter_info, TABLE_LIST &create_table)
+{
+  bool res= vers_check_system_fields(thd, alter_info, create_table);
+  if (res)
+    return true;
+
+  if (!period_info.name)
+    return false;
+
+  if (tmp_table())
+  {
+    my_error(ER_PERIOD_TEMPORARY_NOT_ALLOWED, MYF(0));
+    return true;
+  }
+
+  Table_period_info::start_end_t &period= period_info.period;
+  const Create_field *row_start= NULL;
+  const Create_field *row_end= NULL;
+  List_iterator<Create_field> it(alter_info->create_list);
+  while (const Create_field *f= it++)
+  {
+    if (period.start.streq(f->field_name)) row_start= f;
+    else if (period.end.streq(f->field_name)) row_end= f;
+
+    if (period_info.name.streq(f->field_name))
+    {
+      my_error(ER_DUP_FIELDNAME, MYF(0), f->field_name.str);
+      return true;
+    }
+  }
+
+  res= check_period_field(row_start, period.start.str, period_info.name.str);
+  res= res || check_period_field(row_end, period.end.str, period_info.name.str);
+  if (res)
+    return true;
+
+  if (row_start->type_handler() != row_end->type_handler()
+      || row_start->length != row_end->length)
+  {
+    my_error(ER_PERIOD_TYPES_MISMATCH, MYF(0), period_info.name.str);
+    res= true;
+  }
+
+  return res;
+}
+
+bool
+Table_scope_and_contents_source_st::fix_create_fields(THD *thd,
+                                                      Alter_info *alter_info,
+                                                      const TABLE_LIST &create_table,
+                                                      bool create_select)
+{
+  if (vers_fix_system_fields(thd, alter_info, create_table, create_select))
+    return true;
+
+  if (!period_info.name)
+    return false;
+
+  Table_period_info::start_end_t &period= period_info.period;
+  List_iterator<Create_field> it(alter_info->create_list);
+  while (Create_field *f= it++)
+  {
+    if (period.start.streq(f->field_name) || period.end.streq(f->field_name))
+    {
+      f->period= &period_info;
+      f->flags|= NOT_NULL_FLAG;
+    }
+  }
+  return false;
+}

sql/handler.h

@@ -1968,57 +1968,61 @@ enum vers_sys_type_t
   VERS_TRX_ID
 };
 
-extern const LEX_CSTRING null_clex_str;
+struct Table_period_info
-
-struct Vers_parse_info
 {
-  Vers_parse_info() :
+  Table_period_info() {}
-    check_unit(VERS_UNDEFINED),
+  Table_period_info(const char *name_arg, size_t size) :
-    versioned_fields(false),
+    name(name_arg, size) {}
-    unversioned_fields(false)
-  {}
 
-  void init() // Deep initialization
+  Lex_ident name;
-  {
-    system_time= start_end_t(null_clex_str, null_clex_str);
-    as_row= start_end_t(null_clex_str, null_clex_str);
-    check_unit= VERS_UNDEFINED;
-    versioned_fields= false;
-    unversioned_fields= false;
-  }
 
   struct start_end_t
   {
-    start_end_t()
+    start_end_t() {};
-    {}
+    start_end_t(const LEX_CSTRING& _start, const LEX_CSTRING& _end) :
-    start_end_t(LEX_CSTRING _start, LEX_CSTRING _end) :
       start(_start),
       end(_end) {}
     Lex_ident start;
     Lex_ident end;
   };
+  start_end_t period;
 
-  start_end_t system_time;
+  bool is_set() const
-  start_end_t as_row;
-  vers_sys_type_t check_unit;
-
-  void set_system_time(Lex_ident start, Lex_ident end)
   {
-    system_time.start= start;
+    DBUG_ASSERT(bool(period.start) == bool(period.end));
-    system_time.end= end;
+    return period.start;
   }
 
+  void set_period(const Lex_ident& start, const Lex_ident& end)
+  {
+    period.start= start;
+    period.end= end;
+  }
+};
+
+struct Vers_parse_info: public Table_period_info
+{
+  Vers_parse_info() :
+    Table_period_info(STRING_WITH_LEN("SYSTEM_TIME")),
+    check_unit(VERS_UNDEFINED),
+    versioned_fields(false),
+    unversioned_fields(false)
+  {}
+
+  Table_period_info::start_end_t as_row;
+  vers_sys_type_t check_unit;
+
 protected:
   friend struct Table_scope_and_contents_source_st;
   void set_start(const LEX_CSTRING field_name)
   {
     as_row.start= field_name;
-    system_time.start= field_name;
+    period.start= field_name;
   }
   void set_end(const LEX_CSTRING field_name)
   {
     as_row.end= field_name;
-    system_time.end= field_name;
+    period.end= field_name;
   }
   bool is_start(const char *name) const;
   bool is_end(const char *name) const;
@@ -2027,7 +2031,7 @@ protected:
   bool fix_implicit(THD *thd, Alter_info *alter_info);
   operator bool() const
   {
-    return as_row.start || as_row.end || system_time.start || system_time.end;
+    return as_row.start || as_row.end || period.start || period.end;
   }
   bool need_check(const Alter_info *alter_info) const;
   bool check_conditions(const Lex_table_name &table_name,
@@ -2151,21 +2155,27 @@ struct Table_scope_and_contents_source_st:
   SQL_I_List<TABLE_LIST> merge_list;
 
   Vers_parse_info vers_info;
+  Table_period_info period_info;
 
   void init()
   {
     Table_scope_and_contents_source_pod_st::init();
     merge_list.empty();
-    vers_info.init();
+    vers_info= {};
+    period_info= {};
   }
 
-  bool vers_fix_system_fields(THD *thd, Alter_info *alter_info,
+  bool fix_create_fields(THD *thd, Alter_info *alter_info,
                          const TABLE_LIST &create_table,
                          bool create_select= false);
+  bool check_fields(THD *thd, Alter_info *alter_info, TABLE_LIST &create_table);
+
+  bool vers_fix_system_fields(THD *thd, Alter_info *alter_info,
+                              const TABLE_LIST &create_table,
+                              bool create_select= false);
 
   bool vers_check_system_fields(THD *thd, Alter_info *alter_info,
                                 const TABLE_LIST &create_table);
-
 };
 
 

sql/share/errmsg-utf8.txt

@@ -7936,3 +7936,11 @@ ER_USER_IS_BLOCKED
 ER_ACCOUNT_HAS_BEEN_LOCKED
         eng "Access denied, this account is locked"
         rum "Acces refuzat, acest cont este blocat"
+ER_PERIOD_TEMPORARY_NOT_ALLOWED
+        eng "Application-time period table cannot be temporary"
+ER_PERIOD_TYPES_MISMATCH
+        eng "Fields of PERIOD FOR %`s have different types"
+ER_MORE_THAN_ONE_PERIOD
+        eng "Cannot specify more than one application-time period"
+ER_PERIOD_FIELD_WRONG_ATTRIBUTES
+        eng "Period field %`s cannot be %s"

sql/sql_lex.h

@@ -4189,7 +4189,7 @@ public:
   void add_key_to_list(LEX_CSTRING *field_name,
                        enum Key::Keytype type, bool check_exists);
   // Add a constraint as a part of CREATE TABLE or ALTER TABLE
-  bool add_constraint(LEX_CSTRING *name, Virtual_column_info *constr,
+  bool add_constraint(const LEX_CSTRING *name, Virtual_column_info *constr,
                       bool if_not_exists)
   {
     constr->name= *name;
@@ -4271,6 +4271,25 @@ public:
   {
     return create_info.vers_info;
   }
+
+  int add_period(Lex_ident name, Lex_ident_sys_st start, Lex_ident_sys_st end)
+  {
+    Table_period_info &info= create_info.period_info;
+    if (info.is_set())
+    {
+       my_error(ER_MORE_THAN_ONE_PERIOD, MYF(0));
+       return 1;
+    }
+    info.set_period(start, end);
+    info.name= name;
+
+    Virtual_column_info *constr= new Virtual_column_info();
+    constr->expr= lt_creator.create(thd, create_item_ident_nosp(thd, &start),
+                                    create_item_ident_nosp(thd, &end));
+    add_constraint(&null_clex_str, constr, false);
+    return 0;
+  }
+
   sp_package *get_sp_package() const;
 
   /**

sql/sql_parse.cc

@@ -4311,8 +4311,8 @@ mysql_execute_command(THD *thd)
       }
       else
       {
-        if (create_info.vers_fix_system_fields(thd, &alter_info, *create_table) ||
+        if (create_info.fix_create_fields(thd, &alter_info, *create_table) ||
-            create_info.vers_check_system_fields(thd, &alter_info, *create_table))
+            create_info.check_fields(thd, &alter_info, *create_table))
           goto end_with_restore_list;
 
         /*

sql/sql_show.cc

@@ -2047,6 +2047,22 @@ end_options:
   append_directory(thd, packet, "INDEX", create_info.index_file_name);
 }
 
+static void append_period(THD *thd, String *packet, const LEX_CSTRING &start,
+                          const LEX_CSTRING &end, const LEX_CSTRING &period,
+                          bool ident)
+{
+  packet->append(STRING_WITH_LEN(",\n  PERIOD FOR "));
+  if (ident)
+    append_identifier(thd, packet, period.str, period.length);
+  else
+    packet->append(period);
+  packet->append(STRING_WITH_LEN(" ("));
+  append_identifier(thd, packet, start.str, start.length);
+  packet->append(STRING_WITH_LEN(", "));
+  append_identifier(thd, packet, end.str, end.length);
+  packet->append(STRING_WITH_LEN(")"));
+}
+
 /*
   Build a CREATE TABLE statement for a table.
 
@@ -2085,6 +2101,7 @@ int show_create_table(THD *thd, TABLE_LIST *table_list, String *packet,
   KEY *key_info;
   TABLE *table= table_list->table;
   TABLE_SHARE *share= table->s;
+  TABLE_SHARE::period_info_t &period= share->period;
   sql_mode_t sql_mode= thd->variables.sql_mode;
   bool explicit_fields= false;
   bool foreign_db_mode=  sql_mode & (MODE_POSTGRESQL | MODE_ORACLE |
@@ -2364,11 +2381,8 @@ int show_create_table(THD *thd, TABLE_LIST *table_list, String *packet,
     DBUG_ASSERT(!explicit_fields || fe->invisible < INVISIBLE_SYSTEM);
     if (explicit_fields)
     {
-      packet->append(STRING_WITH_LEN(",\n  PERIOD FOR SYSTEM_TIME ("));
+      append_period(thd, packet, fs->field_name, fe->field_name,
-      append_identifier(thd,packet,fs->field_name.str, fs->field_name.length);
+                    table->s->vers.name, false);
-      packet->append(STRING_WITH_LEN(", "));
-      append_identifier(thd,packet,fe->field_name.str, fe->field_name.length);
-      packet->append(STRING_WITH_LEN(")"));
     }
     else
     {
@@ -2377,6 +2391,15 @@ int show_create_table(THD *thd, TABLE_LIST *table_list, String *packet,
     }
   }
 
+  if (period.name)
+  {
+    append_period(thd, packet,
+                  period.start_field(share)->field_name,
+                  period.end_field(share)->field_name,
+                  period.name, true);
+  }
+
+
   /*
     Get possible foreign key definitions stored in InnoDB and append them
     to the CREATE TABLE statement

sql/sql_table.cc

@@ -2998,7 +2998,8 @@ CHARSET_INFO* get_sql_field_charset(Column_definition *sql_field,
    by adding the features DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP.
 
    If the first TIMESTAMP column appears to be nullable, or to have an
-   explicit default, or to be a virtual column, then no promition is done.
+   explicit default, or to be a virtual column, or to be part of table period,
+   then no promotion is done.
 
    @param column_definitions The list of column definitions, in the physical
                              order in which they appear in the table.
@@ -3019,6 +3020,7 @@ void promote_first_timestamp_column(List<Create_field> *column_definitions)
           column_definition->default_value == NULL &&   // no constant default,
           column_definition->unireg_check == Field::NONE && // no function default
           column_definition->vcol_info == NULL &&
+          column_definition->period == NULL &&
           !(column_definition->flags & VERS_SYSTEM_FIELD)) // column isn't generated
       {
         DBUG_PRINT("info", ("First TIMESTAMP column '%s' was promoted to "

sql/sql_yacc.yy

@@ -6641,7 +6641,12 @@ period_for_system_time:
           PERIOD_SYM FOR_SYSTEM_TIME_SYM '(' ident ',' ident ')'
           {
             Vers_parse_info &info= Lex->vers_get_info();
-            info.set_system_time($4, $6);
+            info.set_period($4, $6);
+          }
+        | PERIOD_SYM FOR_SYM ident '(' ident ',' ident ')'
+          {
+            if (Lex->add_period($3, $5, $7))
+              MYSQL_YYABORT;
           }
         ;
 

sql/sql_yacc_ora.yy

@@ -6579,7 +6579,12 @@ period_for_system_time:
           PERIOD_SYM FOR_SYSTEM_TIME_SYM '(' ident ',' ident ')'
           {
             Vers_parse_info &info= Lex->vers_get_info();
-            info.set_system_time($4, $6);
+            info.set_period($4, $6);
+          }
+        | PERIOD_SYM FOR_SYM ident '(' ident ',' ident ')'
+          {
+            if (Lex->add_period($3, $5, $7))
+              MYSQL_YYABORT;
           }
         ;
 

sql/table.cc

@@ -50,6 +50,17 @@
 #define MYSQL57_GENERATED_FIELD 128
 #define MYSQL57_GCOL_HEADER_SIZE 4
 
+struct extra2_fields
+{
+  LEX_CUSTRING version;
+  LEX_CUSTRING options;
+  Lex_ident engine;
+  LEX_CUSTRING gis;
+  LEX_CUSTRING field_flags;
+  const uchar *system_period;
+  LEX_CUSTRING application_period;
+};
+
 static Virtual_column_info * unpack_vcol_info_from_frm(THD *, MEM_ROOT *,
               TABLE *, String *, Virtual_column_info **, bool *);
 static bool check_vcol_forward_refs(Field *, Virtual_column_info *,
@@ -1352,6 +1363,103 @@ void TABLE::find_constraint_correlated_indexes()
 }
 
 
+bool TABLE_SHARE::init_period_from_extra2(period_info_t &period,
+                                          const uchar *data)
+{
+  period.start_fieldno= uint2korr(data);
+  period.end_fieldno= uint2korr(data + frm_fieldno_size);
+  return period.start_fieldno >= fields || period.end_fieldno >= fields;
+}
+
+
+static
+bool read_extra2(const uchar *frm_image, size_t len, extra2_fields *fields)
+{
+  const uchar *extra2= frm_image + 64;
+
+  DBUG_ENTER("read_extra2");
+
+  memset(fields, 0, sizeof(extra2_fields));
+
+  if (*extra2 != '/')   // old frm had '/' there
+  {
+    const uchar *e2end= extra2 + len;
+    while (extra2 + 3 <= e2end)
+    {
+      uchar type= *extra2++;
+      size_t length= *extra2++;
+      if (!length)
+      {
+        if (extra2 + 2 >= e2end)
+          DBUG_RETURN(true);
+        length= uint2korr(extra2);
+        extra2+= 2;
+        if (length < 256)
+          DBUG_RETURN(true);
+      }
+      if (extra2 + length > e2end)
+        DBUG_RETURN(true);
+      switch (type) {
+        case EXTRA2_TABLEDEF_VERSION:
+          if (fields->version.str) // see init_from_sql_statement_string()
+          {
+            if (length != fields->version.length)
+              DBUG_RETURN(true);
+          }
+          else
+          {
+            fields->version.str= extra2;
+            fields->version.length= length;
+          }
+          break;
+        case EXTRA2_ENGINE_TABLEOPTS:
+          if (fields->options.str)
+            DBUG_RETURN(true);
+          fields->options.str= extra2;
+          fields->options.length= length;
+          break;
+        case EXTRA2_DEFAULT_PART_ENGINE:
+          fields->engine.set((char*)extra2, length);
+          break;
+        case EXTRA2_GIS:
+#ifdef HAVE_SPATIAL
+          if (fields->gis.str)
+            DBUG_RETURN(true);
+          fields->gis.str= extra2;
+          fields->gis.length= length;
+#endif /*HAVE_SPATIAL*/
+          break;
+        case EXTRA2_PERIOD_FOR_SYSTEM_TIME:
+          if (fields->system_period || length != 2 * sizeof(uint16))
+            DBUG_RETURN(true);
+          fields->system_period = extra2;
+          break;
+        case EXTRA2_FIELD_FLAGS:
+          if (fields->field_flags.str)
+            DBUG_RETURN(true);
+          fields->field_flags.str= extra2;
+          fields->field_flags.length= length;
+          break;
+        case EXTRA2_APPLICATION_TIME_PERIOD:
+          if (fields->application_period.str)
+            DBUG_RETURN(true);
+          fields->application_period.str= extra2;
+          fields->application_period.length= length;
+          break;
+        default:
+          /* abort frm parsing if it's an unknown but important extra2 value */
+          if (type >= EXTRA2_ENGINE_IMPORTANT)
+            DBUG_RETURN(true);
+      }
+      extra2+= length;
+    }
+    if (extra2 != e2end)
+      DBUG_RETURN(true);
+  }
+  DBUG_RETURN(false);
+}
+
+
 /**
   Read data from a binary .frm file image into a TABLE_SHARE
 
@@ -1380,7 +1488,7 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
   uint i;
   bool use_hash, mysql57_null_bits= 0;
   char *keynames, *names, *comment_pos;
-  const uchar *forminfo, *extra2;
+  const uchar *forminfo;
   const uchar *frm_image_end = frm_image + frm_length;
   uchar *record, *null_flags, *null_pos, *UNINIT_VAR(mysql57_vcol_null_pos);
   const uchar *disk_buff, *strpos;
@@ -1395,21 +1503,18 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
   my_bitmap_map *bitmaps;
   bool null_bits_are_used;
   uint vcol_screen_length;
-  size_t UNINIT_VAR(options_len);
   uchar *vcol_screen_pos;
-  const uchar *options= 0;
+  LEX_CUSTRING options;
-  LEX_CUSTRING gis_options= { NULL, 0};
   KEY first_keyinfo;
   uint len;
   uint ext_key_parts= 0;
   plugin_ref se_plugin= 0;
-  const uchar *system_period= 0;
   bool vers_can_native= false;
-  const uchar *extra2_field_flags= 0;
-  size_t extra2_field_flags_length= 0;
 
   MEM_ROOT *old_root= thd->mem_root;
   Virtual_column_info **table_check_constraints;
+  extra2_fields extra2;
+
   DBUG_ENTER("TABLE_SHARE::init_from_binary_frm_image");
 
   keyinfo= &first_keyinfo;
@@ -1438,90 +1543,27 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
 
   /* Length of the MariaDB extra2 segment in the form file. */
   len = uint2korr(frm_image+4);
-  extra2= frm_image + 64;
 
-  if (*extra2 != '/')   // old frm had '/' there
+  if (read_extra2(frm_image, len, &extra2))
-  {
+    goto err;
-    const uchar *e2end= extra2 + len;
+
-    while (extra2 + 3 <= e2end)
+  tabledef_version.length= extra2.version.length;
-    {
+  tabledef_version.str= (uchar*)memdup_root(&mem_root, extra2.version.str,
-      uchar type= *extra2++;
+                                                       extra2.version.length);
-      size_t length= *extra2++;
+  if (!tabledef_version.str)
-      if (!length)
+    goto err;
-      {
+
-        if (extra2 + 2 >= e2end)
+  /* remember but delay parsing until we have read fields and keys */
-          goto err;
+  options= extra2.options;
-        length= uint2korr(extra2);
+
-        extra2+= 2;
-        if (length < 256)
-          goto err;
-      }
-      if (extra2 + length > e2end)
-        goto err;
-      switch (type) {
-      case EXTRA2_TABLEDEF_VERSION:
-        if (tabledef_version.str) // see init_from_sql_statement_string()
-        {
-          if (length != tabledef_version.length ||
-              memcmp(extra2, tabledef_version.str, length))
-            goto err;
-        }
-        else
-        {
-          tabledef_version.length= length;
-          tabledef_version.str= (uchar*)memdup_root(&mem_root, extra2, length);
-          if (!tabledef_version.str)
-            goto err;
-        }
-        break;
-      case EXTRA2_ENGINE_TABLEOPTS:
-        if (options)
-          goto err;
-        /* remember but delay parsing until we have read fields and keys */
-        options= extra2;
-        options_len= length;
-        break;
-      case EXTRA2_DEFAULT_PART_ENGINE:
 #ifdef WITH_PARTITION_STORAGE_ENGINE
-        {
+  if (extra2.engine)
-          LEX_CSTRING name= { (char*)extra2, length };
+  {
-          share->default_part_plugin= ha_resolve_by_name(NULL, &name, false);
+    share->default_part_plugin= ha_resolve_by_name(NULL, &extra2.engine, false);
-          if (!share->default_part_plugin)
+    if (!share->default_part_plugin)
-            goto err;
-        }
-#endif
-        break;
-      case EXTRA2_GIS:
-#ifdef HAVE_SPATIAL
-        {
-          if (gis_options.str)
-            goto err;
-          gis_options.str= extra2;
-          gis_options.length= length;
-        }
-#endif /*HAVE_SPATIAL*/
-        break;
-      case EXTRA2_PERIOD_FOR_SYSTEM_TIME:
-        if (system_period || length != 2 * sizeof(uint16))
-          goto err;
-        system_period = extra2;
-        break;
-      case EXTRA2_FIELD_FLAGS:
-        if (extra2_field_flags)
-          goto err;
-        extra2_field_flags= extra2;
-        extra2_field_flags_length= length;
-        break;
-      default:
-        /* abort frm parsing if it's an unknown but important extra2 value */
-        if (type >= EXTRA2_ENGINE_IMPORTANT)
-          goto err;
-      }
-      extra2+= length;
-    }
-    if (extra2 != e2end)
       goto err;
   }
+#endif
 
   if (frm_length < FRM_HEADER_SIZE + len ||
       !(pos= uint4korr(frm_image + FRM_HEADER_SIZE + len)))
@@ -1798,11 +1840,11 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
 
     if (share->db_create_options & HA_OPTION_TEXT_CREATE_OPTIONS_legacy)
     {
-      if (options)
+      if (options.str)
         goto err;
-      options_len= uint4korr(next_chunk);
+      options.length= uint4korr(next_chunk);
-      options= next_chunk + 4;
+      options.str= next_chunk + 4;
-      next_chunk+= options_len + 4;
+      next_chunk+= options.length + 4;
     }
     DBUG_ASSERT(next_chunk <= buff_end);
   }
@@ -1830,7 +1872,7 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
 
   disk_buff= frm_image + pos + FRM_FORMINFO_SIZE;
   share->fields= uint2korr(forminfo+258);
-  if (extra2_field_flags && extra2_field_flags_length != share->fields)
+  if (extra2.field_flags.str && extra2.field_flags.length != share->fields)
     goto err;
   pos= uint2korr(forminfo+260);   /* Length of all screens */
   n_length= uint2korr(forminfo+268);
@@ -1965,27 +2007,36 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
   }
 
   /* Set system versioning information. */
-  if (system_period == NULL)
+  vers.name= Lex_ident(STRING_WITH_LEN("SYSTEM_TIME"));
+  if (extra2.system_period == NULL)
   {
     versioned= VERS_UNDEFINED;
-    row_start_field= 0;
+    vers.start_fieldno= 0;
-    row_end_field= 0;
+    vers.end_fieldno= 0;
   }
   else
   {
     DBUG_PRINT("info", ("Setting system versioning informations"));
-    uint16 row_start= uint2korr(system_period);
+    if (init_period_from_extra2(vers, extra2.system_period))
-    uint16 row_end= uint2korr(system_period + sizeof(uint16));
-    if (row_start >= share->fields || row_end >= share->fields)
       goto err;
-    DBUG_PRINT("info", ("Columns with system versioning: [%d, %d]", row_start, row_end));
+    DBUG_PRINT("info", ("Columns with system versioning: [%d, %d]",
+                        vers.start_fieldno, vers.end_fieldno));
     versioned= VERS_TIMESTAMP;
     vers_can_native= plugin_hton(se_plugin)->flags & HTON_NATIVE_SYS_VERSIONING;
-    row_start_field= row_start;
-    row_end_field= row_end;
     status_var_increment(thd->status_var.feature_system_versioning);
   } // if (system_period == NULL)
 
+  if (extra2.application_period.str)
+  {
+    period.name.length= extra2.application_period.length - 2 * frm_fieldno_size;
+    period.name.str= strmake_root(&mem_root,
+                                  (char*)extra2.application_period.str,
+                                  period.name.length);
+    const uchar *field_pos= extra2.application_period.str + period.name.length;
+    if (init_period_from_extra2(period, field_pos))
+      goto err;
+  }
+
   for (i=0 ; i < share->fields; i++, strpos+=field_pack_length, field_ptr++)
   {
     uint interval_nr= 0, recpos;
@@ -2067,7 +2118,7 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
           goto err; // Not supported field type
         if (handler->Column_definition_attributes_frm_unpack(&attr, share,
                                                              strpos,
-                                                             &gis_options))
+                                                             &extra2.gis))
           goto err;
       }
 
@@ -2178,9 +2229,9 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
 
     if (versioned)
     {
-      if (i == row_start_field)
+      if (i == vers.start_fieldno)
         flags|= VERS_SYS_START_FLAG;
-      else if (i == row_end_field)
+      else if (i == vers.end_fieldno)
         flags|= VERS_SYS_END_FLAG;
 
       if (flags & VERS_SYSTEM_FIELD)
@@ -2229,9 +2280,9 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
     reg_field->comment=comment;
     reg_field->vcol_info= vcol_info;
     reg_field->flags|= flags;
-    if (extra2_field_flags)
+    if (extra2.field_flags.str)
     {
-      uchar flags= *extra2_field_flags++;
+      uchar flags= *extra2.field_flags.str++;
       if (flags & VERS_OPTIMIZED_UPDATE)
         reg_field->flags|= VERS_UPDATE_UNVERSIONED_FLAG;
 
@@ -2740,10 +2791,10 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
               (uint) (share->table_check_constraints -
                       share->field_check_constraints));
 
-  if (options)
+  if (options.str)
   {
-    DBUG_ASSERT(options_len);
+    DBUG_ASSERT(options.length);
-    if (engine_table_options_frm_read(options, options_len, share))
+    if (engine_table_options_frm_read(options.str, options.length, share))
       goto err;
   }
   if (parse_engine_table_options(thd, handler_file->partition_ht(), share))
@@ -6631,9 +6682,9 @@ void TABLE::mark_columns_needed_for_delete()
 
   if (s->versioned)
   {
-    bitmap_set_bit(read_set, s->vers_start_field()->field_index);
+    bitmap_set_bit(read_set, s->vers.start_field(s)->field_index);
-    bitmap_set_bit(read_set, s->vers_end_field()->field_index);
+    bitmap_set_bit(read_set, s->vers.end_field(s)->field_index);
-    bitmap_set_bit(write_set, s->vers_end_field()->field_index);
+    bitmap_set_bit(write_set, s->vers.end_field(s)->field_index);
   }
 }
 

sql/table.h

@@ -768,20 +768,36 @@ struct TABLE_SHARE
 
   /**
     System versioning support.
-   */
+  */
+  struct period_info_t
+  {
+    uint16 start_fieldno;
+    uint16 end_fieldno;
+    Lex_ident name;
+    Field *start_field(TABLE_SHARE *s) const
+    {
+      return s->field[start_fieldno];
+    }
+    Field *end_field(TABLE_SHARE *s) const
+    {
+      return s->field[end_fieldno];
+    }
+  };
 
   vers_sys_type_t versioned;
-  uint16 row_start_field;
+  period_info_t vers;
-  uint16 row_end_field;
+  period_info_t period;
+
+  bool init_period_from_extra2(period_info_t &period, const uchar *data);
 
   Field *vers_start_field()
   {
-    return field[row_start_field];
+    return field[vers.start_fieldno];
   }
 
   Field *vers_end_field()
   {
-    return field[row_end_field];
+    return field[vers.end_fieldno];
   }
 
   /**
@@ -1546,13 +1562,13 @@ public:
   Field *vers_start_field() const
   {
     DBUG_ASSERT(s && s->versioned);
-    return field[s->row_start_field];
+    return field[s->vers.start_fieldno];
   }
 
   Field *vers_end_field() const
   {
     DBUG_ASSERT(s && s->versioned);
-    return field[s->row_end_field];
+    return field[s->vers.end_fieldno];
   }
 
   ulonglong vers_start_id() const;
@@ -1758,6 +1774,9 @@ class IS_table_read_plan;
 /** The threshold size a blob field buffer before it is freed */
 #define MAX_TDC_BLOB_SIZE 65536
 
+/** number of bytes used by field positional indexes in frm */
+constexpr uint frm_fieldno_size= 2;
+
 class select_unit;
 class TMP_TABLE_PARAM;
 
@@ -1868,6 +1887,8 @@ struct vers_select_conds_t
   Vers_history_point start;
   Vers_history_point end;
 
+  const TABLE_SHARE::period_info_t *period;
+
   void empty()
   {
     type= SYSTEM_TIME_UNSPECIFIED;

sql/unireg.cc

@@ -106,25 +106,19 @@ static uchar *extra2_write_field_properties(uchar *pos,
   return pos;
 }
 
-static const bool ROW_START = true;
+static
-static const bool ROW_END = false;
-
-static inline
 uint16
-vers_get_field(HA_CREATE_INFO *create_info, List<Create_field> &create_fields, bool row_start)
+get_fieldno_by_name(HA_CREATE_INFO *create_info, List<Create_field> &create_fields,
+                    const Lex_ident field_name)
 {
-  DBUG_ASSERT(create_info->versioned());
-
   List_iterator<Create_field> it(create_fields);
   Create_field *sql_field = NULL;
 
-  const Lex_ident row_field= row_start ? create_info->vers_info.as_row.start
+  DBUG_ASSERT(field_name);
-                                   : create_info->vers_info.as_row.end;
-  DBUG_ASSERT(row_field);
 
   for (unsigned field_no = 0; (sql_field = it++); ++field_no)
   {
-    if (row_field.streq(sql_field->field_name))
+    if (field_name.streq(sql_field->field_name))
     {
       DBUG_ASSERT(field_no <= uint16(~0U));
       return uint16(field_no);
@@ -176,6 +170,8 @@ LEX_CUSTRING build_frm_image(THD *thd, const LEX_CSTRING *table,
   ulong data_offset;
   uint options_len;
   uint gis_extra2_len= 0;
+  uint period_info_len= create_info->period_info.name.length
+                        + 2 * frm_fieldno_size;
   uchar fileinfo[FRM_HEADER_SIZE],forminfo[FRM_FORMINFO_SIZE];
   const partition_info *part_info= IF_PARTITIONING(thd->work_part_info, 0);
   bool error;
@@ -287,6 +283,11 @@ LEX_CUSTRING build_frm_image(THD *thd, const LEX_CSTRING *table,
     extra2_size+= 1 + 1 + 2 * sizeof(uint16);
   }
 
+  if (create_info->period_info.name)
+  {
+    extra2_size+= 1 + (period_info_len > 255 ? 3 : 1) + period_info_len;
+  }
+
   bool has_extra2_field_flags_= has_extra2_field_flags(create_fields);
   if (has_extra2_field_flags_)
   {
@@ -349,13 +350,33 @@ LEX_CUSTRING build_frm_image(THD *thd, const LEX_CSTRING *table,
   }
 #endif /*HAVE_SPATIAL*/
 
+  // PERIOD
+  if (create_info->period_info.name)
+  {
+    *pos++= EXTRA2_APPLICATION_TIME_PERIOD;
+
+    Lex_ident &period_name= create_info->period_info.name;
+    pos= extra2_write_len(pos, period_info_len);
+    memcpy(pos, period_name.str, period_name.length);
+    pos+= period_name.length;
+
+    int2store(pos, get_fieldno_by_name(create_info, create_fields,
+                                       create_info->period_info.period.start));
+    pos+= frm_fieldno_size;
+    int2store(pos, get_fieldno_by_name(create_info, create_fields,
+                                       create_info->period_info.period.end));
+    pos+= frm_fieldno_size;
+  }
+
   if (create_info->versioned())
   {
     *pos++= EXTRA2_PERIOD_FOR_SYSTEM_TIME;
     *pos++= 2 * sizeof(uint16);
-    int2store(pos, vers_get_field(create_info, create_fields, ROW_START));
+    int2store(pos, get_fieldno_by_name(create_info, create_fields,
+                                       create_info->vers_info.as_row.start));
     pos+= sizeof(uint16);
-    int2store(pos, vers_get_field(create_info, create_fields, ROW_END));
+    int2store(pos, get_fieldno_by_name(create_info, create_fields,
+                                       create_info->vers_info.as_row.end));
     pos+= sizeof(uint16);
   }
 

sql/unireg.h

@@ -171,6 +171,7 @@ enum extra2_frm_value_type {
   EXTRA2_DEFAULT_PART_ENGINE=1,
   EXTRA2_GIS=2,
   EXTRA2_PERIOD_FOR_SYSTEM_TIME=4,
+  EXTRA2_APPLICATION_TIME_PERIOD=8,
 
 #define EXTRA2_ENGINE_IMPORTANT 128
 

storage/innobase/handler/ha_innodb.cc

@@ -10959,9 +10959,9 @@ create_table_info_t::create_table_def()
 		ulint vers_row = 0;
 
 		if (m_form->versioned()) {
-			if (i == m_form->s->row_start_field) {
+			if (i == m_form->s->vers.start_fieldno) {
 				vers_row = DATA_VERS_START;
-			} else if (i == m_form->s->row_end_field) {
+			} else if (i == m_form->s->vers.end_fieldno) {
 				vers_row = DATA_VERS_END;
 			} else if (!(field->flags
 				     & VERS_UPDATE_UNVERSIONED_FLAG)) {

storage/innobase/handler/handler0alter.cc

@@ -6233,10 +6233,10 @@ new_clustered_failed:
 			}
 
 			if (altered_table->versioned()) {
-				if (i == altered_table->s->row_start_field) {
+				if (i == altered_table->s->vers.start_fieldno) {
 					field_type |= DATA_VERS_START;
 				} else if (i ==
-					   altered_table->s->row_end_field) {
+					   altered_table->s->vers.end_fieldno) {
 					field_type |= DATA_VERS_END;
 				} else if (!(field->flags
 					     & VERS_UPDATE_UNVERSIONED_FLAG)) {
@@ -9007,9 +9007,9 @@ static void get_type(const Field& f, ulint& prtype, ulint& mtype, ulint& len)
 	if (!f.real_maybe_null()) prtype |= DATA_NOT_NULL;
 	if (f.binary()) prtype |= DATA_BINARY_TYPE;
 	if (f.table->versioned()) {
-		if (&f == f.table->field[f.table->s->row_start_field]) {
+		if (&f == f.table->field[f.table->s->vers.start_fieldno]) {
 			prtype |= DATA_VERS_START;
-		} else if (&f == f.table->field[f.table->s->row_end_field]) {
+		} else if (&f == f.table->field[f.table->s->vers.end_fieldno]) {
 			prtype |= DATA_VERS_END;
 		} else if (!(f.flags & VERS_UPDATE_UNVERSIONED_FLAG)) {
 			prtype |= DATA_VERSIONED;