1 files changed, 119 insertions, 57 deletions

diff --git a/sql/rpl_utility.cc b/sql/rpl_utility.cc
index 1d7cc808f0c..d1ce5bf3b7b 100644
--- a/sql/rpl_utility.cc
+++ b/sql/rpl_utility.cc
@@ -14,18 +14,51 @@
    Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA */
 
 #include "rpl_utility.h"
+#include "rpl_rli.h"
 
-uint32
-field_length_from_packed(enum_field_types const field_type, 
-                         byte const *const data)
+/*********************************************************************
+ *                   table_def member definitions                    *
+ *********************************************************************/
+
+/*
+  This function returns the field size in raw bytes based on the type
+  and the encoded field data from the master's raw data.
+*/
+uint32 table_def::calc_field_size(uint col, uchar *master_data) const
 {
   uint32 length;
 
-  switch (field_type) {
-  case MYSQL_TYPE_DECIMAL:
+  switch (type(col)) {
   case MYSQL_TYPE_NEWDECIMAL:
-    length= ~(uint32) 0;
+    length= my_decimal_get_binary_size(m_field_metadata[col] >> 8, 
+             m_field_metadata[col] - ((m_field_metadata[col] >> 8) << 8));
+    break;
+  case MYSQL_TYPE_DECIMAL:
+  case MYSQL_TYPE_FLOAT:
+  case MYSQL_TYPE_DOUBLE:
+    length= m_field_metadata[col];
+    break;
+  case MYSQL_TYPE_SET:
+  case MYSQL_TYPE_ENUM:
+  case MYSQL_TYPE_STRING:
+  {
+    if (((m_field_metadata[col] & 0xff00) == (MYSQL_TYPE_SET << 8)) ||
+        ((m_field_metadata[col] & 0xff00) == (MYSQL_TYPE_ENUM << 8)))
+      length= m_field_metadata[col] & 0x00ff;
+    else
+    {
+      length= m_field_metadata[col] & 0x00ff;
+      DBUG_ASSERT(length > 0);
+      if (length > 255)
+      {
+        DBUG_ASSERT(uint2korr(master_data) > 0);
+        length= uint2korr(master_data) + 2;
+      }
+      else
+        length= (uint) *master_data + 1;
+    }
     break;
+  }
   case MYSQL_TYPE_YEAR:
   case MYSQL_TYPE_TINY:
     length= 1;
@@ -44,12 +77,6 @@ field_length_from_packed(enum_field_types const field_type,
     length= 8;
     break;
 #endif
-  case MYSQL_TYPE_FLOAT:
-    length= sizeof(float);
-    break;
-  case MYSQL_TYPE_DOUBLE:
-    length= sizeof(double);
-    break;
   case MYSQL_TYPE_NULL:
     length= 0;
     break;
@@ -57,8 +84,6 @@ field_length_from_packed(enum_field_types const field_type,
     length= 3;
     break;
   case MYSQL_TYPE_DATE:
-    length= 4;
-    break;
   case MYSQL_TYPE_TIME:
     length= 3;
     break;
@@ -68,47 +93,78 @@ field_length_from_packed(enum_field_types const field_type,
   case MYSQL_TYPE_DATETIME:
     length= 8;
     break;
-    break;
   case MYSQL_TYPE_BIT:
-    length= ~(uint32) 0;
-    break;
-  default:
-    /* This case should never be chosen */
-    DBUG_ASSERT(0);
-    /* If something goes awfully wrong, it's better to get a string than die */
-  case MYSQL_TYPE_STRING:
-    length= uint2korr(data);
+  {
+    uint from_len= (m_field_metadata[col] >> 8U) & 0x00ff;
+    uint from_bit_len= m_field_metadata[col] & 0x00ff;
+    DBUG_ASSERT(from_bit_len <= 7);
+    length= from_len + ((from_bit_len > 0) ? 1 : 0);
     break;
-
-  case MYSQL_TYPE_ENUM:
-  case MYSQL_TYPE_SET:
-  case MYSQL_TYPE_VAR_STRING:
+  }
   case MYSQL_TYPE_VARCHAR:
-    length= ~(uint32) 0;                               // NYI
+  {
+    length= m_field_metadata[col] > 255 ? 2 : 1; // c&p of Field_varstring::data_length()
+    DBUG_ASSERT(uint2korr(master_data) > 0);
+    length+= length == 1 ? (uint32) *master_data : uint2korr(master_data);
     break;
-
+  }
   case MYSQL_TYPE_TINY_BLOB:
   case MYSQL_TYPE_MEDIUM_BLOB:
   case MYSQL_TYPE_LONG_BLOB:
   case MYSQL_TYPE_BLOB:
   case MYSQL_TYPE_GEOMETRY:
-    length= ~(uint32) 0;                               // NYI
+  {
+#if 1
+    /*
+      BUG#29549: 
+      This is currently broken for NDB, which is using big-endian
+      order when packing length of BLOB. Once they have decided how to
+      fix the issue, we can enable the code below to make sure to
+      always read the length in little-endian order.
+    */
+    Field_blob fb(m_field_metadata[col]);
+    length= fb.get_packed_size(master_data, TRUE);
+#else
+    /*
+      Compute the length of the data. We cannot use get_length() here
+      since it is dependent on the specific table (and also checks the
+      packlength using the internal 'table' pointer) and replication
+      is using a fixed format for storing data in the binlog.
+    */
+    switch (m_field_metadata[col]) {
+    case 1:
+      length= *master_data;
+      break;
+    case 2:
+      length= sint2korr(master_data);
+      break;
+    case 3:
+      length= uint3korr(master_data);
+      break;
+    case 4:
+      length= uint4korr(master_data);
+      break;
+    default:
+      DBUG_ASSERT(0);		// Should not come here
+      break;
+    }
+
+    length+= m_field_metadata[col];
+#endif
     break;
   }
-
+  default:
+    length= -1;
+  }
   return length;
 }
 
-/*********************************************************************
- *                   table_def member definitions                    *
- *********************************************************************/
-
 /*
   Is the definition compatible with a table?
 
 */
 int
-table_def::compatible_with(RELAY_LOG_INFO const *rli_arg, TABLE *table)
+table_def::compatible_with(Relay_log_info const *rli_arg, TABLE *table)
   const
 {
   /*
@@ -116,25 +172,10 @@ table_def::compatible_with(RELAY_LOG_INFO const *rli_arg, TABLE *table)
   */
   uint const cols_to_check= min(table->s->fields, size());
   int error= 0;
-  RELAY_LOG_INFO const *rli= const_cast<RELAY_LOG_INFO*>(rli_arg);
+  Relay_log_info const *rli= const_cast<Relay_log_info*>(rli_arg);
 
   TABLE_SHARE const *const tsh= table->s;
 
-  /*
-    To get proper error reporting for all columns of the table, we
-    both check the width and iterate over all columns.
-  */
-  if (tsh->fields < size())
-  {
-    DBUG_ASSERT(tsh->db.str && tsh->table_name.str);
-    error= 1;
-    slave_print_msg(ERROR_LEVEL, rli, ER_BINLOG_ROW_WRONG_TABLE_DEF,
-                    "Table width mismatch - "
-                    "received %u columns, %s.%s has %u columns",
-                    (uint) size(), tsh->db.str, tsh->table_name.str,
-                    tsh->fields);
-  }
-
   for (uint col= 0 ; col < cols_to_check ; ++col)
   {
     if (table->field[col]->type() != type(col))
@@ -142,11 +183,32 @@ table_def::compatible_with(RELAY_LOG_INFO const *rli_arg, TABLE *table)
       DBUG_ASSERT(col < size() && col < tsh->fields);
       DBUG_ASSERT(tsh->db.str && tsh->table_name.str);
       error= 1;
-      slave_print_msg(ERROR_LEVEL, rli, ER_BINLOG_ROW_WRONG_TABLE_DEF,
-                      "Column %d type mismatch - "
-                      "received type %d, %s.%s has type %d",
-                      col, type(col), tsh->db.str, tsh->table_name.str,
-                      table->field[col]->type());
+      char buf[256];
+      my_snprintf(buf, sizeof(buf), "Column %d type mismatch - "
+                  "received type %d, %s.%s has type %d",
+                  col, type(col), tsh->db.str, tsh->table_name.str,
+                  table->field[col]->type());
+      rli->report(ERROR_LEVEL, ER_BINLOG_ROW_WRONG_TABLE_DEF,
+                  ER(ER_BINLOG_ROW_WRONG_TABLE_DEF), buf);
+    }
+    /*
+      Check the slave's field size against that of the master.
+    */
+    if (!error && 
+        !table->field[col]->compatible_field_size(field_metadata(col)))
+    {
+      error= 1;
+      char buf[256];
+      my_snprintf(buf, sizeof(buf), "Column %d size mismatch - "
+                  "master has size %d, %s.%s on slave has size %d."
+                  " Master's column size should be <= the slave's "
+                  "column size.", col,
+                  table->field[col]->pack_length_from_metadata(
+                                       m_field_metadata[col]),
+                  tsh->db.str, tsh->table_name.str, 
+                  table->field[col]->row_pack_length());
+      rli->report(ERROR_LEVEL, ER_BINLOG_ROW_WRONG_TABLE_DEF,
+                  ER(ER_BINLOG_ROW_WRONG_TABLE_DEF), buf);
     }
   }