diff --git a/mysql-test/suite/sys_vars/r/ctas_compatibility_mode.result b/mysql-test/suite/sys_vars/r/ctas_compatibility_mode.result
index e7c4dc15322a..5c47c95f1e15 100644
--- a/mysql-test/suite/sys_vars/r/ctas_compatibility_mode.result
+++ b/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
diff --git a/mysql-test/suite/sys_vars/t/ctas_compatibility_mode.test b/mysql-test/suite/sys_vars/t/ctas_compatibility_mode.test
index 07de6ef442cb..2637502bf0f2 100644
--- a/mysql-test/suite/sys_vars/t/ctas_compatibility_mode.test
+++ b/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
diff --git a/sql/parse_tree_nodes.h b/sql/parse_tree_nodes.h
index a4123343727f..4c7fb486798b 100644
--- a/sql/parse_tree_nodes.h
+++ b/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>
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 4119753ca65c..6af0b346fa69 100644
--- a/sql/sql_table.cc
+++ b/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);
       }
     }
   }