ndb - fix TUP filtering of LIKE / NOT_LIKE

ndb/include/util/NdbSqlUtil.hpp

@@ -45,9 +45,9 @@ public:
   typedef int Cmp(const void* info, const void* p1, unsigned n1, const void* p2, unsigned n2, bool full);
 
   /**
-   * Prototype for "like" comparison.  Defined for the 3 char string
-   * types.  Second argument must have same type-specific format.
-   * Returns >0 on match, 0 on no match, <0 on bad data.
+   * Prototype for "like" comparison.  Defined for string types.  Second
+   * argument must have same type-specific format.  Returns 0 on match,
+   * +1 on no match, and -1 on bad data.
    *
    * Uses default special chars ( \ % _ ).
    *

ndb/src/common/util/NdbSqlUtil.cpp

@@ -795,14 +795,18 @@ NdbSqlUtil::cmpTimestamp(const void* info, const void* p1, unsigned n1, const vo
 
 // like
 
+static const int ndb_wild_prefix = '\\';
+static const int ndb_wild_one = '_';
+static const int ndb_wild_many = '%';
+
 int
 NdbSqlUtil::likeChar(const void* info, const void* p1, unsigned n1, const void* p2, unsigned n2)
 {
   const char* v1 = (const char*)p1;
   const char* v2 = (const char*)p2;
   CHARSET_INFO* cs = (CHARSET_INFO*)(info);
-  int k = (cs->coll->wildcmp)(cs, v1, v1 + n1, v2, v2 + n2, wild_prefix, wild_one, wild_many);
-  return k;
+  int k = (cs->coll->wildcmp)(cs, v1, v1 + n1, v2, v2 + n2, ndb_wild_prefix, ndb_wild_one, ndb_wild_many);
+  return k == 0 ? 0 : +1;
 }
 
 int
@@ -825,8 +829,8 @@ NdbSqlUtil::likeVarchar(const void* info, const void* p1, unsigned n1, const voi
       const char* w1 = (const char*)v1 + lb;
       const char* w2 = (const char*)v2 + lb;
       CHARSET_INFO* cs = (CHARSET_INFO*)(info);
-      int k = (cs->coll->wildcmp)(cs, w1, w1 + m1, w2, w2 + m2, wild_prefix, wild_one, wild_many);
-      return k;
+      int k = (cs->coll->wildcmp)(cs, w1, w1 + m1, w2, w2 + m2, ndb_wild_prefix, ndb_wild_one, ndb_wild_many);
+      return k == 0 ? 0 : +1;
     }
   }
   return -1;
@@ -852,8 +856,8 @@ NdbSqlUtil::likeLongvarchar(const void* info, const void* p1, unsigned n1, const
       const char* w1 = (const char*)v1 + lb;
       const char* w2 = (const char*)v2 + lb;
       CHARSET_INFO* cs = (CHARSET_INFO*)(info);
-      int k = (cs->coll->wildcmp)(cs, w1, w1 + m1, w2, w2 + m2, wild_prefix, wild_one, wild_many);
-      return k;
+      int k = (cs->coll->wildcmp)(cs, w1, w1 + m1, w2, w2 + m2, ndb_wild_prefix, ndb_wild_one, ndb_wild_many);
+      return k == 0 ? 0 : +1;
     }
   }
   return -1;

ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp

@@ -1861,56 +1861,57 @@ int Dbtup::interpreterNextLab(Signal* signal,
 
 	bool r1_null = ah.isNULL();
 	bool r2_null = argLen == 0;
-	int res;
+	int res1;
         if (cond != Interpreter::LIKE &&
             cond != Interpreter::NOT_LIKE) {
           if (r1_null || r2_null) {
             // NULL==NULL and NULL<not-NULL
-            res = r1_null && r2_null ? 0 : r1_null ? -1 : 1;
+            res1 = r1_null && r2_null ? 0 : r1_null ? -1 : 1;
           } else {
-            res = (*sqlType.m_cmp)(cs, s1, attrLen, s2, argLen, true);
+            res1 = (*sqlType.m_cmp)(cs, s1, attrLen, s2, argLen, true);
           }
 	} else {
           if (r1_null || r2_null) {
             // NULL like NULL is true (has no practical use)
-            res =  r1_null && r2_null ? 0 : -1;
+            res1 =  r1_null && r2_null ? 0 : -1;
           } else {
-            res = (*sqlType.m_like)(cs, s1, attrLen, s2, argLen);
+            res1 = (*sqlType.m_like)(cs, s1, attrLen, s2, argLen);
           }
         }
 
+        int res = 0;
         switch ((Interpreter::BinaryCondition)cond) {
         case Interpreter::EQ:
-          res = (res == 0);
+          res = (res1 == 0);
           break;
         case Interpreter::NE:
-          res = (res != 0);
+          res = (res1 != 0);
           break;
         // note the condition is backwards
         case Interpreter::LT:
-          res = (res > 0);
+          res = (res1 > 0);
           break;
         case Interpreter::LE:
-          res = (res >= 0);
+          res = (res1 >= 0);
           break;
         case Interpreter::GT:
-          res = (res < 0);
+          res = (res1 < 0);
           break;
         case Interpreter::GE:
-          res = (res <= 0);
+          res = (res1 <= 0);
           break;
         case Interpreter::LIKE:
-          res = (res > 0);
+          res = (res1 == 0);
           break;
         case Interpreter::NOT_LIKE:
-          res = (res == 0);
+          res = (res1 == 1);
           break;
 	  // XXX handle invalid value
         }
 #ifdef TRACE_INTERPRETER
-	ndbout_c("cond=%u diff=%d vc=%d nopad=%d attr(%d) = >%.*s<(%d) str=>%.*s<(%d) -> res = %d",
-		 cond, diff, vchr, nopad,
-		 attrId >> 16, attrLen, s1, attrLen, argLen, s2, argLen, res);
+	ndbout_c("cond=%u attr(%d)='%.*s'(%d) str='%.*s'(%d) res1=%d res=%d",
+		 cond, attrId >> 16,
+                 attrLen, s1, attrLen, argLen, s2, argLen, res1, res);
 #endif
         if (res)
           TprogramCounter = brancher(theInstruction, TprogramCounter);