1 files changed, 129 insertions, 66 deletions

diff --git a/sql/item_cmpfunc.h b/sql/item_cmpfunc.h
index 8410c66b034..40d0c2dfe91 100644
--- a/sql/item_cmpfunc.h
+++ b/sql/item_cmpfunc.h
@@ -357,7 +357,7 @@ public:
   }
   Item *neg_transformer(THD *thd);
   virtual Item *negated_item();
-  bool subst_argument_checker(byte **arg) { return TRUE; }
+  bool subst_argument_checker(uchar **arg) { return TRUE; }
 };
 
 class Item_func_not :public Item_bool_func
@@ -572,7 +572,7 @@ public:
     return this;
   }
   bool eq(const Item *item, bool binary_cmp) const;
-  bool subst_argument_checker(byte **arg) { return TRUE; }
+  bool subst_argument_checker(uchar **arg) { return TRUE; }
 };
 
 
@@ -713,49 +713,6 @@ public:
   bool is_null();
 };
 
-
-class Item_func_case :public Item_func
-{
-  int first_expr_num, else_expr_num;
-  enum Item_result cached_result_type;
-  String tmp_value;
-  uint ncases;
-  Item_result cmp_type;
-  DTCollation cmp_collation;
-public:
-  Item_func_case(List<Item> &list, Item *first_expr_arg, Item *else_expr_arg)
-    :Item_func(), first_expr_num(-1), else_expr_num(-1),
-    cached_result_type(INT_RESULT)
-  {
-    ncases= list.elements;
-    if (first_expr_arg)
-    {
-      first_expr_num= list.elements;
-      list.push_back(first_expr_arg);
-    }
-    if (else_expr_arg)
-    {
-      else_expr_num= list.elements;
-      list.push_back(else_expr_arg);
-    }
-    set_arguments(list);
-  }
-  double val_real();
-  longlong val_int();
-  String *val_str(String *);
-  my_decimal *val_decimal(my_decimal *);
-  bool fix_fields(THD *thd, Item **ref);
-  void fix_length_and_dec();
-  uint decimal_precision() const;
-  table_map not_null_tables() const { return 0; }
-  enum Item_result result_type () const { return cached_result_type; }
-  const char *func_name() const { return "case"; }
-  void print(String *str);
-  Item *find_item(String *str);
-  CHARSET_INFO *compare_collation() { return cmp_collation.collation; }
-};
-
-
 /* Functions to handle the optimized IN */
 
 
@@ -778,7 +735,7 @@ public:
      count(elements), used_count(elements) {}
   virtual ~in_vector() {}
   virtual void set(uint pos,Item *item)=0;
-  virtual byte *get_value(Item *item)=0;
+  virtual uchar *get_value(Item *item)=0;
   void sort()
   {
     qsort2(base,used_count,size,compare,collation);
@@ -810,6 +767,7 @@ public:
   {
     return test(compare(collation, base + pos1*size, base + pos2*size));
   }
+  virtual Item_result result_type()= 0;
 };
 
 class in_string :public in_vector
@@ -820,7 +778,7 @@ public:
   in_string(uint elements,qsort2_cmp cmp_func, CHARSET_INFO *cs);
   ~in_string();
   void set(uint pos,Item *item);
-  byte *get_value(Item *item);
+  uchar *get_value(Item *item);
   Item* create_item()
   { 
     return new Item_string(collation);
@@ -831,6 +789,7 @@ public:
     Item_string *to= (Item_string*)item;
     to->str_value= *str;
   }
+  Item_result result_type() { return STRING_RESULT; }
 };
 
 class in_longlong :public in_vector
@@ -849,7 +808,7 @@ protected:
 public:
   in_longlong(uint elements);
   void set(uint pos,Item *item);
-  byte *get_value(Item *item);
+  uchar *get_value(Item *item);
   
   Item* create_item()
   { 
@@ -865,6 +824,7 @@ public:
     ((Item_int*) item)->unsigned_flag= (my_bool)
       ((packed_longlong*) base)[pos].unsigned_flag;
   }
+  Item_result result_type() { return INT_RESULT; }
 
   friend int cmp_longlong(void *cmp_arg, packed_longlong *a,packed_longlong *b);
 };
@@ -889,35 +849,37 @@ public:
     :in_longlong(elements), thd(current_thd), warn_item(warn_item_arg),
      lval_cache(0) {};
   void set(uint pos,Item *item);
-  byte *get_value(Item *item);
+  uchar *get_value(Item *item);
   friend int cmp_longlong(void *cmp_arg, packed_longlong *a,packed_longlong *b);
 };
 
+
 class in_double :public in_vector
 {
   double tmp;
 public:
   in_double(uint elements);
   void set(uint pos,Item *item);
-  byte *get_value(Item *item);
+  uchar *get_value(Item *item);
   Item *create_item()
   { 
-    return new Item_float(0.0);
+    return new Item_float(0.0, 0);
   }
   void value_to_item(uint pos, Item *item)
   {
     ((Item_float*)item)->value= ((double*) base)[pos];
   }
-
+  Item_result result_type() { return REAL_RESULT; }
 };
 
+
 class in_decimal :public in_vector
 {
   my_decimal val;
 public:
   in_decimal(uint elements);
   void set(uint pos, Item *item);
-  byte *get_value(Item *item);
+  uchar *get_value(Item *item);
   Item *create_item()
   { 
     return new Item_decimal(0, FALSE);
@@ -928,6 +890,8 @@ public:
     Item_decimal *item_dec= (Item_decimal*)item;
     item_dec->set_decimal_value(dec);
   }
+  Item_result result_type() { return DECIMAL_RESULT; }
+
 };
 
 
@@ -958,7 +922,9 @@ class cmp_item_string :public cmp_item
 protected:
   String *value_res;
 public:
+  cmp_item_string () {}
   cmp_item_string (CHARSET_INFO *cs) { cmp_charset= cs; }
+  void set_charset(CHARSET_INFO *cs) { cmp_charset= cs; }
   friend class cmp_item_sort_string;
   friend class cmp_item_sort_string_in_static;
 };
@@ -969,6 +935,8 @@ protected:
   char value_buff[STRING_BUFFER_USUAL_SIZE];
   String value;
 public:
+  cmp_item_sort_string():
+    cmp_item_string() {}
   cmp_item_sort_string(CHARSET_INFO *cs):
     cmp_item_string(cs),
     value(value_buff, sizeof(value_buff), cs) {}
@@ -990,6 +958,11 @@ public:
     return sortcmp(value_res, l_cmp->value_res, cmp_charset);
   } 
   cmp_item *make_same();
+  void set_charset(CHARSET_INFO *cs)
+  {
+    cmp_charset= cs;
+    value.set_quick(value_buff, sizeof(value_buff), cs);
+  }
 };
 
 class cmp_item_int :public cmp_item
@@ -1104,22 +1077,107 @@ public:
   }
 };
 
+
+/*
+  The class Item_func_case is the CASE ... WHEN ... THEN ... END function
+  implementation.
+
+  When there is no expression between CASE and the first WHEN 
+  (the CASE expression) then this function simple checks all WHEN expressions
+  one after another. When some WHEN expression evaluated to TRUE then the
+  value of the corresponding THEN expression is returned.
+
+  When the CASE expression is specified then it is compared to each WHEN
+  expression individually. When an equal WHEN expression is found
+  corresponding THEN expression is returned.
+  In order to do correct comparisons several comparators are used. One for
+  each result type. Different result types that are used in particular
+  CASE ... END expression are collected in the fix_length_and_dec() member
+  function and only comparators for there result types are used.
+*/
+
+class Item_func_case :public Item_func
+{
+  int first_expr_num, else_expr_num;
+  enum Item_result cached_result_type, left_result_type;
+  String tmp_value;
+  uint ncases;
+  Item_result cmp_type;
+  DTCollation cmp_collation;
+  cmp_item *cmp_items[5]; /* For all result types */
+  cmp_item *case_item;
+public:
+  Item_func_case(List<Item> &list, Item *first_expr_arg, Item *else_expr_arg)
+    :Item_func(), first_expr_num(-1), else_expr_num(-1),
+    cached_result_type(INT_RESULT), left_result_type(INT_RESULT), case_item(0)
+  {
+    ncases= list.elements;
+    if (first_expr_arg)
+    {
+      first_expr_num= list.elements;
+      list.push_back(first_expr_arg);
+    }
+    if (else_expr_arg)
+    {
+      else_expr_num= list.elements;
+      list.push_back(else_expr_arg);
+    }
+    set_arguments(list);
+    bzero(&cmp_items, sizeof(cmp_items));
+  }
+  double val_real();
+  longlong val_int();
+  String *val_str(String *);
+  my_decimal *val_decimal(my_decimal *);
+  bool fix_fields(THD *thd, Item **ref);
+  void fix_length_and_dec();
+  uint decimal_precision() const;
+  table_map not_null_tables() const { return 0; }
+  enum Item_result result_type () const { return cached_result_type; }
+  const char *func_name() const { return "case"; }
+  void print(String *str);
+  Item *find_item(String *str);
+  CHARSET_INFO *compare_collation() { return cmp_collation.collation; }
+  void cleanup();
+};
+
+/*
+  The Item_func_in class implements the in_expr IN(values_list) function.
+
+  The current implementation distinguishes 2 cases:
+  1) all items in the value_list are constants and have the same
+    result type. This case is handled by in_vector class.
+  2) items in the value_list have different result types or there is some
+    non-constant items.
+    In this case Item_func_in employs several cmp_item objects to performs
+    comparisons of in_expr and an item from the values_list. One cmp_item
+    object for each result type. Different result types are collected in the
+    fix_length_and_dec() member function by means of collect_cmp_types()
+    function.
+*/
 class Item_func_in :public Item_func_opt_neg
 {
 public:
-  Item_result cmp_type;
   /* 
     an array of values when the right hand arguments of IN
     are all SQL constant and there are no nulls 
   */
   in_vector *array;
-  cmp_item *in_item;
   bool have_null;
+  /* 
+    true when all arguments of the IN clause are of compatible types
+    and can be used safely as comparisons for key conditions
+  */
+  bool arg_types_compatible;
+  Item_result left_result_type;
+  cmp_item *cmp_items[6]; /* One cmp_item for each result type */
   DTCollation cmp_collation;
 
   Item_func_in(List<Item> &list)
-    :Item_func_opt_neg(list), array(0), in_item(0), have_null(0)
+    :Item_func_opt_neg(list), array(0), have_null(0),
+    arg_types_compatible(FALSE)
   {
+    bzero(&cmp_items, sizeof(cmp_items));
     allowed_arg_cols= 0;  // Fetch this value from first argument
   }
   longlong val_int();
@@ -1128,12 +1186,16 @@ public:
   uint decimal_precision() const { return 1; }
   void cleanup()
   {
+    uint i;
     DBUG_ENTER("Item_func_in::cleanup");
     Item_int_func::cleanup();
     delete array;
-    delete in_item;
     array= 0;
-    in_item= 0;
+    for (i= 0; i <= (uint)DECIMAL_RESULT + 1; i++)
+    {
+      delete cmp_items[i];
+      cmp_items[i]= 0;
+    }
     DBUG_VOID_RETURN;
   }
   optimize_type select_optimize() const
@@ -1170,8 +1232,9 @@ public:
   in_row(uint elements, Item *);
   ~in_row();
   void set(uint pos,Item *item);
-  byte *get_value(Item *item);
+  uchar *get_value(Item *item);
   friend void Item_func_in::fix_length_and_dec();
+  Item_result result_type() { return ROW_RESULT; }
 };
 
 /* Functions used by where clause */
@@ -1375,13 +1438,13 @@ public:
                          COND **conds);
   void top_level_item() { abort_on_null=1; }
   void copy_andor_arguments(THD *thd, Item_cond *item);
-  bool walk(Item_processor processor, byte *arg);
-  Item *transform(Item_transformer transformer, byte *arg);
+  bool walk(Item_processor processor, bool walk_subquery, uchar *arg);
+  Item *transform(Item_transformer transformer, uchar *arg);
   void traverse_cond(Cond_traverser, void *arg, traverse_order order);
   void neg_arguments(THD *thd);
-  bool subst_argument_checker(byte **arg) { return TRUE; }
-  Item *compile(Item_analyzer analyzer, byte **arg_p,
-                Item_transformer transformer, byte *arg_t);
+  bool subst_argument_checker(uchar **arg) { return TRUE; }
+  Item *compile(Item_analyzer analyzer, uchar **arg_p,
+                Item_transformer transformer, uchar *arg_t);
 };
 
 
@@ -1490,8 +1553,8 @@ public:
   void fix_length_and_dec();
   bool fix_fields(THD *thd, Item **ref);
   void update_used_tables();
-  bool walk(Item_processor processor, byte *arg);
-  Item *transform(Item_transformer transformer, byte *arg);
+  bool walk(Item_processor processor, bool walk_subquery, uchar *arg);
+  Item *transform(Item_transformer transformer, uchar *arg);
   void print(String *str);
   CHARSET_INFO *compare_collation() 
   { return fields.head()->collation.collation; }