MDEV-16329 [3/5] use binlog_cache_data directly in most places

* Eliminate most usages of THD::use_trans_table. Only 3 left, and they are
  at quite high levels, and really essential.
* Eliminate is_transactional argument when possible. Lots of places are
  left though, because of some WSREP error handling in
  MYSQL_BIN_LOG::set_write_error.
* Remove junk binlog functions from THD
* binlog_prepare_pending_rows_event is moved to log.cc inside MYSQL_BIN_LOG
  and is not anymore template. Instead it accepls event factory with a type
  code, and a callback to a constructing function in it.

sql/handler.cc

@@ -7269,7 +7269,9 @@ int handler::binlog_log_row(TABLE *table,
   if (thd->variables.option_bits & OPTION_GTID_BEGIN)
     is_trans= 1;
 
-  bool error= (*log_func)(thd, table, &mysql_bin_log, cache_mngr,
+  auto *cache= binlog_get_cache_data(cache_mngr, use_trans_cache(thd, is_trans));
+
+  bool error= (*log_func)(thd, table, &mysql_bin_log, cache,
                           is_trans, before_record, after_record);
   DBUG_RETURN(error ? HA_ERR_RBR_LOGGING_FAILED : 0);
 }

sql/handler.h

@@ -656,8 +656,8 @@ given at all. */
 typedef ulonglong alter_table_operations;
 
 class MYSQL_BIN_LOG;
-class binlog_cache_mngr;
-typedef bool Log_func(THD*, TABLE*, MYSQL_BIN_LOG *, binlog_cache_mngr *, bool,
+class binlog_cache_data;
+typedef bool Log_func(THD*, TABLE*, MYSQL_BIN_LOG *, binlog_cache_data *, bool,
                       const uchar*, const uchar*);
 
 /*

sql/log.h

@@ -350,6 +350,31 @@ class MYSQL_LOG
                                   enum cache_type io_cache_type_arg);
 };
 
+/**
+  @struct Rows_event_factory
+
+  Holds an event type code and a callback function to create it.
+  Should be created by Rows_event_factory::get.
+*/
+struct Rows_event_factory
+{
+  int type_code;
+
+  Rows_log_event *(*create)(THD*, TABLE*, ulong, bool is_transactional);
+
+  template<class RowsEventT>
+  static Rows_event_factory get()
+  {
+    return { RowsEventT::TYPE_CODE,
+             [](THD* thd, TABLE* table, ulong flags, bool is_transactional)
+                     -> Rows_log_event*
+             {
+               return new RowsEventT(thd, table, flags, is_transactional);
+             }
+    };
+  }
+};
+
 /* Tell the io thread if we can delay the master info sync. */
 #define SEMI_SYNC_SLAVE_DELAY_SYNC 1
 /* Tell the io thread if the current event needs a ack. */
@@ -419,6 +444,7 @@ class MYSQL_QUERY_LOG: public MYSQL_LOG
 #define BINLOG_COOKIE_IS_DUMMY(c) \
   ( ((ulong)(c)>>1) == BINLOG_COOKIE_DUMMY_ID )
 
+
 class binlog_cache_mngr;
 class binlog_cache_data;
 struct rpl_gtid;
@@ -723,11 +749,18 @@ class MYSQL_BIN_LOG: public TC_LOG, private MYSQL_LOG
               Format_description_log_event *fdle, bool do_xa);
   int do_binlog_recovery(const char *opt_name, bool do_xa_recovery);
 #if !defined(MYSQL_CLIENT)
+  Rows_log_event*
+  prepare_pending_rows_event(THD *thd, TABLE* table,
+                             binlog_cache_data *cache_data,
+                             uint32 serv_id, size_t needed,
+                             bool is_transactional,
+                             Rows_event_factory event_factory);
 
   int flush_and_set_pending_rows_event(THD *thd, Rows_log_event* event,
-                                       binlog_cache_mngr *cache_mngr,
+                                       binlog_cache_data *cache_data,
                                        bool is_transactional);
-  int remove_pending_rows_event(THD *thd, bool is_transactional);
+
+  static int remove_pending_rows_event(THD *thd, binlog_cache_data *cache_data);
 
 #endif /* !defined(MYSQL_CLIENT) */
   void reset_bytes_written()
@@ -824,7 +857,7 @@ class MYSQL_BIN_LOG: public TC_LOG, private MYSQL_LOG
   void write_binlog_checkpoint_event_already_locked(const char *name, uint len);
   int  write_cache(THD *thd, IO_CACHE *cache);
   void set_write_error(THD *thd, bool is_transactional);
-  bool check_write_error(THD *thd);
+  static bool check_write_error(THD *thd);
 
   void start_union_events(THD *thd, query_id_t query_id_param);
   void stop_union_events(THD *thd);
@@ -1179,10 +1212,11 @@ bool write_annotated_row(THD *thd);
 int binlog_flush_pending_rows_event(THD *thd, bool stmt_end,
                                     bool is_transactional,
                                     MYSQL_BIN_LOG *bin_log,
-                                    binlog_cache_mngr *cache_mngr,
-                                    bool use_trans_cache);
+                                    binlog_cache_data *cache_data);
 Rows_log_event* binlog_get_pending_rows_event(binlog_cache_mngr *cache_mngr,
                                               bool use_trans_cache);
+binlog_cache_data* binlog_get_cache_data(binlog_cache_mngr *cache_mngr,
+                                         bool use_trans_cache);
 
 extern MYSQL_PLUGIN_IMPORT MYSQL_BIN_LOG mysql_bin_log;
 extern handlerton *binlog_hton;

sql/log_event.h

@@ -4927,14 +4927,14 @@ class Write_rows_log_event : public Rows_log_event
 #if defined(MYSQL_SERVER) 
   static bool binlog_row_logging_function(THD *thd, TABLE *table,
                                           MYSQL_BIN_LOG *bin_log,
-                                          binlog_cache_mngr *cache_mngr,
+                                          binlog_cache_data *cache_data,
                                           bool is_transactional,
                                           const uchar *before_record
                                           __attribute__((unused)),
                                           const uchar *after_record)
   {
     DBUG_ASSERT(!table->versioned(VERS_TRX_ID));
-    return thd->binlog_write_row(table, bin_log, cache_mngr, is_transactional,
+    return thd->binlog_write_row(table, bin_log, cache_data, is_transactional,
                                  after_record);
   }
 #endif
@@ -5013,13 +5013,13 @@ class Update_rows_log_event : public Rows_log_event
 #ifdef MYSQL_SERVER
   static bool binlog_row_logging_function(THD *thd, TABLE *table,
                                           MYSQL_BIN_LOG *bin_log,
-                                          binlog_cache_mngr *cache_mngr,
+                                          binlog_cache_data *cache_data,
                                           bool is_transactional,
                                           const uchar *before_record,
                                           const uchar *after_record)
   {
     DBUG_ASSERT(!table->versioned(VERS_TRX_ID));
-    return thd->binlog_update_row(table, bin_log, cache_mngr, is_transactional,
+    return thd->binlog_update_row(table, bin_log, cache_data, is_transactional,
                                   before_record, after_record);
   }
 #endif
@@ -5104,14 +5104,14 @@ class Delete_rows_log_event : public Rows_log_event
 #ifdef MYSQL_SERVER
   static bool binlog_row_logging_function(THD *thd, TABLE *table,
                                           MYSQL_BIN_LOG *bin_log,
-                                          binlog_cache_mngr *cache_mngr,
+                                          binlog_cache_data *cache_data,
                                           bool is_transactional,
                                           const uchar *before_record,
                                           const uchar *after_record
                                           __attribute__((unused)))
   {
     DBUG_ASSERT(!table->versioned(VERS_TRX_ID));
-    return thd->binlog_delete_row(table, bin_log, cache_mngr, is_transactional,
+    return thd->binlog_delete_row(table, bin_log, cache_data, is_transactional,
                                   before_record);
   }
 #endif

sql/sql_class.h

@@ -2947,13 +2947,13 @@ class THD: public THD_count, /* this must be first */
   void binlog_start_trans_and_stmt();
   void binlog_set_stmt_begin();
   int binlog_write_row(TABLE* table, MYSQL_BIN_LOG *bin_log,
-                       binlog_cache_mngr *cache_mngr, bool is_transactional,
+                       binlog_cache_data *cache_data, bool is_transactional,
                        const uchar *buf);
   int binlog_delete_row(TABLE* table,  MYSQL_BIN_LOG *bin_log,
-                        binlog_cache_mngr *cache_mngr, bool is_transactional,
+                        binlog_cache_data *cache_data, bool is_transactional,
                         const uchar *buf);
   int binlog_update_row(TABLE* table, MYSQL_BIN_LOG *bin_log,
-                        binlog_cache_mngr *cache_mngr, bool is_transactional,
+                        binlog_cache_data *cache_data, bool is_transactional,
                         const uchar *old_data, const uchar *new_data);
   bool prepare_handlers_for_update(uint flag);
   bool binlog_write_annotated_row(Log_event_writer *writer);
@@ -2968,14 +2968,12 @@ class THD: public THD_count, /* this must be first */
     Member functions to handle pending event for row-level logging.
   */
   binlog_cache_mngr *binlog_get_cache_mngr() const;
-  void binlog_set_pending_rows_event(Rows_log_event* ev, bool use_trans_cache);
   inline int binlog_flush_pending_rows_event(bool stmt_end)
   {
     return (binlog_flush_pending_rows_event(stmt_end, FALSE) || 
             binlog_flush_pending_rows_event(stmt_end, TRUE));
   }
   int binlog_flush_pending_rows_event(bool stmt_end, bool is_transactional);
-  int binlog_remove_pending_rows_event(bool clear_maps, bool is_transactional);
 
   bool binlog_need_stmt_format(bool is_transactional) const
   {