MDEV-6679 Different optimizer plan for "a BETWEEN 'string' AND ?" and "a BETWEEN ? AND 'string'"

Item_string::eq() and Item_param::eq() in string context behaved differently.
Introducing a new class Item_basic_value to share the eq() code between
literals (Item_int, Item_double, Item_string, Item_null) and Item_param.

mysql-test/r/ctype_utf8.result

@@ -5962,5 +5962,32 @@ NULL
 DEALLOCATE PREPARE stmt;
 SET NAMES utf8;
 #
+# MDEV-6679 Different optimizer plan for "a BETWEEN 'string' AND ?" and "a BETWEEN ? AND 'string'"
+#
+SET NAMES utf8, collation_connection=utf8_swedish_ci;
+CREATE TABLE t1 (a VARCHAR(10) CHARACTER SET utf8, b INT NOT NULL DEFAULT 0, key(a));
+INSERT INTO t1 (a) VALUES ('a'),('b'),('c'),('d'),('¢');
+SET @arg='¢';
+PREPARE stmt FROM "EXPLAIN SELECT * FROM t1 WHERE a BETWEEN _utf8'¢' and ?";
+EXECUTE stmt USING @arg;
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	range	a	a	33	NULL	1	Using index condition
+PREPARE stmt FROM "EXPLAIN SELECT * FROM t1 WHERE a between ? and _utf8'¢'";
+EXECUTE stmt USING @arg;
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	range	a	a	33	NULL	1	Using index condition
+DEALLOCATE PREPARE stmt;
+ALTER TABLE t1 CONVERT TO CHARACTER SET latin1;
+PREPARE stmt FROM "EXPLAIN SELECT * FROM t1 WHERE a BETWEEN _utf8'¢' and ?";
+EXECUTE stmt USING @arg;
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	range	a	a	13	NULL	1	Using index condition
+PREPARE stmt FROM "EXPLAIN SELECT * FROM t1 WHERE a between ? and _utf8'¢'";
+EXECUTE stmt USING @arg;
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	range	a	a	13	NULL	1	Using index condition
+DEALLOCATE PREPARE stmt;
+DROP TABLE t1;
+#
 # End of 10.0 tests
 #

mysql-test/t/ctype_utf8.test

@@ -1681,6 +1681,27 @@ EXECUTE stmt USING @no_such_var;
 DEALLOCATE PREPARE stmt;
 SET NAMES utf8;
 
+--echo #
+--echo # MDEV-6679 Different optimizer plan for "a BETWEEN 'string' AND ?" and "a BETWEEN ? AND 'string'"
+--echo #
+SET NAMES utf8, collation_connection=utf8_swedish_ci;
+CREATE TABLE t1 (a VARCHAR(10) CHARACTER SET utf8, b INT NOT NULL DEFAULT 0, key(a));
+INSERT INTO t1 (a) VALUES ('a'),('b'),('c'),('d'),('¢');
+SET @arg='¢';
+PREPARE stmt FROM "EXPLAIN SELECT * FROM t1 WHERE a BETWEEN _utf8'¢' and ?";
+EXECUTE stmt USING @arg;
+PREPARE stmt FROM "EXPLAIN SELECT * FROM t1 WHERE a between ? and _utf8'¢'";
+EXECUTE stmt USING @arg;
+DEALLOCATE PREPARE stmt;
+
+ALTER TABLE t1 CONVERT TO CHARACTER SET latin1;
+PREPARE stmt FROM "EXPLAIN SELECT * FROM t1 WHERE a BETWEEN _utf8'¢' and ?";
+EXECUTE stmt USING @arg;
+PREPARE stmt FROM "EXPLAIN SELECT * FROM t1 WHERE a between ? and _utf8'¢'";
+EXECUTE stmt USING @arg;
+DEALLOCATE PREPARE stmt;
+DROP TABLE t1;
+
 --echo #
 --echo # End of 10.0 tests
 --echo #

sql/item.cc

@@ -1295,19 +1295,6 @@ Item *Item_param::safe_charset_converter(CHARSET_INFO *tocs)
 }
 
 
-bool Item_string::eq(const Item *item, bool binary_cmp) const
-{
-  if (type() == item->type() && item->basic_const_item())
-  {
-    if (binary_cmp)
-      return !stringcmp(&str_value, &item->str_value);
-    return (collation.collation == item->collation.collation &&
-	    !sortcmp(&str_value, &item->str_value, collation.collation));
-  }
-  return 0;
-}
-
-
 /**
   Get the value of the function as a MYSQL_TIME structure.
   As a extra convenience the time structure is reset on error or NULL values!
@@ -3121,10 +3108,6 @@ my_decimal *Item_string::val_decimal(my_decimal *decimal_value)
 }
 
 
-bool Item_null::eq(const Item *item, bool binary_cmp) const
-{ return item->type() == type(); }
-
-
 double Item_null::val_real()
 {
   // following assert is redundant, because fixed=1 assigned in constructor
@@ -3808,30 +3791,21 @@ Item_param::clone_item()
 
 
 bool
-Item_param::eq(const Item *arg, bool binary_cmp) const
+Item_param::eq(const Item *item, bool binary_cmp) const
 {
-  Item *item;
-  if (!basic_const_item() || !arg->basic_const_item() || arg->type() != type())
+  if (!basic_const_item())
     return FALSE;
-  /*
-    We need to cast off const to call val_int(). This should be OK for
-    a basic constant.
-  */
-  item= (Item*) arg;
 
   switch (state) {
   case NULL_VALUE:
-    return TRUE;
+    return null_eq(item);
   case INT_VALUE:
-    return value.integer == item->val_int() &&
-           unsigned_flag == item->unsigned_flag;
+    return int_eq(value.integer, item);
   case REAL_VALUE:
-    return value.real == item->val_real();
+    return real_eq(value.real, item);
   case STRING_VALUE:
   case LONG_DATA_VALUE:
-    if (binary_cmp)
-      return !stringcmp(&str_value, &item->str_value);
-    return !sortcmp(&str_value, &item->str_value, collation.collation);
+    return str_eq(&str_value, item, binary_cmp);
   default:
     break;
   }
@@ -6115,24 +6089,6 @@ int Item_decimal::save_in_field(Field *field, bool no_conversions)
 }
 
 
-bool Item_int::eq(const Item *arg, bool binary_cmp) const
-{
-  /* No need to check for null value as basic constant can't be NULL */
-  if (arg->basic_const_item() && arg->type() == type())
-  {
-    /*
-      We need to cast off const to call val_int(). This should be OK for
-      a basic constant.
-    */
-    Item *item= (Item*) arg;
-    return (item->val_int() == value &&
-            ((longlong) value >= 0 ||
-             (item->unsigned_flag == unsigned_flag)));
-  }
-  return FALSE;
-}
-
-
 Item *Item_int_with_ref::clone_item()
 {
   DBUG_ASSERT(ref->const_item());
@@ -6250,27 +6206,6 @@ void Item_float::print(String *str, enum_query_type query_type)
 }
 
 
-/*
-  hex item
-  In string context this is a binary string.
-  In number context this is a longlong value.
-*/
-
-bool Item_float::eq(const Item *arg, bool binary_cmp) const
-{
-  if (arg->basic_const_item() && arg->type() == type())
-  {
-    /*
-      We need to cast off const to call val_int(). This should be OK for
-      a basic constant.
-    */
-    Item *item= (Item*) arg;
-    return item->val_real() == value;
-  }
-  return FALSE;
-}
-
-
 inline uint char_val(char X)
 {
   return (uint) (X >= '0' && X <= '9' ? X-'0' :
@@ -6365,19 +6300,6 @@ void Item_hex_string::print(String *str, enum_query_type query_type)
 }
 
 
-bool Item_hex_constant::eq(const Item *arg, bool binary_cmp) const
-{
-  if (arg->basic_const_item() && arg->type() == type() &&
-      arg->cast_to_int_type() == cast_to_int_type())
-  {
-    if (binary_cmp)
-      return !stringcmp(&str_value, &arg->str_value);
-    return !sortcmp(&str_value, &arg->str_value, collation.collation);
-  }
-  return FALSE;
-}
-
-
 /*
   bin item.
   In string context this is a binary string.

sql/item.h

@@ -1663,11 +1663,66 @@ class Field_enumerator
 
 class sp_head;
 
-class Item_basic_constant :public Item
+
+/**
+  A common class for Item_basic_constant and Item_param
+*/
+class Item_basic_value :public Item
+{
+  bool is_basic_value(const Item *item, Type type_arg) const
+  {
+    return item->basic_const_item() && item->type() == type_arg;
+  }
+  bool is_basic_value(Type type_arg) const
+  {
+    return basic_const_item() && type() == type_arg;
+  }
+  bool str_eq(const String *value,
+              const String *other, CHARSET_INFO *cs, bool binary_cmp) const
+  {
+    return binary_cmp ?
+      value->bin_eq(other) :
+      collation.collation == cs && value->eq(other, collation.collation);
+  }
+protected:
+  Item_basic_value(): Item() {}
+  /*
+    In the xxx_eq() methods below we need to cast off "const" to
+    call val_xxx(). This is OK for Item_basic_constant and Item_param.
+  */
+  bool null_eq(const Item *item) const
+  {
+    DBUG_ASSERT(is_basic_value(NULL_ITEM));
+    return item->type() == NULL_ITEM;
+  }
+  bool str_eq(const String *value, const Item *item, bool binary_cmp) const
+  {
+    DBUG_ASSERT(is_basic_value(STRING_ITEM));
+    return is_basic_value(item, STRING_ITEM) &&
+           str_eq(value, ((Item_basic_value*)item)->val_str(NULL),
+                  item->collation.collation, binary_cmp);
+  }
+  bool real_eq(double value, const Item *item) const
+  {
+    DBUG_ASSERT(is_basic_value(REAL_ITEM));
+    return is_basic_value(item, REAL_ITEM) &&
+           value == ((Item_basic_value*)item)->val_real();
+  }
+  bool int_eq(longlong value, const Item *item) const
+  {
+    DBUG_ASSERT(is_basic_value(INT_ITEM));
+    return is_basic_value(item, INT_ITEM) &&
+           value == ((Item_basic_value*)item)->val_int() &&
+           (value >= 0 || item->unsigned_flag == unsigned_flag);
+  }
+};
+
+
+class Item_basic_constant :public Item_basic_value
 {
   table_map used_table_map;
 public:
-  Item_basic_constant(): Item(), used_table_map(0) {};
+  Item_basic_constant(): Item_basic_value(), used_table_map(0) {};
   void set_used_tables(table_map map) { used_table_map= map; }
   table_map used_tables() const { return used_table_map; }
   /* to prevent drop fixed flag (no need parent cleanup call) */
@@ -2263,7 +2318,7 @@ class Item_null :public Item_basic_constant
     collation.set(cs, DERIVATION_IGNORABLE);
   }
   enum Type type() const { return NULL_ITEM; }
-  bool eq(const Item *item, bool binary_cmp) const;
+  bool eq(const Item *item, bool binary_cmp) const { return null_eq(item); }
   double val_real();
   longlong val_int();
   String *val_str(String *str);
@@ -2306,7 +2361,7 @@ class Item_null_result :public Item_null
 
 /* Item represents one placeholder ('?') of prepared statement */
 
-class Item_param :public Item,
+class Item_param :public Item_basic_value,
                   private Settable_routine_parameter
 {
   char cnvbuf[MAX_FIELD_WIDTH];
@@ -2496,7 +2551,8 @@ class Item_int :public Item_num
   Item_num *neg() { value= -value; return this; }
   uint decimal_precision() const
   { return (uint) (max_length - MY_TEST(value < 0)); }
-  bool eq(const Item *, bool binary_cmp) const;
+  bool eq(const Item *item, bool binary_cmp) const
+  { return int_eq(value, item); }
   bool check_partition_func_processor(uchar *bool_arg) { return FALSE;}
   bool check_vcol_func_processor(uchar *arg) { return FALSE;}
 };
@@ -2617,7 +2673,8 @@ class Item_float :public Item_num
   { return new Item_float(name, value, decimals, max_length); }
   Item_num *neg() { value= -value; return this; }
   virtual void print(String *str, enum_query_type query_type);
-  bool eq(const Item *, bool binary_cmp) const;
+  bool eq(const Item *item, bool binary_cmp) const
+  { return real_eq(value, item); }
 };
 
 
@@ -2729,7 +2786,10 @@ class Item_string :public Item_basic_constant
   enum Item_result result_type () const { return STRING_RESULT; }
   enum_field_types field_type() const { return MYSQL_TYPE_VARCHAR; }
   bool basic_const_item() const { return 1; }
-  bool eq(const Item *item, bool binary_cmp) const;
+  bool eq(const Item *item, bool binary_cmp) const
+  {
+    return str_eq(&str_value, item, binary_cmp);
+  }
   Item *clone_item() 
   {
     return new Item_string(name, str_value.ptr(), 
@@ -2931,7 +2991,12 @@ class Item_hex_constant: public Item_basic_constant
   bool check_partition_func_processor(uchar *int_arg) {return FALSE;}
   bool check_vcol_func_processor(uchar *arg) { return FALSE;}
   bool basic_const_item() const { return 1; }
-  bool eq(const Item *item, bool binary_cmp) const;
+  bool eq(const Item *item, bool binary_cmp) const
+  {
+    return item->basic_const_item() && item->type() == type() &&
+           item->cast_to_int_type() == cast_to_int_type() &&
+           str_value.bin_eq(&item->str_value);
+  }
   String *val_str(String*) { DBUG_ASSERT(fixed == 1); return &str_value; }
 };
 

sql/sql_string.h

@@ -540,6 +540,15 @@ class String
     }
     return TRUE;
   }
+  bool bin_eq(const String *other) const
+  {
+    return length() == other->length() &&
+           !memcmp(ptr(), other->ptr(), length());
+  }
+  bool eq(const String *other, CHARSET_INFO *cs) const
+  {
+    return !sortcmp(this, other, cs);
+  }
 };