MDEV-31646 preserve DMLs in case of online binlog fault

A 32-bit server build has a limitation of 2^32 bytes max for IO_CACHE.
This is quite a reachable value for a single transaction.

If DML reaches it, and the engine is not rollback-capable, then the last
record operation will be lost after ALTER TABLE.

To avoid it, report the error to the ALTER TABLE side and ignore it on the
DML side. Thus, a DML will not fail because of an online alter log failure.

mysql-test/main/alter_table_online_debug.result

@@ -1367,7 +1367,7 @@ NOT FOUND /Slave SQL/ in mysqld.1.err
 # MDEV-31646 Online alter applies binlog cache limit to cache writes
 #
 create table t (pk int primary key, a varchar(100)) engine=MyISAM;
-insert into t select seq, repeat('x', 100) from seq_1_to_500;
+insert into t select seq, repeat('x', 100) from seq_1_to_50;
 set @cache.size= @@max_binlog_cache_size;
 set global max_binlog_cache_size= 4096;
 set debug_sync= 'now wait_for do_updates';
@@ -1385,29 +1385,56 @@ Level	Code	Message
 connection default;
 drop table t;
 set debug_sync= reset;
+set global max_binlog_cache_size= @cache.size;
 # Now make sure that smaller limits will be processed fine
-set @old_dbug=@@debug_dbug;
-set debug_dbug="+d,online_alter_small_cache";
+set debug_dbug="+d,online_alter_small_cache_2";
 create table t (pk int primary key, a varchar(100)) engine=MyISAM;
-insert into t select seq, repeat('x', 100) from seq_1_to_500;
-set @cache.size= @@max_binlog_cache_size;
-set global max_binlog_cache_size= 4096;
+insert into t select seq, repeat('x', 100) from seq_1_to_50;
 connection con1;
-set debug_sync= 'alter_table_online_progress signal do_updates wait_for go';
+set debug_dbug="+d,online_alter_small_cache_2";
+set debug_sync= 'alter_table_online_before_lock signal do_updates wait_for go';
 alter table t add b int, algorithm=copy, lock=none;
 connection default;
 set debug_sync= 'now wait_for do_updates';
 update t set a = repeat('y', 100);
-ERROR HY000: Multi-statement transaction required more than 'max_binlog_cache_size' bytes of storage; increase this mariadbd variable and try again
+Warnings:
+Warning	1026	Error writing file 'online-alter-binlog' (errno: 27 "File too large")
 show warnings;
 Level	Code	Message
-Error	1197	Multi-statement transaction required more than 'max_binlog_cache_size' bytes of storage; increase this mariadbd variable and try again
-Error	1534	Writing one row to the row-based binary log failed
-Warning	1196	Some non-transactional changed tables couldn't be rolled back
+Warning	1026	Error writing file 'online-alter-binlog' (errno: 27 "File too large")
 set debug_sync= 'now signal go';
 connection con1;
+ERROR HY000: IO Write error: (27, File too large) 
 show warnings;
 Level	Code	Message
+Error	1811	IO Write error: (27, File too large) 
+set debug_dbug= @old_dbug;
+connection default;
+drop table t;
+set debug_sync= reset;
+set debug_dbug= @old_dbug;
+set debug_dbug="+d,online_alter_small_cache_1";
+create table t (pk int primary key, a varchar(100)) engine=MyISAM;
+insert into t select seq, repeat('x', 100) from seq_1_to_50;
+connection con1;
+set debug_dbug="+d,online_alter_small_cache_1";
+set debug_sync= 'alter_table_online_before_lock signal do_updates wait_for go';
+alter table t add b int, algorithm=copy, lock=none;
+connection default;
+set debug_sync= 'now wait_for do_updates';
+update t set a = repeat('y', 100);
+Warnings:
+Warning	1026	Error writing file 'online-alter-binlog' (errno: 27 "File too large")
+show warnings;
+Level	Code	Message
+Warning	1026	Error writing file 'online-alter-binlog' (errno: 27 "File too large")
+set debug_sync= 'now signal go';
+connection con1;
+ERROR HY000: IO Write error: (27, File too large) 
+show warnings;
+Level	Code	Message
+Error	1811	IO Write error: (27, File too large) 
+set debug_dbug= @old_dbug;
 connection default;
 drop table t;
 set debug_sync= reset;

mysql-test/main/alter_table_online_debug.test

@@ -1561,7 +1561,7 @@ let SEARCH_PATTERN= Slave SQL;
 --echo # MDEV-31646 Online alter applies binlog cache limit to cache writes
 --echo #
 create table t (pk int primary key, a varchar(100)) engine=MyISAM;
-insert into t select seq, repeat('x', 100) from seq_1_to_500;
+insert into t select seq, repeat('x', 100) from seq_1_to_50;
 
 set @cache.size= @@max_binlog_cache_size;
 set global max_binlog_cache_size= 4096;
@@ -1586,38 +1586,39 @@ show warnings;
 --connection default
 drop table t;
 set debug_sync= reset;
+set global max_binlog_cache_size= @cache.size;
 
 --echo # Now make sure that smaller limits will be processed fine
-
-set @old_dbug=@@debug_dbug;
-set debug_dbug="+d,online_alter_small_cache";
-create table t (pk int primary key, a varchar(100)) engine=MyISAM;
-insert into t select seq, repeat('x', 100) from seq_1_to_500;
-
-set @cache.size= @@max_binlog_cache_size;
-set global max_binlog_cache_size= 4096;
-
---connection con1
-set debug_sync= 'alter_table_online_progress signal do_updates wait_for go';
---send
-alter table t add b int, algorithm=copy, lock=none;
-
---connection default
-set debug_sync= 'now wait_for do_updates';
---error ER_TRANS_CACHE_FULL
-update t set a = repeat('y', 100);
-show warnings;
-
-set debug_sync= 'now signal go';
-
---connection con1
---reap
-show warnings;
-
---connection default
-drop table t;
-set debug_sync= reset;
-set debug_dbug= @old_dbug;
+let $i=2;
+while ($i) {
+  eval set debug_dbug="+d,online_alter_small_cache_$i";
+  create table t (pk int primary key, a varchar(100)) engine=MyISAM;
+  insert into t select seq, repeat('x', 100) from seq_1_to_50;
+
+  --connection con1
+  eval set debug_dbug="+d,online_alter_small_cache_$i";
+  set debug_sync= 'alter_table_online_before_lock signal do_updates wait_for go';
+  send alter table t add b int, algorithm=copy, lock=none;
+
+  --connection default
+  set debug_sync= 'now wait_for do_updates';
+  update t set a = repeat('y', 100);
+  show warnings;
+
+  set debug_sync= 'now signal go';
+
+  --connection con1
+  --error ER_IO_WRITE_ERROR
+  --reap
+  show warnings;
+
+  set debug_dbug= @old_dbug;
+  --connection default
+  drop table t;
+  set debug_sync= reset;
+  set debug_dbug= @old_dbug;
+  dec $i;
+}
 
 --disconnect con1
 --disconnect con2

sql/log.cc

@@ -2273,7 +2273,7 @@ int binlog_log_row_online_alter(TABLE* table, const uchar *before_record,
 {
   THD *thd= table->in_use;
 
-  if (!table->online_alter_cache)
+  if (unlikely(!table->online_alter_cache))
   {
     table->online_alter_cache= online_alter_binlog_get_cache_data(thd, table);
     trans_register_ha(thd, false, binlog_hton, 0);
@@ -2284,6 +2284,9 @@ int binlog_log_row_online_alter(TABLE* table, const uchar *before_record,
   // We need to log all columns for the case if alter table changes primary key
   DBUG_ASSERT(!before_record || bitmap_is_set_all(table->read_set));
   MY_BITMAP *old_rpl_write_set= table->rpl_write_set;
+
+  Dummy_error_handler dummy_handler;
+  thd->push_internal_handler(&dummy_handler);
   table->rpl_write_set= &table->s->all_set;
 
   int error= (*log_func)(thd, table, table->s->online_alter_binlog,
@@ -2291,8 +2294,14 @@ int binlog_log_row_online_alter(TABLE* table, const uchar *before_record,
                          before_record, after_record);
 
   table->rpl_write_set= old_rpl_write_set;
+  thd->pop_internal_handler();
 
-  return unlikely(error) ? HA_ERR_RBR_LOGGING_FAILED : 0;
+  if (unlikely(error))
+    push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE, ER_DISK_FULL,
+                        "Broken online alter log. "
+                        "ALTER TABLE will finish with error.");
+
+  return 0;
 }
 
 static void
@@ -2487,7 +2496,7 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
     thd->reset_binlog_for_next_statement();
     DBUG_RETURN(error);
   }
-  if (!wsrep_emulate_bin_log && Event_log::check_write_error(thd))
+  if (!wsrep_emulate_bin_log && MYSQL_BIN_LOG::check_write_error(thd))
   {
     /*
       "all == true" means that a "rollback statement" triggered the error and
@@ -2552,7 +2561,7 @@ void binlog_reset_cache(THD *thd)
 }
 
 
-void Event_log::set_write_error(THD *thd, bool is_transactional)
+void MYSQL_BIN_LOG::set_write_error(THD *thd, bool is_transactional)
 {
   DBUG_ENTER("MYSQL_BIN_LOG::set_write_error");
 
@@ -2592,7 +2601,7 @@ void Event_log::set_write_error(THD *thd, bool is_transactional)
   DBUG_VOID_RETURN;
 }
 
-bool Event_log::check_write_error(THD *thd)
+bool MYSQL_BIN_LOG::check_write_error(THD *thd)
 {
   DBUG_ENTER("MYSQL_BIN_LOG::check_write_error");
 
@@ -3806,9 +3815,9 @@ bool MYSQL_BIN_LOG::open_index_file(const char *index_file_name_arg,
 }
 
 
-bool Event_log::open(enum cache_type io_cache_type_arg)
+bool Event_log::open(enum cache_type io_cache_type_arg, size_t buffer_size)
 {
-  bool error= init_io_cache(&log_file, -1, LOG_BIN_IO_SIZE, io_cache_type_arg,
+  bool error= init_io_cache(&log_file, -1, buffer_size, io_cache_type_arg,
                             0, 0, MYF(MY_WME | MY_NABP | MY_WAIT_IF_FULL));
 
   log_state= LOG_OPENED;
@@ -6395,9 +6404,14 @@ static online_alter_cache_data *binlog_setup_cache_data(MEM_ROOT *root, TABLE_SH
 {
   static ulong online_alter_cache_use= 0, online_alter_cache_disk_use= 0;
 
+  my_off_t file_size= SIZE_T_MAX; // maximum possible cache size
+  size_t cache_size= binlog_cache_size;
+  DBUG_EXECUTE_IF("online_alter_small_cache_1",
+                  cache_size= file_size= IO_SIZE;);
+
   auto cache= new (root) online_alter_cache_data();
   if (!cache || open_cached_file(&cache->cache_log, mysql_tmpdir,
-                         LOG_PREFIX, (size_t)binlog_cache_size, MYF(MY_WME)))
+                         LOG_PREFIX, cache_size, MYF(MY_WME)))
   {
     delete cache;
     return NULL;
@@ -6407,10 +6421,8 @@ static online_alter_cache_data *binlog_setup_cache_data(MEM_ROOT *root, TABLE_SH
   cache->hton= share->db_type();
   cache->sink_log= share->online_alter_binlog;
 
-  my_off_t binlog_max_size= SIZE_T_MAX; // maximum possible cache size
-  DBUG_EXECUTE_IF("online_alter_small_cache", binlog_max_size= 4096;);
 
-  cache->set_binlog_cache_info(binlog_max_size,
+  cache->set_binlog_cache_info(file_size,
                                &online_alter_cache_use,
                                &online_alter_cache_disk_use);
   cache->store_prev_position();
@@ -6550,7 +6562,8 @@ Event_log::flush_and_set_pending_rows_event(THD *thd, Rows_log_event* event,
     if (writer.write(pending))
     {
       set_write_error(thd, is_transactional);
-      if (check_write_error(thd) && cache_data &&
+      if (dynamic_cast<MYSQL_BIN_LOG*>(this) &&
+          MYSQL_BIN_LOG::check_write_error(thd) && cache_data &&
           stmt_has_updated_non_trans_table(thd))
         cache_data->set_incident();
       delete pending;
@@ -7696,7 +7709,6 @@ class CacheWriter: public Log_event_writer
 static int binlog_online_alter_end_trans(THD *thd, bool all, bool commit)
 {
   DBUG_ENTER("binlog_online_alter_end_trans");
-  int error= 0;
 #ifdef HAVE_REPLICATION
   if (thd->online_alter_cache_list.empty())
     DBUG_RETURN(0);
@@ -7705,12 +7717,22 @@ static int binlog_online_alter_end_trans(THD *thd, bool all, bool commit)
 
   for (auto &cache: thd->online_alter_cache_list)
   {
+    int error= 0;
     auto *binlog= cache.sink_log;
     DBUG_ASSERT(binlog);
+
+    if (binlog->get_write_error())
+      continue;
+
     bool non_trans= cache.hton->flags & HTON_NO_ROLLBACK // Aria
                     || !cache.hton->rollback;
     bool do_commit= (commit && is_ending_transaction) || non_trans;
 
+    // Ignore any potential error on the DML side. It should be handled by
+    // the ALTER TABLE side.
+    Dummy_error_handler dh;
+    thd->push_internal_handler(&dh);
+
     if (commit || non_trans)
     {
       // Do not set STMT_END for last event to leave table open in altering thd
@@ -7729,6 +7751,9 @@ static int binlog_online_alter_end_trans(THD *thd, bool all, bool commit)
         mysql_mutex_lock(binlog->get_log_lock());
         error= binlog->write_cache_raw(thd, &cache.cache_log);
         mysql_mutex_unlock(binlog->get_log_lock());
+
+        if (unlikely(error))
+          binlog->set_write_error(my_errno);
       }
     }
     else if (!commit) // rollback
@@ -7742,14 +7767,14 @@ static int binlog_online_alter_end_trans(THD *thd, bool all, bool commit)
       cache.store_prev_position();
     }
 
+    thd->pop_internal_handler();
 
     if (error)
     {
-      my_error(ER_ERROR_ON_WRITE, MYF(ME_ERROR_LOG),
-               binlog->get_name(), errno);
-      binlog_online_alter_cleanup(thd->online_alter_cache_list,
-                                  is_ending_transaction);
-      DBUG_RETURN(error);
+      push_warning_printf(thd, Sql_state_errno_level::WARN_LEVEL_WARN,
+                          ER_ERROR_ON_WRITE, ER_THD(thd, ER_ERROR_ON_WRITE),
+                          binlog->get_name(), errno);
+      DBUG_ASSERT(binlog->get_write_error());
     }
   }
 
@@ -7759,7 +7784,7 @@ static int binlog_online_alter_end_trans(THD *thd, bool all, bool commit)
   for (TABLE *table= thd->open_tables; table; table= table->next)
     table->online_alter_cache= NULL;
 #endif // HAVE_REPLICATION
-  DBUG_RETURN(error);
+  DBUG_RETURN(0);
 }
 
 SAVEPOINT** find_savepoint_in_list(THD *thd, LEX_CSTRING name,
@@ -12366,6 +12391,11 @@ get_gtid_list_event(IO_CACHE *cache, Gtid_list_log_event **out_gtid_list)
   return errormsg;
 }
 
+void Cache_flip_event_log::set_write_error(THD *thd, bool is_transactional)
+{
+  set_write_error(my_errno);
+}
+
 
 struct st_mysql_storage_engine binlog_storage_engine=
 { MYSQL_HANDLERTON_INTERFACE_VERSION };

sql/log.h

@@ -400,8 +400,7 @@ class Event_log: public MYSQL_LOG
   int flush_and_set_pending_rows_event(THD *thd, Rows_log_event* event,
                                        binlog_cache_data *cache_data,
                                        bool is_transactional);
-  void set_write_error(THD *thd, bool is_transactional);
-  static bool check_write_error(THD *thd);
+  virtual void set_write_error(THD *thd, bool is_transactional) = 0;
   int write_cache(THD *thd, IO_CACHE *cache);
   int write_cache_raw(THD *thd, IO_CACHE *cache);
   char* get_name() { return name; }
@@ -420,7 +419,8 @@ class Event_log: public MYSQL_LOG
                      MY_MUTEX_INIT_SLOW);
   }
 
-  bool open(enum cache_type io_cache_type_arg);
+  bool open(enum cache_type io_cache_type_arg,
+            size_t buffer_size= LOG_BIN_IO_SIZE);
   virtual IO_CACHE *get_log_file() { return &log_file; }
 
   longlong write_description_event(enum_binlog_checksum_alg checksum_alg,
@@ -445,17 +445,20 @@ class Event_log: public MYSQL_LOG
  */
 class Cache_flip_event_log: public Event_log {
   IO_CACHE alt_buf;
+IF_DBUG(public:,)
   IO_CACHE *current, *alt;
   std::atomic<uint> ref_count;
+  std::atomic<int> online_write_error;
 public:
 
   Cache_flip_event_log() : Event_log(), alt_buf{},
-                           current(&log_file), alt(&alt_buf), ref_count(1) {}
-  bool open(enum cache_type io_cache_type_arg)
+                           current(&log_file), alt(&alt_buf), ref_count(1),
+                           online_write_error(0) {}
+  bool open(size_t buffer_size)
   {
     log_file.dir= mysql_tmpdir;
     alt_buf.dir= log_file.dir;
-    bool res= Event_log::open(io_cache_type_arg);
+    bool res= Event_log::open(WRITE_CACHE);
     if (res)
       return res;
 
@@ -464,7 +467,7 @@ class Cache_flip_event_log: public Event_log {
     if (!name)
       return false;
 
-    res= init_io_cache(&alt_buf, -1, LOG_BIN_IO_SIZE, io_cache_type_arg, 0, 0,
+    res= init_io_cache(&alt_buf, -1, buffer_size, WRITE_CACHE, 0, 0,
                        MYF(MY_WME | MY_NABP | MY_WAIT_IF_FULL)) != 0;
     return res;
   }
@@ -509,6 +512,17 @@ class Cache_flip_event_log: public Event_log {
       delete this;
     }
   }
+  void set_write_error(THD *thd, bool is_transactional) override;
+
+  void set_write_error(int error)
+  {
+    online_write_error.store(error, std::memory_order_relaxed);
+  }
+
+  int get_write_error() const
+  {
+    return online_write_error.load(std::memory_order_relaxed);
+  }
 
 private:
   void cleanup()
@@ -986,6 +1000,8 @@ class MYSQL_BIN_LOG: public TC_LOG, private Event_log
   bool write_incident(THD *thd);
   void write_binlog_checkpoint_event_already_locked(const char *name, uint len);
   bool write_table_map(THD *thd, TABLE *table, bool with_annotate);
+  void set_write_error(THD *thd, bool is_transactional) override;
+  static bool check_write_error(THD *thd);
   void start_union_events(THD *thd, query_id_t query_id_param);
   void stop_union_events(THD *thd);
   bool is_query_in_union(THD *thd, query_id_t query_id_param);

sql/sql_table.cc

@@ -11662,6 +11662,14 @@ static int online_alter_read_from_binlog(THD *thd, rpl_group_info *rgi,
   thd->push_internal_handler(&hdeh);
   do
   {
+    error= log->get_write_error();
+    if (unlikely(error))
+    {
+      my_error(ER_IO_WRITE_ERROR, MYF(0), (ulong)error, strerror(error), "");
+      error= 1;
+      break;
+    }
+
     const auto *descr_event= rgi->rli->relay_log.description_event_for_exec;
     auto *ev= Log_event::read_log_event(log_file, descr_event, false);
     error= log_file->error;
@@ -11889,7 +11897,10 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
       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);
+
+      size_t buffer_size= LOG_BIN_IO_SIZE;
+      DBUG_EXECUTE_IF("online_alter_small_cache_2", buffer_size= IO_SIZE;);
+      error= from->s->online_alter_binlog->open(buffer_size);
 
       if (error)
       {
@@ -12099,6 +12110,9 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
 
     error= online_alter_read_from_binlog(thd, &rgi, binlog);
 
+    // flip() makes reinit_io_cache, so it should be here
+    DBUG_EXECUTE_IF("online_alter_small_cache_2",
+                  from->s->online_alter_binlog->current->end_of_file= IO_SIZE;);
     DEBUG_SYNC(thd, "alter_table_online_before_lock");
 
     int lock_error=