PS-9238: Make MySQL 5.7 compatible with

CREATE TABLE AS SELECT [...] START TRANSACTION to improve 8.0 -> 5.7
replication reliability

https://perconadev.atlassian.net/browse/PS-9238

Part 2.1 of the fix.
When ctas_compatibility_mode=OFF the binlog sequence is:
BEGIN
CREATE TABLE ... START TRANSACTION
ROW EVENT
COMMIT

MTS sees the above as one group.

When the source produces binlog with CREATE TABLE ... START TRANSACTION
but the intermediate replica is configured with
ctas_compatibility_mode=ON it has to execute and binlog CTAS in legacy
mode (pre 8.0.20). It means the intermediate commit will be done after
CREATE TABLE. This commit will call pre_commit hook, which will call
Slave_worker::commit_positions() where we check the validity of
ptr_group->checkpoint_seqno. checkpoint_seqno however is set when group
end is detected (COMMIT event), but it is too late.

We detect CTAS case and set checkpoint_seqno to be prepared for
intermediate commit.

sql/log_event.cc

@@ -3035,6 +3035,15 @@ Slave_worker *Log_event::get_slave_worker(Relay_log_info *rli) {
 #ifndef NDEBUG
     w_rr++;
 #endif
+  } else if (opt_ctas_compatibility_mode && is_ctas()) {
+    /*
+      In ctas compatibility there will be intermediate commit
+      after CREATE TABLE. It will call pre_commit hook, which will call
+      Slave_worker::commit_positions() where we check the validity of
+      ptr_group->checkpoint_seqno.
+    */
+    ptr_group->checkpoint_seqno = rli->rli_checkpoint_seqno;
+    rli->rli_checkpoint_seqno++;
   }
 
   return ret_worker;

sql/log_event.h

@@ -1099,6 +1099,7 @@ class Log_event {
              false otherwise
   */
   virtual bool ends_group() const { return false; }
+  virtual bool is_ctas() const { return false; }
 #ifdef MYSQL_SERVER
   /**
      Apply the event to the database.
@@ -1471,6 +1472,12 @@ class Query_log_event : public virtual binary_log::Query_event,
             native_strncasecmp(query, STRING_WITH_LEN("ROLLBACK TO "))) ||
            !strncmp(query, STRING_WITH_LEN("XA ROLLBACK"));
   }
+
+  bool is_ctas() const override {
+    return (strstr(query, "CREATE TABLE") != nullptr) &&
+           (strstr(query, "START TRANSACTION") != nullptr);
+  }
+
   static size_t get_query(const char *buf, size_t length,
                           const Format_description_event *fd_event,
                           const char **query_arg);