MDEV-9745 Crash with CASE WHEN TRUE THEN COALESCE(CAST(NULL AS UNSIGNED)) ELSE 4 END

This is a backport of the patch for MDEV-9653 (fixed earlier in 10.1.13).

The code in Item_func_case::fix_length_and_dec() did not
calculate max_length and decimals properly.

In case of any numeric result (DECIMAL, REAL, INT) a generic method
Item_func_case::agg_num_lengths() was called, which could erroneously result
into a DECIMAL item with max_length==0 and decimals==0, so the constructor of
Field_new_decimals tried to create a field of DECIMAL(0,0) type,
which caused a crash.

Unlike Item_func_case, the code responsible for merging attributes in
Item_func_coalesce::fix_length_and_dec() works fine: it has specific execution
branches for all distinct numeric types and correctly creates a DECIMAL(1,0)
column instead of DECIMAL(0,0) for the same set of arguments.

The fix does the following:
- Moves the attribute merging code from Item_func_coalesce::fix_length_and_dec()
  to a new method Item_func_hybrid_result_type::fix_attributes()
- Removes the wrong code from Item_func_case::fix_length_and_dec()
  and reuses fix_attributes() in both Item_func_coalesce::fix_length_and_dec()
  and Item_func_case::fix_length_and_dec()
- Fixes count_real_length() and count_decimal_length() to get an array
  of Items as an argument, instead of using Item::args directly.
  This is needed for Item_func_case::fix_length_and_dec().
- Moves methods Item_func::count_xxx_length() from "public" to "protected".
- Removes Item_func_case::agg_num_length(), as it's not used any more.
- Additionally removes Item_func_case::agg_str_length(),
  as it also was not used (dead code).

mysql-test/r/case.result

@@ -231,3 +231,16 @@ case t1.f1 when '00:00:00' then 1 end
 1
 NULL
 drop table t1;
+#
+# MDEV-9745 Crash with CASE WHEN TRUE THEN COALESCE(CAST(NULL AS UNSIGNED)) ELSE 4 END
+#
+CREATE TABLE t1 SELECT CASE WHEN TRUE THEN COALESCE(CAST(NULL AS UNSIGNED)) ELSE 4 END AS a;
+DESCRIBE t1;
+Field	Type	Null	Key	Default	Extra
+a	decimal(1,0)	YES		NULL	
+DROP TABLE t1;
+CREATE TABLE t1 SELECT CASE WHEN TRUE THEN COALESCE(CAST(NULL AS UNSIGNED)) ELSE 40 END AS a;
+DESCRIBE t1;
+Field	Type	Null	Key	Default	Extra
+a	decimal(2,0)	YES		NULL	
+DROP TABLE t1;

mysql-test/t/case.test

@@ -193,3 +193,12 @@ insert t1 values ('00:00:00'),('00:01:00');
 select case t1.f1 when '00:00:00' then 1 end from t1;
 drop table t1;
 
+--echo #
+--echo # MDEV-9745 Crash with CASE WHEN TRUE THEN COALESCE(CAST(NULL AS UNSIGNED)) ELSE 4 END
+--echo #
+CREATE TABLE t1 SELECT CASE WHEN TRUE THEN COALESCE(CAST(NULL AS UNSIGNED)) ELSE 4 END AS a;
+DESCRIBE t1;
+DROP TABLE t1;
+CREATE TABLE t1 SELECT CASE WHEN TRUE THEN COALESCE(CAST(NULL AS UNSIGNED)) ELSE 40 END AS a;
+DESCRIBE t1;
+DROP TABLE t1;

sql/item_cmpfunc.cc

@@ -3015,24 +3015,6 @@ bool Item_func_case::fix_fields(THD *thd, Item **ref)
 }
 
 
-void Item_func_case::agg_str_lengths(Item* arg)
-{
-  fix_char_length(max(max_char_length(), arg->max_char_length()));
-  set_if_bigger(decimals, arg->decimals);
-  unsigned_flag= unsigned_flag && arg->unsigned_flag;
-}
-
-
-void Item_func_case::agg_num_lengths(Item *arg)
-{
-  uint len= my_decimal_length_to_precision(arg->max_length, arg->decimals,
-                                           arg->unsigned_flag) - arg->decimals;
-  set_if_bigger(max_length, len); 
-  set_if_bigger(decimals, arg->decimals);
-  unsigned_flag= unsigned_flag && arg->unsigned_flag; 
-}
-
-
 /**
   Check if (*place) and new_value points to different Items and call
   THD::change_item_tree() if needed.
@@ -3098,17 +3080,7 @@ void Item_func_case::fix_length_and_dec()
   }
   else
   {
-    collation.set_numeric();
-    max_length=0;
-    decimals=0;
-    unsigned_flag= TRUE;
-    for (uint i= 0; i < ncases; i+= 2)
-      agg_num_lengths(args[i + 1]);
-    if (else_expr_num != -1) 
-      agg_num_lengths(args[else_expr_num]);
-    max_length= my_decimal_precision_to_length_no_truncation(max_length +
-                                                             decimals, decimals,
-                                               unsigned_flag);
+    fix_attributes(agg, nagg);
   }
   
   /*
@@ -3336,19 +3308,32 @@ void Item_func_coalesce::fix_length_and_dec()
 {
   cached_field_type= agg_field_type(args, arg_count);
   agg_result_type(&cached_result_type, args, arg_count);
+  fix_attributes(args, arg_count);
+}
+
+
+#if MYSQL_VERSION_ID > 100100
+#error Rename this to Item_hybrid_func::fix_attributes() when mering to 10.1
+#endif
+void Item_func_hybrid_result_type::fix_attributes(Item **items, uint nitems)
+{
   switch (cached_result_type) {
   case STRING_RESULT:
-    if (count_string_result_length(cached_field_type, args, arg_count))
+    if (count_string_result_length(field_type(),
+                                   items, nitems))
       return;          
     break;
   case DECIMAL_RESULT:
-    count_decimal_length();
+    collation.set_numeric();
+    count_decimal_length(items, nitems);
     break;
   case REAL_RESULT:
-    count_real_length();
+    collation.set_numeric();
+    count_real_length(items, nitems);
     break;
   case INT_RESULT:
-    count_only_length(args, arg_count);
+    collation.set_numeric();
+    count_only_length(items, nitems);
     decimals= 0;
     break;
   case ROW_RESULT:

sql/item_cmpfunc.h

@@ -1255,8 +1255,6 @@ public:
   Item *find_item(String *str);
   CHARSET_INFO *compare_collation() { return cmp_collation.collation; }
   void cleanup();
-  void agg_str_lengths(Item *arg);
-  void agg_num_lengths(Item *arg);
 };
 
 /*

sql/item_func.cc

@@ -641,16 +641,16 @@ void Item_func::count_datetime_length(Item **item, uint nitems)
   result length/precision depends on argument ones.
 */
 
-void Item_func::count_decimal_length()
+void Item_func::count_decimal_length(Item **item, uint nitems)
 {
   int max_int_part= 0;
   decimals= 0;
   unsigned_flag= 1;
-  for (uint i=0 ; i < arg_count ; i++)
+  for (uint i=0 ; i < nitems ; i++)
   {
-    set_if_bigger(decimals, args[i]->decimals);
-    set_if_bigger(max_int_part, args[i]->decimal_int_part());
-    set_if_smaller(unsigned_flag, args[i]->unsigned_flag);
+    set_if_bigger(decimals, item[i]->decimals);
+    set_if_bigger(max_int_part, item[i]->decimal_int_part());
+    set_if_smaller(unsigned_flag, item[i]->unsigned_flag);
   }
   int precision= min(max_int_part + decimals, DECIMAL_MAX_PRECISION);
   fix_char_length(my_decimal_precision_to_length_no_truncation(precision,
@@ -681,19 +681,19 @@ void Item_func::count_only_length(Item **item, uint nitems)
   result length/precision depends on argument ones.
 */
 
-void Item_func::count_real_length()
+void Item_func::count_real_length(Item **item, uint nitems)
 {
   uint32 length= 0;
   decimals= 0;
   max_length= 0;
-  for (uint i=0 ; i < arg_count ; i++)
+  for (uint i=0 ; i < nitems ; i++)
   {
     if (decimals != NOT_FIXED_DEC)
     {
-      set_if_bigger(decimals, args[i]->decimals);
-      set_if_bigger(length, (args[i]->max_length - args[i]->decimals));
+      set_if_bigger(decimals, item[i]->decimals);
+      set_if_bigger(length, (item[i]->max_length - item[i]->decimals));
     }
-    set_if_bigger(max_length, args[i]->max_length);
+    set_if_bigger(max_length, item[i]->max_length);
   }
   if (decimals != NOT_FIXED_DEC)
   {
@@ -811,7 +811,7 @@ void Item_num_op::fix_length_and_dec(void)
   if (r0 == REAL_RESULT || r1 == REAL_RESULT ||
       r0 == STRING_RESULT || r1 ==STRING_RESULT)
   {
-    count_real_length();
+    count_real_length(args, arg_count);
     max_length= float_length(decimals);
     cached_result_type= REAL_RESULT;
   }

sql/item_func.h

@@ -39,6 +39,13 @@ protected:
     0 means get this number from first argument
   */
   uint allowed_arg_cols;
+
+  void count_only_length(Item **item, uint nitems);
+  void count_real_length(Item **item, uint nitems);
+  void count_decimal_length(Item **item, uint nitems);
+  void count_datetime_length(Item **item, uint nitems);
+  bool count_string_result_length(enum_field_types field_type,
+                                  Item **item, uint nitems);
 public:
   uint arg_count;
   table_map used_tables_cache, not_null_tables_cache;
@@ -146,16 +153,10 @@ public:
   virtual void print(String *str, enum_query_type query_type);
   void print_op(String *str, enum_query_type query_type);
   void print_args(String *str, uint from, enum_query_type query_type);
-  void count_only_length(Item **item, uint nitems);
-  void count_real_length();
-  void count_decimal_length();
   inline bool get_arg0_date(MYSQL_TIME *ltime, ulonglong fuzzy_date)
   {
     return (null_value=args[0]->get_date(ltime, fuzzy_date));
   }
-  void count_datetime_length(Item **item, uint nitems);
-  bool count_string_result_length(enum_field_types field_type,
-                                  Item **item, uint nitems);
   inline bool get_arg0_time(MYSQL_TIME *ltime)
   {
     null_value= args[0]->get_time(ltime);
@@ -436,7 +437,7 @@ class Item_func_hybrid_result_type: public Item_func
   }
 protected:
   Item_result cached_result_type;
-
+  void fix_attributes(Item **item, uint nitems);
 public:
   Item_func_hybrid_result_type() :Item_func(), cached_result_type(REAL_RESULT)
   { collation.set_numeric(); }