Timestamp-based versioning for InnoDB [closes #209]

* Removed integer_fields check
* Reworked Vers_parse_info::check_sys_fields()
* Misc renames
* versioned as vers_sys_type_t

* Removed versioned_by_sql(), versioned_by_engine()

versioned() works as before;
versioned(VERS_TIMESTAMP) is versioned_by_sql();
versioned(VERS_TRX_ID) is versioned_by_engine().

* create_tmp_table() fix
* Foreign constraints for timestamp-based
* Range auto-specifier fix
* SQL: 1-row partition rotation fix [fixes #260]
* Fix 'drop system versioning, algorithm=inplace'

26 files changed, 318 insertions, 246 deletions

diff --git a/sql/field.cc b/sql/field.cc
index d4120e22a1d..4014fca818f 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -2008,7 +2008,7 @@ bool Field_vers_trx_id::get_date(MYSQL_TIME *ltime, ulonglong fuzzydate, ulonglo
   DBUG_ASSERT(ltime);
   if (!table || !table->s)
     return true;
-  DBUG_ASSERT(table->versioned_by_engine() ||
+  DBUG_ASSERT(table->versioned(VERS_TRX_ID) ||
     (table->versioned() && table->s->table_category == TABLE_CATEGORY_TEMPORARY));
   if (!trx_id)
     return true;
diff --git a/sql/field.h b/sql/field.h
index 64294afc7bb..addd1e95f38 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -4659,14 +4659,14 @@ bool check_expression(Virtual_column_info *vcol, LEX_CSTRING *name,
 inline
 ulonglong TABLE::vers_end_id() const
 {
-  DBUG_ASSERT(versioned_by_engine());
+  DBUG_ASSERT(versioned(VERS_TRX_ID));
   return static_cast<ulonglong>(vers_end_field()->val_int());
 }
 
 inline
 ulonglong TABLE::vers_start_id() const
 {
-  DBUG_ASSERT(versioned_by_engine());
+  DBUG_ASSERT(versioned(VERS_TRX_ID));
   return static_cast<ulonglong>(vers_start_field()->val_int());
 }
 
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index faf61c26c66..da274bbdf67 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -2118,7 +2118,7 @@ void ha_partition::update_create_info(HA_CREATE_INFO *create_info)
     HA_STATUS_AUTO is optimized so it will not always be forwarded
     to all partitions, but HA_STATUS_VARIABLE will.
   */
-  info(HA_STATUS_VARIABLE);
+  info(HA_STATUS_VARIABLE | HA_STATUS_OPEN);
 
   info(HA_STATUS_AUTO);
 
@@ -3626,7 +3626,7 @@ int ha_partition::open(const char *name, int mode, uint test_if_locked)
                             m_part_info->part_expr->get_monotonicity_info();
   else if (m_part_info->list_of_part_fields)
     m_part_func_monotonicity_info= MONOTONIC_STRICT_INCREASING;
-  info(HA_STATUS_VARIABLE | HA_STATUS_CONST);
+  info(HA_STATUS_VARIABLE | HA_STATUS_CONST | HA_STATUS_OPEN);
   DBUG_RETURN(0);
 
 err_handler:
diff --git a/sql/ha_partition.h b/sql/ha_partition.h
index 0b9f1dc3953..209b9531469 100644
--- a/sql/ha_partition.h
+++ b/sql/ha_partition.h
@@ -1487,7 +1487,7 @@ public:
     {
       handler *file= m_file[part_id];
       DBUG_ASSERT(bitmap_is_set(&(m_part_info->read_partitions), part_id));
-      file->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK);
+      file->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK | HA_STATUS_OPEN);
       part_recs+= file->stats.records;
     }
     return part_recs;
diff --git a/sql/handler.cc b/sql/handler.cc
index 1c237e22015..6d275a4396f 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -4317,6 +4317,9 @@ handler::check_if_supported_inplace_alter(TABLE *altered_table,
 
   HA_CREATE_INFO *create_info= ha_alter_info->create_info;
 
+  if (altered_table->versioned(VERS_TIMESTAMP))
+    DBUG_RETURN(HA_ALTER_INPLACE_NOT_SUPPORTED);
+
   Alter_inplace_info::HA_ALTER_FLAGS inplace_offline_operations=
     Alter_inplace_info::ALTER_COLUMN_EQUAL_PACK_LENGTH |
     Alter_inplace_info::ALTER_COLUMN_NAME |
@@ -5777,7 +5780,7 @@ bool ha_show_status(THD *thd, handlerton *db_type, enum ha_stat_type stat)
 
 bool handler::check_table_binlog_row_based(bool binlog_row)
 {
-  if (table->versioned_by_engine())
+  if (table->versioned(VERS_TRX_ID))
     return false;
   if (unlikely((table->in_use->variables.sql_log_bin_off)))
     return 0;                            /* Called by partitioning engine */
@@ -6758,7 +6761,7 @@ bool Vers_parse_info::is_end(const Create_field &f) const
 }
 
 static Create_field *vers_init_sys_field(THD *thd, const char *field_name,
-                                         int flags, bool integer_fields)
+                                         int flags)
 {
   Create_field *f= new (thd->mem_root) Create_field();
   if (!f)
@@ -6769,17 +6772,8 @@ static Create_field *vers_init_sys_field(THD *thd, const char *field_name,
   f->field_name.length= strlen(field_name);
   f->charset= system_charset_info;
   f->flags= flags | VERS_HIDDEN_FLAG;
-  if (integer_fields)
-  {
-    f->set_handler(&type_handler_longlong);
-    f->flags|= UNSIGNED_FLAG;
-    f->length= MY_INT64_NUM_DECIMAL_DIGITS - 1;
-  }
-  else
-  {
-    f->set_handler(&type_handler_timestamp2);
-    f->length= MAX_DATETIME_PRECISION;
-  }
+  f->set_handler(&type_handler_timestamp2);
+  f->length= MAX_DATETIME_PRECISION;
 
   if (f->check(thd))
     return NULL;
@@ -6788,10 +6782,9 @@ static Create_field *vers_init_sys_field(THD *thd, const char *field_name,
 }
 
 static bool vers_create_sys_field(THD *thd, const char *field_name,
-                                  Alter_info *alter_info, int flags,
-                                  bool integer_fields)
+                                  Alter_info *alter_info, int flags)
 {
-  Create_field *f= vers_init_sys_field(thd, field_name, flags, integer_fields);
+  Create_field *f= vers_init_sys_field(thd, field_name, flags);
   if (!f)
     return true;
 
@@ -6803,9 +6796,9 @@ static bool vers_create_sys_field(THD *thd, const char *field_name,
 
 static bool vers_change_sys_field(THD *thd, const char *field_name,
                                   Alter_info *alter_info, int flags,
-                                  bool integer_fields, const char *change)
+                                  const char *change)
 {
-  Create_field *f= vers_init_sys_field(thd, field_name, flags, integer_fields);
+  Create_field *f= vers_init_sys_field(thd, field_name, flags);
   if (!f)
     return true;
 
@@ -6821,8 +6814,7 @@ static bool vers_change_sys_field(THD *thd, const char *field_name,
 const LString Vers_parse_info::default_start= "sys_trx_start";
 const LString Vers_parse_info::default_end= "sys_trx_end";
 
-bool Vers_parse_info::fix_implicit(THD *thd, Alter_info *alter_info,
-                                   bool integer_fields, int *added)
+bool Vers_parse_info::fix_implicit(THD *thd, Alter_info *alter_info, int *added)
 {
   // If user specified some of these he must specify the others too. Do nothing.
   if (*this)
@@ -6833,8 +6825,8 @@ bool Vers_parse_info::fix_implicit(THD *thd, Alter_info *alter_info,
   system_time= start_end_t(default_start, default_end);
   as_row= system_time;
 
-  if (vers_create_sys_field(thd, default_start, alter_info, VERS_SYS_START_FLAG, integer_fields) ||
-      vers_create_sys_field(thd, default_end, alter_info, VERS_SYS_END_FLAG, integer_fields))
+  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))
   {
     return true;
   }
@@ -7008,9 +7000,8 @@ bool Table_scope_and_contents_source_st::vers_fix_system_fields(
     } // while (Create_field *f= it++)
   } // if (vers_tables)
 
-  bool integer_fields= vers_native(thd);
   int added= 0;
-  if (vers_info.fix_implicit(thd, alter_info, integer_fields, &added))
+  if (vers_info.fix_implicit(thd, alter_info, &added))
     return true;
 
   DBUG_ASSERT(added >= 0);
@@ -7045,12 +7036,18 @@ bool Table_scope_and_contents_source_st::vers_fix_system_fields(
     vers_cols == 0 &&
     (plain_cols == 0 || !vers_info))
   {
-    my_error(ER_VERS_NO_COLS_DEFINED, MYF(0), create_table.table_name);
+    my_error(ER_VERS_TABLE_MUST_HAVE_COLUMNS, MYF(0), create_table.table_name);
     return true;
   }
 
-  return vers_info.check_with_conditions(create_table.table_name) ||
-         vers_info.check_generated_type(create_table.table_name, alter_info, integer_fields);
+  if (vers_info.check_with_conditions(create_table.table_name))
+    return true;
+
+  bool native= vers_native(thd);
+  if (vers_info.check_sys_fields(create_table.table_name, alter_info, native))
+    return true;
+
+  return false;
 }
 
 static bool add_field_to_drop_list(THD *thd, Alter_info *alter_info,
@@ -7092,9 +7089,6 @@ bool Vers_parse_info::fix_alter_info(THD *thd, Alter_info *alter_info,
                                           TABLE *table)
 {
   TABLE_SHARE *share= table->s;
-  bool integer_fields= create_info->vers_native(thd);
-  if (create_info->db_type->db_type == DB_TYPE_PARTITION_DB && table->file->native_versioned())
-    integer_fields= true;
   const char *table_name= share->table_name.str;
 
   if (!need_check() && !share->versioned)
@@ -7268,9 +7262,7 @@ bool Vers_parse_info::fix_alter_info(THD *thd, Alter_info *alter_info,
         }
 
         if (vers_change_sys_field(thd, name, alter_info,
-                                  f->flags &
-                                      (VERS_SYS_START_FLAG | VERS_SYS_END_FLAG),
-                                  integer_fields, name))
+                                  f->flags & VERS_SYSTEM_FIELD, name))
         {
           return true;
         }
@@ -7285,10 +7277,19 @@ bool Vers_parse_info::fix_alter_info(THD *thd, Alter_info *alter_info,
     return false;
   }
 
-  return fix_implicit(thd, alter_info, integer_fields) ||
-         (with_system_versioning &&
-          (check_with_conditions(table_name) ||
-           check_generated_type(table_name, alter_info, integer_fields)));
+  if (fix_implicit(thd, alter_info))
+    return true;
+
+  if (with_system_versioning)
+  {
+    if (check_with_conditions(table_name))
+      return true;
+    bool native= create_info->vers_native(thd);
+    if (check_sys_fields(table_name, alter_info, native))
+      return true;
+  }
+
+  return false;
 }
 
 bool
@@ -7374,38 +7375,76 @@ bool Vers_parse_info::check_with_conditions(const char *table_name) const
   return false;
 }
 
-bool Vers_parse_info::check_generated_type(const char *table_name,
-                                           Alter_info *alter_info,
-                                           bool integer_fields) const
+bool Vers_parse_info::check_sys_fields(const char *table_name,
+                                       Alter_info *alter_info,
+                                       bool native) const
 {
   List_iterator<Create_field> it(alter_info->create_list);
+  vers_sys_type_t found= VERS_UNDEFINED;
+  uint found_flag= 0;
   while (Create_field *f= it++)
   {
-    if (is_start(*f) || is_end(*f))
+    vers_sys_type_t check_unit= VERS_UNDEFINED;
+    uint sys_flag= f->flags & VERS_SYSTEM_FIELD;
+
+    if (!sys_flag)
+      continue;
+
+    if (sys_flag & found_flag)
     {
-      if (integer_fields)
-      {
-        if (f->type_handler() != &type_handler_longlong || !(f->flags & UNSIGNED_FLAG) ||
-            f->length != (MY_INT64_NUM_DECIMAL_DIGITS - 1))
-        {
-          my_error(ER_VERS_FIELD_WRONG_TYPE, MYF(0), f->field_name.str,
-                   "BIGINT(20) UNSIGNED", table_name);
-          return true;
-        }
-      }
-      else
+      my_error(ER_VERS_DUPLICATE_ROW_START_END, MYF(0),
+                found_flag & VERS_SYS_START_FLAG ? "START" : "END", f->field_name.str);
+      return true;
+    }
+
+    sys_flag|= found_flag;
+
+    if ((f->type_handler() == &type_handler_datetime2 ||
+          f->type_handler() == &type_handler_timestamp2) &&
+        f->length == MAX_DATETIME_FULL_WIDTH)
+    {
+      check_unit= VERS_TIMESTAMP;
+    }
+    else if (native
+      && f->type_handler() == &type_handler_longlong
+      && (f->flags & UNSIGNED_FLAG)
+      && f->length == (MY_INT64_NUM_DECIMAL_DIGITS - 1))
+    {
+      check_unit= VERS_TRX_ID;
+    }
+    else
+    {
+      if (!found)
+        found= VERS_TIMESTAMP;
+      goto error;
+    }
+
+    if (check_unit)
+    {
+      if (found)
       {
-        if (!(f->type_handler() == &type_handler_datetime2 ||
-              f->type_handler() == &type_handler_timestamp2) ||
-            f->length != MAX_DATETIME_FULL_WIDTH)
+        if (found == check_unit)
         {
-          my_error(ER_VERS_FIELD_WRONG_TYPE, MYF(0), f->field_name.str,
-                   "TIMESTAMP(6)", table_name);
-          return true;
+          if (found == VERS_TRX_ID && !use_transaction_registry)
+          {
+            my_error(ER_VERS_TRT_IS_DISABLED, MYF(0));
+            return true;
+          }
+          return false;
         }
+      error:
+        my_error(ER_VERS_FIELD_WRONG_TYPE, MYF(0), f->field_name.str,
+                  found == VERS_TIMESTAMP ?
+                    "TIMESTAMP(6)" :
+                    "BIGINT(20) UNSIGNED",
+                  table_name);
+        return true;
       }
+      found= check_unit;
     }
   }
 
-  return false;
+  my_error(ER_MISSING, MYF(0), table_name, found_flag & VERS_SYS_START_FLAG ?
+    "ROW END" : found_flag ? "ROW START" : "ROW START/END");
+  return true;
 }
diff --git a/sql/handler.h b/sql/handler.h
index c29a10c393f..d73661b9a77 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -1757,7 +1757,7 @@ protected:
   bool is_end(const char *name) const;
   bool is_start(const Create_field &f) const;
   bool is_end(const Create_field &f) const;
-  bool fix_implicit(THD *thd, Alter_info *alter_info, bool integer_fields, int *added= NULL);
+  bool fix_implicit(THD *thd, Alter_info *alter_info, int *added= NULL);
   operator bool() const
   {
     return as_row.start || as_row.end || system_time.start || system_time.end;
@@ -1772,8 +1772,8 @@ protected:
       *this;
   }
   bool check_with_conditions(const char *table_name) const;
-  bool check_generated_type(const char *table_name, Alter_info *alter_info,
-                            bool integer_fields) const;
+  bool check_sys_fields(const char *table_name, Alter_info *alter_info,
+                        bool native) const;
 
 public:
   static const LString default_start;
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 397d1b6bef5..1256da729f0 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -12507,7 +12507,7 @@ Rows_log_event::write_row(rpl_group_info *rgi,
 
   // Handle INSERT.
   // Set vers fields when replicating from not system-versioned table.
-  if (m_type == WRITE_ROWS_EVENT_V1 && table->versioned_by_sql())
+  if (m_type == WRITE_ROWS_EVENT_V1 && table->versioned(VERS_TIMESTAMP))
   {
     ulong sec_part;
     bitmap_set_bit(table->read_set, table->vers_start_field()->field_index);
@@ -13442,7 +13442,7 @@ int Delete_rows_log_event::do_exec_row(rpl_group_info *rgi)
     if (!error)
     {
       m_table->mark_columns_per_binlog_row_image();
-      if (m_vers_from_plain && m_table->versioned_by_sql())
+      if (m_vers_from_plain && m_table->versioned(VERS_TIMESTAMP))
       {
         Field *end= m_table->vers_end_field();
         bitmap_set_bit(m_table->write_set, end->field_index);
@@ -13711,7 +13711,7 @@ Update_rows_log_event::do_exec_row(rpl_group_info *rgi)
   memcpy(m_table->write_set->bitmap, m_cols_ai.bitmap, (m_table->write_set->n_bits + 7) / 8);
 
   m_table->mark_columns_per_binlog_row_image();
-  if (m_vers_from_plain && m_table->versioned_by_sql())
+  if (m_vers_from_plain && m_table->versioned(VERS_TIMESTAMP))
   {
     bitmap_set_bit(m_table->write_set,
                    m_table->vers_start_field()->field_index);
@@ -13721,7 +13721,7 @@ Update_rows_log_event::do_exec_row(rpl_group_info *rgi)
   error= m_table->file->ha_update_row(m_table->record[1], m_table->record[0]);
   if (error == HA_ERR_RECORD_IS_THE_SAME)
     error= 0;
-  if (m_vers_from_plain && m_table->versioned_by_sql())
+  if (m_vers_from_plain && m_table->versioned(VERS_TIMESTAMP))
   {
     store_record(m_table, record[2]);
     error= vers_insert_history_row(m_table);
diff --git a/sql/log_event.h b/sql/log_event.h
index 2dedd8bbe9a..4d84bc34a47 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -4746,7 +4746,7 @@ public:
                                           __attribute__((unused)),
                                           const uchar *after_record)
   {
-    DBUG_ASSERT(!table->versioned_by_engine());
+    DBUG_ASSERT(!table->versioned(VERS_TRX_ID));
     return thd->binlog_write_row(table, is_transactional, after_record);
   }
 #endif
@@ -4828,7 +4828,7 @@ public:
                                           const uchar *before_record,
                                           const uchar *after_record)
   {
-    DBUG_ASSERT(!table->versioned_by_engine());
+    DBUG_ASSERT(!table->versioned(VERS_TRX_ID));
     return thd->binlog_update_row(table, is_transactional,
                                   before_record, after_record);
   }
@@ -4918,7 +4918,7 @@ public:
                                           const uchar *after_record
                                           __attribute__((unused)))
   {
-    DBUG_ASSERT(!table->versioned_by_engine());
+    DBUG_ASSERT(!table->versioned(VERS_TRX_ID));
     return thd->binlog_delete_row(table, is_transactional,
                                   before_record);
   }
diff --git a/sql/mysqld.h b/sql/mysqld.h
index aedf2850186..2463f569c94 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -181,22 +181,22 @@ extern const char *log_output_str;
 extern const char *log_backup_output_str;
 
 /* System Versioning begin */
-enum vers_range_type_t
+enum vers_system_time_t
 {
-  FOR_SYSTEM_TIME_UNSPECIFIED = 0,
-  FOR_SYSTEM_TIME_ALL,
-  FOR_SYSTEM_TIME_AS_OF,
-  FOR_SYSTEM_TIME_FROM_TO,
-  FOR_SYSTEM_TIME_BETWEEN,
-  FOR_SYSTEM_TIME_BEFORE
+  SYSTEM_TIME_UNSPECIFIED = 0,
+  SYSTEM_TIME_ALL,
+  SYSTEM_TIME_AS_OF,
+  SYSTEM_TIME_FROM_TO,
+  SYSTEM_TIME_BETWEEN,
+  SYSTEM_TIME_BEFORE
 };
 
-struct st_vers_asof_timestamp
+struct vers_asof_timestamp_t
 {
   ulong type;
   MYSQL_TIME ltime;
-  st_vers_asof_timestamp() :
-    type(FOR_SYSTEM_TIME_UNSPECIFIED)
+  vers_asof_timestamp_t() :
+    type(SYSTEM_TIME_UNSPECIFIED)
   {}
 };
 
diff --git a/sql/partition_info.cc b/sql/partition_info.cc
index 1f7ba06f875..31af3815cd5 100644
--- a/sql/partition_info.cc
+++ b/sql/partition_info.cc
@@ -953,7 +953,7 @@ bool partition_info::vers_setup_expression(THD * thd, uint32 alter_add)
 {
   DBUG_ASSERT(part_type == VERSIONING_PARTITION);
 
-  if (!table->versioned_by_sql())
+  if (!table->versioned(VERS_TIMESTAMP))
   {
     my_error(ER_VERS_ENGINE_UNSUPPORTED, MYF(0), table->s->table_name.str);
     return true;
@@ -1084,7 +1084,7 @@ bool partition_info::vers_scan_min_max(THD *thd, partition_element *part)
             part->partition_name);
           break;
         }
-        if (table->versioned_by_engine())
+        if (table->versioned(VERS_TRX_ID))
         {
           uchar buf[8];
           Field_timestampf fld(buf, NULL, 0, Field::NONE, &table->vers_end_field()->field_name, NULL, 6);
@@ -1261,8 +1261,6 @@ bool partition_info::vers_setup_stats(THD * thd, bool is_create_table_ind)
         table->s->stat_serial++;
 
       table->s->hist_part_id= vers_info->hist_part->id;
-      if (!is_create_table_ind && (vers_limit_exceed() || vers_interval_exceed()))
-        vers_part_rotate(thd);
     }
     mysql_mutex_lock(&table->s->LOCK_rotation);
     mysql_cond_broadcast(&table->s->COND_rotation);
diff --git a/sql/partition_info.h b/sql/partition_info.h
index 359eae6e8cb..1e291a94e3f 100644
--- a/sql/partition_info.h
+++ b/sql/partition_info.h
@@ -512,7 +512,7 @@ public:
     bool updated;
     mysql_rwlock_wrlock(&table->s->LOCK_stat_serial);
     el->empty= false;
-    if (table->versioned_by_engine())
+    if (table->versioned(VERS_TRX_ID))
     {
       // transaction is not yet pushed to VTQ, so we use now-time
       my_time_t end_ts= my_time_t(0);
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 3050b56a725..e71f7914a17 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -7816,22 +7816,22 @@ ER_INVISIBLE_NOT_NULL_WITHOUT_DEFAULT
 # MariaDB error numbers related to System Versioning
 
 ER_VERSIONING_REQUIRED
-        eng "System Versioning required: %s"
+        eng "System versioning required: %s"
 
 ER_UPDATE_INFO_WITH_SYSTEM_VERSIONING
         eng "Rows matched: %ld  Changed: %ld  Inserted: %ld  Warnings: %ld"
 
 ER_VERS_FIELD_WRONG_TYPE
-        eng "%`s must be of type %s for versioned table %`s"
+        eng "%`s must be of type %s for temporal table %`s"
 
 ER_VERS_ENGINE_UNSUPPORTED
-        eng "Engine does not support System Versioning for %`s"
+        eng "Transaction system versioning for %`s is not supported"
 
 ER_VERS_RANGE_UNITS_MISMATCH
         eng "Range units mismatch"
 
 ER_NON_VERSIONED_FIELD_IN_VERSIONED_QUERY
-        eng "Attempt to read unversioned field %`s in historical query"
+        eng "Attempt to read non-temporal field %`s in historical query"
 
 ER_PARTITION_WRONG_TYPE
         eng "Wrong partition type, expected type: %`s"
@@ -7852,10 +7852,10 @@ WARN_VERS_PART_NON_HISTORICAL
         eng "Partition %`s contains non-historical data"
 
 ER_VERS_ALTER_NOT_ALLOWED
-        eng "Not allowed for versioned %`s.%`s. Change @@system_versioning_alter_history to proceed with ALTER."
+        eng "Not allowed for temporal %`s.%`s. Change @@system_versioning_alter_history to proceed with ALTER."
 
 ER_VERS_ALTER_ENGINE_PROHIBITED
-        eng "Not allowed for versioned %`s.%`s. Change to/from native versioning engine is prohibited."
+        eng "Not allowed for temporal %`s.%`s. Change to/from native system versioning engine is prohibited."
 
 ER_VERS_RANGE_PROHIBITED
         eng "SYSTEM_TIME range selector is prohibited"
@@ -7881,11 +7881,11 @@ ER_VERS_VTMD_ERROR
 ER_VERS_DIFFERENT_TABLES
         eng "Wrong parameters for %`s: system fields selected from different tables"
 
-ER_VERS_NO_COLS_DEFINED
-        eng "Table %`s has no versioned columns"
+ER_VERS_TABLE_MUST_HAVE_COLUMNS
+        eng "Table %`s must have at least 1 temporal column"
 
 ER_VERS_NOT_VERSIONED
-        eng "Table %`s is not versioned"
+        eng "Table %`s is not temporal"
 
 ER_MISSING
         eng "Wrong parameters for %`s: missing '%s'"
@@ -7921,10 +7921,10 @@ ER_VERS_GENERATED_ALWAYS_NOT_EMPTY
         eng "Can not modify column %`s to GENERATED ALWAYS AS ROW START/END for non-empty table"
 
 ER_VERS_TRT_IS_DISABLED
-        eng "Some versioned DML requires `transaction_registry` to be set to ON."
+        eng "Temporal operation requires `mysql.transaction_registry` (@@system_versioning_transaction_registry)."
 
 ER_VERS_DUPLICATE_ROW_START_END
 	eng "Duplicate ROW %s column %`s"
 
 ER_VERS_ALREADY_VERSIONED
-        eng "Table %`s is already system-versioned table"
+        eng "Table %`s is already temporal"
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 34bd7513f63..d07e626ec6a 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -7640,7 +7640,7 @@ insert_fields(THD *thd, Name_resolution_context *context, const char *db_name,
         TABLE *table= f->field->table;
         DBUG_ASSERT(table && table->pos_in_table_list);
         TABLE_LIST *tl= table->pos_in_table_list;
-        vers_range_type_t vers_type= tl->vers_conditions.type;
+        vers_system_time_t vers_type= tl->vers_conditions.type;
 
         enum_sql_command sql_command= thd->lex->sql_command;
         unsigned int create_options= thd->lex->create_info.options;
@@ -7652,8 +7652,8 @@ insert_fields(THD *thd, Name_resolution_context *context, const char *db_name,
               ((fl & VERS_HIDDEN_FLAG) && (
                 vers_hide == VERS_HIDE_IMPLICIT ||
                   (vers_hide == VERS_HIDE_AUTO && (
-                    vers_type == FOR_SYSTEM_TIME_UNSPECIFIED ||
-                    vers_type == FOR_SYSTEM_TIME_AS_OF))))) :
+                    vers_type == SYSTEM_TIME_UNSPECIFIED ||
+                    vers_type == SYSTEM_TIME_AS_OF))))) :
             (fl & VERS_HIDDEN_FLAG))
         {
           if (sql_command != SQLCOM_CREATE_TABLE ||
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 30e428ab20d..41b8efc4464 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -155,7 +155,6 @@ extern bool volatile shutdown_in_progress;
 
 extern "C" LEX_STRING * thd_query_string (MYSQL_THD thd);
 extern "C" size_t thd_query_safe(MYSQL_THD thd, char *buf, size_t buflen);
-void thd_vers_update_trt(THD *thd, bool value);
 
 /**
   @class CSET_STRING
@@ -707,7 +706,7 @@ typedef struct system_variables
   uint column_compression_zlib_level;
   ulong in_subquery_conversion_threshold;
 
-  st_vers_asof_timestamp vers_asof_timestamp;
+  vers_asof_timestamp_t vers_asof_timestamp;
   my_bool vers_force;
   ulong vers_hide;
   my_bool vers_innodb_algorithm_simple;
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 118066e7d93..9c535e2124a 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -253,7 +253,7 @@ static bool record_should_be_deleted(THD *thd, TABLE *table, SQL_SELECT *sel,
 inline
 int TABLE::delete_row()
 {
-  if (!versioned_by_sql() || !vers_end_field()->is_max())
+  if (!versioned(VERS_TIMESTAMP) || !vers_end_field()->is_max())
     return file->ha_delete_row(record[0]);
 
   store_record(this, record[1]);
@@ -328,8 +328,8 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
       DBUG_RETURN(TRUE);
 
     // trx_sees() in InnoDB reads sys_trx_start
-    if (!table->versioned_by_sql()) {
-      DBUG_ASSERT(table_list->vers_conditions.type == FOR_SYSTEM_TIME_BEFORE);
+    if (!table->versioned(VERS_TIMESTAMP)) {
+      DBUG_ASSERT(table_list->vers_conditions.type == SYSTEM_TIME_BEFORE);
       bitmap_set_bit(table->read_set, table->vers_end_field()->field_index);
     }
   }
@@ -429,7 +429,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   if (!with_select && !using_limit && const_cond_result &&
       (!thd->is_current_stmt_binlog_format_row() &&
        !has_triggers)
-      && !table->versioned_by_sql())
+      && !table->versioned(VERS_TIMESTAMP))
   {
     /* Update the table->file->stats.records number */
     table->file->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK);
@@ -837,7 +837,7 @@ cleanup:
         errcode= query_error_code(thd, killed_status == NOT_KILLED);
 
       ScopedStatementReplication scoped_stmt_rpl(
-          table->versioned_by_engine() ? thd : NULL);
+          table->versioned(VERS_TRX_ID) ? thd : NULL);
       /*
         [binlog]: If 'handler::delete_all_rows()' was called and the
         storage engine does not inject the rows itself, we replicate
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index ab739733c63..b118cac09c2 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1161,7 +1161,7 @@ values_loop_end:
           errcode= query_error_code(thd, thd->killed == NOT_KILLED);
 
         ScopedStatementReplication scoped_stmt_rpl(
-            table->versioned_by_engine() ? thd : NULL);
+            table->versioned(VERS_TRX_ID) ? thd : NULL);
        /* bug#22725:
 
 	A query which per-row-loop can not be interrupted with
@@ -1582,7 +1582,7 @@ bool mysql_prepare_insert(THD *thd, TABLE_LIST *table_list,
   if (!table)
     table= table_list->table;
 
-  if (table->versioned_by_sql() && duplic == DUP_REPLACE)
+  if (table->versioned(VERS_TIMESTAMP) && duplic == DUP_REPLACE)
   {
     // Additional memory may be required to create historical items.
     if (table_list->set_insert_values(thd->mem_root))
@@ -1648,7 +1648,7 @@ static int last_uniq_key(TABLE *table,uint keynr)
 
 int vers_insert_history_row(TABLE *table)
 {
-  DBUG_ASSERT(table->versioned_by_sql());
+  DBUG_ASSERT(table->versioned(VERS_TIMESTAMP));
   restore_record(table,record[1]);
 
   // Set Sys_end to now()
@@ -1866,7 +1866,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
             info->updated++;
             if (table->versioned())
             {
-              if (table->versioned_by_sql())
+              if (table->versioned(VERS_TIMESTAMP))
               {
                 store_record(table, record[2]);
                 if ((error= vers_insert_history_row(table)))
@@ -1940,7 +1940,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
             !table->file->referenced_by_foreign_key() &&
             (!table->triggers || !table->triggers->has_delete_triggers()))
         {
-          if (table->versioned_by_engine())
+          if (table->versioned(VERS_TRX_ID))
           {
             bitmap_set_bit(table->write_set, table->vers_start_field()->field_index);
             table->vers_start_field()->set_notnull();
@@ -1953,7 +1953,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
           if (error != HA_ERR_RECORD_IS_THE_SAME)
           {
             info->deleted++;
-            if (table->versioned_by_sql())
+            if (table->versioned(VERS_TIMESTAMP))
             {
               store_record(table, record[2]);
               error= vers_insert_history_row(table);
@@ -1978,7 +1978,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
                                                 TRG_ACTION_BEFORE, TRUE))
             goto before_trg_err;
 
-          if (!table->versioned_by_sql())
+          if (!table->versioned(VERS_TIMESTAMP))
             error= table->file->ha_delete_row(table->record[1]);
           else
           {
@@ -1996,7 +1996,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
           }
           if (error)
             goto err;
-          if (!table->versioned_by_sql())
+          if (!table->versioned(VERS_TIMESTAMP))
             info->deleted++;
           else
             info->updated++;
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index e4662f2d52e..5beec82e063 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -3163,11 +3163,14 @@ int vers_get_partition_id(partition_info *part_info,
   DBUG_ASSERT(part_info);
   Field *sys_trx_end= part_info->part_field_array[STAT_TRX_END];
   DBUG_ASSERT(sys_trx_end);
-  DBUG_ASSERT(part_info->table);
+  TABLE *table= part_info->table;
+  DBUG_ASSERT(table);
   Vers_part_info *vers_info= part_info->vers_info;
-  DBUG_ASSERT(vers_info && vers_info->initialized());
-  DBUG_ASSERT(sys_trx_end->table == part_info->table && part_info->table->versioned());
-  DBUG_ASSERT(part_info->table->vers_end_field() == sys_trx_end);
+  DBUG_ASSERT(vers_info);
+  DBUG_ASSERT(vers_info->initialized());
+  DBUG_ASSERT(sys_trx_end->table == table);
+  DBUG_ASSERT(table->versioned());
+  DBUG_ASSERT(table->vers_end_field() == sys_trx_end);
 
   // new rows have NULL in sys_trx_end
   if (sys_trx_end->is_max() || sys_trx_end->is_null())
@@ -3177,7 +3180,6 @@ int vers_get_partition_id(partition_info *part_info,
   else // row is historical
   {
     THD *thd= current_thd;
-    TABLE *table= part_info->table;
 
     switch (thd->lex->sql_command)
     {
@@ -3198,7 +3200,7 @@ int vers_get_partition_id(partition_info *part_info,
         mysql_mutex_unlock(&table->s->LOCK_rotation);
         // transaction is not yet pushed to VTQ, so we use now-time
         ulong sec_part;
-        my_time_t end_ts= sys_trx_end->table->versioned_by_engine() ?
+        my_time_t end_ts= sys_trx_end->table->versioned(VERS_TRX_ID) ?
           my_time_t(0) : sys_trx_end->get_timestamp(&sec_part);
         if (part_info->vers_limit_exceed() || part_info->vers_interval_exceed(end_ts))
         {
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 95cae826015..c5d6954171e 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -673,13 +673,13 @@ setup_without_group(THD *thd, Ref_ptr_array ref_pointer_array,
 
 bool vers_select_conds_t::init_from_sysvar(THD *thd)
 {
-  st_vers_asof_timestamp &in= thd->variables.vers_asof_timestamp;
-  type= (vers_range_type_t) in.type;
-  unit_start= UNIT_TIMESTAMP;
+  vers_asof_timestamp_t &in= thd->variables.vers_asof_timestamp;
+  type= (vers_system_time_t) in.type;
+  unit_start= VERS_TIMESTAMP;
   from_query= false;
-  if (type != FOR_SYSTEM_TIME_UNSPECIFIED && type != FOR_SYSTEM_TIME_ALL)
+  if (type != SYSTEM_TIME_UNSPECIFIED && type != SYSTEM_TIME_ALL)
   {
-    DBUG_ASSERT(type == FOR_SYSTEM_TIME_AS_OF);
+    DBUG_ASSERT(type == SYSTEM_TIME_AS_OF);
     start= new (thd->mem_root)
         Item_datetime_literal(thd, &in.ltime, TIME_SECOND_PART_DIGITS);
     if (!start)
@@ -844,7 +844,7 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables, COND **where_expr
           DBUG_RETURN(-1);
         }
         else
-          vers_conditions.init(FOR_SYSTEM_TIME_ALL);
+          vers_conditions.init(SYSTEM_TIME_ALL);
       }
 #endif
 
@@ -886,7 +886,7 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables, COND **where_expr
         break;
       }
 
-      if (vers_conditions == FOR_SYSTEM_TIME_ALL)
+      if (vers_conditions == SYSTEM_TIME_ALL)
         continue;
     } // if (vers_conditions)
 
@@ -912,17 +912,17 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables, COND **where_expr
     bool tmp_from_ib=
         table->table->s->table_category == TABLE_CATEGORY_TEMPORARY &&
         table->table->vers_start_field()->type() == MYSQL_TYPE_LONGLONG;
-    bool timestamps_only= table->table->versioned_by_sql() && !tmp_from_ib;
+    bool timestamps_only= table->table->versioned(VERS_TIMESTAMP) && !tmp_from_ib;
 
     if (vers_conditions)
     {
       /* TODO: do resolve fix_length_and_dec(), fix_fields(). This requires
         storing vers_conditions as Item and make some magic related to
-        UNIT_TIMESTAMP/UNIT_TRX_ID at stage of fix_fields()
+        vers_system_time_t/VERS_TRX_ID at stage of fix_fields()
         (this is large refactoring). */
       vers_conditions.resolve_units(timestamps_only);
-      if (timestamps_only && (vers_conditions.unit_start == UNIT_TRX_ID ||
-        vers_conditions.unit_end == UNIT_TRX_ID))
+      if (timestamps_only && (vers_conditions.unit_start == VERS_TRX_ID ||
+        vers_conditions.unit_end == VERS_TRX_ID))
       {
         my_error(ER_VERS_ENGINE_UNSUPPORTED, MYF(0), table->table_name);
         DBUG_RETURN(-1);
@@ -934,54 +934,62 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables, COND **where_expr
     // have uint64 type of sys_trx_(start|end) field.
     // They need special handling.
     TABLE *t= table->table;
-    if (tmp_from_ib || t->versioned_by_sql() ||
+    if (tmp_from_ib || t->versioned(VERS_TIMESTAMP) ||
         thd->variables.vers_innodb_algorithm_simple)
     {
       if (vers_conditions)
       {
         if (vers_conditions.start)
         {
+          if (!vers_conditions.unit_start)
+            vers_conditions.unit_start= t->s->versioned;
           switch (vers_conditions.unit_start)
           {
-          case UNIT_TIMESTAMP:
+          case VERS_TIMESTAMP:
           {
             vers_conditions.start= newx Item_datetime_from_unixtime_typecast(
               thd, vers_conditions.start, 6);
             break;
           }
-          case UNIT_TRX_ID:
+          case VERS_TRX_ID:
           {
             vers_conditions.start= newx Item_longlong_typecast(
               thd, vers_conditions.start);
             break;
           }
-          default:;
+          default:
+            DBUG_ASSERT(0);
+            break;
           }
         }
 
         if (vers_conditions.end)
         {
+          if (!vers_conditions.unit_end)
+            vers_conditions.unit_end= t->s->versioned;
           switch (vers_conditions.unit_end)
           {
-          case UNIT_TIMESTAMP:
+          case VERS_TIMESTAMP:
           {
             vers_conditions.end= newx Item_datetime_from_unixtime_typecast(
               thd, vers_conditions.end, 6);
             break;
           }
-          case UNIT_TRX_ID:
+          case VERS_TRX_ID:
           {
             vers_conditions.end= newx Item_longlong_typecast(
               thd, vers_conditions.end);
             break;
           }
-          default:;
+          default:
+            DBUG_ASSERT(0);
+            break;
           }
         }
       }
       switch (vers_conditions.type)
       {
-      case FOR_SYSTEM_TIME_UNSPECIFIED:
+      case SYSTEM_TIME_UNSPECIFIED:
         if (t->vers_start_field()->real_type() != MYSQL_TYPE_LONGLONG)
         {
           MYSQL_TIME max_time;
@@ -997,25 +1005,25 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables, COND **where_expr
           cond1= newx Item_func_eq(thd, row_end, curr);
         }
         break;
-      case FOR_SYSTEM_TIME_AS_OF:
+      case SYSTEM_TIME_AS_OF:
         cond1= newx Item_func_le(thd, row_start,
           vers_conditions.start);
         cond2= newx Item_func_gt(thd, row_end,
           vers_conditions.start);
         break;
-      case FOR_SYSTEM_TIME_FROM_TO:
+      case SYSTEM_TIME_FROM_TO:
         cond1= newx Item_func_lt(thd, row_start,
           vers_conditions.end);
         cond2= newx Item_func_ge(thd, row_end,
           vers_conditions.start);
         break;
-      case FOR_SYSTEM_TIME_BETWEEN:
+      case SYSTEM_TIME_BETWEEN:
         cond1= newx Item_func_le(thd, row_start,
           vers_conditions.end);
         cond2= newx Item_func_ge(thd, row_end,
           vers_conditions.start);
         break;
-      case FOR_SYSTEM_TIME_BEFORE:
+      case SYSTEM_TIME_BEFORE:
         cond1= newx Item_func_lt(thd, row_end,
           vers_conditions.start);
         break;
@@ -1031,32 +1039,32 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables, COND **where_expr
 
       switch (vers_conditions.type)
       {
-      case FOR_SYSTEM_TIME_UNSPECIFIED:
+      case SYSTEM_TIME_UNSPECIFIED:
         curr= newx Item_int(thd, ULONGLONG_MAX);
         cond1= newx Item_func_eq(thd, row_end, curr);
         break;
-      case FOR_SYSTEM_TIME_AS_OF:
-        trx_id0= vers_conditions.unit_start == UNIT_TIMESTAMP ?
+      case SYSTEM_TIME_AS_OF:
+        trx_id0= vers_conditions.unit_start == VERS_TIMESTAMP ?
           newx Item_func_vtq_id(thd, vers_conditions.start, TR_table::FLD_TRX_ID) :
           vers_conditions.start;
         cond1= newx Item_func_vtq_trx_sees_eq(thd, trx_id0, row_start);
         cond2= newx Item_func_vtq_trx_sees(thd, row_end, trx_id0);
         break;
-      case FOR_SYSTEM_TIME_FROM_TO:
-      case FOR_SYSTEM_TIME_BETWEEN:
-        trx_id0= vers_conditions.unit_start == UNIT_TIMESTAMP ?
+      case SYSTEM_TIME_FROM_TO:
+      case SYSTEM_TIME_BETWEEN:
+        trx_id0= vers_conditions.unit_start == VERS_TIMESTAMP ?
           newx Item_func_vtq_id(thd, vers_conditions.start, TR_table::FLD_TRX_ID, true) :
           vers_conditions.start;
-        trx_id1= vers_conditions.unit_end == UNIT_TIMESTAMP ?
+        trx_id1= vers_conditions.unit_end == VERS_TIMESTAMP ?
           newx Item_func_vtq_id(thd, vers_conditions.end, TR_table::FLD_TRX_ID, false) :
           vers_conditions.end;
-        cond1= vers_conditions.type == FOR_SYSTEM_TIME_FROM_TO ?
+        cond1= vers_conditions.type == SYSTEM_TIME_FROM_TO ?
           newx Item_func_vtq_trx_sees(thd, trx_id1, row_start) :
           newx Item_func_vtq_trx_sees_eq(thd, trx_id1, row_start);
         cond2= newx Item_func_vtq_trx_sees_eq(thd, row_end, trx_id0);
         break;
-      case FOR_SYSTEM_TIME_BEFORE:
-        trx_id0= vers_conditions.unit_start == UNIT_TIMESTAMP ?
+      case SYSTEM_TIME_BEFORE:
+        trx_id0= vers_conditions.unit_start == VERS_TIMESTAMP ?
           newx Item_func_vtq_id(thd, vers_conditions.start, TR_table::FLD_TRX_ID) :
           vers_conditions.start;
         cond1= newx Item_func_lt(thd, row_end, trx_id0);
@@ -1094,7 +1102,7 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables, COND **where_expr
     }
 
     if (outer_table)
-      outer_table->vers_conditions.type= FOR_SYSTEM_TIME_ALL;
+      outer_table->vers_conditions.type= SYSTEM_TIME_ALL;
   }
 
   DBUG_RETURN(0);
@@ -17457,6 +17465,8 @@ create_tmp_table(THD *thd, TMP_TABLE_PARAM *param, List<Item> &fields,
   Field **tmp_from_field=from_field;
   Field *sys_trx_start= NULL;
   Field *sys_trx_end= NULL;
+  vers_sys_type_t versioned= VERS_UNDEFINED;
+
   while ((item=li++))
   {
     Item::Type type= item->type();
@@ -17594,9 +17604,15 @@ create_tmp_table(THD *thd, TMP_TABLE_PARAM *param, List<Item> &fields,
           if (s->versioned)
           {
             if (field->flags & VERS_SYS_START_FLAG)
+            {
               sys_trx_start= new_field;
+              versioned= s->versioned;
+            }
             else if (field->flags & VERS_SYS_END_FLAG)
+            {
               sys_trx_end= new_field;
+              versioned= s->versioned;
+            }
           }
         }
       }
@@ -17604,9 +17620,16 @@ create_tmp_table(THD *thd, TMP_TABLE_PARAM *param, List<Item> &fields,
       {
         Item_type_holder *ith= (Item_type_holder*)item;
         if (ith->field_flags() & VERS_SYS_START_FLAG)
+        {
           sys_trx_start= new_field;
+          goto set_versioned;
+        }
         else if (ith->field_flags() & VERS_SYS_END_FLAG)
+        {
           sys_trx_end= new_field;
+        set_versioned:
+          versioned= ith->vers_trx_id() ? VERS_TRX_ID : VERS_TIMESTAMP;
+        }
       }
       if (type == Item::SUM_FUNC_ITEM)
       {
@@ -17691,9 +17714,10 @@ create_tmp_table(THD *thd, TMP_TABLE_PARAM *param, List<Item> &fields,
 
   if (sys_trx_start && sys_trx_end)
   {
+    DBUG_ASSERT(versioned);
     sys_trx_start->flags|= VERS_SYS_START_FLAG | VERS_HIDDEN_FLAG;
     sys_trx_end->flags|= VERS_SYS_END_FLAG | VERS_HIDDEN_FLAG;
-    share->versioned= true;
+    share->versioned= versioned;
     share->field= table->field;
     share->row_start_field= sys_trx_start->field_index;
     share->row_end_field= sys_trx_end->field_index;
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 32f5f546dcc..1296ac3ae2f 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -1372,7 +1372,7 @@ mysqld_show_create(THD *thd, TABLE_LIST *table_list)
   bool versioned= table_list->vers_conditions;
   if (versioned)
   {
-    DBUG_ASSERT(table_list->vers_conditions == FOR_SYSTEM_TIME_AS_OF);
+    DBUG_ASSERT(table_list->vers_conditions == SYSTEM_TIME_AS_OF);
     VTMD_table vtmd(*table_list);
     if (vtmd.setup_select(thd))
       goto exit;
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index a6686eec2a4..e7c43d08ea3 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -9048,7 +9048,7 @@ bool mysql_alter_table(THD *thd, const char *new_db, const char *new_name,
       DBUG_RETURN(true);
     }
 
-    if (table_list->table->versioned_by_engine() &&
+    if (table_list->table->versioned(VERS_TRX_ID) &&
         alter_info->requested_algorithm ==
             Alter_info::ALTER_TABLE_ALGORITHM_DEFAULT &&
         !table_list->table->s->partition_info_str)
@@ -9184,8 +9184,7 @@ bool mysql_alter_table(THD *thd, const char *new_db, const char *new_name,
   if (check_engine(thd, alter_ctx.new_db, alter_ctx.new_name, create_info))
     DBUG_RETURN(true);
 
-  if (create_info->vers_info.fix_alter_info(thd, alter_info, create_info,
-                                                 table))
+  if (create_info->vers_info.fix_alter_info(thd, alter_info, create_info, table))
   {
     DBUG_RETURN(true);
   }
@@ -9799,7 +9798,7 @@ bool mysql_alter_table(THD *thd, const char *new_db, const char *new_name,
                                  alter_info->keys_onoff,
                                  &alter_ctx))
     {
-      if (vers_survival_mod && new_versioned && table->versioned_by_sql())
+      if (vers_survival_mod && new_versioned && table->versioned(VERS_TIMESTAMP))
       {
         // Failure of this function may result in corruption of an original table.
         vers_reset_alter_copy(thd, table);
@@ -10384,7 +10383,7 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
       error= 1;
       break;
     }
-    if (keep_versioned && to->versioned_by_engine() &&
+    if (keep_versioned && to->versioned(VERS_TRX_ID) &&
         thd->variables.vers_alter_history != VERS_ALTER_HISTORY_SURVIVE)
     {
       to->vers_write= false;
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index d14d50267fd..509cb984f84 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -930,7 +930,7 @@ update_begin:
 
           if (has_vers_fields && table->versioned())
           {
-            if (table->versioned_by_sql())
+            if (table->versioned(VERS_TIMESTAMP))
             {
               store_record(table, record[2]);
               if ((error = vers_insert_history_row(table)))
@@ -1119,7 +1119,7 @@ update_end:
         errcode= query_error_code(thd, killed_status == NOT_KILLED);
 
       ScopedStatementReplication scoped_stmt_rpl(
-          table->versioned_by_engine() ? thd : NULL);
+          table->versioned(VERS_TRX_ID) ? thd : NULL);
 
       if (thd->binlog_query(THD::ROW_QUERY_TYPE,
                             thd->query(), thd->query_length(),
@@ -1145,7 +1145,7 @@ update_end:
   if (error < 0 && !thd->lex->analyze_stmt)
   {
     char buff[MYSQL_ERRMSG_SIZE];
-    if (!table->versioned_by_sql())
+    if (!table->versioned(VERS_TIMESTAMP))
       my_snprintf(buff, sizeof(buff), ER_THD(thd, ER_UPDATE_INFO), (ulong) found,
                   (ulong) updated,
                   (ulong) thd->get_stmt_da()->current_statement_warn_count());
@@ -2330,7 +2330,7 @@ int multi_update::send_data(List<Item> &not_used_values)
           }
           else if (has_vers_fields && table->versioned())
           {
-            if (table->versioned_by_sql())
+            if (table->versioned(VERS_TIMESTAMP))
             {
               store_record(table, record[2]);
               if (vers_insert_history_row(table))
@@ -2655,7 +2655,7 @@ int multi_update::do_updates()
 
           if (has_vers_fields && table->versioned())
           {
-            if (table->versioned_by_sql())
+            if (table->versioned(VERS_TIMESTAMP))
             {
               store_record(table, record[2]);
               if ((local_error= vers_insert_history_row(table)))
@@ -2788,7 +2788,7 @@ bool multi_update::send_eof()
       bool force_stmt= false;
       for (TABLE *table= all_tables->table; table; table= table->next)
       {
-        if (table->versioned_by_engine())
+        if (table->versioned(VERS_TRX_ID))
         {
           force_stmt= true;
           break;
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 1b9b72a5898..9f2059bafc1 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -740,8 +740,8 @@ bool LEX::set_bincmp(CHARSET_INFO *cs, bool bin)
   } while(0)
 
 
-void vers_select_conds_t::init(vers_range_type_t t, vers_range_unit_t u_start,
-                               Item *s, vers_range_unit_t u_end, Item *e)
+void vers_select_conds_t::init(vers_system_time_t t, vers_sys_type_t u_start,
+                               Item *s, vers_sys_type_t u_end, Item *e)
 {
   type= t;
   unit_start= u_start;
@@ -880,7 +880,7 @@ Virtual_column_info *add_virtual_expression(THD *thd, Item *expr)
   enum Window_frame::Frame_exclusion frame_exclusion;
   enum trigger_order_type trigger_action_order_type;
   DDL_options_st object_ddl_options;
-  enum vers_range_unit_t vers_range_unit;
+  enum vers_sys_type_t vers_range_unit;
   enum Column_definition::enum_column_versioning vers_column_versioning;
 }
 
@@ -9171,15 +9171,15 @@ select_options:
 opt_trans_or_timestamp:
           /* empty */
           {
-            $$ = UNIT_AUTO;
+            $$ = VERS_UNDEFINED;
           }
         | TRANSACTION_SYM
           {
-            $$ = UNIT_TRX_ID;
+            $$ = VERS_TRX_ID;
           }
         | TIMESTAMP
           {
-            $$ = UNIT_TIMESTAMP;
+            $$ = VERS_TIMESTAMP;
           }
         ;
 
@@ -9197,21 +9197,21 @@ opt_for_system_time_clause:
 system_time_expr:
           AS OF_SYM opt_trans_or_timestamp simple_expr
           {
-            Lex->vers_conditions.init(FOR_SYSTEM_TIME_AS_OF, $3, $4);
+            Lex->vers_conditions.init(SYSTEM_TIME_AS_OF, $3, $4);
           }
         | ALL
           {
-            Lex->vers_conditions.init(FOR_SYSTEM_TIME_ALL);
+            Lex->vers_conditions.init(SYSTEM_TIME_ALL);
           }
         | FROM opt_trans_or_timestamp simple_expr
           TO_SYM opt_trans_or_timestamp simple_expr
           {
-            Lex->vers_conditions.init(FOR_SYSTEM_TIME_FROM_TO, $2, $3, $5, $6);
+            Lex->vers_conditions.init(SYSTEM_TIME_FROM_TO, $2, $3, $5, $6);
           }
         | BETWEEN_SYM opt_trans_or_timestamp simple_expr
           AND_SYM opt_trans_or_timestamp simple_expr
           {
-            Lex->vers_conditions.init(FOR_SYSTEM_TIME_BETWEEN, $2, $3, $5, $6);
+            Lex->vers_conditions.init(SYSTEM_TIME_BETWEEN, $2, $3, $5, $6);
           }
         ;
 
@@ -13504,7 +13504,7 @@ truncate_end:
           opt_lock_wait_timeout
           | TO_SYM SYSTEM_TIME_SYM opt_trans_or_timestamp simple_expr
           {
-            Lex->vers_conditions.init(FOR_SYSTEM_TIME_BEFORE, $3, $4);
+            Lex->vers_conditions.init(SYSTEM_TIME_BEFORE, $3, $4);
             Lex->last_table()->vers_conditions= Lex->vers_conditions;
           }
         ;
@@ -13824,8 +13824,8 @@ show_param:
               MYSQL_YYABORT;
             lex->create_info.storage_media= HA_SM_DEFAULT;
 
-            if (lex->vers_conditions.type != FOR_SYSTEM_TIME_UNSPECIFIED &&
-                lex->vers_conditions.type != FOR_SYSTEM_TIME_AS_OF)
+            if (lex->vers_conditions.type != SYSTEM_TIME_UNSPECIFIED &&
+                lex->vers_conditions.type != SYSTEM_TIME_AS_OF)
             {
               my_yyabort_error((ER_VERS_RANGE_PROHIBITED, MYF(0)));
             }
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index d4971c0d8d5..02237921c11 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -391,7 +391,7 @@ const char *Sys_var_vers_asof::asof_keywords[]= {"DEFAULT", NULL};
 static Sys_var_vers_asof Sys_vers_asof_timestamp(
        "system_versioning_asof", "Default value for the FOR SYSTEM_TIME AS OF clause",
        SESSION_VAR(vers_asof_timestamp.type), NO_CMD_LINE,
-       Sys_var_vers_asof::asof_keywords, DEFAULT(FOR_SYSTEM_TIME_UNSPECIFIED));
+       Sys_var_vers_asof::asof_keywords, DEFAULT(SYSTEM_TIME_UNSPECIFIED));
 
 static Sys_var_mybool Sys_vers_force(
        "system_versioning_force", "Force system versioning for all created tables",
diff --git a/sql/sys_vars.ic b/sql/sys_vars.ic
index a2ea75d0dc2..200669a5bf5 100644
--- a/sql/sys_vars.ic
+++ b/sql/sys_vars.ic
@@ -2628,17 +2628,17 @@ public:
     bool res= var->value->get_date(&ltime, 0);
     if (!res)
     {
-      var->save_result.ulonglong_value= FOR_SYSTEM_TIME_AS_OF;
+      var->save_result.ulonglong_value= SYSTEM_TIME_AS_OF;
     }
     return res;
   }
 
 private:
-  bool update(set_var *var, st_vers_asof_timestamp &out)
+  bool update(set_var *var, vers_asof_timestamp_t &out)
   {
     bool res= false;
     out.type= static_cast<enum_var_type>(var->save_result.ulonglong_value);
-    if (out.type == FOR_SYSTEM_TIME_AS_OF)
+    if (out.type == SYSTEM_TIME_AS_OF)
     {
       res= var->value->get_date(&out.ltime, 0);
     }
@@ -2648,27 +2648,28 @@ private:
 public:
   virtual bool global_update(THD *thd, set_var *var)
   {
-    return update(var, global_var(st_vers_asof_timestamp));
+    return update(var, global_var(vers_asof_timestamp_t));
   }
   virtual bool session_update(THD *thd, set_var *var)
   {
-    return update(var, session_var(thd, st_vers_asof_timestamp));
+    return update(var, session_var(thd, vers_asof_timestamp_t));
   }
 
 private:
-  uchar *value_ptr(THD *thd, st_vers_asof_timestamp &val)
+  uchar *value_ptr(THD *thd, vers_asof_timestamp_t &val)
   {
     switch (val.type)
     {
-    case FOR_SYSTEM_TIME_UNSPECIFIED:
-    case FOR_SYSTEM_TIME_ALL:
+    case SYSTEM_TIME_UNSPECIFIED:
+    case SYSTEM_TIME_ALL:
       return (uchar*) thd->strdup(asof_keywords[val.type]);
-    case FOR_SYSTEM_TIME_AS_OF:
+    case SYSTEM_TIME_AS_OF:
     {
       uchar *buf= (uchar*) thd->alloc(MAX_DATE_STRING_REP_LENGTH);
       if (buf &&!my_datetime_to_str(&val.ltime, (char*) buf, 6))
       {
-        my_error(ER_WRONG_VALUE_FOR_VAR, MYF(0), "versioning_asof_timestamp", "NULL (wrong datetime)");
+        // TODO: figure out variable name
+        my_error(ER_WRONG_VALUE_FOR_VAR, MYF(0), "system_versioning_asof_timestamp", "NULL (wrong datetime)");
         return (uchar*) thd->strdup("Error: wrong datetime");
       }
       return buf;
@@ -2676,13 +2677,13 @@ private:
     default:
       break;
     }
-    my_error(ER_WRONG_VALUE_FOR_VAR, MYF(0), "versioning_asof_timestamp", "NULL (wrong range type)");
+    my_error(ER_WRONG_VALUE_FOR_VAR, MYF(0), "system_versioning_asof_timestamp", "NULL (wrong range type)");
     return (uchar*) thd->strdup("Error: wrong range type");
   }
 
 public:
   virtual uchar *session_value_ptr(THD *thd, const LEX_CSTRING *base)
-  { return value_ptr(thd, session_var(thd, st_vers_asof_timestamp)); }
+  { return value_ptr(thd, session_var(thd, vers_asof_timestamp_t)); }
   virtual uchar *global_value_ptr(THD *thd, const LEX_CSTRING *base)
-  { return value_ptr(thd, global_var(st_vers_asof_timestamp)); }
+  { return value_ptr(thd, global_var(vers_asof_timestamp_t)); }
 };
diff --git a/sql/table.cc b/sql/table.cc
index 23ecec81c0d..2601480c6f5 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -1198,6 +1198,7 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
   const uchar *system_period= 0;
   bool vtmd_used= false;
   share->vtmd= false;
+  bool vers_can_native= false;
   const uchar *extra2_field_flags= 0;
   size_t extra2_field_flags_length= 0;
 
@@ -1770,9 +1771,9 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
   /* Set system versioning information. */
   if (system_period == NULL)
   {
-    versioned= false;
-    row_start_field = 0;
-    row_end_field = 0;
+    versioned= VERS_UNDEFINED;
+    row_start_field= 0;
+    row_end_field= 0;
   }
   else
   {
@@ -1782,7 +1783,8 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
     if (row_start >= share->fields || row_end >= share->fields)
       goto err;
     DBUG_PRINT("info", ("Columns with system versioning: [%d, %d]", row_start, row_end));
-    versioned= true;
+    versioned= VERS_TIMESTAMP;
+    vers_can_native= plugin_hton(se_plugin)->flags & HTON_NATIVE_SYS_VERSIONING;
     vers_init();
     row_start_field= row_start;
     row_end_field= row_end;
@@ -2018,6 +2020,27 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
         flags|= VERS_SYS_START_FLAG;
       else if (i == row_end_field)
         flags|= VERS_SYS_END_FLAG;
+
+      if (flags & VERS_SYSTEM_FIELD)
+      {
+        switch (field_type)
+        {
+        case MYSQL_TYPE_TIMESTAMP2:
+        case MYSQL_TYPE_DATETIME2:
+          break;
+        case MYSQL_TYPE_LONGLONG:
+          if (vers_can_native)
+          {
+            versioned= VERS_TRX_ID;
+            break;
+          }
+        default:
+          my_error(ER_VERS_FIELD_WRONG_TYPE, MYF(0), fieldnames.type_names[i],
+            versioned == VERS_TIMESTAMP ? "TIMESTAMP(6)" : "BIGINT(20) UNSIGNED",
+            table_name.str);
+          goto err;
+        }
+      }
     }
 
     /* Convert pre-10.2.2 timestamps to use Field::default_value */
@@ -7753,7 +7776,7 @@ void TABLE::vers_update_fields()
   bitmap_set_bit(write_set, vers_start_field()->field_index);
   bitmap_set_bit(write_set, vers_end_field()->field_index);
 
-  if (versioned_by_sql())
+  if (versioned(VERS_TIMESTAMP))
   {
     if (!vers_write)
       return;
@@ -8883,25 +8906,25 @@ bool TR_table::check()
 
 void vers_select_conds_t::resolve_units(bool timestamps_only)
 {
-  DBUG_ASSERT(type != FOR_SYSTEM_TIME_UNSPECIFIED);
+  DBUG_ASSERT(type != SYSTEM_TIME_UNSPECIFIED);
   DBUG_ASSERT(start);
-  if (unit_start == UNIT_AUTO)
+  if (unit_start == VERS_UNDEFINED)
   {
     if (start->type() == Item::FIELD_ITEM)
-      unit_start= UNIT_TIMESTAMP;
+      unit_start= VERS_TIMESTAMP;
     else
       unit_start= (!timestamps_only && (start->result_type() == INT_RESULT ||
         start->result_type() == REAL_RESULT)) ?
-          UNIT_TRX_ID : UNIT_TIMESTAMP;
+          VERS_TRX_ID : VERS_TIMESTAMP;
   }
-  if (end && unit_end == UNIT_AUTO)
+  if (end && unit_end == VERS_UNDEFINED)
   {
     if (start->type() == Item::FIELD_ITEM)
-      unit_start= UNIT_TIMESTAMP;
+      unit_start= VERS_TIMESTAMP;
     else
       unit_end= (!timestamps_only && (end->result_type() == INT_RESULT ||
         end->result_type() == REAL_RESULT)) ?
-          UNIT_TRX_ID : UNIT_TIMESTAMP;
+          VERS_TRX_ID : VERS_TIMESTAMP;
   }
 }
 
diff --git a/sql/table.h b/sql/table.h
index 4fffae7e763..83bc9d4597f 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -574,6 +574,13 @@ struct TABLE_STATISTICS_CB
 
 class Vers_min_max_stats;
 
+enum vers_sys_type_t
+{
+  VERS_UNDEFINED= 0,
+  VERS_TIMESTAMP,
+  VERS_TRX_ID
+};
+
 #ifndef UINT32_MAX
 #define UINT32_MAX             (4294967295U)
 #endif
@@ -757,7 +764,7 @@ struct TABLE_SHARE
     System versioning support.
    */
 
-  bool versioned;
+  vers_sys_type_t versioned;
   bool vtmd;
   uint16 row_start_field;
   uint16 row_end_field;
@@ -1517,29 +1524,16 @@ public:
    */
   bool vers_write;
 
-  bool versioned() const
+  bool versioned(vers_sys_type_t type= VERS_UNDEFINED) const
   {
     DBUG_ASSERT(s);
-    return s->versioned;
+    return type ? s->versioned == type : s->versioned;
   }
 
-  bool versioned_write() const
+  bool versioned_write(vers_sys_type_t type= VERS_UNDEFINED) const
   {
     DBUG_ASSERT(versioned() || !vers_write);
-    return vers_write;
-  }
-
-  /* Versioned by SQL layer */
-  bool versioned_by_sql() const
-  {
-    DBUG_ASSERT(s && file);
-    return s->versioned && !file->native_versioned();
-  }
-
-  bool versioned_by_engine() const
-  {
-    DBUG_ASSERT(s && file);
-    return s->versioned && file->native_versioned();
+    return versioned(type) ? vers_write : false;
   }
 
   bool vers_vtmd() const
@@ -1861,53 +1855,46 @@ class Item_in_subselect;
   4) jtbm semi-join (jtbm_subselect != NULL)
 */
 
-enum vers_range_unit_t
-{
-  UNIT_AUTO = 0,
-  UNIT_TIMESTAMP,
-  UNIT_TRX_ID
-};
-
 /** last_leaf_for_name_resolutioning support. */
 struct vers_select_conds_t
 {
-  vers_range_type_t type;
-  vers_range_unit_t unit_start, unit_end;
+  vers_system_time_t type;
+  vers_sys_type_t unit_start, unit_end;
   bool from_query:1;
   Item *start, *end;
 
   void empty()
   {
-    type= FOR_SYSTEM_TIME_UNSPECIFIED;
-    unit_start= unit_end= UNIT_AUTO;
+    type= SYSTEM_TIME_UNSPECIFIED;
+    unit_start= unit_end= VERS_UNDEFINED;
     from_query= false;
     start= end= NULL;
   }
 
   Item *fix_dec(Item *item);
 
-  void init(vers_range_type_t t, vers_range_unit_t u_start= UNIT_AUTO,
-            Item * s= NULL, vers_range_unit_t u_end= UNIT_AUTO,
+  void init(vers_system_time_t t, vers_sys_type_t u_start= VERS_UNDEFINED,
+            Item * s= NULL, vers_sys_type_t u_end= VERS_UNDEFINED,
             Item * e= NULL);
 
   bool init_from_sysvar(THD *thd);
 
-  bool operator== (vers_range_type_t b)
+  bool operator== (vers_system_time_t b)
   {
     return type == b;
   }
-  bool operator!= (vers_range_type_t b)
+  bool operator!= (vers_system_time_t b)
   {
     return type != b;
   }
   operator bool() const
   {
-    return type != FOR_SYSTEM_TIME_UNSPECIFIED;
+    return type != SYSTEM_TIME_UNSPECIFIED;
   }
   void resolve_units(bool timestamps_only);
   bool user_defined() const
   {
-    return !from_query && type != FOR_SYSTEM_TIME_UNSPECIFIED;
+    return !from_query && type != SYSTEM_TIME_UNSPECIFIED;
   }
 };