MDEV-9238 Wrap create_virtual_tmp_table() into a class, split into different steps

9 files changed, 876 insertions, 358 deletions

diff --git a/sql/field.cc b/sql/field.cc
index 29c9625e3bc..e81a06a588b 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -9651,117 +9651,6 @@ void Column_definition::create_length_to_internal_length(void)
 }
 
 
-/**
-  Init for a tmp table field. To be extended if need be.
-*/
-void Column_definition::init_for_tmp_table(enum_field_types sql_type_arg,
-                                           uint32 length_arg,
-                                           uint32 decimals_arg,
-                                           bool maybe_null, bool is_unsigned,
-                                           uint pack_length_arg)
-{
-  DBUG_ENTER("Create_field::init_for_tmp_table");
-
-  field_name= "";
-  sql_type= sql_type_arg;
-  char_length= length= length_arg;;
-  unireg_check= Field::NONE;
-  interval= 0;
-  charset= &my_charset_bin;
-  geom_type= Field::GEOM_GEOMETRY;
-
-  DBUG_PRINT("enter", ("sql_type: %d, length: %u, pack_length: %u",
-                       sql_type_arg, length_arg, pack_length_arg));
-
-  /*
-    These pack flags are crafted to get it correctly through the
-    branches of make_field().
-   */
-  switch (sql_type_arg)
-  {
-  case MYSQL_TYPE_VARCHAR:
-  case MYSQL_TYPE_VAR_STRING:
-  case MYSQL_TYPE_STRING:
-  case MYSQL_TYPE_SET:
-    pack_flag= 0;
-    break;
-
-  case MYSQL_TYPE_GEOMETRY:
-    pack_flag= FIELDFLAG_GEOM;
-    break;
-
-  case MYSQL_TYPE_ENUM:
-    pack_flag= FIELDFLAG_INTERVAL;
-    break;
-
-  case MYSQL_TYPE_NEWDECIMAL:
-    DBUG_ASSERT(decimals_arg <= DECIMAL_MAX_SCALE);
-  case MYSQL_TYPE_DECIMAL:
-  case MYSQL_TYPE_FLOAT:
-  case MYSQL_TYPE_DOUBLE:
-    pack_flag= FIELDFLAG_NUMBER |
-      (decimals_arg & FIELDFLAG_MAX_DEC) << FIELDFLAG_DEC_SHIFT;
-    break;
-
-  case MYSQL_TYPE_TINY_BLOB:
-  case MYSQL_TYPE_MEDIUM_BLOB:
-  case MYSQL_TYPE_LONG_BLOB:
-  case MYSQL_TYPE_BLOB:
-    pack_flag= FIELDFLAG_BLOB;
-    break;
-
-  case MYSQL_TYPE_BIT:
-    pack_flag= FIELDFLAG_NUMBER | FIELDFLAG_TREAT_BIT_AS_CHAR;
-    break;
-
-  default:
-    pack_flag= FIELDFLAG_NUMBER;
-    break;
-  }
-
-  /*
-    Set the pack flag correctly for the blob-like types. This sets the
-    packtype to something that make_field can use. If the pack type is
-    not set correctly, the packlength will be reeeeally wierd (like
-    129 or so).
-   */
-  switch (sql_type_arg)
-  {
-  case MYSQL_TYPE_ENUM:
-  case MYSQL_TYPE_SET:
-  case MYSQL_TYPE_TINY_BLOB:
-  case MYSQL_TYPE_MEDIUM_BLOB:
-  case MYSQL_TYPE_LONG_BLOB:
-  case MYSQL_TYPE_BLOB:
-  case MYSQL_TYPE_GEOMETRY:
-    // If you are going to use the above types, you have to pass a
-    // pack_length as parameter. Assert that is really done.
-    DBUG_ASSERT(pack_length_arg != ~0U);
-    pack_flag|= pack_length_to_packflag(pack_length_arg);
-    break;
-  default:
-    /* Nothing */
-    break;
-  }
-
-  pack_flag|=
-    (maybe_null ? FIELDFLAG_MAYBE_NULL : 0) |
-    (is_unsigned ? 0 : FIELDFLAG_DECIMAL);
-
-  DBUG_PRINT("debug", ("pack_flag: %s%s%s%s%s%s, pack_type: %d",
-                       FLAGSTR(pack_flag, FIELDFLAG_BINARY),
-                       FLAGSTR(pack_flag, FIELDFLAG_NUMBER),
-                       FLAGSTR(pack_flag, FIELDFLAG_INTERVAL),
-                       FLAGSTR(pack_flag, FIELDFLAG_GEOM),
-                       FLAGSTR(pack_flag, FIELDFLAG_BLOB),
-                       FLAGSTR(pack_flag, FIELDFLAG_DECIMAL),
-                       f_packtype(pack_flag)));
-  vcol_info= 0;
-
-  DBUG_VOID_RETURN;
-}
-
-
 static inline bool is_item_func(Item* x)
 {
   return x != NULL && x->type() == Item::FUNC_ITEM;
diff --git a/sql/field.h b/sql/field.h
index 9f743f90628..1a0c347cf67 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -3481,12 +3481,6 @@ public:
   Column_definition(THD *thd, Field *field, Field *orig_field);
   void create_length_to_internal_length(void);
 
-  /* 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,
-                          uint pack_length = ~0U);
-
   bool check(THD *thd);
 
   bool stored_in_db() const { return !vcol_info || vcol_info->stored_in_db; }
diff --git a/sql/item.h b/sql/item.h
index 64410f33cd7..e64e3191d51 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -759,6 +759,17 @@ public:
   {
     return get_handler_by_field_type(field_type());
   }
+  Field *make_num_distinct_aggregator_field(MEM_ROOT *mem_root,
+                                            const Item *item) const
+  {
+    return type_handler()->make_num_distinct_aggregator_field(mem_root, this);
+  }
+  Field *make_conversion_table_field(TABLE *table,
+                                     uint metadata, const Field *target) const
+  {
+    DBUG_ASSERT(0); // Should not be called in Item context
+    return NULL;
+  }
   /* result_type() of an item specifies how the value should be returned */
   Item_result result_type() const { return type_handler()->result_type(); }
   /* ... while cmp_type() specifies how it should be compared */
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index 0b8cf2e4e2e..073bec4eaa0 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -684,32 +684,6 @@ int Aggregator_distinct::composite_key_cmp(void* arg, uchar* key1, uchar* key2)
 }
 
 
-static enum enum_field_types 
-calc_tmp_field_type(enum enum_field_types table_field_type, 
-                    Item_result result_type)
-{
-  /* Adjust tmp table type according to the chosen aggregation type */
-  switch (result_type) {
-  case STRING_RESULT:
-  case REAL_RESULT:
-    if (table_field_type != MYSQL_TYPE_FLOAT)
-      table_field_type= MYSQL_TYPE_DOUBLE;
-    break;
-  case INT_RESULT:
-    table_field_type= MYSQL_TYPE_LONGLONG;
-    /* fallthrough */
-  case DECIMAL_RESULT:
-    if (table_field_type != MYSQL_TYPE_LONGLONG)
-      table_field_type= MYSQL_TYPE_NEWDECIMAL;
-    break;
-  case ROW_RESULT:
-  default:
-    DBUG_ASSERT(0);
-  }
-  return table_field_type;
-}
-
-
 /***************************************************************************/
 
 C_MODE_START
@@ -886,8 +860,6 @@ bool Aggregator_distinct::setup(THD *thd)
   }
   else
   {
-    List<Column_definition> field_list;
-    Column_definition field_def;                /* field definition */
     Item *arg;
     DBUG_ENTER("Aggregator_distinct::setup");
     /* It's legal to call setup() more than once when in a subquery */
@@ -899,8 +871,6 @@ bool Aggregator_distinct::setup(THD *thd)
       PS/SP. Hence all further allocations are performed in the runtime
       mem_root.
     */
-    if (field_list.push_back(&field_def, thd->mem_root))
-      DBUG_RETURN(TRUE);
 
     item_sum->null_value= item_sum->maybe_null= 1;
     item_sum->quick_group= 0;
@@ -918,17 +888,8 @@ bool Aggregator_distinct::setup(THD *thd)
     if (always_null)
       DBUG_RETURN(FALSE);
 
-    enum enum_field_types field_type;
-
-    field_type= calc_tmp_field_type(arg->field_type(),
-                              arg->result_type());
-    field_def.init_for_tmp_table(field_type, 
-                                 arg->max_length,
-                                 arg->decimals, 
-                                 arg->maybe_null,
-                                 arg->unsigned_flag);
-
-    if (! (table= create_virtual_tmp_table(thd, field_list)))
+    Field *field= arg->make_num_distinct_aggregator_field(thd->mem_root, arg);
+    if (!field || !(table= create_virtual_tmp_table(thd, field)))
       DBUG_RETURN(TRUE);
 
     /* XXX: check that the case of CHAR(0) works OK */
diff --git a/sql/rpl_utility.cc b/sql/rpl_utility.cc
index 1c48d6bc937..244363e43ac 100644
--- a/sql/rpl_utility.cc
+++ b/sql/rpl_utility.cc
@@ -922,6 +922,51 @@ table_def::compatible_with(THD *thd, rpl_group_info *rgi,
   return true;
 }
 
+
+/**
+  A wrapper to Virtual_tmp_table, to get access to its constructor,
+  which is protected for safety purposes (against illegal use on stack).
+*/
+class Virtual_conversion_table: public Virtual_tmp_table
+{
+public:
+  Virtual_conversion_table(THD *thd) :Virtual_tmp_table(thd) { }
+  /**
+    Add a new field into the virtual table.
+    @param sql_type     - The real_type of the field.
+    @param metadata     - The RBR binary log metadata for this field.
+    @param target_field - The field from the target table, to get extra
+                          attributes from (e.g. typelib in case of ENUM).
+  */
+  bool add(enum_field_types sql_type,
+           uint16 metadata, const Field *target_field)
+  {
+    const Type_handler *handler= Type_handler::get_handler_by_real_type(sql_type);
+    if (!handler)
+    {
+      sql_print_error("In RBR mode, Slave received unknown field type field %d "
+                      " for column Name: %s.%s.%s.",
+                      (int) sql_type,
+                      target_field->table->s->db.str,
+                      target_field->table->s->table_name.str,
+                      target_field->field_name);
+      return true;
+    }
+    Field *tmp= handler->make_conversion_table_field(this, metadata,
+                                                     target_field);
+    if (!tmp)
+      return true;
+    Virtual_tmp_table::add(tmp);
+    DBUG_PRINT("debug", ("sql_type: %d, target_field: '%s', max_length: %d, decimals: %d,"
+                         " maybe_null: %d, unsigned_flag: %d, pack_length: %u",
+                         sql_type, target_field->field_name,
+                         tmp->field_length, tmp->decimals(), TRUE,
+                         tmp->flags, tmp->pack_length()));
+    return false;
+  }
+};
+
+
 /**
   Create a conversion table.
 
@@ -937,8 +982,7 @@ TABLE *table_def::create_conversion_table(THD *thd, rpl_group_info *rgi,
 {
   DBUG_ENTER("table_def::create_conversion_table");
 
-  List<Column_definition> field_list;
-  TABLE *conv_table= NULL;
+  Virtual_conversion_table *conv_table;
   Relay_log_info *rli= rgi->rli;
   /*
     At slave, columns may differ. So we should create
@@ -946,101 +990,35 @@ TABLE *table_def::create_conversion_table(THD *thd, rpl_group_info *rgi,
     conversion table.
   */
   uint const cols_to_create= MY_MIN(target_table->s->fields, size());
+  if (!(conv_table= new(thd) Virtual_conversion_table(thd)) ||
+      conv_table->init(cols_to_create))
+    goto err;
   for (uint col= 0 ; col < cols_to_create; ++col)
   {
-    Column_definition *field_def=
-      (Create_field*) alloc_root(thd->mem_root, sizeof(Create_field));
-    bool unsigned_flag= 0;
-    if (field_list.push_back(field_def, thd->mem_root))
-      DBUG_RETURN(NULL);
-
-    uint decimals= 0;
-    TYPELIB* interval= NULL;
-    uint pack_length= 0;
-    uint32 max_length=
-      max_display_length_for_field(type(col), field_metadata(col));
-
-    switch(type(col)) {
-      int precision;
-    case MYSQL_TYPE_ENUM:
-    case MYSQL_TYPE_SET:
-      interval= static_cast<Field_enum*>(target_table->field[col])->typelib;
-      pack_length= field_metadata(col) & 0x00ff;
-      break;
-
-    case MYSQL_TYPE_NEWDECIMAL:
-      /*
-        The display length of a DECIMAL type is not the same as the
-        length that should be supplied to make_field, so we correct
-        the length here.
-       */
-      precision= field_metadata(col) >> 8;
-      decimals= field_metadata(col) & 0x00ff;
-      max_length=
-        my_decimal_precision_to_length(precision, decimals, FALSE);
-      break;
-
-    case MYSQL_TYPE_DECIMAL:
-      sql_print_error("In RBR mode, Slave received incompatible DECIMAL field "
-                      "(old-style decimal field) from Master while creating "
-                      "conversion table. Please consider changing datatype on "
-                      "Master to new style decimal by executing ALTER command for"
-                      " column Name: %s.%s.%s.",
-                      target_table->s->db.str,
-                      target_table->s->table_name.str,
-                      target_table->field[col]->field_name);
+    if (conv_table->add(type(col), field_metadata(col),
+                        target_table->field[col]))
+    {
+      DBUG_PRINT("debug", ("binlog_type: %d, metadata: %04X, target_field: '%s'"
+                           " make_conversion_table_field() failed",
+                           binlog_type(col), field_metadata(col),
+                           target_table->field[col]->field_name));
       goto err;
-
-    case MYSQL_TYPE_TINY_BLOB:
-    case MYSQL_TYPE_MEDIUM_BLOB:
-    case MYSQL_TYPE_LONG_BLOB:
-    case MYSQL_TYPE_BLOB:
-    case MYSQL_TYPE_GEOMETRY:
-      pack_length= field_metadata(col) & 0x00ff;
-      break;
-
-    case MYSQL_TYPE_TINY:
-    case MYSQL_TYPE_SHORT:
-    case MYSQL_TYPE_INT24:
-    case MYSQL_TYPE_LONG:
-    case MYSQL_TYPE_LONGLONG:
-      /*
-        As we don't know if the integer was signed or not on the master,
-        assume we have same sign on master and slave.  This is true when not
-        using conversions so it should be true also when using conversions.
-      */
-      unsigned_flag= ((Field_num*) target_table->field[col])->unsigned_flag;
-      break;
-    default:
-      break;
     }
-
-    DBUG_PRINT("debug", ("sql_type: %d, target_field: '%s', max_length: %d, decimals: %d,"
-                         " maybe_null: %d, unsigned_flag: %d, pack_length: %u",
-                         binlog_type(col), target_table->field[col]->field_name,
-                         max_length, decimals, TRUE, unsigned_flag,
-                         pack_length));
-    field_def->init_for_tmp_table(type(col),
-                                  max_length,
-                                  decimals,
-                                  TRUE,         // maybe_null
-                                  unsigned_flag,
-                                  pack_length);
-    field_def->charset= target_table->field[col]->charset();
-    field_def->interval= interval;
   }
 
-  conv_table= create_virtual_tmp_table(thd, field_list);
+  if (conv_table->open())
+    goto err; // Could not allocate record buffer?
 
-err:
-  if (conv_table == NULL)
-  {
-    rli->report(ERROR_LEVEL, ER_SLAVE_CANT_CREATE_CONVERSION, rgi->gtid_info(),
-                ER_THD(thd, ER_SLAVE_CANT_CREATE_CONVERSION),
-                target_table->s->db.str,
-                target_table->s->table_name.str);
-  }
   DBUG_RETURN(conv_table);
+
+err:
+  if (conv_table)
+    delete conv_table;
+  rli->report(ERROR_LEVEL, ER_SLAVE_CANT_CREATE_CONVERSION, rgi->gtid_info(),
+              ER_THD(thd, ER_SLAVE_CANT_CREATE_CONVERSION),
+              target_table->s->db.str,
+              target_table->s->table_name.str);
+  DBUG_RETURN(NULL);
 }
 #endif /* MYSQL_CLIENT */
 
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 62ca735b2c8..5b6d98cdf45 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -16016,9 +16016,9 @@ Field *create_tmp_field(THD *thd, TABLE *table,Item *item, Item::Type type,
     a tmp_set bitmap to be used by things like filesort.
 */
 
-void setup_tmp_table_column_bitmaps(TABLE *table, uchar *bitmaps)
+void
+setup_tmp_table_column_bitmaps(TABLE *table, uchar *bitmaps, uint field_count)
 {
-  uint field_count= table->s->fields;
   uint bitmap_size= bitmap_buffer_size(field_count);
 
   DBUG_ASSERT(table->s->vfields == 0 && table->def_vcol_set == 0);
@@ -16042,6 +16042,13 @@ void setup_tmp_table_column_bitmaps(TABLE *table, uchar *bitmaps)
 }
 
 
+void
+setup_tmp_table_column_bitmaps(TABLE *table, uchar *bitmaps)
+{
+  setup_tmp_table_column_bitmaps(table, bitmaps, table->s->fields);
+}
+
+
 /**
   Create a temp table according to a field list.
 
@@ -16926,139 +16933,108 @@ err:
 
 /****************************************************************************/
 
-/**
-  Create a reduced TABLE object with properly set up Field list from a
-  list of field definitions.
-
-    The created table doesn't have a table handler associated with
-    it, has no keys, no group/distinct, no copy_funcs array.
-    The sole purpose of this TABLE object is to use the power of Field
-    class to read/write data to/from table->record[0]. Then one can store
-    the record in any container (RB tree, hash, etc).
-    The table is created in THD mem_root, so are the table's fields.
-    Consequently, if you don't BLOB fields, you don't need to free it.
-
-  @param thd         connection handle
-  @param field_list  list of column definitions
+void *Virtual_tmp_table::operator new(size_t size, THD *thd) throw()
+{
+  return (Virtual_tmp_table *) alloc_root(thd->mem_root, size);
+}
 
-  @return
-    0 if out of memory, TABLE object in case of success
-*/
 
-TABLE *create_virtual_tmp_table(THD *thd, List<Column_definition> &field_list)
+bool Virtual_tmp_table::init(uint field_count)
 {
-  uint field_count= field_list.elements;
-  uint blob_count= 0;
-  Field **field;
-  Column_definition *cdef;            /* column definition */
-  uint record_length= 0;
-  uint null_count= 0;                 /* number of columns which may be null */
-  uint null_pack_length;              /* NULL representation array length */
   uint *blob_field;
   uchar *bitmaps;
-  TABLE *table;
-  TABLE_SHARE *share;
-
-  if (!multi_alloc_root(thd->mem_root,
-                        &table, sizeof(*table),
-                        &share, sizeof(*share),
+  if (!multi_alloc_root(in_use->mem_root,
+                        &s, sizeof(*s),
                         &field, (field_count + 1) * sizeof(Field*),
-                        &blob_field, (field_count+1) *sizeof(uint),
-                        &bitmaps, bitmap_buffer_size(field_count)*5,
+                        &blob_field, (field_count + 1) * sizeof(uint),
+                        &bitmaps, bitmap_buffer_size(field_count) * 5,
                         NullS))
-    return 0;
+    return true;
+  bzero(s, sizeof(*s));
+  s->blob_field= blob_field;
+  setup_tmp_table_column_bitmaps(this, bitmaps, field_count);
+  m_alloced_field_count= field_count;
+  return false;
+};
 
-  bzero(table, sizeof(*table));
-  bzero(share, sizeof(*share));
-  table->field= field;
-  table->s= share;
-  table->temp_pool_slot= MY_BIT_NONE;
-  share->blob_field= blob_field;
-  share->fields= field_count;
-  setup_tmp_table_column_bitmaps(table, bitmaps);
 
+bool Virtual_tmp_table::add(List<Column_definition> &field_list)
+{
   /* Create all fields and calculate the total length of record */
+  Column_definition *cdef;            /* column definition */
   List_iterator_fast<Column_definition> it(field_list);
-  while ((cdef= it++))
+  for ( ; (cdef= it++); )
   {
-    *field= cdef->make_field(share, thd->mem_root, 0,
+    Field *tmp;
+    if (!(tmp= cdef->make_field(s, in_use->mem_root, 0,
                              (uchar*) (f_maybe_null(cdef->pack_flag) ? "" : 0),
                              f_maybe_null(cdef->pack_flag) ? 1 : 0,
-                             cdef->field_name);
-    if (!*field)
-      goto error;
-    (*field)->init(table);
-    record_length+= (*field)->pack_length();
-    if (! ((*field)->flags & NOT_NULL_FLAG))
-      null_count++;
-
-    if ((*field)->flags & BLOB_FLAG)
-      share->blob_field[blob_count++]= (uint) (field - table->field);
-
-    field++;
+                             cdef->field_name)))
+      return true;
+    add(tmp);
   }
-  *field= NULL;                             /* mark the end of the list */
-  share->blob_field[blob_count]= 0;            /* mark the end of the list */
-  share->blob_fields= blob_count;
+  return false;
+}
 
-  null_pack_length= (null_count + 7)/8;
-  share->reclength= record_length + null_pack_length;
-  share->rec_buff_length= ALIGN_SIZE(share->reclength + 1);
-  table->record[0]= (uchar*) thd->alloc(share->rec_buff_length);
-  if (!table->record[0])
-    goto error;
 
-  if (null_pack_length)
-  {
-    table->null_flags= (uchar*) table->record[0];
-    share->null_fields= null_count;
-    share->null_bytes= share->null_bytes_for_compare= null_pack_length;
-  }
+void Virtual_tmp_table::setup_field_pointers()
+{
+  uchar *null_pos= record[0];
+  uchar *field_pos= null_pos + s->null_bytes;
+  uint null_bit= 1;
 
-  table->in_use= thd;           /* field->reset() may access table->in_use */
+  for (Field **cur_ptr= field; *cur_ptr; ++cur_ptr)
   {
-    /* Set up field pointers */
-    uchar *null_pos= table->record[0];
-    uchar *field_pos= null_pos + share->null_bytes;
-    uint null_bit= 1;
-
-    for (field= table->field; *field; ++field)
+    Field *cur_field= *cur_ptr;
+    if ((cur_field->flags & NOT_NULL_FLAG))
+      cur_field->move_field(field_pos);
+    else
     {
-      Field *cur_field= *field;
-      if ((cur_field->flags & NOT_NULL_FLAG))
-        cur_field->move_field(field_pos);
-      else
+      cur_field->move_field(field_pos, (uchar*) null_pos, null_bit);
+      null_bit<<= 1;
+      if (null_bit == (uint)1 << 8)
       {
-        cur_field->move_field(field_pos, (uchar*) null_pos, null_bit);
-        null_bit<<= 1;
-        if (null_bit == (uint)1 << 8)
-        {
-          ++null_pos;
-          null_bit= 1;
-        }
+        ++null_pos;
+        null_bit= 1;
       }
-      if (cur_field->type() == MYSQL_TYPE_BIT &&
-          cur_field->key_type() == HA_KEYTYPE_BIT)
+    }
+    if (cur_field->type() == MYSQL_TYPE_BIT &&
+        cur_field->key_type() == HA_KEYTYPE_BIT)
+    {
+      /* This is a Field_bit since key_type is HA_KEYTYPE_BIT */
+      static_cast<Field_bit*>(cur_field)->set_bit_ptr(null_pos, null_bit);
+      null_bit+= cur_field->field_length & 7;
+      if (null_bit > 7)
       {
-        /* This is a Field_bit since key_type is HA_KEYTYPE_BIT */
-        static_cast<Field_bit*>(cur_field)->set_bit_ptr(null_pos, null_bit);
-        null_bit+= cur_field->field_length & 7;
-        if (null_bit > 7)
-        {
-          null_pos++;
-          null_bit-= 8;
-        }
+        null_pos++;
+        null_bit-= 8;
       }
-      cur_field->reset();
-
-      field_pos+= cur_field->pack_length();
     }
+    cur_field->reset();
+    field_pos+= cur_field->pack_length();
   }
-  return table;
-error:
-  for (field= table->field; *field; ++field)
-    delete *field;                         /* just invokes field destructor */
-  return 0;
+}
+
+
+bool Virtual_tmp_table::open()
+{
+  // Make sure that we added all the fields we planned to:
+  DBUG_ASSERT(s->fields == m_alloced_field_count);
+  field[s->fields]= NULL;            // mark the end of the list
+  s->blob_field[s->blob_fields]= 0;  // mark the end of the list
+
+  uint null_pack_length= (s->null_fields + 7) / 8; // NULL-bit array length
+  s->reclength+= null_pack_length;
+  s->rec_buff_length= ALIGN_SIZE(s->reclength + 1);
+  if (!(record[0]= (uchar*) in_use->alloc(s->rec_buff_length)))
+    return true;
+  if (null_pack_length)
+  {
+    null_flags= (uchar*) record[0];
+    s->null_bytes= s->null_bytes_for_compare= null_pack_length;
+  }
+  setup_field_pointers();
+  return false;
 }
 
 
diff --git a/sql/sql_select.h b/sql/sql_select.h
index a65aa9fdeaa..1b1bb6ded71 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -1861,7 +1861,180 @@ Field *create_tmp_field(THD *thd, TABLE *table,Item *item, Item::Type type,
   All methods presume that there is at least one field to change.
 */
 
-TABLE *create_virtual_tmp_table(THD *thd, List<Column_definition> &field_list);
+
+class Virtual_tmp_table: public TABLE
+{
+  /**
+    Destruct collected fields. This method is called on errors only,
+    when we could not make the virtual temporary table completely,
+    e.g. when some of the fields could not be created or added.
+
+    This is needed to avoid memory leaks, as some fields can be BLOB
+    variants and thus can have String onboard. Strings must be destructed
+    as they store data not the heap (not on MEM_ROOT).
+  */
+  void destruct_fields()
+  {
+    for (uint i= 0; i < s->fields; i++)
+      delete field[i];  // to invoke the field destructor
+    s->fields= 0;       // safety
+  }
+
+protected:
+  /**
+     The number of the fields that are going to be in the table.
+     We remember the number of the fields at init() time, and
+     at open() we check that all of the fields were really added.
+  */
+  uint m_alloced_field_count;
+
+  /**
+    Setup field pointers and null-bit pointers.
+  */
+  void setup_field_pointers();
+
+public:
+  /**
+    Create a new empty virtual temporary table on the thread mem_root.
+    After creation, the caller must:
+    - call init()
+    - populate the table with new fields using add().
+    - call open().
+    @param thd         - Current thread.
+  */
+  static void *operator new(size_t size, THD *thd) throw();
+
+  Virtual_tmp_table(THD *thd)
+  {
+    bzero(this, sizeof(*this));
+    temp_pool_slot= MY_BIT_NONE;
+    in_use= thd;
+  }
+
+  ~Virtual_tmp_table()
+  {
+    destruct_fields();
+  }
+
+  /**
+    Allocate components for the given number of fields.
+     - fields[]
+     - s->blob_fields[],
+     - bitmaps: def_read_set, def_write_set, tmp_set, eq_join_set, cond_set.
+    @param field_count - The number of fields we plan to add to the table.
+    @returns false     - on success.
+    @returns true      - on error.
+  */
+  bool init(uint field_count);
+
+  /**
+    Add one Field to the end of the field array, update members:
+    s->reclength, s->fields, s->blob_fields, s->null_fuelds.
+  */
+  bool add(Field *new_field)
+  {
+    DBUG_ASSERT(s->fields < m_alloced_field_count);
+    new_field->init(this);
+    field[s->fields]= new_field;
+    s->reclength+= new_field->pack_length();
+    if (!(new_field->flags & NOT_NULL_FLAG))
+      s->null_fields++;
+    if (new_field->flags & BLOB_FLAG)
+    {
+      // Note, s->blob_fields was incremented in Field_blob::Field_blob
+      DBUG_ASSERT(s->blob_fields);
+      DBUG_ASSERT(s->blob_fields <= m_alloced_field_count);
+      s->blob_field[s->blob_fields - 1]= s->fields;
+    }
+    s->fields++;
+    return false;
+  }
+
+  /**
+    Add fields from a Column_definition list
+    @returns false - on success.
+    @returns true  - on error.
+  */
+  bool add(List<Column_definition> &field_list);
+
+  /**
+    Open a virtual table for read/write:
+    - Setup end markers in TABLE::field and TABLE_SHARE::blob_fields,
+    - Allocate a buffer in TABLE::record[0].
+    - Set field pointers (Field::ptr, Field::null_pos, Field::null_bit) to
+      the allocated record.
+    This method is called when all of the fields have been added to the table.
+    After calling this method the table is ready for read and write operations.
+    @return false - on success
+    @return true  - on error (e.g. could not allocate the record buffer).
+  */
+  bool open();
+};
+
+
+/**
+  Create a reduced TABLE object with properly set up Field list from a
+  list of field definitions.
+
+    The created table doesn't have a table handler associated with
+    it, has no keys, no group/distinct, no copy_funcs array.
+    The sole purpose of this TABLE object is to use the power of Field
+    class to read/write data to/from table->record[0]. Then one can store
+    the record in any container (RB tree, hash, etc).
+    The table is created in THD mem_root, so are the table's fields.
+    Consequently, if you don't BLOB fields, you don't need to free it.
+
+  @param thd         connection handle
+  @param field_list  list of column definitions
+
+  @return
+    0 if out of memory, or a
+    TABLE object ready for read and write in case of success
+*/
+
+inline TABLE *
+create_virtual_tmp_table(THD *thd, List<Column_definition> &field_list)
+{
+  Virtual_tmp_table *table;
+  if (!(table= new(thd) Virtual_tmp_table(thd)))
+    return NULL;
+  if (table->init(field_list.elements) ||
+      table->add(field_list) ||
+      table->open())
+  {
+    delete table;
+    return NULL;
+  }
+  return table;
+}
+
+
+/**
+  Create a new virtual temporary table consisting of a single field.
+  SUM(DISTINCT expr) and similar numeric aggregate functions use this.
+  @param thd    - Current thread
+  @param field  - The field that will be added into the table.
+  @return NULL  - On error.
+  @return !NULL - A pointer to the created table that is ready
+                  for read and write.
+*/
+inline TABLE *
+create_virtual_tmp_table(THD *thd, Field *field)
+{
+  Virtual_tmp_table *table;
+  DBUG_ASSERT(field);
+  if (!(table= new(thd) Virtual_tmp_table(thd)))
+    return NULL;
+  if (table->init(1) ||
+      table->add(field) ||
+      table->open())
+  {
+    delete table;
+    return NULL;
+  }
+  return table;
+}
+
 
 int test_if_item_cache_changed(List<Cached_item> &list);
 int join_init_read_record(JOIN_TAB *tab);
diff --git a/sql/sql_type.cc b/sql/sql_type.cc
index 113a242b01e..4e492e7099d 100644
--- a/sql/sql_type.cc
+++ b/sql/sql_type.cc
@@ -16,6 +16,9 @@
 
 #include "sql_type.h"
 #include "sql_const.h"
+#include "sql_class.h"
+#include "item.h"
+#include "log.h"
 
 static Type_handler_tiny        type_handler_tiny;
 static Type_handler_short       type_handler_short;
@@ -220,3 +223,424 @@ Type_handler::get_handler_by_real_type(enum_field_types type)
   return &type_handler_string;
 }
 
+
+/**
+  Create a DOUBLE field by default.
+*/
+Field *
+Type_handler::make_num_distinct_aggregator_field(MEM_ROOT *mem_root,
+                                                 const Item *item) const
+{
+  return new(mem_root)
+         Field_double(NULL, item->max_length,
+                      (uchar *) (item->maybe_null ? "" : 0),
+                      item->maybe_null ? 1 : 0, Field::NONE,
+                      item->name, item->decimals, 0, item->unsigned_flag);
+}
+
+
+Field *
+Type_handler_float::make_num_distinct_aggregator_field(MEM_ROOT *mem_root,
+                                                       const Item *item)
+                                                       const
+{
+  return new(mem_root)
+         Field_float(NULL, item->max_length,
+                     (uchar *) (item->maybe_null ? "" : 0),
+                     item->maybe_null ? 1 : 0, Field::NONE,
+                     item->name, item->decimals, 0, item->unsigned_flag);
+}
+
+
+Field *
+Type_handler_decimal_result::make_num_distinct_aggregator_field(
+                                                            MEM_ROOT *mem_root,
+                                                            const Item *item)
+                                                            const
+{
+  DBUG_ASSERT(item->decimals <= DECIMAL_MAX_SCALE);
+  return new (mem_root)
+         Field_new_decimal(NULL, item->max_length,
+                           (uchar *) (item->maybe_null ? "" : 0),
+                           item->maybe_null ? 1 : 0, Field::NONE,
+                           item->name, item->decimals, 0, item->unsigned_flag);
+}
+
+
+Field *
+Type_handler_int_result::make_num_distinct_aggregator_field(MEM_ROOT *mem_root,
+                                                            const Item *item)
+                                                            const
+{
+  /**
+    Make a longlong field for all INT-alike types. It could create
+    smaller fields for TINYINT, SMALLINT, MEDIUMINT, INT though.
+  */
+  return new(mem_root)
+         Field_longlong(NULL, item->max_length,
+                        (uchar *) (item->maybe_null ? "" : 0),
+                        item->maybe_null ? 1 : 0, Field::NONE,
+                        item->name, 0, item->unsigned_flag);
+}
+
+
+/***********************************************************************/
+
+#define TMPNAME ""
+
+Field *Type_handler_tiny::make_conversion_table_field(TABLE *table,
+                                                      uint metadata,
+                                                      const Field *target)
+                                                      const
+{
+  /*
+    As we don't know if the integer was signed or not on the master,
+    assume we have same sign on master and slave.  This is true when not
+    using conversions so it should be true also when using conversions.
+  */
+  bool unsigned_flag= ((Field_num*) target)->unsigned_flag;
+  return new (table->in_use->mem_root)
+         Field_tiny(NULL, 4 /*max_length*/, (uchar *) "", 1, Field::NONE,
+                    TMPNAME, 0/*zerofill*/, unsigned_flag);
+}
+
+
+Field *Type_handler_short::make_conversion_table_field(TABLE *table,
+                                                       uint metadata,
+                                                       const Field *target)
+                                                       const
+{
+  bool unsigned_flag= ((Field_num*) target)->unsigned_flag;
+  return new (table->in_use->mem_root)
+         Field_short(NULL, 6 /*max_length*/, (uchar *) "", 1, Field::NONE,
+                     TMPNAME, 0/*zerofill*/, unsigned_flag);
+}
+
+
+Field *Type_handler_int24::make_conversion_table_field(TABLE *table,
+                                                       uint metadata,
+                                                       const Field *target)
+                                                       const
+{
+  bool unsigned_flag= ((Field_num*) target)->unsigned_flag;
+  return new (table->in_use->mem_root)
+         Field_medium(NULL, 9 /*max_length*/, (uchar *) "", 1, Field::NONE,
+                      TMPNAME, 0/*zerofill*/, unsigned_flag);
+}
+
+
+Field *Type_handler_long::make_conversion_table_field(TABLE *table,
+                                                      uint metadata,
+                                                      const Field *target)
+                                                      const
+{
+  bool unsigned_flag= ((Field_num*) target)->unsigned_flag;
+  return new (table->in_use->mem_root)
+         Field_long(NULL, 11 /*max_length*/, (uchar *) "", 1, Field::NONE,
+         TMPNAME, 0/*zerofill*/, unsigned_flag);
+}
+
+
+Field *Type_handler_longlong::make_conversion_table_field(TABLE *table,
+                                                          uint metadata,
+                                                          const Field *target)
+                                                          const
+{
+  bool unsigned_flag= ((Field_num*) target)->unsigned_flag;
+  return new (table->in_use->mem_root)
+         Field_longlong(NULL, 20 /*max_length*/,(uchar *) "", 1, Field::NONE,
+                        TMPNAME, 0/*zerofill*/, unsigned_flag);
+}
+
+
+
+Field *Type_handler_float::make_conversion_table_field(TABLE *table,
+                                                       uint metadata,
+                                                       const Field *target)
+                                                       const
+{
+  return new (table->in_use->mem_root)
+         Field_float(NULL, 12 /*max_length*/, (uchar *) "", 1, Field::NONE,
+                     TMPNAME, 0/*dec*/, 0/*zerofill*/, 0/*unsigned_flag*/);
+}
+
+
+Field *Type_handler_double::make_conversion_table_field(TABLE *table,
+                                                        uint metadata,
+                                                        const Field *target)
+                                                        const
+{
+  return new (table->in_use->mem_root)
+         Field_double(NULL, 22 /*max_length*/, (uchar *) "", 1, Field::NONE,
+                      TMPNAME, 0/*dec*/, 0/*zerofill*/, 0/*unsigned_flag*/);
+}
+
+
+Field *Type_handler_newdecimal::make_conversion_table_field(TABLE *table,
+                                                            uint metadata,
+                                                            const Field *target)
+                                                            const
+{
+  int  precision= metadata >> 8;
+  uint decimals= metadata & 0x00ff;
+  uint32 max_length= my_decimal_precision_to_length(precision, decimals, false);
+  DBUG_ASSERT(decimals <= DECIMAL_MAX_SCALE);
+  return new (table->in_use->mem_root)
+         Field_new_decimal(NULL, max_length, (uchar *) "", 1, Field::NONE,
+                           TMPNAME, decimals, 0/*zerofill*/, 0/*unsigned*/);
+}
+
+
+Field *Type_handler_olddecimal::make_conversion_table_field(TABLE *table,
+                                                            uint metadata,
+                                                            const Field *target)
+                                                            const
+{
+  sql_print_error("In RBR mode, Slave received incompatible DECIMAL field "
+                  "(old-style decimal field) from Master while creating "
+                  "conversion table. Please consider changing datatype on "
+                  "Master to new style decimal by executing ALTER command for"
+                  " column Name: %s.%s.%s.",
+                  target->table->s->db.str,
+                  target->table->s->table_name.str,
+                  target->field_name);
+  return NULL;
+}
+
+
+Field *Type_handler_year::make_conversion_table_field(TABLE *table,
+                                                      uint metadata,
+                                                      const Field *target)
+                                                      const
+{
+  return new(table->in_use->mem_root)
+         Field_year(NULL, 4, (uchar *) "", 1, Field::NONE, TMPNAME);
+}
+
+
+Field *Type_handler_null::make_conversion_table_field(TABLE *table,
+                                                      uint metadata,
+                                                      const Field *target)
+                                                      const
+{
+  return new(table->in_use->mem_root)
+         Field_null(NULL, 0, Field::NONE, TMPNAME, target->charset());
+}
+
+
+Field *Type_handler_timestamp::make_conversion_table_field(TABLE *table,
+                                                           uint metadata,
+                                                           const Field *target)
+                                                           const
+{
+  // We assume TIMESTAMP(0)
+  return new(table->in_use->mem_root)
+         Field_timestamp(NULL, MAX_DATETIME_WIDTH, (uchar *) "", 1,
+                         Field::NONE, TMPNAME, table->s);
+}
+
+
+Field *Type_handler_timestamp2::make_conversion_table_field(TABLE *table,
+                                                            uint metadata,
+                                                            const Field *target)
+                                                            const
+{
+  return new(table->in_use->mem_root)
+         Field_timestampf(NULL, (uchar *) "", 1, Field::NONE,
+                          TMPNAME, table->s, metadata);
+}
+
+
+Field *Type_handler_newdate::make_conversion_table_field(TABLE *table,
+                                                         uint metadata,
+                                                         const Field *target)
+                                                         const
+{
+  return new(table->in_use->mem_root)
+         Field_newdate(NULL, (uchar *) "", 1, Field::NONE, TMPNAME);
+}
+
+
+Field *Type_handler_date::make_conversion_table_field(TABLE *table,
+                                                      uint metadata,
+                                                      const Field *target)
+                                                      const
+{
+  return new(table->in_use->mem_root)
+         Field_date(NULL, (uchar *) "", 1, Field::NONE, TMPNAME);
+}
+
+
+Field *Type_handler_time::make_conversion_table_field(TABLE *table,
+                                                      uint metadata,
+                                                      const Field *target)
+                                                      const
+{
+  return new(table->in_use->mem_root)
+         Field_time(NULL, MAX_TIME_WIDTH, (uchar *) "", 1,
+                    Field::NONE, TMPNAME);
+}
+
+
+Field *Type_handler_time2::make_conversion_table_field(TABLE *table,
+                                                       uint metadata,
+                                                       const Field *target)
+                                                       const
+{
+  return new(table->in_use->mem_root)
+         Field_timef(NULL, (uchar *) "", 1, Field::NONE, TMPNAME, metadata);
+}
+
+
+Field *Type_handler_datetime::make_conversion_table_field(TABLE *table,
+                                                          uint metadata,
+                                                          const Field *target)
+                                                          const
+{
+  return new(table->in_use->mem_root)
+         Field_datetime(NULL, MAX_DATETIME_WIDTH, (uchar *) "", 1,
+                        Field::NONE, TMPNAME);
+}
+
+
+Field *Type_handler_datetime2::make_conversion_table_field(TABLE *table,
+                                                           uint metadata,
+                                                           const Field *target)
+                                                           const
+{
+  return new(table->in_use->mem_root)
+         Field_datetimef(NULL, (uchar *) "", 1,
+                         Field::NONE, TMPNAME, metadata);
+}
+
+
+Field *Type_handler_bit::make_conversion_table_field(TABLE *table,
+                                                     uint metadata,
+                                                     const Field *target)
+                                                     const
+{
+  DBUG_ASSERT((metadata & 0xff) <= 7);
+  uint32 max_length= 8 * (metadata >> 8U) + (metadata & 0x00ff);
+  return new(table->in_use->mem_root)
+         Field_bit_as_char(NULL, max_length, (uchar *) "", 1,
+                           Field::NONE, TMPNAME);
+}
+
+
+Field *Type_handler_string::make_conversion_table_field(TABLE *table,
+                                                        uint metadata,
+                                                        const Field *target)
+                                                        const
+{
+  /* This is taken from Field_string::unpack. */
+  uint32 max_length= (((metadata >> 4) & 0x300) ^ 0x300) + (metadata & 0x00ff);
+  return new(table->in_use->mem_root)
+         Field_string(NULL, max_length, (uchar *) "", 1,
+                      Field::NONE, TMPNAME, target->charset());
+}
+
+
+Field *Type_handler_varchar::make_conversion_table_field(TABLE *table,
+                                                         uint metadata,
+                                                         const Field *target)
+                                                         const
+{
+  return new(table->in_use->mem_root)
+         Field_varstring(NULL, metadata, HA_VARCHAR_PACKLENGTH(metadata),
+                         (uchar *) "", 1, Field::NONE, TMPNAME,
+                         table->s, target->charset());
+}
+
+
+Field *Type_handler_tiny_blob::make_conversion_table_field(TABLE *table,
+                                                           uint metadata,
+                                                           const Field *target)
+                                                           const
+{
+  return new(table->in_use->mem_root)
+         Field_blob(NULL, (uchar *) "", 1, Field::NONE, TMPNAME,
+                    table->s, 1, target->charset());
+}
+
+
+Field *Type_handler_blob::make_conversion_table_field(TABLE *table,
+                                                      uint metadata,
+                                                      const Field *target)
+                                                      const
+{
+  return new(table->in_use->mem_root)
+         Field_blob(NULL, (uchar *) "", 1, Field::NONE, TMPNAME,
+                    table->s, 2, target->charset());
+}
+
+
+Field *Type_handler_medium_blob::make_conversion_table_field(TABLE *table,
+                                                           uint metadata,
+                                                           const Field *target)
+                                                           const
+{
+  return new(table->in_use->mem_root)
+         Field_blob(NULL, (uchar *) "", 1, Field::NONE, TMPNAME,
+                    table->s, 3, target->charset());
+}
+
+
+Field *Type_handler_long_blob::make_conversion_table_field(TABLE *table,
+                                                           uint metadata,
+                                                           const Field *target)
+                                                           const
+{
+  return new(table->in_use->mem_root)
+         Field_blob(NULL, (uchar *) "", 1, Field::NONE, TMPNAME,
+                    table->s, 4, target->charset());
+}
+
+
+#ifdef HAVE_SPATIAL
+Field *Type_handler_geometry::make_conversion_table_field(TABLE *table,
+                                                          uint metadata,
+                                                          const Field *target)
+                                                          const
+{
+  DBUG_ASSERT(target->type() == MYSQL_TYPE_GEOMETRY);
+  /*
+    We do not do not update feature_gis statistics here:
+    status_var_increment(target->table->in_use->status_var.feature_gis);
+    as this is only a temporary field.
+    The statistics was already incremented when "target" was created.
+  */
+  return new(table->in_use->mem_root)
+         Field_geom(NULL, (uchar *) "", 1, Field::NONE, TMPNAME, table->s, 4,
+                    ((const Field_geom*) target)->geom_type,
+                    ((const Field_geom*) target)->srid);
+}
+#endif
+
+Field *Type_handler_enum::make_conversion_table_field(TABLE *table,
+                                                      uint metadata,
+                                                      const Field *target)
+                                                      const
+{
+  DBUG_ASSERT(target->type() == MYSQL_TYPE_STRING);
+  DBUG_ASSERT(target->real_type() == MYSQL_TYPE_ENUM);
+  return new(table->in_use->mem_root)
+         Field_enum(NULL, target->field_length,
+                    (uchar *) "", 1, Field::NONE, TMPNAME,
+                    metadata & 0x00ff/*pack_length()*/,
+                    ((const Field_enum*) target)->typelib, target->charset());
+}
+
+
+Field *Type_handler_set::make_conversion_table_field(TABLE *table,
+                                                     uint metadata,
+                                                     const Field *target)
+                                                     const
+{
+  DBUG_ASSERT(target->type() == MYSQL_TYPE_STRING);
+  DBUG_ASSERT(target->real_type() == MYSQL_TYPE_SET);
+  return new(table->in_use->mem_root)
+         Field_set(NULL, target->field_length,
+                   (uchar *) "", 1, Field::NONE, TMPNAME,
+                   metadata & 0x00ff/*pack_length()*/,
+                   ((const Field_enum*) target)->typelib, target->charset());
+}
diff --git a/sql/sql_type.h b/sql/sql_type.h
index d2f85384d28..1bc14a0c208 100644
--- a/sql/sql_type.h
+++ b/sql/sql_type.h
@@ -23,6 +23,10 @@
 
 #include "mysqld.h"
 
+class Field;
+class Item;
+struct TABLE;
+
 class Type_handler
 {
 protected:
@@ -39,6 +43,42 @@ public:
                                             CHARSET_INFO *cs) const
   { return this; }
   virtual ~Type_handler() {}
+  /**
+    Makes a temporary table Field to handle numeric aggregate functions,
+    e.g. SUM(DISTINCT expr), AVG(DISTINCT expr), etc.
+  */
+  virtual Field *make_num_distinct_aggregator_field(MEM_ROOT *,
+                                                    const Item *) const;
+  /**
+    Makes a temporary table Field to handle RBR replication type conversion.
+    @param TABLE    - The conversion table the field is going to be added to.
+                      It's used to access to table->in_use->mem_root,
+                      to create the new field on the table memory root,
+                      as well as to increment statistics in table->share
+                      (e.g. table->s->blob_count).
+    @param metadata - Metadata from the binary log.
+    @param target   - The field in the target table on the slave.
+
+    Note, the data types of "target" and of "this" are not necessarily
+    always the same, in general case it's possible that:
+            this->field_type() != target->field_type()
+    and/or
+            this->real_type( ) != target->real_type()
+
+    This method decodes metadata according to this->real_type()
+    and creates a new field also according to this->real_type().
+
+    In some cases it lurks into "target", to get some extra information, e.g.:
+    - unsigned_flag for numeric fields
+    - charset() for string fields
+    - typelib and field_length for SET and ENUM
+    - geom_type and srid for GEOMETRY
+    This information is not available in the binary log, so
+    we assume that these fields are the same on the master and on the slave.
+  */
+  virtual Field *make_conversion_table_field(TABLE *TABLE,
+                                             uint metadata,
+                                             const Field *target) const= 0;
 };
 
 
@@ -59,6 +99,7 @@ public:
   Item_result result_type() const { return DECIMAL_RESULT; }
   Item_result cmp_type() const { return DECIMAL_RESULT; }
   virtual ~Type_handler_decimal_result() {};
+  Field *make_num_distinct_aggregator_field(MEM_ROOT *, const Item *) const;
 };
 
 
@@ -68,6 +109,7 @@ public:
   Item_result result_type() const { return INT_RESULT; }
   Item_result cmp_type() const { return INT_RESULT; }
   virtual ~Type_handler_int_result() {}
+  Field *make_num_distinct_aggregator_field(MEM_ROOT *, const Item *) const;
 };
 
 
@@ -117,6 +159,8 @@ class Type_handler_tiny: public Type_handler_int_result
 public:
   virtual ~Type_handler_tiny() {}
   enum_field_types field_type() const { return MYSQL_TYPE_TINY; }
+  Field *make_conversion_table_field(TABLE *TABLE, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -125,6 +169,8 @@ class Type_handler_short: public Type_handler_int_result
 public:
   virtual ~Type_handler_short() {}
   enum_field_types field_type() const { return MYSQL_TYPE_SHORT; }
+  Field *make_conversion_table_field(TABLE *TABLE, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -133,6 +179,8 @@ class Type_handler_long: public Type_handler_int_result
 public:
   virtual ~Type_handler_long() {}
   enum_field_types field_type() const { return MYSQL_TYPE_LONG; }
+  Field *make_conversion_table_field(TABLE *TABLE, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -141,6 +189,8 @@ class Type_handler_longlong: public Type_handler_int_result
 public:
   virtual ~Type_handler_longlong() {}
   enum_field_types field_type() const { return MYSQL_TYPE_LONGLONG; }
+  Field *make_conversion_table_field(TABLE *TABLE, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -149,6 +199,8 @@ class Type_handler_int24: public Type_handler_int_result
 public:
   virtual ~Type_handler_int24() {}
   enum_field_types field_type() const { return MYSQL_TYPE_INT24; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -157,6 +209,8 @@ class Type_handler_year: public Type_handler_int_result
 public:
   virtual ~Type_handler_year() {}
   enum_field_types field_type() const { return MYSQL_TYPE_YEAR; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -165,6 +219,8 @@ class Type_handler_bit: public Type_handler_int_result
 public:
   virtual ~Type_handler_bit() {}
   enum_field_types field_type() const { return MYSQL_TYPE_BIT; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -173,6 +229,9 @@ class Type_handler_float: public Type_handler_real_result
 public:
   virtual ~Type_handler_float() {}
   enum_field_types field_type() const { return MYSQL_TYPE_FLOAT; }
+  Field *make_num_distinct_aggregator_field(MEM_ROOT *, const Item *) const;
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -181,6 +240,8 @@ class Type_handler_double: public Type_handler_real_result
 public:
   virtual ~Type_handler_double() {}
   enum_field_types field_type() const { return MYSQL_TYPE_DOUBLE; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -189,6 +250,8 @@ class Type_handler_time: public Type_handler_temporal_result
 public:
   virtual ~Type_handler_time() {}
   enum_field_types field_type() const { return MYSQL_TYPE_TIME; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -198,6 +261,8 @@ public:
   virtual ~Type_handler_time2() {}
   enum_field_types field_type() const { return MYSQL_TYPE_TIME; }
   enum_field_types real_field_type() const { return MYSQL_TYPE_TIME2; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -206,6 +271,8 @@ class Type_handler_date: public Type_handler_temporal_result
 public:
   virtual ~Type_handler_date() {}
   enum_field_types field_type() const { return MYSQL_TYPE_DATE; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -214,7 +281,8 @@ class Type_handler_newdate: public Type_handler_temporal_result
 public:
   virtual ~Type_handler_newdate() {}
   enum_field_types field_type() const { return MYSQL_TYPE_DATE; }
-  enum_field_types real_field_type() const { return MYSQL_TYPE_NEWDATE; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -223,6 +291,8 @@ class Type_handler_datetime: public Type_handler_temporal_result
 public:
   virtual ~Type_handler_datetime() {}
   enum_field_types field_type() const { return MYSQL_TYPE_DATETIME; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -232,6 +302,8 @@ public:
   virtual ~Type_handler_datetime2() {}
   enum_field_types field_type() const { return MYSQL_TYPE_DATETIME; }
   enum_field_types real_field_type() const { return MYSQL_TYPE_DATETIME2; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -240,6 +312,8 @@ class Type_handler_timestamp: public Type_handler_temporal_result
 public:
   virtual ~Type_handler_timestamp() {}
   enum_field_types field_type() const { return MYSQL_TYPE_TIMESTAMP; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -249,6 +323,8 @@ public:
   virtual ~Type_handler_timestamp2() {}
   enum_field_types field_type() const { return MYSQL_TYPE_TIMESTAMP; }
   enum_field_types real_field_type() const { return MYSQL_TYPE_TIMESTAMP2; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -257,6 +333,8 @@ class Type_handler_olddecimal: public Type_handler_decimal_result
 public:
   virtual ~Type_handler_olddecimal() {}
   enum_field_types field_type() const { return MYSQL_TYPE_DECIMAL; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -265,6 +343,8 @@ class Type_handler_newdecimal: public Type_handler_decimal_result
 public:
   virtual ~Type_handler_newdecimal() {}
   enum_field_types field_type() const { return MYSQL_TYPE_NEWDECIMAL; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -273,6 +353,8 @@ class Type_handler_null: public Type_handler_string_result
 public:
   virtual ~Type_handler_null() {}
   enum_field_types field_type() const { return MYSQL_TYPE_NULL; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -281,6 +363,8 @@ class Type_handler_string: public Type_handler_string_result
 public:
   virtual ~Type_handler_string() {}
   enum_field_types field_type() const { return MYSQL_TYPE_STRING; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -289,6 +373,8 @@ class Type_handler_varchar: public Type_handler_string_result
 public:
   virtual ~Type_handler_varchar() {}
   enum_field_types field_type() const { return MYSQL_TYPE_VARCHAR; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -297,6 +383,8 @@ class Type_handler_tiny_blob: public Type_handler_string_result
 public:
   virtual ~Type_handler_tiny_blob() {}
   enum_field_types field_type() const { return MYSQL_TYPE_TINY_BLOB; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -305,6 +393,8 @@ class Type_handler_medium_blob: public Type_handler_string_result
 public:
   virtual ~Type_handler_medium_blob() {}
   enum_field_types field_type() const { return MYSQL_TYPE_MEDIUM_BLOB; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -313,6 +403,8 @@ class Type_handler_long_blob: public Type_handler_string_result
 public:
   virtual ~Type_handler_long_blob() {}
   enum_field_types field_type() const { return MYSQL_TYPE_LONG_BLOB; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -321,15 +413,21 @@ class Type_handler_blob: public Type_handler_string_result
 public:
   virtual ~Type_handler_blob() {}
   enum_field_types field_type() const { return MYSQL_TYPE_BLOB; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
+#ifdef HAVE_SPATIAL
 class Type_handler_geometry: public Type_handler_string_result
 {
 public:
   virtual ~Type_handler_geometry() {}
   enum_field_types field_type() const { return MYSQL_TYPE_GEOMETRY; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
+#endif
 
 
 class Type_handler_enum: public Type_handler_string_result
@@ -338,6 +436,8 @@ public:
   virtual ~Type_handler_enum() {}
   enum_field_types field_type() const { return MYSQL_TYPE_VARCHAR; }
   virtual enum_field_types real_field_type() const { return MYSQL_TYPE_ENUM; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -347,6 +447,8 @@ public:
   virtual ~Type_handler_set() {}
   enum_field_types field_type() const { return MYSQL_TYPE_VARCHAR; }
   virtual enum_field_types real_field_type() const { return MYSQL_TYPE_SET; }
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
 };
 
 
@@ -414,6 +516,16 @@ public:
       m_type_handler->type_handler_adjusted_to_max_octet_length(max_octet_length,
                                                                 cs);
   }
+  Field *make_num_distinct_aggregator_field(MEM_ROOT *mem_root,
+                                            const Item *item) const
+  {
+    return m_type_handler->make_num_distinct_aggregator_field(mem_root, item);
+  }
+  Field *make_conversion_table_field(TABLE *table, uint metadata,
+                                     const Field *target) const
+  {
+    return m_type_handler->make_conversion_table_field(table, metadata, target);
+  }
 };