MDEV-34705: Binlog-in-engine: Implement DELETE_DOMAIN_ID for FLUSH

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

mysql-test/suite/binlog_in_engine/binlog_flush_purge.result

@@ -128,5 +128,34 @@ binlog-000029.ibb	262144
 SET SESSION binlog_format= MIXED;
 DROP TABLE t1;
 SET GLOBAL max_binlog_total_size= @old_max_total;
-SET GLOBAL slave_connections_needed_for_purge= @old_min_slaves;
 SET GLOBAL binlog_expire_logs_seconds= @old_expire;
+*** Test FLUSH BINARY LOGS DELETE_DOMAIN_ID.
+SET SESSION gtid_domain_id= 1;
+SET SESSION gtid_seq_no= 1000;
+CREATE TABLE t1 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1, 0);
+INSERT INTO t1 VALUES (2, 2), (3, 0), (4, 5), (5, 0), (6, 3), (7, 4), (8, 8);
+SET SESSION gtid_domain_id= 2;
+SET SESSION gtid_seq_no= 100;
+ALTER TABLE t1 ADD INDEX b_idx(b);
+SET SESSION gtid_domain_id= 1;
+INSERT INTO t1 VALUES (10, 0), (11, 0), (12, 0);
+SELECT @@GLOBAL.gtid_binlog_state;
+@@GLOBAL.gtid_binlog_state
+0-1-2508,1-1-1003,2-1-100
+FLUSH BINARY LOGS DELETE_DOMAIN_ID=(2);
+ERROR HY000: Could not delete gtid domain. Reason: binlog files may contain gtids from the domain ('2') being deleted. Make sure to first purge those files.
+SELECT @@GLOBAL.gtid_binlog_state;
+@@GLOBAL.gtid_binlog_state
+0-1-2508,1-1-1003,2-1-100
+FLUSH BINARY LOGS;
+PURGE BINARY LOGS TO 'binlog-000030.ibb';
+FLUSH BINARY LOGS DELETE_DOMAIN_ID=(2);
+SELECT @@GLOBAL.gtid_binlog_state;
+@@GLOBAL.gtid_binlog_state
+0-1-2508,1-1-1003
+# restart
+SELECT @@GLOBAL.gtid_binlog_state;
+@@GLOBAL.gtid_binlog_state
+0-1-2508,1-1-1003
+DROP TABLE t1;

mysql-test/suite/binlog_in_engine/binlog_flush_purge.test

@@ -133,14 +133,18 @@ while ($i < $num_insert) {
 }
 COMMIT;
 --enable_query_log
+# We need to wait for 25 to be pre-allocated here, so we know that 23 has been
+# fully written to disk. Otherwise 23 may still be in the buffer pool, and the
+# file date can be older than @now and then the PURGE ... BEFORE @now below
+# fails.
+--let $binlog_name= binlog-000025.ibb
+--let $binlog_size= 262144
+--source include/wait_for_engine_binlog.inc
 PURGE BINARY LOGS BEFORE @now;
 --let $binlog_name= binlog-000022.ibb
 --let $binlog_size= 262144
 --let $wait_notfound= 1
 --source include/wait_for_engine_binlog.inc
---let $binlog_name= binlog-000025.ibb
---let $binlog_size= 262144
---source include/wait_for_engine_binlog.inc
 SHOW BINARY LOGS;
 
 --echo *** Test PURGE BINARY LOGS TO
@@ -160,7 +164,6 @@ while ($i < $num_insert) {
 }
 COMMIT;
 --enable_query_log
---source include/wait_for_engine_binlog.inc
 --let $binlog_name= binlog-000029.ibb
 --let $binlog_size= 262144
 --source include/wait_for_engine_binlog.inc
@@ -185,5 +188,41 @@ SET SESSION binlog_format= MIXED;
 DROP TABLE t1;
 
 SET GLOBAL max_binlog_total_size= @old_max_total;
-SET GLOBAL slave_connections_needed_for_purge= @old_min_slaves;
 SET GLOBAL binlog_expire_logs_seconds= @old_expire;
+
+
+--echo *** Test FLUSH BINARY LOGS DELETE_DOMAIN_ID.
+
+SET SESSION gtid_domain_id= 1;
+SET SESSION gtid_seq_no= 1000;
+CREATE TABLE t1 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1, 0);
+INSERT INTO t1 VALUES (2, 2), (3, 0), (4, 5), (5, 0), (6, 3), (7, 4), (8, 8);
+SET SESSION gtid_domain_id= 2;
+SET SESSION gtid_seq_no= 100;
+ALTER TABLE t1 ADD INDEX b_idx(b);
+SET SESSION gtid_domain_id= 1;
+INSERT INTO t1 VALUES (10, 0), (11, 0), (12, 0);
+SELECT @@GLOBAL.gtid_binlog_state;
+--error ER_BINLOG_CANT_DELETE_GTID_DOMAIN
+FLUSH BINARY LOGS DELETE_DOMAIN_ID=(2);
+SELECT @@GLOBAL.gtid_binlog_state;
+FLUSH BINARY LOGS;
+--let $binlog_name= binlog-000031.ibb
+--let $binlog_size= 262144
+--source include/wait_for_engine_binlog.inc
+PURGE BINARY LOGS TO 'binlog-000030.ibb';
+FLUSH BINARY LOGS DELETE_DOMAIN_ID=(2);
+SELECT @@GLOBAL.gtid_binlog_state;
+
+# Test that deletion of domains in the state got persisted to disk.
+--let $binlog_name= binlog-000032.ibb
+--let $binlog_size= 262144
+--source include/wait_for_engine_binlog.inc
+--source include/restart_mysqld.inc
+SELECT @@GLOBAL.gtid_binlog_state;
+
+DROP TABLE t1;
+
+# No need to restore @@GLOBAL.slave_connections_needed_for_purge, as we
+# restarted the server.

sql/handler.h

@@ -1571,6 +1571,14 @@ struct handlerton
     Used to implement FLUSH BINARY LOGS.
   */
   bool (*binlog_flush)();
+  /*
+    Read the binlog state at the start of the very first (not purged) binlog
+    file, and return it in *out_state. This is used to check validity of
+    FLUSH BINARY LOGS DELETE_DOMAIN_ID=(<list>).
+
+    Returns true on error, false on ok.
+  */
+  bool (*binlog_get_init_state)(rpl_binlog_state_base *out_state);
   /* Engine implementation of RESET MASTER. */
   bool (*reset_binlogs)();
   /*

sql/log.cc

@@ -8209,6 +8209,7 @@ static int do_delete_gtid_domain(DYNAMIC_ARRAY *domain_drop_lex)
   IO_CACHE cache;
   const char* errmsg= NULL;
   char errbuf[MYSQL_ERRMSG_SIZE]= {0};
+  rpl_binlog_state_base init_state;
 
   if (!domain_drop_lex)
     return 0; // still "effective" having empty domain sequence to delete
@@ -8229,8 +8230,16 @@ static int do_delete_gtid_domain(DYNAMIC_ARRAY *domain_drop_lex)
                   errmsg= "injected error";);
   if (errmsg)
     goto end;
+
+  init_state.init();
+  if (init_state.load_nolock(glev->list, glev->count))
+  {
+    my_error(ER_OUT_OF_RESOURCES, MYF(0));
+    rc= -1;
+    goto err;
+  }
   errmsg= rpl_global_gtid_binlog_state.drop_domain(domain_drop_lex,
-                                                   glev, errbuf);
+                                                   &init_state, errbuf);
 
 end:
   if (errmsg)
@@ -8245,6 +8254,7 @@ end:
       rc= 1;
     }
   }
+err:
   delete glev;
 
   return rc;
@@ -8305,6 +8315,67 @@ int MYSQL_BIN_LOG::rotate_and_purge(bool force_rotate,
 }
 
 
+/**
+  Remove a list of domains from the in-memory global binlog state, after
+  checking that deletion is safe. "Safe" in this context means that there
+  are no GTID present with the domain in any of the existing binlog files
+  (ie. the binlog files where that domain was used have all been purged).
+  This is checked by comparing the binlog state at the beginning of the
+  earliest current binlog file with the current binlog state.
+
+  @param  domain_drop_lex  gtid domain id sequence from lex.
+                           Passed as a pointer to dynamic array must be not empty
+                           unless pointer value NULL.
+  @retval zero             on success
+  @retval > 0              ineffective call none from the *non* empty
+                           gtid domain sequence is deleted
+  @retval < 0              on error
+*/
+static int
+binlog_engine_delete_gtid_domain(DYNAMIC_ARRAY *domain_drop_lex)
+{
+  int rc= 0;
+  const char* errmsg= NULL;
+  char errbuf[MYSQL_ERRMSG_SIZE]= {0};
+  rpl_binlog_state_base init_state;
+
+  if (!domain_drop_lex)
+    return 0; // still "effective" having empty domain sequence to delete
+
+  DBUG_ASSERT(domain_drop_lex->elements > 0);
+  DBUG_ASSERT(opt_binlog_engine_hton);
+  mysql_mutex_assert_owner(mysql_bin_log.get_log_lock());
+
+  if (!opt_binlog_engine_hton->binlog_get_init_state)
+  {
+    my_error(ER_ENGINE_BINLOG_NO_DELETE_DOMAIN, MYF(0));
+    return -1;
+  }
+
+  init_state.init();
+  if ((*opt_binlog_engine_hton->binlog_get_init_state)(&init_state))
+  {
+    my_error(ER_BINLOG_CANNOT_READ_STATE, MYF(0));
+    return -1;
+  }
+  errmsg= rpl_global_gtid_binlog_state.drop_domain(domain_drop_lex,
+                                                   &init_state, errbuf);
+  if (errmsg)
+  {
+    if (strlen(errmsg) > 0)
+    {
+      my_error(ER_BINLOG_CANT_DELETE_GTID_DOMAIN, MYF(0), errmsg);
+      rc= -1;
+    }
+    else
+    {
+      rc= 1;
+    }
+  }
+  return rc;
+}
+
+
 /* Implementation of FLUSH BINARY LOGS for binlog implemented in engine. */
 int
 MYSQL_BIN_LOG::flush_binlogs_engine(DYNAMIC_ARRAY *domain_drop_lex)
@@ -8314,7 +8385,9 @@ MYSQL_BIN_LOG::flush_binlogs_engine(DYNAMIC_ARRAY *domain_drop_lex)
 
   mysql_mutex_lock(&LOCK_log);
 
-  // ToDo: Implement DELETE_DOMAIN_ID option. Ask the engine to load the oldest GTID state in the binlog, check that it matches the current GTID state in the to-be-deleted domains, then update the GTID state so the engine can write the state with domains deleted after it does the FLUSH. See also do_delete_gtid_domain().
+  if ((error= binlog_engine_delete_gtid_domain(domain_drop_lex)) &&
+      error < 0)
+    error= 1;
 
   if ((*opt_binlog_engine_hton->binlog_flush)())
     error= 1;
@@ -8325,11 +8398,6 @@ MYSQL_BIN_LOG::flush_binlogs_engine(DYNAMIC_ARRAY *domain_drop_lex)
   mysql_mutex_unlock(&LOCK_after_binlog_sync);
   mysql_mutex_unlock(&LOCK_commit_ordered);
 
-  if (!error)
-  {
-    /* ToDo: Do purge, once implemented. */
-  }
-
   DBUG_RETURN(error);
 }
 

sql/rpl_gtid.cc

@@ -2203,11 +2203,11 @@ rpl_binlog_state::append_state(String *str)
 /**
   Remove domains supplied by the first argument from binlog state.
   Removal is done for any domain whose last gtids (from all its servers) match
-  ones in Gtid list event of the 2nd argument.
+  ones in the binlog state at the start of the current binlog, passed in as the
+  2nd argument.
 
   @param  ids               gtid domain id sequence, may contain dups
-  @param  glev              pointer to Gtid list event describing
-                            the match condition
+  @param  init_state        Binlog state at the start of the current binlog
   @param  errbuf [out]      pointer to possible error message array
 
   @retval NULL              as success when at least one domain is removed
@@ -2217,12 +2217,12 @@ rpl_binlog_state::append_state(String *str)
 */
 const char*
 rpl_binlog_state::drop_domain(DYNAMIC_ARRAY *ids,
-                              Gtid_list_log_event *glev,
+                              rpl_binlog_state_base *init_state,
                               char* errbuf)
 {
   DYNAMIC_ARRAY domain_unique; // sequece (unsorted) of unique element*:s
   rpl_binlog_state::element* domain_unique_buffer[16];
-  ulong k, l;
+  ulong k;
   const char* errmsg= NULL;
 
   DBUG_ENTER("rpl_binlog_state::drop_domain");
@@ -2249,45 +2249,46 @@ rpl_binlog_state::drop_domain(DYNAMIC_ARRAY *ids,
     B and C may require the user's attention so any (incl the A's suspected)
     inconsistency is diagnosed and *warned*.
   */
-  for (l= 0, errbuf[0]= 0; l < glev->count; l++, errbuf[0]= 0)
-  {
-    rpl_gtid* rb_state_gtid= find_nolock(glev->list[l].domain_id,
-                                         glev->list[l].server_id);
+
+  errbuf[0]= 0;
+  init_state->iterate([this, errbuf](const rpl_gtid *gtid) {
+    rpl_gtid* rb_state_gtid= find_nolock(gtid->domain_id, gtid->server_id);
     if (!rb_state_gtid)
       sprintf(errbuf,
               "missing gtids from the '%u-%u' domain-server pair which is "
               "referred to in the gtid list describing an earlier state. Ignore "
               "if the domain ('%u') was already explicitly deleted",
-              glev->list[l].domain_id, glev->list[l].server_id,
-              glev->list[l].domain_id);
-    else if (rb_state_gtid->seq_no < glev->list[l].seq_no)
+              gtid->domain_id, gtid->server_id,
+              gtid->domain_id);
+    else if (rb_state_gtid->seq_no < gtid->seq_no)
       sprintf(errbuf,
               "having a gtid '%u-%u-%llu' which is less than "
               "the '%u-%u-%llu' of the gtid list describing an earlier state. "
               "The state may have been affected by manually injecting "
               "a lower sequence number gtid or via replication",
               rb_state_gtid->domain_id, rb_state_gtid->server_id,
-              rb_state_gtid->seq_no, glev->list[l].domain_id,
-              glev->list[l].server_id, glev->list[l].seq_no);
+              rb_state_gtid->seq_no, gtid->domain_id,
+              gtid->server_id, gtid->seq_no);
     if (strlen(errbuf)) // use strlen() as cheap flag
       push_warning_printf(current_thd, Sql_condition::WARN_LEVEL_WARN,
                           ER_BINLOG_CANT_DELETE_GTID_DOMAIN,
                           "The current gtid binlog state is incompatible with "
                           "a former one %s.", errbuf);
-  }
+    errbuf[0]= 0;
+    return false;    // No error
+  });
 
   /*
     For each domain_id from ids
       If the domain is already absent from the binlog state
         Warn && continue
-      If any GTID with that domain in binlog state is missing from glev.list
+      If any GTID with that domain in binlog state is missing from init_state
         Error out binlog state can't change
   */
   for (ulong i= 0; i < ids->elements; i++)
   {
     rpl_binlog_state::element *elem= NULL;
     uint32 *ptr_domain_id;
-    bool all_found;
 
     ptr_domain_id= (uint32*) dynamic_array_ptr(ids, i);
     elem= (rpl_binlog_state::element *)
@@ -2302,25 +2303,21 @@ rpl_binlog_state::drop_domain(DYNAMIC_ARRAY *ids,
       continue;
     }
 
-    all_found= true;
-    for (k= 0; k < elem->hash.records && all_found; k++)
+    for (k= 0; k < elem->hash.records; k++)
     {
       rpl_gtid *d_gtid= (rpl_gtid *)my_hash_element(&elem->hash, k);
-      bool match_found= false;
-      for (ulong l= 0; l < glev->count && !match_found; l++)
-        match_found= match_found || (*d_gtid == glev->list[l]);
-      if (!match_found)
-        all_found= false;
+      rpl_gtid *state_gtid=
+        init_state->find_nolock(d_gtid->domain_id, d_gtid->server_id);
+      if (!state_gtid || state_gtid->seq_no != d_gtid->seq_no)
+      {
+        sprintf(errbuf, "binlog files may contain gtids from the domain ('%u') "
+                "being deleted. Make sure to first purge those files",
+                *ptr_domain_id);
+        errmsg= errbuf;
+        goto end;
+      }
     }
 
-    if (!all_found)
-    {
-      sprintf(errbuf, "binlog files may contain gtids from the domain ('%u') "
-              "being deleted. Make sure to first purge those files",
-              *ptr_domain_id);
-      errmsg= errbuf;
-      goto end;
-    }
     // compose a sequence of unique pointers to domain object
     for (k= 0; k < domain_unique.elements; k++)
     {

sql/rpl_gtid.h

@@ -329,7 +329,8 @@ struct rpl_binlog_state : public rpl_binlog_state_base
   bool append_state(String *str);
   rpl_gtid *find(uint32 domain_id, uint32 server_id);
   rpl_gtid *find_most_recent(uint32 domain_id);
-  const char* drop_domain(DYNAMIC_ARRAY *ids, Gtid_list_log_event *glev, char*);
+  const char* drop_domain(DYNAMIC_ARRAY *ids, rpl_binlog_state_base *init_state,
+                          char*);
 };
 
 

sql/share/errmsg-utf8.txt

@@ -12298,3 +12298,7 @@ ER_ENGINE_BINLOG_REQUIRES_GTID
         eng "GTID starting position is required on master with --binlog-storage-engine enabled"
 ER_ENGINE_BINLOG_NO_RESET_FILE_NUMBER
 	eng "RESET MASTER TO is not available when --binlog-storage-engine is enabled"
+ER_ENGINE_BINLOG_NO_DELETE_DOMAIN
+	eng "The binlog engine does not support DELETE_DOMAIN_ID"
+ER_BINLOG_CANNOT_READ_STATE
+	eng "Error reading GTID state from the binlog"

storage/innobase/handler/ha_innodb.cc

@@ -4135,6 +4135,7 @@ static int innodb_init(void* p)
         innobase_hton->get_binlog_reader= innodb_get_binlog_reader;
         innobase_hton->get_binlog_file_list= innodb_get_binlog_file_list;
         innobase_hton->binlog_flush= innodb_binlog_flush;
+        innobase_hton->binlog_get_init_state= innodb_binlog_get_init_state;
         innobase_hton->reset_binlogs= innodb_reset_binlogs;
         innobase_hton->binlog_purge= innodb_binlog_purge;
 

storage/innobase/handler/innodb_binlog.cc

@@ -373,6 +373,7 @@ struct chunk_data_cache : public chunk_data_base {
 
 
 class gtid_search {
+public:
   /*
     Note that this enum is set up to be compatible with int results -1/0/1 for
     error/not found/fount from read_gtid_state_from_page().
@@ -383,7 +384,6 @@ class gtid_search {
     READ_NOT_FOUND= 0,
     READ_FOUND= 1
   };
-public:
   gtid_search();
   ~gtid_search();
   enum Read_Result read_gtid_state_file_no(rpl_binlog_state_base *state,
@@ -2209,6 +2209,25 @@ innodb_find_binlogs(uint64_t *out_first, uint64_t *out_last)
 }
 
 
+bool
+innodb_binlog_get_init_state(rpl_binlog_state_base *out_state)
+{
+  gtid_search search_obj;
+  uint64_t dummy_file_end, dummy_diff_state_interval;
+  bool err= false;
+
+  mysql_mutex_lock(&purge_binlog_mutex);
+  uint64_t file_no= earliest_binlog_file_no;
+  enum gtid_search::Read_Result res=
+    search_obj.read_gtid_state_file_no(out_state, file_no, 0, &dummy_file_end,
+                                       &dummy_diff_state_interval);
+  mysql_mutex_unlock(&purge_binlog_mutex);
+  if (res != gtid_search::READ_FOUND)
+    err= true;
+  return err;
+
+}
+
 bool
 innodb_reset_binlogs()
 {

storage/innobase/include/innodb_binlog.h

@@ -108,6 +108,7 @@ extern bool innobase_binlog_write_direct
   (IO_CACHE *cache, handler_binlog_event_group_info *binlog_info,
    const rpl_gtid *gtid);
 extern bool innodb_find_binlogs(uint64_t *out_first, uint64_t *out_last);
+extern bool innodb_binlog_get_init_state(rpl_binlog_state_base *out_state);
 extern bool innodb_reset_binlogs();
 extern int innodb_binlog_purge(handler_binlog_purge_info *purge_info);