MDEV-34705: Binlog in Engine: Allocate next binlog tablespace as needed.

Only works for two tablespace files though. For the third, we need to
implement closing the first one, so that the tablespace id can be reused.
Signed-off-by: Kristian Nielsen <knielsen@knielsen-hq.org>

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

@@ -10,3 +10,15 @@ COMMIT;
 SELECT * FROM t1 ORDER BY a;
 DROP TABLE t1;
 SELECT @@GLOBAL.binlog_checksum;
+
+CREATE TABLE t2 (a INT PRIMARY KEY, b VARCHAR(2048)) ENGINE=InnoDB;
+
+SET SESSION binlog_format= ROW;
+--let $i= 0
+while ($i < 500) {
+  eval INSERT INTO t2 VALUES ($i, REPEAT("x", 2048));
+  inc $i;
+}
+SET SESSION binlog_format= MIXED;
+
+DROP TABLE t2;

storage/innobase/fsp/fsp0fsp.cc

@@ -490,7 +490,8 @@ void fil_space_t::modify_check(const mtr_t& mtr) const
   default:
     /* We may only write redo log for a persistent tablespace. */
     ut_ad(purpose == FIL_TYPE_TABLESPACE);
-    ut_ad(mtr.is_named_space(id));
+    ut_ad(mtr.is_named_space(id) ||
+          id == SRV_SPACE_ID_BINLOG0 || id == SRV_SPACE_ID_BINLOG1);
   }
 }
 #endif
@@ -3770,12 +3771,15 @@ void fsp_shrink_temp_space()
 fil_space_t* binlog_space;
 buf_block_t *binlog_cur_block;
 uint32_t binlog_cur_page_no;
-uint32_t binlog_cur_page_offset;
+/* ToDo: This needs to be initialized to the current point in any existing binlog. */
+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;
 
 /** Create a binlog tablespace file
 @param[in] name	 file name
 @return DB_SUCCESS or error code */
-dberr_t fsp_binlog_tablespace_create(const char* name)
+dberr_t fsp_binlog_tablespace_create(uint64_t file_no)
 {
 	pfs_os_file_t	fh;
 	bool		ret;
@@ -3784,6 +3788,9 @@ dberr_t fsp_binlog_tablespace_create(const char* name)
 	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);
+
 	os_file_create_subdirs_if_needed(name);
 
 	/* ToDo: Do we need here an mtr.log_file_op(FILE_CREATE) like in fil_ibd_create(()? */
@@ -3812,7 +3819,8 @@ dberr_t fsp_binlog_tablespace_create(const char* name)
 	}
 
 	mysql_mutex_lock(&fil_system.mutex);
-	uint32_t space_id= SRV_SPACE_ID_BINLOG0;
+        /* ToDo: Need to ensure file (N-2) is no longer active before creating (N). */
+	uint32_t space_id= SRV_SPACE_ID_BINLOG0 + (file_no & 1);
 	if (!(binlog_space= fil_space_t::create(space_id,
                                                 ( FSP_FLAGS_FCRC32_MASK_MARKER |
 						  FSP_FLAGS_FCRC32_PAGE_SSIZE()),
@@ -3882,12 +3890,6 @@ void fsp_binlog_append(const uchar *data, uint32_t len, mtr_t *mtr)
 void fsp_binlog_write_cache(IO_CACHE *cache, size_t main_size, mtr_t *mtr)
 {
   fil_space_t *space= binlog_space;
-  if (!space) {
-    fsp_binlog_tablespace_create("./binlog-000000.ibb");
-    space= binlog_space;
-  }
-  /* ToDo: Is this set_named_space() needed / appropriate? */
-  mtr->set_named_space(space);
   const uint32_t page_end= (uint32_t)srv_page_size - FIL_PAGE_DATA_END;
   uint32_t page_no= binlog_cur_page_no;
   uint32_t page_offset= binlog_cur_page_offset;
@@ -3906,6 +3908,18 @@ void fsp_binlog_write_cache(IO_CACHE *cache, size_t main_size, mtr_t *mtr)
   size_t gtid_remain= remain - main_size;
   while (remain > 0) {
     if (page_offset == FIL_PAGE_DATA) {
+      if (UNIV_UNLIKELY(!space) || page_no >= space->size) {
+        /*
+          Create a new binlog tablespace.
+          ToDo: pre-create the next tablespace in a background thread, avoiding
+          a large stall here in the common case where pre-allocation is fast
+          enough.
+        */
+        dberr_t res2= fsp_binlog_tablespace_create(binlog_file_no++);
+        ut_a(res2 == DB_SUCCESS /* ToDo: Error handling. */);
+        space= binlog_space;
+        binlog_cur_page_no= page_no= 0;
+      }
       block= fsp_page_create(space, page_no, mtr);
     } else {
       dberr_t err;
@@ -3993,8 +4007,7 @@ void fsp_binlog_test(const uchar *data, uint32_t len)
   mtr_t mtr;
   mtr.start();
   if (!binlog_space)
-    fsp_binlog_tablespace_create("./binlog-000000.ibb");
-  mtr.set_named_space(binlog_space);
+    fsp_binlog_tablespace_create(0);
   fsp_binlog_append(data, len, &mtr);
   mtr.commit();
 }

storage/innobase/include/mtr0log.h

@@ -25,7 +25,7 @@ Mini-transaction log record encoding and decoding
 #include "mtr0mtr.h"
 
 /** The smallest invalid page identifier for persistent tablespaces */
-constexpr page_id_t end_page_id{SRV_SPACE_ID_BINLOG1, 0};
+constexpr page_id_t end_page_id{SRV_SPACE_ID_BINLOG1 + 1, 0};
 
 /** The minimum 2-byte integer (0b10xxxxxx xxxxxxxx) */
 constexpr uint32_t MIN_2BYTE= 1 << 7;
@@ -390,7 +390,9 @@ inline byte *mtr_t::log_write(const page_id_t id, const buf_page_t *bpage,
 {
   static_assert(!(type & 15) && type != RESERVED &&
                 type <= FILE_CHECKPOINT, "invalid type");
-  ut_ad(type >= FILE_CREATE || is_named_space(id.space()));
+  ut_ad(type >= FILE_CREATE || is_named_space(id.space()) ||
+        id.space() == SRV_SPACE_ID_BINLOG0 ||
+        id.space() == SRV_SPACE_ID_BINLOG1);
   ut_ad(!bpage || bpage->id() == id);
   ut_ad(id < end_page_id);
   constexpr bool have_len= type != INIT_PAGE && type != FREE_PAGE;

storage/innobase/mtr/mtr0mtr.cc

@@ -1568,7 +1568,9 @@ void mtr_t::set_modified(const buf_block_t &block)
 void mtr_t::init(buf_block_t *b)
 {
   const page_id_t id{b->page.id()};
-  ut_ad(is_named_space(id.space()));
+  ut_ad(is_named_space(id.space()) ||
+        id.space() == SRV_SPACE_ID_BINLOG0 ||
+        id.space() == SRV_SPACE_ID_BINLOG1);
   ut_ad(!m_freed_pages == !m_freed_space);
   ut_ad(memo_contains_flagged(b, MTR_MEMO_PAGE_X_FIX));