MDEV-15405 Mixed replication fails with "Could not execute Delete_rows_v1...

MDEV-15405 Mixed replication fails with "Could not execute Delete_rows_v1 event" upon DELETE HISTORY

Allow slave thread to set time for system versioning

Note that every binlog event stores now(0), while microseconds
are only stored when they're actually used in the query.

Meaning for unversioned->versioned replication, there will be
no microseconds. Need to compensate for that.

mysql-test/suite/versioning/r/rpl_mix.result

+include/master-slave.inc
+[connection master]
+CREATE TABLE t1 (pk INT PRIMARY KEY, i INT) WITH SYSTEM VERSIONING;
+INSERT INTO t1 VALUES (1,10),(2,20);
+UPDATE t1 SET i = 100;
+SET BINLOG_FORMAT= ROW;
+DELETE HISTORY FROM t1;
+connection slave;
+connection master;
+drop table t1;
+include/rpl_end.inc

mysql-test/suite/versioning/t/rpl_mix.test

+--source include/have_binlog_format_mixed.inc
+--source include/master-slave.inc
+
+#
+# MDEV-15405 Mixed replication fails with "Could not execute Delete_rows_v1 event" upon DELETE HISTORY
+#
+
+CREATE TABLE t1 (pk INT PRIMARY KEY, i INT) WITH SYSTEM VERSIONING;
+INSERT INTO t1 VALUES (1,10),(2,20);
+UPDATE t1 SET i = 100;
+# This is to imitiate any real-life situation which causes a switch to row format
+SET BINLOG_FORMAT= ROW;
+DELETE HISTORY FROM t1;
+--sync_slave_with_master
+
+connection master;
+drop table t1;
+
+--source include/rpl_end.inc

sql/log_event.cc

@@ -1312,7 +1312,7 @@ Log_event::Log_event(const char* buf,
   thd = 0;
 #endif
   when = uint4korr(buf);
-  when_sec_part= 0;
+  when_sec_part= ~0UL;
   server_id = uint4korr(buf + SERVER_ID_OFFSET);
   data_written= uint4korr(buf + EVENT_LEN_OFFSET);
   if (description_event->binlog_version==1)
@@ -5090,7 +5090,7 @@ bool Query_log_event::print_query_header(IO_CACHE* file,
   }
 
   end=int10_to_str((long) when, strmov(buff,"SET TIMESTAMP="),10);
-  if (when_sec_part)
+  if (when_sec_part && when_sec_part <= TIME_MAX_SECOND_PART)
   {
     *end++= '.';
     end=int10_to_str(when_sec_part, end, 10);

sql/sql_class.h

@@ -3434,8 +3434,8 @@ class THD :public Statement,
     ulong sec_part;
   } system_time;
 
-  ulong systime_sec_part() { return system_time.sec_part; }
-  my_time_t systime() { return system_time.sec; }
+  ulong systime_sec_part() { query_start_sec_part_used=1; return system_time.sec_part; }
+  my_time_t systime() { query_start_used=1; return system_time.sec; }
 
 private:
   void set_system_time()
@@ -3462,6 +3462,25 @@ class THD :public Statement,
     }
   }
 
+  void set_system_time_from_user_time(bool with_sec_part)
+  {
+    if (with_sec_part)
+    {
+      system_time.sec= start_time;
+      system_time.sec_part= start_time_sec_part;
+    }
+    else
+    {
+      if (system_time.sec == start_time)
+        system_time.sec_part++;
+      else
+      {
+        system_time.sec= start_time;
+        system_time.sec_part= 0;
+      }
+    }
+  }
+
 public:
   inline void set_start_time()
   {
@@ -3488,10 +3507,21 @@ class THD :public Statement,
     user_time= t;
     set_time();
   }
+  /*
+    this is only used by replication and BINLOG command.
+    usecs > TIME_MAX_SECOND_PART means "was not in binlog"
+  */
   inline void set_time(my_time_t t, ulong sec_part)
   {
-    my_hrtime_t hrtime= { hrtime_from_time(t) + sec_part };
-    set_time(hrtime);
+    start_time= t;
+    start_time_sec_part= sec_part > TIME_MAX_SECOND_PART ? 0 : sec_part;
+    user_time.val= hrtime_from_time(start_time) + start_time_sec_part;
+    if (slave_thread)
+      set_system_time_from_user_time(sec_part <= TIME_MAX_SECOND_PART);
+    else // BINLOG command
+      set_system_time();
+    PSI_CALL_set_thread_start_time(start_time);
+    start_utime= utime_after_lock= microsecond_interval_timer();
   }
   void set_time_after_lock()
   {