WL#874 "Extended LOAD DATA".

Now one can use user variables as target for data loaded from file
(besides table's columns). Also LOAD DATA got new SET-clause in which
one can specify values for table columns as expressions.

For example the following is possible:
LOAD DATA INFILE 'words.dat' INTO TABLE t1 (a, @b) SET c = @b + 1;

This patch also implements new way of replicating LOAD DATA.
Now we do it similarly to other queries.
We store LOAD DATA query in new Execute_load_query event
(which is last in the sequence of events representing LOAD DATA).
When we are executing this event we simply rewrite part of query which
holds name of file (we use name of temporary file) and then execute it
as usual query. In the beggining of this sequence we use Begin_load_query
event which is almost identical to Append_file event


client/mysqlbinlog.cc:
  Added support of two new binary log events Begin_load_query_log_event and
  Execute_load_query_log_Event which are used to replicate LOAD DATA INFILE.
mysql-test/r/ctype_ucs.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results and made it more robust for future similar 
  changes.
mysql-test/r/insert_select.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results and made it more robust for future similar 
  changes.
mysql-test/r/loaddata.result:
  Added tests for new LOAD DATA features.
mysql-test/r/mix_innodb_myisam_binlog.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results (don't dare to get rid from binlog positions
  completely since it seems that this test uses them).
mysql-test/r/mysqlbinlog.result:
  New approach for binlogging of LOAD DATA statement. Now we store it as
  usual query and rewrite part in which file is specified when needed.
  So now mysqlbinlog output for LOAD DATA much more closer to its initial
  form. Updated test'd results accordingly.
mysql-test/r/mysqldump.result:
  Made test more robust to other tests failures.
mysql-test/r/rpl000015.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results accordingly.
mysql-test/r/rpl_change_master.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results.
mysql-test/r/rpl_charset.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results accordingly
mysql-test/r/rpl_deadlock.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results accordingly
mysql-test/r/rpl_error_ignored_table.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results and made it more robust for future similar 
  changes.
mysql-test/r/rpl_flush_log_loop.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results accordingly.
mysql-test/r/rpl_flush_tables.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results and made it more robust for future similar 
  changes.
mysql-test/r/rpl_loaddata.result:
  New way of replicating LOAD DATA. Now we do it similarly to other
  queries. We store LOAD DATA query in new Execute_load_query event
  (which is last in the sequence of events representing LOAD DATA).
  When we are executing this event we simply rewrite part of query which
  holds name of file (we use name of temporary file) and then execute it
  as usual query. In the beggining of this sequence we use Begin_load_query
  event which is almost identical to Append_file event...
  
  Updated test's results wwith new binlog positions.
mysql-test/r/rpl_loaddata_rule_m.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results and made it more robust for future similar 
  changes.
  Since now LOAD DATA is replicated much in the same way as usual query
  --binlog_do/ignore_db work for it inthe same way as for usual queries.
mysql-test/r/rpl_loaddata_rule_s.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results accordingly.
mysql-test/r/rpl_loaddatalocal.result:
  Added nice test for case when it is important that LOAD DATA LOCAL
  ignores duplicates.
mysql-test/r/rpl_log.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results accordingly (don't dare to get rid from binlog 
  positions completely since it seems that this test uses them).
mysql-test/r/rpl_log_pos.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results accordingly.
mysql-test/r/rpl_max_relay_size.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results accordingly.
mysql-test/r/rpl_multi_query.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results accordingly.
mysql-test/r/rpl_relayrotate.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results accordingly.
mysql-test/r/rpl_replicate_do.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results accordingly.
mysql-test/r/rpl_reset_slave.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results accordingly.
mysql-test/r/rpl_rotate_logs.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results accordingly.
mysql-test/r/rpl_server_id1.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results accordingly.
mysql-test/r/rpl_server_id2.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results accordingly.
mysql-test/r/rpl_temporary.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results and made it more robust for future similar 
  changes.
mysql-test/r/rpl_timezone.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results and made it more robust for future similar 
  changes.
mysql-test/r/rpl_until.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results accordingly and tweaked test a bit to bring it
  back to good shape.
mysql-test/r/rpl_user_variables.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results and made it more robust for future similar 
  changes.
mysql-test/r/user_var.result:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test's results and made it more robust for future similar 
  changes.
mysql-test/t/ctype_ucs.test:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test accordingly and made it more robust for future similar
  changes.
mysql-test/t/insert_select.test:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test accordingly and made it more robust for future similar
  changes.
mysql-test/t/loaddata.test:
  Added test cases for new LOAD DATA functionality.
mysql-test/t/mix_innodb_myisam_binlog.test:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test accordingly.
mysql-test/t/mysqlbinlog.test:
  New way of replicating LOAD DATA local. Now we do it similarly to other
  queries. We store LOAD DATA query in new Execute_load_query event
  (which is last in the sequence of events representing LOAD DATA).
  When we are executing this event we simply rewrite part of query which
  holds name of file (we use name of temporary file) and then execute it
  as usual query. In the beggining of this sequence we use Begin_load_query
  event which is almost identical to Append_file event...
  
  Thus we need new binlog positions for LOAD DATA events.
mysql-test/t/mysqlbinlog2.test:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test accordingly.
mysql-test/t/mysqldump.test:
  Made test more robust for failures of other tests.
mysql-test/t/rpl_charset.test:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test accordingly.
mysql-test/t/rpl_deadlock.test:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test accordingly.
mysql-test/t/rpl_error_ignored_table.test:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test accordingly and made it more robust for future similar
  changes.
mysql-test/t/rpl_flush_tables.test:
  Addition of two new types of binary log events shifted binlog positions.
  Made test more robust for future similar changes.
mysql-test/t/rpl_loaddata.test:
  New way of replicating LOAD DATA. Now we do it similarly to other
  queries. We store LOAD DATA query in new Execute_load_query event
  (which is last in the sequence of events representing LOAD DATA).
  When we are executing this event we simply rewrite part of query which
  holds name of file (we use name of temporary file) and then execute it
  as usual query. In the beggining of this sequence we use Begin_load_query
  event which is almost identical to Append_file event...
  
  Apropritely updated comments in test.
mysql-test/t/rpl_loaddata_rule_m.test:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test accordingly and made it more robust for future similar 
  changes.
  Since now LOAD DATA is replicated much in the same way as usual query
  --binlog_do/ignore_db work for it inthe same way as for usual queries.
mysql-test/t/rpl_loaddata_rule_s.test:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test accordingly.
mysql-test/t/rpl_loaddatalocal.test:
  Added nice test for case when it is important that LOAD DATA LOCAL
  ignores duplicates.
mysql-test/t/rpl_log.test:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test accordingly (don't dare to get rid from binlog positions
  completely since it seems that this test uses them).
mysql-test/t/rpl_log_pos.test:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test accordingly.
mysql-test/t/rpl_multi_query.test:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test accordingly.
mysql-test/t/rpl_temporary.test:
  Addition of two new types of binary log events shifted binlog positions.
  Made test more robust for future similar changes.
mysql-test/t/rpl_timezone.test:
  Addition of two new types of binary log events shifted binlog positions.
  Made test more robust for future similar changes.
mysql-test/t/rpl_until.test:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test accordingly and tweaked it a bit to bring it back to good
  shape.
mysql-test/t/rpl_user_variables.test:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test accordingly and made it more robust for future similar
  changes.
mysql-test/t/user_var.test:
  Addition of two new types of binary log events shifted binlog positions.
  Updated test accordingly and made it more robust for future similar
  changes.
sql/item_func.cc:
  Added Item_user_var_as_out_param class that represents user variable
  which used as out parameter in LOAD DATA.
  
  Moved code from Item_func_set_user_var::update_hash() function to
  separate static function to be able to reuse it in this new class.
sql/item_func.h:
  Added Item_user_var_as_out_param class that represents user variable
  which used as out parameter in LOAD DATA.
sql/log_event.cc:
  New way of replicating LOAD DATA. Now we do it similarly to other
  queries. We store LOAD DATA query in new Execute_load_query event
  (which is last in the sequence of events representing LOAD DATA).
  When we are executing this event we simply rewrite part of query which
  holds name of file (we use name of temporary file) and then execute it
  as usual query. In the beggining of this sequence we use Begin_load_query
  event which is almost identical to Append_file event.
sql/log_event.h:
  New way of replicating LOAD DATA. Now we do it similarly to other
  queries. We store LOAD DATA query in new Execute_load_query event
  (which is last in the sequence of events representing LOAD DATA).
  When we are executing this event we simply rewrite part of query which
  holds name of file (we use name of temporary file) and then execute it
  as usual query. In the beggining of this sequence we use Begin_load_query
  event which is almost identical to Append_file event.
sql/mysql_priv.h:
  Now mysql_load() has two more arguments. They are needed to pass list of
  columns and corresponding expressions from new LOAD DATA's SET clause.
sql/share/errmsg.txt:
  Added new error message which is used to forbid loading of data from
  fixed length rows to variables.
sql/sql_lex.h:
  Added LEX::fname_start/fname_end members. 
  They are pointers to part of LOAD DATA statement which should be
  rewritten during replication (file name + little extra).
sql/sql_load.cc:
  Added support for extended LOAD DATA. 
  Now one can use user variables as target for data loaded from file 
  (besides table's columns). Also LOAD DATA got new SET-clause in which
  one can specify values for table columns as expressions.
  
  Updated mysql_load()/read_fixed_length()/read_sep_field() to support
  this functionality (now they can read data from file to both columns and
  variables and assign do calculations and assignments specified in SET
  clause).
  
  We also use new approach for LOAD DATA binlogging/replication.
sql/sql_parse.cc:
  mysql_execute_command():
    Since now we have SET clause in LOAD DATA we should also check
    permissions for tables used in its expressions. Also mysql_load()
    has two more arguments to pass information about this clause.
sql/sql_repl.cc:
  New way of replicating LOAD DATA. Now we do it similarly to other
  queries. We store LOAD DATA query in new Execute_load_query event
  (which is last in the sequence of events representing LOAD DATA).
  When we are executing this event we simply rewrite part of query which
  holds name of file (we use name of temporary file) and then execute it
  as usual query. In the beggining of this sequence we use Begin_load_query
  event which is almost identical to Append_file event.
sql/sql_repl.h:
  struct st_load_file_info:
    Removed memebers which are no longer needed for LOAD DATA binnlogging.
sql/sql_yacc.yy:
  Added support for extended LOAD DATA syntax. Now one can use
  user variables as target for data loaded from file (besides table's 
  columns). Also LOAD DATA got new SET-clause in which one can specify
  values for table columns as expressions.
  
  For example the following is possible:
  LOAD DATA INFILE 'words.dat' INTO TABLE t1 (a, @b) SET c = @b + 1;
  
  Also now we save pointers to the beginning and to the end of part of 
  LOAD DATA statement which should be rewritten during replication.

12 files changed, 868 insertions, 212 deletions

diff --git a/sql/item_func.cc b/sql/item_func.cc
index 680f3608d0d..0a993b6f3d8 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -3294,12 +3294,28 @@ Item_func_set_user_var::fix_length_and_dec()
 }
 
 
-bool Item_func_set_user_var::update_hash(void *ptr, uint length,
-					 Item_result type,
-					 CHARSET_INFO *cs,
-					 Derivation dv)
+/*
+  Set value to user variable.
+
+  SYNOPSYS
+    update_hash()
+    entry    - pointer to structure representing variable
+    set_null - should we set NULL value ?
+    ptr      - pointer to buffer with new value
+    length   - length of new value
+    type     - type of new value
+    cs       - charset info for new value
+    dv       - derivation for new value
+
+  RETURN VALUE
+    False - success, True - failure
+*/
+
+static bool
+update_hash(user_var_entry *entry, bool set_null, void *ptr, uint length,
+            Item_result type, CHARSET_INFO *cs, Derivation dv)
 {
-  if ((null_value=args[0]->null_value))
+  if (set_null)
   {
     char *pos= (char*) entry+ ALIGN_SIZE(sizeof(user_var_entry));
     if (entry->value && entry->value != pos)
@@ -3332,7 +3348,7 @@ bool Item_func_set_user_var::update_hash(void *ptr, uint length,
 	  entry->value=0;
 	if (!(entry->value=(char*) my_realloc(entry->value, length,
 					      MYF(MY_ALLOW_ZERO_PTR))))
-	  goto err;
+	  return 1;
       }
     }
     if (type == STRING_RESULT)
@@ -3348,11 +3364,21 @@ bool Item_func_set_user_var::update_hash(void *ptr, uint length,
     entry->collation.set(cs, dv);
   }
   return 0;
+}
 
- err:
-  current_thd->fatal_error();			// Probably end of memory
-  null_value= 1;
-  return 1;
+
+bool
+Item_func_set_user_var::update_hash(void *ptr, uint length, Item_result type,
+                                    CHARSET_INFO *cs, Derivation dv)
+{
+  if (::update_hash(entry, (null_value= args[0]->null_value),
+                    ptr, length, type, cs, dv))
+  {
+    current_thd->fatal_error();     // Probably end of memory
+    null_value= 1;
+    return 1;
+  }
+  return 0;
 }
 
 
@@ -3878,6 +3904,77 @@ bool Item_func_get_user_var::eq(const Item *item, bool binary_cmp) const
 }
 
 
+bool Item_user_var_as_out_param::fix_fields(THD *thd, TABLE_LIST *tables,
+                                            Item **ref)
+{
+  DBUG_ASSERT(fixed == 0);
+  if (Item::fix_fields(thd, tables, ref) ||
+      !(entry= get_variable(&thd->user_vars, name, 1)))
+    return TRUE;
+  entry->type= STRING_RESULT;
+  /*
+    Let us set the same collation which is used for loading
+    of fields in LOAD DATA INFILE.
+    (Since Item_user_var_as_out_param is used only there).
+  */
+  entry->collation.set(thd->variables.collation_database);
+  entry->update_query_id= thd->query_id;
+  return FALSE;
+}
+
+
+void Item_user_var_as_out_param::set_null_value(CHARSET_INFO* cs)
+{
+  if (::update_hash(entry, TRUE, 0, 0, STRING_RESULT, cs,
+                    DERIVATION_IMPLICIT))
+    current_thd->fatal_error();			// Probably end of memory
+}
+
+
+void Item_user_var_as_out_param::set_value(const char *str, uint length,
+                                           CHARSET_INFO* cs)
+{
+  if (::update_hash(entry, FALSE, (void*)str, length, STRING_RESULT, cs,
+                    DERIVATION_IMPLICIT))
+    current_thd->fatal_error();			// Probably end of memory
+}
+
+
+double Item_user_var_as_out_param::val_real()
+{
+  DBUG_ASSERT(0);
+  return 0.0;
+}
+
+
+longlong Item_user_var_as_out_param::val_int()
+{
+  DBUG_ASSERT(0);
+  return 0;
+}
+
+
+String* Item_user_var_as_out_param::val_str(String *str)
+{
+  DBUG_ASSERT(0);
+  return 0;
+}
+
+
+my_decimal* Item_user_var_as_out_param::val_decimal(my_decimal *decimal_buffer)
+{
+  DBUG_ASSERT(0);
+  return 0;
+}
+
+
+void Item_user_var_as_out_param::print(String *str)
+{
+  str->append('@');
+  str->append(name.str,name.length);
+}
+
+
 longlong Item_func_inet_aton::val_int()
 {
   DBUG_ASSERT(fixed == 1);
diff --git a/sql/item_func.h b/sql/item_func.h
index eb9aa4ca0c7..93633e75619 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -1131,6 +1131,35 @@ public:
 };
 
 
+/*
+  This item represents user variable used as out parameter (e.g in LOAD DATA),
+  and it is supposed to be used only for this purprose. So it is simplified
+  a lot. Actually you should never obtain its value.
+
+  The only two reasons for this thing being an Item is possibility to store it
+  in List<Item> and desire to place this code somewhere near other functions
+  working with user variables.
+*/
+class Item_user_var_as_out_param :public Item
+{
+  LEX_STRING name;
+  user_var_entry *entry;
+public:
+  Item_user_var_as_out_param(LEX_STRING a) : name(a) {}
+  /* We should return something different from FIELD_ITEM here */
+  enum Type type() const { return STRING_ITEM;}
+  double val_real();
+  longlong val_int();
+  String *val_str(String *str);
+  my_decimal *val_decimal(my_decimal *decimal_buffer);
+  /* fix_fields() binds variable name with its entry structure */
+  bool fix_fields(THD *thd, struct st_table_list *tables, Item **ref);
+  void print(String *str);
+  void set_null_value(CHARSET_INFO* cs);
+  void set_value(const char *str, uint length, CHARSET_INFO* cs);
+};
+
+
 class Item_func_inet_aton : public Item_int_func
 {
 public:
diff --git a/sql/log_event.cc b/sql/log_event.cc
index ff7445029d0..a943181e999 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -272,6 +272,8 @@ const char* Log_event::get_type_str()
   case XID_EVENT: return "Xid";
   case USER_VAR_EVENT: return "User var";
   case FORMAT_DESCRIPTION_EVENT: return "Format_desc";
+  case BEGIN_LOAD_QUERY_EVENT: return "Begin_load_query";
+  case EXECUTE_LOAD_QUERY_EVENT: return "Execute_load_query";
   default: return "Unknown";				/* impossible */
   }
 }
@@ -783,10 +785,10 @@ Log_event* Log_event::read_log_event(const char* buf, uint event_len,
 
   switch(buf[EVENT_TYPE_OFFSET]) {
   case QUERY_EVENT:
-    ev  = new Query_log_event(buf, event_len, description_event);
+    ev  = new Query_log_event(buf, event_len, description_event, QUERY_EVENT);
     break;
   case LOAD_EVENT:
-    ev = new Create_file_log_event(buf, event_len, description_event);
+    ev = new Load_log_event(buf, event_len, description_event);
     break;
   case NEW_LOAD_EVENT:
     ev = new Load_log_event(buf, event_len, description_event);
@@ -832,6 +834,12 @@ Log_event* Log_event::read_log_event(const char* buf, uint event_len,
   case FORMAT_DESCRIPTION_EVENT:
     ev = new Format_description_log_event(buf, event_len, description_event); 
     break;
+  case BEGIN_LOAD_QUERY_EVENT:
+    ev = new Begin_load_query_log_event(buf, event_len, description_event);
+    break;
+  case EXECUTE_LOAD_QUERY_EVENT:
+    ev = new Execute_load_query_log_event(buf, event_len, description_event);
+    break;
   default:
     DBUG_PRINT("error",("Unknown evernt code: %d",(int) buf[EVENT_TYPE_OFFSET]));
     ev= NULL;
@@ -1085,10 +1093,13 @@ bool Query_log_event::write(IO_CACHE* file)
     Calculate length of whole event
     The "1" below is the \0 in the db's length
   */
-  event_length= (uint) (start-buf) + db_len + 1 + q_len;
+  event_length= (uint) (start-buf) + get_post_header_size_for_derived() + db_len + 1 + q_len;
 
   return (write_header(file, event_length) ||
-          my_b_safe_write(file, (byte*) buf, (uint) (start-buf)) ||
+          my_b_safe_write(file, (byte*) buf, QUERY_HEADER_LEN) ||
+          write_post_header_for_derived(file) ||
+          my_b_safe_write(file, (byte*) start_of_status,
+                          (uint) (start-start_of_status)) ||
           my_b_safe_write(file, (db) ? (byte*) db : (byte*)"", db_len + 1) ||
           my_b_safe_write(file, (byte*) query, q_len)) ? 1 : 0;
 }
@@ -1150,7 +1161,8 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg,
 */
 
 Query_log_event::Query_log_event(const char* buf, uint event_len,
-                                 const Format_description_log_event *description_event)
+                                 const Format_description_log_event *description_event,
+                                 Log_event_type event_type)
   :Log_event(buf, description_event), data_buf(0), query(NullS), catalog(NullS), 
    db(NullS), catalog_len(0), status_vars_len(0),
    flags2_inited(0), sql_mode_inited(0), charset_inited(0),
@@ -1163,7 +1175,7 @@ Query_log_event::Query_log_event(const char* buf, uint event_len,
   DBUG_ENTER("Query_log_event::Query_log_event(char*,...)");
 
   common_header_len= description_event->common_header_len;
-  post_header_len= description_event->post_header_len[QUERY_EVENT-1]; 
+  post_header_len= description_event->post_header_len[event_type-1];
   DBUG_PRINT("info",("event_len=%ld, common_header_len=%d, post_header_len=%d",
                      event_len, common_header_len, post_header_len));
   
@@ -1196,13 +1208,12 @@ Query_log_event::Query_log_event(const char* buf, uint event_len,
                         (uint) status_vars_len));
     tmp-= 2;
   }
-  /* we have parsed everything we know in the post header */
-#ifndef DBUG_OFF
-  if (tmp) /* this is probably a master newer than us */
-    DBUG_PRINT("info", ("Query_log_event has longer post header than we know\
-  (%d more bytes)", tmp));
-#endif
-  
+  /*
+    We have parsed everything we know in the post header for QUERY_EVENT,
+    the rest of post header is either comes from older version MySQL or
+    dedicated to derived events (e.g. Execute_load_query...)
+  */
+
   /* variable-part: the status vars; only in MySQL 5.0  */
   
   start= (char*) (buf+post_header_len);
@@ -1281,8 +1292,8 @@ Query_log_event::Query_log_event(const char* buf, uint event_len,
 */
 
 #ifdef MYSQL_CLIENT
-void Query_log_event::print(FILE* file, bool short_form,
-                            LAST_EVENT_INFO* last_event_info)
+void Query_log_event::print_query_header(FILE* file, bool short_form,
+                                         LAST_EVENT_INFO* last_event_info)
 {
   // TODO: print the catalog ??
   char buff[40],*end;				// Enough for SET TIMESTAMP
@@ -1292,8 +1303,8 @@ void Query_log_event::print(FILE* file, bool short_form,
   if (!short_form)
   {
     print_header(file);
-    fprintf(file, "\tQuery\tthread_id=%lu\texec_time=%lu\terror_code=%d\n",
-	    (ulong) thread_id, (ulong) exec_time, error_code);
+    fprintf(file, "\t%s\tthread_id=%lu\texec_time=%lu\terror_code=%d\n",
+	    get_type_str(), (ulong) thread_id, (ulong) exec_time, error_code);
   }
 
   if (!(flags & LOG_EVENT_SUPPRESS_USE_F) && db)
@@ -1399,7 +1410,13 @@ void Query_log_event::print(FILE* file, bool short_form,
       memcpy(last_event_info->charset, charset, 6);
     }
   }
+}
+
 
+void Query_log_event::print(FILE* file, bool short_form,
+                            LAST_EVENT_INFO* last_event_info)
+{
+  print_query_header(file, short_form, last_event_info);
   my_fwrite(file, (byte*) query, q_len, MYF(MY_NABP | MY_WME));
   fputs(";\n", file);
 }
@@ -1413,6 +1430,12 @@ void Query_log_event::print(FILE* file, bool short_form,
 #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
 int Query_log_event::exec_event(struct st_relay_log_info* rli)
 {
+  return exec_event(rli, query, q_len);
+}
+
+
+int Query_log_event::exec_event(struct st_relay_log_info* rli, const char *query_arg, uint32 q_len_arg)
+{
   int expected_error,actual_error= 0;
   /*
     Colleagues: please never free(thd->catalog) in MySQL. This would lead to
@@ -1444,8 +1467,8 @@ int Query_log_event::exec_event(struct st_relay_log_info* rli)
   if (db_ok(thd->db, replicate_do_db, replicate_ignore_db))
   {
     thd->set_time((time_t)when);
-    thd->query_length= q_len;
-    thd->query = (char*)query;
+    thd->query_length= q_len_arg;
+    thd->query= (char*)query_arg;
     VOID(pthread_mutex_lock(&LOCK_thread_count));
     thd->query_id = next_query_id();
     VOID(pthread_mutex_unlock(&LOCK_thread_count));
@@ -1506,7 +1529,7 @@ int Query_log_event::exec_event(struct st_relay_log_info* rli)
       }
 
       /* Execute the query (note that we bypass dispatch_command()) */
-      mysql_parse(thd, thd->query, q_len);
+      mysql_parse(thd, thd->query, thd->query_length);
 
     }
     else
@@ -1518,7 +1541,7 @@ int Query_log_event::exec_event(struct st_relay_log_info* rli)
         we exit gracefully; otherwise we warn about the bad error and tell DBA
         to check/fix it.
       */
-      if (mysql_test_parse_for_slave(thd, thd->query, q_len))
+      if (mysql_test_parse_for_slave(thd, thd->query, thd->query_length))
         clear_all_errors(thd, rli);        /* Can ignore query */
       else
       {
@@ -1560,7 +1583,7 @@ Default database: '%s'. Query: '%s'",
 			expected_error,
 			actual_error ? thd->net.last_error: "no error",
 			actual_error,
-			print_slave_db_safe(db), query);
+			print_slave_db_safe(db), query_arg);
       thd->query_error= 1;
     }
     /*
@@ -1581,7 +1604,7 @@ Default database: '%s'. Query: '%s'",
 			"Error '%s' on query. Default database: '%s'. Query: '%s'",
 			(actual_error ? thd->net.last_error :
 			 "unexpected success or fatal error"),
-			print_slave_db_safe(thd->db), query);
+			print_slave_db_safe(thd->db), query_arg);
       thd->query_error= 1;
     }
 
@@ -1828,11 +1851,6 @@ int Start_log_event_v3::exec_event(struct st_relay_log_info* rli)
     server starts or when FLUSH LOGS), or to create artificial events to parse
     binlogs from MySQL 3.23 or 4.x.
     When in a client, only the 2nd use is possible.
-
-  TODO
-    Update this code with the new event for LOAD DATA, once they are pushed (in
-    4.1 or 5.0). If it's in 5.0, only the "case 4" block should be updated.
-
 */
 
 Format_description_log_event::
@@ -1866,6 +1884,8 @@ Format_description_log_event(uint8 binlog_ver, const char* server_ver)
       post_header_len[DELETE_FILE_EVENT-1]= DELETE_FILE_HEADER_LEN;
       post_header_len[NEW_LOAD_EVENT-1]= post_header_len[LOAD_EVENT-1];
       post_header_len[FORMAT_DESCRIPTION_EVENT-1]= FORMAT_DESCRIPTION_HEADER_LEN;
+      post_header_len[BEGIN_LOAD_QUERY_EVENT-1]= post_header_len[APPEND_BLOCK_EVENT-1];
+      post_header_len[EXECUTE_LOAD_QUERY_EVENT-1]= EXECUTE_LOAD_QUERY_HEADER_LEN;
     }
     break;
 
@@ -2071,12 +2091,9 @@ int Format_description_log_event::exec_event(struct st_relay_log_info* rli)
 */
 
 #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
-void Load_log_event::pack_info(Protocol *protocol)
+uint Load_log_event::get_query_buffer_length()
 {
-  char *buf, *pos;
-  uint buf_len;
-
-  buf_len=
+  return
     5 + db_len + 3 +                        // "use DB; "
     18 + fname_len + 2 +                    // "LOAD DATA INFILE 'file''"
     7 +					    // LOCAL
@@ -2089,11 +2106,15 @@ void Load_log_event::pack_info(Protocol *protocol)
     19 + sql_ex.line_start_len*4 + 2 +      // " LINES STARTING BY 'str'"
     15 + 22 +                               // " IGNORE xxx  LINES"
     3 + (num_fields-1)*2 + field_block_len; // " (field1, field2, ...)"
+}
 
-  if (!(buf= my_malloc(buf_len, MYF(MY_WME))))
-    return;
-  pos= buf;
-  if (db && db_len)
+
+void Load_log_event::print_query(bool need_db, char *buf,
+                                 char **end, char **fn_start, char **fn_end)
+{
+  char *pos= buf;
+
+  if (need_db && db && db_len)
   {
     pos= strmov(pos, "use `");
     memcpy(pos, db, db_len);
@@ -2101,6 +2122,10 @@ void Load_log_event::pack_info(Protocol *protocol)
   }
 
   pos= strmov(pos, "LOAD DATA ");
+
+  if (fn_start)
+    *fn_start= pos;
+
   if (check_fname_outside_temp_buf())
     pos= strmov(pos, "LOCAL ");
   pos= strmov(pos, "INFILE '");
@@ -2112,7 +2137,12 @@ void Load_log_event::pack_info(Protocol *protocol)
   else if (sql_ex.opt_flags & IGNORE_FLAG)
     pos= strmov(pos, " IGNORE ");
 
-  pos= strmov(pos ,"INTO TABLE `");
+  pos= strmov(pos ,"INTO");
+
+  if (fn_end)
+    *fn_end= pos;
+
+  pos= strmov(pos ," TABLE `");
   memcpy(pos, table_name, table_name_len);
   pos+= table_name_len;
 
@@ -2161,7 +2191,18 @@ void Load_log_event::pack_info(Protocol *protocol)
     *pos++= ')';
   }
 
-  protocol->store(buf, pos-buf, &my_charset_bin);
+  *end= pos;
+}
+
+
+void Load_log_event::pack_info(Protocol *protocol)
+{
+  char *buf, *end;
+
+  if (!(buf= my_malloc(get_query_buffer_length(), MYF(MY_WME))))
+    return;
+  print_query(TRUE, buf, &end, 0, 0);
+  protocol->store(buf, end-buf, &my_charset_bin);
   my_free(buf, MYF(0));
 }
 #endif /* defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) */
@@ -2368,11 +2409,6 @@ int Load_log_event::copy_log_event(const char *buf, ulong event_len,
   fname_len = strlen(fname);
   // null termination is accomplished by the caller doing buf[event_len]=0
 
-  /*
-    In 5.0 this event will have the same format, as we are planning to log LOAD
-    DATA INFILE in a completely different way (as a plain-text query) since 4.1
-    or 5.0 (Dmitri's WL#874)
-  */
   DBUG_RETURN(0);
 }
 
@@ -2532,7 +2568,6 @@ void Load_log_event::set_fields(const char* affected_db,
 int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli, 
 			       bool use_rli_only_for_errors)
 {
-  char *load_data_query= 0;
   thd->db_length= db_len;
   thd->db= (char*) rewrite_db(db, &thd->db_length);
   DBUG_ASSERT(thd->query == 0);
@@ -2594,21 +2629,30 @@ int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli,
     else
     {
       char llbuff[22];
+      char *end;
       enum enum_duplicates handle_dup;
       bool ignore= 0;
+      char *load_data_query;
+
       /*
-        Make a simplified LOAD DATA INFILE query, for the information of the
-        user in SHOW PROCESSLIST. Note that db is known in the 'db' column.
+        Forge LOAD DATA INFILE query which will be used in SHOW PROCESS LIST
+        and written to slave's binlog if binlogging is on.
       */
-      if ((load_data_query= (char *) my_alloca(18 + strlen(fname) + 14 +
-                                               strlen(tables.table_name) + 8)))
+      if (!(load_data_query= (char *)thd->alloc(get_query_buffer_length() + 1)))
       {
-        thd->query_length= (uint)(strxmov(load_data_query,
-                                          "LOAD DATA INFILE '", fname,
-                                          "' INTO TABLE `", tables.table_name,
-                                          "` <...>", NullS) - load_data_query);
-        thd->query= load_data_query;
+        /*
+          This will set thd->fatal_error in case of OOM. So we surely will notice
+          that something is wrong.
+        */
+        goto error;
       }
+
+      print_query(FALSE, load_data_query, &end, (char **)&thd->lex->fname_start,
+                  (char **)&thd->lex->fname_end);
+      *end= 0;
+      thd->query_length= end - load_data_query;
+      thd->query= load_data_query;
+
       if (sql_ex.opt_flags & REPLACE_FLAG)
 	handle_dup= DUP_REPLACE;
       else if (sql_ex.opt_flags & IGNORE_FLAG)
@@ -2654,6 +2698,7 @@ int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli,
       List<Item> field_list;
       set_fields(thd->db,field_list);
       thd->variables.pseudo_thread_id= thread_id;
+      List<Item> set_fields;
       if (net)
       {
 	// mysql_load will use thd->net to read the file
@@ -2663,9 +2708,13 @@ int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli,
 	*/
 	thd->net.pkt_nr = net->pkt_nr;
       }
-      if (mysql_load(thd, &ex, &tables, field_list, handle_dup, ignore,
-                     net != 0, TL_WRITE))
-	thd->query_error = 1;
+      /*
+        It is safe to use set_fields twice because we are not going to
+        update it inside mysql_load().
+      */
+      if (mysql_load(thd, &ex, &tables, field_list, set_fields, set_fields,
+                     handle_dup, ignore, net != 0, TL_WRITE))
+        thd->query_error= 1;
       if (thd->cuted_fields)
       {
 	/* log_pos is the position of the LOAD event in the master log */
@@ -2691,7 +2740,8 @@ int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli,
     if (net)
       skip_load_data_infile(net);
   }
-	    
+
+error:
   thd->net.vio = 0; 
   char *save_db= thd->db;
   VOID(pthread_mutex_lock(&LOCK_thread_count));
@@ -2700,8 +2750,6 @@ int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli,
   thd->query_length= thd->db_length= 0;
   VOID(pthread_mutex_unlock(&LOCK_thread_count));
   close_thread_tables(thd);
-  if (load_data_query)
-    my_afree(load_data_query);
   if (thd->query_error)
   {
     /* this err/sql_errno code is copy-paste from net_send_error() */
@@ -4016,8 +4064,8 @@ void Append_block_log_event::print(FILE* file, bool short_form,
     return;
   print_header(file);
   fputc('\n', file);
-  fprintf(file, "#Append_block: file_id: %d  block_len: %d\n",
-	  file_id, block_len);
+  fprintf(file, "#%s: file_id: %d  block_len: %d\n",
+	  get_type_str(), file_id, block_len);
 }
 #endif /* MYSQL_CLIENT */
 
@@ -4036,14 +4084,21 @@ void Append_block_log_event::pack_info(Protocol *protocol)
 			     block_len));
   protocol->store(buf, length, &my_charset_bin);
 }
-#endif /* defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) */
 
 
 /*
+  Append_block_log_event::get_open_mode()
+*/
+
+int Append_block_log_event::get_open_mode() const
+{
+  return O_WRONLY | O_APPEND | O_BINARY;
+}
+
+/*
   Append_block_log_event::exec_event()
 */
 
-#if defined( HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
 int Append_block_log_event::exec_event(struct st_relay_log_info* rli)
 {
   char proc_info[17+FN_REFLEN+10], *fname= proc_info+17;
@@ -4055,14 +4110,18 @@ int Append_block_log_event::exec_event(struct st_relay_log_info* rli)
   memcpy(p, ".data", 6);
   strnmov(proc_info, "Making temp file ", 17); // no end 0
   thd->proc_info= proc_info;
-  if ((fd = my_open(fname, O_WRONLY|O_APPEND|O_BINARY, MYF(MY_WME))) < 0)
+  if ((fd = my_open(fname, get_open_mode(), MYF(MY_WME))) < 0)
   {
-    slave_print_error(rli,my_errno, "Error in Append_block event: could not open file '%s'", fname);
+    slave_print_error(rli, my_errno,
+                      "Error in %s event: could not open file '%s'",
+                      get_type_str(), fname);
     goto err;
   }
   if (my_write(fd, (byte*) block, block_len, MYF(MY_WME+MY_NABP)))
   {
-    slave_print_error(rli,my_errno, "Error in Append_block event: write to '%s' failed", fname);
+    slave_print_error(rli, my_errno,
+                      "Error in %s event: write to '%s' failed",
+                      get_type_str(), fname);
     goto err;
   }
   error=0;
@@ -4335,6 +4394,216 @@ err:
 
 
 /**************************************************************************
+	Begin_load_query_log_event methods
+**************************************************************************/
+
+#ifndef MYSQL_CLIENT
+Begin_load_query_log_event::
+Begin_load_query_log_event(THD* thd_arg, const char* db_arg, char* block_arg,
+                           uint block_len_arg, bool using_trans)
+  :Append_block_log_event(thd_arg, db_arg, block_arg, block_len_arg,
+                          using_trans)
+{
+   file_id= thd_arg->file_id= mysql_bin_log.next_file_id();
+}
+#endif
+
+
+Begin_load_query_log_event::
+Begin_load_query_log_event(const char* buf, uint len,
+                           const Format_description_log_event* desc_event)
+  :Append_block_log_event(buf, len, desc_event)
+{
+}
+
+
+#if defined( HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
+int Begin_load_query_log_event::get_open_mode() const
+{
+  return O_CREAT | O_WRONLY | O_BINARY | O_TRUNC;
+}
+#endif /* defined( HAVE_REPLICATION) && !defined(MYSQL_CLIENT) */
+
+
+/**************************************************************************
+	Execute_load_query_log_event methods
+**************************************************************************/
+
+
+#ifndef MYSQL_CLIENT
+Execute_load_query_log_event::
+Execute_load_query_log_event(THD* thd_arg, const char* query_arg,
+                     ulong query_length_arg, uint fn_pos_start_arg,
+                     uint fn_pos_end_arg,
+                     enum_load_dup_handling dup_handling_arg,
+                     bool using_trans, bool suppress_use):
+  Query_log_event(thd_arg, query_arg, query_length_arg, using_trans,
+                  suppress_use),
+  file_id(thd_arg->file_id), fn_pos_start(fn_pos_start_arg),
+  fn_pos_end(fn_pos_end_arg), dup_handling(dup_handling_arg)
+{
+}
+#endif /* !MYSQL_CLIENT */
+
+
+Execute_load_query_log_event::
+Execute_load_query_log_event(const char* buf, uint event_len,
+                             const Format_description_log_event* desc_event):
+  Query_log_event(buf, event_len, desc_event, EXECUTE_LOAD_QUERY_EVENT),
+  file_id(0), fn_pos_start(0), fn_pos_end(0)
+{
+  if (!Query_log_event::is_valid())
+    return;
+
+  buf+= desc_event->common_header_len;
+
+  fn_pos_start= uint4korr(buf + ELQ_FN_POS_START_OFFSET);
+  fn_pos_end= uint4korr(buf + ELQ_FN_POS_END_OFFSET);
+  dup_handling= (enum_load_dup_handling)(*(buf + ELQ_DUP_HANDLING_OFFSET));
+
+  if (fn_pos_start > q_len || fn_pos_end > q_len ||
+      dup_handling > LOAD_DUP_REPLACE)
+    return;
+
+  file_id= uint4korr(buf + ELQ_FILE_ID_OFFSET);
+}
+
+
+ulong Execute_load_query_log_event::get_post_header_size_for_derived()
+{
+  return EXECUTE_LOAD_QUERY_EXTRA_HEADER_LEN;
+}
+
+
+bool
+Execute_load_query_log_event::write_post_header_for_derived(IO_CACHE* file)
+{
+  char buf[EXECUTE_LOAD_QUERY_EXTRA_HEADER_LEN];
+  int4store(buf, file_id);
+  int4store(buf + 4, fn_pos_start);
+  int4store(buf + 4 + 4, fn_pos_end);
+  *(buf + 4 + 4 + 4)= (char)dup_handling;
+  return my_b_safe_write(file, (byte*) buf, EXECUTE_LOAD_QUERY_EXTRA_HEADER_LEN);
+}
+
+
+#ifdef MYSQL_CLIENT
+void Execute_load_query_log_event::print(FILE* file, bool short_form,
+                                         LAST_EVENT_INFO* last_event_info)
+{
+  print(file, short_form, last_event_info, 0);
+}
+
+
+void Execute_load_query_log_event::print(FILE* file, bool short_form,
+                                         LAST_EVENT_INFO* last_event_info,
+                                         const char *local_fname)
+{
+  print_query_header(file, short_form, last_event_info);
+
+  if (local_fname)
+  {
+    my_fwrite(file, (byte*) query, fn_pos_start, MYF(MY_NABP | MY_WME));
+    fprintf(file, " LOCAL INFILE \'");
+    fprintf(file, local_fname);
+    fprintf(file, "\'");
+    if (dup_handling == LOAD_DUP_REPLACE)
+      fprintf(file, " REPLACE");
+    fprintf(file, " INTO");
+    my_fwrite(file, (byte*) query + fn_pos_end, q_len-fn_pos_end,
+        MYF(MY_NABP | MY_WME));
+    fprintf(file, ";\n");
+  }
+  else
+  {
+    my_fwrite(file, (byte*) query, q_len, MYF(MY_NABP | MY_WME));
+    fprintf(file, ";\n");
+  }
+
+  if (!short_form)
+    fprintf(file, "# file_id: %d \n", file_id);
+}
+#endif
+
+
+#if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
+void Execute_load_query_log_event::pack_info(Protocol *protocol)
+{
+  char *buf, *pos;
+  if (!(buf= my_malloc(9 + db_len + 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, "`; ");
+  }
+  if (query && q_len)
+  {
+    memcpy(pos, query, q_len);
+    pos+= q_len;
+  }
+  pos= strmov(pos, " ;file_id=");
+  pos= int10_to_str((long) file_id, pos, 10);
+  protocol->store(buf, pos-buf, &my_charset_bin);
+  my_free(buf, MYF(MY_ALLOW_ZERO_PTR));
+}
+
+
+int
+Execute_load_query_log_event::exec_event(struct st_relay_log_info* rli)
+{
+  char *p;
+  char *buf;
+  char *fname;
+  char *fname_end;
+  int error;
+
+  /* Replace filename and LOCAL keyword in query before executing it */
+  if (!(buf = my_malloc(q_len + 1 - (fn_pos_end - fn_pos_start) +
+                        (FN_REFLEN + 10) + 10 + 8 + 5, MYF(MY_WME))))
+  {
+    slave_print_error(rli, my_errno, "Not enough memory");
+    return 1;
+  }
+
+  p= buf;
+  memcpy(p, query, fn_pos_start);
+  p+= fn_pos_start;
+  fname= (p= strmake(p, " INFILE \'", 9));
+  p= slave_load_file_stem(p, file_id, server_id);
+  fname_end= (p= strmake(p, ".data", 5));
+  *(p++)='\'';
+  switch (dup_handling)
+  {
+  case LOAD_DUP_IGNORE:
+    p= strmake(p, " IGNORE", 7);
+    break;
+  case LOAD_DUP_REPLACE:
+    p= strmake(p, " REPLACE", 8);
+    break;
+  default:
+    /* Ordinary load data */
+    break;
+  }
+  p= strmake(p, " INTO", 5);
+  p= strmake(p, query+fn_pos_end, q_len-fn_pos_end);
+
+  error= Query_log_event::exec_event(rli, buf, p-buf);
+
+  /* Forging file name for deletion in same buffer */
+  *fname_end= 0;
+
+  (void) my_delete(fname, MYF(MY_WME));
+
+  my_free(buf, MYF(MY_ALLOW_ZERO_PTR));
+  return error;
+}
+#endif
+
+
+/**************************************************************************
 	sql_ex_info methods
 **************************************************************************/
 
diff --git a/sql/log_event.h b/sql/log_event.h
index 7a041959f92..43a801da851 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -163,10 +163,12 @@ struct sql_ex_info
 
   See the #defines below for the format specifics.
 
-  The events which really update data are Query_log_event and
-  Load_log_event/Create_file_log_event/Execute_load_log_event (these 3 act
-  together to replicate LOAD DATA INFILE, with the help of
-  Append_block_log_event which prepares temporary files to load into the table).
+  The events which really update data are Query_log_event,
+  Execute_load_query_log_event and old Load_log_event and
+  Execute_load_log_event events (Execute_load_query is used together with
+  Begin_load_query and Append_block events to replicate LOAD DATA INFILE.
+  Create_file/Append_block/Execute_load (which includes Load_log_event)
+  were used to replicate LOAD DATA before the 5.0.3).
 
  ****************************************************************************/
 
@@ -194,6 +196,8 @@ struct sql_ex_info
 #define EXEC_LOAD_HEADER_LEN   4
 #define DELETE_FILE_HEADER_LEN 4
 #define FORMAT_DESCRIPTION_HEADER_LEN (START_V3_HEADER_LEN+1+LOG_EVENT_TYPES)
+#define EXECUTE_LOAD_QUERY_EXTRA_HEADER_LEN (4 + 4 + 4 + 1)
+#define EXECUTE_LOAD_QUERY_HEADER_LEN  (QUERY_HEADER_LEN + EXECUTE_LOAD_QUERY_EXTRA_HEADER_LEN)
 
 /*
    Event header offsets;
@@ -284,6 +288,12 @@ struct sql_ex_info
 /* DF = "Delete File" */
 #define DF_FILE_ID_OFFSET  0
 
+/* ELQ = "Execute Load Query" */
+#define ELQ_FILE_ID_OFFSET QUERY_HEADER_LEN
+#define ELQ_FN_POS_START_OFFSET ELQ_FILE_ID_OFFSET + 4
+#define ELQ_FN_POS_END_OFFSET ELQ_FILE_ID_OFFSET + 8
+#define ELQ_DUP_HANDLING_OFFSET ELQ_FILE_ID_OFFSET + 12
+
 /* 4 bytes which all binlogs should begin with */
 #define BINLOG_MAGIC        "\xfe\x62\x69\x6e"
 
@@ -387,6 +397,8 @@ enum Log_event_type
   RAND_EVENT, USER_VAR_EVENT,
   FORMAT_DESCRIPTION_EVENT,
   XID_EVENT,
+  BEGIN_LOAD_QUERY_EVENT,
+  EXECUTE_LOAD_QUERY_EVENT,
 
   /*
     add new events here - right above this comment!
@@ -711,13 +723,17 @@ public:
 #ifdef HAVE_REPLICATION
   void pack_info(Protocol* protocol);
   int exec_event(struct st_relay_log_info* rli);
+  int exec_event(struct st_relay_log_info* rli, const char *query_arg,
+                 uint32 q_len_arg);
 #endif /* HAVE_REPLICATION */
 #else
+  void print_query_header(FILE* file, bool short_form = 0, LAST_EVENT_INFO* last_event_info= 0);
   void print(FILE* file, bool short_form = 0, LAST_EVENT_INFO* last_event_info= 0);
 #endif
 
   Query_log_event(const char* buf, uint event_len,
-                  const Format_description_log_event *description_event);
+                  const Format_description_log_event *description_event,
+                  Log_event_type event_type);
   ~Query_log_event()
   {
     if (data_buf)
@@ -728,6 +744,14 @@ public:
   Log_event_type get_type_code() { return QUERY_EVENT; }
   bool write(IO_CACHE* file);
   bool is_valid() const { return query != 0; }
+
+  /*
+    Returns number of bytes additionaly written to post header by derived
+    events (so far it is only Execute_load_query event).
+  */
+  virtual ulong get_post_header_size_for_derived() { return 0; }
+  /* Writes derived event-specific part of post header. */
+  virtual bool write_post_header_for_derived(IO_CACHE* file) { return FALSE; }
 };
 
 #ifdef HAVE_REPLICATION
@@ -779,6 +803,10 @@ public:
  ****************************************************************************/
 class Load_log_event: public Log_event
 {
+private:
+  uint get_query_buffer_length();
+  void print_query(bool need_db, char *buf, char **end,
+                   char **fn_start, char **fn_end);
 protected:
   int copy_log_event(const char *buf, ulong event_len,
                      int body_offset, const Format_description_log_event* description_event);
@@ -1312,6 +1340,7 @@ public:
 #ifdef HAVE_REPLICATION
   int exec_event(struct st_relay_log_info* rli);
   void pack_info(Protocol* protocol);
+  virtual int get_open_mode() const;
 #endif /* HAVE_REPLICATION */
 #else
   void print(FILE* file, bool short_form = 0, LAST_EVENT_INFO* last_event_info= 0);
@@ -1395,6 +1424,93 @@ public:
 };
 
 
+/***************************************************************************
+
+  Begin load query Log Event class
+
+  Event for the first block of file to be loaded, its only difference from
+  Append_block event is that this event creates or truncates existing file
+  before writing data.
+
+****************************************************************************/
+class Begin_load_query_log_event: public Append_block_log_event
+{
+public:
+#ifndef MYSQL_CLIENT
+  Begin_load_query_log_event(THD* thd_arg, const char *db_arg,
+                             char* block_arg, uint block_len_arg,
+                             bool using_trans);
+#ifdef HAVE_REPLICATION
+  Begin_load_query_log_event(THD* thd);
+  int get_open_mode() const;
+#endif /* HAVE_REPLICATION */
+#endif
+  Begin_load_query_log_event(const char* buf, uint event_len,
+                             const Format_description_log_event* description_event);
+  ~Begin_load_query_log_event() {}
+  Log_event_type get_type_code() { return BEGIN_LOAD_QUERY_EVENT; }
+};
+
+
+/*
+  Elements of this enum describe how LOAD DATA handles duplicates.
+*/
+enum enum_load_dup_handling { LOAD_DUP_ERROR= 0, LOAD_DUP_IGNORE,
+                              LOAD_DUP_REPLACE };
+
+/****************************************************************************
+
+  Execute load query Log Event class
+
+  Event responsible for LOAD DATA execution, it similar to Query_log_event
+  but before executing the query it substitutes original filename in LOAD DATA
+  query with name of temporary file.
+
+****************************************************************************/
+class Execute_load_query_log_event: public Query_log_event
+{
+public:
+  uint file_id;       // file_id of temporary file
+  uint fn_pos_start;  // pointer to the part of the query that should
+                      // be substituted
+  uint fn_pos_end;    // pointer to the end of this part of query
+  /*
+    We have to store type of duplicate handling explicitly, because
+    for LOAD DATA it also depends on LOCAL option. And this part
+    of query will be rewritten during replication so this information
+    may be lost...
+  */
+  enum_load_dup_handling dup_handling;
+
+#ifndef MYSQL_CLIENT
+  Execute_load_query_log_event(THD* thd, const char* query_arg,
+                       ulong query_length, uint fn_pos_start_arg,
+                       uint fn_pos_end_arg,
+                       enum_load_dup_handling dup_handling_arg,
+                       bool using_trans, bool suppress_use);
+#ifdef HAVE_REPLICATION
+  void pack_info(Protocol* protocol);
+  int exec_event(struct st_relay_log_info* rli);
+#endif /* HAVE_REPLICATION */
+#else
+  void print(FILE* file, bool short_form = 0,
+             LAST_EVENT_INFO* last_event_info= 0);
+  /* Prints the query as LOAD DATA LOCAL and with rewritten filename */
+  void print(FILE* file, bool short_form, LAST_EVENT_INFO* last_event_info,
+             const char *local_fname);
+#endif
+  Execute_load_query_log_event(const char* buf, uint event_len,
+                               const Format_description_log_event *description_event);
+  ~Execute_load_query_log_event() {}
+
+  Log_event_type get_type_code() { return EXECUTE_LOAD_QUERY_EVENT; }
+  bool is_valid() const { return Query_log_event::is_valid() && file_id != 0; }
+
+  ulong get_post_header_size_for_derived();
+  bool write_post_header_for_derived(IO_CACHE* file);
+ };
+
+
 #ifdef MYSQL_CLIENT
 class Unknown_log_event: public Log_event
 {
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index cd8587f0230..3fe77e29193 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -942,8 +942,10 @@ bool eval_const_cond(COND *cond);
 
 /* sql_load.cc */
 bool mysql_load(THD *thd, sql_exchange *ex, TABLE_LIST *table_list,
-	        List<Item> &fields, enum enum_duplicates handle_duplicates,
-                bool ignore, bool local_file, thr_lock_type lock_type);
+	        List<Item> &fields_vars, List<Item> &set_fields,
+                List<Item> &set_values_list,
+                enum enum_duplicates handle_duplicates, bool ignore,
+                bool local_file, thr_lock_type lock_type);
 int write_record(THD *thd, TABLE *table, COPY_INFO *info);
 
 /* sql_manager.cc */
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index f8e246fc45c..d4caeebb70c 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -5330,3 +5330,5 @@ ER_SP_BAD_SQLSTATE 42000
 	eng "Bad SQLSTATE: '%s'"
 ER_STARTUP
 	eng "%s: ready for connections.\nVersion: '%s'  socket: '%s'  port: %d  %s"
+ER_LOAD_FROM_FIXED_SIZE_ROWS_TO_VAR
+        eng "Can't load value from file with fixed size rows to variable"
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 4cbf9f802d0..afd48943439 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -810,6 +810,12 @@ typedef struct st_lex
   */
   TABLE_LIST **query_tables_own_last;
 
+  /*
+    Pointers to part of LOAD DATA statement that should be rewritten
+    during replication ("LOCAL 'filename' REPLACE INTO" part).
+  */
+  uchar *fname_start, *fname_end;
+
   st_lex() :result(0), sql_command(SQLCOM_END), query_tables_own_last(0)
   {
     extern byte *sp_lex_sp_key(const byte *ptr, uint *plen, my_bool first);
diff --git a/sql/sql_load.cc b/sql/sql_load.cc
index 174ccdfab5b..b27ba9c095f 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -72,18 +72,44 @@ public:
 };
 
 static int read_fixed_length(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
-			     List<Item> &fields, READ_INFO &read_info,
+                             List<Item> &fields_vars, List<Item> &set_fields,
+                             List<Item> &set_values, READ_INFO &read_info,
 			     ulong skip_lines,
 			     bool ignore_check_option_errors);
 static int read_sep_field(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
-			  List<Item> &fields, READ_INFO &read_info,
+                          List<Item> &fields_vars, List<Item> &set_fields,
+                          List<Item> &set_values, READ_INFO &read_info,
 			  String &enclosed, ulong skip_lines,
 			  bool ignore_check_option_errors);
 
+
+/*
+  Execute LOAD DATA query
+
+  SYNOPSYS
+    mysql_load()
+      thd - current thread
+      ex  - sql_exchange object representing source file and its parsing rules
+      table_list  - list of tables to which we are loading data
+      fields_vars - list of fields and variables to which we read
+                    data from file
+      set_fields  - list of fields mentioned in set clause
+      set_values  - expressions to assign to fields in previous list
+      handle_duplicates - indicates whenever we should emit error or
+                          replace row if we will meet duplicates.
+      ignore -          - indicates whenever we should ignore duplicates
+      read_file_from_client - is this LOAD DATA LOCAL ?
+      lock_type - what type of concurrency do we allow then we are inserting data
+
+  RETURN VALUES
+    TRUE - error / FALSE - success
+*/
+
 bool mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
-	       List<Item> &fields, enum enum_duplicates handle_duplicates,
-               bool ignore,
-	       bool read_file_from_client,thr_lock_type lock_type)
+	        List<Item> &fields_vars, List<Item> &set_fields,
+                List<Item> &set_values,
+                enum enum_duplicates handle_duplicates, bool ignore,
+                bool read_file_from_client, thr_lock_type lock_type)
 {
   char name[FN_REFLEN];
   File file;
@@ -130,48 +156,80 @@ bool mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
     my_error(ER_NON_UPDATABLE_TABLE, MYF(0), table_list->alias, "LOAD");
     DBUG_RETURN(TRUE);
   }
+  /*
+    Let us emit an error if we are loading data to table which is used
+    in subselect in SET clause like we do it for INSERT.
+
+    The main thing to fix to remove this restriction is to ensure that the
+    table is marked to be 'used for insert' in which case we should never
+    mark this table as as 'const table' (ie, one that has only one row).
+  */
+  if (unique_table(table_list, table_list->next_global))
+  {
+    my_error(ER_UPDATE_TABLE_USED, MYF(0), table_list->table_name);
+    DBUG_RETURN(TRUE);
+  }
+
   table= table_list->table;
   transactional_table= table->file->has_transactions();
 
-  if (!fields.elements)
+  if (!fields_vars.elements)
   {
     Field **field;
     for (field=table->field; *field ; field++)
-      fields.push_back(new Item_field(*field));
+      fields_vars.push_back(new Item_field(*field));
+    table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
+    /*
+      Let us also prepare SET clause, altough it is probably empty
+      in this case.
+    */
+    if (setup_fields(thd, 0, table_list, set_fields, 1, 0, 0) ||
+        setup_fields(thd, 0, table_list, set_values, 1, 0, 0))
+      DBUG_RETURN(TRUE);
   }
   else
   {						// Part field list
-    thd->dupp_field=0;
     /* TODO: use this conds for 'WITH CHECK OPTIONS' */
-    if (setup_fields(thd, 0, table_list, fields, 1, 0, 0))
+    if (setup_fields(thd, 0, table_list, fields_vars, 1, 0, 0) ||
+        setup_fields(thd, 0, table_list, set_fields, 1, 0, 0) ||
+        check_that_all_fields_are_given_values(thd, table))
       DBUG_RETURN(TRUE);
-    if (thd->dupp_field)
-    {
-      my_error(ER_FIELD_SPECIFIED_TWICE, MYF(0), thd->dupp_field->field_name);
-      DBUG_RETURN(TRUE);
-    }
-    if (check_that_all_fields_are_given_values(thd, table))
+    /*
+      Check whenever TIMESTAMP field with auto-set feature specified
+      explicitly.
+    */
+    if (table->timestamp_field &&
+        table->timestamp_field->query_id == thd->query_id)
+      table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
+    /*
+      Fix the expressions in SET clause. This should be done after
+      check_that_all_fields_are_given_values() and setting use_timestamp
+      since it may update query_id for some fields.
+    */
+    if (setup_fields(thd, 0, table_list, set_values, 1, 0, 0))
       DBUG_RETURN(TRUE);
   }
 
   uint tot_length=0;
-  bool use_blobs=0,use_timestamp=0;
-  List_iterator_fast<Item> it(fields);
+  bool use_blobs= 0, use_vars= 0;
+  List_iterator_fast<Item> it(fields_vars);
+  Item *item;
 
-  Item_field *field;
-  while ((field=(Item_field*) it++))
+  while ((item= it++))
   {
-    if (field->field->flags & BLOB_FLAG)
+    if (item->type() == Item::FIELD_ITEM)
     {
-      use_blobs=1;
-      tot_length+=256;				// Will be extended if needed
+      Field *field= ((Item_field*)item)->field;
+      if (field->flags & BLOB_FLAG)
+      {
+        use_blobs= 1;
+        tot_length+= 256;			// Will be extended if needed
+      }
+      else
+        tot_length+= field->field_length;
     }
     else
-      tot_length+=field->field->field_length;
-    if (!field_term->length() && !(field->field->flags & NOT_NULL_FLAG))
-      field->field->set_notnull();
-    if (field->field == table->timestamp_field)
-      use_timestamp=1;
+      use_vars= 1;
   }
   if (use_blobs && !ex->line_term->length() && !field_term->length())
   {
@@ -179,6 +237,11 @@ bool mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
 	       MYF(0));
     DBUG_RETURN(TRUE);
   }
+  if (use_vars && !field_term->length() && !enclosed->length())
+  {
+    my_error(ER_LOAD_FROM_FIXED_SIZE_ROWS_TO_VAR, MYF(0));
+    DBUG_RETURN(TRUE);
+  }
 
   /* We can't give an error in the middle when using LOCAL files */
   if (read_file_from_client && handle_duplicates == DUP_ERROR)
@@ -251,12 +314,6 @@ bool mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
   if (mysql_bin_log.is_open())
   {
     lf_info.thd = thd;
-    lf_info.ex = ex;
-    lf_info.db = db;
-    lf_info.table_name = table_list->table_name;
-    lf_info.fields = &fields;
-    lf_info.ignore= ignore;
-    lf_info.handle_dup = handle_duplicates;
     lf_info.wrote_create_file = 0;
     lf_info.last_pos_in_file = HA_POS_ERROR;
     lf_info.log_delayed= transactional_table;
@@ -264,8 +321,6 @@ bool mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
   }
 #endif /*!EMBEDDED_LIBRARY*/
 
-  restore_record(table, s->default_values);
-
   thd->count_cuted_fields= CHECK_FIELD_WARN;		/* calc cuted fields */
   thd->cuted_fields=0L;
   /* Skip lines if there is a line terminator */
@@ -282,8 +337,6 @@ bool mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
 
   if (!(error=test(read_info.error)))
   {
-    if (use_timestamp)
-      table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
 
     table->next_number_field=table->found_next_number_field;
     if (ignore ||
@@ -300,12 +353,13 @@ bool mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
                               MODE_STRICT_ALL_TABLES)));
 
     if (!field_term->length() && !enclosed->length())
-      error= read_fixed_length(thd, info, table_list, fields,read_info,
+      error= read_fixed_length(thd, info, table_list, fields_vars,
+                               set_fields, set_values, read_info,
 			       skip_lines, ignore);
     else
-      error= read_sep_field(thd, info, table_list, fields, read_info,
-			    *enclosed, skip_lines,
-			    ignore);
+      error= read_sep_field(thd, info, table_list, fields_vars,
+                            set_fields, set_values, read_info,
+			    *enclosed, skip_lines, ignore);
     if (table->file->end_bulk_insert())
       error=1;					/* purecov: inspected */
     ha_enable_transaction(thd, TRUE);
@@ -380,13 +434,19 @@ bool mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
   {
     /*
       As already explained above, we need to call end_io_cache() or the last
-      block will be logged only after Execute_load_log_event (which is wrong),
-      when read_info is destroyed.
+      block will be logged only after Execute_load_query_log_event (which is
+      wrong), when read_info is destroyed.
     */
     read_info.end_io_cache();
     if (lf_info.wrote_create_file)
     {
-      Execute_load_log_event e(thd, db, transactional_table);
+      Execute_load_query_log_event e(thd, thd->query, thd->query_length,
+        (char*)thd->lex->fname_start - (char*)thd->query,
+        (char*)thd->lex->fname_end - (char*)thd->query,
+        (handle_duplicates == DUP_REPLACE) ?  LOAD_DUP_REPLACE :
+                                              (ignore ? LOAD_DUP_IGNORE :
+                                                        LOAD_DUP_ERROR),
+        transactional_table, FALSE);
       mysql_bin_log.write(&e);
     }
   }
@@ -410,10 +470,11 @@ err:
 
 static int
 read_fixed_length(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
-		  List<Item> &fields, READ_INFO &read_info, ulong skip_lines,
-		  bool ignore_check_option_errors)
+                  List<Item> &fields_vars, List<Item> &set_fields,
+                  List<Item> &set_values, READ_INFO &read_info,
+                  ulong skip_lines, bool ignore_check_option_errors)
 {
-  List_iterator_fast<Item> it(fields);
+  List_iterator_fast<Item> it(fields_vars);
   Item_field *sql_field;
   TABLE *table= table_list->table;
   ulonglong id;
@@ -421,11 +482,7 @@ read_fixed_length(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
   DBUG_ENTER("read_fixed_length");
 
   id= 0;
-  
-  /* No fields can be null in this format. mark all fields as not null */
-  while ((sql_field= (Item_field*) it++))
-      sql_field->field->set_notnull();
-
+ 
   while (!read_info.read_fixed_length())
   {
     if (thd->killed)
@@ -450,16 +507,28 @@ read_fixed_length(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
     read_info.row_end[0]=0;
 #endif
     no_trans_update= !table->file->has_transactions();
+
+    restore_record(table, s->default_values);
+    /*
+      There is no variables in fields_vars list in this format so
+      this conversion is safe.
+    */
     while ((sql_field= (Item_field*) it++))
     {
       Field *field= sql_field->field;                  
+      /*
+        No fields specified in fields_vars list can be null in this format.
+        Mark field as not null, we should do this for each row because of
+        restore_record...
+      */
+      field->set_notnull();
+
       if (pos == read_info.row_end)
       {
         thd->cuted_fields++;			/* Not enough fields */
         push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN, 
                             ER_WARN_TOO_FEW_RECORDS, 
                             ER(ER_WARN_TOO_FEW_RECORDS), thd->row_count);
-	      field->reset();
       }
       else
       {
@@ -483,6 +552,9 @@ read_fixed_length(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
                           ER(ER_WARN_TOO_MANY_RECORDS), thd->row_count); 
     }
 
+    if (fill_record(thd, set_fields, set_values, ignore_check_option_errors))
+      DBUG_RETURN(1);
+
     switch (table_list->view_check_option(thd,
                                           ignore_check_option_errors)) {
     case VIEW_CHECK_SKIP:
@@ -527,12 +599,13 @@ continue_loop:;
 
 static int
 read_sep_field(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
-	       List<Item> &fields, READ_INFO &read_info,
+               List<Item> &fields_vars, List<Item> &set_fields,
+               List<Item> &set_values, READ_INFO &read_info,
 	       String &enclosed, ulong skip_lines,
 	       bool ignore_check_option_errors)
 {
-  List_iterator_fast<Item> it(fields);
-  Item_field *sql_field;
+  List_iterator_fast<Item> it(fields_vars);
+  Item *item;
   TABLE *table= table_list->table;
   uint enclosed_length;
   ulonglong id;
@@ -550,60 +623,95 @@ read_sep_field(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
       thd->send_kill_message();
       DBUG_RETURN(1);
     }
-    while ((sql_field=(Item_field*) it++))
+
+    restore_record(table, s->default_values);
+
+    while ((item= it++))
     {
       uint length;
       byte *pos;
 
       if (read_info.read_field())
 	break;
+
+      /* If this line is to be skipped we don't want to fill field or var */
+      if (skip_lines)
+        continue;
+
       pos=read_info.row_start;
       length=(uint) (read_info.row_end-pos);
-      Field *field=sql_field->field;
 
       if (!read_info.enclosed &&
 	  (enclosed_length && length == 4 && !memcmp(pos,"NULL",4)) ||
 	  (length == 1 && read_info.found_null))
       {
-	field->reset();
-	field->set_null();
-	if (!field->maybe_null())
-	{
-	  if (field->type() == FIELD_TYPE_TIMESTAMP)
-	    ((Field_timestamp*) field)->set_time();
-	  else if (field != table->next_number_field)      
-	    field->set_warning((uint) MYSQL_ERROR::WARN_LEVEL_WARN, 
-			       ER_WARN_NULL_TO_NOTNULL, 1);
+        if (item->type() == Item::FIELD_ITEM)
+        {
+          Field *field= ((Item_field *)item)->field;
+          field->reset();
+          field->set_null();
+          if (!field->maybe_null())
+          {
+            if (field->type() == FIELD_TYPE_TIMESTAMP)
+              ((Field_timestamp*) field)->set_time();
+            else if (field != table->next_number_field)
+              field->set_warning((uint) MYSQL_ERROR::WARN_LEVEL_WARN,
+                                 ER_WARN_NULL_TO_NOTNULL, 1);
+          }
 	}
+        else
+          ((Item_user_var_as_out_param *)item)->set_null_value(
+                                                  read_info.read_charset);
 	continue;
       }
-      field->set_notnull();
-      read_info.row_end[0]=0;			// Safe to change end marker
-      field->store((char*) read_info.row_start,length,read_info.read_charset);
+
+      if (item->type() == Item::FIELD_ITEM)
+      {
+        Field *field= ((Item_field *)item)->field;
+        field->set_notnull();
+        read_info.row_end[0]=0;			// Safe to change end marker
+        field->store((char*) pos, length, read_info.read_charset);
+      }
+      else
+        ((Item_user_var_as_out_param *)item)->set_value((char*) pos, length,
+                                                read_info.read_charset);
     }
     if (read_info.error)
       break;
     if (skip_lines)
     {
-      if (!--skip_lines)
-	thd->cuted_fields= 0L;			// Reset warnings
+      skip_lines--;
       continue;
     }
-    if (sql_field)
-    {						// Last record
-      if (sql_field == (Item_field*) fields.head())
+    if (item)
+    {
+      /* Have not read any field, thus input file is simply ended */
+      if (item == fields_vars.head())
 	break;
-      for (; sql_field ; sql_field=(Item_field*) it++)
+      for (; item ; item= it++)
       {
-	sql_field->field->set_null();
-	sql_field->field->reset();
-	thd->cuted_fields++;
- 	push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN, 
-                	    ER_WARN_TOO_FEW_RECORDS,
-                	    ER(ER_WARN_TOO_FEW_RECORDS), thd->row_count);
+        if (item->type() == Item::FIELD_ITEM)
+        {
+          /*
+            QQ: We probably should not throw warning for each field.
+            But how about intention to always have the same number
+            of warnings in THD::cuted_fields (and get rid of cuted_fields
+            in the end ?)
+          */
+          thd->cuted_fields++;
+          push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                              ER_WARN_TOO_FEW_RECORDS,
+                              ER(ER_WARN_TOO_FEW_RECORDS), thd->row_count);
+        }
+        else
+          ((Item_user_var_as_out_param *)item)->set_null_value(
+                                                  read_info.read_charset);
       }
     }
 
+    if (fill_record(thd, set_fields, set_values, ignore_check_option_errors))
+      DBUG_RETURN(1);
+
     switch (table_list->view_check_option(thd,
                                           ignore_check_option_errors)) {
     case VIEW_CHECK_SKIP:
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index d5ed8c8efb0..77580512c5a 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -3337,14 +3337,10 @@ unsent_create_error:
   {
     DBUG_ASSERT(first_table == all_tables && first_table != 0);
     uint privilege= (lex->duplicates == DUP_REPLACE ?
-		     INSERT_ACL | DELETE_ACL : INSERT_ACL);
+		     INSERT_ACL | DELETE_ACL : INSERT_ACL) |
+                    (lex->local_file ? 0 : FILE_ACL);
 
-    if (!lex->local_file)
-    {
-      if (check_access(thd, privilege | FILE_ACL, first_table->db, 0, 0, 0))
-	goto error;
-    }
-    else
+    if (lex->local_file)
     {
       if (!(thd->client_capabilities & CLIENT_LOCAL_FILES) ||
 	  ! opt_local_infile)
@@ -3352,12 +3348,14 @@ unsent_create_error:
 	my_message(ER_NOT_ALLOWED_COMMAND, ER(ER_NOT_ALLOWED_COMMAND), MYF(0));
 	goto error;
       }
-      if (check_one_table_access(thd, privilege, all_tables))
-	goto error;
     }
+
+    if (check_one_table_access(thd, privilege, all_tables))
+      goto error;
+
     res= mysql_load(thd, lex->exchange, first_table, lex->field_list,
-                    lex->duplicates, lex->ignore, (bool) lex->local_file,
-		    lex->lock_option);
+                    lex->update_list, lex->value_list, lex->duplicates,
+                    lex->ignore, (bool) lex->local_file, lex->lock_option);
     break;
   }
 
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 0284ab542df..634b6ab0995 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -1511,17 +1511,16 @@ int log_loaded_block(IO_CACHE* file)
   lf_info->last_pos_in_file = file->pos_in_file;
   if (lf_info->wrote_create_file)
   {
-    Append_block_log_event a(lf_info->thd, lf_info->db, buffer, block_len,
-			     lf_info->log_delayed);
+    Append_block_log_event a(lf_info->thd, lf_info->thd->db, buffer,
+                             block_len, lf_info->log_delayed);
     mysql_bin_log.write(&a);
   }
   else
   {
-    Create_file_log_event c(lf_info->thd,lf_info->ex,lf_info->db,
-			    lf_info->table_name, *lf_info->fields,
-			    lf_info->handle_dup, lf_info->ignore, buffer,
-			    block_len, lf_info->log_delayed);
-    mysql_bin_log.write(&c);
+    Begin_load_query_log_event b(lf_info->thd, lf_info->thd->db,
+                                 buffer, block_len,
+                                 lf_info->log_delayed);
+    mysql_bin_log.write(&b);
     lf_info->wrote_create_file = 1;
     DBUG_SYNC_POINT("debug_lock.created_file_event",10);
   }
diff --git a/sql/sql_repl.h b/sql/sql_repl.h
index b901d7cfe0c..9eb6456ee20 100644
--- a/sql/sql_repl.h
+++ b/sql/sql_repl.h
@@ -63,12 +63,7 @@ typedef struct st_load_file_info
 {
   THD* thd;
   my_off_t last_pos_in_file;
-  sql_exchange* ex;
-  List <Item> *fields;
-  enum enum_duplicates handle_dup;
-  char* db;
-  char* table_name;
-  bool wrote_create_file, log_delayed, ignore;
+  bool wrote_create_file, log_delayed;
 } LOAD_FILE_INFO;
 
 int log_loaded_block(IO_CACHE* file);
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 56dd6409eba..e2d0dba7a24 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -686,7 +686,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 	text_string opt_gconcat_separator
 
 %type <num>
-	type int_type real_type order_dir opt_field_spec lock_option
+	type int_type real_type order_dir lock_option
 	udf_type if_exists opt_local opt_table_options table_options
         table_option opt_if_not_exists opt_no_write_to_binlog opt_var_type
         opt_var_ident_type delete_option opt_temporary all_or_any opt_distinct
@@ -714,6 +714,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 	signed_literal now_or_signed_literal opt_escape
 	sp_opt_default
 	simple_ident_nospvar simple_ident_q
+        field_or_var
 
 %type <item_num>
 	NUM_literal
@@ -809,6 +810,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
         prepare prepare_src execute deallocate 
 	statement sp_suid opt_view_list view_list or_replace algorithm
 	sp_c_chistics sp_a_chistics sp_chistic sp_c_chistic xa
+        load_data opt_field_or_var_spec fields_or_vars opt_load_data_set_spec
 END_OF_INPUT
 
 %type <NONE> call sp_proc_stmts sp_proc_stmts1 sp_proc_stmt
@@ -5703,11 +5705,6 @@ insert_field_spec:
 	   }
 	   ident_eq_list;
 
-opt_field_spec:
-	/* empty */	  { }
-	| '(' fields ')'  { }
-	| '(' ')'	  { };
-
 fields:
 	fields ',' insert_ident { Lex->field_list.push_back($3); }
 	| insert_ident		{ Lex->field_list.push_back($1); };
@@ -6409,34 +6406,49 @@ use:	USE_SYM ident
 
 /* import, export of files */
 
-load:	LOAD DATA_SYM load_data_lock opt_local INFILE TEXT_STRING_sys
+load:   LOAD DATA_SYM 
+        {
+          LEX *lex=Lex;
+          lex->fname_start= lex->ptr;
+        }
+        load_data
+        {}
+        |
+        LOAD TABLE_SYM table_ident FROM MASTER_SYM
+        {
+          Lex->sql_command = SQLCOM_LOAD_MASTER_TABLE;
+          if (!Select->add_table_to_list(YYTHD, $3, NULL, TL_OPTION_UPDATING))
+            YYABORT;
+        };
+
+load_data:
+	load_data_lock opt_local INFILE TEXT_STRING_sys
 	{
 	  LEX *lex=Lex;
 	  lex->sql_command= SQLCOM_LOAD;
-	  lex->lock_option= $3;
-	  lex->local_file=  $4;
+	  lex->lock_option= $1;
+	  lex->local_file=  $2;
 	  lex->duplicates= DUP_ERROR;
 	  lex->ignore= 0;
-	  if (!(lex->exchange= new sql_exchange($6.str,0)))
+	  if (!(lex->exchange= new sql_exchange($4.str, 0)))
 	    YYABORT;
+        }
+        opt_duplicate INTO
+        {
+	  LEX *lex=Lex;
+	  lex->fname_end= lex->ptr;
 	  lex->field_list.empty();
+	  lex->update_list.empty();
+	  lex->value_list.empty();
 	}
-	opt_duplicate INTO TABLE_SYM table_ident opt_field_term opt_line_term
-	opt_ignore_lines opt_field_spec
+	TABLE_SYM table_ident opt_field_term opt_line_term
+	opt_ignore_lines opt_field_or_var_spec opt_load_data_set_spec
 	{
-	  if (!Select->add_table_to_list(YYTHD, $11, NULL, TL_OPTION_UPDATING))
+	  if (!Select->add_table_to_list(YYTHD, $10, NULL, TL_OPTION_UPDATING))
 	    YYABORT;
 	}
         |
-	LOAD TABLE_SYM table_ident FROM MASTER_SYM
-        {
-	  Lex->sql_command = SQLCOM_LOAD_MASTER_TABLE;
-	  if (!Select->add_table_to_list(YYTHD, $3, NULL, TL_OPTION_UPDATING))
-	    YYABORT;
-
-        }
-        |
-	LOAD DATA_SYM FROM MASTER_SYM
+	FROM MASTER_SYM
         {
 	  Lex->sql_command = SQLCOM_LOAD_MASTER_DATA;
         };
@@ -6516,6 +6528,29 @@ opt_ignore_lines:
             Lex->exchange->skip_lines= atol($2.str);
           };
 
+opt_field_or_var_spec:
+	/* empty */	          { }
+	| '(' fields_or_vars ')'  { }
+	| '(' ')'	          { };
+
+fields_or_vars:
+        fields_or_vars ',' field_or_var
+          { Lex->field_list.push_back($3); }
+        | field_or_var
+          { Lex->field_list.push_back($1); }
+        ;
+        
+field_or_var:
+        simple_ident_nospvar {$$= $1;}
+        | '@' ident_or_text
+          { $$= new Item_user_var_as_out_param($2); }
+        ;
+
+opt_load_data_set_spec:
+        /* empty */           { }
+        | SET insert_update_list  { };
+
+
 /* Common definitions */
 
 text_literal: