MDEV-34705: Binlog-in-engine: mariadb-backup integration

InnoDB binlog files are now backed up along with other InnoDB data by
mariadb-backup.

The files are copied after backup locks have been released. Backup files
created later than the backup LSN are skipped. Then during --prepare, any
data missing from the hot-copied binlog files will be restored by the
binlog recovery code, and any excess data written after the backup LSN will
be zeroed out.

A couple test cases test taking a consistent backup of a server with active
traffic during the backup, by provisioning a slave from the restored binlog
position and checking that the slave can replicate from the original master
and get identical data.

Signed-off-by: Kristian Nielsen <knielsen@knielsen-hq.org>

extra/mariabackup/backup_copy.cc

@@ -61,6 +61,7 @@ Street, Fifth Floor, Boston, MA 02110-1335 USA
 #include "backup_debug.h"
 #include "backup_mysql.h"
 #include <btr0btr.h>
+#include <innodb_binlog.h>
 #ifdef _WIN32
 #include <direct.h> /* rmdir */
 #endif
@@ -1668,6 +1669,7 @@ copy_back()
 	datadir_iter_t *it = NULL;
 	datadir_node_t node;
 	const char *dst_dir;
+	ds_ctxt *ds_binlogs = NULL;
 
 	memset(&node, 0, sizeof(node));
 
@@ -1793,6 +1795,10 @@ copy_back()
 
 	ds_destroy(ds_tmp);
 
+	/* Prepare destination directory for any InnoDB binlog files. */
+	dst_dir = dst_dir_buf.make(opt_binlog_directory);
+	ds_binlogs = ds_create(dst_dir, DS_TYPE_LOCAL);
+
 	/* copy the rest of tablespaces */
 	ds_tmp = ds_create(mysql_data_home, DS_TYPE_LOCAL);
 
@@ -1854,6 +1860,16 @@ copy_back()
 
 		filename = base_name(node.filepath);
 
+		/* Copy InnoDB binlog files into --binlog-directory. */
+		uint64_t file_no;
+		if (is_binlog_name(filename, &file_no)) {
+			if (!(ret = copy_or_move_file(ds_binlogs, filename, filename,
+						      dst_dir, 1))) {
+				goto cleanup;
+			}
+			continue;
+		}
+
 		/* skip .qp files */
 		if (filename_matches(filename, ext_list)) {
 			continue;
@@ -1914,6 +1930,11 @@ cleanup:
 
 	ds_tmp = NULL;
 
+	if (ds_binlogs != NULL) {
+		ds_destroy(ds_binlogs);
+		ds_binlogs = NULL;
+	}
+
 	return(ret);
 }
 

extra/mariabackup/backup_mysql.cc

@@ -384,6 +384,7 @@ bool get_mysql_vars(MYSQL *connection)
   char *aria_log_dir_path_var= NULL;
   char *page_zip_level_var= NULL;
   char *ignore_db_dirs= NULL;
+  char *binlog_directory_var= NULL;
   char *endptr;
   ulong server_version= mysql_get_server_version(connection);
 
@@ -410,6 +411,7 @@ bool get_mysql_vars(MYSQL *connection)
       {"innodb_compression_level", &page_zip_level_var},
       {"ignore_db_dirs", &ignore_db_dirs},
       {"aria_log_dir_path", &aria_log_dir_path_var},
+      {"binlog_directory", &binlog_directory_var},
       {NULL, NULL}};
 
   read_mysql_variables(connection, "SHOW VARIABLES", mysql_vars, true);
@@ -551,6 +553,10 @@ bool get_mysql_vars(MYSQL *connection)
   if (ignore_db_dirs)
     xb_load_list_string(ignore_db_dirs, ",", register_ignore_db_dirs_filter);
 
+  if (binlog_directory_var && *binlog_directory_var)
+    opt_binlog_directory= my_strdup(PSI_NOT_INSTRUMENTED, binlog_directory_var,
+                                    MYF(MY_FAE));
+
 out:
   free_mysql_variables(mysql_vars);
 

extra/mariabackup/common_engine.cc

@@ -9,6 +9,9 @@
 #include <memory>
 #include <chrono>
 
+#include "innodb_binlog.h"
+
+
 namespace common_engine {
 
 class Table {
@@ -298,17 +301,21 @@ class BackupImpl {
 		}
 		bool copy_log_tables(bool finalize);
 		bool copy_stats_tables();
+		bool copy_engine_binlogs(const char *binlog_dir, lsn_t backup_lsn);
 		bool wait_for_finish();
 		bool close_log_tables();
 	private:
 
 		void process_table_job(Table *table, bool no_lock, bool delete_table,
 			bool finalize, unsigned thread_num);
+		void process_binlog_job(std::string src, std::string dst,
+					lsn_t backup_lsn, unsigned thread_num);
 
 		const char *m_datadir_path;
 		ds_ctxt_t *m_ds;
 		std::vector<MYSQL *> &m_con_pool;
 		TasksGroup m_process_table_jobs;
+		std::unique_ptr<byte []> m_page_buf;
 
 		post_copy_table_hook_t m_table_post_copy_hook;
 		std::unordered_map<table_key_t, std::unique_ptr<LogTable>> m_log_tables;
@@ -337,6 +344,29 @@ exit:
 	m_process_table_jobs.finish_task(result);
 }
 
+void BackupImpl::process_binlog_job(std::string src, std::string dst,
+				    lsn_t backup_lsn, unsigned thread_num) {
+	int result = 0;
+	const char *c_src= src.c_str();
+	bool is_empty= true;
+	lsn_t start_lsn;
+	int binlog_found;
+
+	if (!m_process_table_jobs.get_result())
+		goto exit;
+
+	binlog_found= get_binlog_header(c_src, m_page_buf.get(), start_lsn, is_empty);
+	if (binlog_found > 0 && !is_empty && start_lsn <= backup_lsn) {
+		if (!m_ds->copy_file(c_src, dst.c_str(), thread_num))
+			goto exit;
+	}
+
+	result = 1;
+
+exit:
+	m_process_table_jobs.finish_task(result);
+}
+
 bool BackupImpl::scan(const std::unordered_set<table_key_t> &exclude_tables,
 	std::unordered_set<table_key_t> *out_processed_tables, bool no_lock,
 	bool collect_log_and_stats) {
@@ -461,6 +491,26 @@ bool BackupImpl::copy_stats_tables() {
 	return true;
 }
 
+bool BackupImpl::copy_engine_binlogs(const char *binlog_dir, lsn_t backup_lsn) {
+	std::vector<std::string>files;
+	std::string dir(binlog_dir && binlog_dir[0] ? binlog_dir : m_datadir_path);
+	foreach_file_in_datadir(dir.c_str(),
+				[&](const char *name)->bool {
+					uint64_t file_no;
+					if (is_binlog_name(name, &file_no))
+						files.emplace_back(name);
+					return true;
+				});
+	m_page_buf.reset(new byte [ibb_page_size]);
+	for (auto &file : files) {
+		std::string path(dir + "/" + file);
+		m_process_table_jobs.push_task(
+			std::bind(&BackupImpl::process_binlog_job, this, path,
+				  file, backup_lsn, std::placeholders::_1));
+	}
+	return true;
+}
+
 bool BackupImpl::wait_for_finish() {
 	/* Wait for threads to exit */
 	return m_process_table_jobs.wait_for_finish();
@@ -499,6 +549,10 @@ bool Backup::copy_stats_tables() {
 	return m_backup_impl->copy_stats_tables();
 }
 
+bool Backup::copy_engine_binlogs(const char *binlog_dir, lsn_t backup_lsn) {
+	return m_backup_impl->copy_engine_binlogs(binlog_dir, backup_lsn);
+}
+
 bool Backup::wait_for_finish() {
 	return m_backup_impl->wait_for_finish();
 }

extra/mariabackup/common_engine.h

@@ -28,6 +28,7 @@ class Backup {
 				bool no_lock, bool collect_log_and_stats);
 		bool copy_log_tables(bool finalize);
 		bool copy_stats_tables();
+		bool copy_engine_binlogs(const char *binlog_dir, lsn_t backup_lsn);
 		bool wait_for_finish();
 		bool close_log_tables();
 		void set_post_copy_table_hook(const post_copy_table_hook_t &hook);

extra/mariabackup/xtrabackup.cc

@@ -1386,7 +1386,8 @@ enum options_xtrabackup
   OPT_XB_IGNORE_INNODB_PAGE_CORRUPTION,
   OPT_INNODB_FORCE_RECOVERY,
   OPT_INNODB_CHECKPOINT,
-  OPT_ARIA_LOG_DIR_PATH
+  OPT_ARIA_LOG_DIR_PATH,
+  OPT_BINLOG_DIRECTORY
 };
 
 struct my_option xb_client_options[]= {
@@ -2053,6 +2054,12 @@ struct my_option xb_server_options[] =
      (G_PTR *) &xtrabackup_help, (G_PTR *) &xtrabackup_help, 0,
      GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
 
+    {"binlog-directory", OPT_BINLOG_DIRECTORY,
+   "Directory containing binlog files, if different from datadir."
+   "Has effect only if server is using --binlog-storage-engine=innodb",
+   &opt_binlog_directory, &opt_binlog_directory,
+   0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },
+
   { 0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
 };
 
@@ -2367,6 +2374,10 @@ xb_get_one_option(const struct my_option *opt,
     if (my_handle_options_init_variables)
       fprintf(stderr, "Obsolete option: %s. Ignored\n", opt->name);
     break;
+  case OPT_BINLOG_DIRECTORY:
+
+    ADD_PRINT_PARAM_OPT(opt_binlog_directory);
+    break;
 #define MYSQL_CLIENT
 #include "sslopt-case.h"
 #undef MYSQL_CLIENT
@@ -2553,6 +2564,10 @@ static bool innodb_init_param()
 		srv_undo_dir = (char*) ".";
 	}
 
+	if (!opt_binlog_directory || !xtrabackup_backup) {
+		opt_binlog_directory = (char *) ".";
+	}
+
 	compile_time_assert(SRV_FORCE_IGNORE_CORRUPT == 1);
 
 	/*
@@ -5396,6 +5411,17 @@ class BackupStages {
 					nullptr);
 			);
 
+			// Copy InnoDB binlog files.
+			if (!m_common_backup.copy_engine_binlogs(opt_binlog_directory,
+                                                                 recv_sys.lsn)) {
+				msg("Error on copy InnoDB binlog files");
+				return false;
+			}
+			if (!m_common_backup.wait_for_finish()) {
+				msg("InnoDB binlog file backup process is finished with error");
+				return false;
+			}
+
 			backup_finish(backup_datasinks.m_data);
 			return true;
 		}

mysql-test/suite/binlog_in_engine/include/mariabackup_slave_provision.inc

@@ -0,0 +1,105 @@
+--source include/have_innodb_binlog.inc
+# Test does a lot of queries that take a lot of CPU under Valgrind.
+--source include/not_valgrind.inc
+
+RESET MASTER;
+
+let $basedir=$MYSQLTEST_VARDIR/tmp/backup;
+
+CREATE TABLE t1(a varchar(60) PRIMARY KEY, b VARCHAR(60)) ENGINE INNODB;
+INSERT INTO t1 VALUES(1, NULL);
+CREATE TABLE t2 (val INT) ENGINE=InnoDB;
+INSERT INTO t2 VALUES (0);
+
+--disable_query_log
+--delimiter //
+CREATE PROCEDURE gen_load()
+  MODIFIES SQL DATA
+  BEGIN
+    DECLARE i INT;
+    DECLARE flag TYPE OF t2.val;
+    SET i = 0;
+    load_loop: LOOP
+      SELECT val INTO flag FROM t2;
+      IF NOT (flag=0) THEN
+        LEAVE load_loop;
+      END IF;
+      START TRANSACTION;
+      INSERT INTO t1 VALUES (CONCAT("AbAdCaFe", LPAD(i, 6, "0")), @@SESSION.last_gtid);
+      COMMIT;
+      SET i = i + 1;
+    END LOOP;
+  END
+//
+--delimiter ;
+--enable_query_log
+
+connect (con1,localhost,root,,);
+--echo *** Start a background load...
+send CALL gen_load();
+
+--connection default
+--echo *** Doing backup...
+--exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf --backup --target-dir=$basedir $backup_args
+--echo *** Doing prepare...
+--exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf --prepare --target-dir=$basedir
+
+--echo *** Stop the background load...
+UPDATE t2 SET val=1;
+--connection con1
+reap;
+--connection default
+disconnect con1;
+--let $count_master= `SELECT COUNT(*) FROM t1`
+
+--echo *** Provision a new slave from the backup
+--connect (server2,127.0.0.1,root,,,$SERVER_MYPORT_2)
+--let $datadir_2= `SELECT @@datadir`
+
+--echo *** Stopping provisioned server
+--source include/shutdown_mysqld.inc
+
+--echo *** Removing old datadir for provisioned server
+--rmdir $datadir_2
+
+--echo *** Provision new server from backup
+--exec $XTRABACKUP --copy-back --datadir=$datadir_2 --target-dir=$basedir $copy_back_args
+
+--let $restart_parameters= --skip-slave-start
+--source include/start_mysqld.inc
+
+--let $gtid_pos= `SELECT @@GLOBAL.gtid_binlog_pos`
+
+--replace_result $gtid_pos GTID_POS
+eval SET GLOBAL gtid_slave_pos= '$gtid_pos';
+--replace_result $SERVER_MYPORT_1 SERVER_MYPORT_1
+eval CHANGE MASTER TO
+  master_ssl_verify_server_cert=0,
+  master_port=$SERVER_MYPORT_1, master_host='127.0.0.1', master_user='root',
+  master_use_gtid= Slave_pos;
+START SLAVE;
+
+--connection default
+--source include/save_master_gtid.inc
+
+--connection server2
+--source include/sync_with_master_gtid.inc
+--let $count_slave= `SELECT COUNT(*) FROM t1`
+if ($count_master != $count_slave) {
+  --echo *** ERROR: Table on master has $count_master rows, but table on provisioned slave has $count_slave rows
+  --die Row difference on provisioned slave.
+}
+
+# Cleanup
+
+--connection server2
+STOP SLAVE;
+RESET SLAVE ALL;
+DROP PROCEDURE gen_load;
+DROP TABLE t1, t2;
+
+--connection default
+DROP PROCEDURE gen_load;
+DROP TABLE t1, t2;
+
+rmdir $basedir;

mysql-test/suite/binlog_in_engine/mariabackup_slave_provision_binlog_dir.cnf

@@ -0,0 +1,5 @@
+!include my.cnf
+
+[mysqld]
+binlog_directory=binlogs
+loose_innodb_log_file_size=96M

mysql-test/suite/binlog_in_engine/mariabackup_slave_provision_binlog_dir.result

@@ -0,0 +1,40 @@
+RESET MASTER;
+CREATE TABLE t1(a varchar(60) PRIMARY KEY, b VARCHAR(60)) ENGINE INNODB;
+INSERT INTO t1 VALUES(1, NULL);
+CREATE TABLE t2 (val INT) ENGINE=InnoDB;
+INSERT INTO t2 VALUES (0);
+connect  con1,localhost,root,,;
+*** Start a background load...
+CALL gen_load();
+connection default;
+*** Doing backup...
+*** Doing prepare...
+*** Stop the background load...
+UPDATE t2 SET val=1;
+connection con1;
+connection default;
+disconnect con1;
+*** Provision a new slave from the backup
+connect  server2,127.0.0.1,root,,,$SERVER_MYPORT_2;
+*** Stopping provisioned server
+*** Removing old datadir for provisioned server
+*** Provision new server from backup
+# restart: --skip-slave-start
+SET GLOBAL gtid_slave_pos= 'GTID_POS';
+CHANGE MASTER TO
+master_ssl_verify_server_cert=0,
+master_port=SERVER_MYPORT_1, master_host='127.0.0.1', master_user='root',
+master_use_gtid= Slave_pos;
+START SLAVE;
+connection default;
+include/save_master_gtid.inc
+connection server2;
+include/sync_with_master_gtid.inc
+connection server2;
+STOP SLAVE;
+RESET SLAVE ALL;
+DROP PROCEDURE gen_load;
+DROP TABLE t1, t2;
+connection default;
+DROP PROCEDURE gen_load;
+DROP TABLE t1, t2;

mysql-test/suite/binlog_in_engine/mariabackup_slave_provision_binlog_dir.test

@@ -0,0 +1,4 @@
+--source include/not_embedded.inc
+--let $backup_args=
+--let $copy_back_args= --binlog-directory=binlogs
+--source include/mariabackup_slave_provision.inc

mysql-test/suite/binlog_in_engine/mariabackup_slave_provision_nolock.cnf

@@ -0,0 +1,4 @@
+!include my.cnf
+
+[mysqld]
+loose_innodb_log_file_size=96M

mysql-test/suite/binlog_in_engine/mariabackup_slave_provision_nolock.result

@@ -0,0 +1,40 @@
+RESET MASTER;
+CREATE TABLE t1(a varchar(60) PRIMARY KEY, b VARCHAR(60)) ENGINE INNODB;
+INSERT INTO t1 VALUES(1, NULL);
+CREATE TABLE t2 (val INT) ENGINE=InnoDB;
+INSERT INTO t2 VALUES (0);
+connect  con1,localhost,root,,;
+*** Start a background load...
+CALL gen_load();
+connection default;
+*** Doing backup...
+*** Doing prepare...
+*** Stop the background load...
+UPDATE t2 SET val=1;
+connection con1;
+connection default;
+disconnect con1;
+*** Provision a new slave from the backup
+connect  server2,127.0.0.1,root,,,$SERVER_MYPORT_2;
+*** Stopping provisioned server
+*** Removing old datadir for provisioned server
+*** Provision new server from backup
+# restart: --skip-slave-start
+SET GLOBAL gtid_slave_pos= 'GTID_POS';
+CHANGE MASTER TO
+master_ssl_verify_server_cert=0,
+master_port=SERVER_MYPORT_1, master_host='127.0.0.1', master_user='root',
+master_use_gtid= Slave_pos;
+START SLAVE;
+connection default;
+include/save_master_gtid.inc
+connection server2;
+include/sync_with_master_gtid.inc
+connection server2;
+STOP SLAVE;
+RESET SLAVE ALL;
+DROP PROCEDURE gen_load;
+DROP TABLE t1, t2;
+connection default;
+DROP PROCEDURE gen_load;
+DROP TABLE t1, t2;

mysql-test/suite/binlog_in_engine/mariabackup_slave_provision_nolock.test

@@ -0,0 +1,4 @@
+--source include/not_embedded.inc
+--let $backup_args= --no-lock
+--let $copy_back_args=
+--source include/mariabackup_slave_provision.inc

mysql-test/suite/rpl/t/rpl_binlog_directory.test

@@ -24,6 +24,7 @@ INSERT INTO t1 VALUES (1, 10);
 --mkdir $master_datadir/binlog_dir
 --copy_file $master_datadir/master-bin.000001 $master_datadir/binlog_dir/master-bin.000001
 --copy_file $master_datadir/master-bin.000001.idx $master_datadir/binlog_dir/master-bin.000001.idx
+--move_file $master_datadir/master-bin.state $master_datadir/binlog_dir/master-bin.state
 --let $rpl_server_parameters= --binlog-directory=binlog_dir
 --source include/rpl_start_server.inc
 
@@ -31,6 +32,7 @@ INSERT INTO t1 VALUES (2, 11);
 
 # Move master back to using the standard binlog directory.
 --source include/rpl_stop_server.inc
+--move_file $master_datadir/binlog_dir/master-bin.state $master_datadir/master-bin.state
 --let $rpl_server_parameters=
 --source include/rpl_start_server.inc
 

storage/innobase/fsp/fsp_binlog.cc

@@ -1023,6 +1023,7 @@ fsp_log_binlog_write(mtr_t *mtr, fsp_binlog_page_entry *page,
 {
   uint64_t file_no= page->file_no;
   uint32_t page_no= page->page_no;
+  ut_ad(page->latched);
   if (page_offset + len >= ibb_page_size - BINLOG_PAGE_DATA_END)
     page->complete= true;
   if (page->flushed_clean)
@@ -1736,7 +1737,13 @@ read_more_data:
     if (0)
       static_assert(BINLOG_PAGE_DATA == 0,
                     "Replace static_assert with code from above comment");
-    else if (s.in_page_offset >= ibb_page_size - (BINLOG_PAGE_DATA_END + 3) ||
+
+    /* Check for end-of-file. */
+    if (cur_end_offset == ~(uint64_t)0 ||
+        (s.page_no << ibb_page_size_shift) + s.in_page_offset >= cur_end_offset)
+      return sofar;
+
+    if (s.in_page_offset >= ibb_page_size - (BINLOG_PAGE_DATA_END + 3) ||
              page_ptr[s.in_page_offset] == FSP_BINLOG_TYPE_FILLER)
     {
       ut_ad(s.in_page_offset >= ibb_page_size - BINLOG_PAGE_DATA_END ||
@@ -1744,11 +1751,6 @@ read_more_data:
       goto go_next_page;
     }
 
-    /* Check for end-of-file. */
-    if (cur_end_offset == ~(uint64_t)0 ||
-        (s.page_no << ibb_page_size_shift) + s.in_page_offset >= cur_end_offset)
-      return sofar;
-
     type= page_ptr[s.in_page_offset];
     if (type == 0)
     {
@@ -1853,7 +1855,8 @@ skip_chunk:
     s.skip_current= false;
   }
 
-  if (s.in_page_offset >= ibb_page_size - (BINLOG_PAGE_DATA_END + 3))
+  if (s.in_page_offset >= ibb_page_size - (BINLOG_PAGE_DATA_END + 3) &&
+      (s.page_no << ibb_page_size_shift) + s.in_page_offset < cur_end_offset)
   {
 go_next_page:
     /* End of page reached, move to the next page. */

storage/innobase/handler/innodb_binlog.cc

@@ -29,6 +29,7 @@ InnoDB implementation of binlog.
 #include "log0log.h"
 #include "small_vector.h"
 
+#include "mysys_err.h"
 #include "rpl_gtid_base.h"
 #include "handler.h"
 #include "log.h"
@@ -237,9 +238,6 @@ class ha_innodb_binlog_reader : public handler_binlog_reader {
   /* Used to read the header of the commit record. */
   byte rd_buf[5*COMPR_INT_MAX64];
 private:
-  int read_from_file(uint64_t end_offset, uchar *buf, uint32_t len);
-  int read_from_page(uchar *page_ptr, uint64_t end_offset,
-                     uchar *buf, uint32_t len);
   int read_data(uchar *buf, uint32_t len);
 
 public:
@@ -539,17 +537,15 @@ static int read_gtid_state_from_page(rpl_binlog_state_base *state,
     1  File found (but may be empty according to out_empty).
 */
 int
-binlog_recovery::get_header(uint64_t file_no, lsn_t &out_lsn, bool &out_empty)
+get_binlog_header(const char *binlog_path, byte *page_buf,
-  noexcept
+                  lsn_t &out_lsn, bool &out_empty) noexcept
 {
-  char full_path[OS_FILE_MAX_PATH];
   binlog_header_data header;
 
   out_empty= true;
   out_lsn= 0;
 
-  binlog_name_make(full_path, file_no, binlog_dir);
+  File fh= my_open(binlog_path, O_RDONLY | O_BINARY, MYF(0));
-  File fh= my_open(full_path, O_RDONLY | O_BINARY, MYF(0));
   if (fh < (File)0)
     return (my_errno == ENOENT ? 0 : -1);
   size_t read= my_pread(fh, page_buf, ibb_page_size, 0, MYF(0));
@@ -579,6 +575,16 @@ binlog_recovery::get_header(uint64_t file_no, lsn_t &out_lsn, bool &out_empty)
 }
 
 
+int
+binlog_recovery::get_header(uint64_t file_no, lsn_t &out_lsn, bool &out_empty)
+  noexcept
+{
+  char full_path[OS_FILE_MAX_PATH];
+  binlog_name_make(full_path, file_no, binlog_dir);
+  return get_binlog_header(full_path, page_buf, out_lsn, out_empty);
+}
+
+
 bool binlog_recovery::init_recovery(bool space_id, uint32_t page_no,
                                     uint16_t offset,
                                     lsn_t start_lsn, lsn_t end_lsn,
@@ -836,9 +842,24 @@ binlog_recovery::open_cur_file() noexcept
   if (cur_file_fh >= (File)0)
     my_close(cur_file_fh, MYF(0));
   binlog_name_make(full_path, cur_file_no, binlog_dir);
-  cur_file_fh= my_open(full_path, O_RDWR | O_BINARY, MYF(MY_WME));
+  cur_file_fh= my_open(full_path, O_RDWR | O_BINARY, MYF(0));
   if (cur_file_fh < (File)0)
-    return true;
+  {
+    /*
+      If we are on page 0 and the binlog file does not exist, then we should
+      create it (and recover its content).
+      Otherwise, it is an error, we cannot recover it as we are missing the
+      start of it.
+    */
+    if (my_errno != ENOENT ||
+        cur_page_no != 0 ||
+        (cur_file_fh= my_open(full_path, O_RDWR | O_CREAT | O_TRUNC |
+                              O_BINARY, MYF(0))) < (File)0)
+    {
+      my_error(EE_FILENOTFOUND, MYF(MY_WME), full_path, my_errno);
+      return true;
+    }
+  }
   cur_phys_size= (uint64_t)my_seek(cur_file_fh, 0, MY_SEEK_END, MYF(0));
   return false;
 }
@@ -934,13 +955,12 @@ binlog_recovery::close_file() noexcept
 bool
 binlog_recovery::next_file() noexcept
 {
-  if (flush_page())
+  if (cur_page_offset && flush_page())
     return true;
   if (close_file())
     return true;
   ++cur_file_no;
   cur_page_no= 0;
-  cur_page_offset= 0;
   return false;
 }
 
@@ -948,7 +968,7 @@ binlog_recovery::next_file() noexcept
 bool
 binlog_recovery::next_page() noexcept
 {
-  if (flush_page())
+  if (cur_page_offset && flush_page())
     return true;
   ++cur_page_no;
   return false;
@@ -1022,7 +1042,8 @@ binlog_recovery::apply_redo(bool space_id, uint32_t page_no, uint16_t offset,
   /* Test for moving to the next page. */
   else if (page_no != cur_page_no)
   {
-    if (cur_page_offset < ibb_page_size - BINLOG_PAGE_DATA_END &&
+    if (cur_page_offset > BINLOG_PAGE_DATA &&
+        cur_page_offset < ibb_page_size - BINLOG_PAGE_DATA_END &&
         !srv_force_recovery)
     {
       sql_print_error("InnoDB: Missing recovery record in file_no=%"
@@ -1114,7 +1135,7 @@ binlog_recovery::update_page_from_record(uint16_t offset,
   Check if this is an InnoDB binlog file name.
   Return the index/file_no if so.
 */
-static bool
+bool
 is_binlog_name(const char *name, uint64_t *out_idx)
 {
   const size_t base_len= sizeof(BINLOG_NAME_BASE) - 1;  // Length without '\0' terminator
@@ -1263,7 +1284,15 @@ innodb_binlog_init(size_t binlog_size, const char *directory)
   }
 
   start_binlog_prealloc_thread();
-  binlog_sync_initial();
+  if (res < 0)
+  {
+    /*
+      We are creating binlogs anew from scratch.
+      Write and fsync the initial file-header, so that recovery will know where
+      to start in case of a crash.
+    */
+    binlog_sync_initial();
+  }
 
   return false;
 }
@@ -1463,13 +1492,26 @@ find_pos_in_binlog(uint64_t file_no, size_t file_size, byte *page_buf,
     ut_a(p <= page_end);
   }
 
-  *out_page_no= p_0 - 1;
+  /*
-  *out_pos_in_page= (uint32_t)(p - page_buf);
+    Normalize the position, so that we store (page_no+1, BINLOG_PAGE_DATA)
-
+    and not (page_no, page_size - BINLOG_PAGE_DATA_END).
-  if (*out_pos_in_page >= page_size - BINLOG_PAGE_DATA_END)
+  */
-    ret= fsp_binlog_open(file_name, fh, file_no, file_size, p_0, nullptr);
+  byte *partial_page;
+  if (p == page_end)
+  {
+    *out_page_no= p_0;
+    *out_pos_in_page= BINLOG_PAGE_DATA;
+    partial_page= nullptr;
+  }
   else
-    ret= fsp_binlog_open(file_name, fh, file_no, file_size, p_0 - 1, page_buf);
+  {
+    *out_page_no= p_0 - 1;
+    *out_pos_in_page= (uint32_t)(p - page_buf);
+    partial_page= page_buf;
+  }
+
+  ret= fsp_binlog_open(file_name, fh, file_no, file_size,
+                       *out_page_no, partial_page);
   uint64_t pos= (*out_page_no << page_size_shift) | *out_pos_in_page;
   binlog_cur_written_offset[idx].store(pos, std::memory_order_relaxed);
   binlog_cur_end_offset[idx].store(pos, std::memory_order_relaxed);

storage/innobase/include/innodb_binlog.h

@@ -168,6 +168,9 @@ binlog_name_make_short(char *name_buf, uint64_t file_no)
 }
 
 
+extern bool is_binlog_name(const char *name, uint64_t *out_idx);
+extern int get_binlog_header(const char *binlog_path, byte *page_buf,
+                             lsn_t &out_lsn, bool &out_empty) noexcept;
 extern void innodb_binlog_startup_init();
 extern bool innodb_binlog_init(size_t binlog_size, const char *directory);
 extern void innodb_binlog_close(bool shutdown);