MDEV-32570 (client): switch print_event_info table_map_event to use m_table_map instead

To be consistent with existing behavior,
print_event_info::table_map_event, which was added in MDEV-32570 is
switched to use the existing print_event_info::m_table_map. This
simplifies logic to track its life-cycle, and is consistent with
using print_event_info::m_table_map_ignored.

client/mysqlbinlog.cc

@@ -968,6 +968,14 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
         print_event_info->deactivate_current_event_group();
     }
 
+    /*
+      Start a new GTID event with a fresh table map state. This is because
+      table maps are cached for each transaction, e.g. for
+      Partial_rows_log_event printing.
+    */
+    print_event_info->m_table_map.clear_tables();
+    print_event_info->m_table_map_ignored.clear_tables();
+
     /*
       Where we always ensure the initial binlog state is valid, we only
       continually monitor the GTID stream for validity if we are in GTID
@@ -1185,18 +1193,7 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
       break;
     case TABLE_MAP_EVENT:
     {
-      /*
-        Always keep the Table_map_log_event around in case a group of
-        Partial_rows_log_events is seen, where we will write the content of
-        the Table_map_log_event for the last fragment so it can be re-applied.
-
-        TODO: Refactor the existing logic now that the Tmle is always
-              persistent
-      */
       Table_map_log_event *map= ((Table_map_log_event *)ev);
-      if (print_event_info->table_map_event != map)
-        delete print_event_info->table_map_event;
-      print_event_info->table_map_event= map;
       destroy_evt= FALSE;
 
       if (shall_skip_database(map->get_db_name()) ||
@@ -1205,6 +1202,14 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
         print_event_info->m_table_map_ignored.set_table(map->get_table_id(), map);
         goto end;
       }
+
+      /*
+        Always keep the Table_map_log_event around in case a group of
+        Partial_rows_log_events is seen, where we will write the content of
+        the Table_map_log_event for the last fragment so it can be re-applied.
+      */
+      print_event_info->m_table_map.set_table(map->get_table_id(), map);
+
 #ifdef WHEN_FLASHBACK_REVIEW_READY
       /* Create review table for Flashback */
       if (opt_flashback_review)
@@ -2465,13 +2470,11 @@ static Exit_status dump_log_entries(const char* logname)
   print_event_info.short_form= short_form;
   print_event_info.print_row_count= print_row_count;
   print_event_info.file= result_file;
-  print_event_info.table_map_event= NULL;
   fflush(result_file);
   rc= (remote_opt ? dump_remote_log_entries(&print_event_info, logname) :
        dump_local_log_entries(&print_event_info, logname));
 
-  if (print_event_info.table_map_event)
-    delete print_event_info.table_map_event;
+  print_event_info.m_table_map.clear_tables();
 
   if (rc == ERROR_STOP)
     return rc;

sql/log_event.h

@@ -903,6 +903,11 @@ typedef struct st_print_event_info
   enum_base64_output_mode base64_output_mode;
   my_off_t hexdump_from;
 
+  /*
+    The Table_map_log_event(s) for the current event group.  We always need it
+    around in case we are printing a group of Partial_rows_log_events, where we
+    will write a Table_map_log_event for the last fragment.
+  */
   table_mapping m_table_map;
   table_mapping m_table_map_ignored;
   bool flags2_inited;
@@ -932,13 +937,6 @@ typedef struct st_print_event_info
   bool skip_replication;
   bool print_table_metadata;
 
-  /*
-    The Table_map_log_event for the current event group.  We always need it
-    around in case we are printing a group of Partial_rows_log_events, where we
-    will write a Table_map_log_event for the last fragment.
-  */
-  Table_map_log_event *table_map_event;
-
   /*
      These two caches are used by the row-based replication events to
      collect the header information and the main body of the events

sql/log_event_client.cc

@@ -1770,17 +1770,14 @@ bool Log_event::print_base64(IO_CACHE* file,
     {
     case TABLE_MAP_EVENT:
     {
-      Table_map_log_event *map; 
-      map= new Table_map_log_event(ptr, size, 
-                                   glob_description_event);
 #ifdef WHEN_FLASHBACK_REVIEW_READY
+      Table_map_log_event *map= static_cast<Table_map_log_event*>(this);
       if (need_flashback_review)
       {
         map->set_review_dbname(m_review_dbname.ptr());
         map->set_review_tablename(m_review_tablename.ptr());
       }
 #endif
-      print_event_info->m_table_map.set_table(map->get_table_id(), map);
       break;
     }
     case WRITE_ROWS_EVENT:
@@ -3619,15 +3616,35 @@ bool Partial_rows_log_event::print(FILE *file,
 
   if (!print_event_info->short_form || print_event_info->print_row_count)
   {
-    DBUG_ASSERT(print_event_info->table_map_event);
+    DBUG_ASSERT(print_event_info->m_table_map.count() ||
+                print_event_info->m_table_map_ignored.count());
     /*
-      For the last fragment, re-write the Table_map_log_event into the start of
-      the BINLOG base64 statement. See the comment in the header file for
+      For the last fragment, re-write the Table_map_log_event(s) into the start
+      of the BINLOG base64 statement. See the comment in the header file for
       Partial_rows_log_event for more details why.
     */
     if (seq_no == total_fragments)
-      if (print_event_info->table_map_event->print_body(print_event_info))
-        goto err;
+    {
+      ulong n_tables= print_event_info->m_table_map.count();
+      {
+        ulonglong table_ids[n_tables];
+        print_event_info->m_table_map.get_table_ids(&table_ids[0], n_tables);
+        for (ulong i = 0; i < n_tables; i++)
+        {
+          DBUG_PRINT("info", ("Table ID: %llu", table_ids[i]));
+          Table_map_log_event *tbl_map_to_print;
+
+          if (print_event_info->m_table_map_ignored.get_table(table_ids[i]))
+            continue;
+
+          tbl_map_to_print=
+              print_event_info->m_table_map.get_table(table_ids[i]);
+          if (tbl_map_to_print &&
+              tbl_map_to_print->print_body(print_event_info))
+            goto err;
+        }
+      }
+    }
 
     if (print_base64(body, print_event_info, do_print_encoded))
       goto err;

sql/rpl_tblmap.cc

@@ -169,6 +169,9 @@ void table_mapping::clear_tables()
   for (uint i= 0; i < m_table_ids.records; i++)
   {
     entry *e= (entry *)my_hash_element(&m_table_ids, i);
+#ifdef MYSQL_CLIENT
+    free_table_map_log_event(e->table);
+#endif
     e->next= m_free;
     m_free= e;
   }
@@ -176,4 +179,18 @@ void table_mapping::clear_tables()
   DBUG_VOID_RETURN;
 }
 
+void table_mapping::get_table_ids(ulonglong *table_ids, size_t n_table_ids)
+{
+  DBUG_ENTER("table_mapping::get_table_ids()");
+  DBUG_ASSERT(n_table_ids == m_table_ids.records);
+
+  for (uint i = 0; i < m_table_ids.records; i++)
+  {
+    entry *e = (entry *)my_hash_element(&m_table_ids, i);
+    table_ids[i] = e->table_id;
+  }
+
+  DBUG_VOID_RETURN;
+}
+
 #endif

sql/rpl_tblmap.h

@@ -76,6 +76,7 @@ public:
   int       remove_table(ulonglong table_id);
   void      clear_tables();
   ulong     count() const { return m_table_ids.records; }
+  void      get_table_ids(ulonglong *table_ids, size_t n_table_ids);
 
 private:
   /*