ndb - bug#19956 (occurs only in v5.0)

mysql-test/r/ndb_blob.result:
  bug#19956 : var* pk
mysql-test/t/ndb_blob.test:
  bug#19956 : var* pk
ndb/src/ndbapi/NdbDictionaryImpl.hpp:
  bug#19956 - check var* length bytes and zero rest of value if any blob attribute
ndb/src/ndbapi/NdbOperationSearch.cpp:
  bug#19956 - check var* length bytes and zero rest of value if any blob attribute

mysql-test/r/ndb_blob.result

@@ -500,3 +500,69 @@ select count(*) from t1;
 count(*)
 0
 drop table t1;
+create table t1 (
+a varchar(40) not null,
+b mediumint not null,
+t text,
+c varchar(2) not null,
+d bigint not null,
+primary key (a,b,c),
+key (c,a),
+unique key (d)
+) engine=ndb;
+insert into t1 (a,b,c,d,t) values ('a',1110,'a',1,@v1);
+insert into t1 (a,b,c,d,t) values ('b',1110,'a',2,@v2);
+insert into t1 (a,b,c,d,t) values ('a',1110,'b',3,@v3);
+insert into t1 (a,b,c,d,t) values ('b',1110,'b',4,@v4);
+select a,b,c,d,sha1(t) from t1 order by c,a;
+a	b	c	d	sha1(t)
+a	1110	a	1	558a30713786aa72f66abc1e6a521d55aacdeeb5
+b	1110	a	2	b238654911689bfb626a3ef9dba4a1ca074e6a5e
+a	1110	b	3	2b6515f29c20b8e9e17cc597527e516c0de8d612
+b	1110	b	4	NULL
+select a,b,c,d,sha1(t) from t1 where a='a' and b=1110 and c='a';
+a	b	c	d	sha1(t)
+a	1110	a	1	558a30713786aa72f66abc1e6a521d55aacdeeb5
+select a,b,c,d,sha1(t) from t1 where a='a' and b=1110 and c='b';
+a	b	c	d	sha1(t)
+a	1110	b	3	2b6515f29c20b8e9e17cc597527e516c0de8d612
+update t1 set t=@v4 where a='b' and b=1110 and c='a';
+update t1 set t=@v2 where a='b' and b=1110 and c='b';
+select a,b,c,d,sha1(t) from t1 order by c,a;
+a	b	c	d	sha1(t)
+a	1110	a	1	558a30713786aa72f66abc1e6a521d55aacdeeb5
+b	1110	a	2	NULL
+a	1110	b	3	2b6515f29c20b8e9e17cc597527e516c0de8d612
+b	1110	b	4	b238654911689bfb626a3ef9dba4a1ca074e6a5e
+update t1 set t=@v2 where d=2;
+update t1 set t=@v4 where d=4;
+select a,b,c,d,sha1(t) from t1 order by c,a;
+a	b	c	d	sha1(t)
+a	1110	a	1	558a30713786aa72f66abc1e6a521d55aacdeeb5
+b	1110	a	2	b238654911689bfb626a3ef9dba4a1ca074e6a5e
+a	1110	b	3	2b6515f29c20b8e9e17cc597527e516c0de8d612
+b	1110	b	4	NULL
+update t1 set t=@v4 where a='b' and c='a';
+update t1 set t=@v2 where a='b' and c='b';
+select a,b,c,d,sha1(t) from t1 order by c,a;
+a	b	c	d	sha1(t)
+a	1110	a	1	558a30713786aa72f66abc1e6a521d55aacdeeb5
+b	1110	a	2	NULL
+a	1110	b	3	2b6515f29c20b8e9e17cc597527e516c0de8d612
+b	1110	b	4	b238654911689bfb626a3ef9dba4a1ca074e6a5e
+update t1 set t=@v2 where b+d=1112;
+update t1 set t=@v4 where b+d=1114;
+select a,b,c,d,sha1(t) from t1 order by c,a;
+a	b	c	d	sha1(t)
+a	1110	a	1	558a30713786aa72f66abc1e6a521d55aacdeeb5
+b	1110	a	2	b238654911689bfb626a3ef9dba4a1ca074e6a5e
+a	1110	b	3	2b6515f29c20b8e9e17cc597527e516c0de8d612
+b	1110	b	4	NULL
+delete from t1 where a='a' and b=1110 and c='a';
+delete from t1 where a='b' and c='a';
+delete from t1 where d=3;
+delete from t1 where b+d=1114;
+select count(*) from t1;
+count(*)
+0
+drop table t1;

mysql-test/t/ndb_blob.test

@@ -428,4 +428,60 @@ truncate t1;
 select count(*) from t1;
 drop table t1;
 
+# -- bug#19956 - var* key, complex key
+
+create table t1 (
+  a varchar(40) not null,
+  b mediumint not null,
+  t text,
+  c varchar(2) not null,
+  d bigint not null,
+  primary key (a,b,c),
+  key (c,a),
+  unique key (d)
+) engine=ndb;
+
+--disable_query_log
+set @s1 = 'rggurloniukyehuxdbfkkyzlceixzrehqhvxvxbpwizzvjzpucqmzrhzxzfau';
+set @s2 = 'ykyymbzqgqlcjhlhmyqelfoaaohvtbekvifukdtnvcrrjveevfakxarxexomz';
+set @s3 = 'dbnfqyzgtqxalcrwtfsqabknvtfcbpoonxsjiqvmhnfikxxhcgoexlkoezvah';
+set @v1 = repeat(@s1,123);
+set @v2 = repeat(@s2,234);
+set @v3 = repeat(@s3,345);
+set @v4 = NULL;
+--enable_query_log
+
+insert into t1 (a,b,c,d,t) values ('a',1110,'a',1,@v1);
+insert into t1 (a,b,c,d,t) values ('b',1110,'a',2,@v2);
+insert into t1 (a,b,c,d,t) values ('a',1110,'b',3,@v3);
+insert into t1 (a,b,c,d,t) values ('b',1110,'b',4,@v4);
+select a,b,c,d,sha1(t) from t1 order by c,a;
+
+select a,b,c,d,sha1(t) from t1 where a='a' and b=1110 and c='a';
+select a,b,c,d,sha1(t) from t1 where a='a' and b=1110 and c='b';
+
+update t1 set t=@v4 where a='b' and b=1110 and c='a';
+update t1 set t=@v2 where a='b' and b=1110 and c='b';
+select a,b,c,d,sha1(t) from t1 order by c,a;
+
+update t1 set t=@v2 where d=2;
+update t1 set t=@v4 where d=4;
+select a,b,c,d,sha1(t) from t1 order by c,a;
+
+update t1 set t=@v4 where a='b' and c='a';
+update t1 set t=@v2 where a='b' and c='b';
+select a,b,c,d,sha1(t) from t1 order by c,a;
+
+update t1 set t=@v2 where b+d=1112;
+update t1 set t=@v4 where b+d=1114;
+select a,b,c,d,sha1(t) from t1 order by c,a;
+
+delete from t1 where a='a' and b=1110 and c='a';
+delete from t1 where a='b' and c='a';
+delete from t1 where d=3;
+delete from t1 where b+d=1114;
+select count(*) from t1;
+
+drop table t1;
+
 # End of 4.1 tests

ndb/src/ndbapi/NdbDictionaryImpl.hpp

@@ -91,6 +91,10 @@ public:
   NdbDictionary::Column * m_facade;
 
   static NdbDictionary::Column * create_psuedo(const char *);
+
+  // Get total length in bytes, used by NdbOperation
+  // backported from 5.1
+  bool get_var_length(const void* value, Uint32& len) const;
 };
 
 class NdbTableImpl : public NdbDictionary::Table, public NdbDictObjectImpl {
@@ -489,6 +493,27 @@ NdbColumnImpl::getBlobType() const {
 	  m_type == NdbDictionary::Column::Text);
 }
 
+inline
+bool
+NdbColumnImpl::get_var_length(const void* value, Uint32& len) const
+{
+  Uint32 max_len = m_attrSize * m_arraySize;
+  switch (m_type) {
+  case NdbDictionary::Column::Varchar:
+  case NdbDictionary::Column::Varbinary:
+    len = 1 + *((Uint8*)value);
+    break;
+  case NdbDictionary::Column::Longvarchar:
+  case NdbDictionary::Column::Longvarbinary:
+    len = 2 + uint2korr((char*)value);
+    break;
+  default:
+    len = max_len;
+    return true;
+  }
+  return (len <= max_len);
+}
+
 inline
 NdbTableImpl &
 NdbTableImpl::getImpl(NdbDictionary::Table & t){

ndb/src/ndbapi/NdbOperationSearch.cpp

@@ -71,6 +71,7 @@ NdbOperation::equal_impl(const NdbColumnImpl* tAttrInfo,
   Uint32 tKeyInfoPosition;
   const char* aValue = aValuePassed;
   Uint64 tempData[512];
+  Uint64 tempData2[512];
 
   if ((theStatus == OperationDefined) &&
       (aValue != NULL) &&
@@ -131,6 +132,19 @@ NdbOperation::equal_impl(const NdbColumnImpl* tAttrInfo,
     OperationType tOpType = theOperationType;
     Uint32 sizeInBytes = tAttrInfo->m_attrSize * tAttrInfo->m_arraySize;
 
+    Uint32 real_len;
+    if (! tAttrInfo->get_var_length(aValue, real_len)) {
+      setErrorCodeAbort(4209);
+      DBUG_RETURN(-1);
+    }
+
+    // 5.0 fixed storage + NdbBlob uses full size => pad var* with nulls
+    if (real_len < sizeInBytes && m_currentTable->m_noOfBlobs != 0) {
+      memcpy(tempData2, aValue, real_len);
+      memset((char*)tempData2 + real_len, 0, sizeInBytes - real_len);
+      aValue = (char*)tempData2;
+    }
+
     {
       /************************************************************************
        * Check if the pointer of the value passed is aligned on a 4 byte