MDEV-32771 Server crash upon online alter with concurrent XA

In case of a non-recovery XA rollback/commit in the same connection,
thon->rollback is called instead of rollback_by_xid, Though previously,
thd_ha_data was moved to thd->transaction->xid_state.xid in hton->prepare.

Like it wasn't enough, XA PREPARE can be skipped upon user and thus we
can end up in hton->commit/rollback with and unprepared XA, so checking
xid_state.is_explicit_XA is not enough -- we should check
xid_state.get_state_code() == XA_PREPARED, which will also guarantee
is_explicit_XA() == true.

mysql-test/main/alter_table_online_debug.result

@@ -1675,6 +1675,116 @@ connect con1, localhost, root,,;
 connection default;
 drop table t;
 set debug_sync= reset;
+# MDEV-32771 Server crash upon online alter with concurrent XA
+create table t (a int primary key);
+insert t values(1),(2),(3);
+# First, check that nothing from the rollbacked statement commits
+set debug_sync= 'alter_table_online_downgraded signal downgraded wait_for go';
+alter table t add b int default (555), algorithm=copy;
+connection con1;
+set debug_sync= 'now wait_for downgraded';
+xa start 'xid';
+update t set a = 0;
+ERROR 23000: Duplicate entry '0' for key 'PRIMARY'
+xa end 'xid';
+xa prepare 'xid';
+xa commit 'xid';
+set debug_sync= 'now signal go';
+connection default;
+select * from t;
+a	b
+1	555
+2	555
+3	555
+set debug_sync= 'alter_table_online_downgraded signal downgraded wait_for go';
+alter table t add c int default(777), algorithm=copy;
+connection con1;
+set debug_sync= 'now wait_for downgraded';
+xa start 'xid';
+update t set a = 0;
+ERROR 23000: Duplicate entry '0' for key 'PRIMARY'
+xa end 'xid';
+xa prepare 'xid';
+xa rollback 'xid';
+set debug_sync= 'now signal go';
+connection default;
+select * from t;
+a	b	c
+1	555	777
+2	555	777
+3	555	777
+# Same, but add one successful statement into transaction
+set debug_sync= 'alter_table_online_downgraded signal downgraded wait_for go';
+alter table t drop b, algorithm=copy;
+connection con1;
+set debug_sync= 'now wait_for downgraded';
+xa start 'xid';
+update t set a = 10 where a = 1;
+update t set a = 0;
+ERROR 23000: Duplicate entry '0' for key 'PRIMARY'
+xa end 'xid';
+xa prepare 'xid';
+xa rollback 'xid';
+set debug_sync= 'now signal go';
+connection default;
+select * from t;
+a	c
+1	777
+2	777
+3	777
+set debug_sync= 'alter_table_online_downgraded signal downgraded wait_for go';
+alter table t drop primary key, algorithm=copy;
+connection con1;
+set debug_sync= 'now wait_for downgraded';
+xa start 'xid';
+# This statement will take effect.
+update t set a = 10 where a = 1;
+update t set a = 0;
+ERROR 23000: Duplicate entry '0' for key 'PRIMARY'
+xa end 'xid';
+xa prepare 'xid';
+xa commit 'xid';
+set debug_sync= 'now signal go';
+connection default;
+select * from t;
+a	c
+10	777
+2	777
+3	777
+# The only statement succeeds (test both commit and rollback)
+set debug_sync= 'alter_table_online_downgraded signal downgraded wait_for go';
+alter table t add d text default ('qwe'), algorithm=copy;
+connection con1;
+set debug_sync= 'now wait_for downgraded';
+xa start 'xid';
+update t set a = 0;
+xa end 'xid';
+xa prepare 'xid';
+xa rollback 'xid';
+set debug_sync= 'now signal go';
+connection default;
+select * from t;
+a	c	d
+10	777	qwe
+2	777	qwe
+3	777	qwe
+set debug_sync= 'alter_table_online_downgraded signal downgraded wait_for go';
+alter table t drop c, algorithm=copy;
+connection con1;
+set debug_sync= 'now wait_for downgraded';
+xa start 'xid';
+update t set a = 0;
+xa end 'xid';
+xa prepare 'xid';
+xa commit 'xid';
+set debug_sync= 'now signal go';
+connection default;
+select * from t;
+a	d
+0	qwe
+0	qwe
+0	qwe
+drop table t;
 set global default_storage_engine= MyISAM;
 disconnect con1;
 disconnect con2;

mysql-test/main/alter_table_online_debug.test

@@ -1922,6 +1922,125 @@ select * from t;
 --connection default
 drop table t;
 set debug_sync= reset;
+
+--echo # MDEV-32771 Server crash upon online alter with concurrent XA
+
+create table t (a int primary key);
+insert t values(1),(2),(3);
+
+--echo # First, check that nothing from the rollbacked statement commits
+
+set debug_sync= 'alter_table_online_downgraded signal downgraded wait_for go';
+send alter table t add b int default (555), algorithm=copy;
+
+--connection con1
+set debug_sync= 'now wait_for downgraded';
+xa start 'xid';
+--error ER_DUP_ENTRY
+update t set a = 0;
+xa end 'xid';
+xa prepare 'xid';
+xa commit 'xid';
+set debug_sync= 'now signal go';
+
+--connection default
+--reap
+select * from t;
+
+set debug_sync= 'alter_table_online_downgraded signal downgraded wait_for go';
+send alter table t add c int default(777), algorithm=copy;
+
+--connection con1
+set debug_sync= 'now wait_for downgraded';
+xa start 'xid';
+--error ER_DUP_ENTRY
+update t set a = 0;
+xa end 'xid';
+xa prepare 'xid';
+xa rollback 'xid';
+set debug_sync= 'now signal go';
+
+--connection default
+--reap
+select * from t;
+
+--echo # Same, but add one successful statement into transaction
+
+set debug_sync= 'alter_table_online_downgraded signal downgraded wait_for go';
+send alter table t drop b, algorithm=copy;
+
+--connection con1
+set debug_sync= 'now wait_for downgraded';
+xa start 'xid';
+update t set a = 10 where a = 1;
+--error ER_DUP_ENTRY
+update t set a = 0;
+xa end 'xid';
+xa prepare 'xid';
+xa rollback 'xid';
+set debug_sync= 'now signal go';
+
+--connection default
+--reap
+select * from t;
+
+set debug_sync= 'alter_table_online_downgraded signal downgraded wait_for go';
+send alter table t drop primary key, algorithm=copy;
+
+--connection con1
+set debug_sync= 'now wait_for downgraded';
+xa start 'xid';
+--echo # This statement will take effect.
+update t set a = 10 where a = 1;
+--error ER_DUP_ENTRY
+update t set a = 0;
+xa end 'xid';
+xa prepare 'xid';
+xa commit 'xid';
+set debug_sync= 'now signal go';
+
+--connection default
+--reap
+select * from t;
+
+
+--echo # The only statement succeeds (test both commit and rollback)
+
+set debug_sync= 'alter_table_online_downgraded signal downgraded wait_for go';
+send alter table t add d text default ('qwe'), algorithm=copy;
+
+--connection con1
+set debug_sync= 'now wait_for downgraded';
+xa start 'xid';
+update t set a = 0;
+xa end 'xid';
+xa prepare 'xid';
+xa rollback 'xid';
+set debug_sync= 'now signal go';
+
+--connection default
+--reap
+select * from t;
+
+set debug_sync= 'alter_table_online_downgraded signal downgraded wait_for go';
+send alter table t drop c, algorithm=copy;
+
+--connection con1
+set debug_sync= 'now wait_for downgraded';
+xa start 'xid';
+update t set a = 0;
+xa end 'xid';
+xa prepare 'xid';
+xa commit 'xid';
+set debug_sync= 'now signal go';
+
+--connection default
+--reap
+select * from t;
+
+drop table t;
+
+
 eval set global default_storage_engine= $default_storage_engine;
 
 --disconnect con1

sql/online_alter.cc

@@ -315,17 +315,39 @@ int online_alter_savepoint_rollback(handlerton *hton, THD *thd, sv_id_t sv_id)
 
 static int online_alter_commit(handlerton *hton, THD *thd, bool all)
 {
-  int res= online_alter_end_trans(get_cache_list(hton, thd), thd,
-                                  ending_trans(thd, all), true);
+  int res;
+  bool is_ending_transaction= ending_trans(thd, all);
+  if (is_ending_transaction 
+      && thd->transaction->xid_state.get_state_code() == XA_PREPARED)
+  {
+    res= hton->commit_by_xid(hton, thd->transaction->xid_state.get_xid());
+    // cleanup was already done by prepare()
+  }
+  else
+  {
+    res= online_alter_end_trans(get_cache_list(hton, thd), thd,
+                                is_ending_transaction, true);
     cleanup_tables(thd);
+  }
   return res;
 };
 
 static int online_alter_rollback(handlerton *hton, THD *thd, bool all)
 {
-  int res= online_alter_end_trans(get_cache_list(hton, thd), thd,
-                                  ending_trans(thd, all), false);
+  int res;
+  bool is_ending_transaction= ending_trans(thd, all);
+  if (is_ending_transaction
+      && thd->transaction->xid_state.get_state_code() == XA_PREPARED)
+  {
+    res= hton->rollback_by_xid(hton, thd->transaction->xid_state.get_xid());
+    // cleanup was already done by prepare()
+  }
+  else
+  {
+    res= online_alter_end_trans(get_cache_list(hton, thd), thd,
+                                is_ending_transaction, false);
     cleanup_tables(thd);
+  }
   return res;
 };