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.

5 files changed, 176 insertions, 18 deletions

diff --git a/sql/sql_class.h b/sql/sql_class.h
index 1627b6ec02d..42c873e9fc3 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -2622,7 +2622,9 @@ public:
 
 
 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 @@ public:
     {}
   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();
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index eb9e1e5b3af..8604f876f37 100644
--- a/sql/sql_insert.cc
+++ b/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,
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 7403bb5a1a4..9131cec9d04 100644
--- a/sql/sql_lex.h
+++ b/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()
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 9ec03ea1d5f..fbe9c9753d9 100644
--- a/sql/sql_parse.cc
+++ b/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);
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 5a3ad0b3eba..ed367582ba5 100644
--- a/sql/sql_yacc.yy
+++ b/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
           {