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

Problem:
5.7 does not understand the new syntax and complains about
CREATE TABLE ... START TRANSACTION received in binlog.

Cause:
Commit 4fe3f2f (8.0.21) introduced changes that allow the execution
of CREATE TABLE AS SELECT as a single transaction. Before that change,
CREATE TABLE was the 1st transaction, then rows were inserted, which was
unsafe. To achieve this we don't commit the transaction after
CREATE TABLE. The commit is done after rows are inserted. For async
replica to understand this protocol, START TRANSACTION clause was added
to CREATE TABLE in binlog. When the replica sees this, it knows not to
commit after CREATE TABLE.

Solution:
Introduced ctas_compatibility_mode global, read-only variable, OFF by
default. If switched to ON, it enforces the behavior from before the
above-mentioned commit.

Implementation details:
The fix contains 2 parts:
1. Query_result_create::binlog_show_create_table() - When SE supports
atomic DDL, CREATE TABLE query was binlogged with
direct=false / is_trans=true flags. This caused the event to be cached
in trx_cache instead of stmt_cache.
MYSQL_BIN_LOG::commit() logic skips trx_cache commit, if this is not
the transaction commit (it does only stmt_cache commit). Then, when rows
are inserted, it was detected that there is ongoing transaction
(trx_cache not empty), so no new BEGIN statement was generated
in binlog.
Effectively CREATE TABLE and rows insertion were done in one
transaction.

The fix is to revert to the default behavior, i.e.,
Query_result_create::binlog_show_create_table() creates the event in
stmt_cache, which is committed after table creation. When rows are being
inserted, it is detected that trx_cache is empty, so BEGIN statement is
added to binlog.

2. store_create_info()—When executing CTAS, the START TRANSACTION clause
was added to the rewritten query to inform the replica about what was
happening.

The fix is not to add the START TRANSACTION clause.

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

@@ -24,6 +24,8 @@ binlog_expire_logs_auto_purge
 binlog_expire_logs_auto_purge
 binlog_rotate_encryption_master_key_at_startup
 binlog_rotate_encryption_master_key_at_startup
+ctas_compatibility_mode
+ctas_compatibility_mode
 cte_max_recursion_depth
 cte_max_recursion_depth
 default_collation_for_utf8mb4

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

@@ -0,0 +1,11 @@
+include/assert.inc [ctas_compatibility_mode default should be OFF]
+SET GLOBAL ctas_compatibility_mode = ON;
+ERROR HY000: Variable 'ctas_compatibility_mode' is a read only variable
+CREATE TABLE t1 (a int);
+INSERT INTO t1 VALUES (0);
+# restart: --ctas-compatibility-mode=ON --log-bin=ctas_binlog
+CREATE TABLE t2 AS SELECT * FROM t1;
+# restart:
+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;

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

@@ -0,0 +1,45 @@
+#
+# The default is OFF
+#
+--let $assert_text= ctas_compatibility_mode default should be OFF
+--let $assert_cond= "[SHOW GLOBAL VARIABLES LIKE "ctas_compatibility_mode", Value, 1]" = "OFF"
+--source include/assert.inc
+
+#
+# Check that it is read-only
+#
+--error ER_INCORRECT_GLOBAL_LOCAL_VAR
+SET GLOBAL ctas_compatibility_mode = ON; 
+
+#
+# Check that compatibility mode works 
+#
+CREATE TABLE t1 (a int);
+INSERT INTO t1 VALUES (0);
+
+--let $binlog_file = ctas_binlog
+--let $restart_parameters = "restart: --ctas-compatibility-mode=ON --log-bin=$binlog_file"
+--source include/restart_mysqld.inc
+
+CREATE TABLE t2 AS SELECT * FROM t1;
+
+--let $restart_parameters = "restart:"
+--source include/restart_mysqld.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"
+--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"
+--let $assert_select = BEGIN
+--let $assert_count = 1
+--source include/assert_grep.inc
+
+# cleanup
+DROP TABLE t1, t2;
+--remove_file $MYSQLD_DATADIR/$binlog_file.000001
+--remove_file $MYSQLTEST_VARDIR/tmp/$binlog_file.sql

sql/mysqld.cc

@@ -1242,6 +1242,7 @@ bool opt_log_replica_updates = false;
 char *opt_replica_skip_errors;
 bool opt_replica_allow_batching = true;
 
+bool opt_ctas_compatibility_mode = true;
 /**
   compatibility option:
     - index usage hints (USE INDEX without a FOR clause) behave as in 5.0

sql/mysqld.h

@@ -197,6 +197,7 @@ extern Rpl_acf_configuration_handler *rpl_acf_configuration_handler;
 extern Source_IO_monitor *rpl_source_io_monitor;
 extern int32_t opt_regexp_time_limit;
 extern int32_t opt_regexp_stack_limit;
+extern bool opt_ctas_compatibility_mode;
 #ifdef _WIN32
 extern bool opt_no_monitor;
 #endif  // _WIN32

sql/sql_insert.cc

@@ -3039,8 +3039,8 @@ int Query_result_create::binlog_show_create_table(THD *thd) {
 
     bool is_trans = false;
     bool direct = true;
-    if (get_default_handlerton(thd, thd->lex->create_info->db_type)->flags &
-        HTON_SUPPORTS_ATOMIC_DDL) {
+    if ((get_default_handlerton(thd, thd->lex->create_info->db_type)->flags &
+        HTON_SUPPORTS_ATOMIC_DDL) && !opt_ctas_compatibility_mode) {
       is_trans = true;
       direct = false;
     }

sql/sql_show.cc

@@ -2363,7 +2363,8 @@ bool store_create_info(THD *thd, Table_ref *table_list, String *packet,
     This is done only while binlogging CREATE TABLE AS SELECT.
   */
   if (!thd->lex->query_block->field_list_is_empty() &&
-      (create_info_arg->db_type->flags & HTON_SUPPORTS_ATOMIC_DDL)) {
+      (create_info_arg->db_type->flags & HTON_SUPPORTS_ATOMIC_DDL) &&
+      !opt_ctas_compatibility_mode) {
     packet->append(STRING_WITH_LEN(" START TRANSACTION"));
   }
 

sql/sys_vars.cc

@@ -4292,6 +4292,13 @@ static Sys_var_int32 Sys_regexp_stack_limit(
     GLOBAL_VAR(opt_regexp_stack_limit), CMD_LINE(REQUIRED_ARG),
     VALID_RANGE(0, INT32_MAX), DEFAULT(8000000), BLOCK_SIZE(1));
 
+static Sys_var_bool Sys_ctas_compatibility_mode(
+    "ctas_compatibility_mode",
+    "Execute and binlog CTAS in pre 8.0.21 way, i.e. with intermediate commit "
+    "after the table creation.",
+    READ_ONLY GLOBAL_VAR(opt_ctas_compatibility_mode), CMD_LINE(OPT_ARG),
+    DEFAULT(false));
+
 static Sys_var_bool Sys_replica_compressed_protocol(
     "replica_compressed_protocol",
     "Use compression in the source/replica protocol.",