mirror/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-01-29 02:05:57 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

IB: 0.2 part III

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

Warning: breaks 'insert' and 'update' tests since they require part IV.
This commit is contained in:
Aleksey Midenkov 2016-09-29 11:12:46 +00:00
parent 23f4e40839
commit 1ec7dbe176
23 changed files with 354 additions and 143 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
sql/field.cc
View file

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

22
sql/field.h
View file

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

12
sql/handler.h
View file

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

6
sql/share/errmsg-utf8.txt
View file

@ -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"

4
sql/sql_base.cc
View file

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

10
sql/sql_delete.cc
View file

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

49
sql/sql_insert.cc
View file

@ -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;
}
DBUG_ASSERT(0);
const int error= table->file->ha_write_row(table->record[0]);
if (!error)
++*inserted;
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))

2
sql/sql_insert.h
View file

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

4
sql/sql_select.cc
View file

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

4
sql/sql_show.cc
View file

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

71
sql/sql_update.cc
View file

@ -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())
if (table->versioned_by_sql())
{
error= 1;
break;
}
if ( (error= vers_insert_history_row(table, &updated_sys_ver)) )
{
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())
if (table->versioned_by_sql())
{
goto err2;
}
if ( (local_error= vers_insert_history_row(table, &updated_sys_ver)) )
{
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

94
sql/table.cc
View file

@ -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");
if (vers_start_field()->set_time())
DBUG_ASSERT(0);
if (vers_end_field()->set_max_timestamp())
DBUG_ASSERT(0);
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())
DBUG_ASSERT(0);
DBUG_VOID_RETURN;
}

17
sql/table.h
View file

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

10
storage/innobase/dict/dict0mem.cc
View file

@ -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.

22
storage/innobase/handler/ha_innodb.cc
View file

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

2
storage/innobase/include/data0type.h
View file

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

8
storage/innobase/include/dict0mem.h
View file

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

4
storage/innobase/include/row0ins.h
View file

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

44
storage/innobase/include/row0ins.ic Normal file
View file

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

5
storage/innobase/include/row0mysql.h
View file

@ -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));
/*********************************************************************//**

16
storage/innobase/row/row0ins.cc
View file

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

49
storage/innobase/row/row0mysql.cc
View file

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

18
storage/xtradb/include/row0ins.ic
View file

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