WL#15339 step 1: Remove transaction buffering when plugin is SUSPENDED

Foundation Commit
=================
Description:
-----------
- To recover transactions from Binlog, Rapid plugin needs to track
  GTIDs of the committed transactions so that tracked GTIDs can be
  used as exclude set to use Binlog Iterator API (WL#13901).
- Tracked Gtid_set has its own `Sid_map` (and not `global_sid_map`).
- In after commit hook, the GTID of the committing transaction is
  computed from the argument of type Trans_gtid_info. Trans_gtid_info
  has sidno which is relative to global_sid_map.
- Since plugin uses its own Gtid_set to track transactions, `sidno`
  is insufficient to compute correct GTID. The type needs to have
  `sid` (rpl_sid type) which is global server identifier and must be
  used to compute GTID and keep in private `Sid_map`.

Changes:
-------
-> Added member `sid` (type rpl_sid) in Trans_gtid_info.
-> Changes in `class Last_used_gtid_tracker_ctx`:
  - Added member `m_last_used_sid` (type rpl_sid)
  - Modifed `set_last_used_gtid()` to set the `sid` of the transaction
    along with GTID.
  - Added `get_last_used_sid()` to get the `sid` of the last
   transaction.
-> In caller of after commit hook, populate `sid` from
  `last_used_gtid_tracker`.

Change-Id: I4f669c5a84c24bce764c7462c245dceb93eaa912

Author: Abhinav Agarwal

sql/replication.h

@@ -119,6 +119,7 @@ typedef struct Trans_context_info {
   This represents the GTID context of the transaction.
  */
 typedef struct Trans_gtid_info {
+  rpl_sid sid;        // transaction sid
   ulong type;         // enum values in enum_gtid_type
   int sidno;          // transaction sidno
   long long int gno;  // transaction gno

sql/rpl_context.cc

@@ -195,15 +195,21 @@ Last_used_gtid_tracker_ctx::Last_used_gtid_tracker_ctx() {
 
 Last_used_gtid_tracker_ctx::~Last_used_gtid_tracker_ctx() = default;
 
-void Last_used_gtid_tracker_ctx::set_last_used_gtid(const Gtid &gtid) {
+void Last_used_gtid_tracker_ctx::set_last_used_gtid(const Gtid &gtid,
+                                                    const rpl_sid &sid) {
   (*m_last_used_gtid).set(gtid.sidno, gtid.gno);
+  m_last_used_sid.copy_from(sid);
 }
 
 void Last_used_gtid_tracker_ctx::get_last_used_gtid(Gtid &gtid) {
   gtid.sidno = (*m_last_used_gtid).sidno;
   gtid.gno = (*m_last_used_gtid).gno;
 }
 
+void Last_used_gtid_tracker_ctx::get_last_used_sid(rpl_sid &sid) {
+  m_last_used_sid.copy_to(sid.bytes);
+}
+
 Transaction_compression_ctx::Transaction_compression_ctx(PSI_memory_key key)
     : m_managed_buffer_memory_resource(psi_memory_resource(key)),
       m_managed_buffer_sequence(Grow_calculator_t(),

sql/rpl_context.h

@@ -255,8 +255,9 @@ class Last_used_gtid_tracker_ctx {
    Set the last used GTID the session.
 
    @param[in]  gtid  the used gtid.
+   @param[in]  sid   the used sid.
   */
-  void set_last_used_gtid(const Gtid &gtid);
+  void set_last_used_gtid(const Gtid &gtid, const rpl_sid &sid);
 
   /**
    Get the last used GTID the session.
@@ -265,8 +266,16 @@ class Last_used_gtid_tracker_ctx {
   */
   void get_last_used_gtid(Gtid &gtid);
 
+  /**
+   Get the last used SID of the session.
+
+   @param[out]  sid the used sid.
+  */
+  void get_last_used_sid(rpl_sid &sid);
+
  private:
   std::unique_ptr<Gtid> m_last_used_gtid;
+  rpl_sid m_last_used_sid;
 };
 
 class Transaction_compression_ctx {

sql/rpl_gtid_state.cc

@@ -96,7 +96,8 @@ enum_return_status Gtid_state::acquire_ownership(THD *thd, const Gtid &gtid) {
     thd->owned_gtid = gtid;
     thd->owned_gtid.dbug_print(nullptr, "set owned_gtid in acquire_ownership");
     thd->owned_sid = sid_map->sidno_to_sid(gtid.sidno);
-    thd->rpl_thd_ctx.last_used_gtid_tracker_ctx().set_last_used_gtid(gtid);
+    thd->rpl_thd_ctx.last_used_gtid_tracker_ctx().set_last_used_gtid(
+        gtid, thd->owned_sid);
   }
   RETURN_OK;
 err:

sql/rpl_handler.cc

@@ -816,6 +816,9 @@ int Trans_delegate::after_commit(THD *thd, bool all) {
   param.gtid_info.sidno = gtid.sidno;
   param.gtid_info.gno = gtid.gno;
 
+  thd->rpl_thd_ctx.last_used_gtid_tracker_ctx().get_last_used_sid(
+      param.gtid_info.sid);
+
   bool is_real_trans =
       (all || !thd->get_transaction()->is_active(Transaction_ctx::SESSION));
   if (is_real_trans) param.flags |= TRANS_IS_REAL_TRANS;