WL#5151: Conversion between different types when

         replicating

Replace c_ptr() calls with c_ptr_safe() calls to
avoid valgrind warnings.

Adding code to to handle the case that no metadata
was present in the table map for the column.

Allow first parameter to unpack_row() to be NULL,
in which case no source tables is used and hence
no checks nor conversions are done.

Clarifying some comments and fixing documentation
for unpack_row().

sql/rpl_record.cc

@@ -150,13 +150,20 @@ pack_row(TABLE *table, MY_BITMAP const* cols,
    the various member functions of Field and subclasses expect to
    write.
 
-   The row is assumed to only consist of the fields for which the corresponding
-   bit in bitset @c cols is set; the other parts of the record are left alone.
+   The row is assumed to only consist of the fields for which the
+   corresponding bit in bitset @c cols is set; the other parts of the
+   record are left alone.
 
    At most @c colcnt columns are read: if the table is larger than
    that, the remaining fields are not filled in.
 
-   @param rli     Relay log info
+   @note The relay log information can be NULL, which means that no
+   checking or comparison with the source table is done, simply
+   because it is not used.  This feature is used by MySQL Backup to
+   unpack a row from from the backup image, but can be used for other
+   purposes as well.
+
+   @param rli     Relay log info, which can be NULL
    @param table   Table to unpack into
    @param colcnt  Number of columns to read from record
    @param row_data
@@ -170,10 +177,8 @@ pack_row(TABLE *table, MY_BITMAP const* cols,
 
    @retval 0 No error
 
-   @retval ER_NO_DEFAULT_FOR_FIELD
-   Returned if one of the fields existing on the slave but not on the
-   master does not have a default value (and isn't nullable)
-
+   @retval HA_ERR_GENERIC
+   A generic, internal, error caused the unpacking to fail.
  */
 #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
 int
@@ -184,8 +189,8 @@ unpack_row(Relay_log_info const *rli,
 {
   DBUG_ENTER("unpack_row");
   DBUG_ASSERT(row_data);
+  DBUG_ASSERT(table);
   size_t const master_null_byte_count= (bitmap_bits_set(cols) + 7) / 8;
-  int error= 0;
 
   uchar const *null_ptr= row_data;
   uchar const *pack_ptr= row_data + master_null_byte_count;
@@ -201,14 +206,21 @@ unpack_row(Relay_log_info const *rli,
   // The "current" null bits
   unsigned int null_bits= *null_ptr++;
   uint i= 0;
-  table_def *tabledef;
-  TABLE *conv_table;
-  bool table_found= rli->get_table_data(table, &tabledef, &conv_table);
+  table_def *tabledef= NULL;
+  TABLE *conv_table= NULL;
+  bool table_found= rli && rli->get_table_data(table, &tabledef, &conv_table);
   DBUG_PRINT("debug", ("Table data: table_found: %d, tabldef: %p, conv_table: %p",
                        table_found, tabledef, conv_table));
   DBUG_ASSERT(table_found);
-  if (!table_found)
-    return HA_ERR_GENERIC;
+
+  /*
+    If rli is NULL it means that there is no source table and that the
+    row shall just be unpacked without doing any checks. This feature
+    is used by MySQL Backup, but can be used for other purposes as
+    well.
+   */
+  if (rli && !table_found)
+    DBUG_RETURN(HA_ERR_GENERIC);
 
   for (field_ptr= begin_ptr ; field_ptr < end_ptr && *field_ptr ; ++field_ptr)
   {
@@ -221,7 +233,7 @@ unpack_row(Relay_log_info const *rli,
       conv_table ? conv_table->field[field_ptr - begin_ptr] : NULL;
     Field *const f=
       conv_field ? conv_field : *field_ptr;
-    DBUG_PRINT("debug", ("Conversion %srequired for field '%s' (#%d)",
+    DBUG_PRINT("debug", ("Conversion %srequired for field '%s' (#%ld)",
                          conv_field ? "" : "not ",
                          (*field_ptr)->field_name, field_ptr - begin_ptr));
     DBUG_ASSERT(f != NULL);
@@ -285,7 +297,7 @@ unpack_row(Relay_log_info const *rli,
         conv_field->val_str(&value_string);
         DBUG_PRINT("debug", ("Copying field '%s' of type '%s' with value '%s'",
                              (*field_ptr)->field_name,
-                             source_type.c_ptr(), value_string.c_ptr()));
+                             source_type.c_ptr_safe(), value_string.c_ptr_safe()));
 #endif
         copy.set(*field_ptr, f, TRUE);
         (*copy.do_copy)(&copy);
@@ -296,7 +308,7 @@ unpack_row(Relay_log_info const *rli,
         (*field_ptr)->val_str(&value_string);
         DBUG_PRINT("debug", ("Value of field '%s' of type '%s' is now '%s'",
                              (*field_ptr)->field_name,
-                             target_type.c_ptr(), value_string.c_ptr()));
+                             target_type.c_ptr_safe(), value_string.c_ptr_safe()));
 #endif
       }
 
@@ -344,7 +356,7 @@ unpack_row(Relay_log_info const *rli,
       *master_reclength = table->s->reclength;
   }
   
-  DBUG_RETURN(error);
+  DBUG_RETURN(0);
 }
 
 /**

sql/rpl_utility.cc

@@ -582,7 +582,7 @@ can_convert_field_to(Field *field,
   String field_type(field_type_buf, sizeof(field_type_buf), field->charset());
   field->sql_type(field_type);
   DBUG_PRINT("enter", ("field_type: %s, target_type: %d, source_type: %d, source_metadata: 0x%x",
-                       field_type.c_ptr(), field->real_type(), source_type, metadata));
+                       field_type.c_ptr_safe(), field->real_type(), source_type, metadata));
 #endif
   /*
     If the real type is the same, we need to check the metadata to
@@ -590,6 +590,19 @@ can_convert_field_to(Field *field,
    */
   if (field->real_type() == source_type)
   {
+    if (metadata == 0) // Metadata can only be zero if no metadata was provided
+    {
+      /*
+        If there is no metadata, we either have an old event where no
+        metadata were supplied, or a type that does not require any
+        metadata. In either case, conversion can be done but no
+        conversion table is necessary.
+       */
+      DBUG_PRINT("debug", ("Base types are identical, but there is no metadata"));
+      *order_var= 0;
+      DBUG_RETURN(true);
+    }
+
     DBUG_PRINT("debug", ("Base types are identical, doing field size comparison"));
     if (field->compatible_field_size(metadata, rli, mflags, order_var))
       DBUG_RETURN(is_conversion_ok(*order_var, rli));
@@ -816,7 +829,7 @@ table_def::compatible_with(THD *thd, Relay_log_info *rli,
       rli->report(ERROR_LEVEL, ER_SLAVE_CONVERSION_FAILED,
                   ER(ER_SLAVE_CONVERSION_FAILED),
                   col, db_name, tbl_name,
-                  source_type.c_ptr(), target_type.c_ptr());
+                  source_type.c_ptr_safe(), target_type.c_ptr_safe());
       return false;
     }
   }
@@ -836,7 +849,7 @@ table_def::compatible_with(THD *thd, Relay_log_info *rli,
         DBUG_PRINT("debug", ("Field %s - conversion required."
                              " Source type: '%s', Target type: '%s'",
                              tmp_table->field[col]->field_name,
-                             source_type.c_ptr(), target_type.c_ptr()));
+                             source_type.c_ptr_safe(), target_type.c_ptr_safe()));
       }
   }
 #endif
@@ -920,7 +933,7 @@ TABLE *table_def::create_conversion_table(THD *thd, Relay_log_info *rli, TABLE *
     field_def->init_for_tmp_table(type(col),
                                   max_length,
                                   decimals,
-                                  maybe_null(col), // maybe_null
+                                  TRUE,         // maybe_null
                                   FALSE,        // unsigned_flag
                                   pack_length);
     field_def->charset= target_table->field[col]->charset();