MDEV-8707 Wrong result for

SELECT..WHERE varchar_column=DATE'2001-01-01' AND varchar_column='2001-01-01'

mysql-test/r/ctype_latin1.result

@@ -8039,5 +8039,69 @@ Warnings:
 Note	1003	select `test`.`t1`.`a` AS `a` from `test`.`t1` where ((`test`.`t1`.`a` = 'a') and (`test`.`t1`.`a` between 'a' and <cache>(('c' collate latin1_bin))))
 DROP TABLE t1;
 #
+# MDEV-8707 Wrong result for SELECT..WHERE varchar_column=DATE'2001-01-01' AND varchar_column='2001-01-01'
+#
+SET NAMES latin1;
+CREATE TABLE t1 (a VARCHAR(40));
+INSERT INTO t1 VALUES ('2001-01-01'),('2001-01-01x');
+SELECT * FROM t1 WHERE a=DATE'2001-01-01' AND a='2001-01-01';
+a
+2001-01-01
+SELECT * FROM t1 WHERE a='2001-01-01' AND a=DATE'2001-01-01';
+a
+2001-01-01
+SELECT * FROM t1 WHERE (a,a)=('2001-01-01x',DATE'2001-01-01');
+a
+2001-01-01x
+Warnings:
+Warning	1292	Truncated incorrect date value: '2001-01-01x'
+SELECT * FROM t1 WHERE (a,a)=(DATE'2001-01-01','2001-01-01x');
+a
+2001-01-01x
+Warnings:
+Warning	1292	Truncated incorrect date value: '2001-01-01x'
+SELECT * FROM t1 WHERE (a,a)=('2001-01-01',DATE'2001-01-01');
+a
+2001-01-01
+SELECT * FROM t1 WHERE (a,a)=(DATE'2001-01-01','2001-01-01');
+a
+2001-01-01
+DROP TABLE t1;
+CREATE TABLE t1 (a ENUM('2001-01-01','2001-01-01x'));
+INSERT INTO t1 VALUES ('2001-01-01'),('2001-01-01x');
+SELECT * FROM t1 WHERE a=DATE'2001-01-01' AND a='2001-01-01';
+a
+2001-01-01
+SELECT * FROM t1 WHERE a='2001-01-01' AND a=DATE'2001-01-01';
+a
+2001-01-01
+SELECT * FROM t1 WHERE (a,a)=('2001-01-01x',DATE'2001-01-01');
+a
+2001-01-01x
+Warnings:
+Warning	1292	Truncated incorrect date value: '2001-01-01x'
+SELECT * FROM t1 WHERE (a,a)=(DATE'2001-01-01','2001-01-01x');
+a
+2001-01-01x
+Warnings:
+Warning	1292	Truncated incorrect date value: '2001-01-01x'
+SELECT * FROM t1 WHERE (a,a)=('2001-01-01',DATE'2001-01-01');
+a
+2001-01-01
+SELECT * FROM t1 WHERE (a,a)=(DATE'2001-01-01','2001-01-01');
+a
+2001-01-01
+DROP TABLE t1;
+CREATE TABLE t1 (a VARCHAR(40),b VARCHAR(40));
+INSERT INTO t1 VALUES ('2001-01-01','2001-01-01x');
+SELECT * FROM t1 WHERE a=b AND a=DATE'2001-01-01';
+a	b
+DROP TABLE t1;
+CREATE TABLE t1 (a ENUM('2001-01-01','2001-01-01x'),b ENUM('2001-01-01','2001-01-01x'));
+INSERT INTO t1 VALUES ('2001-01-01','2001-01-01x');
+SELECT * FROM t1 WHERE a=b AND a=DATE'2001-01-01';
+a	b
+DROP TABLE t1;
+#
 # End of 10.1 tests
 #

mysql-test/t/ctype_latin1.test

@@ -298,6 +298,37 @@ SELECT * FROM t1 WHERE a BETWEEN 'a' AND 'c' COLLATE latin1_bin AND a='a';
 EXPLAIN EXTENDED SELECT * FROM t1 WHERE a BETWEEN 'a' AND 'c' COLLATE latin1_bin AND a='a';
 DROP TABLE t1;
 
+--echo #
+--echo # MDEV-8707 Wrong result for SELECT..WHERE varchar_column=DATE'2001-01-01' AND varchar_column='2001-01-01'
+--echo #
+SET NAMES latin1;
+CREATE TABLE t1 (a VARCHAR(40));
+INSERT INTO t1 VALUES ('2001-01-01'),('2001-01-01x');
+SELECT * FROM t1 WHERE a=DATE'2001-01-01' AND a='2001-01-01';
+SELECT * FROM t1 WHERE a='2001-01-01' AND a=DATE'2001-01-01';
+SELECT * FROM t1 WHERE (a,a)=('2001-01-01x',DATE'2001-01-01');
+SELECT * FROM t1 WHERE (a,a)=(DATE'2001-01-01','2001-01-01x');
+SELECT * FROM t1 WHERE (a,a)=('2001-01-01',DATE'2001-01-01');
+SELECT * FROM t1 WHERE (a,a)=(DATE'2001-01-01','2001-01-01');
+DROP TABLE t1;
+CREATE TABLE t1 (a ENUM('2001-01-01','2001-01-01x'));
+INSERT INTO t1 VALUES ('2001-01-01'),('2001-01-01x');
+SELECT * FROM t1 WHERE a=DATE'2001-01-01' AND a='2001-01-01';
+SELECT * FROM t1 WHERE a='2001-01-01' AND a=DATE'2001-01-01';
+SELECT * FROM t1 WHERE (a,a)=('2001-01-01x',DATE'2001-01-01');
+SELECT * FROM t1 WHERE (a,a)=(DATE'2001-01-01','2001-01-01x');
+SELECT * FROM t1 WHERE (a,a)=('2001-01-01',DATE'2001-01-01');
+SELECT * FROM t1 WHERE (a,a)=(DATE'2001-01-01','2001-01-01');
+DROP TABLE t1;
+CREATE TABLE t1 (a VARCHAR(40),b VARCHAR(40));
+INSERT INTO t1 VALUES ('2001-01-01','2001-01-01x');
+SELECT * FROM t1 WHERE a=b AND a=DATE'2001-01-01';
+DROP TABLE t1;
+CREATE TABLE t1 (a ENUM('2001-01-01','2001-01-01x'),b ENUM('2001-01-01','2001-01-01x'));
+INSERT INTO t1 VALUES ('2001-01-01','2001-01-01x');
+SELECT * FROM t1 WHERE a=b AND a=DATE'2001-01-01';
+DROP TABLE t1;
+
 --echo #
 --echo # End of 10.1 tests
 --echo #

sql/field.cc

@@ -1246,6 +1246,21 @@ double Field::pos_in_interval_val_str(Field *min, Field *max, uint data_offset)
 }
 
 
+bool Field::test_if_equality_guarantees_uniqueness(const Item *item) const
+{
+  DBUG_ASSERT(cmp_type() != STRING_RESULT); // For STRING_RESULT see Field_str
+  /*
+    We use result_type() rather than cmp_type() in the below condition,
+    because it covers a special case that string literals guarantee uniqueness
+    for temporal columns, so the query:
+      WHERE temporal_column='string'
+    cannot return multiple distinct temporal values.
+    QQ: perhaps we could allow INT/DECIMAL/DOUBLE types for temporal items.
+  */
+  return result_type() == item->result_type();
+}
+
+
 /*
   This handles all numeric and BIT data types.
 */ 
@@ -1843,6 +1858,32 @@ Field_str::Field_str(uchar *ptr_arg,uint32 len_arg, uchar *null_ptr_arg,
 }
 
 
+bool Field_str::test_if_equality_guarantees_uniqueness(const Item *item) const
+{
+  /*
+    Can't guarantee uniqueness when comparing a CHAR/VARCHAR/TEXT,
+    BINARY/VARBINARY/BLOB, ENUM,SET columns to an item with cmp_type()
+    of INT_RESULT, DOUBLE_RESULT, DECIMAL_RESULT or TIME_RESULT.
+    Example:
+      SELECT * FROM t1 WHERE varchar_column=DATE'2001-01-01'
+    return non-unuque values, e.g. '2001-01-01' and '2001-01-01x'.
+  */
+  if (!field_charset->coll->propagate(field_charset, 0, 0) ||
+      item->cmp_type() != STRING_RESULT)
+    return false;
+  /*
+    Can't guarantee uniqueness when comparing to
+    an item of a different collation.
+    Example:
+      SELECT * FROM t1
+      WHERE latin1_bin_column = _latin1'A' COLLATE latin1_swedish_ci
+    return non-unique values 'a' and 'A'.
+  */
+  DTCollation tmp(field_charset, field_derivation, repertoire());
+  return !tmp.aggregate(item->collation) && tmp.collation == field_charset;
+}
+
+
 void Field_num::make_field(Send_field *field)
 {
   Field::make_field(field);

sql/field.h

@@ -1012,6 +1012,21 @@ public:
     return (double) 0.5; 
   }
 
+  /*
+    Check if comparison between the field and an item unambiguously
+    identifies a distinct field value.
+
+    Example1: SELECT * FROM t1 WHERE int_column=10;
+              This example returns distinct integer value of 10.
+
+    Example2: SELECT * FROM t1 WHERE varchar_column=DATE'2001-01-01'
+              This example returns non-distinct values.
+              Comparison as DATE will return '2001-01-01' and '2001-01-01x',
+              but these two values are not equal to each other as VARCHARs.
+    See also the function with the same name in sql_select.cc.
+  */
+  virtual bool test_if_equality_guarantees_uniqueness(const Item *const_item)
+                                                      const;
   virtual bool can_optimize_keypart_ref(const Item_bool_func *cond,
                                         const Item *item) const;
   virtual bool can_optimize_hash_join(const Item_bool_func *cond,
@@ -1187,6 +1202,7 @@ public:
   {
     return pos_in_interval_val_str(min, max, length_size());
   }
+  bool test_if_equality_guarantees_uniqueness(const Item *const_item) const;
 };
 
 /* base class for Field_string, Field_varstring and Field_blob */

sql/item.cc

@@ -1814,7 +1814,7 @@ void Item::split_sum_func2(THD *thd, Item **ref_pointer_array,
 
 
 static bool
-left_is_superset(DTCollation *left, DTCollation *right)
+left_is_superset(const DTCollation *left, const DTCollation *right)
 {
   /* Allow convert to Unicode */
   if (left->collation->state & MY_CS_UNICODE &&
@@ -1873,7 +1873,7 @@ left_is_superset(DTCollation *left, DTCollation *right)
   @endcode
 */
 
-bool DTCollation::aggregate(DTCollation &dt, uint flags)
+bool DTCollation::aggregate(const DTCollation &dt, uint flags)
 {
   if (!my_charset_same(collation, dt.collation))
   {

sql/item.h

@@ -127,6 +127,13 @@ public:
     derivation= derivation_arg;
     set_repertoire_from_charset(collation_arg);
   }
+  DTCollation(CHARSET_INFO *collation_arg,
+              Derivation derivation_arg,
+              uint repertoire_arg)
+   :collation(collation_arg),
+    derivation(derivation_arg),
+    repertoire(repertoire_arg)
+  { }
   void set(const DTCollation &dt)
   { 
     collation= dt.collation;
@@ -160,7 +167,7 @@ public:
   }
   void set(Derivation derivation_arg)
   { derivation= derivation_arg; }
-  bool aggregate(DTCollation &dt, uint flags= 0);
+  bool aggregate(const DTCollation &dt, uint flags= 0);
   bool set(DTCollation &dt1, DTCollation &dt2, uint flags= 0)
   { set(dt1); return aggregate(dt2, flags); }
   const char *derivation_name() const

sql/item_cmpfunc.cc

@@ -553,8 +553,10 @@ int Arg_comparator::set_compare_func(Item_func_or_sum *item, Item_result type)
 	my_error(ER_OPERAND_COLUMNS, MYF(0), (*a)->element_index(i)->cols());
 	return 1;
       }
-      if (comparators[i].set_cmp_func(owner, (*a)->addr(i), (*b)->addr(i),
+      if (comparators[i].set_cmp_func_and_arg_cmp_context(owner,
-                                      set_null))
+                                                          (*a)->addr(i),
+                                                          (*b)->addr(i),
+                                                          set_null))
         return 1;
     }
     break;

sql/item_cmpfunc.h

@@ -79,6 +79,17 @@ public:
                         item_cmp_type((*a1)->cmp_type(),
                                       (*a2)->cmp_type()));
   }
+  int set_cmp_func_and_arg_cmp_context(Item_func_or_sum *owner_arg,
+                                       Item **a1, Item **a2,
+                                       bool set_null_arg)
+  {
+    set_null= set_null_arg;
+    Item_result type= item_cmp_type((*a1)->cmp_type(), (*a2)->cmp_type());
+    int rc= set_cmp_func(owner_arg, a1, a2, type);
+    if (!rc)
+      (*a1)->cmp_context= (*a2)->cmp_context= type;
+    return rc;
+  }
   inline int compare() { return (this->*func)(); }
 
   int compare_string();		 // compare args[0] & args[1]

sql/sql_select.cc

@@ -12513,7 +12513,6 @@ finish:
 */
 
 static bool check_simple_equality(THD *thd, Item *left_item, Item *right_item,
-                                  const Item_bool_func *item,
                                   COND_EQUAL *cond_equal)
 {
   Item *orig_left_item= left_item;
@@ -12639,27 +12638,35 @@ static bool check_simple_equality(THD *thd, Item *left_item, Item *right_item,
     }
 
     if (const_item &&
-        field_item->result_type() == const_item->result_type())
+        field_item->field->test_if_equality_guarantees_uniqueness(const_item))
     {
-      bool copyfl;
+      /*
+        field_item and const_item are arguments of a scalar or a row
+        comparison function:
+          WHERE column=constant
+          WHERE (column, ...) = (constant, ...)
 
-      if (field_item->cmp_type() == STRING_RESULT)
+        The owner comparison function has previously called fix_fields(),
-      {
+        so field_item and const_item should be directly comparable items,
-        CHARSET_INFO *cs= field_item->field->charset();
+        field_item->cmp_context and const_item->cmp_context should be set.
-        if (!item)
+        In case of string comparison, charsets and collations of
-        {
+        field_item and const_item should have already be aggregated
-          Item_func_eq *eq_item;
+        for comparison, all necessary character set converters installed
-          if (!(eq_item= new (thd->mem_root) Item_func_eq(thd, orig_left_item,
+        and fixed.
-                                                          orig_right_item)) ||
+
-              eq_item->set_cmp_func_and_arg_cmp_context())
+        In case of string comparison, const_item can be either:
-            return FALSE;
+        - a weaker constant that does not need to be converted to field_item:
-          eq_item->quick_fix_field();
+            WHERE latin1_field = 'latin1_const'
-          item= eq_item;
+            WHERE varbinary_field = 'latin1_const'
-        }  
+            WHERE latin1_bin_field = 'latin1_general_ci_const'
-        if ((cs != item->compare_collation()) ||
+        - a stronger constant that does not need to be converted to field_item:
-            !cs->coll->propagate(cs, 0, 0))
+            WHERE latin1_field = binary 0xDF
-          return FALSE;
+            WHERE latin1_field = 'a' COLLATE latin1_bin
-      }
+        - a result of conversion (e.g. from the session character set)
+          to the character set of field_item:
+            WHERE latin1_field = 'utf8_string_with_latin1_repertoire'
+      */
+      bool copyfl;
 
       Item_equal *item_equal = find_item_equal(cond_equal,
                                                field_item->field, &copyfl);
@@ -12737,7 +12744,7 @@ static bool check_row_equality(THD *thd, Item *left_row, Item_row *right_row,
     }
     else
     { 
-      is_converted= check_simple_equality(thd, left_item, right_item, 0,
+      is_converted= check_simple_equality(thd, left_item, right_item,
                                           cond_equal);
     }  
  
@@ -12799,7 +12806,7 @@ bool Item_func_eq::check_equality(THD *thd, COND_EQUAL *cond_equal,
                               (Item_row *) right_item,
                               cond_equal, eq_list);
   }
-  return check_simple_equality(thd, left_item, right_item, this, cond_equal);
+  return check_simple_equality(thd, left_item, right_item, cond_equal);
 }
 
                           
@@ -15457,6 +15464,8 @@ Item_func_isnull::remove_eq_conds(THD *thd, Item::cond_result *cond_value,
 /*
   psergey-todo: this returns false for int_column='1234' (here '1234' is a
   constant. Need to discuss this with Bar).
+
+  See also Field::test_if_equality_guaranees_uniqueness(const Item *item);
 */
 static bool
 test_if_equality_guarantees_uniqueness(Item *l, Item *r)