MDEV-32126 Assertion fails upon online ALTER and binary log enabled

Assertion `!writer.checksum_len || writer.remains == 0' fails upon
concurrent online ALTER and transactions with failing statements and binary
log enabled.
Also another assertion, `pos != (~(my_off_t) 0)', fails in my_seek, upon
reinit_io_cache, on a simplified test. This means that IO_CACHE wasn't
properly initialized, or had an error before.

The overall problem is a deep interference with the effect of an installed
binlog_hton: the assumption about that thd->binlog_get_cache_mngr() is,
sufficiently, NULL, when we shouldn't run the binlog part of
binlog_commit/binlog_rollback, is wrong: as turns out, sometimes the binlog
handlerton can be not installed in current thd, but binlog_commit can be
called on behalf of binlog, as in the bug reported.

One separate condition found is XA recovery of the orphaned transaction,
when binlog_commit is also called, but it has nothing to do with
online alter.

Solution:
Extract online alter operations into a separate handlerton.

mysql-test/main/alter_table_online_debug.result

@@ -1518,6 +1518,51 @@ connection default;
 drop table t1;
 set debug_sync= reset;
 drop table iso_levels;
+# MDEV-32126 Assertion fails upon online ALTER and binary log enabled
+create temporary table tmp (id int, primary key(id)) engine=innodb;
+create table t1 (a int, b text) engine=innodb;
+create table t2 (a int, b int, c char(8), d text, unique(a)) engine=innodb;
+insert into t2 values (1,1,'f','e'),(1000,1000,'c','b');
+connection default;
+set debug_sync= 'alter_table_online_before_lock signal go_trx wait_for go_alter';
+alter table t2 force, algorithm=copy, lock=none;
+connection con2;
+set debug_sync= 'now wait_for go_trx';
+start transaction;
+insert into t1 values (3,'a');
+insert into t2 values (3,3,'a','x'), (3,3,'a','x');
+ERROR 23000: Duplicate entry '3' for key 'a'
+insert into t2 values (3,3,'a','x');
+commit;
+set debug_sync= 'now signal go_alter';
+connection default;
+truncate t2;
+set @@binlog_format=mixed;
+connection con2;
+start transaction;
+create temporary table tmp (id int, primary key(id)) engine=innodb;
+insert into t1 values (1, repeat('x',8000)),(2, repeat('x',8000));
+update t2 set b = null order by b limit 2;
+insert into t1 values (3, repeat('x',8000));
+delete from t1;
+insert into t2 values (1,1,'f','e'),(1000,1000,'c','b');
+commit;
+connection default;
+set debug_sync= 'alter_table_online_before_lock signal go_trx wait_for go_alter';
+alter table t2 force, algorithm=copy, lock=none;
+connection con2;
+set debug_sync= 'now wait_for go_trx';
+start transaction;
+drop temporary table if exists tmp;
+insert into t2 values (3,3,'a','x'), (3,3,'a','x');
+ERROR 23000: Duplicate entry '3' for key 'a'
+insert into t2 values (3,3,'a','x');
+commit;
+set debug_sync= 'now signal go_alter';
+connection default;
+drop table t1, t2;
+set @@binlog_format=default;
+set debug_sync= reset;
 disconnect con1;
 disconnect con2;
 #

mysql-test/main/alter_table_online_debug.test

@@ -1736,6 +1736,60 @@ set debug_sync= reset;
 drop table iso_levels;
 
 
+--echo # MDEV-32126 Assertion fails upon online ALTER and binary log enabled
+create temporary table tmp (id int, primary key(id)) engine=innodb;
+create table t1 (a int, b text) engine=innodb;
+create table t2 (a int, b int, c char(8), d text, unique(a)) engine=innodb;
+insert into t2 values (1,1,'f','e'),(1000,1000,'c','b');
+--connection default
+set debug_sync= 'alter_table_online_before_lock signal go_trx wait_for go_alter';
+send alter table t2 force, algorithm=copy, lock=none;
+
+--connection con2
+set debug_sync= 'now wait_for go_trx';
+start transaction;
+insert into t1 values (3,'a');
+--error ER_DUP_ENTRY
+insert into t2 values (3,3,'a','x'), (3,3,'a','x');
+insert into t2 values (3,3,'a','x');
+commit;
+set debug_sync= 'now signal go_alter';
+--connection default
+--reap
+truncate t2;
+set @@binlog_format=mixed;
+--connection con2
+start transaction;
+create temporary table tmp (id int, primary key(id)) engine=innodb;
+insert into t1 values (1, repeat('x',8000)),(2, repeat('x',8000));
+update t2 set b = null order by b limit 2;
+insert into t1 values (3, repeat('x',8000));
+delete from t1;
+insert into t2 values (1,1,'f','e'),(1000,1000,'c','b');
+commit;
+
+
+--connection default
+set debug_sync= 'alter_table_online_before_lock signal go_trx wait_for go_alter';
+send alter table t2 force, algorithm=copy, lock=none;
+
+--connection con2
+set debug_sync= 'now wait_for go_trx';
+start transaction;
+drop temporary table if exists tmp;
+--error ER_DUP_ENTRY
+insert into t2 values (3,3,'a','x'), (3,3,'a','x');
+insert into t2 values (3,3,'a','x');
+commit;
+set debug_sync= 'now signal go_alter';
+
+--connection default
+--reap
+
+drop table t1, t2;
+set @@binlog_format=default;
+set debug_sync= reset;
+
 
 --disconnect con1
 --disconnect con2

sql/CMakeLists.txt

@@ -218,6 +218,8 @@ MYSQL_ADD_PLUGIN(partition ha_partition.cc STORAGE_ENGINE DEFAULT STATIC_ONLY
 RECOMPILE_FOR_EMBEDDED)
 MYSQL_ADD_PLUGIN(sql_sequence ha_sequence.cc STORAGE_ENGINE MANDATORY STATIC_ONLY
 RECOMPILE_FOR_EMBEDDED)
+MYSQL_ADD_PLUGIN(online_alter_log log.cc STORAGE_ENGINE MANDATORY STATIC_ONLY
+NOT_EMBEDDED)
 
 ADD_LIBRARY(sql STATIC ${SQL_SOURCE})
 MAYBE_DISABLE_IPO(sql)

sql/handler.h

@@ -557,6 +557,7 @@ enum legacy_db_type
   DB_TYPE_BLACKHOLE_DB=19,
   DB_TYPE_PARTITION_DB=20,
   DB_TYPE_BINLOG=21,
+  DB_TYPE_ONLINE_ALTER=22,
   DB_TYPE_PBXT=23,
   DB_TYPE_PERFORMANCE_SCHEMA=28,
   DB_TYPE_S3=41,
@@ -1096,6 +1097,9 @@ enum ha_stat_type { HA_ENGINE_STATUS, HA_ENGINE_LOGS, HA_ENGINE_MUTEX };
 extern MYSQL_PLUGIN_IMPORT st_plugin_int *hton2plugin[MAX_HA];
 
 struct handlerton;
+
+extern handlerton *online_alter_hton;
+
 #define view_pseudo_hton ((handlerton *)1)
 
 /*
@@ -1916,7 +1920,6 @@ struct THD_TRANS
 
 };
 
-
 /**
   Either statement transaction or normal transaction - related
   thread-specific storage engine data.
@@ -1969,7 +1972,9 @@ class Ha_trx_info
   bool is_trx_read_write() const
   {
     DBUG_ASSERT(is_started());
-    return m_flags & (int) TRX_READ_WRITE;
+    bool result= m_flags & (int) TRX_READ_WRITE;
+    DBUG_ASSERT(!result || m_ht != online_alter_hton);
+    return result;
   }
   bool is_started() const { return m_ht != NULL; }
   /** Mark this transaction read-write if the argument is read-write. */

sql/log.cc

@@ -2185,7 +2185,6 @@ int binlog_rollback_by_xid(handlerton *hton, XID *xid)
 
   DBUG_ASSERT(thd->lex->sql_command == SQLCOM_XA_ROLLBACK ||
               (thd->transaction->xid_state.get_state_code() == XA_ROLLBACK_ONLY));
-
   rc= binlog_rollback(hton, thd, TRUE);
   thd->ha_data[hton->slot].ha_info[1].reset();
 
@@ -2276,9 +2275,9 @@ int binlog_log_row_online_alter(TABLE* table, const uchar *before_record,
   if (!table->online_alter_cache)
   {
     table->online_alter_cache= online_alter_binlog_get_cache_data(thd, table);
-    trans_register_ha(thd, false, binlog_hton, 0);
+    trans_register_ha(thd, false, online_alter_hton, 0);
     if (thd->in_multi_stmt_transaction_mode())
-      trans_register_ha(thd, true, binlog_hton, 0);
+      trans_register_ha(thd, true, online_alter_hton, 0);
   }
 
   // We need to log all columns for the case if alter table changes primary key
@@ -2339,12 +2338,7 @@ int binlog_commit(THD *thd, bool all, bool ro_1pc)
   PSI_stage_info org_stage;
   DBUG_ENTER("binlog_commit");
 
-  IF_DBUG(bool commit_online= !thd->online_alter_cache_list.empty(),);
-
   bool is_ending_transaction= ending_trans(thd, all);
-  error= binlog_online_alter_end_trans(thd, all, true);
-  if (error)
-    DBUG_RETURN(error);
 
   binlog_cache_mngr *const cache_mngr= thd->binlog_get_cache_mngr();
   /*
@@ -2352,7 +2346,7 @@ int binlog_commit(THD *thd, bool all, bool ro_1pc)
   */
   if (!cache_mngr)
   {
-    DBUG_ASSERT(WSREP(thd) || commit_online ||
+    DBUG_ASSERT(WSREP(thd) ||
                 (thd->lex->sql_command != SQLCOM_XA_PREPARE &&
                 !(thd->lex->sql_command == SQLCOM_XA_COMMIT &&
                   thd->lex->xa_opt == XA_ONE_PHASE)));
@@ -2450,17 +2444,13 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
   DBUG_ENTER("binlog_rollback");
 
   bool is_ending_trans= ending_trans(thd, all);
-
-  bool rollback_online= !thd->online_alter_cache_list.empty();
-  if (rollback_online)
-    binlog_online_alter_end_trans(thd, all, 0);
   int error= 0;
   binlog_cache_mngr *const cache_mngr= thd->binlog_get_cache_mngr();
 
   if (!cache_mngr)
   {
-    DBUG_ASSERT(WSREP(thd) || rollback_online);
-    DBUG_ASSERT(thd->lex->sql_command != SQLCOM_XA_ROLLBACK || rollback_online);
+    DBUG_ASSERT(WSREP(thd));
+    DBUG_ASSERT(thd->lex->sql_command != SQLCOM_XA_ROLLBACK);
 
     DBUG_RETURN(0);
   }
@@ -12538,3 +12528,61 @@ void wsrep_register_binlog_handler(THD *thd, bool trx)
 }
 
 #endif /* WITH_WSREP */
+
+static int online_alter_close_connection(handlerton *hton, THD *thd)
+{
+  DBUG_ASSERT(thd->online_alter_cache_list.empty());
+  return 0;
+}
+
+handlerton *online_alter_hton;
+
+int online_alter_log_init(void *p)
+{
+  online_alter_hton= (handlerton *)p;
+  online_alter_hton->db_type= DB_TYPE_ONLINE_ALTER;
+  online_alter_hton->savepoint_offset= sizeof(my_off_t);
+  online_alter_hton->close_connection= online_alter_close_connection;
+
+  online_alter_hton->savepoint_set= // Done by online_alter_savepoint_set
+          [](handlerton *, THD *, void *){ return 0; };
+  online_alter_hton->savepoint_rollback= // Done by online_alter_savepoint_rollback
+          [](handlerton *, THD *, void *){ return 0; };
+  online_alter_hton->savepoint_rollback_can_release_mdl=
+          [](handlerton *hton, THD *thd){ return true; };
+
+  online_alter_hton->commit= [](handlerton *, THD *thd, bool all)
+          { return binlog_online_alter_end_trans(thd, all, true); };
+  online_alter_hton->rollback= [](handlerton *, THD *thd, bool all)
+          { return binlog_online_alter_end_trans(thd, all, false); };
+  online_alter_hton->commit_by_xid= [](handlerton *hton, XID *xid)
+          { return binlog_online_alter_end_trans(current_thd, true, true); };
+  online_alter_hton->rollback_by_xid= [](handlerton *hton, XID *xid)
+          { return binlog_online_alter_end_trans(current_thd, true, false); };
+
+  online_alter_hton->drop_table= [](handlerton *, const char*) { return -1; };
+  online_alter_hton->flags= HTON_NOT_USER_SELECTABLE | HTON_HIDDEN
+                            | HTON_NO_ROLLBACK;
+  return 0;
+}
+
+struct st_mysql_storage_engine online_alter_storage_engine=
+{ MYSQL_HANDLERTON_INTERFACE_VERSION };
+
+maria_declare_plugin(online_alter_log)
+{
+  MYSQL_STORAGE_ENGINE_PLUGIN,
+  &online_alter_storage_engine,
+  "online_alter_log",
+  "MariaDB PLC",
+  "This is a pseudo storage engine to represent the online alter log in a transaction",
+  PLUGIN_LICENSE_GPL,
+  online_alter_log_init,
+  NULL,
+  0x0100, // 1.0
+  NULL,   // no status vars
+  NULL,   // no sysvars
+  "1.0",
+  MariaDB_PLUGIN_MATURITY_STABLE
+}
+maria_declare_plugin_end;

sql/log.h

@@ -1351,6 +1351,7 @@ binlog_cache_data* binlog_get_cache_data(binlog_cache_mngr *cache_mngr,
 
 extern MYSQL_PLUGIN_IMPORT MYSQL_BIN_LOG mysql_bin_log;
 extern handlerton *binlog_hton;
+extern handlerton *online_alter_hton;
 extern LOGGER logger;
 
 extern const char *log_bin_index;