MDEV-8707 Wrong result for

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

7 files changed, 113 insertions, 27 deletions

diff --git a/sql/field.cc b/sql/field.cc
index 891ab931e3f..f48eb46657b 100644
--- a/sql/field.cc
+++ b/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);
diff --git a/sql/field.h b/sql/field.h
index cc701ef16c6..de9a6cd2a96 100644
--- a/sql/field.h
+++ b/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 */
diff --git a/sql/item.cc b/sql/item.cc
index c624ebfef73..769ca9d2a49 100644
--- a/sql/item.cc
+++ b/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))
   {
diff --git a/sql/item.h b/sql/item.h
index 86dcd7abdbb..127f1cf138f 100644
--- a/sql/item.h
+++ b/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
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index c9f50379a67..ca7dcc03ef1 100644
--- a/sql/item_cmpfunc.cc
+++ b/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),
-                                      set_null))
+      if (comparators[i].set_cmp_func_and_arg_cmp_context(owner,
+                                                          (*a)->addr(i),
+                                                          (*b)->addr(i),
+                                                          set_null))
         return 1;
     }
     break;
diff --git a/sql/item_cmpfunc.h b/sql/item_cmpfunc.h
index 88f329cc519..2d986b39c39 100644
--- a/sql/item_cmpfunc.h
+++ b/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]
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 9d95bdfd617..8d60be54914 100644
--- a/sql/sql_select.cc
+++ b/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,28 +12638,36 @@ 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))
     {
+      /*
+        field_item and const_item are arguments of a scalar or a row
+        comparison function:
+          WHERE column=constant
+          WHERE (column, ...) = (constant, ...)
+
+        The owner comparison function has previously called fix_fields(),
+        so field_item and const_item should be directly comparable items,
+        field_item->cmp_context and const_item->cmp_context should be set.
+        In case of string comparison, charsets and collations of
+        field_item and const_item should have already be aggregated
+        for comparison, all necessary character set converters installed
+        and fixed.
+
+        In case of string comparison, const_item can be either:
+        - a weaker constant that does not need to be converted to field_item:
+            WHERE latin1_field = 'latin1_const'
+            WHERE varbinary_field = 'latin1_const'
+            WHERE latin1_bin_field = 'latin1_general_ci_const'
+        - a stronger constant that does not need to be converted to field_item:
+            WHERE latin1_field = binary 0xDF
+            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;
 
-      if (field_item->cmp_type() == STRING_RESULT)
-      {
-        CHARSET_INFO *cs= field_item->field->charset();
-        if (!item)
-        {
-          Item_func_eq *eq_item;
-          if (!(eq_item= new (thd->mem_root) Item_func_eq(thd, orig_left_item,
-                                                          orig_right_item)) ||
-              eq_item->set_cmp_func_and_arg_cmp_context())
-            return FALSE;
-          eq_item->quick_fix_field();
-          item= eq_item;
-        }  
-        if ((cs != item->compare_collation()) ||
-            !cs->coll->propagate(cs, 0, 0))
-          return FALSE;
-      }
-
       Item_equal *item_equal = find_item_equal(cond_equal,
                                                field_item->field, &copyfl);
       if (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)