Optimize Seconds_Behind_Source Under MTS Parallel Replication

mysql-test/suite/rpl/r/rpl_seconds_behind_master_commit_timestamp.result

@@ -0,0 +1,51 @@
+include/rpl/init_source_replica.inc
+Warnings:
+Note	####	Sending passwords in plain text without SSL/TLS is extremely insecure.
+Note	####	Storing MySQL user name or password information in the connection metadata repository is not secure and is therefore not recommended. Please consider using the USER and PASSWORD connection options for START REPLICA; see the 'START REPLICA Syntax' in the MySQL Manual for more information.
+[connection master]
+CALL mtr.add_suppression("Unsafe statement written to the binary log using statement format");
+CALL mtr.add_suppression("Unsafe statement written to the binary log using statement format");
+SET @save_replica_parallel_workers = @@GLOBAL.replica_parallel_workers;
+SET GLOBAL replica_parallel_workers = 4;
+CREATE TABLE t (a INT PRIMARY KEY, b INT);
+include/rpl/sync_to_replica.inc
+#
+# Test case 1: Long transaction blocked on replica, SBM must not
+#   include the time the transaction spent executing on source.
+#   immediate_commit_timestamp records commit time, not start time.
+#
+LOCK TABLE t WRITE;
+INSERT INTO t(a, b) SELECT 1, SLEEP(10);
+SELECT a, b FROM t WHERE a = 1;
+a	b
+1	0
+include/assert.inc [Seconds_Behind_Source must not exceed 9 seconds during blocked long transaction]
+UNLOCK TABLES;
+#
+# Test case 2: MTS checkpoint timing. sbm_block_checkpoint prevents
+#   checkpoint so a completed group stays in GAQ.
+#   sbm_force_checkpoint forces checkpoint on the next QUERY_EVENT.
+#   This verifies the checkpoint path uses immediate_commit_timestamp
+#   and SBM does not spike.
+#
+SET GLOBAL debug= "+d,sbm_block_checkpoint";
+INSERT INTO t(a, b) SELECT 2, 0;
+LOCK TABLE t WRITE;
+SET GLOBAL debug= "+d,sbm_force_checkpoint";
+BEGIN;
+INSERT INTO t(a, b) SELECT 3, 0;
+SELECT SLEEP(10);
+SLEEP(10)
+0
+COMMIT;
+include/assert.inc [Seconds_Behind_Source must not exceed 9 seconds during blocked long transaction under MTS]
+UNLOCK TABLES;
+#
+# Cleanup
+#
+include/rpl/stop_replica.inc
+SET GLOBAL replica_parallel_workers = @save_replica_parallel_workers;
+include/rpl/start_replica.inc
+DROP TABLE t;
+include/rpl/sync_to_replica.inc
+include/rpl/deinit.inc

mysql-test/suite/rpl/t/rpl_seconds_behind_master_commit_timestamp-master.opt

@@ -0,0 +1 @@
+--binlog-rows-query-log-events=ON

mysql-test/suite/rpl/t/rpl_seconds_behind_master_commit_timestamp.test

@@ -0,0 +1,108 @@
+--source include/have_debug.inc
+--source include/rpl/init_source_replica.inc
+
+--connection master
+CALL mtr.add_suppression("Unsafe statement written to the binary log using statement format");
+
+--connection slave
+CALL mtr.add_suppression("Unsafe statement written to the binary log using statement format");
+
+--connection slave
+SET @save_replica_parallel_workers = @@GLOBAL.replica_parallel_workers;
+SET GLOBAL replica_parallel_workers = 4;
+
+--connection master
+CREATE TABLE t (a INT PRIMARY KEY, b INT);
+--source include/rpl/sync_to_replica.inc
+
+--echo #
+--echo # Test case 1: Long transaction blocked on replica, SBM must not
+--echo #   include the time the transaction spent executing on source.
+--echo #   immediate_commit_timestamp records commit time, not start time.
+--echo #
+
+--connection slave
+LOCK TABLE t WRITE;
+
+--connection master
+--disable_warnings
+INSERT INTO t(a, b) SELECT 1, SLEEP(10);
+--enable_warnings
+SELECT a, b FROM t WHERE a = 1;
+
+--connection slave
+--let $wait_timeout= 30
+--let $show_statement= SHOW PROCESSLIST
+--let $field= Info
+--let $condition= = 'INSERT INTO t(a, b) SELECT 1, SLEEP(10)'
+--source include/wait_show_condition.inc
+
+--let $sbm= query_get_value("SHOW REPLICA STATUS", Seconds_Behind_Source, 1)
+--let $assert_text= Seconds_Behind_Source must not exceed 9 seconds during blocked long transaction
+--let $assert_cond= $sbm <= 9
+--source include/assert.inc
+
+UNLOCK TABLES;
+--let $wait_timeout= 30
+--let $wait_condition= SELECT COUNT(*) = 1 FROM t WHERE a = 1
+--source include/wait_condition.inc
+
+--echo #
+--echo # Test case 2: MTS checkpoint timing. sbm_block_checkpoint prevents
+--echo #   checkpoint so a completed group stays in GAQ.
+--echo #   sbm_force_checkpoint forces checkpoint on the next QUERY_EVENT.
+--echo #   This verifies the checkpoint path uses immediate_commit_timestamp
+--echo #   and SBM does not spike.
+--echo #
+
+--connection slave
+SET GLOBAL debug= "+d,sbm_block_checkpoint";
+
+--connection master
+INSERT INTO t(a, b) SELECT 2, 0;
+
+--connection slave
+--let $wait_timeout= 30
+--let $wait_condition= SELECT COUNT(*) = 1 FROM t WHERE a = 2
+--source include/wait_condition.inc
+
+LOCK TABLE t WRITE;
+SET GLOBAL debug= "+d,sbm_force_checkpoint";
+
+--connection master
+BEGIN;
+INSERT INTO t(a, b) SELECT 3, 0;
+SELECT SLEEP(10);
+COMMIT;
+
+--connection slave
+--let $wait_timeout= 30
+--let $show_statement= SHOW PROCESSLIST
+--let $field= Info
+--let $condition= = 'INSERT INTO t(a, b) SELECT 3, 0'
+--source include/wait_show_condition.inc
+
+--let $sbm= query_get_value("SHOW REPLICA STATUS", Seconds_Behind_Source, 1)
+--let $assert_text= Seconds_Behind_Source must not exceed 9 seconds during blocked long transaction under MTS
+--let $assert_cond= $sbm <= 9
+--source include/assert.inc
+
+UNLOCK TABLES;
+--let $wait_timeout= 30
+--let $wait_condition= SELECT COUNT(*) = 1 FROM t WHERE a = 3
+--source include/wait_condition.inc
+
+--echo #
+--echo # Cleanup
+--echo #
+
+--connection slave
+--source include/rpl/stop_replica.inc
+SET GLOBAL replica_parallel_workers = @save_replica_parallel_workers;
+--source include/rpl/start_replica.inc
+
+--connection master
+DROP TABLE t;
+--source include/rpl/sync_to_replica.inc
+
+--source include/rpl/deinit.inc

sql/binlog.cc

@@ -1376,6 +1376,21 @@ bool MYSQL_BIN_LOG::write_transaction(THD *thd, binlog_cache_data *cache_data,
     log).
   */
   ulonglong immediate_commit_timestamp = my_micro_time();
+#ifndef NDEBUG
+  if (DBUG_EVALUATE_IF("inc_event_time_by_1_hour", 1, 0) &&
+      DBUG_EVALUATE_IF("dec_event_time_by_1_hour", 1, 0)) {
+    /**
+      This assertion guarantees that these debug flags are not
+      used at the same time (they would cancel each other).
+    */
+    assert(0);
+  } else {
+    DBUG_EXECUTE_IF("inc_event_time_by_1_hour",
+                    immediate_commit_timestamp += 3600000000;);
+    DBUG_EXECUTE_IF("dec_event_time_by_1_hour",
+                    immediate_commit_timestamp -= 3600000000;);
+  }
+#endif
 
   /*
     When the original_commit_timestamp session variable is set to a value

sql/log_event.cc

@@ -2641,6 +2641,19 @@ Slave_worker *Log_event::get_slave_worker(Relay_log_info *rli) {
 
           Gtid_log_event *gtid_log_ev = static_cast<Gtid_log_event *>(this);
           rli->started_processing(gtid_log_ev);
+
+          // Set group ts from immediate_commit_timestamp at GTID event time,
+          // eliminating the ts=0 window between starts_group and ends_group.
+          // This prevents SBM from spuriously dropping to 0 when checkpoint
+          // sees an in-progress group whose ts has not been set yet.
+          if (gtid_log_ev->has_commit_timestamps) {
+            Slave_job_group *grp =
+                gaq->get_job_group(gaq->assigned_group_index);
+            if (grp != nullptr) {
+              grp->ts = static_cast<time_t>(
+                  gtid_log_ev->immediate_commit_timestamp / 1000000);
+            }
+          }
         }
 
         if (schedule_next_event(this, rli)) {
@@ -2813,8 +2826,13 @@ Slave_worker *Log_event::get_slave_worker(Relay_log_info *rli) {
       ret_worker->checkpoint_notified = true;
     }
     ptr_group->checkpoint_seqno = rli->rli_checkpoint_seqno;
-    ptr_group->ts = common_header->when.tv_sec +
-                    (time_t)exec_time;  // Seconds_behind_source related
+    // Only set ts from ends_group event if it was not already set from
+    // immediate_commit_timestamp at GTID event time. This preserves the
+    // commit timestamp precision and avoids overwriting with when+exec_time.
+    if (ptr_group->ts == 0) {
+      ptr_group->ts = common_header->when.tv_sec +
+                      (time_t)exec_time;  // Seconds_behind_source related
+    }
     rli->rli_checkpoint_seqno++;
     /*
       Coordinator should not use the main memroot however its not

sql/rpl_replica.cc

@@ -4957,6 +4957,13 @@ static int exec_relay_log_event(THD *thd, Relay_log_info *rli,
       need force to compute checkpoint.
     */
     bool force = rli->rli_checkpoint_seqno >= rli->checkpoint_group;
+    DBUG_EXECUTE_IF("sbm_force_checkpoint", {
+      if (ev->get_type_code() == mysql::binlog::event::QUERY_EVENT) {
+        force = true;
+        DBUG_SET_INITIAL("-d,sbm_force_checkpoint");
+        DBUG_SET_INITIAL("-d,sbm_block_checkpoint");
+      }
+    });
     if (force || rli->is_time_for_mta_checkpoint()) {
       mysql_mutex_unlock(&rli->data_lock);
       if (mta_checkpoint_routine(rli, force)) {
@@ -5004,8 +5011,22 @@ static int exec_relay_log_event(THD *thd, Relay_log_info *rli,
           ev->get_type_code() ==
               mysql::binlog::event::FORMAT_DESCRIPTION_EVENT ||
           ev->server_id == 0)) {
-      rli->last_master_timestamp =
-          ev->common_header->when.tv_sec + (time_t)ev->exec_time;
+      // Prefer immediate_commit_timestamp from GTID event for more precise
+      // SBM calculation. Fall back to when+exec_time for old masters that
+      // do not send commit timestamps, or for non-GTID events.
+      if (is_any_gtid_event(ev)) {
+        auto *gtid_ev = static_cast<Gtid_log_event *>(ev);
+        if (gtid_ev->has_commit_timestamps) {
+          rli->last_master_timestamp = static_cast<time_t>(
+              gtid_ev->immediate_commit_timestamp / 1000000);
+        } else {
+          rli->last_master_timestamp =
+              ev->common_header->when.tv_sec + (time_t)ev->exec_time;
+        }
+      } else {
+        rli->last_master_timestamp =
+            ev->common_header->when.tv_sec + (time_t)ev->exec_time;
+      }
       assert(rli->last_master_timestamp >= 0);
     }
 
@@ -6636,6 +6657,7 @@ bool mta_checkpoint_routine(Relay_log_info *rli, bool force) {
   DBUG_EXECUTE_IF("mta_checkpoint", {
     rpl_replica_debug_point(DBUG_RPL_S_MTS_CHECKPOINT_START, rli->info_thd);
   };);
+  DBUG_EXECUTE_IF("sbm_block_checkpoint", return error;);
 #endif
 
   /*