MDEV-4989: Support for GTID in mysqlbinlog10.7-MDEV-4989

New Feature:
============
This commit extends the mariadb-binlog capabilities to allow events
to be filtered by GTID ranges. More specifically, the following
capabilities are addressed:
   1) GTIDs can be used to filter results on local binlog files
   2) GTIDs can be used to filter results from remote servers
   3) For a given GTID range, its start-position is exclusive and its
      stop-position is inclusive
   4) After the events have been written, the session server id and
      domain id are reset to their former values
   5) Output filtered by GTID ranges can be piped to the MariaDB client

To facilitate these features, the --start-position and --stop-position
arguments have been extended to additionally accept values formatted
as a list of GTID positions, e.g. `--start-position=0-1-0,1-2-55`

Reviewed By:
============
<TODO>

1 files changed, 39 insertions, 5 deletions

diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 7bcff12a735..00fc65cf82f 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -163,6 +163,8 @@ struct binlog_send_info {
   bool should_stop;
   size_t dirlen;
 
+  Gtid_event_filter *gtid_event_filter;
+
   binlog_send_info(THD *thd_arg, String *packet_arg, ushort flags_arg,
                    char *lfn)
     : thd(thd_arg), net(&thd_arg->net), packet(packet_arg),
@@ -185,6 +187,8 @@ struct binlog_send_info {
     error_text[0] = 0;
     bzero(&error_gtid, sizeof(error_gtid));
     until_binlog_state.init();
+
+    gtid_event_filter= NULL;
   }
 };
 
@@ -1751,6 +1755,7 @@ send_event_to_slave(binlog_send_info *info, Log_event_type event_type,
     }
   }
 
+
   /* Skip GTID event groups until we reach slave position within a domain_id. */
   if (event_type == GTID_EVENT && info->using_gtid_state)
   {
@@ -1758,7 +1763,7 @@ send_event_to_slave(binlog_send_info *info, Log_event_type event_type,
     slave_connection_state::entry *gtid_entry;
     rpl_gtid *gtid;
 
-    if (gtid_state->count() > 0 || until_gtid_state)
+    if (gtid_state->count() > 0 || until_gtid_state || info->gtid_event_filter)
     {
       rpl_gtid event_gtid;
 
@@ -1899,6 +1904,17 @@ send_event_to_slave(binlog_send_info *info, Log_event_type event_type,
           }
         }
       }
+
+      /*
+        Should this result be excluded from the output?
+      */
+      if (info->gtid_event_filter &&
+          info->gtid_event_filter->exclude(&event_gtid))
+      {
+        info->gtid_skip_group=
+            (flags2 & Gtid_log_event::FL_STANDALONE ? GTID_SKIP_STANDALONE
+                                                    : GTID_SKIP_TRANSACTION);
+      }
     }
   }
 
@@ -2110,7 +2126,9 @@ err:
 static int init_binlog_sender(binlog_send_info *info,
                               LOG_INFO *linfo,
                               const char *log_ident,
-                              my_off_t *pos)
+                              my_off_t *pos,
+                              rpl_gtid *start_gtids,
+                              size_t n_start_gtids)
 {
   THD *thd= info->thd;
   int error;
@@ -2130,7 +2148,8 @@ static int init_binlog_sender(binlog_send_info *info,
 
   info->current_checksum_alg= get_binlog_checksum_value_at_connect(thd);
   info->mariadb_slave_capability= get_mariadb_slave_capability(thd);
-  info->using_gtid_state= get_slave_connect_state(thd, &connect_gtid_state);
+  info->using_gtid_state= get_slave_connect_state(thd, &connect_gtid_state) ||
+                          start_gtids != NULL;
   DBUG_EXECUTE_IF("simulate_non_gtid_aware_master",
                   info->using_gtid_state= false;);
 
@@ -2247,6 +2266,17 @@ static int init_binlog_sender(binlog_send_info *info,
     info->clear_initial_log_pos= true;
   }
 
+  if (start_gtids != NULL)
+  {
+    Domain_gtid_event_filter *filter= new Domain_gtid_event_filter();
+    my_off_t i;
+    for(i = 0; i < n_start_gtids; i++)
+    {
+      filter->add_start_gtid(&start_gtids[i]);
+    }
+    info->gtid_event_filter= filter;
+  }
+
   return 0;
 }
 
@@ -2840,7 +2870,8 @@ static int send_one_binlog_file(binlog_send_info *info,
 }
 
 void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
-                       ushort flags)
+                       ushort flags, rpl_gtid *start_gtids,
+                       uint32 n_start_gtids)
 {
   LOG_INFO linfo;
 
@@ -2860,7 +2891,8 @@ void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
 
   bzero((char*) &log,sizeof(log));
 
-  if (init_binlog_sender(info, &linfo, log_ident, &pos))
+  if (init_binlog_sender(info, &linfo, log_ident, &pos, start_gtids,
+                         n_start_gtids))
     goto err;
 
   has_transmit_started= true;
@@ -3022,6 +3054,8 @@ err:
   thd->reset_current_linfo();
   thd->variables.max_allowed_packet= old_max_allowed_packet;
   delete info->fdev;
+  delete info->gtid_event_filter;
+  my_free(start_gtids);
 
   if (likely(info->error == 0))
   {