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.

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,19 +5437,21 @@ 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);
 
   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);
 }

sql/field.h

@@ -677,17 +677,16 @@ class Field: public Value_source
   { 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 @@ class Field_longlong :public Field_num {
   {
     return unpack_int64(to, from, from_end);
   }
+
+  bool set_max();
+  bool is_max();
 };
 
 
@@ -2582,8 +2584,8 @@ class Field_timestampf :public Field_timestamp_with_dec {
   {
     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); }

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 */

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"

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));

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;
     }

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))

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);
 

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();

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"));
   }