MDEV-25004 vers_force_trx option to force transactional System Versioning

Works like vers_force but forces trx_id-based system-versioned tables if the storage supports it (currently InnoDB-only). Otherwise creates timestamp-based system-versioned table.

Author: midenok

sql/handler.cc

@@ -6976,7 +6976,6 @@ static Create_field *vers_init_sys_field(THD *thd, const char *field_name, int f
  f->flags= flags | NOT_NULL_FLAG;
  if (integer)
  {
- DBUG_ASSERT(0); // Not implemented yet
  f->set_handler(&type_handler_vers_trx_id);
  f->length= MY_INT64_NUM_DECIMAL_DIGITS - 1;
  f->flags|= UNSIGNED_FLAG;
@@ -6994,10 +6993,13 @@ static Create_field *vers_init_sys_field(THD *thd, const char *field_name, int f
  return f;
 }
 
-static bool vers_create_sys_field(THD *thd, const char *field_name,
- Alter_info *alter_info, int flags)
+bool Vers_parse_info::create_sys_field(THD *thd, const char *field_name,
+  Alter_info *alter_info, int flags)
 {
- Create_field *f= vers_init_sys_field(thd, field_name, flags, false);
+ DBUG_ASSERT(can_native >= 0); /* Requires vers_check_native() called */
+ Create_field *f= vers_init_sys_field(thd, field_name, flags,
+ DBUG_EVALUATE_IF("sysvers_force_trx",
+ (bool) can_native, false));
  if (!f)
  return true;
 
@@ -7021,23 +7023,34 @@ 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_ROW_START) ||
- vers_create_sys_field(thd, default_end, alter_info, VERS_ROW_END))
+ if (create_sys_field(thd, default_start, alter_info, VERS_ROW_START) ||
+ create_sys_field(thd, default_end, alter_info, VERS_ROW_END))
  {
  return true;
  }
  return false;
 }
 
 
+void Table_scope_and_contents_source_st::vers_check_native()
+{
+ vers_info.can_native= (db_type->db_type == DB_TYPE_PARTITION_DB ||
+ ha_check_storage_engine_flag(db_type,
+ HTON_NATIVE_SYS_VERSIONING));
+}
+
+
 bool Table_scope_and_contents_source_st::vers_fix_system_fields(
  THD *thd, Alter_info *alter_info, const TABLE_LIST &create_table)
 {
  DBUG_ASSERT(!(alter_info->flags & ALTER_DROP_SYSTEM_VERSIONING));
 
- DBUG_EXECUTE_IF("sysvers_force", if (!tmp_table()) {
- alter_info->flags|= ALTER_ADD_SYSTEM_VERSIONING;
- options|= HA_VERSIONED_TABLE; });
+ if (DBUG_EVALUATE_IF("sysvers_force", true, false) ||
+ DBUG_EVALUATE_IF("sysvers_force_trx", true, false))
+ {
+ alter_info->flags|= ALTER_ADD_SYSTEM_VERSIONING;
+ options|= HA_VERSIONED_TABLE;
+ }
 
  if (!vers_info.need_check(alter_info))
  return false;
@@ -7068,7 +7081,9 @@ bool Table_scope_and_contents_source_st::vers_fix_system_fields(
  {
  f->flags|= VERS_UPDATE_UNVERSIONED_FLAG;
  }
- } // while (Create_field *f= it++)
+ } // while
+
+ vers_check_native();
 
  if (vers_info.fix_implicit(thd, alter_info))
  return true;
@@ -7121,11 +7136,7 @@ bool Table_scope_and_contents_source_st::vers_check_system_fields(
  if (!(alter_info->flags & ALTER_ADD_SYSTEM_VERSIONING) && !versioned_fields)
  return false;
 
- bool can_native= ha_check_storage_engine_flag(db_type,
- HTON_NATIVE_SYS_VERSIONING)
- || db_type->db_type == DB_TYPE_PARTITION_DB;
-
- return vers_info.check_sys_fields(table_name, db, alter_info, can_native);
+ return vers_info.check_sys_fields(table_name, db, alter_info);
 }
 
 
@@ -7138,7 +7149,8 @@ bool Vers_parse_info::fix_alter_info(THD *thd, Alter_info *alter_info,
  if (!need_check(alter_info) && !share->versioned)
  return false;
 
- if (DBUG_EVALUATE_IF("sysvers_force", 0, share->tmp_table))
+ if (DBUG_EVALUATE_IF("sysvers_force", 0, share->tmp_table) ||
+ DBUG_EVALUATE_IF("sysvers_force_trx", 0, share->tmp_table))
  {
  my_error(ER_VERS_TEMPORARY, MYF(0));
  return true;
@@ -7383,8 +7395,7 @@ bool Create_field::vers_check_bigint(const Lex_table_name &table_name) const
 
 bool Vers_parse_info::check_sys_fields(const Lex_table_name &table_name,
  const Lex_table_name &db,
- Alter_info *alter_info,
- bool can_native) const
+ Alter_info *alter_info) const
 {
  if (check_conditions(table_name, db))
  return true;

sql/handler.h

@@ -1881,7 +1881,8 @@ struct Vers_parse_info
 {
  Vers_parse_info() :
  versioned_fields(false),
- unversioned_fields(false)
+ unversioned_fields(false),
+ can_native(-1)
  {}
 
  void init() // Deep initialization
@@ -1890,6 +1891,7 @@ struct Vers_parse_info
  as_row= start_end_t(null_clex_str, null_clex_str);
  versioned_fields= false;
  unversioned_fields= false;
+ can_native= -1;
  }
 
  struct start_end_t
@@ -1936,6 +1938,9 @@ struct Vers_parse_info
  bool need_check(const Alter_info *alter_info) const;
  bool check_conditions(const Lex_table_name &table_name,
  const Lex_table_name &db) const;
+ bool create_sys_field(THD *thd, const char *field_name,
+ Alter_info *alter_info, int flags);
+
 public:
  static const Lex_ident default_start;
  static const Lex_ident default_end;
@@ -1945,15 +1950,15 @@ struct Vers_parse_info
  bool fix_create_like(Alter_info &alter_info, HA_CREATE_INFO &create_info,
  TABLE_LIST &src_table, TABLE_LIST &table);
  bool check_sys_fields(const Lex_table_name &table_name,
- const Lex_table_name &db, Alter_info *alter_info,
- bool can_native) const;
+ const Lex_table_name &db, Alter_info *alter_info) const;
 
  /**
  At least one field was specified 'WITH/WITHOUT SYSTEM VERSIONING'.
  Useful for error handling.
  */
  bool versioned_fields : 1;
  bool unversioned_fields : 1;
+ int can_native;
 };
 
 /**
@@ -2059,14 +2064,15 @@ struct Table_scope_and_contents_source_st:
  vers_info.init();
  }
 
+ void vers_check_native();
+
  bool vers_fix_system_fields(THD *thd, Alter_info *alter_info,
  const TABLE_LIST &create_table);
 
  bool vers_check_system_fields(THD *thd, Alter_info *alter_info,
  const Lex_table_name &table_name,
  const Lex_table_name &db,
  int select_count= 0);
-
 };
 
 

sql/sql_table.cc

@@ -9637,6 +9637,7 @@ bool mysql_alter_table(THD *thd, const LEX_CSTRING *new_db,
  if (check_engine(thd, alter_ctx.new_db.str, alter_ctx.new_name.str, create_info))
  DBUG_RETURN(true);
 
+ create_info->vers_check_native();
  if (create_info->vers_info.fix_alter_info(thd, alter_info, create_info, table))
  {
  DBUG_RETURN(true);

storage/innobase/row/row0mysql.cc

@@ -1672,7 +1672,7 @@ row_fts_update_or_delete(
 
 if (new_doc_id == 0) {
 ib::error() << "InnoDB FTS: Doc ID cannot be 0";
-return(DB_FTS_INVALID_DOCID);
+DBUG_RETURN(DB_FTS_INVALID_DOCID);
 }
 row_fts_do_update(trx, table, old_doc_id, new_doc_id);
 }