Merge mkindahl@bk-internal.mysql.com:/home/bk/mysql-5.1-new-rpl

into  kindahl-laptop.dnsalias.net:/home/bkroot/mysql-5.1-rpl


sql/field.cc:
  Auto merged
sql/field.h:
  Auto merged
sql/log_event.cc:
  Auto merged
sql/sql_show.cc:
  Auto merged
sql/rpl_record.cc:
  Manual merge.

13 files changed, 527 insertions, 226 deletions

diff --git a/mysql-test/extra/rpl_tests/rpl_extraMaster_Col.test b/mysql-test/extra/rpl_tests/rpl_extraMaster_Col.test
index 26b1e6d5ba0..a03aa337c2b 100644
--- a/mysql-test/extra/rpl_tests/rpl_extraMaster_Col.test
+++ b/mysql-test/extra/rpl_tests/rpl_extraMaster_Col.test
@@ -466,7 +466,7 @@ binary data';
    select * from t2 order by f1;
    select * from t3 order by f1;
    select * from t4 order by f1;
-   select * from t31 order by f1;
+   select * from t31 order by f3;
    
 connection master;
 
diff --git a/mysql-test/suite/rpl/r/rpl_extraColmaster_innodb.result b/mysql-test/suite/rpl/r/rpl_extraColmaster_innodb.result
index 18906617925..ef6e2a6de25 100644
--- a/mysql-test/suite/rpl/r/rpl_extraColmaster_innodb.result
+++ b/mysql-test/suite/rpl/r/rpl_extraColmaster_innodb.result
diff --git a/mysql-test/suite/rpl/r/rpl_extraColmaster_myisam.result b/mysql-test/suite/rpl/r/rpl_extraColmaster_myisam.result
index 5bc1a13b107..475ea4cdcba 100644
--- a/mysql-test/suite/rpl/r/rpl_extraColmaster_myisam.result
+++ b/mysql-test/suite/rpl/r/rpl_extraColmaster_myisam.result
diff --git a/mysql-test/suite/rpl/r/rpl_row_extraColmaster_ndb.result b/mysql-test/suite/rpl/r/rpl_row_extraColmaster_ndb.result
index 0b1f2e6c8bf..fb83c5088e5 100644
--- a/mysql-test/suite/rpl/r/rpl_row_extraColmaster_ndb.result
+++ b/mysql-test/suite/rpl/r/rpl_row_extraColmaster_ndb.result
diff --git a/mysql-test/suite/rpl/t/disabled.def b/mysql-test/suite/rpl/t/disabled.def
index 72b68fa59db..94ce28ce7a3 100644
--- a/mysql-test/suite/rpl/t/disabled.def
+++ b/mysql-test/suite/rpl/t/disabled.def
@@ -13,4 +13,3 @@
 rpl_ddl                  : BUG#26418 2007-03-01 mleich Slave out of sync after CREATE/DROP TEMPORARY TABLE + ROLLBACK on master
 rpl_auto_increment_11932   : Bug#29809 2007-07-16 ingo Slave SQL errors in warnings file
 rpl_stm_extraColmaster_ndb : WL#3915 : Statement-based replication not supported in ndb. Enable test when supported.
-rpl_row_extraColmaster_ndb : BUG#29549 : Replication of BLOBs fail for NDB
diff --git a/mysql-test/suite/rpl/t/rpl_row_mysqlbinlog.test b/mysql-test/suite/rpl/t/rpl_row_mysqlbinlog.test
index dd46d64f684..be76ac9f3f6 100644
--- a/mysql-test/suite/rpl/t/rpl_row_mysqlbinlog.test
+++ b/mysql-test/suite/rpl/t/rpl_row_mysqlbinlog.test
@@ -344,5 +344,6 @@ FLUSH LOGS;
 --exec rm $MYSQLTEST_VARDIR/tmp/local.sql
 
 DROP TABLE IF EXISTS t1, t2, t3, t04, t05, t4, t5; 
+sync_slave_with_master;
 
 # End of 4.1 tests
diff --git a/mysql-test/suite/rpl_ndb/t/disabled.def b/mysql-test/suite/rpl_ndb/t/disabled.def
index f372d44cb90..bb093adaa8c 100644
--- a/mysql-test/suite/rpl_ndb/t/disabled.def
+++ b/mysql-test/suite/rpl_ndb/t/disabled.def
@@ -15,8 +15,6 @@ rpl_ndb_2innodb          : BUG#19227 2006-04-20 pekka pk delete apparently not r
 rpl_ndb_2myisam          : BUG#19227 Seems to pass currently
 rpl_ndb_2other           : BUG#21842 2007-08-30 tsmith test has never worked on bigendian (sol10-sparc-a, powermacg5
 rpl_ndb_dd_partitions    : BUG#19259 2006-04-21 rpl_ndb_dd_partitions fails on s/AMD
-rpl_ndb_innodb2ndb       : Bug#29549 rpl_ndb_myisam2ndb,rpl_ndb_innodb2ndb failed on Solaris for pack_length issue
-rpl_ndb_myisam2ndb       : Bug#29549 rpl_ndb_myisam2ndb,rpl_ndb_innodb2ndb failed on Solaris for pack_length issue
 rpl_ndb_ddl              : BUG#28798 2007-05-31  lars  Valgrind failure in NDB
 rpl_ndb_mix_innodb       : BUG#28123 rpl_ndb_mix_innodb.test casue slave to core on sol10-sparc-a
 rpl_ndb_ctype_ucs2_def   : BUG#27404 util thd mysql_parse sig11 when mysqld default multibyte charset
diff --git a/mysql-test/t/partition.test b/mysql-test/t/partition.test
index 42db23dadef..bf08f48c3b9 100644
--- a/mysql-test/t/partition.test
+++ b/mysql-test/t/partition.test
@@ -3,7 +3,8 @@
 # Simple test for the partition storage engine
 # Taken fromm the select test
 #
--- source include/have_partition.inc
+source include/have_partition.inc;
+source include/have_archive.inc;
 
 --disable_warnings
 drop table if exists t1;
diff --git a/sql/field.cc b/sql/field.cc
index 052f1cdf6d2..a872beab726 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -1394,20 +1394,85 @@ int Field::store(const char *to, uint length, CHARSET_INFO *cs,
 
 
 /**
+   Pack the field into a format suitable for storage and transfer.
+
+   To implement packing functionality, only the virtual function
+   should be overridden. The other functions are just convenience
+   functions and hence should not be overridden.
+
+   The value of <code>low_byte_first</code> is dependent on how the
+   packed data is going to be used: for local use, e.g., temporary
+   store on disk or in memory, use the native format since that is
+   faster. For data that is going to be transfered to other machines
+   (e.g., when writing data to the binary log), data should always be
+   stored in little-endian format.
+
+   @note The default method for packing fields just copy the raw bytes
+   of the record into the destination, but never more than
+   <code>max_length</code> characters.
+
+   @param to
+   Pointer to memory area where representation of field should be put.
+
+   @param from
+   Pointer to memory area where record representation of field is
+   stored.
+
+   @param max_length
+   Maximum length of the field, as given in the column definition. For
+   example, for <code>CHAR(1000)</code>, the <code>max_length</code>
+   is 1000. This information is sometimes needed to decide how to pack
+   the data.
+
+   @param low_byte_first
+   @c TRUE if integers should be stored little-endian, @c FALSE if
+   native format should be used. Note that for little-endian machines,
+   the value of this flag is a moot point since the native format is
+   little-endian.
+*/
+uchar *
+Field::pack(uchar *to, const uchar *from, uint max_length,
+            bool low_byte_first __attribute__((unused)))
+{
+  uint32 length= pack_length();
+  set_if_smaller(length, max_length);
+  memcpy(to, from, length);
+  return to+length;
+}
+
+/**
    Unpack a field from row data.
 
-   This method is used to unpack a field from a master whose size 
-   of the field is less than that of the slave.
-  
+   This method is used to unpack a field from a master whose size of
+   the field is less than that of the slave.
+
+   The <code>param_data</code> parameter is a two-byte integer (stored
+   in the least significant 16 bits of the unsigned integer) usually
+   consisting of two parts: the real type in the most significant byte
+   and a original pack length in the least significant byte.
+
+   The exact layout of the <code>param_data</code> field is given by
+   the <code>Table_map_log_event::save_field_metadata()</code>.
+
+   This is the default method for unpacking a field. It just copies
+   the memory block in byte order (of original pack length bytes or
+   length of field, whichever is smaller).
+
    @param   to         Destination of the data
    @param   from       Source of the data
-   @param   param_data Pack length of the field data
+   @param   param_data Real type and original pack length of the field
+                       data
+
+   @param low_byte_first
+   If this flag is @c true, all composite entities (e.g., lengths)
+   should be unpacked in little-endian format; otherwise, the entities
+   are unpacked in native order.
 
    @return  New pointer into memory based on from + length of the data
 */
-const uchar *Field::unpack(uchar* to,
-                           const uchar *from, 
-                           uint param_data)
+const uchar *
+Field::unpack(uchar* to, const uchar *from, uint param_data,
+              bool low_byte_first __attribute__((unused)))
 {
   uint length=pack_length();
   int from_type= 0;
@@ -1420,19 +1485,18 @@ const uchar *Field::unpack(uchar* to,
     from_type= (param_data & 0xff00) >> 8U;  // real_type.
     param_data= param_data & 0x00ff;        // length.
   }
+
+  if ((param_data == 0) ||
+      (length == param_data) ||
+      (from_type != real_type()))
+  {
+    memcpy(to, from, length);
+    return from+length;
+  }
+
   uint len= (param_data && (param_data < length)) ?
             param_data : length;
-  /*
-    If the length is the same, use old unpack method.
-    If the param_data is 0, use the old unpack method.
-      This is possible if the table map was generated from a down-level
-      master or if the data was not available on the master.
-    If the real_types are not the same, use the old unpack method.
-  */
-  if ((length == param_data) ||
-      (param_data == 0) ||
-      (from_type != real_type()))
-    return(unpack(to, from));
+
   memcpy(to, from, param_data > length ? length : len);
   return from+len;
 }
@@ -2805,10 +2869,15 @@ uint Field_new_decimal::is_equal(Create_field *new_field)
 
    @return  New pointer into memory based on from + length of the data
 */
-const uchar *Field_new_decimal::unpack(uchar* to, 
-                                       const uchar *from, 
-                                       uint param_data)
+const uchar *
+Field_new_decimal::unpack(uchar* to,
+                          const uchar *from,
+                          uint param_data,
+                          bool low_byte_first)
 {
+  if (param_data == 0)
+    return Field::unpack(to, from, param_data, low_byte_first);
+
   uint from_precision= (param_data & 0xff00) >> 8U;
   uint from_decimal= param_data & 0x00ff;
   uint length=pack_length();
@@ -3950,6 +4019,49 @@ void Field_longlong::sql_type(String &res) const
 }
 
 
+/*
+  Floating-point numbers
+ */
+
+uchar *
+Field_real::pack(uchar *to, const uchar *from,
+                 uint max_length, bool low_byte_first)
+{
+  DBUG_ENTER("Field_real::pack");
+  DBUG_ASSERT(max_length >= pack_length());
+  DBUG_PRINT("debug", ("pack_length(): %u", pack_length()));
+#ifdef WORDS_BIGENDIAN
+  if (low_byte_first != table->s->db_low_byte_first)
+  {
+    const uchar *dptr= from + pack_length();
+    while (dptr-- > from)
+      *to++ = *dptr;
+    DBUG_RETURN(to);
+  }
+  else
+#endif
+    DBUG_RETURN(Field::pack(to, from, max_length, low_byte_first));
+}
+
+const uchar *
+Field_real::unpack(uchar *to, const uchar *from,
+                   uint param_data, bool low_byte_first)
+{
+  DBUG_ENTER("Field_real::unpack");
+  DBUG_PRINT("debug", ("pack_length(): %u", pack_length()));
+#ifdef WORDS_BIGENDIAN
+  if (low_byte_first != table->s->db_low_byte_first)
+  {
+    const uchar *dptr= from + pack_length();
+    while (dptr-- > from)
+      *to++ = *dptr;
+    DBUG_RETURN(from + pack_length());
+  }
+  else
+#endif
+    DBUG_RETURN(Field::unpack(to, from, param_data, low_byte_first));
+}
+
 /****************************************************************************
   single precision float
 ****************************************************************************/
@@ -6337,6 +6449,11 @@ int Field_longstr::store_decimal(const my_decimal *d)
   return store(str.ptr(), str.length(), str.charset());
 }
 
+uint32 Field_longstr::max_data_length() const
+{
+  return field_length + (field_length > 255 ? 2 : 1);
+}
+
 
 double Field_string::val_real(void)
 {
@@ -6481,7 +6598,9 @@ void Field_string::sql_type(String &res) const
 }
 
 
-uchar *Field_string::pack(uchar *to, const uchar *from, uint max_length)
+uchar *Field_string::pack(uchar *to, const uchar *from,
+                          uint max_length,
+                          bool low_byte_first __attribute__((unused)))
 {
   uint length=      min(field_length,max_length);
   uint local_char_length= max_length/field_charset->mbmaxlen;
@@ -6489,11 +6608,15 @@ uchar *Field_string::pack(uchar *to, const uchar *from, uint max_length)
     local_char_length= my_charpos(field_charset, from, from+length,
                                   local_char_length);
   set_if_smaller(length, local_char_length);
-  while (length && from[length-1] == ' ')
+  while (length && from[length-1] == field_charset->pad_char)
     length--;
+
+  // Length always stored little-endian
   *to++= (uchar) length;
   if (field_length > 255)
     *to++= (uchar) (length >> 8);
+
+  // Store the actual bytes of the string
   memcpy(to, from, length);
   return to+length;
 }
@@ -6515,34 +6638,27 @@ uchar *Field_string::pack(uchar *to, const uchar *from, uint max_length)
 
    @return  New pointer into memory based on from + length of the data
 */
-const uchar *Field_string::unpack(uchar *to,
-                                  const uchar *from,
-                                  uint param_data)
-{
-  uint from_len= param_data & 0x00ff;                 // length.
-  uint length= 0;
-  uint f_length;
-  f_length= (from_len < field_length) ? from_len : field_length;
-  DBUG_ASSERT(f_length <= 255);
-  length= (uint) *from++;
-  bitmap_set_bit(table->write_set,field_index);
-  store((const char *)from, length, system_charset_info);
-  return from+length;
-}
-
-
-const uchar *Field_string::unpack(uchar *to, const uchar *from)
-{
+const uchar *
+Field_string::unpack(uchar *to,
+                     const uchar *from,
+                     uint param_data,
+                     bool low_byte_first __attribute__((unused)))
+{
+  uint from_length=
+    param_data ? min(param_data & 0x00ff, field_length) : field_length;
   uint length;
-  if (field_length > 255)
+
+  if (from_length > 255)
   {
     length= uint2korr(from);
     from+= 2;
   }
   else
     length= (uint) *from++;
-  memcpy(to, from, (int) length);
-  bfill(to+length, field_length - length, ' ');
+
+  memcpy(to, from, length);
+  // Pad the string with the pad character of the fields charset
+  bfill(to + length, field_length - length, field_charset->pad_char);
   return from+length;
 }
 
@@ -6960,22 +7076,30 @@ uint32 Field_varstring::data_length()
   Here the number of length bytes are depending on the given max_length
 */
 
-uchar *Field_varstring::pack(uchar *to, const uchar *from, uint max_length)
+uchar *Field_varstring::pack(uchar *to, const uchar *from,
+                             uint max_length,
+                             bool low_byte_first __attribute__((unused)))
 {
   uint length= length_bytes == 1 ? (uint) *from : uint2korr(from);
   set_if_smaller(max_length, field_length);
   if (length > max_length)
     length=max_length;
-  *to++= (char) (length & 255);
+
+  /* Length always stored little-endian */
+  *to++= length & 0xFF;
   if (max_length > 255)
-    *to++= (char) (length >> 8);
-  if (length)
+    *to++= (length >> 8) & 0xFF;
+
+  /* Store bytes of string */
+  if (length > 0)
     memcpy(to, from+length_bytes, length);
   return to+length;
 }
 
 
-uchar *Field_varstring::pack_key(uchar *to, const uchar *key, uint max_length)
+uchar *
+Field_varstring::pack_key(uchar *to, const uchar *key, uint max_length,
+                          bool low_byte_first __attribute__((unused)))
 {
   uint length=  length_bytes == 1 ? (uint) *key : uint2korr(key);
   uint local_char_length= ((field_charset->mbmaxlen > 1) ?
@@ -7014,8 +7138,9 @@ uchar *Field_varstring::pack_key(uchar *to, const uchar *key, uint max_length)
     Pointer to end of 'key' (To the next key part if multi-segment key)
 */
 
-const uchar *Field_varstring::unpack_key(uchar *to, const uchar *key,
-                                         uint max_length)
+const uchar *
+Field_varstring::unpack_key(uchar *to, const uchar *key, uint max_length,
+                            bool low_byte_first __attribute__((unused)))
 {
   /* get length of the blob key */
   uint32 length= *key++;
@@ -7044,8 +7169,9 @@ const uchar *Field_varstring::unpack_key(uchar *to, const uchar *key,
     end of key storage
 */
 
-uchar *Field_varstring::pack_key_from_key_image(uchar *to, const uchar *from,
-                                                uint max_length)
+uchar *
+Field_varstring::pack_key_from_key_image(uchar *to, const uchar *from, uint max_length,
+                                         bool low_byte_first __attribute__((unused)))
 {
   /* Key length is always stored as 2 bytes */
   uint length= uint2korr(from);
@@ -7065,6 +7191,9 @@ uchar *Field_varstring::pack_key_from_key_image(uchar *to, const uchar *from,
 
    This method is used to unpack a varstring field from a master
    whose size of the field is less than that of the slave.
+
+   @note
+   The string length is always packed little-endian.
   
    @param   to         Destination of the data
    @param   from       Source of the data
@@ -7072,9 +7201,10 @@ uchar *Field_varstring::pack_key_from_key_image(uchar *to, const uchar *from,
 
    @return  New pointer into memory based on from + length of the data
 */
-const uchar *Field_varstring::unpack(uchar *to, 
-                                     const uchar *from,
-                                     uint param_data)
+const uchar *
+Field_varstring::unpack(uchar *to, const uchar *from,
+                        uint param_data,
+                        bool low_byte_first __attribute__((unused)))
 {
   uint length;
   uint l_bytes= (param_data && (param_data < field_length)) ? 
@@ -7086,28 +7216,7 @@ const uchar *Field_varstring::unpack(uchar *to,
     if (length_bytes == 2)
       to[1]= 0;
   }
-  else
-  {
-    length= uint2korr(from);
-    to[0]= *from++;
-    to[1]= *from++;
-  }
-  if (length)
-    memcpy(to+ length_bytes, from, length);
-  return from+length;
-}
-
-
-/*
-  unpack field packed with Field_varstring::pack()
-*/
-
-const uchar *Field_varstring::unpack(uchar *to, const uchar *from)
-{
-  uint length;
-  if (length_bytes == 1)
-    length= (uint) (*to= *from++);
-  else
+  else /* l_bytes == 2 */
   {
     length= uint2korr(from);
     to[0]= *from++;
@@ -7356,9 +7465,9 @@ void Field_blob::store_length(uchar *i_ptr,
 }
 
 
-uint32 Field_blob::get_length(const uchar *pos, bool low_byte_first)
+uint32 Field_blob::get_length(const uchar *pos, uint packlength_arg, bool low_byte_first)
 {
-  switch (packlength) {
+  switch (packlength_arg) {
   case 1:
     return (uint32) pos[0];
   case 2:
@@ -7789,26 +7898,37 @@ void Field_blob::sql_type(String &res) const
   }
 }
 
-
-uchar *Field_blob::pack(uchar *to, const uchar *from, uint max_length)
+uchar *Field_blob::pack(uchar *to, const uchar *from,
+                        uint max_length, bool low_byte_first)
 {
+  DBUG_ENTER("Field_blob::pack");
+  DBUG_PRINT("enter", ("to: 0x%lx; from: 0x%lx;"
+                       " max_length: %u; low_byte_first: %d",
+                       (ulong) to, (ulong) from,
+                       max_length, low_byte_first));
+  DBUG_DUMP("record", from, table->s->reclength);
   uchar *save= ptr;
   ptr= (uchar*) from;
   uint32 length=get_length();			// Length of from string
-  if (length > max_length)
-  {
-    length=max_length;
-    store_length(to,packlength,length,TRUE);
-  }
-  else
-    memcpy(to,from,packlength);			// Copy length
-  if (length)
+
+  /*
+    Store max length, which will occupy packlength bytes. If the max
+    length given is smaller than the actual length of the blob, we
+    just store the initial bytes of the blob.
+  */
+  store_length(to, packlength, min(length, max_length), low_byte_first);
+
+  /*
+    Store the actual blob data, which will occupy 'length' bytes.
+   */
+  if (length > 0)
   {
     get_ptr((uchar**) &from);
     memcpy(to+packlength, from,length);
   }
   ptr=save;					// Restore org row pointer
-  return to+packlength+length;
+  DBUG_DUMP("packed", to, packlength + length);
+  DBUG_RETURN(to+packlength+length);
 }
 
 
@@ -7823,28 +7943,29 @@ uchar *Field_blob::pack(uchar *to, const uchar *from, uint max_length)
 
    @param   to         Destination of the data
    @param   from       Source of the data
-   @param   param_data not used
+   @param   param_data @c TRUE if base types should be stored in little-
+                       endian format, @c FALSE if native format should
+                       be used.
 
    @return  New pointer into memory based on from + length of the data
 */
 const uchar *Field_blob::unpack(uchar *to, 
                                 const uchar *from,
-                                uint param_data)
-{
-  return unpack(to, from);
-}
-
-
-const uchar *Field_blob::unpack(uchar *to, const uchar *from)
-{
-  uint32 length=get_length(from);
-  memcpy(to,from,packlength);
-  from+=packlength;
-  if (length)
-    memcpy_fixed(to+packlength, &from, sizeof(from));
-  else
-    bzero(to+packlength,sizeof(from));
-  return from+length;
+                                uint param_data,
+                                bool low_byte_first)
+{
+  DBUG_ENTER("Field_blob::unpack");
+  DBUG_PRINT("enter", ("to: 0x%lx; from: 0x%lx;"
+                       " param_data: %u; low_byte_first: %d",
+                       (ulong) to, (ulong) from, param_data, low_byte_first));
+  uint const master_packlength=
+    param_data > 0 ? param_data & 0xFF : packlength;
+  uint32 const length= get_length(from, master_packlength, low_byte_first);
+  DBUG_DUMP("packed", from, length + master_packlength);
+  store(reinterpret_cast<const char*>(from) + master_packlength,
+        length, field_charset);
+  DBUG_DUMP("record", to, table->s->reclength);
+  DBUG_RETURN(from + master_packlength + length);
 }
 
 /* Keys for blobs are like keys on varchars */
@@ -7894,7 +8015,9 @@ int Field_blob::pack_cmp(const uchar *b, uint key_length_arg,
 
 /* Create a packed key that will be used for storage from a MySQL row */
 
-uchar *Field_blob::pack_key(uchar *to, const uchar *from, uint max_length)
+uchar *
+Field_blob::pack_key(uchar *to, const uchar *from, uint max_length,
+                     bool low_byte_first __attribute__((unused)))
 {
   uchar *save= ptr;
   ptr= (uchar*) from;
@@ -7939,8 +8062,9 @@ uchar *Field_blob::pack_key(uchar *to, const uchar *from, uint max_length)
     Pointer into 'from' past the last byte copied from packed key.
 */
 
-const uchar *Field_blob::unpack_key(uchar *to, const uchar *from,
-                                    uint max_length)
+const uchar *
+Field_blob::unpack_key(uchar *to, const uchar *from, uint max_length,
+                       bool low_byte_first __attribute__((unused)))
 {
   /* get length of the blob key */
   uint32 length= *from++;
@@ -7963,8 +8087,9 @@ const uchar *Field_blob::unpack_key(uchar *to, const uchar *from,
 
 /* Create a packed key that will be used for storage from a MySQL key */
 
-uchar *Field_blob::pack_key_from_key_image(uchar *to, const uchar *from,
-                                           uint max_length)
+uchar *
+Field_blob::pack_key_from_key_image(uchar *to, const uchar *from, uint max_length,
+                                    bool low_byte_first __attribute__((unused)))
 {
   uint length=uint2korr(from);
   if (length > max_length)
@@ -8941,9 +9066,11 @@ void Field_bit::sql_type(String &res) const
 }
 
 
-uchar *Field_bit::pack(uchar *to, const uchar *from, uint max_length)
+uchar *
+Field_bit::pack(uchar *to, const uchar *from, uint max_length,
+                bool low_byte_first __attribute__((unused)))
 {
-  DBUG_ASSERT(max_length);
+  DBUG_ASSERT(max_length > 0);
   uint length;
   if (bit_len > 0)
   {
@@ -8978,28 +9105,44 @@ uchar *Field_bit::pack(uchar *to, const uchar *from, uint max_length)
 /**
    Unpack a bit field from row data.
 
-   This method is used to unpack a bit field from a master whose size 
+   This method is used to unpack a bit field from a master whose size
    of the field is less than that of the slave.
-  
+
    @param   to         Destination of the data
    @param   from       Source of the data
    @param   param_data Bit length (upper) and length (lower) values
 
    @return  New pointer into memory based on from + length of the data
 */
-const uchar *Field_bit::unpack(uchar *to,
-                               const uchar *from,
-                               uint param_data)
+const uchar *
+Field_bit::unpack(uchar *to, const uchar *from, uint param_data,
+                  bool low_byte_first __attribute__((unused)))
 {
   uint const from_len= (param_data >> 8U) & 0x00ff;
   uint const from_bit_len= param_data & 0x00ff;
   /*
-    If the master and slave have the same sizes, then use the old
-    unpack() method.
+    If the parameter data is zero (i.e., undefined), or if the master
+    and slave have the same sizes, then use the old unpack() method.
   */
-  if ((from_bit_len == bit_len) &&
-      (from_len == bytes_in_rec)) 
-    return(unpack(to, from));
+  if (param_data == 0 ||
+      (from_bit_len == bit_len) && (from_len == bytes_in_rec))
+  {
+    if (bit_len > 0)
+    {
+      /*
+        set_rec_bits is a macro, don't put the post-increment in the
+        argument since that might cause strange side-effects.
+
+        For the choice of the second argument, see the explanation for
+        Field_bit::pack().
+      */
+      set_rec_bits(*from, bit_ptr + (to - ptr), bit_ofs, bit_len);
+      from++;
+    }
+    memcpy(to, from, bytes_in_rec);
+    return from + bytes_in_rec;
+  }
+
   /*
     We are converting a smaller bit field to a larger one here.
     To do that, we first need to construct a raw value for the original
@@ -9027,25 +9170,6 @@ const uchar *Field_bit::unpack(uchar *to,
 }
 
 
-const uchar *Field_bit::unpack(uchar *to, const uchar *from)
-{
-  if (bit_len > 0)
-  {
-    /*
-      set_rec_bits is a macro, don't put the post-increment in the
-      argument since that might cause strange side-effects.
-
-      For the choice of the second argument, see the explanation for
-      Field_bit::pack().
-    */
-    set_rec_bits(*from, bit_ptr + (to - ptr), bit_ofs, bit_len);
-    from++;
-  }
-  memcpy(to, from, bytes_in_rec);
-  return from + bytes_in_rec;
-}
-
-
 void Field_bit::set_default()
 {
   if (bit_len > 0)
diff --git a/sql/field.h b/sql/field.h
index 8aad6783291..381721aa3a8 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -175,6 +175,17 @@ public:
   */
   virtual uint32 data_length() { return pack_length(); }
   virtual uint32 sort_length() const { return pack_length(); }
+
+  /**
+     Get the maximum size of the data in packed format.
+
+     @return Maximum data length of the field when packed using the
+     Field::pack() function.
+   */
+  virtual uint32 max_data_length() const {
+    return pack_length();
+  };
+
   virtual int reset(void) { bzero(ptr,pack_length()); return 0; }
   virtual void reset_fields() {}
   virtual void set_default()
@@ -357,32 +368,45 @@ public:
     return str;
   }
   virtual bool send_binary(Protocol *protocol);
-  virtual uchar *pack(uchar *to, const uchar *from, uint max_length=~(uint) 0)
+
+  virtual uchar *pack(uchar *to, const uchar *from,
+                      uint max_length, bool low_byte_first);
+  /**
+     @overload Field::pack(uchar*, const uchar*, uint, bool)
+  */
+  uchar *pack(uchar *to, const uchar *from)
   {
-    uint32 length=pack_length();
-    memcpy(to,from,length);
-    return to+length;
+    DBUG_ENTER("Field::pack");
+    uchar *result= this->pack(to, from, UINT_MAX, table->s->db_low_byte_first);
+    DBUG_RETURN(result);
   }
-  virtual const uchar *unpack(uchar* to, const uchar *from, uint param_data);
-  virtual const uchar *unpack(uchar* to, const uchar *from)
+
+  virtual const uchar *unpack(uchar* to, const uchar *from,
+                              uint param_data, bool low_byte_first);
+  /**
+     @overload Field::unpack(uchar*, const uchar*, uint, bool)
+  */
+  const uchar *unpack(uchar* to, const uchar *from)
   {
-    uint length=pack_length();
-    memcpy(to,from,length);
-    return from+length;
+    DBUG_ENTER("Field::unpack");
+    const uchar *result= unpack(to, from, 0U, table->s->db_low_byte_first);
+    DBUG_RETURN(result);
   }
-  virtual uchar *pack_key(uchar* to, const uchar *from, uint max_length)
+
+  virtual uchar *pack_key(uchar* to, const uchar *from,
+                          uint max_length, bool low_byte_first)
   {
-    return pack(to,from,max_length);
+    return pack(to, from, max_length, low_byte_first);
   }
   virtual uchar *pack_key_from_key_image(uchar* to, const uchar *from,
-					uint max_length)
+					uint max_length, bool low_byte_first)
   {
-    return pack(to,from,max_length);
+    return pack(to, from, max_length, low_byte_first);
   }
   virtual const uchar *unpack_key(uchar* to, const uchar *from,
-                                  uint max_length)
+                                  uint max_length, bool low_byte_first)
   {
-    return unpack(to,from);
+    return unpack(to, from, max_length, low_byte_first);
   }
   virtual uint packed_col_length(const uchar *to, uint length)
   { return length;}
@@ -567,6 +591,7 @@ public:
     {}
 
   int store_decimal(const my_decimal *d);
+  uint32 max_data_length() const;
 };
 
 /* base class for float and double and decimal (old one) */
@@ -587,6 +612,10 @@ public:
   int truncate(double *nr, double max_length);
   uint32 max_display_length() { return field_length; }
   uint size_of() const { return sizeof(*this); }
+  virtual const uchar *unpack(uchar* to, const uchar *from,
+                              uint param_data, bool low_byte_first);
+  virtual uchar *pack(uchar* to, const uchar *from,
+                      uint max_length, bool low_byte_first);
 };
 
 
@@ -615,6 +644,16 @@ public:
   void overflow(bool negative);
   bool zero_pack() const { return 0; }
   void sql_type(String &str) const;
+  virtual const uchar *unpack(uchar* to, const uchar *from,
+                              uint param_data, bool low_byte_first)
+  {
+    return Field::unpack(to, from, param_data, low_byte_first);
+  }
+  virtual uchar *pack(uchar* to, const uchar *from,
+                      uint max_length, bool low_byte_first)
+  {
+    return Field::pack(to, from, max_length, low_byte_first);
+  }
 };
 
 
@@ -665,7 +704,8 @@ public:
   uint row_pack_length() { return pack_length(); }
   int compatible_field_size(uint field_metadata);
   uint is_equal(Create_field *new_field);
-  virtual const uchar *unpack(uchar* to, const uchar *from, uint param_data);
+  virtual const uchar *unpack(uchar* to, const uchar *from,
+                              uint param_data, bool low_byte_first);
 };
 
 
@@ -696,6 +736,20 @@ public:
   uint32 pack_length() const { return 1; }
   void sql_type(String &str) const;
   uint32 max_display_length() { return 4; }
+
+  virtual uchar *pack(uchar* to, const uchar *from,
+                      uint max_length, bool low_byte_first)
+  {
+    *to= *from;
+    return to + 1;
+  }
+
+  virtual const uchar *unpack(uchar* to, const uchar *from,
+                              uint param_data, bool low_byte_first)
+  {
+    *to= *from;
+    return from + 1;
+  }
 };
 
 
@@ -731,8 +785,47 @@ public:
   uint32 pack_length() const { return 2; }
   void sql_type(String &str) const;
   uint32 max_display_length() { return 6; }
-};
 
+  virtual uchar *pack(uchar* to, const uchar *from,
+                      uint max_length, bool low_byte_first)
+  {
+    int16 val;
+#ifdef WORDS_BIGENDIAN
+    if (table->s->db_low_byte_first)
+      val = sint2korr(from);
+    else
+#endif
+      shortget(val, from);
+
+#ifdef WORDS_BIGENDIAN
+    if (low_byte_first)
+      int2store(to, val);
+    else
+#endif
+      shortstore(to, val);
+    return to + sizeof(val);
+  }
+
+  virtual const uchar *unpack(uchar* to, const uchar *from,
+                              uint param_data, bool low_byte_first)
+  {
+    int16 val;
+#ifdef WORDS_BIGENDIAN
+    if (low_byte_first)
+      val = sint2korr(from);
+    else
+#endif
+      shortget(val, from);
+
+#ifdef WORDS_BIGENDIAN
+    if (table->s->db_low_byte_first)
+      int2store(to, val);
+    else
+#endif
+      shortstore(to, val);
+    return from + sizeof(val);
+  }
+};
 
 class Field_medium :public Field_num {
 public:
@@ -761,6 +854,18 @@ public:
   uint32 pack_length() const { return 3; }
   void sql_type(String &str) const;
   uint32 max_display_length() { return 8; }
+
+  virtual uchar *pack(uchar* to, const uchar *from,
+                      uint max_length, bool low_byte_first)
+  {
+    return Field::pack(to, from, max_length, low_byte_first);
+  }
+
+  virtual const uchar *unpack(uchar* to, const uchar *from,
+                              uint param_data, bool low_byte_first)
+  {
+    return Field::unpack(to, from, param_data, low_byte_first);
+  }
 };
 
 
@@ -796,6 +901,45 @@ public:
   uint32 pack_length() const { return 4; }
   void sql_type(String &str) const;
   uint32 max_display_length() { return MY_INT32_NUM_DECIMAL_DIGITS; }
+  virtual uchar *pack(uchar* to, const uchar *from,
+                      uint max_length, bool low_byte_first)
+  {
+    int32 val;
+#ifdef WORDS_BIGENDIAN
+    if (table->s->db_low_byte_first)
+      val = sint4korr(from);
+    else
+#endif
+      longget(val, from);
+
+#ifdef WORDS_BIGENDIAN
+    if (low_byte_first)
+      int4store(to, val);
+    else
+#endif
+      longstore(to, val);
+    return to + sizeof(val);
+  }
+
+  virtual const uchar *unpack(uchar* to, const uchar *from,
+                              uint param_data, bool low_byte_first)
+  {
+    int32 val;
+#ifdef WORDS_BIGENDIAN
+    if (low_byte_first)
+      val = sint4korr(from);
+    else
+#endif
+      longget(val, from);
+
+#ifdef WORDS_BIGENDIAN
+    if (table->s->db_low_byte_first)
+      int4store(to, val);
+    else
+#endif
+      longstore(to, val);
+    return from + sizeof(val);
+  }
 };
 
 
@@ -838,6 +982,45 @@ public:
   void sql_type(String &str) const;
   bool can_be_compared_as_longlong() const { return TRUE; }
   uint32 max_display_length() { return 20; }
+  virtual uchar *pack(uchar* to, const uchar *from,
+                      uint max_length, bool low_byte_first)
+  {
+    int64 val;
+#ifdef WORDS_BIGENDIAN
+    if (table->s->db_low_byte_first)
+      val = sint8korr(from);
+    else
+#endif
+      longlongget(val, from);
+
+#ifdef WORDS_BIGENDIAN
+    if (low_byte_first)
+      int8store(to, val);
+    else
+#endif
+      longlongstore(to, val);
+    return to + sizeof(val);
+  }
+
+  virtual const uchar *unpack(uchar* to, const uchar *from,
+                              uint param_data, bool low_byte_first)
+  {
+    int64 val;
+#ifdef WORDS_BIGENDIAN
+    if (low_byte_first)
+      val = sint8korr(from);
+    else
+#endif
+      longlongget(val, from);
+
+#ifdef WORDS_BIGENDIAN
+    if (table->s->db_low_byte_first)
+      int8store(to, val);
+    else
+#endif
+      longlongstore(to, val);
+    return from + sizeof(val);
+  }
 };
 #endif
 
@@ -1218,9 +1401,10 @@ public:
   int cmp(const uchar *,const uchar *);
   void sort_string(uchar *buff,uint length);
   void sql_type(String &str) const;
-  uchar *pack(uchar *to, const uchar *from, uint max_length=~(uint) 0);
-  virtual const uchar *unpack(uchar* to, const uchar *from, uint param_data);
-  const uchar *unpack(uchar* to, const uchar *from);
+  virtual uchar *pack(uchar *to, const uchar *from,
+                      uint max_length, bool low_byte_first);
+  virtual const uchar *unpack(uchar* to, const uchar *from,
+                              uint param_data, bool low_byte_first);
   uint pack_length_from_metadata(uint field_metadata)
   { return (field_metadata & 0x00ff); }
   uint row_pack_length() { return (field_length + 1); }
@@ -1298,13 +1482,15 @@ public:
   uint get_key_image(uchar *buff,uint length, imagetype type);
   void set_key_image(const uchar *buff,uint length);
   void sql_type(String &str) const;
-  uchar *pack(uchar *to, const uchar *from, uint max_length=~(uint) 0);
-  uchar *pack_key(uchar *to, const uchar *from, uint max_length);
+  virtual uchar *pack(uchar *to, const uchar *from,
+                      uint max_length, bool low_byte_first);
+  uchar *pack_key(uchar *to, const uchar *from, uint max_length, bool low_byte_first);
   uchar *pack_key_from_key_image(uchar* to, const uchar *from,
-                                 uint max_length);
-  virtual const uchar *unpack(uchar* to, const uchar *from, uint param_data);
-  const uchar *unpack(uchar* to, const uchar *from);
-  const uchar *unpack_key(uchar* to, const uchar *from, uint max_length);
+                                 uint max_length, bool low_byte_first);
+  virtual const uchar *unpack(uchar* to, const uchar *from,
+                              uint param_data, bool low_byte_first);
+  const uchar *unpack_key(uchar* to, const uchar *from,
+                          uint max_length, bool low_byte_first);
   int pack_cmp(const uchar *a, const uchar *b, uint key_length,
                my_bool insert_or_update);
   int pack_cmp(const uchar *b, uint key_length,my_bool insert_or_update);
@@ -1397,7 +1583,7 @@ public:
   { return (uint32) (packlength); }
   uint row_pack_length() { return pack_length_no_ptr(); }
   uint32 sort_length() const;
-  inline uint32 max_data_length() const
+  virtual uint32 max_data_length() const
   {
     return (uint32) (((ulonglong) 1 << (packlength*8)) -1);
   }
@@ -1425,13 +1611,13 @@ public:
      @returns The length in the row plus the size of the data.
   */
   uint32 get_packed_size(const uchar *ptr_arg, bool low_byte_first)
-    {return packlength + get_length(ptr_arg, low_byte_first);}
+    {return packlength + get_length(ptr_arg, packlength, low_byte_first);}
 
   inline uint32 get_length(uint row_offset= 0)
-  { return get_length(ptr+row_offset, table->s->db_low_byte_first); }
-  uint32 get_length(const uchar *ptr, bool low_byte_first);
+  { return get_length(ptr+row_offset, this->packlength, table->s->db_low_byte_first); }
+  uint32 get_length(const uchar *ptr, uint packlength, bool low_byte_first);
   uint32 get_length(const uchar *ptr_arg)
-  { return get_length(ptr_arg, table->s->db_low_byte_first); }
+  { return get_length(ptr_arg, this->packlength, table->s->db_low_byte_first); }
   void put_length(uchar *pos, uint32 length);
   inline void get_ptr(uchar **str)
     {
@@ -1472,13 +1658,16 @@ public:
     memcpy_fixed(ptr+packlength,&tmp,sizeof(char*));
     return 0;
   }
-  uchar *pack(uchar *to, const uchar *from, uint max_length= ~(uint) 0);
-  uchar *pack_key(uchar *to, const uchar *from, uint max_length);
+  virtual uchar *pack(uchar *to, const uchar *from,
+                      uint max_length, bool low_byte_first);
+  uchar *pack_key(uchar *to, const uchar *from,
+                  uint max_length, bool low_byte_first);
   uchar *pack_key_from_key_image(uchar* to, const uchar *from,
-                                 uint max_length);
-  virtual const uchar *unpack(uchar *to, const uchar *from, uint param_data);
-  const uchar *unpack(uchar *to, const uchar *from);
-  const uchar *unpack_key(uchar* to, const uchar *from, uint max_length);
+                                 uint max_length, bool low_byte_first);
+  virtual const uchar *unpack(uchar *to, const uchar *from,
+                              uint param_data, bool low_byte_first);
+  const uchar *unpack_key(uchar* to, const uchar *from,
+                          uint max_length, bool low_byte_first);
   int pack_cmp(const uchar *a, const uchar *b, uint key_length,
                my_bool insert_or_update);
   int pack_cmp(const uchar *b, uint key_length,my_bool insert_or_update);
@@ -1631,6 +1820,7 @@ public:
   enum_field_types type() const { return MYSQL_TYPE_BIT; }
   enum ha_base_keytype key_type() const { return HA_KEYTYPE_BIT; }
   uint32 key_length() const { return (uint32) (field_length + 7) / 8; }
+  uint32 max_data_length() const { return (field_length + 7) / 8; }
   uint32 max_display_length() { return field_length; }
   uint size_of() const { return sizeof(*this); }
   Item_result result_type () const { return INT_RESULT; }
@@ -1672,9 +1862,10 @@ public:
   { return (bytes_in_rec + ((bit_len > 0) ? 1 : 0)); }
   int compatible_field_size(uint field_metadata);
   void sql_type(String &str) const;
-  uchar *pack(uchar *to, const uchar *from, uint max_length=~(uint) 0);
-  virtual const uchar *unpack(uchar *to, const uchar *from, uint param_data);
-  const uchar *unpack(uchar* to, const uchar *from);
+  virtual uchar *pack(uchar *to, const uchar *from,
+                      uint max_length, bool low_byte_first);
+  virtual const uchar *unpack(uchar *to, const uchar *from,
+                              uint param_data, bool low_byte_first);
   virtual void set_default();
 
   Field *new_key_field(MEM_ROOT *root, struct st_table *new_table,
diff --git a/sql/log.cc b/sql/log.cc
index 95204e89d0e..2354c5829bf 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -3577,9 +3577,6 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info)
 	(!binlog_filter->db_ok(local_db)))
     {
       VOID(pthread_mutex_unlock(&LOCK_log));
-      DBUG_PRINT("info",("OPTION_BIN_LOG is %s, db_ok('%s') == %d",
-                         (thd->options & OPTION_BIN_LOG) ? "set" : "clear",
-                         local_db, binlog_filter->db_ok(local_db)));
       DBUG_RETURN(0);
     }
 #endif /* HAVE_REPLICATION */
diff --git a/sql/rpl_record.cc b/sql/rpl_record.cc
index 65c8e106112..71cc431890b 100644
--- a/sql/rpl_record.cc
+++ b/sql/rpl_record.cc
@@ -65,6 +65,8 @@ pack_row(TABLE *table, MY_BITMAP const* cols,
   my_ptrdiff_t const rec_offset= record - table->record[0];
   my_ptrdiff_t const def_offset= table->s->default_values - table->record[0];
 
+  DBUG_ENTER("pack_row");
+
   /*
     We write the null bits and the packed records using one pass
     through all the fields. The null bytes are written little-endian,
@@ -96,26 +98,14 @@ pack_row(TABLE *table, MY_BITMAP const* cols,
           For big-endian machines, we have to make sure that the
           length is stored in little-endian format, since this is the
           format used for the binlog.
-
-          We do this by setting the db_low_byte_first, which is used
-          inside some store_length() to decide what order to write the
-          bytes in.
-
-          In reality, db_log_byte_first is only set for legacy table
-          type Isam, but in the event of a bug, we need to guarantee
-          the endianess when writing to the binlog.
-
-          This is currently broken for NDB due to BUG#29549, so we
-          will fix it when NDB has fixed their way of handling BLOBs.
         */
-#if 0
-        bool save= table->s->db_low_byte_first;
-        table->s->db_low_byte_first= TRUE;
-#endif
-        pack_ptr= field->pack(pack_ptr, field->ptr + offset);
-#if 0
-        table->s->db_low_byte_first= save;
-#endif
+        const uchar *old_pack_ptr= pack_ptr;
+        pack_ptr= field->pack(pack_ptr, field->ptr + offset,
+                              field->max_data_length(), TRUE);
+        DBUG_PRINT("debug", ("field: %s; pack_ptr: 0x%lx;"
+                             " pack_ptr':0x%lx; bytes: %d",
+                             field->field_name, (ulong) old_pack_ptr,
+                             (ulong) pack_ptr, pack_ptr - old_pack_ptr));
       }
 
       null_mask <<= 1;
@@ -143,8 +133,8 @@ pack_row(TABLE *table, MY_BITMAP const* cols,
     packed data. If it doesn't, something is very wrong.
   */
   DBUG_ASSERT(null_ptr == row_data + null_byte_count);
-
-  return static_cast<size_t>(pack_ptr - row_data);
+  DBUG_DUMP("row_data", row_data, pack_ptr - row_data);
+  DBUG_RETURN(static_cast<size_t>(pack_ptr - row_data));
 }
 #endif
 
@@ -242,18 +232,14 @@ unpack_row(Relay_log_info const *rli,
           Use the master's size information if available else call
           normal unpack operation.
         */
-#if 0
-        bool save= table->s->db_low_byte_first;
-        table->s->db_low_byte_first= TRUE;
-#endif
         uint16 const metadata= tabledef->field_metadata(i);
-        if (tabledef && metadata)
-          pack_ptr= f->unpack(f->ptr, pack_ptr, metadata);
-        else
-          pack_ptr= f->unpack(f->ptr, pack_ptr);
-#if 0
-        table->s->db_low_byte_first= save;
-#endif
+        uchar const *const old_pack_ptr= pack_ptr;
+        pack_ptr= f->unpack(f->ptr, pack_ptr, metadata, TRUE);
+	DBUG_PRINT("debug", ("field: %s; metadata: 0x%x;"
+                             " pack_ptr: 0x%lx; pack_ptr': 0x%lx; bytes: %d",
+                             f->field_name, metadata,
+                             (ulong) old_pack_ptr, (ulong) pack_ptr,
+                             pack_ptr - old_pack_ptr));
       }
 
       null_mask <<= 1;
@@ -289,6 +275,8 @@ unpack_row(Relay_log_info const *rli,
    */
   DBUG_ASSERT(null_ptr == row_data + master_null_byte_count);
 
+  DBUG_DUMP("row_data", row_data, pack_ptr - row_data);
+
   *row_end = pack_ptr;
   if (master_reclength)
   {
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index f7813b73089..a89777a5e23 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -29,6 +29,8 @@
 #include "event_data_objects.h"
 #include <my_dir.h>
 
+#define STR_OR_NIL(S) ((S) ? (S) : "<nil>")
+
 #ifdef WITH_PARTITION_STORAGE_ENGINE
 #include "ha_partition.h"
 #endif
@@ -3115,8 +3117,8 @@ int get_all_tables(THD *thd, TABLE_LIST *tables, COND *cond)
   schema_table_idx= get_schema_table_idx(schema_table);
   get_lookup_field_values(thd, cond, tables, &lookup_field_vals);
   DBUG_PRINT("INDEX VALUES",("db_name='%s', table_name='%s'",
-                             lookup_field_vals.db_value.str,
-                             lookup_field_vals.table_value.str));
+                             STR_OR_NIL(lookup_field_vals.db_value.str),
+                             STR_OR_NIL(lookup_field_vals.table_value.str)));
 
   if (!lookup_field_vals.wild_db_value && !lookup_field_vals.wild_table_value)
   {