MDEV-17551 assert or crashed table when using blobs

The bug was that when long item-strings was converted to VARCHAR,
type_handler::string_type_handler() didn't take into account max
VARCHAR length.  The resulting Aria temporary table was created with
a VARCHAR field of length 1 when it should have been 65537. This caused
MariaDB to send impossible records to ma_write() and Aria reported
eventually the table as crashed.

Fixed by updating Type_handler::string_type_handler() to not create too long
VARCHAR fields. To make things extra safe, I also added checks in when
writing dynamic Aria records to ensure we find the wrong record during write
instead of during read.

mysql-test/main/type_varchar.result

@@ -686,3 +686,22 @@ DROP TABLE t1,t2;
 #
 # End of 10.0 tests
 #
+#
+# MDEV-17551
+# Assertion `(&(&share->intern_lock)->m_mutex)->count > 0 &&
+# pthread_equal(pthread_self(), (&(&share->intern_lock)->m_mutex)->
+# thread)' failed in _ma_state_info_write or ER_CRASHED_ON_USAGE
+# upon SELECT with UNION
+#
+CREATE TABLE t1 (b BLOB, vb BLOB AS (b) VIRTUAL);
+INSERT INTO t1 (b) VALUES ('foobar');
+SELECT 'foo' AS f1, CONVERT( 'bar' USING latin1 ) AS f2 FROM t1
+UNION
+SELECT b AS f1, CONVERT( vb USING latin1 ) AS f2 FROM t1;
+f1	f2
+foo	bar
+foobar	foobar
+DROP TABLE t1;
+#
+# End of 10.3 tests
+#

mysql-test/main/type_varchar.test

@@ -328,3 +328,22 @@ DROP TABLE t1,t2;
 --echo #
 --echo # End of 10.0 tests
 --echo #
+
+--echo #
+--echo # MDEV-17551
+--echo # Assertion `(&(&share->intern_lock)->m_mutex)->count > 0 &&
+--echo # pthread_equal(pthread_self(), (&(&share->intern_lock)->m_mutex)->
+--echo # thread)' failed in _ma_state_info_write or ER_CRASHED_ON_USAGE
+--echo # upon SELECT with UNION
+--echo #
+
+CREATE TABLE t1 (b BLOB, vb BLOB AS (b) VIRTUAL);
+INSERT INTO t1 (b) VALUES ('foobar');
+SELECT 'foo' AS f1, CONVERT( 'bar' USING latin1 ) AS f2 FROM t1
+  UNION
+SELECT b AS f1, CONVERT( vb USING latin1 ) AS f2 FROM t1;
+DROP TABLE t1;
+
+--echo #
+--echo # End of 10.3 tests
+--echo #

sql/sql_const.h

@@ -64,7 +64,7 @@
     CREATE TABLE t1 (c VARBINARY(65534));
     CREATE TABLE t1 (c VARBINARY(65535));
   Like VARCHAR(65536), they will be converted to BLOB automatically
-  in non-sctict mode.
+  in non-strict mode.
 */
 #define MAX_FIELD_VARCHARLENGTH	(65535-2-1)
 #define MAX_FIELD_BLOBLENGTH UINT_MAX32         /* cf field_blob::get_length() */

sql/sql_type.cc

@@ -327,6 +327,8 @@ Type_handler::string_type_handler(uint max_octet_length)
     return &type_handler_long_blob;
   else if (max_octet_length >= 65536)
     return &type_handler_medium_blob;
+  else if (max_octet_length >= MAX_FIELD_VARCHARLENGTH)
+    return &type_handler_blob;
   return &type_handler_varchar;
 }
 
@@ -1372,6 +1374,7 @@ Field *Type_handler_varchar::make_conversion_table_field(TABLE *table,
                                                          const Field *target)
                                                          const
 {
+  DBUG_ASSERT(HA_VARCHAR_PACKLENGTH(metadata) <= MAX_FIELD_VARCHARLENGTH);
   return new(table->in_use->mem_root)
          Field_varstring(NULL, metadata, HA_VARCHAR_PACKLENGTH(metadata),
                          (uchar *) "", 1, Field::NONE, &empty_clex_str,
@@ -2364,6 +2367,8 @@ Field *Type_handler_varchar::make_table_field(const LEX_CSTRING *name,
                                               TABLE *table) const
 
 {
+  DBUG_ASSERT(HA_VARCHAR_PACKLENGTH(attr.max_length) <=
+              MAX_FIELD_VARCHARLENGTH);
   return new (table->in_use->mem_root)
          Field_varstring(addr.ptr, attr.max_length,
                          HA_VARCHAR_PACKLENGTH(attr.max_length),

storage/maria/ma_check.c

@@ -5426,7 +5426,12 @@ int _ma_sort_write_record(MARIA_SORT_PARAM *sort_param)
       info->cur_row.checksum= (*share->calc_check_checksum)(info,
                                                               sort_param->
                                                               record);
-      reclength= _ma_rec_pack(info,from,sort_param->record);
+      if (!(reclength= _ma_rec_pack(info,from,sort_param->record)))
+      {
+        _ma_check_print_error(param,"Got error %d when packing record",
+                              my_errno);
+        DBUG_RETURN(1);
+      }
       flag=0;
 
       do

storage/maria/ma_close.c

@@ -114,8 +114,10 @@ int maria_close(register MARIA_HA *info)
                         share->deleting ? FLUSH_IGNORE_CHANGED : FLUSH_RELEASE))
         error= my_errno;
       unmap_file(info);
-      if (((share->changed && share->base.born_transactional) ||
-           maria_is_crashed(info) || (share->temporary && !share->deleting)))
+      if (!internal_table &&
+          (((share->changed && share->base.born_transactional) ||
+            maria_is_crashed(info) ||
+            (share->temporary && !share->deleting))))
       {
         if (save_global_changed)
         {

storage/maria/ma_dynrec.c

@@ -224,6 +224,8 @@ my_bool _ma_write_dynamic_record(MARIA_HA *info, const uchar *record)
 {
   ulong reclength= _ma_rec_pack(info,info->rec_buff + MARIA_REC_BUFF_OFFSET,
                                 record);
+  if (!reclength)
+    return 1;
   return (write_dynamic_record(info,info->rec_buff + MARIA_REC_BUFF_OFFSET,
                                reclength));
 }
@@ -234,6 +236,8 @@ my_bool _ma_update_dynamic_record(MARIA_HA *info, MARIA_RECORD_POS pos,
 {
   uint length= _ma_rec_pack(info, info->rec_buff + MARIA_REC_BUFF_OFFSET,
                             record);
+  if (!length)
+    return 1;
   return (update_dynamic_record(info, pos,
                                 info->rec_buff + MARIA_REC_BUFF_OFFSET,
                                 length));
@@ -258,12 +262,19 @@ my_bool _ma_write_blob_record(MARIA_HA *info, const uchar *record)
   reclength2= _ma_rec_pack(info,
                            rec_buff+ALIGN_SIZE(MARIA_MAX_DYN_BLOCK_HEADER),
 			   record);
+  if (!reclength2)
+  {
+    error= 1;
+    goto err;
+  }
+
   DBUG_PRINT("info",("reclength: %lu  reclength2: %lu",
 		     reclength, reclength2));
   DBUG_ASSERT(reclength2 <= reclength);
   error= write_dynamic_record(info,
                               rec_buff+ALIGN_SIZE(MARIA_MAX_DYN_BLOCK_HEADER),
                               reclength2);
+err:
   my_safe_afree(rec_buff, reclength);
   return(error != 0);
 }
@@ -293,12 +304,19 @@ my_bool _ma_update_blob_record(MARIA_HA *info, MARIA_RECORD_POS pos,
     my_errno= HA_ERR_OUT_OF_MEM; /* purecov: inspected */
     return(1);
   }
-  reclength2= _ma_rec_pack(info,rec_buff+ALIGN_SIZE(MARIA_MAX_DYN_BLOCK_HEADER),
+  reclength2= _ma_rec_pack(info, rec_buff+
+                           ALIGN_SIZE(MARIA_MAX_DYN_BLOCK_HEADER),
                            record);
+  if (!reclength2)
+  {
+    error= 1;
+    goto err;
+  }
   DBUG_ASSERT(reclength2 <= reclength);
   error=update_dynamic_record(info,pos,
 			      rec_buff+ALIGN_SIZE(MARIA_MAX_DYN_BLOCK_HEADER),
 			      reclength2);
+err:
   my_safe_afree(rec_buff, reclength);
   return(error != 0);
 }
@@ -938,7 +956,12 @@ static my_bool update_dynamic_record(MARIA_HA *info, MARIA_RECORD_POS filepos,
 }
 
 
-	/* Pack a record. Return new reclength */
+/**
+   Pack a record.
+
+   @return new reclength
+   @return 0 in case of wrong data in record
+*/
 
 uint _ma_rec_pack(MARIA_HA *info, register uchar *to,
                   register const uchar *from)
@@ -1042,6 +1065,11 @@ uint _ma_rec_pack(MARIA_HA *info, register uchar *to,
           tmp_length= uint2korr(from);
           store_key_length_inc(to,tmp_length);
         }
+        if (tmp_length > column->length)
+        {
+          my_errno= HA_ERR_WRONG_IN_RECORD;
+          DBUG_RETURN(0);
+        }
         memcpy(to, from+pack_length,tmp_length);
         to+= tmp_length;
         continue;
@@ -1613,7 +1641,9 @@ my_bool _ma_cmp_dynamic_record(register MARIA_HA *info,
       if (!(buffer=(uchar*) my_safe_alloca(buffer_length)))
 	DBUG_RETURN(1);
     }
-    reclength= _ma_rec_pack(info,buffer,record);
+    if (!(reclength= _ma_rec_pack(info,buffer,record)))
+      goto err;
+
     record= buffer;
 
     filepos= info->cur_row.lastpos;