BUG#34141: mysqlbinlog cannot read 4.1 binlogs containing load data infile

Main problem: mysql 5.1 cannot read binlogs from 4.1.
Subproblem 1: There is a mistake in sql_ex_info::init. The read_str()
function updates its first argument to point to the next character to
read. However, it is applied only to a copy of the buffer pointer, so the
real buffer pointer is not updated.
Fix 1: do not take a copy of the buffer pointer. The copy was needed
because sql_ex_info::init does not use the const attribute on some of its
arguments. So we add the const attribute, too.
Subproblem 2: The first BINLOG statement is asserted to be a
FORMAT_DESCRIPTION_LOG_EVENT, but 4.1 binlogs begin with START_EVENT_V3.
Fix 2: allow START_EVENT_V3 too.


mysql-test/suite/binlog/std_data/binlog_old_version_4_1.000001:
  New BitKeeper file ``mysql-test/suite/binlog/std_data/binlog_old_version_4_1.000001''
mysql-test/suite/binlog/r/binlog_old_versions.result:
  Updated result file.
mysql-test/suite/binlog/t/binlog_old_versions.test:
  Added a test reading an old 4.1 binlog.
sql/log_event.cc:
  1. Added const keyword at the following places:
   - input buffer for pretty_print_str
   - input buffer for write_str
   - input buffer, end pointer, and return value from sql_ex_info::init
  2. Fixed the bug by not taking a copy of buf before calling read_str in
  sql_ex_info::init().
sql/log_event.h:
  Added const keyword to fields of the sql_ex_info struct.
  Added const keyword to arguments and return value of sql_ex_info::init
sql/sql_binlog.cc:
  The first BINLOG statement must describe the format for future BINLOG
  statements. Otherwise, we do not know how to read the BINLOG statement.
  Problem: only FORMAT_DESCRIPTION_EVENT is currently allowed as the first
  event. Binlogs from 4.1 begin with a START_EVENT_V3, which serves the
  same purpose.
  Fix: We now allow the first BINLOG statement to be a START_EVENT_V3, as
  well as a FORMAT_DESCRIPTION_EVENT.

3 files changed, 21 insertions, 22 deletions

diff --git a/sql/log_event.cc b/sql/log_event.cc
index 45478020a36..070e6ab4c12 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -212,9 +212,9 @@ uint debug_not_change_ts_if_art_event= 1; // bug#29309 simulation
 */
 
 #ifdef MYSQL_CLIENT
-static void pretty_print_str(IO_CACHE* cache, char* str, int len)
+static void pretty_print_str(IO_CACHE* cache, const char* str, int len)
 {
-  char* end = str + len;
+  const char* end = str + len;
   my_b_printf(cache, "\'");
   while (str < end)
   {
@@ -277,9 +277,9 @@ inline int ignored_error_code(int err_code)
 */
 
 #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
-static char *pretty_print_str(char *packet, char *str, int len)
+static char *pretty_print_str(char *packet, const char *str, int len)
 {
-  char *end= str + len;
+  const char *end= str + len;
   char *pos= packet;
   *pos++= '\'';
   while (str < end)
@@ -388,7 +388,7 @@ static void cleanup_load_tmpdir()
   write_str()
 */
 
-static bool write_str(IO_CACHE *file, char *str, uint length)
+static bool write_str(IO_CACHE *file, const char *str, uint length)
 {
   uchar tmp[1];
   tmp[0]= (uchar) length;
@@ -6011,7 +6011,8 @@ bool sql_ex_info::write_data(IO_CACHE* file)
   sql_ex_info::init()
 */
 
-char *sql_ex_info::init(char *buf, char *buf_end, bool use_new_format)
+const char *sql_ex_info::init(const char *buf, const char *buf_end,
+                              bool use_new_format)
 {
   cached_new_format = use_new_format;
   if (use_new_format)
@@ -6024,12 +6025,11 @@ char *sql_ex_info::init(char *buf, char *buf_end, bool use_new_format)
       the case when we have old format because we will be reusing net buffer
       to read the actual file before we write out the Create_file event.
     */
-    const char *ptr= buf;
-    if (read_str(&ptr, buf_end, (const char **) &field_term, &field_term_len) ||
-	read_str(&ptr, buf_end, (const char **) &enclosed,   &enclosed_len) ||
-	read_str(&ptr, buf_end, (const char **) &line_term,  &line_term_len) ||
-	read_str(&ptr, buf_end, (const char **) &line_start, &line_start_len) ||
-	read_str(&ptr, buf_end, (const char **) &escaped,    &escaped_len))
+    if (read_str(&buf, buf_end, &field_term, &field_term_len) ||
+        read_str(&buf, buf_end, &enclosed,   &enclosed_len) ||
+        read_str(&buf, buf_end, &line_term,  &line_term_len) ||
+        read_str(&buf, buf_end, &line_start, &line_start_len) ||
+        read_str(&buf, buf_end, &escaped,    &escaped_len))
       return 0;
     opt_flags = *buf++;
   }
diff --git a/sql/log_event.h b/sql/log_event.h
index 31c1ab7173a..c85d620d831 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -152,11 +152,11 @@ struct old_sql_ex
 struct sql_ex_info
 {
   sql_ex_info() {}                            /* Remove gcc warning */
-  char* field_term;
-  char* enclosed;
-  char* line_term;
-  char* line_start;
-  char* escaped;
+  const char* field_term;
+  const char* enclosed;
+  const char* line_term;
+  const char* line_start;
+  const char* escaped;
   int cached_new_format;
   uint8 field_term_len,enclosed_len,line_term_len,line_start_len, escaped_len;
   char opt_flags;
@@ -171,7 +171,7 @@ struct sql_ex_info
 	    line_start_len + escaped_len + 6 : 7);
   }
   bool write_data(IO_CACHE* file);
-  char* init(char* buf,char* buf_end,bool use_new_format);
+  const char* init(const char* buf, const char* buf_end, bool use_new_format);
   bool new_format()
   {
     return ((cached_new_format != -1) ? cached_new_format :
diff --git a/sql/sql_binlog.cc b/sql/sql_binlog.cc
index 543b1af9fc0..778aa46149c 100644
--- a/sql/sql_binlog.cc
+++ b/sql/sql_binlog.cc
@@ -159,14 +159,13 @@ void mysql_client_binlog_statement(THD* thd)
       */
       if (!have_fd_event)
       {
-        if (bufptr[EVENT_TYPE_OFFSET] == FORMAT_DESCRIPTION_EVENT)
+        int type = bufptr[EVENT_TYPE_OFFSET];
+        if (type == FORMAT_DESCRIPTION_EVENT || type == START_EVENT_V3)
           have_fd_event= TRUE;
         else
         {
           my_error(ER_NO_FORMAT_DESCRIPTION_EVENT_BEFORE_BINLOG_STATEMENT,
-                   MYF(0),
-                   Log_event::get_type_str(
-                     (Log_event_type)bufptr[EVENT_TYPE_OFFSET]));
+                   MYF(0), Log_event::get_type_str((Log_event_type)type));
           goto end;
         }
       }