bug#15682 - ndb - invalid handling of varchar in position/rnd_pos

mysql-test/r/ndb_basic.result:
  bug#15682 - invalid handling of varchar in position/rnd_pos
mysql-test/t/ndb_basic.test:
  bug#15682 - invalid handling of varchar in position/rnd_pos
ndb/src/kernel/blocks/dbtc/Dbtc.hpp:
  New error code for invalid key
ndb/src/kernel/blocks/dbtc/DbtcMain.cpp:
  Handle invalid key gracefully
ndb/src/kernel/vm/SimulatedBlock.cpp:
  Handle invalid key gracefully
ndb/src/ndbapi/ndberror.c:
  New error code
sql/ha_ndbcluster.cc:
  Fix varchar keys in position/rnd_pos

mysql-test/r/ndb_basic.result

@@ -673,3 +673,19 @@ select * from atablewithareallylongandirritatingname;
 a
 2
 drop table atablewithareallylongandirritatingname;
+create table t1 (f1 varchar(50), f2 text,f3 int, primary key(f1)) engine=NDB;
+insert into t1 (f1,f2,f3)VALUES("111111","aaaaaa",1);
+insert into t1 (f1,f2,f3)VALUES("222222","bbbbbb",2);
+select * from t1 order by f1;
+f1	f2	f3
+111111	aaaaaa	1
+222222	bbbbbb	2
+select * from t1 order by f2;
+f1	f2	f3
+111111	aaaaaa	1
+222222	bbbbbb	2
+select * from t1 order by f3;
+f1	f2	f3
+111111	aaaaaa	1
+222222	bbbbbb	2
+drop table t1;

mysql-test/t/ndb_basic.test

@@ -615,3 +615,14 @@ create table atablewithareallylongandirritatingname (a int);
 insert into atablewithareallylongandirritatingname values (2);
 select * from atablewithareallylongandirritatingname;
 drop table atablewithareallylongandirritatingname;
+
+#
+# Bug#15682
+#
+create table t1 (f1 varchar(50), f2 text,f3 int, primary key(f1)) engine=NDB;
+insert into t1 (f1,f2,f3)VALUES("111111","aaaaaa",1);
+insert into t1 (f1,f2,f3)VALUES("222222","bbbbbb",2);
+select * from t1 order by f1;
+select * from t1 order by f2;
+select * from t1 order by f3;
+drop table t1;

ndb/src/kernel/blocks/dbtc/Dbtc.hpp

@@ -141,6 +141,8 @@
 #define ZALREADYEXIST 630
 #define ZINCONSISTENTHASHINDEX 892
 #define ZNOTUNIQUE 893
+
+#define ZINVALID_KEY 290
 #endif
 
 class Dbtc: public SimulatedBlock {

ndb/src/kernel/blocks/dbtc/DbtcMain.cpp

@@ -2313,7 +2313,10 @@ Dbtc::handle_special_hash(Uint32 dstHash[4], Uint32* src, Uint32 srcLen,
   {
     keyPartLenPtr = keyPartLen;
     dstPos = xfrm_key(tabPtrI, src, dst, sizeof(Tmp) >> 2, keyPartLenPtr);
-    ndbrequire(dstPos);
+    if (unlikely(dstPos == 0))
+    {
+      goto error;
+    }
   } 
   else 
   {
@@ -2334,6 +2337,10 @@ Dbtc::handle_special_hash(Uint32 dstHash[4], Uint32* src, Uint32 srcLen,
     dstHash[1] = tmp[1];
   }
   return true;  // success
+
+error:
+  terrorCode = ZINVALID_KEY;
+  return false;
 }
 
 /*
@@ -2941,7 +2948,15 @@ void Dbtc::tckeyreq050Lab(Signal* signal)
   UintR tnoOfStandby;
   UintR tnodeinfo;
 
+  terrorCode = 0;
+
   hash(signal); /* NOW IT IS TIME TO CALCULATE THE HASH VALUE*/
+  
+  if (unlikely(terrorCode))
+  {
+    releaseAtErrorLab(signal);
+    return;
+  }
 
   CacheRecord * const regCachePtr = cachePtr.p;
   TcConnectRecord * const regTcPtr = tcConnectptr.p;

ndb/src/kernel/vm/SimulatedBlock.cpp

@@ -1890,7 +1890,10 @@ SimulatedBlock::xfrm_key(Uint32 tab, const Uint32* src,
 	AttributeDescriptor::getType(keyAttr.attributeDescriptor);
       Uint32 lb, len;
       bool ok = NdbSqlUtil::get_var_length(typeId, srcPtr, srcBytes, lb, len);
-      ndbrequire(ok);
+      if (unlikely(!ok))
+      {
+	return 0;
+      }
       Uint32 xmul = cs->strxfrm_multiply;
       if (xmul == 0)
 	xmul = 1;
@@ -1902,7 +1905,10 @@ SimulatedBlock::xfrm_key(Uint32 tab, const Uint32* src,
       Uint32 dstLen = xmul * (srcBytes - lb);
       ndbrequire(dstLen <= ((dstSize - dstPos) << 2));
       int n = NdbSqlUtil::strnxfrm_bug7284(cs, dstPtr, dstLen, srcPtr + lb, len);
-      ndbrequire(n != -1);
+      if (unlikely(n == -1))
+      {
+	return 0;
+      }
       while ((n & 3) != 0) 
       {
 	dstPtr[n++] = 0;

ndb/src/ndbapi/ndberror.c

@@ -227,6 +227,7 @@ ErrorBundle ErrorCodes[] = {
   { 277,  IE, "277" },
   { 278,  IE, "278" },
   { 287,  IE, "Index corrupted" },
+  { 290,  IE, "Corrupt key in TC, unable to xfrm" },
   { 631,  IE, "631" },
   { 632,  IE, "632" },
   { 702,  IE, "Request to non-master" },
@@ -295,7 +296,6 @@ ErrorBundle ErrorCodes[] = {
   { 4608, AE, "You can not takeOverScan unless you have used openScanExclusive"},
   { 4609, AE, "You must call nextScanResult before trying to takeOverScan"},
   { 4232, AE, "Parallelism can only be between 1 and 240" },
-  { 290,  AE, "Scan not started or has been closed by kernel due to timeout" },
 
   /** 
    * Event schema errors

sql/ha_ndbcluster.cc

@@ -2796,8 +2796,26 @@ void ha_ndbcluster::position(const byte *record)
         }      
         *buff++= 0;
       }
-      memcpy(buff, record + key_part->offset, key_part->length);
-      buff += key_part->length;
+
+      size_t len = key_part->length;
+      const byte * ptr = record + key_part->offset;
+      Field *field = key_part->field;
+      if ((field->type() ==  MYSQL_TYPE_VARCHAR) &&
+	  ((Field_varstring*)field)->length_bytes == 1)
+      {
+	/** 
+	 * Keys always use 2 bytes length
+	 */
+	buff[0] = ptr[0];
+	buff[1] = 0;
+	memcpy(buff+2, ptr + 1, len);	
+	len += 2;
+      }
+      else
+      {
+	memcpy(buff, ptr, len);
+      }
+      buff += len;
     }
   } 
   else