ndb - bug#17421, changes NDB API pushdown LIKE arg to plain char

mysql-test/r/ndb_condition_pushdown.result:
  bug#17421, changes NDB API pushdown LIKE arg to plain char
mysql-test/t/ndb_condition_pushdown.test:
  bug#17421, changes NDB API pushdown LIKE arg to plain char
ndb/include/ndbapi/NdbOperation.hpp:
  bug#17421, changes NDB API pushdown LIKE arg to plain char
ndb/include/util/NdbSqlUtil.hpp:
  bug#17421, changes NDB API pushdown LIKE arg to plain char
ndb/src/common/util/NdbSqlUtil.cpp:
  bug#17421, changes NDB API pushdown LIKE arg to plain char

5 files changed, 117 insertions, 17 deletions

diff --git a/mysql-test/r/ndb_condition_pushdown.result b/mysql-test/r/ndb_condition_pushdown.result
index c516740d02c..27a510b252f 100644
--- a/mysql-test/r/ndb_condition_pushdown.result
+++ b/mysql-test/r/ndb_condition_pushdown.result
@@ -1782,5 +1782,65 @@ select * from t5 where b like '%jo%' order by a;
 a	b
 1	jonas
 3	johan
+drop table t1;
+create table t1 (a int, b varchar(3), primary key using hash(a))
+engine=ndb;
+insert into t1 values (1,'a'), (2,'ab'), (3,'abc');
+set engine_condition_pushdown = off;
+select * from t1 where b like 'ab';
+a	b
+2	ab
+select * from t1 where b like 'ab' or b like 'ab';
+a	b
+2	ab
+select * from t1 where b like 'abc';
+a	b
+3	abc
+select * from t1 where b like 'abc' or b like 'abc';
+a	b
+3	abc
+set engine_condition_pushdown = on;
+select * from t1 where b like 'ab';
+a	b
+2	ab
+select * from t1 where b like 'ab' or b like 'ab';
+a	b
+2	ab
+select * from t1 where b like 'abc';
+a	b
+3	abc
+select * from t1 where b like 'abc' or b like 'abc';
+a	b
+3	abc
+drop table t1;
+create table t1 (a int, b char(3), primary key using hash(a))
+engine=ndb;
+insert into t1 values (1,'a'), (2,'ab'), (3,'abc');
+set engine_condition_pushdown = off;
+select * from t1 where b like 'ab';
+a	b
+2	ab
+select * from t1 where b like 'ab' or b like 'ab';
+a	b
+2	ab
+select * from t1 where b like 'abc';
+a	b
+3	abc
+select * from t1 where b like 'abc' or b like 'abc';
+a	b
+3	abc
+set engine_condition_pushdown = on;
+select * from t1 where b like 'ab';
+a	b
+2	ab
+select * from t1 where b like 'ab' or b like 'ab';
+a	b
+2	ab
+select * from t1 where b like 'abc';
+a	b
+3	abc
+select * from t1 where b like 'abc' or b like 'abc';
+a	b
+3	abc
 set engine_condition_pushdown = @old_ecpd;
 DROP TABLE t1,t2,t3,t4,t5;
diff --git a/mysql-test/t/ndb_condition_pushdown.test b/mysql-test/t/ndb_condition_pushdown.test
index 1e6e105bc61..398ca3c502c 100644
--- a/mysql-test/t/ndb_condition_pushdown.test
+++ b/mysql-test/t/ndb_condition_pushdown.test
@@ -1649,5 +1649,42 @@ set engine_condition_pushdown = on;
 explain select * from t5 where b like '%jo%';
 select * from t5 where b like '%jo%' order by a;
 
+# bug#17421 -1
+drop table t1;
+create table t1 (a int, b varchar(3), primary key using hash(a))
+engine=ndb;
+insert into t1 values (1,'a'), (2,'ab'), (3,'abc');
+# in TUP the constants 'ab' 'abc' were expected in varchar format
+# "like" returned error which became "false"
+# scan filter negates "or" which exposes the bug
+set engine_condition_pushdown = off;
+select * from t1 where b like 'ab';
+select * from t1 where b like 'ab' or b like 'ab';
+select * from t1 where b like 'abc';
+select * from t1 where b like 'abc' or b like 'abc';
+set engine_condition_pushdown = on;
+select * from t1 where b like 'ab';
+select * from t1 where b like 'ab' or b like 'ab';
+select * from t1 where b like 'abc';
+select * from t1 where b like 'abc' or b like 'abc';
+
+# bug#17421 -2
+drop table t1;
+create table t1 (a int, b char(3), primary key using hash(a))
+engine=ndb;
+insert into t1 values (1,'a'), (2,'ab'), (3,'abc');
+# test that incorrect MySQL behaviour is preserved
+# 'ab ' LIKE 'ab' is true in MySQL
+set engine_condition_pushdown = off;
+select * from t1 where b like 'ab';
+select * from t1 where b like 'ab' or b like 'ab';
+select * from t1 where b like 'abc';
+select * from t1 where b like 'abc' or b like 'abc';
+set engine_condition_pushdown = on;
+select * from t1 where b like 'ab';
+select * from t1 where b like 'ab' or b like 'ab';
+select * from t1 where b like 'abc';
+select * from t1 where b like 'abc' or b like 'abc';
+
 set engine_condition_pushdown = @old_ecpd;
 DROP TABLE t1,t2,t3,t4,t5;
diff --git a/ndb/include/ndbapi/NdbOperation.hpp b/ndb/include/ndbapi/NdbOperation.hpp
index 6237740c3c6..4db541f7fe4 100644
--- a/ndb/include/ndbapi/NdbOperation.hpp
+++ b/ndb/include/ndbapi/NdbOperation.hpp
@@ -631,6 +631,10 @@ public:
 		    bool nopad, Uint32 Label);
   int branch_col_ge(Uint32 ColId, const void * val, Uint32 len, 
 		    bool nopad, Uint32 Label);
+  /**
+   * The argument is always plain char, even if the field is varchar
+   * (changed in 5.0.22).
+   */
   int branch_col_like(Uint32 ColId, const void *, Uint32 len, 
 		      bool nopad, Uint32 Label);
   int branch_col_notlike(Uint32 ColId, const void *, Uint32 len, 
diff --git a/ndb/include/util/NdbSqlUtil.hpp b/ndb/include/util/NdbSqlUtil.hpp
index 36a75136c45..ff2d9766f81 100644
--- a/ndb/include/util/NdbSqlUtil.hpp
+++ b/ndb/include/util/NdbSqlUtil.hpp
@@ -45,14 +45,11 @@ 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 string types.  Second
-   * argument must have same type-specific format.  Returns 0 on match,
-   * +1 on no match, and -1 on bad data.
+   * Prototype for "like" comparison.  Defined for string types.  First
+   * argument can be fixed or var* type, second argument is fixed.
+   * Returns 0 on match, +1 on no match, and -1 on bad data.
    *
    * Uses default special chars ( \ % _ ).
-   *
-   * TODO convert special chars to the cs so that ucs2 etc works
-   * TODO allow user-defined escape ( \ )
    */
   typedef int Like(const void* info, const void* p1, unsigned n1, const void* p2, unsigned n2);
 
diff --git a/ndb/src/common/util/NdbSqlUtil.cpp b/ndb/src/common/util/NdbSqlUtil.cpp
index f2506eda6d4..f3d70a5734a 100644
--- a/ndb/src/common/util/NdbSqlUtil.cpp
+++ b/ndb/src/common/util/NdbSqlUtil.cpp
@@ -805,7 +805,9 @@ NdbSqlUtil::likeChar(const void* info, const void* p1, unsigned n1, const void*
   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, ndb_wild_prefix, ndb_wild_one, ndb_wild_many);
+  // strip end spaces to match (incorrect) MySQL behaviour
+  n1 = (*cs->cset->lengthsp)(cs, v1, n1);
+  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;
 }
 
@@ -820,16 +822,16 @@ int
 NdbSqlUtil::likeVarchar(const void* info, const void* p1, unsigned n1, const void* p2, unsigned n2)
 {
   const unsigned lb = 1;
-  if (n1 >= lb && n2 >= lb) {
+  if (n1 >= lb) {
     const uchar* v1 = (const uchar*)p1;
     const uchar* v2 = (const uchar*)p2;
     unsigned m1 = *v1;
-    unsigned m2 = *v2;
-    if (lb + m1 <= n1 && lb + m2 <= n2) {
+    unsigned m2 = n2;
+    if (lb + m1 <= n1) {
       const char* w1 = (const char*)v1 + lb;
-      const char* w2 = (const char*)v2 + lb;
+      const char* w2 = (const char*)v2;
       CHARSET_INFO* cs = (CHARSET_INFO*)(info);
-      int k = (cs->coll->wildcmp)(cs, w1, w1 + m1, w2, w2 + m2, ndb_wild_prefix, ndb_wild_one, ndb_wild_many);
+      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;
     }
   }
@@ -847,16 +849,16 @@ int
 NdbSqlUtil::likeLongvarchar(const void* info, const void* p1, unsigned n1, const void* p2, unsigned n2)
 {
   const unsigned lb = 2;
-  if (n1 >= lb && n2 >= lb) {
+  if (n1 >= lb) {
     const uchar* v1 = (const uchar*)p1;
     const uchar* v2 = (const uchar*)p2;
     unsigned m1 = uint2korr(v1);
-    unsigned m2 = uint2korr(v2);
-    if (lb + m1 <= n1 && lb + m2 <= n2) {
+    unsigned m2 = n2;
+    if (lb + m1 <= n1) {
       const char* w1 = (const char*)v1 + lb;
-      const char* w2 = (const char*)v2 + lb;
+      const char* w2 = (const char*)v2;
       CHARSET_INFO* cs = (CHARSET_INFO*)(info);
-      int k = (cs->coll->wildcmp)(cs, w1, w1 + m1, w2, w2 + m2, ndb_wild_prefix, ndb_wild_one, ndb_wild_many);
+      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;
     }
   }