Bug#28133: Wrong DATE/DATETIME comparison in IN() function.

The IN function was comparing DATE/DATETIME values either as ints or as
strings. Both methods have their disadvantages and may lead to a wrong
result.

Now IN function checks whether all of its arguments has the STRING result
types and at least one of them is a DATE/DATETIME item. If so it uses either
an object of the in_datetime class or an object of the cmp_item_datetime
class to perform its work. If the IN() function arguments are rows then
row columns are checked whether the DATE/DATETIME comparator should be used
to compare them.
The in_datetime class is used to find occurence of the item to be checked
in the vector of the constant DATE/DATETIME values. The cmp_item_datetime
class is used to compare items one by one in the DATE/DATETIME context.
Both classes obtain values from items with help of the get_datetime_value()
function and cache the left item if it is a constant one.



mysql-test/t/type_datetime.test:
  Added a test case for the bug#28133: Wrong DATE/DATETIME comparison in IN() function.
mysql-test/r/type_datetime.result:
  Added a test case for the bug#28133: Wrong DATE/DATETIME comparison in IN() function.
mysql-test/r/func_in.result:
  A test case result is corrected after the fix for the bug#28133.
sql/item_cmpfunc.h:
  Bug#28133: Wrong DATE/DATETIME comparison in IN() function.
  Two DATE/DATETIME comparison classes are added.
  The in_datetime class is used to find occurence of the item to be checked
  in the vector of the constant DATE/DATETIME values. The cmp_item_datetime
  class is used to compare items one by one in the DATE/DATETIME context.
  Both classes obtain values from items with help of the get_datetime_value()
  function and cache the left item if it is a constant one.
sql/item_cmpfunc.cc:
  Bug#28133: Wrong DATE/DATETIME comparison in IN() function.
  Now IN function checks whether all of its arguments has the STRING result
  types and at least one of them is a DATE/DATETIME item. If so it uses either
  an object of the in_datetime class or an object of the cmp_item_datetime
  class to perform its work. If the IN() function arguments are rows then
  row columns are checked whether the DATE/DATETIME comparator should be used
  to compare them.

mysql-test/r/func_in.result

@@ -467,6 +467,7 @@ CREATE TABLE t4 (a DATE);
 INSERT INTO t4 VALUES ('1972-02-06'), ('1972-07-29');
 SELECT * FROM t4 WHERE a IN ('1972-02-06','19772-07-29');
 a
+1972-02-06
 Warnings:
 Warning	1292	Incorrect date value: '19772-07-29' for column 'a' at row 1
 DROP TABLE t1,t2,t3,t4;

mysql-test/r/type_datetime.result

@@ -266,3 +266,49 @@ f2
 SELECT 1 from dual where NOW() BETWEEN CURRENT_DATE() - INTERVAL 1 DAY AND CURRENT_DATE();
 1
 drop table t1;
+create table t1 (f1 date);
+insert into t1 values('01-01-01'),('01-01-02'),('01-01-03');
+select * from t1 where f1 in ('01-01-01','2001-01-02','2001-01-03 00:00:00');
+f1
+2001-01-01
+2001-01-02
+2001-01-03
+create table t2(f2 datetime);
+insert into t2 values('01-01-01 00:00:00'),('01-02-03 12:34:56'),('02-04-06 11:22:33');
+select * from t2 where f2 in ('01-01-01','01-02-03 12:34:56','01-02-03');
+f2
+2001-01-01 00:00:00
+2001-02-03 12:34:56
+select * from t1,t2 where '01-01-02' in (f1, cast(f2 as date));
+f1	f2
+2001-01-02	2001-01-01 00:00:00
+2001-01-02	2001-02-03 12:34:56
+2001-01-02	2002-04-06 11:22:33
+select * from t1,t2 where '01-01-01' in (f1, '01-02-03');
+f1	f2
+2001-01-01	2001-01-01 00:00:00
+2001-01-01	2001-02-03 12:34:56
+2001-01-01	2002-04-06 11:22:33
+select * from t1,t2 where if(1,'01-02-03 12:34:56','') in (f1, f2);
+f1	f2
+2001-01-01	2001-02-03 12:34:56
+2001-01-02	2001-02-03 12:34:56
+2001-01-03	2001-02-03 12:34:56
+create table t3(f3 varchar(20));
+insert into t3 select * from t2;
+select * from t2,t3 where f2 in (f3,'03-04-05');
+f2	f3
+2001-01-01 00:00:00	2001-01-01 00:00:00
+2001-02-03 12:34:56	2001-02-03 12:34:56
+2002-04-06 11:22:33	2002-04-06 11:22:33
+select f1,f2,f3 from t1,t2,t3 where (f1,'1') in ((f2,'1'),(f3,'1'));
+f1	f2	f3
+2001-01-01	2001-01-01 00:00:00	2001-01-01 00:00:00
+2001-01-01	2001-02-03 12:34:56	2001-01-01 00:00:00
+2001-01-01	2002-04-06 11:22:33	2001-01-01 00:00:00
+2001-01-01	2001-01-01 00:00:00	2001-02-03 12:34:56
+2001-01-01	2001-01-01 00:00:00	2002-04-06 11:22:33
+select f1 from t1 where ('1',f1) in (('1','01-01-01'),('1','2001-1-1 0:0:0'),('1','02-02-02'));
+f1
+2001-01-01
+drop table t1,t2,t3;

mysql-test/t/type_datetime.test

@@ -180,3 +180,22 @@ select f2, f3 from t1 where '01-03-10' between f2 and f3;
 select f2 from t1 where DATE(f2) between "2001-4-15" AND "01-4-15";
 SELECT 1 from dual where NOW() BETWEEN CURRENT_DATE() - INTERVAL 1 DAY AND CURRENT_DATE();
 drop table t1;
+
+#
+# Bug#28133: Wrong DATE/DATETIME comparison in IN() function.
+#
+create table t1 (f1 date);
+insert into t1 values('01-01-01'),('01-01-02'),('01-01-03');
+select * from t1 where f1 in ('01-01-01','2001-01-02','2001-01-03 00:00:00');
+create table t2(f2 datetime);
+insert into t2 values('01-01-01 00:00:00'),('01-02-03 12:34:56'),('02-04-06 11:22:33');
+select * from t2 where f2 in ('01-01-01','01-02-03 12:34:56','01-02-03');
+select * from t1,t2 where '01-01-02' in (f1, cast(f2 as date));
+select * from t1,t2 where '01-01-01' in (f1, '01-02-03');
+select * from t1,t2 where if(1,'01-02-03 12:34:56','') in (f1, f2);
+create table t3(f3 varchar(20));
+insert into t3 select * from t2;
+select * from t2,t3 where f2 in (f3,'03-04-05');
+select f1,f2,f3 from t1,t2,t3 where (f1,'1') in ((f2,'1'),(f3,'1'));
+select f1 from t1 where ('1',f1) in (('1','01-01-01'),('1','2001-1-1 0:0:0'),('1','02-02-02'));
+drop table t1,t2,t3;

sql/item_cmpfunc.cc

@@ -780,7 +780,7 @@ get_datetime_value(THD *thd, Item ***item_arg, Item **cache_arg,
       MYSQL_TYPE_DATE ? MYSQL_TIMESTAMP_DATE : MYSQL_TIMESTAMP_DATETIME;
     value= get_date_from_str(thd, str, t_type, warn_item->name, &error);
   }
-  if (item->const_item())
+  if (item->const_item() && cache_arg)
   {
     Item_cache_int *cache= new Item_cache_int();
     /* Mark the cache as non-const to prevent re-caching. */
@@ -2785,7 +2785,6 @@ in_row::in_row(uint elements, Item * item)
   base= (char*) new cmp_item_row[count= elements];
   size= sizeof(cmp_item_row);
   compare= (qsort2_cmp) cmp_row;
-  tmp.store_value(item);
   /*
     We need to reset these as otherwise we will call sort() with
     uninitialized (even if not used) elements
@@ -2837,6 +2836,27 @@ byte *in_longlong::get_value(Item *item)
   return (byte*) &tmp;
 }
 
+void in_datetime::set(uint pos,Item *item)
+{
+  Item **tmp= &item;
+  bool is_null;
+  struct packed_longlong *buff= &((packed_longlong*) base)[pos];
+
+  buff->val= get_datetime_value(thd, &tmp, 0, warn_item, &is_null);
+  buff->unsigned_flag= 1L;
+}
+
+byte *in_datetime::get_value(Item *item)
+{
+  bool is_null;
+  Item **tmp_item= lval_cache ? &lval_cache : &item;
+  tmp.val= get_datetime_value(thd, &tmp_item, &lval_cache, warn_item, &is_null);
+  if (item->null_value)
+    return 0;
+  tmp.unsigned_flag= 1L;
+  return (byte*) &tmp;
+}
+
 in_double::in_double(uint elements)
   :in_vector(elements,sizeof(double),(qsort2_cmp) cmp_double, 0)
 {}
@@ -2941,12 +2961,18 @@ cmp_item_row::~cmp_item_row()
 }
 
 
+void cmp_item_row::alloc_comparators()
+{
+  if (!comparators)
+    comparators= (cmp_item **) current_thd->calloc(sizeof(cmp_item *)*n);
+}
+
+
 void cmp_item_row::store_value(Item *item)
 {
   DBUG_ENTER("cmp_item_row::store_value");
   n= item->cols();
-  if (!comparators)
-    comparators= (cmp_item **) current_thd->calloc(sizeof(cmp_item *)*n);
+  alloc_comparators();
   if (comparators)
   {
     item->bring_value();
@@ -3058,6 +3084,36 @@ cmp_item* cmp_item_decimal::make_same()
 }
 
 
+void cmp_item_datetime::store_value(Item *item)
+{
+  bool is_null;
+  Item **tmp_item= lval_cache ? &lval_cache : &item;
+  value= get_datetime_value(thd, &tmp_item, &lval_cache, warn_item, &is_null);
+}
+
+
+int cmp_item_datetime::cmp(Item *arg)
+{
+  bool is_null;
+  Item **tmp_item= &arg;
+  return value !=
+    get_datetime_value(thd, &tmp_item, 0, warn_item, &is_null);
+}
+
+
+int cmp_item_datetime::compare(cmp_item *ci)
+{
+  cmp_item_datetime *l_cmp= (cmp_item_datetime *)ci;
+  return (value < l_cmp->value) ? -1 : ((value == l_cmp->value) ? 0 : 1);
+}
+
+
+cmp_item *cmp_item_datetime::make_same()
+{
+  return new cmp_item_datetime(warn_item);
+}
+
+
 bool Item_func_in::nulls_in_row()
 {
   Item **arg,**arg_end;
@@ -3133,7 +3189,11 @@ void Item_func_in::fix_length_and_dec()
   Item **arg, **arg_end;
   uint const_itm= 1;
   THD *thd= current_thd;
-  
+  bool datetime_found= FALSE;
+  /* TRUE <=> arguments values will be compared as DATETIMEs. */
+  bool compare_as_datetime= FALSE;
+  Item *date_arg= 0;
+
   if (agg_cmp_type(thd, &cmp_type, args, arg_count))
     return;
 
@@ -3149,58 +3209,148 @@ void Item_func_in::fix_length_and_dec()
       break;
     }
   }
+  /*
+    When comparing rows create the row comparator object beforehand to ease
+    the DATETIME comparison detection procedure.
+  */
+  if (cmp_type == ROW_RESULT)
+  {
+    cmp_item_row *cmp= 0;
+    if (const_itm && !nulls_in_row())
+    {
+      array= new in_row(arg_count-1, 0);
+      cmp= &((in_row*)array)->tmp;
+    }
+    else
+    {
+      if (!(cmp= new cmp_item_row))
+        return;
+      in_item= cmp;
+    }
+    cmp->n= args[0]->cols();
+    cmp->alloc_comparators();
+  }
+  /* All DATE/DATETIME fields/functions has the STRING result type. */
+  if (cmp_type == STRING_RESULT || cmp_type == ROW_RESULT)
+  {
+    uint col, cols= args[0]->cols();
 
+    for (col= 0; col < cols; col++)
+    {
+      bool skip_column= FALSE;
+      /*
+        Check that all items to be compared has the STRING result type and at
+        least one of them is a DATE/DATETIME item.
+      */
+      for (arg= args, arg_end= args + arg_count; arg != arg_end ; arg++)
+      {
+        Item *itm= ((cmp_type == STRING_RESULT) ? arg[0] :
+                    arg[0]->element_index(col));
+        if (itm->result_type() != STRING_RESULT)
+        {
+          skip_column= TRUE;
+          break;
+        }
+        else if (itm->is_datetime())
+        {
+          datetime_found= TRUE;
+          /*
+            Internally all DATE/DATETIME values are converted to the DATETIME
+            type. So try to find a DATETIME item to issue correct warnings.
+          */
+          if (!date_arg)
+            date_arg= itm;
+          else if (itm->field_type() == MYSQL_TYPE_DATETIME)
+          {
+            date_arg= itm;
+            /* All arguments are already checked to have the STRING result. */
+            if (cmp_type == STRING_RESULT)
+              break;
+          }
+        }
+      }
+      if (skip_column)
+        continue;
+      if (datetime_found)
+      {
+        if (cmp_type == ROW_RESULT)
+        {
+          cmp_item **cmp= 0;
+          if (array)
+            cmp= ((in_row*)array)->tmp.comparators + col;
+          else
+            cmp= ((cmp_item_row*)in_item)->comparators + col;
+          *cmp= new cmp_item_datetime(date_arg);
+          /* Reset variables for the next column. */
+          date_arg= 0;
+          datetime_found= FALSE;
+        }
+        else
+          compare_as_datetime= TRUE;
+      }
+    }
+  }
   /*
     Row item with NULLs inside can return NULL or FALSE => 
     they can't be processed as static
   */
   if (const_itm && !nulls_in_row())
   {
-    /*
-      IN must compare INT/DATE/DATETIME/TIMESTAMP columns and constants
-      as int values (the same way as equality does).
-      So we must check here if the column on the left and all the constant 
-      values on the right can be compared as integers and adjust the 
-      comparison type accordingly.
-    */  
-    if (args[0]->real_item()->type() == FIELD_ITEM &&
-        thd->lex->sql_command != SQLCOM_CREATE_VIEW &&
-        thd->lex->sql_command != SQLCOM_SHOW_CREATE &&
-        cmp_type != INT_RESULT)
+    if (compare_as_datetime)
+      array= new in_datetime(date_arg, arg_count - 1);
+    else
     {
-      Field *field= ((Item_field*) (args[0]->real_item()))->field;
-      if (field->can_be_compared_as_longlong())
+      /*
+        IN must compare INT columns and constants as int values (the same
+        way as equality does).
+        So we must check here if the column on the left and all the constant 
+        values on the right can be compared as integers and adjust the 
+        comparison type accordingly.
+      */  
+      if (args[0]->real_item()->type() == FIELD_ITEM &&
+          thd->lex->sql_command != SQLCOM_CREATE_VIEW &&
+          thd->lex->sql_command != SQLCOM_SHOW_CREATE &&
+          cmp_type != INT_RESULT)
       {
-        bool all_converted= TRUE;
-        for (arg=args+1, arg_end=args+arg_count; arg != arg_end ; arg++)
+        Field *field= ((Item_field*) (args[0]->real_item()))->field;
+        if (field->can_be_compared_as_longlong())
         {
-          if (!convert_constant_item (thd, field, &arg[0]))
-            all_converted= FALSE;
+          bool all_converted= TRUE;
+          for (arg=args+1, arg_end=args+arg_count; arg != arg_end ; arg++)
+          {
+            if (!convert_constant_item (thd, field, &arg[0]))
+              all_converted= FALSE;
+          }
+          if (all_converted)
+            cmp_type= INT_RESULT;
         }
-        if (all_converted)
-          cmp_type= INT_RESULT;
       }
-    }
-    switch (cmp_type) {
-    case STRING_RESULT:
-      array=new in_string(arg_count-1,(qsort2_cmp) srtcmp_in, 
-			  cmp_collation.collation);
-      break;
-    case INT_RESULT:
-      array= new in_longlong(arg_count-1);
-      break;
-    case REAL_RESULT:
-      array= new in_double(arg_count-1);
-      break;
-    case ROW_RESULT:
-      array= new in_row(arg_count-1, args[0]);
-      break;
-    case DECIMAL_RESULT:
-      array= new in_decimal(arg_count - 1);
-      break;
-    default:
-      DBUG_ASSERT(0);
-      return;
+      switch (cmp_type) {
+      case STRING_RESULT:
+        array=new in_string(arg_count-1,(qsort2_cmp) srtcmp_in, 
+                            cmp_collation.collation);
+        break;
+      case INT_RESULT:
+        array= new in_longlong(arg_count-1);
+        break;
+      case REAL_RESULT:
+        array= new in_double(arg_count-1);
+        break;
+      case ROW_RESULT:
+        /*
+          The row comparator was created at the beginning but only DATETIME
+          items comparators were initialized. Call store_value() to setup
+          others.
+        */
+        ((in_row*)array)->tmp.store_value(args[0]);
+        break;
+      case DECIMAL_RESULT:
+        array= new in_decimal(arg_count - 1);
+        break;
+      default:
+        DBUG_ASSERT(0);
+        return;
+      }
     }
     if (array && !(thd->is_fatal_error))		// If not EOM
     {
@@ -3219,7 +3369,19 @@ void Item_func_in::fix_length_and_dec()
   }
   else
   {
-    in_item= cmp_item::get_comparator(cmp_type, cmp_collation.collation);
+    if (in_item)
+    {
+      /*
+        The row comparator was created at the beginning but only DATETIME
+        items comparators were initialized. Call store_value() to setup
+        others.
+      */
+      in_item->store_value(args[0]);
+    }
+    else if (compare_as_datetime)
+      in_item= new cmp_item_datetime(date_arg);
+    else
+      in_item= cmp_item::get_comparator(cmp_type, cmp_collation.collation);
     if (cmp_type  == STRING_RESULT)
       in_item->cmp_charset= cmp_collation.collation;
   }

sql/item_cmpfunc.h

@@ -833,6 +833,7 @@ class in_string :public in_vector
 
 class in_longlong :public in_vector
 {
+protected:
   /*
     Here we declare a temporary variable (tmp) of the same type as the
     elements of this vector. tmp is used in finding if a given value is in 
@@ -866,6 +867,30 @@ class in_longlong :public in_vector
   friend int cmp_longlong(void *cmp_arg, packed_longlong *a,packed_longlong *b);
 };
 
+
+/*
+  Class to represent a vector of constant DATE/DATETIME values.
+  Values are obtained with help of the get_datetime_value() function.
+  If the left item is a constant one then its value is cached in the
+  lval_cache variable.
+*/
+class in_datetime :public in_longlong
+{
+public:
+  THD *thd;
+  /* An item used to issue warnings. */
+  Item *warn_item;
+  /* Cache for the left item. */
+  Item *lval_cache;
+
+  in_datetime(Item *warn_item_arg, uint elements)
+    :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);
+  friend int cmp_longlong(void *cmp_arg, packed_longlong *a,packed_longlong *b);
+};
+
 class in_double :public in_vector
 {
   double tmp;
@@ -986,6 +1011,30 @@ class cmp_item_int :public cmp_item
   cmp_item *make_same();
 };
 
+/*
+  Compare items in the DATETIME context.
+  Values are obtained with help of the get_datetime_value() function.
+  If the left item is a constant one then its value is cached in the
+  lval_cache variable.
+*/
+class cmp_item_datetime :public cmp_item
+{
+  ulonglong value;
+public:
+  THD *thd;
+  /* Item used for issuing warnings. */
+  Item *warn_item;
+  /* Cache for the left item. */
+  Item *lval_cache;
+
+  cmp_item_datetime(Item *warn_item_arg)
+    :thd(current_thd), warn_item(warn_item_arg), lval_cache(0) {}
+  void store_value(Item *item);
+  int cmp(Item *arg);
+  int compare(cmp_item *ci);
+  cmp_item *make_same();
+};
+
 class cmp_item_real :public cmp_item
 {
   double value;
@@ -1020,31 +1069,6 @@ class cmp_item_decimal :public cmp_item
 };
 
 
-class cmp_item_row :public cmp_item
-{
-  cmp_item **comparators;
-  uint n;
-public:
-  cmp_item_row(): comparators(0), n(0) {}
-  ~cmp_item_row();
-  void store_value(Item *item);
-  int cmp(Item *arg);
-  int compare(cmp_item *arg);
-  cmp_item *make_same();
-  void store_value_by_template(cmp_item *tmpl, Item *);
-};
-
-
-class in_row :public in_vector
-{
-  cmp_item_row tmp;
-public:
-  in_row(uint elements, Item *);
-  ~in_row();
-  void set(uint pos,Item *item);
-  byte *get_value(Item *item);
-};
-
 /* 
    cmp_item for optimized IN with row (right part string, which never
    be changed)
@@ -1120,6 +1144,34 @@ class Item_func_in :public Item_func_opt_neg
   CHARSET_INFO *compare_collation() { return cmp_collation.collation; }
 };
 
+class cmp_item_row :public cmp_item
+{
+  cmp_item **comparators;
+  uint n;
+public:
+  cmp_item_row(): comparators(0), n(0) {}
+  ~cmp_item_row();
+  void store_value(Item *item);
+  inline void alloc_comparators();
+  int cmp(Item *arg);
+  int compare(cmp_item *arg);
+  cmp_item *make_same();
+  void store_value_by_template(cmp_item *tmpl, Item *);
+  friend void Item_func_in::fix_length_and_dec();
+};
+
+
+class in_row :public in_vector
+{
+  cmp_item_row tmp;
+public:
+  in_row(uint elements, Item *);
+  ~in_row();
+  void set(uint pos,Item *item);
+  byte *get_value(Item *item);
+  friend void Item_func_in::fix_length_and_dec();
+};
+
 /* Functions used by where clause */
 
 class Item_func_isnull :public Item_bool_func