MDEV-16699 heap-use-after-free in group_concat with compressed or GIS columns

Field_blob::store() has special code for GROUP_CONCAT temporary table
(to store blob values in Blob_mem_storage - this prevents them
from being freed/overwritten when a next row is read).

Field_geom and Field_blob_compressed inherit from Field_blob but they
have their own ::store() method without this special Blob_mem_storage
support.

Considering that non-grouping CONCAT() of such fields converts
them to plain BLOB, let's do the same for GROUP_CONCAT. To do it,
Item_func_group_concat::setup will signal that it's creating
a temporary table for GROUP_CONCAT, and Field_blog::make_new_field()
override will create base Field_blob when under group concat.

mysql-test/main/column_compression.result

@@ -2918,10 +2918,8 @@ CREATE TABLE t1 (a VARCHAR(500) COMPRESSED CHARACTER SET utf8mb3) ENGINE=InnoDB;
 INSERT INTO t1 SET a=REPEAT('x',127);
 ALTER TABLE t1 FORCE, ALGORITHM=COPY;
 DROP TABLE t1;
-#
 # End of 10.4 tests
 #
-#
 # MDEV-19727 Add Type_handler::Key_part_spec_init_ft
 #
 CREATE TABLE t1 (a VARCHAR(1000) COMPRESSED, FULLTEXT INDEX(a));
@@ -2929,5 +2927,20 @@ ERROR HY000: Compressed column 'a' can't be used in key specification
 CREATE TABLE t1 (a TEXT COMPRESSED, FULLTEXT INDEX(a));
 ERROR HY000: Compressed column 'a' can't be used in key specification
 #
-# End of 10.5 tests
+# MDEV-16699 heap-use-after-free in group_concat with compressed or GIS columns
 #
+create table t1 (c text compressed);
+insert into t1 values ('foo'),(repeat('a',55000));
+select length(group_concat(c order by 1)) from t1;
+length(group_concat(c order by 1))
+55004
+create table t2 as select group_concat(c order by 1), concat(c), c from t1;
+show create table t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `group_concat(c order by 1)` mediumtext DEFAULT NULL,
+  `concat(c)` text DEFAULT NULL,
+  `c` text /*M!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+drop table t1, t2;
+# End of 10.5 tests

mysql-test/main/column_compression.test

@@ -452,10 +452,7 @@ INSERT INTO t1 SET a=REPEAT('x',127);
 ALTER TABLE t1 FORCE, ALGORITHM=COPY;
 DROP TABLE t1;
 
-
---echo #
 --echo # End of 10.4 tests
---echo #
 
 --echo #
 --echo # MDEV-19727 Add Type_handler::Key_part_spec_init_ft
@@ -474,5 +471,13 @@ CREATE TABLE t1 (a VARCHAR(1000) COMPRESSED, FULLTEXT INDEX(a));
 CREATE TABLE t1 (a TEXT COMPRESSED, FULLTEXT INDEX(a));
 
 --echo #
---echo # End of 10.5 tests
+--echo # MDEV-16699 heap-use-after-free in group_concat with compressed or GIS columns
 --echo #
+create table t1 (c text compressed);
+insert into t1 values ('foo'),(repeat('a',55000));
+select length(group_concat(c order by 1)) from t1;
+create table t2 as select group_concat(c order by 1), concat(c), c from t1;
+show create table t2;
+drop table t1, t2;
+
+--echo # End of 10.5 tests

mysql-test/main/gis.result

@@ -5435,5 +5435,22 @@ AsText(g)
 POINT(1 1)
 DROP TABLE t1;
 #
-# End of 10.5 tests
+# MDEV-16699 heap-use-after-free in group_concat with compressed or GIS columns
 #
+create table t1 (c polygon);
+insert into t1 values
+(PolygonFromText('POLYGON((1 2,1 2))')),
+(PolygonFromText('POLYGON((0 0,1 1,0 0))'));
+select length(group_concat(c, c order by 1,2)) from t1;
+length(group_concat(c, c order by 1,2))
+229
+create table t2 as select group_concat(c, c order by 1,2), concat(c), c from t1;
+show create table t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `group_concat(c, c order by 1,2)` mediumblob DEFAULT NULL,
+  `concat(c)` longblob DEFAULT NULL,
+  `c` polygon DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+drop table t1, t2;
+# End of 10.5 tests

mysql-test/main/gis.test

@@ -3445,5 +3445,15 @@ SELECT AsText(g) FROM t1;
 DROP TABLE t1;
 
 --echo #
---echo # End of 10.5 tests
+--echo # MDEV-16699 heap-use-after-free in group_concat with compressed or GIS columns
 --echo #
+create table t1 (c polygon);
+insert into t1 values
+  (PolygonFromText('POLYGON((1 2,1 2))')),
+  (PolygonFromText('POLYGON((0 0,1 1,0 0))'));
+select length(group_concat(c, c order by 1,2)) from t1;
+create table t2 as select group_concat(c, c order by 1,2), concat(c), c from t1;
+show create table t2;
+drop table t1, t2;
+
+--echo # End of 10.5 tests

sql/field.cc

@@ -8882,6 +8882,24 @@ int Field_blob::key_cmp(const uchar *a,const uchar *b) const
 }
 
 
+#ifndef DBUG_OFF
+/* helper to assert that new_table->blob_storage is NULL */
+static struct blob_storage_check
+{
+  union { bool b; intptr p; } val;
+  blob_storage_check() { val.p= -1; val.b= false; }
+} blob_storage_check;
+#endif
+Field *Field_blob::make_new_field(MEM_ROOT *root, TABLE *newt, bool keep_type)
+{
+  DBUG_ASSERT((intptr(newt->blob_storage) & blob_storage_check.val.p) == 0);
+  if (newt->group_concat)
+    return new (root) Field_blob(field_length, maybe_null(), &field_name,
+                                 charset());
+  return Field::make_new_field(root, newt, keep_type);
+}
+
+
 Field *Field_blob::new_key_field(MEM_ROOT *root, TABLE *new_table,
                                  uchar *new_ptr, uint32 length,
                                  uchar *new_null_ptr, uint new_null_bit)

sql/field.h

@@ -4559,6 +4559,7 @@ public:
     return get_key_image_itRAW(ptr_arg, buff, length);
   }
   void set_key_image(const uchar *buff,uint length) override;
+  Field *make_new_field(MEM_ROOT *, TABLE *new_table, bool keep_type) override;
   Field *new_key_field(MEM_ROOT *root, TABLE *new_table,
                        uchar *new_ptr, uint32 length,
                        uchar *new_null_ptr, uint new_null_bit) override;

sql/item_sum.cc

@@ -4379,6 +4379,7 @@ bool Item_func_group_concat::setup(THD *thd)
   count_field_types(select_lex, tmp_table_param, all_fields, 0);
   tmp_table_param->force_copy_fields= force_copy_fields;
   tmp_table_param->hidden_field_count= (arg_count_order > 0);
+  tmp_table_param->group_concat= true;
   DBUG_ASSERT(table == 0);
   if (order_or_distinct)
   {
@@ -4406,11 +4407,10 @@ bool Item_func_group_concat::setup(THD *thd)
     Note that in the table, we first have the ORDER BY fields, then the
     field list.
   */
-  if (!(table= create_tmp_table(thd, tmp_table_param, all_fields,
+  table= create_tmp_table(thd, tmp_table_param, all_fields, NULL, 0, TRUE,
-                                (ORDER*) 0, 0, TRUE,
+                          (select_lex->options | thd->variables.option_bits),
-                                (select_lex->options |
+                          HA_POS_ERROR, &empty_clex_str);
-                                 thd->variables.option_bits),
+  if (!table)
-                                HA_POS_ERROR, &empty_clex_str)))
     DBUG_RETURN(TRUE);
   table->file->extra(HA_EXTRA_NO_ROWS);
   table->no_rows= 1;
@@ -4421,6 +4421,8 @@ bool Item_func_group_concat::setup(THD *thd)
   */
   if (order_or_distinct && table->s->blob_fields)
     table->blob_storage= new Blob_mem_storage();
+  else
+    table->blob_storage= NULL;
 
   /*
      Need sorting or uniqueness: init tree and choose a function to sort.

sql/sql_class.h

@@ -6148,6 +6148,7 @@ public:
     aggregate functions as normal functions.
   */
   bool precomputed_group_by;
+  bool group_concat;
   bool force_copy_fields;
   /*
     If TRUE, create_tmp_field called from create_tmp_table will convert
@@ -6166,7 +6167,7 @@ public:
      group_length(0), group_null_parts(0),
      using_outer_summary_function(0),
      schema_table(0), materialized_subquery(0), force_not_null_cols(0),
-     precomputed_group_by(0),
+     precomputed_group_by(0), group_concat(0),
      force_copy_fields(0), bit_fields_as_long(0), skip_create_table(0)
   {}
   ~TMP_TABLE_PARAM()

sql/sql_insert.cc

@@ -4509,12 +4509,9 @@ TABLE *select_create::create_table_from_items(THD *thd, List<Item> *items,
   bool save_table_creation_was_logged;
   DBUG_ENTER("select_create::create_table_from_items");
 
+  tmp_table.reset();
   tmp_table.s= &share;
   init_tmp_table_share(thd, &share, "", 0, "", "");
-
-  tmp_table.s->db_create_options=0;
-  tmp_table.null_row= 0;
-  tmp_table.maybe_null= 0;
   tmp_table.in_use= thd;
 
   if (!(thd->variables.option_bits & OPTION_EXPLICIT_DEF_TIMESTAMP))

sql/sql_select.cc

@@ -19029,6 +19029,7 @@ TABLE *Create_tmp_table::start(THD *thd,
   table->copy_blobs= 1;
   table->in_use= thd;
   table->no_rows_with_nulls= param->force_not_null_cols;
+  table->group_concat= param->group_concat;
   table->expr_arena= thd;
 
   table->s= share;

sql/table.h

@@ -1513,7 +1513,6 @@ public:
     Used only in the MODE_NO_AUTO_VALUE_ON_ZERO mode.
   */
   bool auto_increment_field_not_null;
-  bool insert_or_update;             /* Can be used by the handler */
   /*
      NOTE: alias_name_used is only a hint! It works only in need_correct_ident()
      condition. On other cases it is FALSE even if table_name is alias.
@@ -1534,12 +1533,11 @@ public:
 
   REGINFO reginfo;			/* field connections */
   MEM_ROOT mem_root;
-  /**
+  /* this is for temporary tables created inside Item_func_group_concat */
-     Initialized in Item_func_group_concat::setup for appropriate
+  union {
-     temporary table if GROUP_CONCAT is used with ORDER BY | DISTINCT
+    bool group_concat;                  /* used during create_tmp_table() */
-     and BLOB field count > 0.
+    Blob_mem_storage *blob_storage;     /* used after create_tmp_table()  */
-   */
+  };
-  Blob_mem_storage *blob_storage;
   GRANT_INFO grant;
   /*
     The arena which the items for expressions from the table definition