MDEV-34705: Binlog in Engine: Fix re-using ids for binlog tablespaces

Before creating the next binlog tablespace N+2, flush out and close the old
binlog tablespace N, so that the new tablespace can re-use the tablespace
id without conflict.
Signed-off-by: Kristian Nielsen <knielsen@knielsen-hq.org>

mysql-test/suite/binlog/t/binlog_in_engine.test

@@ -14,11 +14,15 @@ SELECT @@GLOBAL.binlog_checksum;
 CREATE TABLE t2 (a INT PRIMARY KEY, b VARCHAR(2048)) ENGINE=InnoDB;
 
 SET SESSION binlog_format= ROW;
+--let num_trx= 1500
+--echo *** Do $num_trx transactions ...
+--disable_query_log
 --let $i= 0
-while ($i < 500) {
+while ($i < 1500) {
   eval INSERT INTO t2 VALUES ($i, REPEAT("x", 2048));
   inc $i;
 }
+--enable_query_log
 SET SESSION binlog_format= MIXED;
 
 DROP TABLE t2;

storage/innobase/fsp/fsp0fsp.cc

@@ -26,6 +26,7 @@ Created 11/29/1995 Heikki Tuuri
 
 #include "fsp0fsp.h"
 #include "buf0buf.h"
+#include "buf0flu.h"
 #include "fil0fil.h"
 #include "fil0crypt.h"
 #include "mtr0log.h"
@@ -3776,8 +3777,53 @@ uint32_t binlog_cur_page_offset= FIL_PAGE_DATA;
 /* ToDo: This needs to discover existing binlogs and start from the next free index. */
 uint64_t binlog_file_no= 0;
 
+
+/* '.' + '/' + "binlog" + '-' + (<=20 digits) + '.' + "ibb" + '\0'. */
+#define BINLOG_NAME_LEN 1 + 1 + 6 + 1 + 20 + 1 + 3 + 1
+static inline void
+binlog_name_make(char name_buf[BINLOG_NAME_LEN], uint64_t file_no)
+{
+  sprintf(name_buf, "./binlog-%06" PRIu64 ".ibb", file_no);
+}
+
+
+/** Write out all pages, flush, and close/detach a binlog tablespace.
+@param[in] file_no	 Index of the binlog tablespace
+@return DB_SUCCESS or error code */
+static dberr_t
+fsp_binlog_tablespace_close(uint64_t file_no)
+{
+  dberr_t res;
+
+  uint32_t space_id= SRV_SPACE_ID_BINLOG0 + (file_no & 1);
+  mysql_mutex_lock(&fil_system.mutex);
+  fil_space_t *space= fil_space_get_by_id(space_id);
+  mysql_mutex_unlock(&fil_system.mutex);
+  if (!space) {
+    res= DB_ERROR;
+    goto end;
+  }
+
+  /*
+    Write out any remaining pages in the buffer pool to the binlog tablespace.
+    Then flush the file to disk, and close the old tablespace.
+  */
+  while (buf_flush_list_space(space))
+    ;
+  os_aio_wait_until_no_pending_writes(false);
+  space->flush<false>();
+  mysql_mutex_lock(&fil_system.mutex);
+  fil_system.detach(space, false);
+  mysql_mutex_unlock(&fil_system.mutex);
+
+  res= DB_SUCCESS;
+end:
+  return res;
+}
+
+
 /** Create a binlog tablespace file
-@param[in] name	 file name
+@param[in] file_no	 Index of the binlog tablespace
 @return DB_SUCCESS or error code */
 dberr_t fsp_binlog_tablespace_create(uint64_t file_no)
 {
@@ -3788,8 +3834,8 @@ dberr_t fsp_binlog_tablespace_create(uint64_t file_no)
 	if(srv_read_only_mode)
 		return DB_ERROR;
 
-        char name[1 + 1 + 6 + 1 + 20 + 1 + 3 + 1];
-        sprintf(name, "./binlog-%06" PRIu64 ".ibb", file_no);
+        char name[BINLOG_NAME_LEN];
+        binlog_name_make(name, file_no);
 
 	os_file_create_subdirs_if_needed(name);
 
@@ -3909,6 +3955,23 @@ void fsp_binlog_write_cache(IO_CACHE *cache, size_t main_size, mtr_t *mtr)
   while (remain > 0) {
     if (page_offset == FIL_PAGE_DATA) {
       if (UNIV_UNLIKELY(!space) || page_no >= space->size) {
+        /*
+          Flush out to disk and close the old (N-2) tablespace so that we can
+          re-use its tablespace id for the new one.
+
+          ToDo: Start this operation in the background as soon as the (N-2)
+          tablespace has been filled, to avoid stalling here.
+
+          ToDo: Handle recovery. Idea: write the current LSN at the start of
+          the binlog tablespace when we create it. At recovery, we should open
+          the (at most) 2 most recent binlog tablespaces. Whenever we have a
+          redo record, skip it if its LSN is smaller than the one stored in the
+          tablespace corresponding to its space_id. This way, it should be safe
+          to re-use tablespace ids between just two, SRV_SPACE_ID_BINLOG0 and
+          SRV_SPACE_ID_BINLOG1.
+        */
+        if (binlog_file_no >= 2)
+          fsp_binlog_tablespace_close(binlog_file_no - 2);
         /*
           Create a new binlog tablespace.
           ToDo: pre-create the next tablespace in a background thread, avoiding