WL#5151: Conversion between different types when replicating

Row-based replication requires the types of columns on the
master and slave to be approximately the same (some safe
conversions between strings are allowed), but does not
allow safe conversions between fields of similar types such
as TINYINT and INT.

This patch implement type conversions between similar fields
on the master and slave.

The conversions are controlled using a new variable
SLAVE_TYPE_CONVERSIONS of type SET('ALL_LOSSY','ALL_NON_LOSSY').

Non-lossy conversions are any conversions that do not run the
risk of losing any information, while lossy conversions can
potentially truncate the value. The column definitions are
checked to decide if the conversion is acceptable.

If neither conversion is enabled, it is required that the
definitions of the columns are identical on master and slave.

Conversion is done by creating an internal conversion table,
unpacking the master data into it, and then copy the data to
the real table on the slave.

.bzrignore:
  New files added
client/Makefile.am:
  New files added
client/mysqlbinlog.cc:
  Functions in rpl_utility.cc is now needed by mysqlbinlog.cc.
libmysqld/Makefile.am:
  New files added
mysql-test/extra/rpl_tests/check_type.inc:
  Test include file to check a single type conversion.
mysql-test/extra/rpl_tests/rpl_extraSlave_Col.test:
  Switching to use INT instead of TEXT for column that should not have matching types.
mysql-test/extra/rpl_tests/rpl_row_basic.test:
  Adding code to enable type conversions for BIT tests since InnoDB
  cannot handle them properly due to incorrect information stored as
  metadata.
mysql-test/extra/rpl_tests/type_conversions.test:
  Test file to check a set of type conversions
  with current settings of slave_type_conversions.
mysql-test/suite/rpl/t/rpl_typeconv.test:
  Test file to test conversions from master to slave with
  all possible values for slave_type_conversions.
  
  The test also checks that the slave_type_conversions
  variable works as expected.
sql/field.cc:
  Changing definition of compatible_field_size to both check if 
  two field with identical base types are compatible and give an
  order between them if they are compatible.
  
  This only implement checking on the slave, so it will not affect
  replication from an old master to a new slave.
sql/field.h:
  Changing prototypes for functions:
  - compatible_field_size()
  - init_for_tmp_table()
  - row_pack_length()
sql/log_event.cc:
  Changing compability checks to build a conversion table if the fields
  are compatible, but does not have the same base type.
sql/log_event_old.cc:
  Changing compability checks to build a conversion table if the fields
  are compatible, but does not have the same base type.
sql/mysql_priv.h:
  Adding global option variable for SLAVE_TYPE_CONVERSIONS
sql/mysqld.cc:
  Adding SLAVE_TYPE_CONVERSIONS global server variable.
sql/rpl_record.cc:
  Changing unpack_row to use the conversion table if present.
sql/rpl_rli.h:
  Removing function get_tabledef and replacing it with get_table_data().
  This function retrieve data for table opened for replication, not just
  table definition.
sql/rpl_utility.cc:
  Function table_def::compatible_with is changed to compare table on master
  and slave for compatibility and generate a conversions table if they are
  compatible.
  
  Computing real type of fields from metadata for ENUM and SET types.
  Computing pack_length correctly for ENUM, SET, and BLOB types.
  
  Adding optimization to not check compatibility if no
  slave type conversions are enabled.
sql/rpl_utility.h:
  Changing prototypes since implementation has changed.
  
  Modifying table_def::type() to return real type instead of stored type.
sql/set_var.cc:
  Adding SLAVE_TYPE_CONVERSIONS variable.
sql/set_var.h:
  Adding SLAVE_TYPE_CONVERSIONS variable.
sql/share/errmsg.txt:
  Adding error messages for slave type conversions.
sql/sql_class.h:
  Adding SLAVE_TYPE_CONVERSIONS variable.
sql/sql_select.cc:
  Correcting create_virtual_tmp_table() to compute null bit positions
  correctly in the presence of bit fields.

1 files changed, 22 insertions, 23 deletions

diff --git a/sql/field.h b/sql/field.h
index 7a9b69eff40..cfa3e57d2a0 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -164,22 +164,13 @@ public:
     table, which is located on disk).
   */
   virtual uint32 pack_length_in_rec() const { return pack_length(); }
-  virtual int compatible_field_size(uint field_metadata,
-                                    const Relay_log_info *);
+  virtual bool compatible_field_size(uint metadata, Relay_log_info *rli,
+                                     int *order);
   virtual uint pack_length_from_metadata(uint field_metadata)
-  { return field_metadata; }
-  /*
-    This method is used to return the size of the data in a row-based
-    replication row record. The default implementation of returning 0 is
-    designed to allow fields that do not use metadata to return TRUE (1)
-    from compatible_field_size() which uses this function in the comparison.
-    The default value for field metadata for fields that do not have 
-    metadata is 0. Thus, 0 == 0 means the fields are compatible in size.
-
-    Note: While most classes that override this method return pack_length(),
-    the classes Field_string, Field_varstring, and Field_blob return 
-    field_length + 1, field_length, and pack_length_no_ptr() respectfully.
-  */
+  {
+    DBUG_ENTER("Field::pack_length_from_metadata");
+    DBUG_RETURN(field_metadata);
+  }
   virtual uint row_pack_length() { return 0; }
   virtual int save_field_metadata(uchar *first_byte)
   { return do_save_field_metadata(first_byte); }
@@ -636,6 +627,13 @@ public:
   int store_decimal(const my_decimal *);
   my_decimal *val_decimal(my_decimal *);
   uint is_equal(Create_field *new_field);
+  uint row_pack_length() { return pack_length(); }
+  uint32 pack_length_from_metadata(uint field_metadata) {
+    uint32 length= pack_length();
+    DBUG_PRINT("result", ("pack_length_from_metadata(%d): %u",
+                          field_metadata, length));
+    return length;
+  }
   int check_int(CHARSET_INFO *cs, const char *str, int length,
                 const char *int_end, int error);
   bool get_int(CHARSET_INFO *cs, const char *from, uint len, 
@@ -806,8 +804,8 @@ public:
   uint32 pack_length() const { return (uint32) bin_size; }
   uint pack_length_from_metadata(uint field_metadata);
   uint row_pack_length() { return pack_length(); }
-  int compatible_field_size(uint field_metadata,
-                            const Relay_log_info *rli);
+  bool compatible_field_size(uint field_metadata, Relay_log_info *rli,
+                             int *order_var);
   uint is_equal(Create_field *new_field);
   virtual const uchar *unpack(uchar* to, const uchar *from,
                               uint param_data, bool low_byte_first);
@@ -1501,9 +1499,9 @@ public:
       return row_pack_length();
     return (((field_metadata >> 4) & 0x300) ^ 0x300) + (field_metadata & 0x00ff);
   }
-  int compatible_field_size(uint field_metadata,
-                            const Relay_log_info *rli);
-  uint row_pack_length() { return (field_length + 1); }
+  bool compatible_field_size(uint field_metadata, Relay_log_info *rli,
+                             int *order_var);
+  uint row_pack_length() { return field_length; }
   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);
@@ -1965,8 +1963,8 @@ public:
   uint pack_length_from_metadata(uint field_metadata);
   uint row_pack_length()
   { return (bytes_in_rec + ((bit_len > 0) ? 1 : 0)); }
-  int compatible_field_size(uint field_metadata,
-                            const Relay_log_info *rli);
+  bool compatible_field_size(uint metadata, Relay_log_info *rli,
+                             int *order_var);
   void sql_type(String &str) const;
   virtual uchar *pack(uchar *to, const uchar *from,
                       uint max_length, bool low_byte_first);
@@ -2069,7 +2067,8 @@ public:
   /* Init for a tmp table field. To be extended if need be. */
   void init_for_tmp_table(enum_field_types sql_type_arg,
                           uint32 max_length, uint32 decimals,
-                          bool maybe_null, bool is_unsigned);
+                          bool maybe_null, bool is_unsigned,
+                          uint pack_length = ~0U);
 
   bool init(THD *thd, char *field_name, enum_field_types type, char *length,
             char *decimals, uint type_modifier, Item *default_value,