upmerge to bug#14548159

11 files changed, 309 insertions, 134 deletions

diff --git a/sql/ha_ndbcluster_binlog.cc b/sql/ha_ndbcluster_binlog.cc
index 040b5778dbd..0f037040902 100644
--- a/sql/ha_ndbcluster_binlog.cc
+++ b/sql/ha_ndbcluster_binlog.cc
@@ -25,6 +25,7 @@
 #include "rpl_injector.h"
 #include "rpl_filter.h"
 #include "slave.h"
+#include "log_event.h"
 #include "ha_ndbcluster_binlog.h"
 #include "NdbDictionary.hpp"
 #include "ndb_cluster_connection.hpp"
@@ -1297,6 +1298,11 @@ int ndbcluster_log_schema_op(THD *thd, NDB_SHARE *share,
   }
 
   char tmp_buf2[FN_REFLEN];
+  char quoted_table1[2 + 2 * FN_REFLEN + 1];
+  char quoted_db1[2 + 2 * FN_REFLEN + 1];
+  char quoted_db2[2 + 2 * FN_REFLEN + 1];
+  char quoted_table2[2 + 2 * FN_REFLEN + 1];
+  int id_length= 0;
   const char *type_str;
   switch (type)
   {
@@ -1306,16 +1312,31 @@ int ndbcluster_log_schema_op(THD *thd, NDB_SHARE *share,
       DBUG_RETURN(0);
     /* redo the drop table query as is may contain several tables */
     query= tmp_buf2;
-    query_length= (uint) (strxmov(tmp_buf2, "drop table `",
-                                  table_name, "`", NullS) - tmp_buf2);
+    id_length= my_strmov_quoted_identifier (thd, (char *) quoted_table1,
+                                            table_name, 0);
+    quoted_table1[id_length]= '\0';
+    query_length= (uint) (strxmov(tmp_buf2, "drop table ",
+                                  quoted_table1, NullS) - tmp_buf2);
     type_str= "drop table";
     break;
   case SOT_RENAME_TABLE:
     /* redo the rename table query as is may contain several tables */
     query= tmp_buf2;
-    query_length= (uint) (strxmov(tmp_buf2, "rename table `",
-                                  db, ".", table_name, "` to `",
-                                  new_db, ".", new_table_name, "`", NullS) - tmp_buf2);
+    id_length= my_strmov_quoted_identifier (thd, (char *) quoted_db1,
+                                            db, 0);
+    quoted_db1[id_length]= '\0';
+    id_length= my_strmov_quoted_identifier (thd, (char *) quoted_table1,
+                                            table_name, 0);
+    quoted_table1[id_length]= '\0';
+    id_length= my_strmov_quoted_identifier (thd, (char *) quoted_db2,
+                                            new_db, 0);
+    quoted_db2[id_length]= '\0';
+    id_length= my_strmov_quoted_identifier (thd, (char *) quoted_table2,
+                                            new_table_name, 0);
+    quoted_table2[id_length]= '\0';
+    query_length= (uint) (strxmov(tmp_buf2, "rename table ",
+                                  quoted_db1, ".", quoted_table_1, " to ",
+                                  quoted_db2, ".", quoted_table2, NullS) - tmp_buf2);
     type_str= "rename table";
     break;
   case SOT_CREATE_TABLE:
diff --git a/sql/log.cc b/sql/log.cc
index 65923eddce0..43db481b7cc 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -39,6 +39,7 @@
 #include "rpl_rli.h"
 #include "sql_audit.h"
 
+#include "sql_show.h"
 #include <my_dir.h>
 #include <stdarg.h>
 #include <m_ctype.h>				// For test_if_number
@@ -2016,17 +2017,24 @@ static int binlog_savepoint_set(handlerton *hton, THD *thd, void *sv)
   DBUG_ENTER("binlog_savepoint_set");
 
   binlog_trans_log_savepos(thd, (my_off_t*) sv);
+
+  // buffer to store quoted identifier
+  char* buffer= (char *)my_malloc(sizeof("SAVEPOINT ")+ 1 + NAME_LEN * 2 + 2,
+                                  MYF(0));
+  String log_query(buffer, sizeof(buffer), system_charset_info);
+  log_query.length(0);
+
   /* Write it to the binary log */
 
-  String log_query;
-  if (log_query.append(STRING_WITH_LEN("SAVEPOINT ")) ||
-      log_query.append("`") ||
-      log_query.append(thd->lex->ident.str, thd->lex->ident.length) ||
-      log_query.append("`"))
+  if (log_query.append(STRING_WITH_LEN("SAVEPOINT ")))
     DBUG_RETURN(1);
+  else
+    append_identifier(thd, &log_query, thd->lex->ident.str,
+                      thd->lex->ident.length);
   int errcode= query_error_code(thd, thd->killed == THD::NOT_KILLED);
   Query_log_event qinfo(thd, log_query.c_ptr_safe(), log_query.length(),
                         TRUE, FALSE, TRUE, errcode);
+  my_free(buffer);
   DBUG_RETURN(mysql_bin_log.write(&qinfo));
 }
 
@@ -2042,15 +2050,20 @@ static int binlog_savepoint_rollback(handlerton *hton, THD *thd, void *sv)
   if (unlikely(trans_has_updated_non_trans_table(thd) ||
                (thd->variables.option_bits & OPTION_KEEP_LOG)))
   {
-    String log_query;
-    if (log_query.append(STRING_WITH_LEN("ROLLBACK TO ")) ||
-        log_query.append("`") ||
-        log_query.append(thd->lex->ident.str, thd->lex->ident.length) ||
-        log_query.append("`"))
+    // buffer to store rollback query with quoted identifier
+    char* buffer= (char *)my_malloc(12 + 1 + NAME_LEN * 2 + 2, MYF(0));
+    String log_query(buffer, sizeof(buffer), system_charset_info);
+    log_query.length(0);
+
+    if (log_query.append(STRING_WITH_LEN("ROLLBACK TO ")))
       DBUG_RETURN(1);
+    else
+      append_identifier(thd, &log_query, thd->lex->ident.str,
+                        thd->lex->ident.length);
     int errcode= query_error_code(thd, thd->killed == THD::NOT_KILLED);
     Query_log_event qinfo(thd, log_query.c_ptr_safe(), log_query.length(),
                           TRUE, FALSE, TRUE, errcode);
+    my_free(buffer);
     DBUG_RETURN(mysql_bin_log.write(&qinfo));
   }
   binlog_trans_log_truncate(thd, *(my_off_t*)sv);
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 73c2342b6fa..25a552bced9 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -45,6 +45,7 @@
 #include "rpl_record.h"
 #include "transaction.h"
 #include <my_dir.h>
+#include "sql_show.h"    // append_identifier
 
 #endif /* MYSQL_CLIENT */
 
@@ -1963,6 +1964,10 @@ Rows_log_event::print_verbose_one_row(IO_CACHE *file, table_def *td,
 void Rows_log_event::print_verbose(IO_CACHE *file,
                                    PRINT_EVENT_INFO *print_event_info)
 {
+  // Quoted length of the identifier can be twice the original length
+  char quoted_db[1 + NAME_LEN * 2 + 2];
+  char quoted_table[1 + NAME_LEN * 2 + 2];
+  int quoted_db_len, quoted_table_len;
   Table_map_log_event *map;
   table_def *td;
   const char *sql_command, *sql_clause1, *sql_clause2;
@@ -1999,9 +2004,23 @@ void Rows_log_event::print_verbose(IO_CACHE *file,
   for (const uchar *value= m_rows_buf; value < m_rows_end; )
   {
     size_t length;
+#ifdef MYSQL_SERVER
+    quoted_db_len= my_strmov_quoted_identifier(this->thd, (char *) quoted_db,
+                                        map->get_db_name(), 0);
+    quoted_table_len= my_strmov_quoted_identifier(this->thd,
+                                                  (char *) quoted_table,
+                                                  map->get_table_name(), 0);
+#else
+    quoted_db_len= my_strmov_quoted_identifier((char *) quoted_db,
+                                               map->get_db_name());
+    quoted_table_len= my_strmov_quoted_identifier((char *) quoted_table,
+                                          map->get_table_name());
+#endif
+    quoted_db[quoted_db_len]= '\0';
+    quoted_table[quoted_table_len]= '\0';
     my_b_printf(file, "### %s %s.%s\n",
                       sql_command,
-                      map->get_db_name(), map->get_table_name());
+                      quoted_db, quoted_table);
     /* Print the first image */
     if (!(length= print_verbose_one_row(file, td, print_event_info,
                                   &m_cols, value,
@@ -2160,24 +2179,20 @@ Log_event::continue_group(Relay_log_info *rli)
 void Query_log_event::pack_info(Protocol *protocol)
 {
   // TODO: show the catalog ??
-  char *buf, *pos;
-  if (!(buf= (char*) my_malloc(9 + db_len + q_len, MYF(MY_WME))))
-    return;
-  pos= buf;
+  String temp_buf;
+  // Add use `DB` to the string if required
   if (!(flags & LOG_EVENT_SUPPRESS_USE_F)
       && db && db_len)
   {
-    pos= strmov(buf, "use `");
-    memcpy(pos, db, db_len);
-    pos= strmov(pos+db_len, "`; ");
+    temp_buf.append("use ");
+    append_identifier(this->thd, &temp_buf, db, db_len);
+    temp_buf.append("; ");
   }
+  // Add the query to the string
   if (query && q_len)
-  {
-    memcpy(pos, query, q_len);
-    pos+= q_len;
-  }
-  protocol->store(buf, pos-buf, &my_charset_bin);
-  my_free(buf);
+    temp_buf.append(query);
+ // persist the buffer in protocol
+  protocol->store(temp_buf.ptr(), temp_buf.length(), &my_charset_bin);
 }
 #endif
 
@@ -3006,6 +3021,8 @@ void Query_log_event::print_query_header(IO_CACHE* file,
 {
   // TODO: print the catalog ??
   char buff[40],*end;				// Enough for SET TIMESTAMP
+  char quoted_id[1+ 2*FN_REFLEN+ 2];
+  int quoted_len= 0;
   bool different_db= 1;
   uint32 tmp;
 
@@ -3024,11 +3041,17 @@ void Query_log_event::print_query_header(IO_CACHE* file,
   }
   else if (db)
   {
+#ifdef MYSQL_SERVER
+    quoted_len= my_strmov_quoted_identifier(this->thd, (char*)quoted_id, db, 0);
+#else
+    quoted_len= my_strmov_quoted_identifier((char*)quoted_id, db);
+#endif
+    quoted_id[quoted_len]= '\0';
     different_db= memcmp(print_event_info->db, db, db_len + 1);
     if (different_db)
       memcpy(print_event_info->db, db, db_len + 1);
     if (db[0] && different_db) 
-      my_b_printf(file, "use %s%s\n", db, print_event_info->delimiter);
+      my_b_printf(file, "use %s%s\n", quoted_id, print_event_info->delimiter);
   }
 
   end=int10_to_str((long) when, strmov(buff,"SET TIMESTAMP="),10);
@@ -4318,7 +4341,8 @@ void Format_description_log_event::calc_server_version_split()
 uint Load_log_event::get_query_buffer_length()
 {
   return
-    5 + db_len + 3 +                        // "use DB; "
+    //the DB name may double if we escape the quote character
+    5 + 2*db_len + 3 +
     18 + fname_len + 2 +                    // "LOAD DATA INFILE 'file''"
     11 +                                    // "CONCURRENT "
     7 +					    // LOCAL
@@ -4337,13 +4361,21 @@ uint Load_log_event::get_query_buffer_length()
 void Load_log_event::print_query(bool need_db, const char *cs, char *buf,
                                  char **end, char **fn_start, char **fn_end)
 {
+  char quoted_id[1 + NAME_LEN * 2 + 2];//quoted  length 
+  int  quoted_id_len= 0;
   char *pos= buf;
 
   if (need_db && db && db_len)
   {
-    pos= strmov(pos, "use `");
-    memcpy(pos, db, db_len);
-    pos= strmov(pos+db_len, "`; ");
+    pos= strmov(pos, "use ");
+#ifdef MYSQL_SERVER
+    quoted_id_len= my_strmov_quoted_identifier(this->thd, (char *) quoted_id,
+                                               db, 0);
+#else
+    quoted_id_len= my_strmov_quoted_identifier((char *) quoted_id, db);
+#endif
+    pos+= quoted_id_len;
+    pos= strmov(pos, "; ");
   }
 
   pos= strmov(pos, "LOAD DATA ");
@@ -4370,17 +4402,15 @@ void Load_log_event::print_query(bool need_db, const char *cs, char *buf,
   if (fn_end)
     *fn_end= pos;
 
-  pos= strmov(pos ," TABLE `");
+  pos= strmov(pos ," TABLE ");
   memcpy(pos, table_name, table_name_len);
   pos+= table_name_len;
 
   if (cs != NULL)
   {
-    pos= strmov(pos ,"` CHARACTER SET ");
+    pos= strmov(pos ," CHARACTER SET ");
     pos= strmov(pos ,  cs);
   }
-  else
-    pos= strmov(pos, "`");
 
   /* We have to create all optional fields as the default is not empty */
   pos= strmov(pos, " FIELDS TERMINATED BY ");
@@ -4420,9 +4450,9 @@ void Load_log_event::print_query(bool need_db, const char *cs, char *buf,
         *pos++= ' ';
         *pos++= ',';
       }
-      memcpy(pos, field, field_lens[i]);
-      pos+=   field_lens[i];
-      field+= field_lens[i]  + 1;
+      quoted_id_len= my_strmov_quoted_identifier(this->thd, quoted_id, field,
+                                                 0);
+      memcpy(pos, quoted_id, quoted_id_len);
     }
     *pos++= ')';
   }
@@ -4670,6 +4700,8 @@ void Load_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info)
 void Load_log_event::print(FILE* file_arg, PRINT_EVENT_INFO* print_event_info,
 			   bool commented)
 {
+  size_t id_len= 0;
+  char temp_buf[1 + 2*FN_REFLEN + 2];
   Write_on_release_cache cache(&print_event_info->head_cache, file_arg);
 
   DBUG_ENTER("Load_log_event::print");
@@ -4695,10 +4727,16 @@ void Load_log_event::print(FILE* file_arg, PRINT_EVENT_INFO* print_event_info,
   }
   
   if (db && db[0] && different_db)
-    my_b_printf(&cache, "%suse %s%s\n", 
-            commented ? "# " : "",
-            db, print_event_info->delimiter);
-
+  {
+#ifdef MYSQL_SERVER
+    id_len= my_strmov_quoted_identifier(this->thd, temp_buf, db, 0);
+#else
+    id_len= my_strmov_quoted_identifier(temp_buf, db);
+#endif
+    temp_buf[id_len]= '\0';
+    my_b_printf(&cache, "%suse %s%s\n",
+            commented ? "# " : "", temp_buf, print_event_info->delimiter);
+  }
   if (flags & LOG_EVENT_THREAD_SPECIFIC_F)
     my_b_printf(&cache,"%sSET @@session.pseudo_thread_id=%lu%s\n",
             commented ? "# " : "", (ulong)thread_id,
@@ -4713,8 +4751,14 @@ void Load_log_event::print(FILE* file_arg, PRINT_EVENT_INFO* print_event_info,
     my_b_printf(&cache,"REPLACE ");
   else if (sql_ex.opt_flags & IGNORE_FLAG)
     my_b_printf(&cache,"IGNORE ");
-  
-  my_b_printf(&cache, "INTO TABLE `%s`", table_name);
+
+#ifdef MYSQL_SERVER
+    id_len= my_strmov_quoted_identifier(this->thd, temp_buf, table_name, 0);
+#else
+    id_len= my_strmov_quoted_identifier(temp_buf, table_name);
+#endif
+  temp_buf[id_len]= '\0';
+  my_b_printf(&cache, "INTO TABLE %s", temp_buf);
   my_b_printf(&cache, " FIELDS TERMINATED BY ");
   pretty_print_str(&cache, sql_ex.field_term, sql_ex.field_term_len);
 
@@ -4747,7 +4791,9 @@ void Load_log_event::print(FILE* file_arg, PRINT_EVENT_INFO* print_event_info,
     {
       if (i)
         my_b_printf(&cache, ",");
-      my_b_printf(&cache, "%s", field);
+      id_len= my_strmov_quoted_identifier((char *) temp_buf, field);
+      temp_buf[id_len]= '\0';
+      my_b_printf(&cache, "%s", temp_buf);
 
       field += field_lens[i]  + 1;
     }
@@ -5684,7 +5730,10 @@ Xid_log_event::do_shall_skip(Relay_log_info *rli)
 void User_var_log_event::pack_info(Protocol* protocol)
 {
   char *buf= 0;
-  uint val_offset= 4 + name_len;
+  char quoted_id[1 + FN_REFLEN * 2 + 2];// quoted identifier
+  int id_len= my_strmov_quoted_identifier(this->thd, quoted_id, name, 0);
+  quoted_id[id_len]= '\0';
+  uint val_offset= 2 + id_len;
   uint event_len= val_offset;
 
   if (is_null)
@@ -5753,10 +5802,8 @@ void User_var_log_event::pack_info(Protocol* protocol)
     }
   }
   buf[0]= '@';
-  buf[1]= '`';
-  memcpy(buf+2, name, name_len);
-  buf[2+name_len]= '`';
-  buf[3+name_len]= '=';
+  memcpy(buf + 1, quoted_id, id_len);
+  buf[1 + id_len]= '=';
   protocol->store(buf, event_len, &my_charset_bin);
   my_free(buf);
 }
@@ -5893,6 +5940,8 @@ bool User_var_log_event::write(IO_CACHE* file)
 #ifdef MYSQL_CLIENT
 void User_var_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info)
 {
+  char quoted_id[1 + NAME_LEN * 2 + 2];// quoted length of the identifier
+  int quoted_len= 0;
   Write_on_release_cache cache(&print_event_info->head_cache, file,
                                Write_on_release_cache::FLUSH_F);
 
@@ -5902,9 +5951,11 @@ void User_var_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info)
     my_b_printf(&cache, "\tUser_var\n");
   }
 
-  my_b_printf(&cache, "SET @`");
-  my_b_write(&cache, (uchar*) name, (uint) (name_len));
-  my_b_printf(&cache, "`");
+  my_b_printf(&cache, "SET @");
+  quoted_len= my_strmov_quoted_identifier((char *) quoted_id,
+                                          (const char *) name);
+  quoted_id[quoted_len]= '\0';
+  my_b_write(&cache, (uchar*) quoted_id, (uint) quoted_len);
 
   if (is_null)
   {
@@ -7212,14 +7263,23 @@ void Execute_load_query_log_event::print(FILE* file,
 void Execute_load_query_log_event::pack_info(Protocol *protocol)
 {
   char *buf, *pos;
-  if (!(buf= (char*) my_malloc(9 + db_len + q_len + 10 + 21, MYF(MY_WME))))
+  if (!(buf= (char*) my_malloc(9 + (db_len * 2) + 2 + q_len + 10 + 21,
+                               MYF(MY_WME))))
     return;
   pos= buf;
   if (db && db_len)
   {
-    pos= strmov(buf, "use `");
-    memcpy(pos, db, db_len);
-    pos= strmov(pos+db_len, "`; ");
+    /*
+      Statically allocates room to store '\0' and an identifier
+      that may have NAME_LEN * 2 due to quoting and there are
+      two quoting characters that wrap them.
+    */
+    char quoted_db[1 + NAME_LEN * 2 + 2];// quoted length of the identifier
+    size_t size= 0;
+    size= my_strmov_quoted_identifier(this->thd, quoted_db, db, 0);
+    pos= strmov(buf, "use ");
+    memcpy(pos, quoted_db, size);
+    pos= strmov(pos + size, "; ");
   }
   if (query && q_len)
   {
@@ -10186,3 +10246,62 @@ Heartbeat_log_event::Heartbeat_log_event(const char* buf, uint event_len,
   log_ident= buf + header_size;
 }
 #endif
+
+#ifdef MYSQL_SERVER
+/*
+  This is a utility function that adds a quoted identifier into the a buffer.
+  This also escapes any existance of the quote string inside the identifier.
+
+  SYNOPSIS
+    my_strmov_quoted_identifier
+    thd                   thread handler
+    buffer                target buffer
+    identifier            the identifier to be quoted
+    length                length of the identifier
+*/
+size_t my_strmov_quoted_identifier(THD* thd, char *buffer,
+                                   const char* identifier,
+                                   uint length)
+{
+  int q= thd ? get_quote_char_for_identifier(thd, identifier, length) : '`';
+  return my_strmov_quoted_identifier_helper(q, buffer, identifier, length);
+}
+#else
+size_t my_strmov_quoted_identifier(char *buffer,  const char* identifier)
+{
+  int q= '`';
+  return my_strmov_quoted_identifier_helper(q, buffer, identifier, 0);
+}
+
+#endif
+
+size_t my_strmov_quoted_identifier_helper(int q, char *buffer,
+                                          const char* identifier,
+                                          uint length)
+{
+  size_t written= 0;
+  char quote_char;
+  uint id_length= (length) ? length : strlen(identifier);
+
+  if (q == EOF)
+  {
+    (void *) strncpy(buffer, identifier, id_length);
+    return id_length;
+  }
+  quote_char= (char) q;
+  *buffer++= quote_char;
+  written++;
+  while (id_length--)
+  {
+    if (*identifier == quote_char)
+    {
+      *buffer++= quote_char;
+      written++;
+    }
+    *buffer++= *identifier++;
+    written++;
+  }
+  *buffer++= quote_char;
+  return ++written;
+}
+
diff --git a/sql/log_event.h b/sql/log_event.h
index badaf2eaa79..ac7aa7e9453 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -4107,6 +4107,23 @@ bool slave_execute_deferred_events(THD *thd);
 
 int append_query_string(THD *thd, CHARSET_INFO *csinfo,
                         String const *from, String *to);
+
+#ifdef MYSQL_SERVER
+/*
+  This is an utility function that adds a quoted identifier into the a buffer.
+  This also escapes any existance of the quote string inside the identifier.
+ */
+size_t my_strmov_quoted_identifier(THD *thd, char *buffer,
+                                   const char* identifier,
+                                   uint length);
+#else
+size_t my_strmov_quoted_identifier(char *buffer, const char* identifier);
+#endif
+size_t my_strmov_quoted_identifier_helper(int q, char *buffer,
+                                          const char* identifier,
+                                          uint length);
+
+
 /**
   @} (end of group Replication)
 */
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index a7d04f12341..7b1893eaf43 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -3831,24 +3831,19 @@ static bool open_table_entry_fini(THD *thd, TABLE_SHARE *share, TABLE *entry)
     entry->file->implicit_emptied= 0;
     if (mysql_bin_log.is_open())
     {
-      char *query, *end;
-      uint query_buf_size= 20 + share->db.length + share->table_name.length +1;
-      if ((query= (char*) my_malloc(query_buf_size,MYF(MY_WME))))
-      {
-        /* this DELETE FROM is needed even with row-based binlogging */
-        end = strxmov(strmov(query, "DELETE FROM `"),
-                      share->db.str,"`.`",share->table_name.str,"`", NullS);
-        int errcode= query_error_code(thd, TRUE);
-        if (thd->binlog_query(THD::STMT_QUERY_TYPE,
-                              query, (ulong)(end-query),
-                              FALSE, FALSE, FALSE, errcode))
-        {
-          my_free(query);
-          return TRUE;
-        }
-        my_free(query);
-      }
-      else
+      bool error= false;
+      String temp_buf;
+      error= temp_buf.append("DELETE FROM ");
+      append_identifier(thd, &temp_buf, share->db.str, strlen(share->db.str));
+      error= temp_buf.append(".");
+      append_identifier(thd, &temp_buf, share->table_name.str,
+                        strlen(share->table_name.str));
+      int errcode= query_error_code(thd, TRUE);
+      if (thd->binlog_query(THD::STMT_QUERY_TYPE,
+                            temp_buf.c_ptr_safe(), temp_buf.length(),
+                            FALSE, FALSE, FALSE, errcode))
+        return TRUE;
+      if (error)
       {
         /*
           As replication is maybe going to be corrupted, we need to warn the
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index 99cb8cf4b3a..c0a0a4d5522 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -37,6 +37,7 @@
 #include <my_dir.h>
 #include <m_ctype.h>
 #include "log.h"
+#include "log_event.h"
 #ifdef __WIN__
 #include <direct.h>
 #endif
@@ -619,12 +620,17 @@ not_silent:
   {
     char *query;
     uint query_length;
+    char db_name_quoted[2 * FN_REFLEN + sizeof("create database ") + 2];
+    int id_len= 0;
 
     if (!thd->query())                          // Only in replication
     {
-      query= 	     tmp_query;
-      query_length= (uint) (strxmov(tmp_query,"create database `",
-                                    db, "`", NullS) - tmp_query);
+      id_len= my_strmov_quoted_identifier(thd, (char *) db_name_quoted, db,
+                                          0);
+      db_name_quoted[id_len]= '\0';
+      query= tmp_query;
+      query_length= (uint) (strxmov(tmp_query,"create database ",
+                                    db_name_quoted, NullS) - tmp_query);
     }
     else
     {
@@ -761,7 +767,7 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
 {
   ulong deleted_tables= 0;
   bool error= true;
-  char	path[FN_REFLEN+16];
+  char	path[2 * FN_REFLEN + 16];
   MY_DIR *dirp;
   uint length;
   bool found_other_files= false;
@@ -883,11 +889,16 @@ update_binlog:
   {
     const char *query;
     ulong query_length;
+    // quoted db name + wraping quote
+    char buffer_temp [2 * FN_REFLEN + 2];
+    int id_len= 0;
     if (!thd->query())
     {
       /* The client used the old obsolete mysql_drop_db() call */
       query= path;
-      query_length= (uint) (strxmov(path, "drop database `", db, "`",
+      id_len= my_strmov_quoted_identifier(thd, buffer_temp, db, strlen(db));
+      buffer_temp[id_len] ='\0';
+      query_length= (uint) (strxmov(path, "DROP DATABASE ", buffer_temp, "",
                                      NullS) - path);
     }
     else
@@ -925,12 +936,13 @@ update_binlog:
   else if (mysql_bin_log.is_open() && !silent)
   {
     char *query, *query_pos, *query_end, *query_data_start;
+    char temp_identifier[ 2 * FN_REFLEN + 2];
     TABLE_LIST *tbl;
-    uint db_len;
+    uint db_len, id_length=0;
 
     if (!(query= (char*) thd->alloc(MAX_DROP_TABLE_Q_LEN)))
       goto exit; /* not much else we can do */
-    query_pos= query_data_start= strmov(query,"drop table ");
+    query_pos= query_data_start= strmov(query,"DROP TABLE ");
     query_end= query + MAX_DROP_TABLE_Q_LEN;
     db_len= strlen(db);
 
@@ -963,10 +975,10 @@ update_binlog:
         }
         query_pos= query_data_start;
       }
-
-      *query_pos++ = '`';
-      query_pos= strmov(query_pos,tbl->table_name);
-      *query_pos++ = '`';
+      id_length= my_strmov_quoted_identifier(thd, (char *)temp_identifier,
+                                      tbl->table_name, 0);
+      temp_identifier[id_length]= '\0';
+      query_pos= strmov(query_pos,(char *)&temp_identifier);
       *query_pos++ = ',';
     }
 
diff --git a/sql/sql_load.cc b/sql/sql_load.cc
index 1819102aa34..611c32696ea 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -38,7 +38,7 @@
 #include "sql_repl.h"
 #include "sp_head.h"
 #include "sql_trigger.h"
-
+#include "sql_show.h"
 class XML_TAG {
 public:
   int level;
@@ -678,23 +678,20 @@ static bool write_execute_load_query_log_event(THD *thd, sql_exchange* ex,
   const char          *tbl= table_name_arg;
   const char          *tdb= (thd->db != NULL ? thd->db : db_arg);
   String              string_buf;
-
-  if (!thd->db || strcmp(db_arg, thd->db)) 
+  if (!thd->db || strcmp(db_arg, thd->db))
   {
     /*
-      If used database differs from table's database, 
-      prefix table name with database name so that it 
+      If used database differs from table's database,
+      prefix table name with database name so that it
       becomes a FQ name.
      */
     string_buf.set_charset(system_charset_info);
-    string_buf.append(db_arg);
-    string_buf.append("`");
+    append_identifier(thd, &string_buf, db_arg, strlen(db_arg));
     string_buf.append(".");
-    string_buf.append("`");
-    string_buf.append(table_name_arg);
-    tbl= string_buf.c_ptr_safe();
   }
-
+  append_identifier(thd, &string_buf, table_name_arg,
+                    strlen(table_name_arg));
+  tbl= string_buf.c_ptr_safe();
   Load_log_event       lle(thd, ex, tdb, tbl, fv, is_concurrent,
                            duplicates, ignore, transactional_table);
 
@@ -719,11 +716,7 @@ static bool write_execute_load_query_log_event(THD *thd, sql_exchange* ex,
       if (n++)
         pfields.append(", ");
       if (item->type() == Item::FIELD_ITEM)
-      {
-        pfields.append("`");
-        pfields.append(item->name);
-        pfields.append("`");
-      }
+        append_identifier(thd, &pfields, item->name, strlen(item->name));
       else
         item->print(&pfields, QT_ORDINARY);
     }
@@ -743,9 +736,7 @@ static bool write_execute_load_query_log_event(THD *thd, sql_exchange* ex,
       val= lv++;
       if (n++)
         pfields.append(", ");
-      pfields.append("`");
-      pfields.append(item->name);
-      pfields.append("`");
+      append_identifier(thd, &pfields, item->name, strlen(item->name));
       pfields.append(val->name);
     }
   }
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 6c940661763..81e18ccd6c3 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -952,7 +952,8 @@ append_identifier(THD *thd, String *packet, const char *name, uint length)
 {
   const char *name_end;
   char quote_char;
-  int q= get_quote_char_for_identifier(thd, name, length);
+  int q;
+  q= thd ? get_quote_char_for_identifier(thd, name, length) : '`';
 
   if (q == EOF)
   {
diff --git a/sql/sql_show.h b/sql/sql_show.h
index 8ad9327c08c..406c75d8cbe 100644
--- a/sql/sql_show.h
+++ b/sql/sql_show.h
@@ -89,6 +89,7 @@ int store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
 int view_store_create_info(THD *thd, TABLE_LIST *table, String *buff);
 
 int copy_event_to_schema_table(THD *thd, TABLE *sch_table, TABLE *event_table);
+int get_quote_char_for_identifier(THD *thd, const char *name, uint length);
 
 void append_identifier(THD *thd, String *packet, const char *name,
 		       uint length);
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 7d70fa8afd2..b19b265a311 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -2053,6 +2053,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
   {
     bool is_trans;
     char *db=table->db;
+    int db_len= table->db_length;
     handlerton *table_type;
     enum legacy_db_type frm_db_type= DB_TYPE_UNKNOWN;
 
@@ -2114,14 +2115,14 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
           Don't write the database name if it is the current one (or if
           thd->db is NULL).
         */
-        built_ptr_query->append("`");
         if (thd->db == NULL || strcmp(db,thd->db) != 0)
         {
-          built_ptr_query->append(db);
-          built_ptr_query->append("`.`");
+          append_identifier(thd, built_ptr_query, db, db_len);
+          built_ptr_query->append(".");
         }
-        built_ptr_query->append(table->table_name);
-        built_ptr_query->append("`,");
+        append_identifier(thd, built_ptr_query, table->table_name,
+                          strlen(table->table_name));
+        built_ptr_query->append(",");
       }
       /*
         This means that a temporary table was droped and as such there
@@ -2180,15 +2181,15 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
           Don't write the database name if it is the current one (or if
           thd->db is NULL).
         */
-        built_query.append("`");
         if (thd->db == NULL || strcmp(db,thd->db) != 0)
         {
-          built_query.append(db);
-          built_query.append("`.`");
+          append_identifier(thd, &built_query, db, db_len);
+          built_query.append(".");
         }
 
-        built_query.append(table->table_name);
-        built_query.append("`,");
+        append_identifier(thd, &built_query, table->table_name,
+                          strlen(table->table_name));
+        built_query.append(",");
       }
     }
     DEBUG_SYNC(thd, "rm_table_no_locks_before_delete_table");
diff --git a/sql/sql_truncate.cc b/sql/sql_truncate.cc
index 67ed608f114..2285131fade 100644
--- a/sql/sql_truncate.cc
+++ b/sql/sql_truncate.cc
@@ -24,6 +24,7 @@
 #include "sql_acl.h"     // DROP_ACL
 #include "sql_parse.h"   // check_one_table_access()
 #include "sql_truncate.h"
+#include "sql_show.h"    //append_identifier()
 
 
 /**
@@ -43,9 +44,8 @@ static bool fk_info_append_fields(String *str, List<LEX_STRING> *fields)
 
   while ((field= it++))
   {
-    res|= str->append("`");
-    res|= str->append(field);
-    res|= str->append("`, ");
+    append_identifier(NULL, str, field->str, field->length);
+    res|= str->append(", ");
   }
 
   str->chop();
@@ -76,19 +76,23 @@ static const char *fk_info_str(THD *thd, FOREIGN_KEY_INFO *fk_info)
     `db`.`tbl`, CONSTRAINT `id` FOREIGN KEY (`fk`) REFERENCES `db`.`tbl` (`fk`)
   */
 
-  res|= str.append('`');
-  res|= str.append(fk_info->foreign_db);
-  res|= str.append("`.`");
-  res|= str.append(fk_info->foreign_table);
-  res|= str.append("`, CONSTRAINT `");
-  res|= str.append(fk_info->foreign_id);
-  res|= str.append("` FOREIGN KEY (");
+  append_identifier(NULL, &str, fk_info->foreign_db->str,
+                    fk_info->foreign_db->length);
+  res|= str.append(".");
+  append_identifier(NULL, &str, fk_info->foreign_table->str,
+                    fk_info->foreign_table->length);
+  res|= str.append(", CONSTRAINT ");
+  append_identifier(NULL, &str, fk_info->foreign_id->str,
+                    fk_info->foreign_id->length);
+  res|= str.append(" FOREIGN KEY (");
   res|= fk_info_append_fields(&str, &fk_info->foreign_fields);
-  res|= str.append(") REFERENCES `");
-  res|= str.append(fk_info->referenced_db);
-  res|= str.append("`.`");
-  res|= str.append(fk_info->referenced_table);
-  res|= str.append("` (");
+  res|= str.append(") REFERENCES ");
+  append_identifier(NULL, &str, fk_info->referenced_db->str,
+                    fk_info->referenced_db->length);
+  res|= str.append(".");
+  append_identifier(NULL, &str, fk_info->referenced_table->str,
+                    fk_info->referenced_table->length);
+  res|= str.append(" (");
   res|= fk_info_append_fields(&str, &fk_info->referenced_fields);
   res|= str.append(')');