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 of the fix.
In the scenario source -> replica1 -> replica2, if replica1 is
configured with ctas_compatibility_mode it should behave accordingly and
produce the binglog compatible with version 8.0.20 (intermediate commit
after CREATE TABLE and no START TRANSACTION clause)

This part is sensitive to create_info->m_transactional_ddl flag which is
set during statement parsing. If ctas_compatibility_mode is set, we
force this flag not to be set.

Additionally, replica side follows the path of regular CREATE TABLE
which binlogs the query as-is. if ctas_compatibility_mode is set,
CREATE TABLE is removed from the query.

mysql-test/suite/sys_vars/r/ctas_compatibility_mode.result

@@ -9,3 +9,25 @@ CREATE TABLE t2 AS SELECT * FROM t1;
 include/assert_grep.inc ["Checking if ctas_compatibility_mode works"]
 include/assert_grep.inc ["Checking if rows are inserted as the separate transaction"]
 DROP TABLE t1, t2;
+include/rpl_init.inc [topology=1->2]
+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.
+CREATE TABLE t1 (a int);
+INSERT INTO t1 VALUES (0);
+include/rpl_sync.inc
+[connection server_2]
+include/rpl_restart_server.inc [server_number=2 parameters: --ctas-compatibility-mode=ON --log-bin=ctas_binlog]
+include/rpl_start_slaves.inc
+[connection server_1]
+CREATE TABLE t2 AS SELECT * FROM t1;
+include/rpl_sync.inc
+[connection server_2]
+include/rpl_restart_server.inc [server_number=2]
+include/rpl_start_slaves.inc
+[connection server_2]
+include/assert_grep.inc ["Checking if ctas_compatibility_mode works on replica"]
+include/assert_grep.inc ["Checking if rows are inserted as the separate transaction on replica"]
+[connection server_1]
+DROP TABLE t1, t2;
+include/rpl_end.inc

mysql-test/suite/sys_vars/t/ctas_compatibility_mode.test

@@ -43,3 +43,62 @@ let $MYSQLD_DATADIR= `select @@datadir;`;
 DROP TABLE t1, t2;
 --remove_file $MYSQLD_DATADIR/$binlog_file.000001
 --remove_file $MYSQLTEST_VARDIR/tmp/$binlog_file.sql
+
+
+#
+# Check the behavior of replica started with ctas_compatibility_mode enabled
+#
+--let $rpl_topology = 1->2
+--source include/rpl_init.inc
+CREATE TABLE t1 (a int);
+INSERT INTO t1 VALUES (0);
+--source include/rpl_sync.inc
+
+# Now restart replica with ctas-compatibility-mode=ON and a custom binlog file
+--let $rpl_connection_name = server_2
+--source include/rpl_connection.inc
+--let $binlog_file = ctas_binlog
+--let $rpl_server_parameters = --ctas-compatibility-mode=ON --log-bin=$binlog_file
+--let $rpl_server_number = 2
+--source include/rpl_restart_server.inc
+--source include/rpl_start_slaves.inc
+
+# Execute CTAS on source server
+--let $rpl_connection_name = server_1
+--source include/rpl_connection.inc
+CREATE TABLE t2 AS SELECT * FROM t1;
+--source include/rpl_sync.inc
+
+# Restart replica with the default parameters
+--let $rpl_connection_name = server_2
+--source include/rpl_connection.inc
+--let $rpl_server_parameters =
+--let $rpl_server_number = 2
+--source include/rpl_restart_server.inc
+--source include/rpl_start_slaves.inc
+
+# We expect that replica started with --ctas-compatibility-mode=ON behaved accordingly
+--let $rpl_connection_name = server_2
+--source include/rpl_connection.inc
+let $MYSQLD_DATADIR= `select @@datadir;`;
+--exec $MYSQL_BINLOG --verbose $MYSQLD_DATADIR/$binlog_file.000001 > $MYSQLTEST_VARDIR/tmp/$binlog_file.sql
+--let $assert_file = $MYSQLTEST_VARDIR/tmp/$binlog_file.sql
+--let $assert_text = "Checking if ctas_compatibility_mode works on replica"
+--let $assert_select = START TRANSACTION
+--let $assert_count = 0
+--source include/assert_grep.inc
+
+--let $assert_text = "Checking if rows are inserted as the separate transaction on replica"
+--let $assert_select = BEGIN
+--let $assert_count = 1
+--source include/assert_grep.inc
+
+# cleanup
+--remove_file $MYSQLD_DATADIR/$binlog_file.000001
+--remove_file $MYSQLTEST_VARDIR/tmp/$binlog_file.sql
+
+--let $rpl_connection_name = server_1
+--source include/rpl_connection.inc
+DROP TABLE t1, t2;
+
+--source include/rpl_end.inc

sql/parse_tree_nodes.h

@@ -46,6 +46,7 @@
 #include "sql/handler.h"
 #include "sql/key_spec.h"
 #include "sql/mem_root_array.h"
+#include "sql/mysqld.h"	 // opt_ctas_compatibility_mode
 #include "sql/opt_explain.h"  // Sql_cmd_explain_other_thread
 #include "sql/parse_location.h"
 #include "sql/parse_tree_helpers.h"  // PT_item_list
@@ -2489,6 +2490,34 @@ typedef PT_traceable_create_table_option<
     HA_CREATE_USED_START_TRANSACTION>
     PT_create_start_transaction_option;
 
+class PT_create_start_transaction_option
+    : public PT_traceable_create_table_option<
+          TYPE_AND_REF(HA_CREATE_INFO::m_transactional_ddl),
+          HA_CREATE_USED_START_TRANSACTION> {
+  typedef PT_create_table_option super;
+
+  decltype(HA_CREATE_INFO::m_transactional_ddl) value;
+
+ public:
+  explicit PT_create_start_transaction_option(
+      decltype(HA_CREATE_INFO::m_transactional_ddl) value)
+      : PT_traceable_create_table_option<
+            TYPE_AND_REF(HA_CREATE_INFO::m_transactional_ddl),
+            HA_CREATE_USED_START_TRANSACTION>(value),
+        value(value) {}
+
+  bool contextualize(Table_ddl_parse_context *pc) override {
+    if (super::contextualize(pc)) return true;
+    if (opt_ctas_compatibility_mode) {
+      pc->create_info->m_transactional_ddl = false;
+    } else {
+      pc->create_info->m_transactional_ddl = value;
+      pc->create_info->used_fields |= HA_CREATE_USED_START_TRANSACTION;
+    }
+    return false;
+  }
+};
+
 typedef PT_traceable_create_table_option<
     TYPE_AND_REF(HA_CREATE_INFO::m_implicit_tablespace_autoextend_size),
     HA_CREATE_USED_AUTOEXTEND_SIZE>

sql/sql_table.cc

@@ -10396,8 +10396,23 @@ bool mysql_create_table(THD *thd, Table_ref *create_table,
           }
         }
       } else {
-        result = write_bin_log(thd, true, thd->query().str, thd->query().length,
-                               is_trans);
+        /*
+          We can get here from replica thread executing
+          CREATE TABLE ... START TRANSACTION. If ctas_compatibility_mode==true
+          we follow the right path because of create_info->m_transactional_ddl
+          being set properly to false in
+          PT_create_start_transaction_option::contextualize(), but we need to
+          remove START TRANSACTION clause from the query before binlogging it.
+        */
+        size_t query_length = thd->query().length;
+        if (opt_ctas_compatibility_mode) {
+          const char *pos = strstr(thd->query().str, "START TRANSACTION");
+          if (pos != nullptr) {
+            query_length = pos - thd->query().str;
+          }
+        }
+        result =
+            write_bin_log(thd, true, thd->query().str, query_length, is_trans);
       }
     }
   }