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.

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,
+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,8 +7023,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_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;
   }
@@ -7030,14 +7032,25 @@ bool Vers_parse_info::fix_implicit(THD *thd, Alter_info *alter_info)
 }
 
 
+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()) {
+  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; });
+    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,8 +1950,7 @@ 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'.
@@ -1954,6 +1958,7 @@ struct Vers_parse_info
   */
   bool versioned_fields : 1;
   bool unversioned_fields : 1;
+  int can_native;
 };
 
 /**
@@ -2059,6 +2064,8 @@ 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);
 
@@ -2066,7 +2073,6 @@ struct Table_scope_and_contents_source_st:
                                 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);
 	}