Fixed handling of NOT LIKE after Item_func::NOTLIKE_FUNC has been removed

sql/ha_ndbcluster.cc

@@ -7010,7 +7010,7 @@ ha_ndbcluster::build_scan_filter_predicate(Ndb_cond * &cond,
     switch ((negated) ? 
             Ndb_item::negate(cond->ndb_item->qualification.function_type)
             : cond->ndb_item->qualification.function_type) {
-    case Item_func::EQ_FUNC:
+    case NDB_EQ_FUNC:
     {
       if (!value || !field) break;
       // Save value in right format for the field type
@@ -7024,7 +7024,7 @@ ha_ndbcluster::build_scan_filter_predicate(Ndb_cond * &cond,
       cond= cond->next->next->next;
       DBUG_RETURN(0);
     }
-    case Item_func::NE_FUNC:
+    case NDB_NE_FUNC:
     {
       if (!value || !field) break;
       // Save value in right format for the field type
@@ -7038,7 +7038,7 @@ ha_ndbcluster::build_scan_filter_predicate(Ndb_cond * &cond,
       cond= cond->next->next->next;
       DBUG_RETURN(0);
     }
-    case Item_func::LT_FUNC:
+    case NDB_LT_FUNC:
     {
       if (!value || !field) break;
       // Save value in right format for the field type
@@ -7064,7 +7064,7 @@ ha_ndbcluster::build_scan_filter_predicate(Ndb_cond * &cond,
       cond= cond->next->next->next;
       DBUG_RETURN(0);
     }
-    case Item_func::LE_FUNC:
+    case NDB_LE_FUNC:
     {
       if (!value || !field) break;
       // Save value in right format for the field type
@@ -7090,7 +7090,7 @@ ha_ndbcluster::build_scan_filter_predicate(Ndb_cond * &cond,
       cond= cond->next->next->next;
       DBUG_RETURN(0);
     }
-    case Item_func::GE_FUNC:
+    case NDB_GE_FUNC:
     {
       if (!value || !field) break;
       // Save value in right format for the field type
@@ -7116,7 +7116,7 @@ ha_ndbcluster::build_scan_filter_predicate(Ndb_cond * &cond,
       cond= cond->next->next->next;
       DBUG_RETURN(0);
     }
-    case Item_func::GT_FUNC:
+    case NDB_GT_FUNC:
     {
       if (!value || !field) break;
       // Save value in right format for the field type
@@ -7142,7 +7142,7 @@ ha_ndbcluster::build_scan_filter_predicate(Ndb_cond * &cond,
       cond= cond->next->next->next;
       DBUG_RETURN(0);
     }
-    case Item_func::LIKE_FUNC:
+    case NDB_LIKE_FUNC:
     {
       if (!value || !field) break;
       if ((value->qualification.value_type != Item::STRING_ITEM) &&
@@ -7161,7 +7161,26 @@ ha_ndbcluster::build_scan_filter_predicate(Ndb_cond * &cond,
       cond= cond->next->next->next;
       DBUG_RETURN(0);
     }
-    case Item_func::ISNULL_FUNC:
+    case NDB_NOTLIKE_FUNC:
+    {
+      if (!value || !field) break;
+      if ((value->qualification.value_type != Item::STRING_ITEM) &&
+          (value->qualification.value_type != Item::VARBIN_ITEM))
+          break;
+      // Save value in right format for the field type
+      value->save_in_field(field);
+      DBUG_PRINT("info", ("Generating NOTLIKE filter: notlike(%d,%s,%d)", 
+                          field->get_field_no(), value->get_val(), 
+                          value->pack_length()));
+      if (filter->cmp(NdbScanFilter::COND_NOT_LIKE, 
+                      field->get_field_no(),
+                      value->get_val(),
+                      value->pack_length()) == -1)
+        DBUG_RETURN(1);
+      cond= cond->next->next->next;
+      DBUG_RETURN(0);
+    }
+    case NDB_ISNULL_FUNC:
       if (!field)
         break;
       DBUG_PRINT("info", ("Generating ISNULL filter"));
@@ -7169,7 +7188,7 @@ ha_ndbcluster::build_scan_filter_predicate(Ndb_cond * &cond,
         DBUG_RETURN(1);
       cond= cond->next->next;
       DBUG_RETURN(0);
-    case Item_func::ISNOTNULL_FUNC:
+    case NDB_ISNOTNULL_FUNC:
     {
       if (!field)
         break;
@@ -7206,7 +7225,7 @@ ha_ndbcluster::build_scan_filter_group(Ndb_cond* &cond, NdbScanFilter *filter)
     case NDB_FUNCTION:
     {
       switch (cond->ndb_item->qualification.function_type) {
-      case Item_func::COND_AND_FUNC:
+      case NDB_COND_AND_FUNC:
       {
         level++;
         DBUG_PRINT("info", ("Generating %s group %u", (negated)?"NAND":"AND",
@@ -7218,7 +7237,7 @@ ha_ndbcluster::build_scan_filter_group(Ndb_cond* &cond, NdbScanFilter *filter)
         cond= cond->next;
         break;
       }
-      case Item_func::COND_OR_FUNC:
+      case NDB_COND_OR_FUNC:
       {
         level++;
         DBUG_PRINT("info", ("Generating %s group %u", (negated)?"NOR":"OR",
@@ -7230,7 +7249,7 @@ ha_ndbcluster::build_scan_filter_group(Ndb_cond* &cond, NdbScanFilter *filter)
         cond= cond->next;
         break;
       }
-      case Item_func::NOT_FUNC:
+      case NDB_NOT_FUNC:
       {
         DBUG_PRINT("info", ("Generating negated query"));
         cond= cond->next;
@@ -7273,8 +7292,8 @@ ha_ndbcluster::build_scan_filter(Ndb_cond * &cond, NdbScanFilter *filter)
     switch (cond->ndb_item->type) {
     case NDB_FUNCTION:
       switch (cond->ndb_item->qualification.function_type) {
-      case Item_func::COND_AND_FUNC:
-      case Item_func::COND_OR_FUNC:
+      case NDB_COND_AND_FUNC:
+      case NDB_COND_OR_FUNC:
         simple_cond= FALSE;
         break;
       default:

sql/ha_ndbcluster.h

@@ -72,10 +72,28 @@ typedef enum ndb_item_type {
   NDB_END_COND = 3 // End marker for condition group
 } NDB_ITEM_TYPE;
 
+typedef enum ndb_func_type {
+  NDB_EQ_FUNC = 0,
+  NDB_NE_FUNC = 1,
+  NDB_LT_FUNC = 2,
+  NDB_LE_FUNC = 3,
+  NDB_GT_FUNC = 4,
+  NDB_GE_FUNC = 5,
+  NDB_ISNULL_FUNC = 6,
+  NDB_ISNOTNULL_FUNC = 7,
+  NDB_LIKE_FUNC = 8,
+  NDB_NOTLIKE_FUNC = 9,
+  NDB_NOT_FUNC = 10,
+  NDB_UNKNOWN_FUNC = 11,
+  NDB_COND_AND_FUNC = 12,
+  NDB_COND_OR_FUNC = 13,
+  NDB_UNSUPPORTED_FUNC = 14
+} NDB_FUNC_TYPE;
+
 typedef union ndb_item_qualification {
   Item::Type value_type; 
-  enum_field_types field_type;       // Instead of Item::FIELD_ITEM
-  Item_func::Functype function_type; // Instead of Item::FUNC_ITEM
+  enum_field_types field_type; // Instead of Item::FIELD_ITEM
+  NDB_FUNC_TYPE function_type; // Instead of Item::FUNC_ITEM
 } NDB_ITEM_QUALIFICATION;
 
 typedef struct ndb_item_field_value {
@@ -91,21 +109,31 @@ typedef union ndb_item_value {
 
 struct negated_function_mapping
 {
-  Item_func::Functype pos_fun;
-  Item_func::Functype neg_fun;
+  NDB_FUNC_TYPE pos_fun;
+  NDB_FUNC_TYPE neg_fun;
 };
 
+/*
+  Define what functions can be negated in condition pushdown.
+  Note, these HAVE to be in the same order as in definition enum
+*/
 static const negated_function_mapping neg_map[]= 
 {
-  {Item_func::EQ_FUNC, Item_func::NE_FUNC},
-  {Item_func::NE_FUNC, Item_func::EQ_FUNC},
-  {Item_func::LT_FUNC, Item_func::GE_FUNC},
-  {Item_func::LE_FUNC, Item_func::GT_FUNC},
-  {Item_func::GT_FUNC, Item_func::LE_FUNC},
-  {Item_func::GE_FUNC, Item_func::LT_FUNC},
-  {Item_func::ISNULL_FUNC, Item_func::ISNOTNULL_FUNC},
-  {Item_func::ISNOTNULL_FUNC, Item_func::ISNULL_FUNC},
-  {Item_func::UNKNOWN_FUNC, Item_func::NOT_FUNC}
+  {NDB_EQ_FUNC, NDB_NE_FUNC},
+  {NDB_NE_FUNC, NDB_EQ_FUNC},
+  {NDB_LT_FUNC, NDB_GE_FUNC},
+  {NDB_LE_FUNC, NDB_GT_FUNC},
+  {NDB_GT_FUNC, NDB_LE_FUNC},
+  {NDB_GE_FUNC, NDB_LT_FUNC},
+  {NDB_ISNULL_FUNC, NDB_ISNOTNULL_FUNC},
+  {NDB_ISNOTNULL_FUNC, NDB_ISNULL_FUNC},
+  {NDB_LIKE_FUNC, NDB_NOTLIKE_FUNC},
+  {NDB_NOTLIKE_FUNC, NDB_LIKE_FUNC},
+  {NDB_NOT_FUNC, NDB_UNSUPPORTED_FUNC},
+  {NDB_UNKNOWN_FUNC, NDB_UNSUPPORTED_FUNC},
+  {NDB_COND_AND_FUNC, NDB_UNSUPPORTED_FUNC},
+  {NDB_COND_OR_FUNC, NDB_UNSUPPORTED_FUNC},
+  {NDB_UNSUPPORTED_FUNC, NDB_UNSUPPORTED_FUNC}
 };
   
 /*
@@ -160,14 +188,14 @@ class Ndb_item {
   Ndb_item(Item_func::Functype func_type, const Item *item_value) 
     : type(NDB_FUNCTION)
   {
-    qualification.function_type= func_type;
+    qualification.function_type= item_func_to_ndb_func(func_type);
     value.item= item_value;
     value.arg_count= ((Item_func *) item_value)->argument_count();
   };
   Ndb_item(Item_func::Functype func_type, uint no_args) 
     : type(NDB_FUNCTION)
   {
-    qualification.function_type= func_type;
+    qualification.function_type= item_func_to_ndb_func(func_type);
     value.arg_count= no_args;
   };
   ~Ndb_item()
@@ -229,13 +257,30 @@ class Ndb_item {
       ((Item *)item)->save_in_field(field, false);
   };
 
-  static Item_func::Functype negate(Item_func::Functype fun)
+  static NDB_FUNC_TYPE item_func_to_ndb_func(Item_func::Functype fun)
+  {
+    switch (fun) {
+    case (Item_func::EQ_FUNC): { return NDB_EQ_FUNC; }
+    case (Item_func::NE_FUNC): { return NDB_NE_FUNC; }
+    case (Item_func::LT_FUNC): { return NDB_LT_FUNC; }
+    case (Item_func::LE_FUNC): { return NDB_LE_FUNC; }
+    case (Item_func::GT_FUNC): { return NDB_GT_FUNC; }
+    case (Item_func::GE_FUNC): { return NDB_GE_FUNC; }
+    case (Item_func::ISNULL_FUNC): { return NDB_ISNULL_FUNC; }
+    case (Item_func::ISNOTNULL_FUNC): { return NDB_ISNOTNULL_FUNC; }
+    case (Item_func::LIKE_FUNC): { return NDB_LIKE_FUNC; }
+    case (Item_func::NOT_FUNC): { return NDB_NOT_FUNC; }
+    case (Item_func::UNKNOWN_FUNC): { return NDB_UNKNOWN_FUNC; }
+    case (Item_func::COND_AND_FUNC): { return NDB_COND_AND_FUNC; }
+    case (Item_func::COND_OR_FUNC): { return NDB_COND_OR_FUNC; }
+    default: { return NDB_UNSUPPORTED_FUNC; }
+    }
+  };
+
+  static NDB_FUNC_TYPE negate(NDB_FUNC_TYPE fun)
   {
-    uint i;
-    for (i=0; 
-         fun != neg_map[i].pos_fun &&
-           neg_map[i].pos_fun != Item_func::UNKNOWN_FUNC;
-         i++);
+    uint i= (uint) fun;
+    DBUG_ASSERT(fun == neg_map[i].pos_fun);
     return  neg_map[i].neg_fun;
   };