MDEV-31043 ER_KEY_NOT_FOUND upon concurrent ALTER and transaction

Non-transactional engines changes are visible immediately the row operation
succeeds, so we should apply the changes to the online buffer immediately
after the statement ends.

mysql-test/main/alter_table_online_debug.result

@@ -1251,8 +1251,54 @@ a
 2
 1
 drop table t;
+#
+# MDEV-31043 ER_KEY_NOT_FOUND upon concurrent ALTER and transaction
+#
+create table t (a int, b int) engine=myisam;
+insert into t values (1,10),(2,20);
+set debug_sync= 'alter_table_online_before_lock signal burnit wait_for goforit';
+alter table t add c int, algorithm=copy, lock=none;
+connection con1;
+set debug_sync= 'now wait_for burnit';
+update t set b = 100;
+start transaction;
+update t set b = 200;
+connection con2;
+delete from t order by a limit 1;
+delete from t order by a limit 1;
+select * from t;
+a	b
+connection con1;
+commit;
+set debug_sync= 'now signal goforit';
+connection default;
+select * from t;
+a	b	c
+drop table t;
+create table t (a int, b int) engine=aria;
+insert into t values (1,10),(2,20);
+set debug_sync= 'alter_table_online_before_lock signal burnit wait_for goforit';
+alter table t add c int, algorithm=copy, lock=none;
+connection con1;
+set debug_sync= 'now wait_for burnit';
+update t set b = 100;
+start transaction;
+update t set b = 200;
+connection con2;
+delete from t order by a limit 1;
+delete from t order by a limit 1;
+select * from t;
+a	b
+connection con1;
+commit;
+set debug_sync= 'now signal goforit';
+connection default;
+select * from t;
+a	b	c
+drop table t;
 set debug_sync= reset;
 disconnect con1;
+disconnect con2;
 #
 # End of 11.2 tests
 #

mysql-test/main/alter_table_online_debug.test

@@ -1440,8 +1440,44 @@ select * from t;
 
 drop table t;
 
+--echo #
+--echo # MDEV-31043 ER_KEY_NOT_FOUND upon concurrent ALTER and transaction
+--echo #
+let $i=2;
+let local_engine=myisam;
+while ($i) {
+eval create table t (a int, b int) engine=$local_engine;
+insert into t values (1,10),(2,20);
+set debug_sync= 'alter_table_online_before_lock signal burnit wait_for goforit';
+--send
+alter table t add c int, algorithm=copy, lock=none;
+
+--connection con1
+
+set debug_sync= 'now wait_for burnit';
+update t set b = 100;
+start transaction;
+update t set b = 200;
+
+--connection con2
+delete from t order by a limit 1;
+delete from t order by a limit 1;
+select * from t;
+--connection con1
+commit;
+set debug_sync= 'now signal goforit';
+
+--connection default
+--reap
+select * from t;
+drop table t;
+dec $i;
+let local_engine=aria;
+}
+
 set debug_sync= reset;
 --disconnect con1
+--disconnect con2
 --echo #
 --echo # End of 11.2 tests
 --echo #

sql/log.cc

@@ -489,7 +489,8 @@ private:
       delete pending();
       set_pending(0);
     }
-    reinit_io_cache(&cache_log, WRITE_CACHE, pos, 0, reset_cache);
+    my_bool res= reinit_io_cache(&cache_log, WRITE_CACHE, pos, 0, reset_cache);
+    DBUG_ASSERT(res == 0);
     cache_log.end_of_file= saved_max_binlog_cache_size;
   }
 
@@ -7727,24 +7728,41 @@ static int binlog_online_alter_end_trans(THD *thd, bool all, bool commit)
   {
     auto *binlog= cache.sink_log;
     DBUG_ASSERT(binlog);
+    bool non_trans= cache.hton->flags & HTON_NO_ROLLBACK // Aria
+                    || !cache.hton->rollback;
+    bool do_commit= (commit && is_ending_transaction) || non_trans;
+
+    if (commit || non_trans)
+    {
+      // Do not set STMT_END for last event to leave table open in altering thd
+      error= binlog_flush_pending_rows_event(thd, false, true, binlog, &cache);
+    }
 
-    bool do_commit= commit || !cache.hton->rollback ||
-                    cache.hton->flags & HTON_NO_ROLLBACK; // Aria
     if (do_commit)
     {
-      // do not set STMT_END for last event to leave table open in altering thd
+      /*
-      error= binlog_flush_pending_rows_event(thd, false, true, binlog, &cache);
+        If the cache wasn't reinited to write, then it remains empty after
-      if (is_ending_transaction)
+        the last write.
+      */
+      if (my_b_bytes_in_cache(&cache.cache_log) && likely(!error))
       {
+        DBUG_ASSERT(cache.cache_log.type != READ_CACHE);
         mysql_mutex_lock(binlog->get_log_lock());
         error= binlog->write_cache(thd, &cache.cache_log);
         mysql_mutex_unlock(binlog->get_log_lock());
       }
-      else
-        cache.store_prev_position();
     }
-    else if (!is_ending_transaction)
+    else if (!commit) // rollback
+    {
+      DBUG_ASSERT(!non_trans);
       cache.restore_prev_position();
+    }
+    else
+    {
+      DBUG_ASSERT(!is_ending_transaction);
+      cache.store_prev_position();
+    }
+
 
     if (error)
     {