control Cache_flip_event_log lifetime with reference count

If online alter fails, TABLE_SHARE can be freed while concurrent
transactions still have row events in their online_alter_cache_data.
On commit they try'll to flush them, writing to TABLE_SHARE's
Cache_flip_event_log, which is already freed.
This causes a crash in main.alter_table_online_debug test

sql/log.cc

@@ -507,7 +507,8 @@ class online_alter_cache_data: public Sql_alloc, public ilist_node<>,
     before_stmt_pos= my_b_write_tell(&cache_log);
   }
 
-  TABLE_SHARE *share;
+  handlerton *hton;
+  Cache_flip_event_log *sink_log;
   SAVEPOINT *sv_list;
 };
 
@@ -2302,6 +2303,7 @@ binlog_online_alter_cleanup(ilist<online_alter_cache_data> &list, bool ending_tr
     while (it != list.end())
     {
       auto &cache= *it++;
+      cache.sink_log->release();
       cache.reset();
       delete &cache;
     }
@@ -6424,21 +6426,24 @@ static online_alter_cache_data *binlog_setup_cache_data(MEM_ROOT *root, TABLE_SH
     return NULL;
   }
 
-  cache->share= share;
+  share->online_alter_binlog->acquire();
+  cache->hton= share->db_type();
+  cache->sink_log= share->online_alter_binlog;
   cache->set_binlog_cache_info(max_binlog_cache_size, &binlog_cache_use,
                                &binlog_cache_disk_use);
   cache->store_prev_position();
   return cache;
 }
 
+
 online_alter_cache_data *online_alter_binlog_get_cache_data(THD *thd, TABLE *table)
 {
   ilist<online_alter_cache_data> &list= thd->online_alter_cache_list;
 
   /* we assume it's very rare to have more than one online ALTER running */
   for (auto &cache: list)
   {
-    if (cache.share == table->s)
+    if (cache.sink_log == table->s->online_alter_binlog)
       return &cache;
   }
 
@@ -7719,11 +7724,11 @@ static int binlog_online_alter_end_trans(THD *thd, bool all, bool commit)
 
   for (auto &cache: thd->online_alter_cache_list)
   {
-    auto *binlog= cache.share->online_alter_binlog;
+    auto *binlog= cache.sink_log;
     DBUG_ASSERT(binlog);
 
-    bool do_commit= commit || !cache.share->db_type()->rollback ||
-                    cache.share->db_type()->flags & HTON_NO_ROLLBACK; // Aria
+    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
@@ -7778,7 +7783,7 @@ int online_alter_savepoint_set(THD *thd, LEX_CSTRING name)
 
   for (auto &cache: thd->online_alter_cache_list)
   {
-    if (cache.share->db_type()->savepoint_set == NULL)
+    if (cache.hton->savepoint_set == NULL)
       continue;
 
     SAVEPOINT *sv= savepoint_add(thd, name, &cache.sv_list, NULL);
@@ -7800,7 +7805,7 @@ int online_alter_savepoint_rollback(THD *thd, LEX_CSTRING name)
 #ifdef HAVE_REPLICATION
   for (auto &cache: thd->online_alter_cache_list)
   {
-    if (cache.share->db_type()->savepoint_set == NULL)
+    if (cache.hton->savepoint_set == NULL)
       continue;
 
     SAVEPOINT **sv= find_savepoint_in_list(thd, name, &cache.sv_list);

sql/log.h

@@ -442,13 +442,14 @@ class Event_log: public MYSQL_LOG
 
   TODO should be unnecessary after MDEV-24676 is done
  */
-class Cache_flip_event_log: public Event_log, public Sql_alloc {
+class Cache_flip_event_log: public Event_log {
   IO_CACHE alt_buf;
   IO_CACHE *current, *alt;
+  std::atomic<uint> ref_count;
 public:
 
   Cache_flip_event_log() : Event_log(), alt_buf{},
-                           current(&log_file), alt(&alt_buf) {}
+                           current(&log_file), alt(&alt_buf), ref_count(1) {}
   bool open(enum cache_type io_cache_type_arg)
   {
     log_file.dir= mysql_tmpdir;
@@ -490,6 +491,25 @@ class Cache_flip_event_log: public Event_log, public Sql_alloc {
     return current;
   }
 
+  void acquire()
+  {
+    IF_DBUG(auto prev= ,)
+    ref_count.fetch_add(1);
+    DBUG_ASSERT(prev != 0);
+  }
+
+  void release()
+  {
+    auto prev= ref_count.fetch_add(-1);
+
+    if (prev == 1)
+    {
+      cleanup();
+      delete this;
+    }
+  }
+
+private:
   void cleanup()
   {
     end_io_cache(&alt_buf);

sql/sql_table.cc

@@ -11795,15 +11795,15 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
 #ifdef HAVE_REPLICATION
     if (online)
     {
-      from->s->online_alter_binlog= new (&from->s->mem_root) Cache_flip_event_log();
+      from->s->online_alter_binlog= new Cache_flip_event_log();
       if (!from->s->online_alter_binlog)
         DBUG_RETURN(1);
       from->s->online_alter_binlog->init_pthread_objects();
       error= from->s->online_alter_binlog->open(WRITE_CACHE);
 
       if (error)
       {
-        from->s->online_alter_binlog->cleanup();
+        from->s->online_alter_binlog->release();
         from->s->online_alter_binlog= NULL;
         goto err;
       }
@@ -12014,6 +12014,19 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
   }
   else if (online) // error was on copy stage
   {
+    /*
+       We can't free the resources properly now, as we can still be in
+       non-exclusive state. So this s->online_alter_binlog will be used
+       until all transactions will release it.
+       Once the transaction commits, it can release online_alter_binlog
+       by decreasing ref_count.
+
+       online_alter_binlog->ref_count can be reached 0 only once.
+       Proof:
+       If share exists, we'll always have ref_count >= 1.
+       Once it reaches destroy(), nobody can acquire it again,
+       therefore, only release() is possible at this moment.
+    */
     from->s->tdc->flush_unused(1); // to free the binlog
   }
 #endif

sql/table.cc

@@ -508,7 +508,7 @@ void TABLE_SHARE::destroy()
 #ifdef HAVE_REPLICATION
   if (online_alter_binlog)
   {
-    online_alter_binlog->cleanup();
+    online_alter_binlog->release();
     online_alter_binlog= NULL;
   }
 #endif