A fix and test case for Bug#9096 "select doesn't return all matched

records if prepared statements is used".
This fix changes equality evaluation method of basic constants from
by-name to by-value, thus effectively enabling use of parameter markers
in some optimizations (constants propagation, evaluation of possible
keys for query).

mysql-test/r/ps.result

@@ -494,3 +494,28 @@ SELECT FOUND_ROWS();
 FOUND_ROWS()
 2
 deallocate prepare stmt;
+drop table if exists t1;
+Warnings:
+Note	1051	Unknown table 't1'
+create table t1 (c1 int(11) not null, c2 int(11) not null,
+primary key  (c1,c2), key c2 (c2), key c1 (c1));
+insert into t1 values (200887, 860);
+insert into t1 values (200887, 200887);
+select * from t1 where (c1=200887 and c2=200887) or c2=860;
+c1	c2
+200887	860
+200887	200887
+prepare stmt from
+"select * from t1 where (c1=200887 and c2=200887) or c2=860";
+execute stmt;
+c1	c2
+200887	860
+200887	200887
+prepare stmt from
+"select * from t1 where (c1=200887 and c2=?) or c2=?";
+set @a=200887, @b=860;
+execute stmt using @a, @b;
+c1	c2
+200887	860
+200887	200887
+deallocate prepare stmt;

mysql-test/t/ps.test

@@ -496,3 +496,29 @@ SELECT FOUND_ROWS();
 execute stmt;                                                                   
 SELECT FOUND_ROWS();                                                            
 deallocate prepare stmt;
+
+#
+# Bug#9096 "select doesn't return all matched records if prepared statements
+# is used"
+# The bug was is bad co-operation of the optimizer's algorithm which determines
+# which keys can be used to execute a query, constants propagation
+# part of the optimizer and parameter markers used by prepared statements.
+
+drop table if exists t1;
+create table t1 (c1 int(11) not null, c2 int(11) not null,
+             primary key  (c1,c2), key c2 (c2), key c1 (c1));
+
+insert into t1 values (200887, 860);
+insert into t1 values (200887, 200887);
+
+select * from t1 where (c1=200887 and c2=200887) or c2=860;
+
+prepare stmt from
+"select * from t1 where (c1=200887 and c2=200887) or c2=860";
+execute stmt;
+prepare stmt from
+"select * from t1 where (c1=200887 and c2=?) or c2=?";
+set @a=200887, @b=860;
+# this query did not return all matching rows
+execute stmt using @a, @b;
+deallocate prepare stmt;

sql/item.cc

@@ -200,6 +200,11 @@ void Item::set_name(const char *str, uint length, CHARSET_INFO *cs)
 
 bool Item::eq(const Item *item, bool binary_cmp) const
 {
+  /*
+    Note, that this is never TRUE if item is a Item_param:
+    for all basic constants we have special checks, and Item_param's
+    type() can be only among basic constant types.
+  */
   return type() == item->type() && name && item->name &&
     !my_strcasecmp(system_charset_info,name,item->name);
 }
@@ -254,7 +259,7 @@ Item *Item_string::safe_charset_converter(CHARSET_INFO *tocs)
 
 bool Item_string::eq(const Item *item, bool binary_cmp) const
 {
-  if (type() == item->type())
+  if (type() == item->type() && item->basic_const_item())
   {
     if (binary_cmp)
       return !stringcmp(&str_value, &item->str_value);
@@ -1356,6 +1361,70 @@ bool Item_param::fix_fields(THD *thd, TABLE_LIST *tables, Item **ref)
   return 0;
 }
 
+
+bool Item_param::basic_const_item() const
+{
+  if (state == NO_VALUE || state == TIME_VALUE)
+    return FALSE;
+  return TRUE;
+}
+
+Item *
+Item_param::new_item()
+{
+  /* see comments in the header file */
+  switch (state) {
+  case NULL_VALUE:
+    return new Item_null(name);
+  case INT_VALUE:
+    return new Item_int(name, value.integer, max_length);
+  case REAL_VALUE:
+    return new Item_real(name, value.real, decimals, max_length);
+  case STRING_VALUE:
+  case LONG_DATA_VALUE:
+    return new Item_string(name, str_value.c_ptr_quick(), str_value.length(),
+                           str_value.charset());
+  case TIME_VALUE:
+    break;
+  case NO_VALUE:
+  default:
+    DBUG_ASSERT(0);
+  };
+  return 0;
+}
+
+
+bool
+Item_param::eq(const Item *arg, bool binary_cmp) const
+{
+  Item *item;
+  if (!basic_const_item() || !arg->basic_const_item() || arg->type() != type())
+    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;
+  case INT_VALUE:
+    return value.integer == item->val_int() &&
+           unsigned_flag == item->unsigned_flag;
+  case REAL_VALUE:
+    return value.real == item->val();
+  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);
+  default:
+    break;
+  }
+  return FALSE;
+}
+
 /* End of Item_param related */
 
 
@@ -1937,6 +2006,23 @@ int Item_int::save_in_field(Field *field, bool no_conversions)
   return field->store(nr);
 }
 
+
+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 && item->unsigned_flag == unsigned_flag;
+  }
+  return FALSE;
+}
+
+
 Item_num *Item_uint::neg()
 {
   return new Item_real(name, - ((double) value), 0, max_length);
@@ -1951,6 +2037,21 @@ int Item_real::save_in_field(Field *field, bool no_conversions)
   return field->store(nr);
 }
 
+
+bool Item_real::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() == value;
+  }
+  return FALSE;
+}
+
 /****************************************************************************
 ** varbinary item
 ** In string context this is a binary string
@@ -2017,6 +2118,17 @@ int Item_varbinary::save_in_field(Field *field, bool no_conversions)
 }
 
 
+bool Item_varbinary::eq(const Item *arg, bool binary_cmp) const
+{
+  if (arg->basic_const_item() && arg->type() == type())
+  {
+    if (binary_cmp)
+      return !stringcmp(&str_value, &arg->str_value);
+    return !sortcmp(&str_value, &arg->str_value, collation.collation);
+  }
+  return FALSE;
+}
+
 /*
   Pack data in buffer for sending
 */

sql/item.h

@@ -234,7 +234,7 @@ class Item {
   virtual table_map not_null_tables() const { return used_tables(); }
   /*
     Returns true if this is a simple constant item like an integer, not
-    a constant expression
+    a constant expression. Used in the optimizer to propagate basic constants.
   */
   virtual bool basic_const_item() const { return 0; }
   /* cloning of constant items (0 if it is not const) */
@@ -581,7 +581,6 @@ class Item_param :public Item
 
   bool convert_str_value(THD *thd);
 
-  Item *new_item() { return new Item_param(pos_in_query); }
   /*
     If value for parameter was not set we treat it as non-const
     so noone will use parameters value in fix_fields still
@@ -590,12 +589,29 @@ class Item_param :public Item
   virtual table_map used_tables() const
   { return state != NO_VALUE ? (table_map)0 : PARAM_TABLE_BIT; }
   void print(String *str) { str->append('?'); }
-  /* parameter never equal to other parameter of other item */
-  bool eq(const Item *item, bool binary_cmp) const { return 0; }
   bool is_null()
   { DBUG_ASSERT(state != NO_VALUE); return state == NULL_VALUE; }
+  bool basic_const_item() const;
+  /*
+    This method is used to make a copy of a basic constant item when
+    propagating constants in the optimizer. The reason to create a new
+    item and not use the existing one is not precisely known (2005/04/16).
+    Probably we are trying to preserve tree structure of items, in other
+    words, avoid pointing at one item from two different nodes of the tree.
+    Return a new basic constant item if parameter value is a basic
+    constant, assert otherwise. This method is called only if
+    basic_const_item returned TRUE.
+  */
+  Item *new_item();
+  /*
+    Implement by-value equality evaluation if parameter value
+    is set and is a basic constant (integer, real or string).
+    Otherwise return FALSE.
+  */
+  bool eq(const Item *item, bool binary_cmp) const;
 };
 
+
 class Item_int :public Item_num
 {
 public:
@@ -622,6 +638,7 @@ class Item_int :public Item_num
   void cleanup() {}
   void print(String *str);
   Item_num *neg() { value= -value; return this; }
+  bool eq(const Item *, bool binary_cmp) const;
 };
 
 
@@ -677,6 +694,7 @@ class Item_real :public Item_num
   void cleanup() {}
   Item *new_item() { return new Item_real(name,value,decimals,max_length); }
   Item_num *neg() { value= -value; return this; }
+  bool eq(const Item *, bool binary_cmp) const;
 };
 
 
@@ -810,6 +828,7 @@ class Item_varbinary :public Item
   enum_field_types field_type() const { return MYSQL_TYPE_STRING; }
   // to prevent drop fixed flag (no need parent cleanup call)
   void cleanup() {}
+  bool eq(const Item *item, bool binary_cmp) const;
 };
 
 

sql/item_func.cc

@@ -790,10 +790,12 @@ void Item_func_neg::fix_length_and_dec()
     maximum number of bytes real or integer may require. Note that all 
     constants are non negative so we don't need to account for removed '-'.
     B) argument returns a string.
+    Use val() to get value as arg_type doesn't mean that item is
+    Item_int or Item_real due to existence of Item_param.
   */
   if (arg_result == STRING_RESULT || 
-      (arg_type == REAL_ITEM && ((Item_real*)args[0])->value >= 0) ||
-      (arg_type == INT_ITEM && ((Item_int*)args[0])->value > 0))
+      (arg_type == REAL_ITEM && args[0]->val() >= 0) ||
+      (arg_type == INT_ITEM && args[0]->val_int() > 0))
     max_length++;
 
   if (args[0]->result_type() == INT_RESULT)
@@ -809,8 +811,7 @@ void Item_func_neg::fix_length_and_dec()
       signed integers)
     */
     if (args[0]->type() != INT_ITEM ||
-	((ulonglong) ((Item_uint*) args[0])->value <=
-	 (ulonglong) LONGLONG_MIN))
+	(((ulonglong) args[0]->val_int()) <= (ulonglong) LONGLONG_MIN))
       hybrid_type= INT_RESULT;
   }
 }