WL#5370 Keep forward-compatibility when changing

        'CREATE TABLE IF NOT EXISTS ... SELECT' behaviour
BUG#55474, BUG#55499, BUG#55598, BUG#55616 and BUG#55777 are fixed
in this patch too.

This is the 5.1 part.
It implements:
- if the table exists, binlog two events: CREATE TABLE IF NOT EXISTS
  and INSERT ... SELECT

- Insert nothing and binlog nothing on master if the existing object
  is a view. It only generates a warning that table already exists.


mysql-test/r/trigger.result:
  Ather this patch, 'CREATE TABLE IF NOT EXISTS ... SELECT' will not
  insert anything if the creating table already exists and is a view.
sql/sql_class.h:
  Declare virtual function write_to_binlog() for select_insert.
  It's used to binlog 'create select'
sql/sql_insert.cc:
  Implement write_to_binlog();
  Use write_to_binlog() instead of binlog_query() to binlog the statement.
  if the table exists, binlog two events: CREATE TABLE IF NOT EXISTS
  and INSERT ... SELECT
sql/sql_lex.h:
  Declare create_select_start_with_brace and create_select_pos.
  They are helpful for binlogging 'create select'
sql/sql_parse.cc:
  Do nothing on master if the existing object is a view.
sql/sql_yacc.yy:
  Record the relative postion of 'SELECT' in the 'CREATE ...SELECT' statement.
  Record whether there is a '(' before the 'SELECT' clause.

mysql-test/extra/rpl_tests/rpl_stm_create_if_not_exists.test

+--echo
+--echo
+connection master;
+
+if ($is_temporary)
+{
+  --let $_temporary=TEMPORARY
+}
+
+CREATE TABLE t2(c1 INT, c2 char(10));
+INSERT INTO t2 VALUES(1, 'abc'), (2, 'abc');
+
+--echo
+--echo # The original query should be binlogged if the table does not exist.
+--echo # ------------------------------------------------------------------
+--echo
+let binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1);
+eval CREATE $_temporary TABLE IF NOT EXISTS t1 (c1 INT , c2 INT, c3 char(10), c4 INT KEY)
+  SELECT 'abc' AS c3, 1 AS c4;
+source include/show_binlog_events.inc;
+
+if (!$is_temporary)
+{
+  let $diff_table= test.t1;
+  source include/rpl_diff_tables.inc;
+}
+
+--echo
+--echo # The statement should be binlogged as two events. one is 
+--echo # 'CREATE $_temporary TABLE IF NOT EXISTS ..', another one is
+--echo # 'INSERT ... SELECT'.
+--echo # ------------------------------------------------------------------
+--echo
+let binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1);
+eval CREATE $_temporary TABLE IF NOT EXISTS t1
+  SELECT 'abc', 2;
+source include/show_binlog_events.inc;
+
+if (!$is_temporary)
+{
+  let $diff_table= test.t1;
+  source include/rpl_diff_tables.inc;
+}
+
+--echo
+--echo # Verify if it can be binlogged with right database name when the table
+--echo # is not in the default database
+--echo
+--disable_warnings
+DROP DATABASE IF EXISTS db1;
+--enable_warnings
+CREATE DATABASE db1;
+USE db1;
+
+let binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1);
+eval CREATE $_temporary TABLE IF NOT EXISTS test.t1
+  SELECT 'abc', 20;
+source include/show_binlog_events.inc;
+
+if (!$is_temporary)
+{
+  let $diff_table= test.t1;
+  source include/rpl_diff_tables.inc;
+}
+USE test;
+DROP DATABASE db1;
+
+--echo
+--echo # It should be binlogged as 'REPLACE ... SELECT'
+--echo # if the original statement has option REPLACE
+--echo
+let binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1);
+eval CREATE $_temporary TABLE IF NOT EXISTS t1
+  REPLACE SELECT '123', 2;
+source include/show_binlog_events.inc;
+
+if (!$is_temporary)
+{
+  let $diff_table= test.t1;
+  source include/rpl_diff_tables.inc;
+}
+
+--echo
+--echo # It should be binlogged as 'INSERT IGNORE... SELECT'
+--echo # if the original statement has option IGNORE 
+--echo
+let binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1);
+eval CREATE $_temporary TABLE IF NOT EXISTS t1
+  IGNORE SELECT '123', 2;
+source include/show_binlog_events.inc;
+
+if (!$is_temporary)
+{
+  let $diff_table= test.t1;
+  source include/rpl_diff_tables.inc;
+}
+
+--echo
+--echo # Nothing should be binlogged if error happens and no any row is inserted
+--echo
+let binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1);
+--error ER_DUP_ENTRY
+eval CREATE $_temporary TABLE IF NOT EXISTS t1
+  SELECT '123', 2;
+source include/show_binlog_events.inc;
+if (!$is_temporary)
+{
+  let $diff_table= test.t1;
+  source include/rpl_diff_tables.inc;
+}
+
+--echo
+--echo # Verify it can binlog well when there are some braces('(')
+--echo
+let binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1);
+eval CREATE $_temporary TABLE IF NOT EXISTS t1
+  (SELECT '123', 3) UNION (SELECT '123', 4);
+eval CREATE $_temporary TABLE IF NOT EXISTS t1
+  REPLACE (SELECT 'abc', 3) UNION (SELECT 'abc', 4);
+eval CREATE $_temporary TABLE IF NOT EXISTS t1
+  IGNORE (SELECT '123', 3) UNION (SELECT '123', 4);
+source include/show_binlog_events.inc;
+
+if (!$is_temporary)
+{
+  let $diff_table= test.t1;
+  source include/rpl_diff_tables.inc;
+}
+
+if (!$is_temporary)
+{
+  --echo
+  --echo # Throw a warning that table already exists and don't insert anything
+  --echo
+  CREATE VIEW t3 AS SELECT * FROM t2;
+  let binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1);
+
+  CREATE TABLE IF NOT EXISTS t3
+    SELECT '123', 2;
+  source include/show_binlog_events.inc;
+  DROP VIEW t3;
+}
+
+--echo
+--echo # The statement can be binlogged correctly when it is in a SP/EVENT/TRIGGER
+--echo
+
+--disable_warnings
+DROP PROCEDURE IF EXISTS p1;
+--enable_warnings
+eval CREATE PROCEDURE p1(IN a INT)
+  CREATE $_temporary TABLE IF NOT EXISTS t1 SELECT '123', a;
+
+let binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1);
+call p1(500);
+call p1(600);
+source include/show_binlog_events.inc;
+if (!$is_temporary)
+{
+  let $diff_table= test.t1;
+  source include/rpl_diff_tables.inc;
+}
+DROP PROCEDURE p1;
+
+--echo
+--echo # The statement can be binlogged correctly when it is in a prepared statement
+--echo
+eval PREPARE stm FROM "CREATE $_temporary TABLE IF NOT EXISTS t1 SELECT '123', ?";
+
+let binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1);
+SET @a= 700;
+EXECUTE stm USING @a;
+SET @a= 800;
+EXECUTE stm USING @a;
+source include/show_binlog_events.inc;
+if (!$is_temporary)
+{
+  let $diff_table= test.t1;
+  source include/rpl_diff_tables.inc;
+}
+
+--echo
+--echo # The statement can be binlogged correctly when it is in a conditional comment
+--echo
+let binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1);
+
+--echo # The whole statement in a conditional comment
+eval /*!CREATE $_temporary TABLE IF NOT EXISTS t1
+  SELECT 'abc', 900*/;
+source include/show_binlog_events.inc;
+let binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1);
+
+--echo
+--echo # There is an long comment before SELECT
+eval /*!CREATE $_temporary /*blabla*/ TABLE IF NOT EXISTS t1
+  SELECT 'abc', 901*/;
+source include/show_binlog_events.inc;
+let binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1);
+
+--echo
+--echo # Conditional comment starts just from SELECT
+eval CREATE $_temporary TABLE IF NOT EXISTS t1
+  /*!SELECT 'abc',*/ 902;
+source include/show_binlog_events.inc;
+let binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1);
+
+--echo
+--echo # Only SELECT keyword is in the conditional comment
+eval CREATE $_temporary TABLE IF NOT EXISTS t1
+  /*!SELECT*/ /*!'abc',*/ 904;
+source include/show_binlog_events.inc;
+let binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1);
+
+--echo
+--echo # Conditional comment is after SELECT keyword
+eval CREATE $_temporary TABLE IF NOT EXISTS t1
+  SELECT /*!'abc',*/ 903;
+source include/show_binlog_events.inc;
+let binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1);
+
+--echo
+--echo # Conditional comment ends just before SELECT keyword
+eval /*!CREATE $_temporary TABLE IF NOT EXISTS t1
+  */SELECT 'abc', 905;
+source include/show_binlog_events.inc;
+
+if (!$is_temporary)
+{
+  let $diff_table= test.t1;
+  source include/rpl_diff_tables.inc;
+}
+
+DROP TABLE t2;
+eval DROP $_temporary TABLE t1;
+

mysql-test/include/rpl_diff_tables.inc

@@ -33,3 +33,4 @@ while (`SELECT "XX$_servers" <> "XX"`)
   --source include/diff_tables.inc
   connection $_slave;
 }
+connection $_master;

mysql-test/r/trigger.result

@@ -1824,11 +1824,8 @@ Note	1050	Table 'v1' already exists
 set @id=last_insert_id();
 select * from t1;
 id	operation
-1	CREATE TABLE ... SELECT, inserting a new key
 select * from t1_op_log;
 operation
-Before INSERT, new=CREATE TABLE ... SELECT, inserting a new key
-After INSERT, new=CREATE TABLE ... SELECT, inserting a new key
 truncate t1_op_log;
 create table if not exists v1 replace
 select @id, "CREATE TABLE ... REPLACE SELECT, deleting a duplicate key";
@@ -1836,13 +1833,8 @@ Warnings:
 Note	1050	Table 'v1' already exists
 select * from t1;
 id	operation
-1	CREATE TABLE ... REPLACE SELECT, deleting a duplicate key
 select * from t1_op_log;
 operation
-Before INSERT, new=CREATE TABLE ... REPLACE SELECT, deleting a duplicate key
-Before DELETE, old=CREATE TABLE ... SELECT, inserting a new key
-After DELETE, old=CREATE TABLE ... SELECT, inserting a new key
-After INSERT, new=CREATE TABLE ... REPLACE SELECT, deleting a duplicate key
 truncate t1;
 truncate t1_op_log;
 insert into v1 (id, operation)

mysql-test/suite/rpl/r/rpl_create_if_not_exists.result

@@ -65,3 +65,12 @@ c1
 DROP TABLE t1;
 DROP TABLE t2;
 DROP TABLE t3;
+
+# Bug#55616 Killing thread or query during CREATE IF NOT EXISTS makes
+# slave SQL thread abort
+
+CREATE TABLE t1 ( i INT );
+CREATE TABLE IF NOT EXISTS t1
+AS SELECT SLEEP(3);
+KILL QUERY master1;
+DROP TABLE t1;

mysql-test/suite/rpl/r/rpl_create_tmp_table_if_not_exists.result

@@ -19,4 +19,9 @@ master-bin.000001	#	Query	#	#	use `test`; CREATE TEMPORARY TABLE IF NOT EXISTS t
 master-bin.000001	#	Query	#	#	use `test`; CREATE TEMPORARY TABLE IF NOT EXISTS tmp1 LIKE tmp
 master-bin.000001	#	Query	#	#	use `test`; CREATE TEMPORARY TABLE IF NOT EXISTS tmp1 LIKE tmp
 master-bin.000001	#	Query	#	#	use `test`; CREATE TEMPORARY TABLE IF NOT EXISTS tmp2 SELECT * FROM tmp
-master-bin.000001	#	Query	#	#	use `test`; CREATE TEMPORARY TABLE IF NOT EXISTS tmp2 SELECT * FROM tmp
+master-bin.000001	#	Query	#	#	BEGIN
+master-bin.000001	#	Query	#	#	use `test`; CREATE TEMPORARY TABLE IF NOT EXISTS `tmp2` (
+  `c1` int(11) DEFAULT NULL
+)
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO `test`.`tmp2` (`c1`) SELECT * FROM tmp
+master-bin.000001	#	Query	#	#	COMMIT

mysql-test/suite/rpl/r/rpl_row_create_table.result

@@ -467,4 +467,10 @@ DROP VIEW IF EXISTS bug48506_t1, bug48506_t2, bug48506_t3;
 DROP TEMPORARY TABLES t7;
 DROP TABLES t4, t5;
 DROP TABLES IF EXISTS bug48506_t4;
+CREATE TABLE t1 SELECT 1;
+CREATE TABLE IF NOT EXISTS t1 SELECT 1;
+Warnings:
+Note	1050	Table 't1' already exists
+Comparing tables master:test.t1 and slave:test.t1
+DROP TABLE t1;
 end of the tests

mysql-test/suite/rpl/t/rpl_create_if_not_exists.test

@@ -119,5 +119,32 @@ SELECT * FROM t2;
 DROP TABLE t1;
 DROP TABLE t2;
 DROP TABLE t3;
+sync_slave_with_master;
+
+--echo
+--echo # Bug#55616 Killing thread or query during CREATE IF NOT EXISTS makes
+--echo # slave SQL thread abort
+--echo
+
+--connection master1
+let $con_id = `SELECT CONNECTION_ID()`;
+
+CREATE TABLE t1 ( i INT );
+send CREATE TABLE IF NOT EXISTS t1
+  AS SELECT SLEEP(3);
+
+connection master;
+let $wait_timeout = 3;
+let $show_statement = SHOW PROCESSLIST;
+let $field = State;
+let $condition = = 'User sleep';
+source include/wait_show_condition.inc;
+
+--replace_result $con_id master1
+eval KILL QUERY $con_id;
+sync_slave_with_master;
+
+connection master;
+DROP TABLE t1;
 
 source include/master-slave-end.inc;

mysql-test/suite/rpl/t/rpl_row_create_table.test

@@ -299,5 +299,18 @@ DROP VIEW IF EXISTS bug48506_t1, bug48506_t2, bug48506_t3;
 DROP TEMPORARY TABLES t7;
 DROP TABLES t4, t5;
 DROP TABLES IF EXISTS bug48506_t4;
+sync_slave_with_master;
+
+#
+# Bug#55598 RBR: CREATE TABLE IF NOT EXISTS and INSERT written to binary log
+# twice
+#
+connection master;
+CREATE TABLE t1 SELECT 1;
+CREATE TABLE IF NOT EXISTS t1 SELECT 1;
+let $diff_table=test.t1;
+source include/rpl_diff_tables.inc;
+DROP TABLE t1;
+
 source include/master-slave-end.inc;
 --echo end of the tests

mysql-test/suite/rpl/t/rpl_stm_create_if_not_exists.test

+#
+--echo # WL#5370 Keep forward-compatibility when changing 'CREATE TABLE IF NOT
+--echo # EXISTS ... SELECT' behaviour
+--echo #
+
+source include/master-slave.inc;
+source include/have_binlog_format_statement.inc;
+
+source extra/rpl_tests/rpl_stm_create_if_not_exists.test;
+
+let $is_temporary=1;
+source extra/rpl_tests/rpl_stm_create_if_not_exists.test;
+
+source include/master-slave-end.inc;

sql/sql_class.h

@@ -2622,7 +2622,9 @@ class select_dump :public select_to_file {
 
 
 class select_insert :public select_result_interceptor {
- public:
+protected:
+  virtual int write_to_binlog(bool is_trans, int errcode);
+public:
   TABLE_LIST *table_list;
   TABLE *table;
   List<Item> *fields;
@@ -2658,6 +2660,8 @@ class select_create: public select_insert {
   MYSQL_LOCK *m_lock;
   /* m_lock or thd->extra_lock */
   MYSQL_LOCK **m_plock;
+
+  virtual int write_to_binlog(bool is_trans, int errcode);
 public:
   select_create (TABLE_LIST *table_arg,
 		 HA_CREATE_INFO *create_info_par,
@@ -2673,7 +2677,7 @@ class select_create: public select_insert {
     {}
   int prepare(List<Item> &list, SELECT_LEX_UNIT *u);
 
-  int binlog_show_create_table(TABLE **tables, uint count);
+  int binlog_show_create_table(TABLE **tables, uint count, int errcode);
   void store_values(List<Item> &values);
   void send_error(uint errcode,const char *err);
   bool send_eof();

sql/sql_insert.cc

@@ -3268,7 +3268,7 @@ bool select_insert::send_eof()
 
   /*
     Write to binlog before commiting transaction.  No statement will
-    be written by the binlog_query() below in RBR mode.  All the
+    be written by the write_to_binlog() below in RBR mode.  All the
     events are in the transaction cache and will be written when
     ha_autocommit_or_rollback() is issued below.
   */
@@ -3280,9 +3280,8 @@ bool select_insert::send_eof()
       thd->clear_error();
     else
       errcode= query_error_code(thd, killed_status == THD::NOT_KILLED);
-    if (thd->binlog_query(THD::ROW_QUERY_TYPE,
-                      thd->query(), thd->query_length(),
-                      trans_table, FALSE, errcode))
+
+    if (write_to_binlog(trans_table, errcode))
     {
       table->file->ha_release_auto_increment();
       DBUG_RETURN(1);
@@ -3356,9 +3355,7 @@ void select_insert::abort() {
         {
           int errcode= query_error_code(thd, thd->killed == THD::NOT_KILLED);
           /* error of writing binary log is ignored */
-          (void) thd->binlog_query(THD::ROW_QUERY_TYPE, thd->query(),
-                                   thd->query_length(),
-                                   transactional_table, FALSE, errcode);
+          write_to_binlog(transactional_table, errcode);
         }
         if (!thd->current_stmt_binlog_row_based && !can_rollback_data())
           thd->transaction.all.modified_non_trans_table= TRUE;
@@ -3373,6 +3370,103 @@ void select_insert::abort() {
   DBUG_VOID_RETURN;
 }
 
+int select_insert::write_to_binlog(bool is_trans, int errcode)
+{
+  /* It is only for statement mode */
+  if (thd->current_stmt_binlog_row_based)
+    return 0;
+
+  return thd->binlog_query(THD::ROW_QUERY_TYPE,
+                           thd->query(), thd->query_length(),
+                           is_trans, FALSE, errcode);
+}
+
+/* Override the select_insert::write_to_binlog */
+int select_create::write_to_binlog(bool is_trans, int errcode)
+{
+  /* It is only for statement mode */
+  if (thd->current_stmt_binlog_row_based)
+    return 0;
+
+  /*
+    WL#5370 Keep the compatibility between 5.1 master and 5.5 slave.
+    Binlog a 'INSERT ... SELECT' statement only when it has the option
+    'IF NOT EXISTS' and the table already exists as a base table.
+  */
+  if ((create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS) &&
+      create_info->table_existed)
+  {
+    String query;
+    int result;
+
+    thd->binlog_start_trans_and_stmt();
+    /* Binlog the CREATE TABLE IF NOT EXISTS statement */
+    result= binlog_show_create_table(&table, 1, 0);
+    if (result)
+      return result;
+
+    uint db_len= strlen(create_table->db);
+    uint table_len= strlen(create_info->alias);
+    uint select_len= thd->query_length() - thd->lex->create_select_pos;
+    uint field_len= (table->s->fields - (field - table->field)) *
+      (MAX_FIELD_NAME + 3);
+
+    /*
+      pre-allocating memory reduces the times of reallocating memory,
+      when calling query.appen().
+      40bytes is enough for other words("INSERT IGNORE INTO", etc.).
+     */
+    if (query.real_alloc(40 + db_len + table_len + field_len + select_len))
+      return 1;
+
+    if (thd->lex->create_select_in_comment)
+      query.append(STRING_WITH_LEN("/*! "));
+    if (thd->lex->ignore)
+      query.append(STRING_WITH_LEN("INSERT IGNORE INTO `"));
+    else if (thd->lex->duplicates == DUP_REPLACE)
+      query.append(STRING_WITH_LEN("REPLACE INTO `"));
+    else
+      query.append(STRING_WITH_LEN("INSERT INTO `"));
+
+    query.append(create_table->db, db_len);
+    query.append(STRING_WITH_LEN("`.`"));
+    query.append(create_info->alias, table_len);
+    query.append(STRING_WITH_LEN("` "));
+
+    /*
+      The insert items.
+      Field is the the rightmost columns that the rows are inster in.
+    */
+    query.append(STRING_WITH_LEN("("));
+    for (Field **f= field ; *f ; f++)
+    {
+      if (f != field)
+        query.append(STRING_WITH_LEN(","));
+
+      query.append(STRING_WITH_LEN("`"));
+      query.append((*f)->field_name, strlen((*f)->field_name));
+      query.append(STRING_WITH_LEN("`"));
+    }
+    query.append(STRING_WITH_LEN(") "));
+
+    /* The SELECT clause*/
+    DBUG_ASSERT(thd->lex->create_select_pos);
+    if (thd->lex->create_select_start_with_brace)
+      query.append(STRING_WITH_LEN("("));
+    if (query.append(thd->query() + thd->lex->create_select_pos, select_len))
+      return 1;
+
+    /*
+      Avoid to use thd->binlog_query() twice, otherwise it will print the unsafe
+      warning twice.
+    */
+    Query_log_event ev(thd, query.c_ptr_safe(), query.length(), is_trans,
+                       FALSE, errcode);
+    return mysql_bin_log.write(&ev);
+  }
+  else
+    return select_insert::write_to_binlog(is_trans, errcode);
+}
 
 /***************************************************************************
   CREATE TABLE (SELECT) ...
@@ -3613,7 +3707,8 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
           !table->s->tmp_table &&
           !ptr->get_create_info()->table_existed)
       {
-        if (int error= ptr->binlog_show_create_table(tables, count))
+        int errcode= query_error_code(thd, thd->killed == THD::NOT_KILLED);
+        if (int error= ptr->binlog_show_create_table(tables, count, errcode))
           return error;
       }
       return 0;
@@ -3654,7 +3749,10 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
                           ER_TABLE_EXISTS_ERROR, ER(ER_TABLE_EXISTS_ERROR),
                           create_table->table_name);
       if (thd->current_stmt_binlog_row_based)
-        binlog_show_create_table(&(create_table->table), 1);
+      {
+        int errcode= query_error_code(thd, thd->killed == THD::NOT_KILLED);
+        binlog_show_create_table(&(create_table->table), 1, errcode);
+      }
       table= create_table->table;
     }
     else
@@ -3722,10 +3820,10 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
 }
 
 int
-select_create::binlog_show_create_table(TABLE **tables, uint count)
+select_create::binlog_show_create_table(TABLE **tables, uint count, int errcode)
 {
   /*
-    Note 1: In RBR mode, we generate a CREATE TABLE statement for the
+    Note 1: We generate a CREATE TABLE statement for the
     created table by calling store_create_info() (behaves as SHOW
     CREATE TABLE).  In the event of an error, nothing should be
     written to the binary log, even if the table is non-transactional;
@@ -3741,7 +3839,6 @@ select_create::binlog_show_create_table(TABLE **tables, uint count)
     schema that will do a close_thread_tables(), destroying the
     statement transaction cache.
   */
-  DBUG_ASSERT(thd->current_stmt_binlog_row_based);
   DBUG_ASSERT(tables && *tables && count > 0);
 
   char buf[2048];
@@ -3759,7 +3856,6 @@ select_create::binlog_show_create_table(TABLE **tables, uint count)
 
   if (mysql_bin_log.is_open())
   {
-    int errcode= query_error_code(thd, thd->killed == THD::NOT_KILLED);
     result= thd->binlog_query(THD::STMT_QUERY_TYPE,
                               query.ptr(), query.length(),
                               /* is_trans */ TRUE,

sql/sql_lex.h

@@ -1817,6 +1817,23 @@ typedef struct st_lex : public Query_tables_list
   */
   bool protect_against_global_read_lock;
 
+  /*
+    The following three variables are used in 'CREATE TABLE IF NOT EXISTS ...
+    SELECT' statement. They are used to binlog the statement.
+
+    create_select_start_with_brace will be set if there is a '(' before
+    the first SELECT clause
+
+    create_select_pos records the relative position of the SELECT clause
+    in the whole statement.
+
+    create_select_in_comment will be set if SELECT keyword is in conditional
+    comment.
+   */
+  bool create_select_start_with_brace;
+  uint create_select_pos;
+  bool create_select_in_comment;
+
   st_lex();
 
   virtual ~st_lex()

sql/sql_parse.cc

@@ -2717,6 +2717,25 @@ mysql_execute_command(THD *thd)
         {
           TABLE_LIST *duplicate;
           create_table= lex->unlink_first_table(&link_to_local);
+
+          if (create_table->view)
+          {
+            if (create_info.options & HA_LEX_CREATE_IF_NOT_EXISTS)
+            {
+              push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
+                                  ER_TABLE_EXISTS_ERROR,
+                                  ER(ER_TABLE_EXISTS_ERROR),
+                                  create_info.alias);
+              my_ok(thd);
+            }
+            else
+            {
+              my_error(ER_TABLE_EXISTS_ERROR, MYF(0), create_info.alias);
+              res= 1;
+            }
+            goto end_with_restore_list;
+          }
+
           if ((duplicate= unique_table(thd, create_table, select_tables, 0)))
           {
             update_non_unique_table_error(create_table, "CREATE", duplicate);

sql/sql_yacc.yy

@@ -3881,17 +3881,26 @@ create2a:
           create3 {}
         |  opt_partitioning
            create_select ')'
-           { Select->set_braces(1);}
+           {
+             Select->set_braces(1);
+             Lex->create_select_start_with_brace= TRUE;
+           }
            union_opt {}
         ;
 
 create3:
           /* empty */ {}
         | opt_duplicate opt_as create_select
-          { Select->set_braces(0);}
+          {
+            Select->set_braces(0);
+            Lex->create_select_start_with_brace= FALSE;
+          }
           union_clause {}
         | opt_duplicate opt_as '(' create_select ')'
-          { Select->set_braces(1);}
+          {
+            Select->set_braces(1);
+            Lex->create_select_start_with_brace= TRUE;
+          }
           union_opt {}
         ;
 
@@ -4516,6 +4525,19 @@ create_select:
             lex->current_select->table_list.save_and_clear(&lex->save_list);
             mysql_init_select(lex);
             lex->current_select->parsing_place= SELECT_LIST;
+
+            if (lex->sql_command == SQLCOM_CREATE_TABLE &&
+                (lex->create_info.options & HA_LEX_CREATE_IF_NOT_EXISTS))
+            {
+              Lex_input_stream *lip= YYLIP;
+
+              if (lex->spcont)
+                lex->create_select_pos= lip->get_tok_start() -
+                  lex->sphead->m_tmp_query;
+              else
+                lex->create_select_pos= lip->get_tok_start() - lip->get_buf();
+              lex->create_select_in_comment= (lip->in_comment == DISCARD_COMMENT);
+            }
           }
           select_options select_item_list
           {