BUG#14548159: NUMEROUS CASES OF INCORRECT IDENTIFIER

QUOTING IN REPLICATION 

Problem: Misquoting or unquoted identifiers may lead to
incorrect statements to be logged to the binary log.

Fix: we use specialized functions to append quoted identifiers in
the statements generated by the server.

mysql-test/r/mysqlbinlog.result

@@ -18,7 +18,7 @@ flush logs;
 /*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
 DELIMITER /*!*/;
 ROLLBACK/*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1000000000/*!*/;
 SET @@session.pseudo_thread_id=999999999/*!*/;
 SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=1, @@session.unique_checks=1, @@session.autocommit=1/*!*/;
@@ -64,7 +64,7 @@ ROLLBACK /* added by mysqlbinlog */;
 /*!40019 SET @@session.max_insert_delayed_threads=0*/;
 /*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
 DELIMITER /*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1000000000/*!*/;
 SET @@session.pseudo_thread_id=999999999/*!*/;
 SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=1, @@session.unique_checks=1, @@session.autocommit=1/*!*/;
@@ -97,7 +97,7 @@ Warning: The option '--position' is deprecated and will be removed in a future r
 /*!40019 SET @@session.max_insert_delayed_threads=0*/;
 /*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
 DELIMITER /*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1000000000/*!*/;
 SET @@session.pseudo_thread_id=999999999/*!*/;
 SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=1, @@session.unique_checks=1, @@session.autocommit=1/*!*/;
@@ -119,7 +119,7 @@ ROLLBACK /* added by mysqlbinlog */;
 /*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
 DELIMITER /*!*/;
 ROLLBACK/*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1000000000/*!*/;
 SET @@session.pseudo_thread_id=999999999/*!*/;
 SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=1, @@session.unique_checks=1, @@session.autocommit=1/*!*/;
@@ -165,7 +165,7 @@ ROLLBACK /* added by mysqlbinlog */;
 /*!40019 SET @@session.max_insert_delayed_threads=0*/;
 /*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
 DELIMITER /*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1000000000/*!*/;
 SET @@session.pseudo_thread_id=999999999/*!*/;
 SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=1, @@session.unique_checks=1, @@session.autocommit=1/*!*/;
@@ -198,7 +198,7 @@ Warning: The option '--position' is deprecated and will be removed in a future r
 /*!40019 SET @@session.max_insert_delayed_threads=0*/;
 /*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
 DELIMITER /*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1000000000/*!*/;
 SET @@session.pseudo_thread_id=999999999/*!*/;
 SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=1, @@session.unique_checks=1, @@session.autocommit=1/*!*/;
@@ -220,7 +220,7 @@ ROLLBACK /* added by mysqlbinlog */;
 /*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
 DELIMITER /*!*/;
 ROLLBACK/*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1108844556/*!*/;
 SET @@session.pseudo_thread_id=999999999/*!*/;
 SET @@session.auto_increment_increment=1, @@session.auto_increment_offset=1/*!*/;
@@ -239,7 +239,7 @@ Warning: The option '--position' is deprecated and will be removed in a future r
 /*!40019 SET @@session.max_insert_delayed_threads=0*/;
 /*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
 DELIMITER /*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1108844556/*!*/;
 SET @@session.pseudo_thread_id=999999999/*!*/;
 SET @@session.auto_increment_increment=1, @@session.auto_increment_offset=1/*!*/;
@@ -299,7 +299,7 @@ ERROR 42000: PROCEDURE test.p1 does not exist
 /*!40019 SET @@session.max_insert_delayed_threads=0*/;
 /*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
 DELIMITER /*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1000000000/*!*/;
 SET @@session.pseudo_thread_id=999999999/*!*/;
 SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=1, @@session.unique_checks=1, @@session.autocommit=1/*!*/;
@@ -349,7 +349,7 @@ flush logs;
 /*!40019 SET @@session.max_insert_delayed_threads=0*/;
 /*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
 DELIMITER /*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1000000000/*!*/;
 SET @@session.pseudo_thread_id=999999999/*!*/;
 SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=1, @@session.unique_checks=1, @@session.autocommit=1/*!*/;
@@ -484,7 +484,7 @@ FLUSH LOGS;
 /*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
 DELIMITER /*!*/;
 ROLLBACK/*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1253783037/*!*/;
 SET @@session.pseudo_thread_id=999999999/*!*/;
 SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=1, @@session.unique_checks=1, @@session.autocommit=1/*!*/;
@@ -581,22 +581,22 @@ SET @@session.lc_time_names=0/*!*/;
 SET @@session.collation_database=DEFAULT/*!*/;
 BEGIN
 /*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1266652094/*!*/;
 SavePoint mixed_cases
 /*!*/;
-use db1/*!*/;
+use `db1`/*!*/;
 SET TIMESTAMP=1266652094/*!*/;
 INSERT INTO db1.t2 VALUES("in savepoint mixed_cases")
 /*!*/;
 SET TIMESTAMP=1266652094/*!*/;
 INSERT INTO db1.t1 VALUES(40)
 /*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1266652094/*!*/;
 ROLLBACK TO mixed_cases
 /*!*/;
-use db1/*!*/;
+use `db1`/*!*/;
 SET TIMESTAMP=1266652094/*!*/;
 INSERT INTO db1.t2 VALUES("after rollback to")
 /*!*/;
@@ -624,7 +624,7 @@ SET @@session.lc_time_names=0/*!*/;
 SET @@session.collation_database=DEFAULT/*!*/;
 BEGIN
 /*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1266652094/*!*/;
 SavePoint mixed_cases
 /*!*/;

mysql-test/r/user_var-binlog.result

@@ -19,7 +19,7 @@ flush logs;
 DELIMITER /*!*/;
 ROLLBACK/*!*/;
 SET @`a b`:=_latin1 0x68656C6C6F COLLATE `latin1_swedish_ci`/*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=10000/*!*/;
 SET @@session.pseudo_thread_id=999999999/*!*/;
 SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=1, @@session.unique_checks=1, @@session.autocommit=1/*!*/;

mysql-test/suite/binlog/r/binlog_base64_flag.result

@@ -35,7 +35,7 @@ DELIMITER /*!*/;
 # at 4
 <#>ROLLBACK/*!*/;
 # at 102
-<#>use test/*!*/;
+<#>use `test`/*!*/;
 SET TIMESTAMP=1196959712/*!*/;
 <#>SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=1, @@session.unique_checks=1, @@session.autocommit=1/*!*/;
 SET @@session.sql_mode=0/*!*/;

mysql-test/suite/binlog/r/binlog_row_mysqlbinlog_verbose.result

 Verbose statements from : write-partial-row.binlog
 select replace(txt,'\r', '') as stmt from raw_binlog_rows where txt like '###%';
 stmt
-### INSERT INTO mysql.ndb_apply_status
+### INSERT INTO `mysql`.`ndb_apply_status`
 ### SET
 ###   @1=1
 ###   @2=25769803786
 ###   @3=''
 ###   @4=0
 ###   @5=0
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=3
 ###   @2=3
 ###   @3=3
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=1
 ###   @2=1
 ###   @3=1
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=2
 ###   @2=2
 ###   @3=2
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=4
 ###   @2=4
 ###   @3=4
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=4
 ###   @3=40
-### DELETE FROM test.ba
+### DELETE FROM `test`.`ba`
 ### WHERE
 ###   @1=2
 drop table raw_binlog_rows;
 Verbose statements from : write-full-row.binlog
 select replace(txt,'\r', '') as stmt from raw_binlog_rows where txt like '###%';
 stmt
-### INSERT INTO mysql.ndb_apply_status
+### INSERT INTO `mysql`.`ndb_apply_status`
 ### SET
 ###   @1=2
 ###   @2=25769803786
 ###   @3=''
 ###   @4=0
 ###   @5=0
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=3
 ###   @2=3
 ###   @3=3
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=1
 ###   @2=1
 ###   @3=1
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=2
 ###   @2=2
 ###   @3=2
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=4
 ###   @2=4
 ###   @3=4
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=4
 ###   @2=4
 ###   @3=40
-### DELETE FROM test.ba
+### DELETE FROM `test`.`ba`
 ### WHERE
 ###   @1=2
 drop table raw_binlog_rows;
 Verbose statements from : update-partial-row.binlog
 select replace(txt,'\r', '') as stmt from raw_binlog_rows where txt like '###%';
 stmt
-### INSERT INTO mysql.ndb_apply_status
+### INSERT INTO `mysql`.`ndb_apply_status`
 ### SET
 ###   @1=3
 ###   @2=25769803786
 ###   @3=''
 ###   @4=0
 ###   @5=0
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=3
 ###   @2=3
 ###   @3=3
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=1
 ###   @2=1
 ###   @3=1
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=2
 ###   @2=2
 ###   @3=2
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=4
 ###   @2=4
 ###   @3=4
-### UPDATE test.ba
+### UPDATE `test`.`ba`
 ### WHERE
 ###   @1=4
 ###   @3=4
 ### SET
 ###   @1=4
 ###   @3=40
-### DELETE FROM test.ba
+### DELETE FROM `test`.`ba`
 ### WHERE
 ###   @1=2
 drop table raw_binlog_rows;
 Verbose statements from : update-full-row.binlog
 select replace(txt,'\r', '') as stmt from raw_binlog_rows where txt like '###%';
 stmt
-### INSERT INTO mysql.ndb_apply_status
+### INSERT INTO `mysql`.`ndb_apply_status`
 ### SET
 ###   @1=4
 ###   @2=25769803786
 ###   @3=''
 ###   @4=0
 ###   @5=0
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=3
 ###   @2=3
 ###   @3=3
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=1
 ###   @2=1
 ###   @3=1
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=2
 ###   @2=2
 ###   @3=2
-### INSERT INTO test.ba
+### INSERT INTO `test`.`ba`
 ### SET
 ###   @1=4
 ###   @2=4
 ###   @3=4
-### UPDATE test.ba
+### UPDATE `test`.`ba`
 ### WHERE
 ###   @1=4
 ###   @2=4
@@ -155,7 +155,7 @@ stmt
 ###   @1=4
 ###   @2=4
 ###   @3=40
-### DELETE FROM test.ba
+### DELETE FROM `test`.`ba`
 ### WHERE
 ###   @1=2
 drop table raw_binlog_rows;

mysql-test/suite/binlog/r/binlog_stm_ctype_ucs.result

@@ -13,7 +13,7 @@ flush logs;
 DELIMITER /*!*/;
 ROLLBACK/*!*/;
 SET @`v`:=_ucs2 0x006100620063 COLLATE `ucs2_general_ci`/*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=10000/*!*/;
 SET @@session.pseudo_thread_id=999999999/*!*/;
 SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=1, @@session.unique_checks=1, @@session.autocommit=1/*!*/;

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

@@ -153,7 +153,7 @@ Warning: The option '--position' is deprecated and will be removed in a future r
 /*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
 DELIMITER /*!*/;
 ROLLBACK/*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1000000000/*!*/;
 SET @@session.pseudo_thread_id=999999999/*!*/;
 SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=1, @@session.unique_checks=1, @@session.autocommit=1/*!*/;
@@ -175,7 +175,7 @@ ROLLBACK /* added by mysqlbinlog */;
 /*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
 DELIMITER /*!*/;
 ROLLBACK/*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1000000000/*!*/;
 SET @@session.pseudo_thread_id=999999999/*!*/;
 SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=1, @@session.unique_checks=1, @@session.autocommit=1/*!*/;
@@ -284,7 +284,7 @@ ROLLBACK /* added by mysqlbinlog */;
 /*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
 DELIMITER /*!*/;
 ROLLBACK/*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1000000000/*!*/;
 SET @@session.pseudo_thread_id=999999999/*!*/;
 SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=1, @@session.unique_checks=1, @@session.autocommit=1/*!*/;
@@ -316,7 +316,7 @@ Warning: The option '--position' is deprecated and will be removed in a future r
 /*!50003 SET @OLD_COMPLETION_TYPE=@@COMPLETION_TYPE,COMPLETION_TYPE=0*/;
 DELIMITER /*!*/;
 ROLLBACK/*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=1000000000/*!*/;
 SET @@session.pseudo_thread_id=999999999/*!*/;
 SET @@session.foreign_key_checks=1, @@session.sql_auto_is_null=1, @@session.unique_checks=1, @@session.autocommit=1/*!*/;

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

@@ -627,7 +627,7 @@ drop database if exists mysqltest1
 SET TIMESTAMP=t/*!*/;
 create database mysqltest1
 /*!*/;
-use mysqltest1/*!*/;
+use `mysqltest1`/*!*/;
 SET TIMESTAMP=t/*!*/;
 create table t1 (a varchar(100))
 /*!*/;
@@ -840,7 +840,7 @@ drop database mysqltest1
 SET TIMESTAMP=t/*!*/;
 drop user "zedjzlcsjhd"@127.0.0.1
 /*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=t/*!*/;
 drop function if exists f1
 /*!*/;
@@ -925,7 +925,7 @@ create database mysqltest
 SET TIMESTAMP=t/*!*/;
 create database mysqltest2
 /*!*/;
-use mysqltest2/*!*/;
+use `mysqltest2`/*!*/;
 SET TIMESTAMP=t/*!*/;
 create table t ( t integer )
 /*!*/;
@@ -943,7 +943,7 @@ insert into t values (1);
 return 0;
 end
 /*!*/;
-use mysqltest/*!*/;
+use `mysqltest`/*!*/;
 SET TIMESTAMP=t/*!*/;
 SELECT `mysqltest2`.`f1`()
 /*!*/;
@@ -953,14 +953,14 @@ drop database mysqltest
 SET TIMESTAMP=t/*!*/;
 drop database mysqltest2
 /*!*/;
-use test/*!*/;
+use `test`/*!*/;
 SET TIMESTAMP=t/*!*/;
 CREATE DEFINER=`root`@`localhost` PROCEDURE `mysqltestbug36570_p1`()
 begin
 select 1;
 end
 /*!*/;
-use mysql/*!*/;
+use `mysql`/*!*/;
 SET TIMESTAMP=t/*!*/;
 CREATE DEFINER=`root`@`localhost` PROCEDURE `test`.` mysqltestbug36570_p2`( a int)
 `label`:

sql/ha_ndbcluster_binlog.cc

@@ -22,6 +22,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"
@@ -1269,6 +1270,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)
   {
@@ -1278,16 +1284,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:

sql/log.cc

@@ -29,6 +29,7 @@
 #include "rpl_filter.h"
 #include "rpl_rli.h"
 
+#include "sql_show.h"
 #include <my_dir.h>
 #include <stdarg.h>
 #include <m_ctype.h>				// For test_if_number
@@ -1708,17 +1709,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, TRUE, errcode);
+  my_free(buffer, MYF(MY_WME));
   DBUG_RETURN(mysql_bin_log.write(&qinfo));
 }
 
@@ -1731,18 +1739,23 @@ static int binlog_savepoint_rollback(handlerton *hton, THD *thd, void *sv)
     non-transactional table. Otherwise, truncate the binlog cache starting
     from the SAVEPOINT command.
   */
-  if (unlikely(trans_has_updated_non_trans_table(thd) || 
+  if (unlikely(trans_has_updated_non_trans_table(thd) ||
                (thd->options & 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, TRUE, errcode);
+    my_free(buffer, MYF(MY_WME));
     DBUG_RETURN(mysql_bin_log.write(&qinfo));
   }
   binlog_trans_log_truncate(thd, *(my_off_t*)sv);

sql/log_event.h

@@ -3987,6 +3987,22 @@ static inline bool copy_event_cache_to_file_and_reinit(IO_CACHE *cache,
 bool slave_execute_deferred_events(THD *thd);
 #endif
 
+#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)
 */

sql/sql_base.cc

@@ -4144,24 +4144,19 @@ retry:
     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, errcode))
-        {
-          my_free(query, MYF(0));
-          goto err;
-        }
-        my_free(query, MYF(0));
-      }
-      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.ptr(), temp_buf.length(),
+                            FALSE, FALSE, errcode))
+        goto err;
+      if(error)
       {
         /*
           As replication is maybe going to be corrupted, we need to warn the

sql/sql_db.cc

@@ -25,6 +25,7 @@
 #include <my_dir.h>
 #include <m_ctype.h>
 #include "log.h"
+#include "log_event.h"
 #ifdef __WIN__
 #include <direct.h>
 #endif
@@ -718,12 +719,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
     {
@@ -889,7 +895,7 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
 {
   long deleted=0;
   int error= 0;
-  char	path[FN_REFLEN+16];
+  char	path[2 * FN_REFLEN + 16];
   MY_DIR *dirp;
   uint length;
   TABLE_LIST* dropped_tables= 0;
@@ -989,11 +995,17 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
   {
     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
@@ -1029,12 +1041,13 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
   else if (mysql_bin_log.is_open())
   {
     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);
 
@@ -1054,10 +1067,10 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
         }
         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++ = ',';
     }
 

sql/sql_insert.cc

@@ -3426,16 +3426,16 @@ int select_create::write_to_binlog(bool is_trans, int errcode)
     if (thd->lex->create_select_in_comment)
       query.append(STRING_WITH_LEN("/*! "));
     if (thd->lex->ignore)
-      query.append(STRING_WITH_LEN("INSERT IGNORE INTO `"));
+      query.append(STRING_WITH_LEN("INSERT IGNORE INTO "));
     else if (thd->lex->duplicates == DUP_REPLACE)
-      query.append(STRING_WITH_LEN("REPLACE INTO `"));
+      query.append(STRING_WITH_LEN("REPLACE INTO "));
     else
-      query.append(STRING_WITH_LEN("INSERT INTO `"));
+      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("` "));
+    append_identifier(thd, &query, create_table->db, db_len);
+    query.append(STRING_WITH_LEN("."));
+    append_identifier(thd, &query, create_info->alias, table_len );
+    query.append(STRING_WITH_LEN(" "));
 
     /*
       The insert items.
@@ -3447,9 +3447,8 @@ int select_create::write_to_binlog(bool is_trans, int errcode)
       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("`"));
+      append_identifier(thd, &query, (*f)->field_name,
+                        strlen((*f)->field_name));
     }
     query.append(STRING_WITH_LEN(") "));
 

sql/sql_load.cc

@@ -25,6 +25,7 @@
 #include "sp_head.h"
 #include "sql_trigger.h"
 
+#include "sql_show.h"
 class READ_INFO {
   File	file;
   uchar	*buffer,			/* Buffer for read text */
@@ -619,23 +620,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, duplicates,
                            ignore, transactional_table);
 
@@ -660,11 +658,7 @@ static bool write_execute_load_query_log_event(THD *thd, sql_exchange* ex,
       if (n++)
         pfields.append(", ");
       if (item->name)
-      {
-        pfields.append("`");
-        pfields.append(item->name);
-        pfields.append("`");
-      }
+        append_identifier(thd, &pfields, item->name, strlen(item->name));
       else
         item->print(&pfields, QT_ORDINARY);
     }
@@ -684,9 +678,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->print(&pfields, QT_ORDINARY);
     }

sql/sql_show.cc

@@ -1006,7 +1006,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)
   {

sql/sql_show.h

@@ -40,5 +40,6 @@ 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);
 
 #endif /* SQL_SHOW_H */

sql/sql_table.cc

@@ -1944,6 +1944,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
   for (table= tables; table; table= table->next_local)
   {
     char *db=table->db;
+    int db_len= table->db_length;
     handlerton *table_type;
     enum legacy_db_type frm_db_type= DB_TYPE_UNKNOWN;
 
@@ -1966,14 +1967,14 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
           built_tmp_query.append("DROP TEMPORARY TABLE IF EXISTS ");
         }
 
-        built_tmp_query.append("`");
         if (thd->db == NULL || strcmp(db,thd->db) != 0)
         {
-          built_tmp_query.append(db);
-          built_tmp_query.append("`.`");
+          append_identifier(thd, &built_tmp_query, db, db_len);
+          built_tmp_query.append(".");
         }
-        built_tmp_query.append(table->table_name);
-        built_tmp_query.append("`,");
+        append_identifier(thd, &built_tmp_query, table->table_name,
+                          strlen(table->table_name));
+        built_tmp_query.append(",");
       }
 
       continue;
@@ -1999,15 +2000,14 @@ int mysql_rm_table_part2(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(",");
     }
 
     if (!drop_temporary)