BUG#48297: Schema name is ignored when LOAD DATA is written into

binlog, replication aborts

In SBR or MBR, the schema name is not being written to the binlog
when executing a LOAD DATA statement. This becomes a problem when
the current database (lets call it db1) is different from the
table's schema (lets call it db2). For instance, take the
following statements:
  
  use db1;
  load data local infile 'infile.txt' into table db2.t

Should this statement be logged without t's schema (db2), when
replaying it, one can get db1.t populated instead of db2.t (if
db1.t exists). On the other hand, if there is no db1.t at all,
replication will stop.

We fix this by always logging the table (in load file) with fully
qualified name when its schema is different from the current
database or when no default database was selected.

3 files changed, 115 insertions, 5 deletions

diff --git a/mysql-test/extra/rpl_tests/rpl_loaddata.test b/mysql-test/extra/rpl_tests/rpl_loaddata.test
index 26916642cae..7db12600456 100644
--- a/mysql-test/extra/rpl_tests/rpl_loaddata.test
+++ b/mysql-test/extra/rpl_tests/rpl_loaddata.test
@@ -158,4 +158,65 @@ LOAD DATA INFILE "../../std_data/words.dat" INTO TABLE t1;
 
 DROP TABLE IF EXISTS t1;
 
+# BUG#48297: Schema name is ignored when LOAD DATA is written into binlog,
+# replication aborts
+-- source include/master-slave-reset.inc
+
+-- let $db1= b48297_db1
+-- let $db2= b42897_db2
+
+-- connection master
+
+-- disable_warnings
+-- eval drop database if exists $db1
+-- eval drop database if exists $db2
+-- enable_warnings
+
+-- eval create database $db1
+-- eval create database $db2
+
+-- eval use $db1
+-- eval CREATE TABLE t1 (c1 VARCHAR(256)) engine=$engine_type;
+
+-- eval use $db2
+
+-- echo ### assertion: works with cross-referenced database
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- eval LOAD DATA LOCAL INFILE '$MYSQLTEST_VARDIR/std_data/loaddata5.dat' INTO TABLE $db1.t1
+
+-- eval use $db1
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- echo ### assertion: works with fully qualified name on current database
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- eval LOAD DATA LOCAL INFILE '$MYSQLTEST_VARDIR/std_data/loaddata5.dat' INTO TABLE $db1.t1
+
+-- echo ### assertion: works without fully qualified name on current database
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- eval LOAD DATA LOCAL INFILE '$MYSQLTEST_VARDIR/std_data/loaddata5.dat' INTO TABLE t1
+
+-- echo ### create connection without default database
+-- echo ### connect (conn2,localhost,root,,*NO-ONE*);
+connect (conn2,localhost,root,,*NO-ONE*);
+-- connection conn2
+-- echo ### assertion: works without stating the default database
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- eval LOAD DATA LOCAL INFILE '$MYSQLTEST_VARDIR/std_data/loaddata5.dat' INTO TABLE $db1.t1
+-- echo ### disconnect and switch back to master connection
+-- disconnect conn2
+-- connection master
+
+-- sync_slave_with_master
+-- eval use $db1
+
+let $diff_table_1=master:$db1.t1;
+let $diff_table_2=slave:$db1.t1;
+source include/diff_tables.inc;
+
+-- connection master
+
+-- eval DROP DATABASE $db1
+-- eval DROP DATABASE $db2
+
+-- sync_slave_with_master
+
 # End of 4.1 tests
diff --git a/mysql-test/suite/rpl/r/rpl_loaddata.result b/mysql-test/suite/rpl/r/rpl_loaddata.result
index 0653936f0ec..ca9c14691b0 100644
--- a/mysql-test/suite/rpl/r/rpl_loaddata.result
+++ b/mysql-test/suite/rpl/r/rpl_loaddata.result
@@ -86,3 +86,32 @@ CREATE TABLE t1 (word CHAR(20) NOT NULL PRIMARY KEY) ENGINE=INNODB;
 LOAD DATA INFILE "../../std_data/words.dat" INTO TABLE t1;
 ERROR 23000: Duplicate entry 'Aarhus' for key 'PRIMARY'
 DROP TABLE IF EXISTS t1;
+stop slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+reset master;
+reset slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+start slave;
+drop database if exists b48297_db1;
+drop database if exists b42897_db2;
+create database b48297_db1;
+create database b42897_db2;
+use b48297_db1;
+CREATE TABLE t1 (c1 VARCHAR(256)) engine=MyISAM;;
+use b42897_db2;
+### assertion: works with cross-referenced database
+LOAD DATA LOCAL INFILE 'MYSQLTEST_VARDIR/std_data/loaddata5.dat' INTO TABLE b48297_db1.t1;
+use b48297_db1;
+### assertion: works with fully qualified name on current database
+LOAD DATA LOCAL INFILE 'MYSQLTEST_VARDIR/std_data/loaddata5.dat' INTO TABLE b48297_db1.t1;
+### assertion: works without fully qualified name on current database
+LOAD DATA LOCAL INFILE 'MYSQLTEST_VARDIR/std_data/loaddata5.dat' INTO TABLE t1;
+### create connection without default database
+### connect (conn2,localhost,root,,*NO-ONE*);
+### assertion: works without stating the default database
+LOAD DATA LOCAL INFILE 'MYSQLTEST_VARDIR/std_data/loaddata5.dat' INTO TABLE b48297_db1.t1;
+### disconnect and switch back to master connection
+use b48297_db1;
+Comparing tables master:b48297_db1.t1 and slave:b48297_db1.t1
+DROP DATABASE b48297_db1;
+DROP DATABASE b42897_db2;
diff --git a/sql/sql_load.cc b/sql/sql_load.cc
index 59ba7b16d62..5500439b619 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -84,7 +84,7 @@ static int read_sep_field(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
 			  bool ignore_check_option_errors);
 #ifndef EMBEDDED_LIBRARY
 static bool write_execute_load_query_log_event(THD *thd, sql_exchange* ex,
-                                               const char* db_arg,
+                                               const char* db_arg, /* table's database */
                                                const char* table_name_arg,
                                                enum enum_duplicates duplicates,
                                                bool ignore,
@@ -501,7 +501,8 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
           
 	  if (thd->transaction.stmt.modified_non_trans_table)
             write_execute_load_query_log_event(thd, ex,
-                                               tdb, table_list->table_name,
+                                               table_list->db, 
+                                               table_list->table_name,
                                                handle_duplicates, ignore,
                                                transactional_table,
                                                errcode);
@@ -548,7 +549,7 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
       {
         int errcode= query_error_code(thd, killed_status == THD::NOT_KILLED);
         write_execute_load_query_log_event(thd, ex,
-                                           tdb, table_list->table_name,
+                                           table_list->db, table_list->table_name,
                                            handle_duplicates, ignore,
                                            transactional_table,
                                            errcode);
@@ -573,7 +574,7 @@ err:
 
 /* Not a very useful function; just to avoid duplication of code */
 static bool write_execute_load_query_log_event(THD *thd, sql_exchange* ex,
-                                               const char* db_arg,
+                                               const char* db_arg,  /* table's database */
                                                const char* table_name_arg,
                                                enum enum_duplicates duplicates,
                                                bool ignore,
@@ -590,8 +591,27 @@ static bool write_execute_load_query_log_event(THD *thd, sql_exchange* ex,
   Item                *item, *val;
   String               pfield, pfields;
   int                  n;
+  const char          *tbl= table_name_arg;
+  const char          *tdb= (thd->db != NULL ? thd->db : db_arg);
+  String              string_buf;
 
-  Load_log_event       lle(thd, ex, db_arg, table_name_arg, fv, duplicates,
+  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 
+      becomes a FQ name.
+     */
+    string_buf.set_charset(system_charset_info);
+    string_buf.append(db_arg);
+    string_buf.append("`");
+    string_buf.append(".");
+    string_buf.append("`");
+    string_buf.append(table_name_arg);
+    tbl= string_buf.c_ptr_safe();
+  }
+
+  Load_log_event       lle(thd, ex, tdb, tbl, fv, duplicates,
                            ignore, transactional_table);
 
   /*