MDEV-20119: Implement the --do-domain-ids, --ignore-domain-ids, and --ignore-server-ids options for mysqlbinlog

New Feature:
============
Extend mariadb-binlog command-line tool to allow for filtering
events using GTID domain and server ids. The functionality mimics
that of a replica server’s DO_DOMAIN_IDS, IGNORE_DOMAIN_IDS, and
IGNORE_SERVER_IDS from CHANGE MASTER TO. For completeness, this
patch additionally adds the option --do-server-ids as an alias for
--server-id, which now accepts a list of server ids instead of a
single one.

Example usage:
  mariadb-binlog --do-domain-ids=2,3,4 --do-server-ids=1,3
  master-bin.000001

Functional Notes:
 1. --do-domain-ids cannot be combined with --ignore-domain-ids
 2. --do-server-ids cannot be combined with --ignore-server-ids
 3. A domain id filter can be combined with a server id filter
 4. When any new filter options are combined with the
--gtid-strict-mode option, events from excluded domains/servers are
not validated.
 5. Domain/server id filters can be combined with GTID ranges (i.e.
specifications of --start-position and --stop-position). However,
because the --stop-position option implicitly undertakes filtering
to only output events within its range of domains, when combined
with --do-domain-ids or --ignore-domain-ids, output will consist of
the intersection between the filters. Specifically, with
--do-domain-ids and --stop-position, only events with domain ids
present in both argument lists will be output. Conversely, with
--ignore-domain-ids and --stop-position, only events with domain ids
present in the --stop-position and absent from the
--ignore-domain-ids options will be output.

Reviewed By
============
Andrei Elkin <andrei.elkin@mariadb.com>

1 files changed, 228 insertions, 41 deletions

diff --git a/sql/rpl_gtid.cc b/sql/rpl_gtid.cc
index ae40fd46f4d..7876ff456c6 100644
--- a/sql/rpl_gtid.cc
+++ b/sql/rpl_gtid.cc
@@ -16,6 +16,7 @@
 
 
 /* Definitions for MariaDB global transaction ID (GTID). */
+#include <type_traits>
 
 #ifndef MYSQL_CLIENT
 #include "mariadb.h"
@@ -3480,32 +3481,42 @@ my_bool Window_gtid_event_filter::has_finished()
   return m_has_stop ? m_has_passed : FALSE;
 }
 
-void free_gtid_filter_element(void *p)
+void free_u32_gtid_filter_element(void *p)
 {
-  gtid_filter_element *gfe = (gtid_filter_element *) p;
+  gtid_filter_element<uint32> *gfe= (gtid_filter_element<uint32> *) p;
   if (gfe->filter)
     delete gfe->filter;
   my_free(gfe);
 }
 
-Id_delegating_gtid_event_filter::Id_delegating_gtid_event_filter()
-    : m_num_stateful_filters(0), m_num_completed_filters(0)
+template <typename T>
+Id_delegating_gtid_event_filter<T>::Id_delegating_gtid_event_filter()
+    : m_num_stateful_filters(0), m_num_completed_filters(0),
+      m_id_restriction_mode(id_restriction_mode::MODE_NOT_SET)
 {
+  void (*free_func)(void *);
+  if (std::is_same<T,uint32>::value)
+    free_func= free_u32_gtid_filter_element;
+  else
+    DBUG_ASSERT(0);
+
   my_hash_init(PSI_INSTRUMENT_ME, &m_filters_by_id_hash, &my_charset_bin, 32,
-               offsetof(gtid_filter_element, identifier),
-               sizeof(gtid_filter_identifier), NULL, free_gtid_filter_element,
+               offsetof(gtid_filter_element<T>, identifier),
+               sizeof(T), NULL, free_func,
                HASH_UNIQUE);
 
   m_default_filter= new Accept_all_gtid_filter();
 }
 
-Id_delegating_gtid_event_filter::~Id_delegating_gtid_event_filter()
+template <typename T>
+Id_delegating_gtid_event_filter<T>::~Id_delegating_gtid_event_filter()
 {
   my_hash_free(&m_filters_by_id_hash);
   delete m_default_filter;
 }
 
-void Id_delegating_gtid_event_filter::set_default_filter(
+template <typename T>
+void Id_delegating_gtid_event_filter<T>::set_default_filter(
     Gtid_event_filter *filter)
 {
   if (m_default_filter)
@@ -3514,17 +3525,19 @@ void Id_delegating_gtid_event_filter::set_default_filter(
   m_default_filter= filter;
 }
 
-gtid_filter_element *
-Id_delegating_gtid_event_filter::find_or_create_filter_element_for_id(
-    gtid_filter_identifier filter_id)
+template <typename T>
+gtid_filter_element<T> *
+Id_delegating_gtid_event_filter<T>::find_or_create_filter_element_for_id(
+    T filter_id)
 {
-  gtid_filter_element *fe= (gtid_filter_element *) my_hash_search(
-      &m_filters_by_id_hash, (const uchar *) &filter_id, 0);
+  gtid_filter_element<T> *fe=
+      (gtid_filter_element<T> *) my_hash_search(
+          &m_filters_by_id_hash, (const uchar *) &filter_id, 0);
 
   if (!fe)
   {
-    gtid_filter_element *new_fe= (gtid_filter_element *) my_malloc(
-        PSI_NOT_INSTRUMENTED, sizeof(gtid_filter_element), MYF(MY_WME));
+    gtid_filter_element<T> *new_fe= (gtid_filter_element<T> *) my_malloc(
+        PSI_NOT_INSTRUMENTED, sizeof(gtid_filter_element<T>), MYF(MY_WME));
     new_fe->filter= NULL;
     new_fe->identifier= filter_id;
     if (my_hash_insert(&m_filters_by_id_hash, (uchar*) new_fe))
@@ -3539,7 +3552,8 @@ Id_delegating_gtid_event_filter::find_or_create_filter_element_for_id(
   return fe;
 }
 
-my_bool Id_delegating_gtid_event_filter::has_finished()
+template <typename T>
+my_bool Id_delegating_gtid_event_filter<T>::has_finished()
 {
   /*
     If all user-defined filters have deactivated, we are effectively
@@ -3549,11 +3563,13 @@ my_bool Id_delegating_gtid_event_filter::has_finished()
          m_num_completed_filters == m_num_stateful_filters;
 }
 
-my_bool Id_delegating_gtid_event_filter::exclude(rpl_gtid *gtid)
+template <typename T>
+my_bool Id_delegating_gtid_event_filter<T>::exclude(rpl_gtid *gtid)
 {
-  gtid_filter_identifier filter_id= get_id_from_gtid(gtid);
-  gtid_filter_element *filter_element= (gtid_filter_element *) my_hash_search(
-      &m_filters_by_id_hash, (const uchar *) &filter_id, 0);
+  T filter_id= get_id_from_gtid(gtid);
+  gtid_filter_element<T> *filter_element=
+      (gtid_filter_element<T> *) my_hash_search(&m_filters_by_id_hash,
+                                                (const uchar *) &filter_id, 0);
   Gtid_event_filter *filter=
       (filter_element ? filter_element->filter : m_default_filter);
   my_bool ret= TRUE;
@@ -3573,11 +3589,124 @@ my_bool Id_delegating_gtid_event_filter::exclude(rpl_gtid *gtid)
   return ret;
 }
 
+
+template <typename F> Gtid_event_filter* create_event_filter()
+{
+  return new F();
+}
+
+template <typename T>
+int Id_delegating_gtid_event_filter<T>::set_id_restrictions(
+    T *id_list, size_t n_ids, id_restriction_mode mode)
+{
+  static const char *WHITELIST_NAME= "do", *BLACKLIST_NAME= "ignore";
+
+  size_t id_ctr;
+  int err;
+  Gtid_event_filter::gtid_event_filter_type filter_type;
+  const char *filter_name, *opposite_filter_name;
+  Gtid_event_filter *(*construct_filter)(void);
+  Gtid_event_filter *(*construct_default_filter)(void);
+
+  DBUG_ASSERT(mode > id_restriction_mode::MODE_NOT_SET);
+
+  /*
+    Set up variables which help this filter either be in whitelist or blacklist
+    mode
+  */
+  if (mode == Gtid_event_filter::id_restriction_mode::WHITELIST_MODE)
+  {
+    filter_type= Gtid_event_filter::ACCEPT_ALL_GTID_FILTER_TYPE;
+    filter_name= WHITELIST_NAME;
+    opposite_filter_name= BLACKLIST_NAME;
+    construct_filter=
+        create_event_filter<Accept_all_gtid_filter>;
+    construct_default_filter=
+        create_event_filter<Reject_all_gtid_filter>;
+  }
+  else if (mode == Gtid_event_filter::id_restriction_mode::BLACKLIST_MODE)
+  {
+    filter_type= Gtid_event_filter::REJECT_ALL_GTID_FILTER_TYPE;
+    filter_name= BLACKLIST_NAME;
+    opposite_filter_name= WHITELIST_NAME;
+    construct_filter=
+        create_event_filter<Reject_all_gtid_filter>;
+    construct_default_filter=
+        create_event_filter<Accept_all_gtid_filter>;
+  }
+  else
+  {
+    DBUG_ASSERT(0);
+  }
+
+  if (m_id_restriction_mode !=
+      Gtid_event_filter::id_restriction_mode::MODE_NOT_SET)
+  {
+    if (mode != m_id_restriction_mode)
+    {
+      /*
+        If a rule specifying the opposite version of this has already been set,
+        error.
+      */
+      sql_print_error("Cannot create %s filtering rule for %s id because "
+                      "%s rule already exists",
+                      filter_name, get_id_type_name(),
+                      opposite_filter_name);
+      err= 1;
+      goto err;
+    }
+
+    /* This filter is specified more than once, only use the latest values */
+    my_hash_reset(&m_filters_by_id_hash);
+  }
+
+  for (id_ctr= 0; id_ctr < n_ids; id_ctr++)
+  {
+    T filter_id= id_list[id_ctr];
+    gtid_filter_element<T> *map_element=
+        find_or_create_filter_element_for_id(filter_id);
+
+    if(map_element == NULL)
+    {
+      /*
+        If map_element is NULL, find_or_create_filter_element_for_id failed and
+        has already written the error message
+      */
+      err= 1;
+      goto err;
+    }
+    else if (map_element->filter == NULL)
+    {
+      map_element->filter= construct_filter();
+      m_num_stateful_filters++;
+    }
+    else
+    {
+      DBUG_ASSERT(map_element->filter->get_filter_type() ==
+                  filter_type);
+    }
+  }
+
+  /*
+    With a whitelist, we by only want to accept the ids which are specified.
+    Everything else should be denied.
+
+    With a blacklist, we by default want to accept everything that is not
+    specified in the list
+  */
+  set_default_filter(construct_default_filter());
+  m_id_restriction_mode= mode;
+  err= 0;
+
+err:
+  return err;
+}
+
 Window_gtid_event_filter *
 Domain_gtid_event_filter::find_or_create_window_filter_for_id(
-    uint32 domain_id)
+    decltype(rpl_gtid::domain_id) domain_id)
 {
-  gtid_filter_element *filter_element=
+  gtid_filter_element<decltype(rpl_gtid::domain_id)> *filter_element=
       find_or_create_filter_element_for_id(domain_id);
   Window_gtid_event_filter *wgef= NULL;
 
@@ -3606,9 +3735,11 @@ Domain_gtid_event_filter::find_or_create_window_filter_for_id(
   return wgef;
 }
 
-static my_bool check_filter_entry_validity(void *entry, void *are_filters_invalid_arg)
+static my_bool check_filter_entry_validity(void *entry,
+                                           void *are_filters_invalid_arg)
 {
-  gtid_filter_element *fe= (gtid_filter_element*) entry;
+  gtid_filter_element<decltype(rpl_gtid::domain_id)> *fe=
+      (gtid_filter_element<decltype(rpl_gtid::domain_id)> *) entry;
 
   if (fe)
   {
@@ -3629,7 +3760,8 @@ static my_bool check_filter_entry_validity(void *entry, void *are_filters_invali
 int Domain_gtid_event_filter::validate_window_filters()
 {
   int are_filters_invalid= 0;
-  my_hash_iterate(&m_filters_by_id_hash, check_filter_entry_validity, &are_filters_invalid);
+  my_hash_iterate(&m_filters_by_id_hash, check_filter_entry_validity,
+                  &are_filters_invalid);
   return are_filters_invalid;
 }
 
@@ -3645,8 +3777,9 @@ int Domain_gtid_event_filter::add_start_gtid(rpl_gtid *gtid)
   }
   else if (!(err= filter_to_update->set_start_gtid(gtid)))
   {
-    gtid_filter_element *fe= (gtid_filter_element *) my_hash_search(
-        &m_filters_by_id_hash, (const uchar *) &(gtid->domain_id), 0);
+    gtid_filter_element<decltype(rpl_gtid::domain_id)> *fe=
+        (gtid_filter_element<decltype(rpl_gtid::domain_id)> *) my_hash_search(
+            &m_filters_by_id_hash, (const uchar *) &(gtid->domain_id), 0);
     insert_dynamic(&m_start_filters, (const void *) &fe);
   }
 
@@ -3665,8 +3798,9 @@ int Domain_gtid_event_filter::add_stop_gtid(rpl_gtid *gtid)
   }
   else if (!(err= filter_to_update->set_stop_gtid(gtid)))
   {
-    gtid_filter_element *fe= (gtid_filter_element *) my_hash_search(
-        &m_filters_by_id_hash, (const uchar *) &(gtid->domain_id), 0);
+    gtid_filter_element<decltype(rpl_gtid::domain_id)> *fe=
+        (gtid_filter_element<decltype(rpl_gtid::domain_id)> *) my_hash_search(
+            &m_filters_by_id_hash, (const uchar *) &(gtid->domain_id), 0);
     insert_dynamic(&m_stop_filters, (const void *) &fe);
 
     /*
@@ -3699,8 +3833,9 @@ rpl_gtid *Domain_gtid_event_filter::get_start_gtids()
 
   for (i = 0; i < n_start_gtids; i++)
   {
-    gtid_filter_element *fe=
-        *(gtid_filter_element **) dynamic_array_ptr(&m_start_filters, i);
+    gtid_filter_element<decltype(rpl_gtid::domain_id)> *fe=
+        *(gtid_filter_element<decltype(rpl_gtid::domain_id)> **)
+            dynamic_array_ptr(&m_start_filters, i);
     DBUG_ASSERT(fe->filter &&
                 fe->filter->get_filter_type() == WINDOW_GTID_FILTER_TYPE);
     Window_gtid_event_filter *wgef=
@@ -3724,8 +3859,9 @@ rpl_gtid *Domain_gtid_event_filter::get_stop_gtids()
 
   for (i = 0; i < n_stop_gtids; i++)
   {
-    gtid_filter_element *fe=
-        *(gtid_filter_element **) dynamic_array_ptr(&m_stop_filters, i);
+    gtid_filter_element<decltype(rpl_gtid::domain_id)> *fe=
+        *(gtid_filter_element<decltype(rpl_gtid::domain_id)> **)
+            dynamic_array_ptr(&m_stop_filters, i);
     DBUG_ASSERT(fe->filter &&
                 fe->filter->get_filter_type() == WINDOW_GTID_FILTER_TYPE);
     Window_gtid_event_filter *wgef=
@@ -3743,8 +3879,9 @@ void Domain_gtid_event_filter::clear_start_gtids()
   uint32 i;
   for (i = 0; i < get_num_start_gtids(); i++)
   {
-    gtid_filter_element *fe=
-        *(gtid_filter_element **) dynamic_array_ptr(&m_start_filters, i);
+    gtid_filter_element<decltype(rpl_gtid::domain_id)> *fe=
+        *(gtid_filter_element<decltype(rpl_gtid::domain_id)> **)
+            dynamic_array_ptr(&m_start_filters, i);
     DBUG_ASSERT(fe->filter &&
                 fe->filter->get_filter_type() == WINDOW_GTID_FILTER_TYPE);
     Window_gtid_event_filter *wgef=
@@ -3775,8 +3912,9 @@ void Domain_gtid_event_filter::clear_stop_gtids()
 
   for (i = 0; i < get_num_stop_gtids(); i++)
   {
-    gtid_filter_element *fe=
-        *(gtid_filter_element **) dynamic_array_ptr(&m_stop_filters, i);
+    gtid_filter_element<decltype(rpl_gtid::domain_id)> *fe=
+        *(gtid_filter_element<decltype(rpl_gtid::domain_id)> **)
+            dynamic_array_ptr(&m_stop_filters, i);
     DBUG_ASSERT(fe->filter &&
                 fe->filter->get_filter_type() == WINDOW_GTID_FILTER_TYPE);
     Window_gtid_event_filter *wgef=
@@ -3822,10 +3960,10 @@ my_bool Domain_gtid_event_filter::exclude(rpl_gtid *gtid)
   */
   if (get_num_stop_gtids())
   {
-    gtid_filter_identifier filter_id= get_id_from_gtid(gtid);
-    gtid_filter_element *filter_element=
-        (gtid_filter_element *) my_hash_search(&m_filters_by_id_hash,
-                                               (const uchar *) &filter_id, 0);
+    decltype(rpl_gtid::domain_id) filter_id= get_id_from_gtid(gtid);
+    gtid_filter_element<decltype(rpl_gtid::domain_id)> *filter_element=
+        (gtid_filter_element<decltype(rpl_gtid::domain_id)> *) my_hash_search(
+            &m_filters_by_id_hash, (const uchar *) &filter_id, 0);
     if (filter_element)
     {
       Gtid_event_filter *filter= filter_element->filter;
@@ -3840,3 +3978,52 @@ my_bool Domain_gtid_event_filter::exclude(rpl_gtid *gtid)
   return include_domain ? Id_delegating_gtid_event_filter::exclude(gtid)
                         : TRUE;
 }
+
+Intersecting_gtid_event_filter::Intersecting_gtid_event_filter(
+    Gtid_event_filter *filter1, Gtid_event_filter *filter2)
+{
+  my_init_dynamic_array(PSI_INSTRUMENT_ME, &m_filters,
+                        sizeof(Gtid_event_filter *), 3, 3, MYF(0));
+  insert_dynamic(&m_filters, (void *) &filter1);
+  insert_dynamic(&m_filters, (void *) &filter2);
+}
+
+Intersecting_gtid_event_filter::~Intersecting_gtid_event_filter()
+{
+  Gtid_event_filter *tmp_filter= NULL;
+  ulong i;
+  for (i= 0; i < m_filters.elements; i++)
+  {
+    tmp_filter= *(Gtid_event_filter **) dynamic_array_ptr(&m_filters, i);
+    delete tmp_filter;
+  }
+  delete_dynamic(&m_filters);
+}
+
+my_bool Intersecting_gtid_event_filter::exclude(rpl_gtid *gtid)
+{
+  Gtid_event_filter *tmp_filter= NULL;
+  ulong i;
+  for (i= 0; i < m_filters.elements; i++)
+  {
+    tmp_filter= *(Gtid_event_filter **) dynamic_array_ptr(&m_filters, i);
+    DBUG_ASSERT(tmp_filter);
+    if (tmp_filter->exclude(gtid))
+      return TRUE;
+  }
+  return FALSE;
+}
+
+my_bool Intersecting_gtid_event_filter::has_finished()
+{
+  Gtid_event_filter *tmp_filter= NULL;
+  ulong i;
+  for (i= 0; i < m_filters.elements; i++)
+  {
+    tmp_filter= *(Gtid_event_filter **) dynamic_array_ptr(&m_filters, i);
+    DBUG_ASSERT(tmp_filter);
+    if (tmp_filter->has_finished())
+      return TRUE;
+  }
+  return FALSE;
+}