optimized IN with Rows (SCRUM)

NULL with row (IN)
cardinality error of row inside row
new[]
fixed layout


mysql-test/r/row.result:
  test of optimized IN with Rows
  test of NULL with row
  test of cardinality error of row inside row
mysql-test/t/row.test:
  test of optimized IN with Rows
  test of NULL with row
  test of cardinality error of row inside row
sql/item_cmpfunc.cc:
  optimized IN with Rows
  NULL with row (IN)
  cardinality error of row inside row
sql/item_cmpfunc.h:
  optimized IN with Rows
  NULL with row (IN)
  cardinality error of row inside row
  fixed layout
sql/sql_list.h:
  fixed layout
  new[]

1 files changed, 148 insertions, 43 deletions

diff --git a/sql/item_cmpfunc.h b/sql/item_cmpfunc.h
index d15e8856ece..fbf0eb3c123 100644
--- a/sql/item_cmpfunc.h
+++ b/sql/item_cmpfunc.h
@@ -17,6 +17,8 @@
 
 /* compare and test functions */
 
+#include "assert.h"
+
 #ifdef __GNUC__
 #pragma interface			/* gcc class implementation */
 #endif
@@ -375,6 +377,7 @@ class in_vector :public Sql_alloc
   uint count;
 public:
   uint used_count;
+  in_vector() {}
   in_vector(uint elements,uint element_length,qsort_cmp cmp_func)
     :base((char*) sql_calloc(elements*element_length)),
      size(element_length), compare(cmp_func), count(elements),
@@ -389,7 +392,6 @@ public:
   int find(Item *item);
 };
 
-
 class in_string :public in_vector
 {
   char buff[80];
@@ -401,7 +403,6 @@ public:
   byte *get_value(Item *item);
 };
 
-
 class in_longlong :public in_vector
 {
   longlong tmp;
@@ -411,7 +412,6 @@ public:
   byte *get_value(Item *item);
 };
 
-
 class in_double :public in_vector
 {
   double tmp;
@@ -421,7 +421,6 @@ public:
   byte *get_value(Item *item);
 };
 
-
 /*
 ** Classes for easy comparing of non const items
 */
@@ -431,73 +430,114 @@ class cmp_item :public Sql_alloc
 public:
   cmp_item() {}
   virtual ~cmp_item() {}
-  virtual void store_value(Item *item)=0;
-  virtual int cmp(Item *item)=0;
+  virtual void store_value(Item *item)= 0;
+  virtual int cmp(Item *item)= 0;
+  // for optimized IN with row
+  virtual int compare(cmp_item *item)= 0;
   static cmp_item* get_comparator(Item *);
+  virtual cmp_item *make_same()= 0;
+  virtual void store_value_by_template(cmp_item *tmpl, Item *item)
+  {
+    store_value(item);
+  }
 };
 
+class cmp_item_string :public cmp_item 
+{
+protected:
+  String *value_res;
+public:
+  friend class cmp_item_sort_string;
+  friend class cmp_item_binary_string;
+  friend class cmp_item_sort_string_in_static;
+  friend class cmp_item_binary_string_in_static;
+};
 
-class cmp_item_sort_string :public cmp_item {
- protected:
+class cmp_item_sort_string :public cmp_item_string
+{
+protected:
   char value_buff[80];
-  String value,*value_res;
+  String value;
 public:
-  cmp_item_sort_string() :value(value_buff,sizeof(value_buff),default_charset_info) {}
+  cmp_item_sort_string():
+    value(value_buff, sizeof(value_buff), default_charset_info) {}
   void store_value(Item *item)
-    {
-      value_res=item->val_str(&value);
-    }
+  {
+    value_res= item->val_str(&value);
+  }
   int cmp(Item *arg)
-    {
-      char buff[80];
-      String tmp(buff,sizeof(buff),default_charset_info),*res;
-      if (!(res=arg->val_str(&tmp)))
-	return 1;				/* Can't be right */
-      return sortcmp(value_res,res);
-    }
+  {
+    char buff[80];
+    String tmp(buff, sizeof(buff), default_charset_info), *res;
+    if (!(res= arg->val_str(&tmp)))
+      return 1;				/* Can't be right */
+    return sortcmp(value_res, res);
+  }
+  int compare(cmp_item *c)
+  {
+    cmp_item_string *cmp= (cmp_item_string *)c;
+    return sortcmp(value_res, cmp->value_res);
+  } 
+  cmp_item *make_same();
 };
 
 class cmp_item_binary_string :public cmp_item_sort_string {
 public:
   cmp_item_binary_string() {}
   int cmp(Item *arg)
-    {
-      char buff[80];
-      String tmp(buff,sizeof(buff),default_charset_info),*res;
-      if (!(res=arg->val_str(&tmp)))
-	return 1;				/* Can't be right */
-      return stringcmp(value_res,res);
-    }
+  {
+    char buff[80];
+    String tmp(buff,sizeof(buff),default_charset_info),*res;
+    if (!(res=arg->val_str(&tmp)))
+      return 1;				/* Can't be right */
+    return stringcmp(value_res,res);
+  }
+  int compare(cmp_item *c)
+  {
+    cmp_item_string *cmp= (cmp_item_string *)c;
+    return stringcmp(value_res, cmp->value_res);
+  }
+  cmp_item *make_same();
 };
 
-
 class cmp_item_int :public cmp_item
 {
   longlong value;
 public:
   void store_value(Item *item)
-    {
-      value=item->val_int();
-    }
+  {
+    value= item->val_int();
+  }
   int cmp(Item *arg)
-    {
-      return value != arg->val_int();
-    }
+  {
+    return value != arg->val_int();
+  }
+  int compare(cmp_item *c)
+  {
+    cmp_item_int *cmp= (cmp_item_int *)c;
+    return (value < cmp->value) ? -1 : ((value == cmp->value) ? 0 : 1);
+  }
+  cmp_item *make_same();
 };
 
-
 class cmp_item_real :public cmp_item
 {
   double value;
 public:
   void store_value(Item *item)
-    {
-      value= item->val();
-    }
+  {
+    value= item->val();
+  }
   int cmp(Item *arg)
-    {
-      return value != arg->val();
-    }
+  {
+    return value != arg->val();
+  }
+  int compare(cmp_item *c)
+  {
+    cmp_item_real *cmp= (cmp_item_real *)c;
+    return (value < cmp->value)? -1 : ((value == cmp->value) ? 0 : 1);
+  }
+  cmp_item *make_same();
 };
 
 class cmp_item_row :public cmp_item
@@ -505,8 +545,75 @@ class cmp_item_row :public cmp_item
   cmp_item **comparators;
   uint n;
 public:
+  cmp_item_row(): comparators(0), n(0) {}
+  ~cmp_item_row()
+  {
+    if(comparators)
+      for(uint i= 0; i < n; i++)
+	if (comparators[i])
+	  delete comparators[i];
+  }
   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 *);
+  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)
+*/
+
+class cmp_item_sort_string_in_static :public cmp_item_string
+{
+ protected:
+  String value;
+public:
+  cmp_item_sort_string_in_static() {}
+  void store_value(Item *item)
+  {
+    value_res= item->val_str(&value);
+  }
+  int cmp(Item *item)
+  {
+    // Should never be called
+    DBUG_ASSERT(0);
+    return 1;
+  }
+  int compare(cmp_item *c)
+  {
+    cmp_item_string *cmp= (cmp_item_string *)c;
+    return sortcmp(value_res, cmp->value_res);
+  }
+  cmp_item * make_same()
+  {
+    return new cmp_item_sort_string_in_static();
+  }
+};
+
+class cmp_item_binary_string_in_static :public cmp_item_sort_string_in_static {
+public:
+  cmp_item_binary_string_in_static() {}
+  int compare(cmp_item *c)
+  {
+    cmp_item_string *cmp= (cmp_item_string *)c;
+    return stringcmp(value_res, cmp->value_res);
+  }
+  cmp_item * make_same()
+  {
+    return new cmp_item_binary_string_in_static();
+  }
 };
 
 class Item_func_in :public Item_int_func
@@ -546,8 +653,6 @@ class Item_func_in :public Item_int_func
   }
 };
 
-
-
 /* Functions used by where clause */
 
 class Item_func_isnull :public Item_bool_func