MDEV-34705: Inplement starting from a specific GTID position

To find the target position, we first loop backwards over binlog files,
reading the initial GTID state written at the start to find the file to
start in. We then binary search on the differential GTID states written
every --innodb-binlog-state-interval bytes.

This patch does only minimal changes to the dump thread code in sql_repl.cc
to be able to send out binlog data to the client. Some re-factoring/cleanup
should be done in a follow-up patch to more cleanly separate the two code
paths, avoid a lot of if-statements and make the binlog-in-engine code path
free of much of the cruft from the legacy binlog implementation.

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

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

@@ -2,6 +2,7 @@
 --source include/have_binlog_format_mixed.inc
 
 CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE=InnoDB;
+--let $gtid_pos= `SELECT @@last_gtid`
 INSERT INTO t1 VALUES (1);
 BEGIN;
 INSERT INTO t1 VALUES (2);
@@ -25,6 +26,6 @@ while ($i < $num_trx) {
 --enable_query_log
 SET SESSION binlog_format= MIXED;
 
---exec $MYSQL_BINLOG --read-from-remote-server --user=root --host=127.0.0.1 --port=$MASTER_MYPORT master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog.txt
+--exec $MYSQL_BINLOG --read-from-remote-server --user=root --host=127.0.0.1 --port=$MASTER_MYPORT master-bin.000001 --start-position=$gtid_pos > $MYSQLTEST_VARDIR/tmp/mysqlbinlog.txt
 
 DROP TABLE t2;

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

@@ -11,11 +11,12 @@ while ($i < 10) {
   inc $i;
 }
 SELECT @@GLOBAL.gtid_binlog_state;
---exec $MYSQL_BINLOG --read-from-remote-server --user=root --host=127.0.0.1 --port=$MASTER_MYPORT master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog1.txt
+--exec $MYSQL_BINLOG --start-position=42-42-42 --gtid-strict-mode=0 --read-from-remote-server --user=root --host=127.0.0.1 --port=$MASTER_MYPORT master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog1.txt
 
 --source include/restart_mysqld.inc
 
 SELECT @@GLOBAL.gtid_binlog_state;
+--let $gtid_pos2= `SELECT @@GLOBAL.gtid_binlog_pos`
 INSERT INTO t1 VALUES (2, 0);
 let $i= 0;
 while ($i < 10) {
@@ -23,11 +24,12 @@ while ($i < 10) {
   inc $i;
 }
 SELECT @@GLOBAL.gtid_binlog_state;
---exec $MYSQL_BINLOG --read-from-remote-server --user=root --host=127.0.0.1 --port=$MASTER_MYPORT master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog2.txt
+--exec $MYSQL_BINLOG --start-position=$gtid_pos2 --read-from-remote-server --user=root --host=127.0.0.1 --port=$MASTER_MYPORT master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog2.txt
 
 --source include/restart_mysqld.inc
 
 SELECT @@GLOBAL.gtid_binlog_state;
+--let $gtid_pos3= `SELECT @@GLOBAL.gtid_binlog_pos`
 INSERT INTO t1 VALUES (3, 0);
 let $i= 0;
 while ($i < 10) {
@@ -35,6 +37,6 @@ while ($i < 10) {
   inc $i;
 }
 
---exec $MYSQL_BINLOG --read-from-remote-server --user=root --host=127.0.0.1 --port=$MASTER_MYPORT master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog3.txt
+--exec $MYSQL_BINLOG --start-position=$gtid_pos3 --read-from-remote-server --user=root --host=127.0.0.1 --port=$MASTER_MYPORT master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog3.txt
 
 DROP TABLE t1;

sql/handler.h

@@ -61,6 +61,8 @@ class Column_definition;
 class select_result;
 class handler_binlog_reader;
 struct rpl_gtid;
+struct slave_connection_state;
+struct rpl_binlog_state_base;
 
 // the following is for checking tables
 
@@ -5840,6 +5842,8 @@ public:
   { }
   virtual ~handler_binlog_reader() { };
   virtual int read_binlog_data(uchar *buf, uint32_t len) = 0;
+  virtual int init_gtid_pos(slave_connection_state *pos,
+                            rpl_binlog_state_base *state) = 0;
 
   int read_log_event(String *packet, uint32_t ev_offset, size_t max_allowed);
 

sql/log_event.h

@@ -2573,6 +2573,7 @@ public:
 #ifdef HAVE_REPLICATION
   void pack_info(Protocol* protocol) override;
 #endif /* HAVE_REPLICATION */
+  bool to_packet(String *packet);
 #else
   bool print(FILE* file, PRINT_EVENT_INFO* print_event_info) override;
 #endif

sql/log_event_server.cc

@@ -2401,22 +2401,28 @@ void Format_description_log_event::pack_info(Protocol *protocol)
 }
 #endif /* defined(HAVE_REPLICATION) */
 
-bool Format_description_log_event::write(Log_event_writer *writer)
+bool
+Format_description_log_event::to_packet(String *packet)
 {
-  bool ret;
+  uchar *p;
-  /*
+  uint32 needed_length=
-    We don't call Start_log_event_v::write() because this would make 2
+    packet->length() + START_V3_HEADER_LEN + 1 + number_of_event_types + 1;
-    my_b_safe_write().
+  if (packet->reserve(needed_length))
-  */
+    return true;
-  uchar buff[START_V3_HEADER_LEN+1];
+  p= (uchar *)packet->ptr() + packet->length();;
-  size_t rec_size= sizeof(buff) + BINLOG_CHECKSUM_ALG_DESC_LEN +
+  packet->length(needed_length);
-                   number_of_event_types;
+  int2store(p, binlog_version);
-  int2store(buff + ST_BINLOG_VER_OFFSET,binlog_version);
+  p += 2;
-  memcpy((char*) buff + ST_SERVER_VER_OFFSET,server_version,ST_SERVER_VER_LEN);
+  memcpy(p, server_version, ST_SERVER_VER_LEN);
+  p+= ST_SERVER_VER_LEN;
   if (!dont_set_created)
     created= get_time();
-  int4store(buff + ST_CREATED_OFFSET,created);
+  int4store(p, created);
-  buff[ST_COMMON_HEADER_LEN_OFFSET]= common_header_len;
+  p+= 4;
+  *p++= common_header_len;
+  memcpy(p, post_header_len, number_of_event_types);
+  p+= number_of_event_types;
+
   /*
     if checksum is requested
     record the checksum-algorithm descriptor next to
@@ -2441,13 +2447,31 @@ bool Format_description_log_event::write(Log_event_writer *writer)
      (A), (V) presence in FD of the checksum-aware server makes the event
      1 + 4 bytes bigger comparing to the former FD.
   */
+  *p++= checksum_byte;
+
+  return false;
+}
+
+bool Format_description_log_event::write(Log_event_writer *writer)
+{
+  bool ret;
+  /*
+    We don't call Start_log_event_v::write() because this would make 2
+    my_b_safe_write().
+  */
+  constexpr uint32_t needed= START_V3_HEADER_LEN + 1 + LOG_EVENT_TYPES + 1;
+  char buff[needed + 1];
+  String packet(buff, sizeof(buff), system_charset_info);
+  packet.length(0);
+  if (to_packet(&packet))
+    return true;
+  size_t rec_size= packet.length();
+  DBUG_ASSERT(needed == rec_size);
 
   uint orig_checksum_len= writer->checksum_len;
   writer->checksum_len= BINLOG_CHECKSUM_LEN;
   ret= write_header(writer, rec_size) ||
-       write_data(writer, buff, sizeof(buff)) ||
+       write_data(writer, packet.ptr(), packet.length()) ||
-       write_data(writer, post_header_len, number_of_event_types) ||
-       write_data(writer, &checksum_byte, sizeof(checksum_byte)) ||
        write_footer(writer);
   writer->checksum_len= orig_checksum_len;
   return ret;

sql/share/errmsg-utf8.txt

@@ -12294,3 +12294,5 @@ ER_PSEUDO_THREAD_ID_OVERWRITE
         eng "Pseudo thread id should not be modified by the client as it will be overwritten"
 ER_CANNOT_INIT_ENGINE_BINLOG_READER
         eng "Cannot initialize binlog reader from storage engine %s"
+ER_ENGINE_BINLOG_REQUIRES_GTID
+        eng "GTID starting position is required on master with --binlog-storage-engine enabled"

sql/sql_repl.cc

@@ -304,6 +304,52 @@ static int fake_gtid_list_event(binlog_send_info *info,
 }
 
 
+static int fake_format_description_event(binlog_send_info *info,
+                                         Format_description_log_event *fdev,
+                                         const char **errmsg,
+                                         uint32 current_pos)
+{
+  my_bool do_checksum;
+  int err;
+  ha_checksum crc;
+  char buf[320];
+  String str(buf, sizeof(buf), system_charset_info);
+  String* packet= info->packet;
+
+  str.length(0);
+  fdev->dont_set_created= true;
+  if (fdev->to_packet(&str))
+  {
+    info->error= ER_UNKNOWN_ERROR;
+    *errmsg= "Failed due to out-of-memory writing Format_description event";
+    return -1;
+  }
+  if ((err= fake_event_header(packet, FORMAT_DESCRIPTION_EVENT,
+                              str.length(), &do_checksum, &crc,
+                              errmsg, BINLOG_CHECKSUM_ALG_CRC32,
+                              current_pos)))
+  {
+    info->error= ER_UNKNOWN_ERROR;
+    return err;
+  }
+
+  packet->append(str);
+  if (do_checksum)
+  {
+    crc= my_checksum(crc, (uchar*)str.ptr(), str.length());
+  }
+
+  if ((err= fake_event_footer(packet, do_checksum, crc, errmsg)) ||
+      (err= fake_event_write(info->net, packet, errmsg)))
+  {
+    info->error= ER_UNKNOWN_ERROR;
+    return err;
+  }
+
+  return 0;
+}
+
+
 /*
   Reset thread transmit packet buffer for event sending
 
@@ -1317,6 +1363,11 @@ end:
 }
 
 
+static const char *gtid_too_old_errmsg=
+  "Could not find GTID state requested by slave in any binlog "
+  "files. Probably the slave state is too old and required binlog files "
+  "have been purged.";
+
 /*
   Helper function for gtid_find_binlog_pos() below.
   Check a binlog file against a slave position. Use a GTID index if present.
@@ -1410,6 +1461,52 @@ end:
 }
 
 
+/*
+  Do some checks on each GTID in the starting GTID state found when searching
+  for the starting GTID position in the binlog.
+*/
+static void
+found_pos_check_gtid(const rpl_gtid *found_gtid, slave_connection_state *state,
+                     slave_connection_state *until_gtid_state)
+{
+  const rpl_gtid *gtid= state->find(found_gtid->domain_id);
+  if (!gtid)
+  {
+    /*
+      Contains_all_slave_gtid() returns false if there is any domain in
+      Gtid_list_event which is not in the requested slave position.
+
+      We may delete a domain from the slave state inside this loop, but
+      we only do this when it is the very last GTID logged for that
+      domain in earlier binlogs, and then we can not encounter it in any
+      further GTIDs in the Gtid_list.
+    */
+    DBUG_ASSERT(0);
+  } else if (gtid->server_id == found_gtid->server_id &&
+             gtid->seq_no == found_gtid->seq_no)
+  {
+    /*
+      The slave requested to start from the very beginning of this
+      domain in this binlog file. So delete the entry from the state,
+      we do not need to skip anything.
+    */
+    state->remove(gtid);
+  }
+
+  if (until_gtid_state &&
+      (gtid= until_gtid_state->find(found_gtid->domain_id)) &&
+      gtid->server_id == found_gtid->server_id &&
+      gtid->seq_no <= found_gtid->seq_no)
+  {
+    /*
+      We've already reached the stop position in UNTIL for this domain,
+      since it is before the start position.
+    */
+    until_gtid_state->remove(gtid);
+  }
+}
+
+
 /*
   Find the name of the binlog file to start reading for a slave that connects
   using GTID state.
@@ -1506,43 +1603,7 @@ gtid_find_binlog_pos(slave_connection_state *state, char *out_name,
           their UNTIL condition.
         */
         for (i= 0; i < count; ++i)
-        {
+          found_pos_check_gtid(&(gtids[i]), state, until_gtid_state);
-          const rpl_gtid *gtid= state->find(gtids[i].domain_id);
-          if (!gtid)
-          {
-            /*
-              Contains_all_slave_gtid() returns false if there is any domain in
-              Gtid_list_event which is not in the requested slave position.
-
-              We may delete a domain from the slave state inside this loop, but
-              we only do this when it is the very last GTID logged for that
-              domain in earlier binlogs, and then we can not encounter it in any
-              further GTIDs in the Gtid_list.
-            */
-            DBUG_ASSERT(0);
-          } else if (gtid->server_id == gtids[i].server_id &&
-                     gtid->seq_no == gtids[i].seq_no)
-          {
-            /*
-              The slave requested to start from the very beginning of this
-              domain in this binlog file. So delete the entry from the state,
-              we do not need to skip anything.
-            */
-            state->remove(gtid);
-          }
-
-          if (until_gtid_state &&
-              (gtid= until_gtid_state->find(gtids[i].domain_id)) &&
-              gtid->server_id == gtids[i].server_id &&
-              gtid->seq_no <= gtids[i].seq_no)
-          {
-            /*
-              We've already reached the stop position in UNTIL for this domain,
-              since it is before the start position.
-            */
-            until_gtid_state->remove(gtid);
-          }
-        }
       }
 
       goto end;
@@ -1551,9 +1612,7 @@ gtid_find_binlog_pos(slave_connection_state *state, char *out_name,
   }
 
   /* We reached the end without finding anything. */
-  errormsg= "Could not find GTID state requested by slave in any binlog "
+  errormsg= gtid_too_old_errmsg;
-    "files. Probably the slave state is too old and required binlog files "
-    "have been purged.";
 
 end:
   if (glev)
@@ -1567,6 +1626,26 @@ end:
 }
 
 
+static const char *
+gtid_find_engine_pos(handler_binlog_reader *binlog_reader,
+                     slave_connection_state *pos,
+                     slave_connection_state *until_gtid_pos,
+                     rpl_binlog_state *until_binlog_state)
+{
+  int res= binlog_reader->init_gtid_pos(pos, until_binlog_state);
+  if (res < 0)
+    return "Error while looking up GTID position in engine binlog";
+  if (res == 0)
+    return gtid_too_old_errmsg;
+  until_binlog_state->iterate(
+    [pos, until_gtid_pos] (const rpl_gtid *gtid) -> bool {
+      found_pos_check_gtid(gtid, pos, until_gtid_pos);
+      return false;
+    });
+  return nullptr;
+}
+
+
 static bool
 gtid_index_lookup_pos(const char *name, uint32 offset, uint32 *out_start_seek,
                       slave_connection_state *out_gtid_state)
@@ -2342,6 +2421,8 @@ static int init_binlog_sender(binlog_send_info *info,
   {
     LEX_CSTRING *engine_name= hton_name(opt_binlog_engine_hton);
     my_error(ER_CANNOT_INIT_ENGINE_BINLOG_READER, MYF(0), engine_name->str);
+    info->errmsg= "Error while initializing engine binlog reader";
+    info->error= ER_CANNOT_INIT_ENGINE_BINLOG_READER;
     return 1;
   }
 
@@ -2369,6 +2450,14 @@ static int init_binlog_sender(binlog_send_info *info,
       info->is_until_before_gtids= get_slave_gtid_until_before_gtids(thd);
     }
   }
+  else if (opt_binlog_engine_hton)
+  {
+    my_error(ER_ENGINE_BINLOG_REQUIRES_GTID, MYF(0));
+    info->errmsg=
+      "Slave must enable GTID mode when master uses --binlog-storage-engine";
+    info->error= ER_ENGINE_BINLOG_REQUIRES_GTID;
+    return 1;
+  }
 
   DBUG_EXECUTE_IF("binlog_force_reconnect_after_22_events",
     {
@@ -2431,14 +2520,31 @@ static int init_binlog_sender(binlog_send_info *info,
       info->error= error;
       return 1;
     }
-    if ((info->errmsg= gtid_find_binlog_pos(&info->gtid_state,
+
-                                            search_file_name,
+    if (opt_binlog_engine_hton)
-                                            info->until_gtid_state,
-                                            &info->until_binlog_state,
-                                            &found_in_index, &start_seek)))
     {
-      info->error= ER_MASTER_FATAL_ERROR_READING_BINLOG;
+      if ((info->errmsg= gtid_find_engine_pos(info->engine_binlog_reader,
-      return 1;
+                                              &info->gtid_state,
+                                              info->until_gtid_state,
+                                              &info->until_binlog_state)))
+      {
+        info->error= ER_MASTER_FATAL_ERROR_READING_BINLOG;
+        return 1;
+      }
+      found_in_index= true;
+      start_seek= 0;  /* Not used when binlog implemented in engine. */
+    }
+    else
+    {
+      if ((info->errmsg= gtid_find_binlog_pos(&info->gtid_state,
+                                              search_file_name,
+                                              info->until_gtid_state,
+                                              &info->until_binlog_state,
+                                              &found_in_index, &start_seek)))
+      {
+        info->error= ER_MASTER_FATAL_ERROR_READING_BINLOG;
+        return 1;
+      }
     }
 
     if (found_in_index)
@@ -2461,7 +2567,8 @@ static int init_binlog_sender(binlog_send_info *info,
   }
   linfo->index_file_offset= 0;
 
-  if (mysql_bin_log.find_log_pos(linfo, name, 1))
+  if (!opt_binlog_engine_hton &&
+      mysql_bin_log.find_log_pos(linfo, name, 1))
   {
     info->errmsg= "Could not find first log file name in binary "
         "log index file";
@@ -2474,7 +2581,8 @@ static int init_binlog_sender(binlog_send_info *info,
   // note: publish that we use file, before we open it
   thd->set_current_linfo(linfo);
 
-  if (check_start_offset(info, linfo->log_file_name, *pos))
+  if (!opt_binlog_engine_hton &&
+      check_start_offset(info, linfo->log_file_name, *pos))
     return 1;
 
   if (*pos > BIN_LOG_HEADER_SIZE)
@@ -2918,11 +3026,13 @@ static int send_events(binlog_send_info *info, IO_CACHE* log, LOG_INFO* linfo,
   ulong ev_offset;
 
   String *packet= info->packet;
+  if (!opt_binlog_engine_hton) {
   linfo->pos= my_b_tell(log);
   info->last_pos= my_b_tell(log);
 
   log->end_of_file= end_pos;
-  while (linfo->pos < end_pos)
+  }
+  while (opt_binlog_engine_hton || linfo->pos < end_pos)
   {
     if (should_stop(info))
       return 0;
@@ -3052,7 +3162,8 @@ static int send_one_binlog_file(binlog_send_info *info,
   mysql_mutex_assert_not_owner(mysql_bin_log.get_log_lock());
 
   /* seek to the requested position, to start the requested dump */
-  if (start_pos != BIN_LOG_HEADER_SIZE)
+  if (!opt_binlog_engine_hton &&
+      start_pos != BIN_LOG_HEADER_SIZE)
   {
     my_b_seek(log, start_pos);
     linfo->pos= start_pos;
@@ -3066,7 +3177,8 @@ static int send_one_binlog_file(binlog_send_info *info,
      * get end pos of current log file, this function
      * will wait if there is nothing available
      */
-    my_off_t end_pos= get_binlog_end_pos(info, log, linfo);
+    my_off_t end_pos= opt_binlog_engine_hton ?
+      UINT32_MAX : get_binlog_end_pos(info, log, linfo);
     if (end_pos <= 1)
     {
       /** end of file or error */
@@ -3141,6 +3253,25 @@ void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
 
   while (!should_stop(info))
   {
+    /* ToDo: do some re-factoring/cleanup so that the code path for binlog-in-engine becomes separate from the legacy code path, sharing common code but avoiding much of the old cruft. */
+    if (opt_binlog_engine_hton) {
+      /* Build a legacy Format_description event for slave. */
+      if (!(info->fdev= new Format_description_log_event
+            (4, 0, BINLOG_CHECKSUM_ALG_OFF)))
+      {
+        info->errmsg= "Out of memory initializing format_description event";
+        info->error= ER_MASTER_FATAL_ERROR_READING_BINLOG;
+        goto err;
+      }
+      if (reset_transmit_packet(info, info->flags, &ev_offset, &info->errmsg) ||
+          fake_format_description_event(info, info->fdev, &info->errmsg, pos))
+      {
+        info->error= ER_MASTER_FATAL_ERROR_READING_BINLOG;
+        goto err;
+      }
+    }
+    else
+    {
     /*
       Tell the client about the log name with a fake Rotate event;
       this is needed even if we also send a Format_description_log_event
@@ -3192,6 +3323,8 @@ void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
       goto err;
     }
 
+    }  /* !opt_binlog_engine_hton */
+
     /*
       We want to corrupt the first event that will be sent to the slave.
       But we do not want the corruption to happen early, eg. when client does

storage/innobase/fsp/fsp0fsp.cc

@@ -4397,6 +4397,8 @@ public:
   ha_innodb_binlog_reader(uint64_t file_no= 0, uint64_t offset= 0);
   ~ha_innodb_binlog_reader();
   virtual int read_binlog_data(uchar *buf, uint32_t len) final;
+  virtual int init_gtid_pos(slave_connection_state *pos,
+                            rpl_binlog_state_base *state) final;
 };
 
 
@@ -5500,9 +5502,46 @@ void fsp_binlog_write_cache(IO_CACHE *cache, size_t main_size, mtr_t *mtr)
                 ((current_binlog_state_interval >> page_size_shift) - 1))) {
         if (page_no == 0) {
           rpl_binlog_state_base full_state;
+          bool err;
           full_state.init();
-          bool err= load_global_binlog_state(&full_state);
+          err= load_global_binlog_state(&full_state);
           ut_a(!err /* ToDo error handling */);
+          if (UNIV_UNLIKELY(file_no == 0 && page_no == 0) &&
+              (full_state.count_nolock() == 1))
+          {
+            /*
+              The gtid state written here includes the GTID for the event group
+              currently being written. This is precise when the event group
+              data begins before this point. If the event group happens to
+              start exactly on a binlog file boundary, it just means we will
+              have to read slightly more binlog data to find the starting point
+              of that GTID.
+
+              But there is an annoying case if this is the very first binlog
+              file created (no migration from legacy binlog). If we start the
+              binlog with some GTID 0-1-1 and write the state "0-1-1" at the
+              start of the first file, then we will be unable to start
+              replicating from the GTID position "0-1-1", corresponding to the
+              *second* event group in the binlog. Because there will be no
+              slightly earlier point to start reading from!
+
+              So we put a slightly awkward special case here to handle that: If
+              at the start of the first file we have a singleton gtid state
+              with seq_no=1, D-S-1, then it must be the very first GTID in the
+              entire binlog, so we write an *empty* gtid state that will always
+              allow to start replicating from the very start of the binlog.
+
+              (If the user would explicitly set the seq_no of the very first
+              GTID in the binlog greater than 1, then starting from that GTID
+              position will still not be possible).
+            */
+            rpl_gtid singleton_gtid;
+            full_state.get_gtid_list_nolock(&singleton_gtid, 1);
+            if (singleton_gtid.seq_no == 1)
+            {
+              full_state.reset_nolock();
+            }
+          }
           err= binlog_gtid_state(&full_state, mtr, block, page_no,
                                  page_offset, space);
           ut_a(!err /* ToDo error handling */);
@@ -5937,6 +5976,270 @@ innodb_get_binlog_reader()
 }
 
 
+class gtid_search {
+public:
+  gtid_search();
+  ~gtid_search();
+  int read_gtid_state_file_no(rpl_binlog_state_base *state, uint64_t file_no,
+                              uint32_t page_no, uint64_t *out_file_end,
+                              uint64_t *out_diff_state_interval);
+  int find_gtid_pos(slave_connection_state *pos,
+                    rpl_binlog_state_base *out_state, uint64_t *out_file_no,
+                    uint64_t *out_offset);
+private:
+  uint64_t cur_open_file_no;
+  uint64_t cur_open_file_length;
+  File cur_open_file;
+};
+
+
+gtid_search::gtid_search()
+  : cur_open_file_no(~(uint64_t)0), cur_open_file_length(0),
+    cur_open_file((File)-1)
+{
+  /* Nothing else. */
+}
+
+
+gtid_search::~gtid_search()
+{
+  if (cur_open_file >= (File)0)
+    my_close(cur_open_file, MYF(0));
+}
+
+
+/*
+  Read a GTID state record from file_no and page_no.
+
+  Returns:
+    1  State record found
+    0  No state record found
+    -1 Error
+*/
+int
+gtid_search::read_gtid_state_file_no(rpl_binlog_state_base *state,
+                                     uint64_t file_no, uint32_t page_no,
+                                     uint64_t *out_file_end,
+                                     uint64_t *out_diff_state_interval)
+{
+  buf_block_t *block;
+
+  *out_file_end= 0;
+  uint64_t active2= active_binlog_file_no.load(std::memory_order_acquire);
+  if (file_no > active2)
+    return 0;
+
+  for (;;)
+  {
+    mtr_t mtr;
+    bool mtr_started= false;
+    uint64_t active= active2;
+    uint64_t end_offset=
+      binlog_cur_end_offset[file_no&1].load(std::memory_order_acquire);
+    if (file_no + 1 >= active &&
+        page_no <= (end_offset >> srv_page_size_shift))
+    {
+      mtr.start();
+      mtr_started= true;
+      uint32_t space_id= SRV_SPACE_ID_BINLOG0 + (file_no & 1);
+      dberr_t err= DB_SUCCESS;
+      block= buf_page_get_gen(page_id_t{space_id, page_no}, 0, RW_S_LATCH,
+                              nullptr, BUF_GET_IF_IN_POOL, &mtr, &err);
+      if (err != DB_SUCCESS) {
+        mtr.commit();
+        return -1;
+      }
+    }
+    else
+      block= nullptr;
+    active2= active_binlog_file_no.load(std::memory_order_acquire);
+    if (UNIV_UNLIKELY(active2 != active))
+    {
+      /* Active moved ahead while we were reading, try again. */
+      if (mtr_started)
+        mtr.commit();
+      continue;
+    }
+    if (file_no + 1 >= active)
+    {
+      *out_file_end= end_offset;
+      if (page_no > (end_offset >> srv_page_size_shift))
+      {
+        ut_ad(!mtr_started);
+        return 0;
+      }
+    }
+
+    if (block)
+    {
+      int res= read_gtid_state_from_page(state, block->page.frame, page_no,
+                                         out_diff_state_interval);
+      ut_ad(mtr_started);
+      if (mtr_started)
+        mtr.commit();
+      return res;
+    }
+    else
+    {
+      if (mtr_started)
+        mtr.commit();
+      if (cur_open_file_no != file_no)
+      {
+        if (cur_open_file >= (File)0)
+        {
+          my_close(cur_open_file, MYF(0));
+          cur_open_file= (File)-1;
+          cur_open_file_length= 0;
+        }
+      }
+      if (cur_open_file < (File)0)
+      {
+        char filename[BINLOG_NAME_LEN];
+        binlog_name_make(filename, file_no);
+        // ToDo: Here, if the file does not exist, it is not an error (no MY_WME), but it could mean we need to return "gtid pos too old". Or at least that we can go no further back and have to return the oldest state we found for the upper layer to deal with in terms of error etc.
+        cur_open_file= my_open(filename, O_RDONLY | O_BINARY, MYF(MY_WME));
+        if (cur_open_file < (File)0)
+          return -1;
+        MY_STAT stat_buf;
+        if (my_fstat(cur_open_file, &stat_buf, MYF(0))) {
+          my_error(ER_CANT_GET_STAT, MYF(0), filename, errno);
+          my_close(cur_open_file, MYF(0));
+          cur_open_file= (File)-1;
+          return -1;
+        }
+        cur_open_file_length= stat_buf.st_size;
+        cur_open_file_no= file_no;
+      }
+      if (!*out_file_end)
+        *out_file_end= cur_open_file_length;
+      return read_gtid_state(state, cur_open_file, page_no,
+                             out_diff_state_interval);
+    }
+  }
+}
+
+
+/*
+  Search for a GTID position in the binlog.
+  Find a binlog file_no and an offset into the file that is guaranteed to
+  be before the target position. It can be a bit earlier, that only means a
+  bit more of the binlog needs to be scanned to find the real position.
+
+  Returns:
+    -1 error
+     0 Position not found (has been purged)
+     1 Position found
+*/
+
+int
+gtid_search::find_gtid_pos(slave_connection_state *pos,
+                           rpl_binlog_state_base *out_state,
+                           uint64_t *out_file_no, uint64_t *out_offset)
+{
+  /*
+    Dirty read, but getting a slightly stale value is no problem, we will just
+    be starting to scan the binlog file at a slightly earlier position than
+    necessary.
+  */
+  uint64_t file_no= active_binlog_file_no.load(std::memory_order_relaxed);
+
+  /* First search backwards for the right file to start from. */
+  uint64_t file_end= 0;
+  uint64_t diff_state_interval= 0;
+  rpl_binlog_state_base base_state, diff_state;
+  base_state.init();
+  for (;;)
+  {
+    int res= read_gtid_state_file_no(&base_state, file_no, 0, &file_end,
+                                     &diff_state_interval);
+    // ToDo: catch the error "file does not exist" specifically and return "position not found" in this case.
+    if (res < 0)
+      return -1;
+    if (res == 0)
+    {
+      ut_ad(0 /* Not expected to find no state, should always be written. */);
+      return -1;
+    }
+    if (base_state.is_before_pos(pos))
+      break;
+    base_state.reset_nolock();
+    if (file_no == 0)
+      return 0;
+    --file_no;
+  }
+
+  /*
+    Then binary search for the last differential state record that is still
+    before the searched position.
+
+    The invariant is that page2 is known to be after the target page, and page0
+    is known to be a valid position to start (but possibly earlier than needed).
+  */
+  uint32_t diff_state_page_interval=
+    (uint32_t)(diff_state_interval >> srv_page_size_shift);
+  ut_ad(diff_state_interval % srv_page_size == 0);
+  if (diff_state_interval % srv_page_size != 0)
+    return -1;  // Corrupt tablespace
+  uint32_t page0= 0;
+  uint32_t page2= (uint32_t)
+    ((file_end + diff_state_interval - 1) >> srv_page_size_shift);
+  /* Round to the next diff_state_interval after file_end. */
+  page2-= page2 % diff_state_page_interval;
+  uint32_t page1= (page0 + page2) / 2;
+  diff_state.init();
+  diff_state.load_nolock(&base_state);
+  while (page1 >= page0 + diff_state_interval)
+  {
+    ut_ad((page1 - page0) % diff_state_interval == 0);
+    diff_state.reset_nolock();
+    diff_state.load_nolock(&base_state);
+    int res= read_gtid_state_file_no(&diff_state, file_no, 0, &file_end,
+                                     &diff_state_interval);
+    if (res < 0)
+      return -1;
+    if (res == 0)
+    {
+      /*
+        If the diff state record was not written here for some reason, just
+        try the one just before. It will be safe, even if not always optimal,
+        and this is an abnormal situation anyway.
+      */
+      page1= page1 - diff_state_page_interval;
+      continue;
+    }
+    if (diff_state.is_before_pos(pos))
+      page0= page1;
+    else
+      page2= page1;
+    page1= (page0 + page2) / 2;
+  }
+  ut_ad(page1 >= page0);
+  out_state->load_nolock(&diff_state);
+  *out_file_no= file_no;
+  *out_offset= (uint64_t)page0 << srv_page_size_shift;
+  return 1;
+}
+
+
+int
+ha_innodb_binlog_reader::init_gtid_pos(slave_connection_state *pos,
+                                       rpl_binlog_state_base *state)
+{
+  gtid_search search_obj;
+  uint64_t file_no;
+  uint64_t offset;
+  int res= search_obj.find_gtid_pos(pos, state, &file_no, &offset);
+  if (res < 0)
+    return -1;
+  if (res > 0)
+  {
+    cur_file_no= file_no;
+    cur_file_offset= offset;
+  }
+  return res;
+}
+
+
 bool
 innobase_binlog_write_direct(IO_CACHE *cache, size_t main_size,
                              const rpl_gtid *gtid)