WL #1056: Eliminate NOT operators from where condition

6 files changed, 188 insertions, 4 deletions

diff --git a/sql/item.h b/sql/item.h
index a126a61e32e..e630d0229d7 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -212,6 +212,9 @@ public:
   virtual void bring_value() {}
 
   Field *tmp_table_field_from_field_type(TABLE *table);
+  
+  /* Used in sql_select.cc:eliminate_not_funcs() */
+  virtual Item *neg_transformer() { return NULL; }
 };
 
 
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index 4e6301d2626..7e2d0961c14 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -23,6 +23,7 @@
 
 #include "mysql_priv.h"
 #include <m_ctype.h>
+#include "sql_select.h"
 
 static Item_result item_store_type(Item_result a,Item_result b)
 {
@@ -476,6 +477,7 @@ longlong Item_func_eq::val_int()
   return value == 0 ? 1 : 0;
 }
 
+
 /* Same as Item_func_eq, but NULL = NULL */
 
 void Item_func_equal::fix_length_and_dec()
@@ -1721,6 +1723,19 @@ void Item_cond::print(String *str)
   str->append(')');
 }
 
+
+void Item_cond::neg_arguments()
+{
+  List_iterator<Item> li(list);
+  Item *item;
+  while ((item= li++))		/* Apply not transformation to the arguments */
+  {
+    Item *new_item= item->neg_transformer();
+    VOID(li.replace(new_item ? new_item : new Item_func_not(item)));
+  }
+}
+
+
 /*
   Evalution of AND(expr, expr, expr ...)
 
@@ -2334,3 +2349,91 @@ longlong Item_cond_xor::val_int()
   }
   return (longlong) result;
 }
+
+/*
+  Apply NOT transformation to the item and return a new one.
+
+  SYNPOSIS
+    neg_transformer()
+
+  DESCRIPTION
+    Transform the item using next rules:
+       a AND b AND ...    -> NOT(a) OR NOT(b) OR ...
+       a OR b OR ...      -> NOT(a) AND NOT(b) AND ...
+       NOT(a)             -> a
+       a = b              -> a != b
+       a != b             -> a = b
+       a < b              -> a >= b
+       a >= b             -> a < b
+       a > b              -> a <= b
+       a <= b             -> a > b
+       IS NULL(a)         -> IS NOT NULL(a)
+       IS NOT NULL(a)     -> IS NULL(a)
+
+  NOTE
+    This method is used in the eliminate_not_funcs() function.
+
+  RETURN
+    New item or
+    NULL if we cannot apply NOT transformation (see Item::neg_transformer()).
+*/
+
+Item *Item_func_not::neg_transformer()		/* NOT(x)  ->  x */
+{
+  /* We should apply negation elimination to the argument of the NOT function */
+  return eliminate_not_funcs(args[0]);
+}
+
+Item *Item_func_eq::neg_transformer()		/* a = b  ->  a != b */
+{
+  return new Item_func_ne(args[0], args[1]);	
+}
+
+Item *Item_func_ne::neg_transformer()		/* a != b  ->  a = b */
+{
+  return new Item_func_eq(args[0], args[1]);
+}
+
+Item *Item_func_lt::neg_transformer()		/* a < b  ->  a >= b */
+{
+  return new Item_func_ge(args[0], args[1]);
+}
+
+Item *Item_func_ge::neg_transformer()		/* a >= b  ->  a < b */
+{
+  return new Item_func_lt(args[0], args[1]);
+}
+
+Item *Item_func_gt::neg_transformer()		/* a > b  ->  a <= b */
+{
+  return new Item_func_le(args[0], args[1]);
+}
+
+Item *Item_func_le::neg_transformer()		/* a <= b  ->  a > b */
+{
+  return new Item_func_gt(args[0], args[1]);
+}
+
+Item *Item_func_isnull::neg_transformer()	/* a IS NULL  ->  a IS NOT NULL */
+{
+  return new Item_func_isnotnull(args[0]);
+}
+
+Item *Item_func_isnotnull::neg_transformer()	/* a IS NOT NULL  ->  a IS NULL */
+{
+  return new Item_func_isnull(args[0]);
+}
+
+Item *Item_cond_and::neg_transformer()		/* NOT(a AND b AND ...)  -> */
+					/* NOT a OR NOT b OR ... */
+{
+  neg_arguments();
+  return new Item_cond_or(list);
+}
+
+Item *Item_cond_or::neg_transformer()		/* NOT(a OR b OR ...)  -> */
+					/* NOT a AND NOT b AND ... */
+{
+  neg_arguments();
+  return new Item_cond_and(list);
+}
diff --git a/sql/item_cmpfunc.h b/sql/item_cmpfunc.h
index 42b73c48606..856d32cadd6 100644
--- a/sql/item_cmpfunc.h
+++ b/sql/item_cmpfunc.h
@@ -153,7 +153,9 @@ class Item_func_not :public Item_bool_func
 public:
   Item_func_not(Item *a) :Item_bool_func(a) {}
   longlong val_int();
+  enum Functype functype() const { return NOT_FUNC; }
   const char *func_name() const { return "not"; }
+  Item *neg_transformer();
 };
 
 class Item_func_not_all :public Item_func_not
@@ -164,6 +166,8 @@ public:
   virtual void top_level_item() { abort_on_null= 1; }
   bool top_level() { return abort_on_null; }
   longlong val_int();
+  enum Functype functype() const { return NOT_ALL_FUNC; }
+  const char *func_name() const { return "not_all"; }
 };
 
 class Item_func_eq :public Item_bool_rowready_func2
@@ -175,6 +179,7 @@ public:
   enum Functype rev_functype() const { return EQ_FUNC; }
   cond_result eq_cmp_result() const { return COND_TRUE; }
   const char *func_name() const { return "="; }
+  Item *neg_transformer();
 };
 
 class Item_func_equal :public Item_bool_rowready_func2
@@ -199,6 +204,7 @@ public:
   enum Functype rev_functype() const { return LE_FUNC; }
   cond_result eq_cmp_result() const { return COND_TRUE; }
   const char *func_name() const { return ">="; }
+  Item *neg_transformer();
 };
 
 
@@ -211,6 +217,7 @@ public:
   enum Functype rev_functype() const { return LT_FUNC; }
   cond_result eq_cmp_result() const { return COND_FALSE; }
   const char *func_name() const { return ">"; }
+  Item *neg_transformer();
 };
 
 
@@ -223,6 +230,7 @@ public:
   enum Functype rev_functype() const { return GE_FUNC; }
   cond_result eq_cmp_result() const { return COND_TRUE; }
   const char *func_name() const { return "<="; }
+  Item *neg_transformer();
 };
 
 
@@ -235,6 +243,7 @@ public:
   enum Functype rev_functype() const { return GT_FUNC; }
   cond_result eq_cmp_result() const { return COND_FALSE; }
   const char *func_name() const { return "<"; }
+  Item *neg_transformer();
 };
 
 
@@ -247,6 +256,7 @@ public:
   cond_result eq_cmp_result() const { return COND_FALSE; }
   optimize_type select_optimize() const { return OPTIMIZE_NONE; }
   const char *func_name() const { return "<>"; }
+  Item *neg_transformer();
 };
 
 
@@ -690,6 +700,7 @@ public:
   }
   table_map not_null_tables() const { return 0; }
   optimize_type select_optimize() const { return OPTIMIZE_NULL; }
+  Item *neg_transformer();
 };
 
 /* Functions used by HAVING for rewriting IN subquery */
@@ -722,6 +733,7 @@ public:
   const char *func_name() const { return "isnotnull"; }
   optimize_type select_optimize() const { return OPTIMIZE_NULL; }
   table_map not_null_tables() const { return 0; }
+  Item *neg_transformer();
 };
 
 
@@ -801,7 +813,8 @@ protected:
 
 public:
   /* Item_cond() is only used to create top level items */
-  Item_cond() : Item_bool_func(), abort_on_null(1) { const_item_cache=0; }
+  Item_cond(): Item_bool_func(), abort_on_null(1)
+  { const_item_cache=0; }
   Item_cond(Item *i1,Item *i2) 
     :Item_bool_func(), abort_on_null(0)
   {
@@ -809,6 +822,8 @@ public:
     list.push_back(i2);
   }
   Item_cond(THD *thd, Item_cond &item);
+  Item_cond(List<Item> &nlist)
+    :Item_bool_func(), list(nlist), abort_on_null(0) {}
   ~Item_cond() { list.delete_elements(); }
   bool add(Item *item) { return list.push_back(item); }
   bool fix_fields(THD *, struct st_table_list *, Item **ref);
@@ -822,8 +837,8 @@ public:
   friend int setup_conds(THD *thd,TABLE_LIST *tables,COND **conds);
   void top_level_item() { abort_on_null=1; }
   void copy_andor_arguments(THD *thd, Item_cond *item);
-
   bool walk(Item_processor processor, byte *arg);
+  void neg_arguments();
 };
 
 
@@ -833,6 +848,7 @@ public:
   Item_cond_and() :Item_cond() {}
   Item_cond_and(Item *i1,Item *i2) :Item_cond(i1,i2) {}
   Item_cond_and(THD *thd, Item_cond_and &item) :Item_cond(thd, item) {}
+  Item_cond_and(List<Item> &list): Item_cond(list) {}
   enum Functype functype() const { return COND_AND_FUNC; }
   longlong val_int();
   const char *func_name() const { return "and"; }
@@ -843,6 +859,7 @@ public:
        item->copy_andor_arguments(thd, this);
     return item;
   }
+  Item *neg_transformer();
 };
 
 class Item_cond_or :public Item_cond
@@ -851,6 +868,7 @@ public:
   Item_cond_or() :Item_cond() {}
   Item_cond_or(Item *i1,Item *i2) :Item_cond(i1,i2) {}
   Item_cond_or(THD *thd, Item_cond_or &item) :Item_cond(thd, item) {}
+  Item_cond_or(List<Item> &list): Item_cond(list) {}
   enum Functype functype() const { return COND_OR_FUNC; }
   longlong val_int();
   const char *func_name() const { return "or"; }
@@ -862,6 +880,7 @@ public:
       item->copy_andor_arguments(thd, this);
     return item;
   }
+  Item *neg_transformer();
 };
 
 
diff --git a/sql/item_func.h b/sql/item_func.h
index 8086e65786d..7c3d499f52e 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -46,7 +46,8 @@ public:
 		  SP_TOUCHES_FUNC,SP_CROSSES_FUNC,SP_WITHIN_FUNC,
 		  SP_CONTAINS_FUNC,SP_OVERLAPS_FUNC,
 		  SP_STARTPOINT,SP_ENDPOINT,SP_EXTERIORRING,
-		  SP_POINTN,SP_GEOMETRYN,SP_INTERIORRINGN};
+		  SP_POINTN,SP_GEOMETRYN,SP_INTERIORRINGN,
+		  NOT_FUNC, NOT_ALL_FUNC};
   enum optimize_type { OPTIMIZE_NONE,OPTIMIZE_KEY,OPTIMIZE_OP, OPTIMIZE_NULL };
   enum Type type() const { return FUNC_ITEM; }
   virtual enum Functype functype() const   { return UNKNOWN_FUNC; }
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 4bab97d4a46..0fc78421824 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -4161,6 +4161,60 @@ propagate_cond_constants(I_List<COND_CMP> *save_list,COND *and_father,
 }
 
 
+/*
+  Eliminate NOT functions from the condition tree.
+
+  SYNPOSIS
+    eliminate_not_funcs()
+    cond	condition tree
+
+  DESCRIPTION
+    Eliminate NOT functions from the condition tree where it's possible.
+    Recursively traverse condition tree to find all NOT functions.
+    Call neg_transformer() method for negated arguments.
+
+  NOTE
+    If neg_transformer() returned a new condition we call fix_fields().
+    We don't delete any items as it's not needed. They will be deleted 
+    later at once.
+
+  RETURN
+    New condition tree
+*/
+
+COND *eliminate_not_funcs(COND *cond)
+{
+  if (!cond)
+    return cond;
+  if (cond->type() == Item::COND_ITEM)		/* OR or AND */
+  {
+    List_iterator<Item> li(*((Item_cond*) cond)->argument_list());
+    Item *item;
+    while ((item= li++))
+    {
+      Item *new_item= eliminate_not_funcs(item);
+      if (item != new_item)
+	VOID(li.replace(new_item));	/* replace item with a new condition */
+    }
+  }
+  else if (cond->type() == Item::FUNC_ITEM &&	/* 'NOT' operation? */
+	   ((Item_func*) cond)->functype() == Item_func::NOT_FUNC)
+  {
+    COND *new_cond= ((Item_func*) cond)->arguments()[0]->neg_transformer();
+    if (new_cond)
+    {
+      /*
+        Here we can delete the NOT function. Something like: delete cond;
+        But we don't need to do it. All items will be deleted later at once.
+      */
+      new_cond->fix_fields(current_thd, 0, &new_cond);
+      cond= new_cond;
+    }
+  }
+  return cond;
+}
+
+
 static COND *
 optimize_cond(COND *conds,Item::cond_result *cond_value)
 {
@@ -4169,8 +4223,11 @@ optimize_cond(COND *conds,Item::cond_result *cond_value)
     *cond_value= Item::COND_TRUE;
     return conds;
   }
-  /* change field = field to field = const for each found field = const */
   DBUG_EXECUTE("where",print_where(conds,"original"););
+  /* eliminate NOT operators */
+  conds= eliminate_not_funcs(conds);
+  DBUG_EXECUTE("where", print_where(conds, "after negation elimination"););
+  /* change field = field to field = const for each found field = const */
   propagate_cond_constants((I_List<COND_CMP> *) 0,conds,conds);
   /*
     Remove all instances of item == item
diff --git a/sql/sql_select.h b/sql/sql_select.h
index 6c17a646ee6..642d53a6de3 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -414,3 +414,4 @@ bool error_if_full_join(JOIN *join);
 void relink_tables(SELECT_LEX *select_lex);
 int report_error(TABLE *table, int error);
 int safe_index_read(JOIN_TAB *tab);
+COND *eliminate_not_funcs(COND *cond);