IB: 0.2 part III

* versioned DML: INSERT, UPDATE, DELETE;
* general refactoring and fixes.

Warning: breaks 'insert' and 'update' tests since they require part IV.

23 files changed, 353 insertions, 142 deletions

diff --git a/sql/field.cc b/sql/field.cc
index cb91e93ec5e..604965be254 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -4356,6 +4356,20 @@ void Field_longlong::sql_type(String &res) const
   add_zerofill_and_unsigned(res);
 }
 
+bool Field_longlong::set_max()
+{
+  ASSERT_COLUMN_MARKED_FOR_WRITE_OR_COMPUTED;
+  int8store(ptr, ULONGLONG_MAX);
+  return FALSE;
+}
+
+bool Field_longlong::is_max()
+{
+  ASSERT_COLUMN_MARKED_FOR_READ;
+  ulonglong j;
+  j = sint8korr(ptr);
+  return j == ULONGLONG_MAX;
+}
 
 /*
   Floating-point numbers
@@ -5423,9 +5437,10 @@ void Field_timestampf::store_TIME(my_time_t timestamp, ulong sec_part)
   my_timestamp_to_binary(&tm, ptr, dec);
 }
 
-bool Field_timestampf::set_max_timestamp()
+bool Field_timestampf::set_max()
 {
-  DBUG_ENTER("Field_timestampf::set_max_timestamp");
+  DBUG_ENTER("Field_timestampf::set_max");
+  ASSERT_COLUMN_MARKED_FOR_WRITE_OR_COMPUTED;
 
   mi_int4store(ptr, 0x7fffffff);
   memset(ptr + 4, 0x0, value_length() - 4);
@@ -5433,9 +5448,10 @@ bool Field_timestampf::set_max_timestamp()
   DBUG_RETURN(FALSE);
 }
 
-bool Field_timestampf::is_max_timestamp()
+bool Field_timestampf::is_max()
 {
-  DBUG_ENTER("Field_timestampf::is_max_timestamp");
+  DBUG_ENTER("Field_timestampf::is_max");
+  ASSERT_COLUMN_MARKED_FOR_READ;
 
   DBUG_RETURN(mi_sint4korr(ptr) == 0x7fffffff);
 }
diff --git a/sql/field.h b/sql/field.h
index 3513a773ef4..4b6607fa099 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -677,17 +677,16 @@ public:
   { DBUG_ASSERT(0); }
 
   /**
-     Is used by System Versioning.
+     Used by System Versioning.
    */
-  virtual bool set_max_timestamp() {
-    return true;
-  }
+  virtual bool set_max()
+  { DBUG_ASSERT(0); return false; }
+
   /**
-     Is used by System Versioning.
+     Used by System Versioning.
    */
-  virtual bool is_max_timestamp() {
-    return false;
-  }
+  virtual bool is_max()
+  { DBUG_ASSERT(0); return false; }
 
   uchar		*ptr;			// Position to field in record
   /**
@@ -2173,6 +2172,9 @@ public:
   {
     return unpack_int64(to, from, from_end);
   }
+
+  bool set_max();
+  bool is_max();
 };
 
 
@@ -2582,8 +2584,8 @@ public:
   {
     return memcmp(a_ptr, b_ptr, pack_length());
   }
-  virtual bool set_max_timestamp();
-  virtual bool is_max_timestamp();
+  bool set_max();
+  bool is_max();
   void store_TIME(my_time_t timestamp, ulong sec_part);
   my_time_t get_timestamp(const uchar *pos, ulong *sec_part) const;
   uint size_of() const { return sizeof(*this); }
diff --git a/sql/handler.h b/sql/handler.h
index 986896cc48b..b50e1b82ed2 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -1385,6 +1385,11 @@ struct handlerton
    */
    int (*discover_table_structure)(handlerton *hton, THD* thd,
                                    TABLE_SHARE *share, HA_CREATE_INFO *info);
+
+   /*
+     Engine supports System Versioning
+   */
+   bool versioned();
 };
 
 
@@ -1432,6 +1437,7 @@ handlerton *ha_default_tmp_handlerton(THD *thd);
 */
 #define HTON_NO_BINLOG_ROW_OPT       (1 << 9)
 #define HTON_SUPPORTS_EXTENDED_KEYS  (1 <<10) //supports extended keys
+#define HTON_SUPPORTS_SYS_VERSIONING (1 << 11) //Engine supports System Versioning
 
 // MySQL compatibility. Unused.
 #define HTON_SUPPORTS_FOREIGN_KEYS   (1 << 0) //Foreign key constraint supported.
@@ -4485,4 +4491,10 @@ void print_keydup_error(TABLE *table, KEY *key, myf errflag);
 
 int del_global_index_stat(THD *thd, TABLE* table, KEY* key_info);
 int del_global_table_stat(THD *thd, LEX_STRING *db, LEX_STRING *table);
+
+inline
+bool handlerton::versioned()
+{
+  return flags & HTON_SUPPORTS_SYS_VERSIONING;
+}
 #endif /* HANDLER_INCLUDED */
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 380d992a89b..f3859b5ba4a 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -7531,3 +7531,9 @@ ER_SYS_START_FIELD_MUST_BE_TIMESTAMP
 
 ER_SYS_END_FIELD_MUST_BE_TIMESTAMP
         eng "System end field must be of type TIMESTAMP"
+
+ER_SYS_START_FIELD_MUST_BE_BIGINT
+        eng "System start field must be of type BIGINT UNSIGNED"
+
+ER_SYS_END_FIELD_MUST_BE_BIGINT
+        eng "System end field must be of type BIGINT UNSIGNED"
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 1af45c12331..89f1798d36d 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -7962,7 +7962,7 @@ fill_record(THD *thd, TABLE *table_arg, List<Item> &fields, List<Item> &values,
                           ER_THD(thd, ER_WARNING_NON_DEFAULT_VALUE_FOR_VIRTUAL_COLUMN),
                           rfield->field_name, table->s->table_name.str);
     }
-    if (table->versioned() && rfield->is_generated() &&
+    if (table->versioned_by_sql() && rfield->is_generated() &&
         !ignore_errors)
     {
       my_error(ER_GENERATED_FIELD_CANNOT_BE_SET_BY_USER, MYF(0));
@@ -8216,7 +8216,7 @@ fill_record(THD *thd, TABLE *table, Field **ptr, List<Item> &values,
       }
     }
 
-    if (table->versioned() && field->is_generated() &&
+    if (table->versioned_by_sql() && field->is_generated() &&
         !ignore_errors)
     {
       my_error(ER_GENERATED_FIELD_CANNOT_BE_SET_BY_USER, MYF(0));
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index b739197ca45..5af9b326eba 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -216,7 +216,7 @@ inline
 int TABLE::delete_row()
 {
   int error;
-  if (!versioned())
+  if (!versioned_by_sql())
     error= file->ha_delete_row(record[0]);
   else
   {
@@ -360,7 +360,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() &&
        !(table->triggers && table->triggers->has_delete_triggers()))
-      && !table->versioned())
+      && !table->versioned_by_sql())
   {
     /* Update the table->file->stats.records number */
     table->file->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK);
@@ -576,8 +576,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   while (!(error=info.read_record(&info)) && !thd->killed &&
 	 ! thd->is_error())
   {
-    if (table->versioned() &&
-        !table->vers_end_field()->is_max_timestamp())
+    if (table->versioned() && !table->vers_end_field()->is_max())
     {
       continue;
     }
@@ -1076,8 +1075,7 @@ int multi_delete::send_data(List<Item> &values)
     if (table->status & (STATUS_NULL_ROW | STATUS_DELETED))
       continue;
 
-    if (table->versioned() &&
-        !table->vers_end_field()->is_max_timestamp())
+    if (table->versioned() && !table->vers_end_field()->is_max())
     {
       continue;
     }
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 39a8c61c231..4a588ea0cc2 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -222,7 +222,7 @@ static int check_insert_fields(THD *thd, TABLE_LIST *table_list,
                table_list->view_db.str, table_list->view_name.str);
       DBUG_RETURN(-1);
     }
-    if (values.elements != table->user_fields())
+    if (values.elements != table->vers_user_fields())
     {
       my_error(ER_WRONG_VALUE_COUNT_ON_ROW, MYF(0), 1L);
       DBUG_RETURN(-1);
@@ -1029,7 +1029,7 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
         }
       }
 
-      if (table->versioned())
+      if (table->versioned_by_sql())
         table->vers_update_fields();
 
       if ((res= table_list->view_check_option(thd,
@@ -1558,7 +1558,7 @@ bool mysql_prepare_insert(THD *thd, TABLE_LIST *table_list,
   if (!table)
     table= table_list->table;
 
-  if (table->versioned() && duplic == DUP_REPLACE)
+  if (table->versioned_by_sql() && duplic == DUP_REPLACE)
   {
     // Additional memory may be required to create historical items.
     if (table_list->set_insert_values(thd->mem_root))
@@ -1622,22 +1622,16 @@ static int last_uniq_key(TABLE *table,uint keynr)
  sets Sys_end to now() and calls ha_write_row() .
 */
 
-int vers_insert_history_row(TABLE *table, ha_rows *inserted)
+int vers_insert_history_row(TABLE *table)
 {
-  DBUG_ASSERT(table->versioned());
+  DBUG_ASSERT(table->versioned_by_sql());
   restore_record(table,record[1]);
 
   // Set Sys_end to now()
   if (table->vers_end_field()->set_time())
-  {
-    return 1;
-  }
-
-  const int error= table->file->ha_write_row(table->record[0]);
-  if (!error)
-    ++*inserted;
+    DBUG_ASSERT(0);
 
-  return error;
+  return table->file->ha_write_row(table->record[0]);
 }
 
 /*
@@ -1846,9 +1840,20 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
           if (error != HA_ERR_RECORD_IS_THE_SAME)
           {
             info->updated++;
-            if (table->versioned() &&
-              (error=vers_insert_history_row(table, &info->copied)))
-              goto err;
+            if (table->versioned())
+            {
+              if (table->versioned_by_sql())
+              {
+                store_record(table, record[2]);
+                if ((error= vers_insert_history_row(table)))
+                {
+                  restore_record(table, record[2]);
+                  goto err;
+                }
+                restore_record(table, record[2]);
+              }
+              info->copied++;
+            }
           }
           else
             error= 0;
@@ -1907,7 +1912,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
         if (last_uniq_key(table,key_nr) &&
             !table->file->referenced_by_foreign_key() &&
             (!table->triggers || !table->triggers->has_delete_triggers()) &&
-            !table->versioned())
+            !table->versioned_by_sql())
         {
           if ((error=table->file->ha_update_row(table->record[1],
                                                 table->record[0])) &&
@@ -1931,7 +1936,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
                                                 TRG_ACTION_BEFORE, TRUE))
             goto before_trg_err;
 
-          if (!table->versioned())
+          if (!table->versioned_by_sql())
             error= table->file->ha_delete_row(table->record[1]);
           else
           {
@@ -1949,7 +1954,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
           }
           if (error)
             goto err;
-          if (!table->versioned())
+          if (!table->versioned_by_sql())
             info->deleted++;
           else
             info->updated++;
@@ -2040,7 +2045,9 @@ int check_that_all_fields_are_given_values(THD *thd, TABLE *entry, TABLE_LIST *t
   for (Field **field=entry->field ; *field ; field++)
   {
     if (!bitmap_is_set(write_set, (*field)->field_index) &&
-        has_no_default_value(thd, *field, table_list))
+        has_no_default_value(thd, *field, table_list) &&
+        !((*field)->flags & (GENERATED_ROW_START_FLAG | GENERATED_ROW_END_FLAG)) &&
+        ((*field)->real_type() != MYSQL_TYPE_ENUM))
       err=1;
   }
   return thd->abort_on_warning ? err : 0;
@@ -3795,7 +3802,7 @@ int select_insert::send_data(List<Item> &values)
     DBUG_RETURN(0);
 
   thd->count_cuted_fields= CHECK_FIELD_WARN;	// Calculate cuted fields
-  if (table->versioned())
+  if (table->versioned_by_sql())
     table->vers_update_fields();
   store_values(values);
   if (table->default_field && table->update_default_fields(0, info.ignore))
diff --git a/sql/sql_insert.h b/sql/sql_insert.h
index a8794e7414b..6efd680d188 100644
--- a/sql/sql_insert.h
+++ b/sql/sql_insert.h
@@ -37,7 +37,7 @@ void upgrade_lock_type_for_insert(THD *thd, thr_lock_type *lock_type,
                                   bool is_multi_insert);
 int check_that_all_fields_are_given_values(THD *thd, TABLE *entry,
                                            TABLE_LIST *table_list);
-int vers_insert_history_row(TABLE *table, ha_rows *inserted);
+int vers_insert_history_row(TABLE *table);
 int write_record(THD *thd, TABLE *table, COPY_INFO *info);
 void kill_delayed_threads(void);
 
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 17e36390146..94f8d78fa74 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -676,7 +676,7 @@ setup_for_system_time(THD *thd, TABLE_LIST *tables, COND **conds, SELECT_LEX *se
 
   for (table= tables; table; table= table->next_local)
   {
-    if (table->table && table->table->versioned())
+    if (table->table && table->table->versioned_by_sql())
       versioned_tables++;
     else if (table->system_versioning.type != FOR_SYSTEM_TIME_UNSPECIFIED)
     {
@@ -690,7 +690,7 @@ setup_for_system_time(THD *thd, TABLE_LIST *tables, COND **conds, SELECT_LEX *se
 
   for (table= tables; table; table= table->next_local)
   {
-    if (table->table && table->table->versioned())
+    if (table->table && table->table->versioned_by_sql())
     {
       Field *fstart= table->table->vers_start_field();
       Field *fend= table->table->vers_end_field();
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 084e98b143f..2bcf0199429 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -2224,7 +2224,7 @@ int show_create_table(THD *thd, TABLE_LIST *table_list, String *packet,
                           hton->index_options);
   }
 
-  if (table->versioned())
+  if (table->versioned_by_sql())
   {
     const Field *fs = table->vers_start_field();
     const Field *fe = table->vers_end_field();
@@ -2273,7 +2273,7 @@ int show_create_table(THD *thd, TABLE_LIST *table_list, String *packet,
     add_table_options(thd, table, create_info_arg,
                       table_list->schema_table != 0, 0, packet);
 
-  if (table->versioned())
+  if (table->versioned_by_sql())
   {
     packet->append(STRING_WITH_LEN(" WITH SYSTEM VERSIONING"));
   }
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index da01bd26873..acbbf559c63 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -744,7 +744,7 @@ int mysql_update(THD *thd,
 
   while (!(error=info.read_record(&info)) && !thd->killed)
   {
-    if (table->versioned() && !table->vers_end_field()->is_max_timestamp())
+    if (table->versioned() && !table->vers_end_field()->is_max())
     {
       continue;
     }
@@ -763,7 +763,7 @@ int mysql_update(THD *thd,
                                                TRG_EVENT_UPDATE))
         break; /* purecov: inspected */
 
-      if (table->versioned())
+      if (table->versioned_by_sql())
         table->vers_update_fields();
 
       found++;
@@ -837,11 +837,17 @@ int mysql_update(THD *thd,
 
           if (table->versioned())
           {
-            store_record(table, record[2]);
-            if ((error = vers_insert_history_row(table, &updated_sys_ver)))
-              break;
-
-            restore_record(table, record[2]);
+            if (table->versioned_by_sql())
+            {
+              store_record(table, record[2]);
+              if ((error = vers_insert_history_row(table)))
+              {
+                restore_record(table, record[2]);
+                break;
+              }
+              restore_record(table, record[2]);
+            }
+            updated_sys_ver++;
           }
         }
         else if (!ignore ||
@@ -1036,7 +1042,7 @@ int mysql_update(THD *thd,
   if (error < 0 && !thd->lex->analyze_stmt)
   {
     char buff[MYSQL_ERRMSG_SIZE];
-    if (!table->versioned())
+    if (!table->versioned_by_sql())
       my_snprintf(buff, sizeof(buff), ER_THD(thd, ER_UPDATE_INFO), (ulong) found,
                   (ulong) updated,
                   (ulong) thd->get_stmt_da()->current_statement_warn_count());
@@ -2134,7 +2140,7 @@ int multi_update::send_data(List<Item> &not_used_values)
     if (table->status & (STATUS_NULL_ROW | STATUS_UPDATED))
       continue;
 
-    if (table->versioned() && !table->vers_end_field()->is_max_timestamp())
+    if (table->versioned() && !table->vers_end_field()->is_max())
     {
       continue;
     }
@@ -2170,7 +2176,7 @@ int multi_update::send_data(List<Item> &not_used_values)
         if (table->default_field && table->update_default_fields(1, ignore))
           DBUG_RETURN(1);
 
-        if (table->versioned())
+        if (table->versioned_by_sql())
           table->vers_update_fields();
 
         if ((error= cur_table->view_check_option(thd, ignore)) !=
@@ -2222,20 +2228,18 @@ int multi_update::send_data(List<Item> &not_used_values)
           }
           else if (table->versioned())
           {
-            restore_record(table,record[1]);
-
-            // Set end time to now()
-            if (table->vers_end_field()->set_time())
-            {
-              error= 1;
-              break;
-            }
-
-            if ( (error= vers_insert_history_row(table, &updated_sys_ver)) )
+            if (table->versioned_by_sql())
             {
-              error= 1;
-              break;
+              store_record(table, record[2]);
+              if (vers_insert_history_row(table))
+              {
+                restore_record(table, record[2]);
+                error= 1;
+                break;
+              }
+              restore_record(table, record[2]);
             }
+            updated_sys_ver++;
           }
           /* non-transactional or transactional table got modified   */
           /* either multi_update class' flag is raised in its branch */
@@ -2507,7 +2511,7 @@ int multi_update::do_updates()
             goto err2;
           }
         }
-        if (table->versioned())
+        if (table->versioned_by_sql())
           table->vers_update_fields();
 
         if ((local_error=table->file->ha_update_row(table->record[1],
@@ -2527,19 +2531,18 @@ int multi_update::do_updates()
 
           if (table->versioned())
           {
-            restore_record(table,record[1]);
-
-            // Set end time to now()
-            if (table->vers_end_field()->set_time())
-            {
-              goto err2;
-            }
-
-            if ( (local_error= vers_insert_history_row(table, &updated_sys_ver)) )
+            if (table->versioned_by_sql())
             {
-              err_table = table;
-              goto err;
+              store_record(table, record[2]);
+              if ((local_error= vers_insert_history_row(table)))
+              {
+                restore_record(table, record[2]);
+                err_table = table;
+                goto err;
+              }
+              restore_record(table, record[2]);
             }
+            updated_sys_ver++;
           }
         }
         else
diff --git a/sql/table.cc b/sql/table.cc
index c9729d17d9c..80a7046be1a 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -2477,35 +2477,6 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
     }
   }
 
-  /* Set system versioning information. */
-  if (system_period == NULL)
-  {
-    share->disable_system_versioning();
-  }
-  else
-  {
-    DBUG_PRINT("info", ("Setting system versioning informations"));
-    uint16 row_start = uint2korr(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));
-    share->enable_system_versioning(row_start, row_end);
-    vers_start_field()->set_generated_row_start();
-    vers_end_field()->set_generated_row_end();
-
-    if (vers_start_field()->type() != MYSQL_TYPE_TIMESTAMP)
-    {
-      my_error(ER_SYS_START_FIELD_MUST_BE_TIMESTAMP, MYF(0), share->table_name);
-      goto err;
-    }
-    if (vers_end_field()->type() != MYSQL_TYPE_TIMESTAMP)
-    {
-      my_error(ER_SYS_END_FIELD_MUST_BE_TIMESTAMP, MYF(0), share->table_name);
-      goto err;
-    }
-  }
-
   /*
     the correct null_bytes can now be set, since bitfields have been taken
     into account
@@ -2547,19 +2518,70 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
     bitmap_clear_all(share->check_set);
   }
 
-  delete handler_file;
 #ifndef DBUG_OFF
   if (use_hash)
     (void) my_hash_check(&share->name_hash);
 #endif
 
   share->db_plugin= se_plugin;
+
+  /* Set system versioning information. */
+  if (system_period == NULL)
+  {
+    share->disable_system_versioning();
+  }
+  else
+  {
+    DBUG_PRINT("info", ("Setting system versioning informations"));
+    uint16 row_start = uint2korr(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));
+    share->enable_system_versioning(row_start, row_end);
+    vers_start_field()->set_generated_row_start();
+    vers_end_field()->set_generated_row_end();
+
+    DBUG_ASSERT(db_type());
+    if (db_type()->versioned())
+    {
+      if (vers_start_field()->type() != MYSQL_TYPE_LONGLONG
+        || !(vers_start_field()->flags & UNSIGNED_FLAG))
+      {
+        my_error(ER_SYS_START_FIELD_MUST_BE_BIGINT, MYF(0), share->table_name);
+        goto err;
+      }
+      if (vers_end_field()->type() != MYSQL_TYPE_LONGLONG
+        || !(vers_end_field()->flags & UNSIGNED_FLAG))
+      {
+        my_error(ER_SYS_END_FIELD_MUST_BE_BIGINT, MYF(0), share->table_name);
+        goto err;
+      }
+    }
+    else
+    {
+      if (vers_start_field()->type() != MYSQL_TYPE_TIMESTAMP)
+      {
+        my_error(ER_SYS_START_FIELD_MUST_BE_TIMESTAMP, MYF(0), share->table_name);
+        goto err;
+      }
+      if (vers_end_field()->type() != MYSQL_TYPE_TIMESTAMP)
+      {
+        my_error(ER_SYS_END_FIELD_MUST_BE_TIMESTAMP, MYF(0), share->table_name);
+        goto err;
+      }
+    } // if (db_type()->versioned())
+  } // if (system_period == NULL)
+
+  delete handler_file;
+
   share->error= OPEN_FRM_OK;
   thd->status_var.opened_shares++;
   thd->mem_root= old_root;
   DBUG_RETURN(0);
 
- err:
+err:
+  share->db_plugin= NULL;
   share->error= OPEN_FRM_CORRUPTED;
   share->open_errno= my_errno;
   delete handler_file;
@@ -7568,14 +7590,14 @@ void TABLE::vers_update_fields()
 {
   DBUG_ENTER("vers_update_fields");
 
+  bitmap_set_bit(write_set, vers_start_field()->field_index);
+  bitmap_set_bit(write_set, vers_end_field()->field_index);
+
   if (vers_start_field()->set_time())
     DBUG_ASSERT(0);
-  if (vers_end_field()->set_max_timestamp())
+  if (vers_end_field()->set_max())
     DBUG_ASSERT(0);
 
-  bitmap_set_bit(write_set, vers_start_field()->field_index);
-  bitmap_set_bit(write_set, vers_end_field()->field_index);
-
   DBUG_VOID_RETURN;
 }
 
diff --git a/sql/table.h b/sql/table.h
index 7416f27756d..ad62c06838b 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1484,7 +1484,7 @@ public:
   bool export_structure(THD *thd, class Row_definition_list *defs);
 
   /**
-    System versioning support.
+    System Versioning support
    */
 
   bool versioned() const
@@ -1492,15 +1492,22 @@ public:
     return s->versioned;
   }
 
+  /* Versioned by SQL layer */
+  bool versioned_by_sql() const
+  {
+    DBUG_ASSERT(s->db_type());
+    return s->versioned && !s->db_type()->versioned();
+  }
+
   Field *vers_start_field() const
   {
-    DBUG_ASSERT(versioned());
+    DBUG_ASSERT(s->versioned);
     return field[s->row_start_field];
   }
 
   Field *vers_end_field() const
   {
-    DBUG_ASSERT(versioned());
+    DBUG_ASSERT(s->versioned);
     return field[s->row_end_field];
   }
 
@@ -1509,9 +1516,9 @@ public:
 /** Number of additional fields used in versioned tables */
 #define VERSIONING_FIELDS 2
 
-  uint user_fields() const
+  uint vers_user_fields() const
   {
-    return versioned() ?
+    return s->versioned ?
       s->fields - VERSIONING_FIELDS :
       s->fields;
   }
diff --git a/storage/innobase/dict/dict0mem.cc b/storage/innobase/dict/dict0mem.cc
index 13db213259c..304a4ae89eb 100644
--- a/storage/innobase/dict/dict0mem.cc
+++ b/storage/innobase/dict/dict0mem.cc
@@ -304,6 +304,16 @@ dict_mem_table_add_col(
 	col = dict_table_get_nth_col(table, i);
 
 	dict_mem_fill_column_struct(col, i, mtype, prtype, len);
+
+	if (prtype & DATA_VERS_ROW_START) {
+		ut_ad(table->flags2 & DICT_TF2_VERSIONED
+			&& !(prtype & DATA_VERS_ROW_END));
+		table->vers_row_start = i;
+	} else if (prtype & DATA_VERS_ROW_END) {
+		ut_ad(table->flags2 & DICT_TF2_VERSIONED
+			&& !(prtype & DATA_VERS_ROW_START));
+		table->vers_row_end = i;
+	}
 }
 
 /** Adds a virtual column definition to a table.
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index acba2461f55..2899654e49d 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -3848,7 +3848,7 @@ innobase_init(
 	innobase_hton->flush_logs = innobase_flush_logs;
 	innobase_hton->show_status = innobase_show_status;
 	innobase_hton->flags =
-		HTON_SUPPORTS_EXTENDED_KEYS | HTON_SUPPORTS_FOREIGN_KEYS;
+		HTON_SUPPORTS_EXTENDED_KEYS | HTON_SUPPORTS_FOREIGN_KEYS | HTON_SUPPORTS_SYS_VERSIONING;
 
 	innobase_hton->release_temporary_latches =
 		innobase_release_temporary_latches;
@@ -9478,6 +9478,11 @@ ha_innobase::update_row(
 
 	error = row_update_for_mysql((byte*) old_row, m_prebuilt);
 
+	if (error == DB_SUCCESS && DICT_TF2_FLAG_IS_SET(m_prebuilt->table, DICT_TF2_VERSIONED)) {
+		if (trx->id != static_cast<trx_id_t>(table->vers_start_field()->val_int()))
+			error = row_insert_for_mysql((byte*) old_row, m_prebuilt, true);
+	}
+
 	if (error == DB_SUCCESS && autoinc) {
 		/* A value for an AUTO_INCREMENT column
 		was specified in the UPDATE statement. */
@@ -11677,8 +11682,17 @@ create_table_info_t::create_table_def()
 	for (i = 0; i < n_cols; i++) {
 		ulint	is_virtual;
 		bool	is_stored = false;
-
 		Field*	field = m_form->field[i];
+		ulint vers_row_start = 0;
+		ulint vers_row_end = 0;
+
+		if (m_flags2 & DICT_TF2_VERSIONED) {
+			if (i == m_form->s->row_start_field) {
+				vers_row_start = DATA_VERS_ROW_START;
+			} else if (i == m_form->s->row_end_field) {
+				vers_row_end = DATA_VERS_ROW_END;
+			}
+		}
 
 		col_type = get_innobase_type_from_mysql_type(
 			&unsigned_type, field);
@@ -11773,7 +11787,8 @@ err_col:
 				dtype_form_prtype(
 					(ulint) field->type()
 					| nulls_allowed | unsigned_type
-					| binary_type | long_true_varchar,
+					| binary_type | long_true_varchar
+					| vers_row_start | vers_row_end,
 					charset_no),
 				col_len);
 		} else {
@@ -11783,6 +11798,7 @@ err_col:
 					(ulint) field->type()
 					| nulls_allowed | unsigned_type
 					| binary_type | long_true_varchar
+					| vers_row_start | vers_row_end
 					| is_virtual,
 					charset_no),
 				col_len, i, 0);
diff --git a/storage/innobase/include/data0type.h b/storage/innobase/include/data0type.h
index 3fa8320b5aa..353fe63cf02 100644
--- a/storage/innobase/include/data0type.h
+++ b/storage/innobase/include/data0type.h
@@ -198,6 +198,8 @@ be less than 256 */
 /** Check whether locking is disabled (never). */
 #define dict_table_is_locking_disabled(table) false
 
+#define DATA_VERS_ROW_START	0x4000	/* System Versioning row start */
+#define DATA_VERS_ROW_END	0x8000	/* System Versioning row end */
 /*-------------------------------------------*/
 
 /* This many bytes we need to store the type information affecting the
diff --git a/storage/innobase/include/dict0mem.h b/storage/innobase/include/dict0mem.h
index f1193d314f8..ae5bb0d0e4d 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/storage/innobase/include/dict0mem.h
@@ -329,8 +329,9 @@ use its own tablespace instead of the system tablespace. */
 index tables) of a FTS table are in HEX format. */
 #define DICT_TF2_FTS_AUX_HEX_NAME	64U
 
-
+/** System Versioning bit. */
 #define DICT_TF2_VERSIONED		512
+
 /* @} */
 
 #define DICT_TF2_FLAG_SET(table, flag)		\
@@ -1487,7 +1488,10 @@ struct dict_table_t {
 
 	/** Virtual column names */
 	const char*				v_col_names;
-
+	unsigned	vers_row_start:10;
+				/*!< System Versioning: row start col index */
+	unsigned	vers_row_end:10;
+				/*!< System Versioning: row end col index */
 	bool		is_system_db;
 				/*!< True if the table belongs to a system
 				database (mysql, information_schema or
diff --git a/storage/innobase/include/row0ins.h b/storage/innobase/include/row0ins.h
index dc6ad2d34ad..561f6f2d8fe 100644
--- a/storage/innobase/include/row0ins.h
+++ b/storage/innobase/include/row0ins.h
@@ -235,4 +235,8 @@ struct ins_node_t{
 #define	INS_NODE_INSERT_ENTRIES 3	/* index entries should be built and
 					inserted */
 
+#ifndef UNIV_NONINL
+#include "row0ins.ic"
+#endif
+
 #endif
diff --git a/storage/innobase/include/row0ins.ic b/storage/innobase/include/row0ins.ic
new file mode 100644
index 00000000000..3192affc327
--- /dev/null
+++ b/storage/innobase/include/row0ins.ic
@@ -0,0 +1,44 @@
+/*****************************************************************************
+
+Copyright (c) 1996, 2009, Oracle and/or its affiliates. All Rights Reserved.
+
+This program is free software; you can redistribute it and/or modify it under
+the terms of the GNU General Public License as published by the Free Software
+Foundation; version 2 of the License.
+
+This program is distributed in the hope that it will be useful, but WITHOUT
+ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License along with
+this program; if not, write to the Free Software Foundation, Inc.,
+51 Franklin Street, Suite 500, Boston, MA 02110-1335 USA
+
+*****************************************************************************/
+
+/**************************************************//**
+@file include/row0ins.ic
+Insert into a table
+
+Created 4/20/1996 Heikki Tuuri
+*******************************************************/
+
+
+UNIV_INLINE
+void set_row_field_8(
+	dtuple_t* row,
+	int field_num,
+	ib_uint64_t data,
+	mem_heap_t* heap)
+{
+	static const ulint fsize = sizeof(data);
+	dfield_t* dfield = dtuple_get_nth_field(row, field_num);
+	ut_ad(dfield->type.len == fsize);
+	if (dfield->len == UNIV_SQL_NULL) {
+		byte* buf = static_cast<byte*>(mem_heap_alloc(heap, fsize));
+		mach_write_to_8(buf, data);
+		dfield_set_data(dfield, buf, fsize);
+	} else {
+		mach_write_to_8(dfield->data, data);
+	}
+}
diff --git a/storage/innobase/include/row0mysql.h b/storage/innobase/include/row0mysql.h
index 59bd786d25c..6fa0a926e3f 100644
--- a/storage/innobase/include/row0mysql.h
+++ b/storage/innobase/include/row0mysql.h
@@ -241,7 +241,10 @@ row_lock_table_for_mysql(
 dberr_t
 row_insert_for_mysql(
 	const byte*		mysql_rec,
-	row_prebuilt_t*		prebuilt)
+	row_prebuilt_t*		prebuilt,
+	bool			historical
+					/*!< in: System Versioning, row is */
+			= false)	/* historical */
 	MY_ATTRIBUTE((warn_unused_result));
 
 /*********************************************************************//**
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc
index d071036e947..4ee2be787ec 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/storage/innobase/row/row0ins.cc
@@ -27,6 +27,11 @@ Created 4/20/1996 Heikki Tuuri
 #include "ha_prototypes.h"
 
 #include "row0ins.h"
+
+#ifdef UNIV_NONINL
+#include "row0ins.ic"
+#endif
+
 #include "dict0dict.h"
 #include "dict0boot.h"
 #include "trx0rec.h"
@@ -3816,17 +3821,6 @@ error_handling:
 	return(thr);
 }
 
-inline
-void set_row_field_8(dtuple_t* row, int field_num, ib_uint64_t data, mem_heap_t* heap)
-{
-	static const ulint fsize = 8;
-	dfield_t* dfield = dtuple_get_nth_field(row, field_num);
-	ut_ad(dfield->type.len == fsize);
-	byte* buf = static_cast<byte*>(mem_heap_alloc(heap, fsize));
-	mach_write_to_8(buf, data);
-	dfield_set_data(dfield, buf, fsize);
-}
-
 /***********************************************************//**
 Inserts a row to SYS_VTQ, low level.
 @return DB_SUCCESS if operation successfully completed, else error
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index c08b608e006..2bf26f242ca 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -1406,7 +1406,8 @@ run_again:
 dberr_t
 row_insert_for_mysql(
 	const byte*	mysql_rec,
-	row_prebuilt_t*	prebuilt)
+	row_prebuilt_t*	prebuilt,
+	bool		historical)
 {
 	trx_savept_t	savept;
 	que_thr_t*	thr;
@@ -1486,6 +1487,16 @@ row_insert_for_mysql(
 	row_mysql_convert_row_to_innobase(node->row, prebuilt, mysql_rec,
 					  &blob_heap);
 
+	if (DICT_TF2_FLAG_IS_SET(node->table, DICT_TF2_VERSIONED)) {
+		ut_ad(table->vers_row_start != table->vers_row_end);
+		if (historical) {
+			set_row_field_8(node->row, table->vers_row_end, trx->id, node->entry_sys_heap);
+		} else {
+			set_row_field_8(node->row, table->vers_row_start, trx->id, node->entry_sys_heap);
+			set_row_field_8(node->row, table->vers_row_end, IB_UINT64_MAX, node->entry_sys_heap);
+		}
+	}
+
 	savept = trx_savept_take(trx);
 
 	thr = que_fork_get_first_thr(prebuilt->ins_graph);
@@ -1537,7 +1548,7 @@ error_exit:
 
 	node->duplicate = NULL;
 
-	if (!trx->vtq_notify_on_commit && DICT_TF2_FLAG_IS_SET(node->table, DICT_TF2_VERSIONED)) {
+	if (DICT_TF2_FLAG_IS_SET(node->table, DICT_TF2_VERSIONED)) {
 		trx->vtq_notify_on_commit = true;
 	}
 
@@ -1935,6 +1946,38 @@ row_update_for_mysql_using_upd_graph(
 					      prebuilt->clust_pcur);
 	}
 
+	if (DICT_TF2_FLAG_IS_SET(node->table, DICT_TF2_VERSIONED))
+	{
+		/* System Versioning: update sys_trx_start to current trx_id */
+		upd_t* uvect = node->update;
+		upd_field_t* ufield;
+		dict_col_t* col;
+		if (node->is_delete) {
+			ufield = &uvect->fields[0];
+			uvect->n_fields = 0;
+			node->is_delete = false;
+			col = &table->cols[table->vers_row_end];
+		} else {
+			ut_ad(uvect->n_fields < node->table->n_cols);
+			ufield = &uvect->fields[uvect->n_fields];
+			col = &table->cols[table->vers_row_start];
+		}
+		UNIV_MEM_INVALID(ufield, sizeof *ufield);
+		ufield->field_no = dict_col_get_clust_pos(col, clust_index);
+		ufield->orig_len = 0;
+		ufield->exp = NULL;
+
+		static const ulint fsize = sizeof(trx_id_t);
+		byte* buf = static_cast<byte*>(mem_heap_alloc(node->heap, fsize));
+		mach_write_to_8(buf, trx->id);
+		dfield_t* dfield = &ufield->new_val;
+		dfield_set_data(dfield, buf, fsize);
+		dict_col_copy_type(col, &dfield->type);
+
+		uvect->n_fields++;
+		ut_ad(node->in_mysql_interface); // otherwise needs to recalculate node->cmpl_info
+	}
+
 	ut_a(node->pcur->rel_pos == BTR_PCUR_ON);
 
 	/* MySQL seems to call rnd_pos before updating each row it
@@ -2132,7 +2175,7 @@ run_again:
 		}
 	}
 
-	if (!trx->vtq_notify_on_commit && DICT_TF2_FLAG_IS_SET(node->table, DICT_TF2_VERSIONED)) {
+	if (DICT_TF2_FLAG_IS_SET(node->table, DICT_TF2_VERSIONED)) {
 		trx->vtq_notify_on_commit = true;
 	}
 
diff --git a/storage/xtradb/include/row0ins.ic b/storage/xtradb/include/row0ins.ic
index 9c191d869a2..3192affc327 100644
--- a/storage/xtradb/include/row0ins.ic
+++ b/storage/xtradb/include/row0ins.ic
@@ -24,3 +24,21 @@ Created 4/20/1996 Heikki Tuuri
 *******************************************************/
 
 
+UNIV_INLINE
+void set_row_field_8(
+	dtuple_t* row,
+	int field_num,
+	ib_uint64_t data,
+	mem_heap_t* heap)
+{
+	static const ulint fsize = sizeof(data);
+	dfield_t* dfield = dtuple_get_nth_field(row, field_num);
+	ut_ad(dfield->type.len == fsize);
+	if (dfield->len == UNIV_SQL_NULL) {
+		byte* buf = static_cast<byte*>(mem_heap_alloc(heap, fsize));
+		mach_write_to_8(buf, data);
+		dfield_set_data(dfield, buf, fsize);
+	} else {
+		mach_write_to_8(dfield->data, data);
+	}
+}