bk commit into 5.1 tree (pekka:1.2150) BUG#18075

[pekka]

Below is the list of changes that have just been committed into a local
5.1 repository of pekka. When pekka does a push these changes will
be propagated to the main repository and, within 24 hours after the
push, to the public repository.
For information on how to access the public repository
see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html

ChangeSet
  1.2150 06/03/09 13:27:58 pekka@stripped +10 -0
  ndb - fix bug#18075 varsize PK + blobs

  storage/ndb/test/ndbapi/testBlobs.cpp
    1.31 06/03/09 13:26:48 pekka@stripped +1 -1
    fast fix bug#18075

  storage/ndb/src/ndbapi/ndberror.c
    1.53 06/03/09 13:26:48 pekka@stripped +2 -1
    fast fix bug#18075

  storage/ndb/src/ndbapi/NdbScanOperation.cpp
    1.77 06/03/09 13:26:48 pekka@stripped +9 -2
    fast fix bug#18075

  storage/ndb/src/ndbapi/NdbRecAttr.cpp
    1.29 06/03/09 13:26:48 pekka@stripped +1 -1
    fast fix bug#18075

  storage/ndb/src/ndbapi/NdbOperationSearch.cpp
    1.29 06/03/09 13:26:48 pekka@stripped +3 -2
    fast fix bug#18075

  storage/ndb/src/ndbapi/NdbBlobImpl.hpp
    1.3 06/03/09 13:26:48 pekka@stripped +2 -0
    fast fix bug#18075

  storage/ndb/src/ndbapi/NdbBlob.cpp
    1.38 06/03/09 13:26:48 pekka@stripped +99 -18
    fast fix bug#18075

  storage/ndb/include/ndbapi/NdbScanOperation.hpp
    1.37 06/03/09 13:26:48 pekka@stripped +1 -1
    fast fix bug#18075

  storage/ndb/include/ndbapi/NdbOperation.hpp
    1.35 06/03/09 13:26:48 pekka@stripped +1 -1
    fast fix bug#18075

  storage/ndb/include/ndbapi/NdbBlob.hpp
    1.20 06/03/09 13:26:48 pekka@stripped +4 -0
    fast fix bug#18075

# This is a BitKeeper patch.  What follows are the unified diffs for the
# set of deltas contained in the patch.  The rest of the patch, the part
# that BitKeeper cares about, is below these diffs.
# User:	pekka
# Host:	orca.ndb.mysql.com
# Root:	/space/pekka/ndb/version/my51-dict

--- 1.34/storage/ndb/include/ndbapi/NdbOperation.hpp	2005-11-07 12:19:05 +01:00
+++ 1.35/storage/ndb/include/ndbapi/NdbOperation.hpp	2006-03-09 13:26:48 +01:00
@@ -881,7 +881,7 @@
   Uint32 ptr2int() { return theReceiver.getId(); };
 
   // get table or index key from prepared signals
-  int getKeyFromTCREQ(Uint32* data, unsigned size);
+  int getKeyFromTCREQ(Uint32* data, Uint32 & size);
 
 /******************************************************************************
  * These are the private variables that are defined in the operation objects.

--- 1.36/storage/ndb/include/ndbapi/NdbScanOperation.hpp	2006-03-07 16:36:12 +01:00
+++ 1.37/storage/ndb/include/ndbapi/NdbScanOperation.hpp	2006-03-09 13:26:48 +01:00
@@ -240,7 +240,7 @@
   void receiver_completed(NdbReceiver*);
   void execCLOSE_SCAN_REP();
 
-  int getKeyFromKEYINFO20(Uint32* data, unsigned size);
+  int getKeyFromKEYINFO20(Uint32* data, Uint32 & size);
   NdbOperation*	takeOverScanOp(OperationType opType, NdbTransaction*);
   
   bool m_ordered;

--- 1.28/storage/ndb/src/ndbapi/NdbOperationSearch.cpp	2005-11-07 12:19:09 +01:00
+++ 1.29/storage/ndb/src/ndbapi/NdbOperationSearch.cpp	2006-03-09 13:26:48 +01:00
@@ -472,7 +472,8 @@
 NdbOperation::reorderKEYINFO()
 {
   Uint32 data[4000];
-  getKeyFromTCREQ(data, 4000);
+  Uint32 size = 4000;
+  getKeyFromTCREQ(data, size);
   Uint32 pos = 1;
   Uint32 k;
   for (k = 0; k < m_accessTable->m_noOfKeys; k++) {
@@ -501,7 +502,7 @@
 }
 
 int
-NdbOperation::getKeyFromTCREQ(Uint32* data, unsigned size)
+NdbOperation::getKeyFromTCREQ(Uint32* data, Uint32 & size)
 {
   assert(size >= theTupKeyLen && theTupKeyLen > 0);
   size = theTupKeyLen;

--- 1.28/storage/ndb/src/ndbapi/NdbRecAttr.cpp	2006-01-16 14:26:28 +01:00
+++ 1.29/storage/ndb/src/ndbapi/NdbRecAttr.cpp	2006-03-09 13:26:48 +01:00
@@ -199,7 +199,7 @@
       out << hex << "H'" << r.u_32_value() << dec;
       break;
     case NdbDictionary::Column::Unsigned:
-      out << r.u_32_value();
+      out << *((Uint32*)r.aRef() + j);
       break;
     case NdbDictionary::Column::Smallunsigned:
       out << r.u_short_value();

--- 1.76/storage/ndb/src/ndbapi/NdbScanOperation.cpp	2006-03-08 08:34:54 +01:00
+++ 1.77/storage/ndb/src/ndbapi/NdbScanOperation.cpp	2006-03-09 13:26:48 +01:00
@@ -912,13 +912,20 @@
  *     the scan process. 
  ****************************************************************************/
 int
-NdbScanOperation::getKeyFromKEYINFO20(Uint32* data, unsigned size)
+NdbScanOperation::getKeyFromKEYINFO20(Uint32* data, Uint32 & size)
 {
   NdbRecAttr * tRecAttr = m_curr_row;
   if(tRecAttr)
   {
     const Uint32 * src = (Uint32*)tRecAttr->aRef();
-    memcpy(data, src, 4*size);
+
+    assert(tRecAttr->get_size_in_bytes() > 0);
+    assert(tRecAttr->get_size_in_bytes() < 65536);
+    const Uint32 len = (tRecAttr->get_size_in_bytes() + 3)/4-1;
+
+    assert(size >= len);
+    memcpy(data, src, 4*len);
+    size = len;
     return 0;
   }
   return -1;

--- 1.30/storage/ndb/test/ndbapi/testBlobs.cpp	2006-02-09 11:34:38 +01:00
+++ 1.31/storage/ndb/test/ndbapi/testBlobs.cpp	2006-03-09 13:26:48 +01:00
@@ -223,7 +223,7 @@
 {
   NdbDictionary::Table tab(g_opt.m_tname);
   if (g_dic->getTable(g_opt.m_tname) != 0)
-    CHK(g_dic->dropTable(tab) == 0);
+    CHK(g_dic->dropTable(g_opt.m_tname) == 0);
   return 0;
 }
 

--- 1.19/storage/ndb/include/ndbapi/NdbBlob.hpp	2006-02-05 19:00:43 +01:00
+++ 1.20/storage/ndb/include/ndbapi/NdbBlob.hpp	2006-03-09 13:26:48 +01:00
@@ -292,6 +292,7 @@
   };
   Buf theKeyBuf;
   Buf theAccessKeyBuf;
+  Buf thePackKeyBuf;
   Buf theHeadInlineBuf;
   Buf theHeadInlineCopyBuf;     // for writeTuple
   Buf thePartBuf;
@@ -328,6 +329,9 @@
   Uint32 getPartNumber(Uint64 pos);
   Uint32 getPartCount();
   Uint32 getDistKey(Uint32 part);
+  // pack / unpack
+  int packKeyValue(const NdbTableImpl* aTable, const Buf& srcBuf);
+  int unpackKeyValue(const NdbTableImpl* aTable, Buf& dstBuf);
   // getters and setters
   int getTableKeyValue(NdbOperation* anOp);
   int setTableKeyValue(NdbOperation* anOp);

--- 1.37/storage/ndb/src/ndbapi/NdbBlob.cpp	2006-02-20 11:55:47 +01:00
+++ 1.38/storage/ndb/src/ndbapi/NdbBlob.cpp	2006-03-09 13:26:48 +01:00
@@ -310,7 +310,7 @@
 void
 NdbBlob::Buf::copyfrom(const NdbBlob::Buf& src)
 {
-  assert(size == src.size);
+  size = src.size;
   memcpy(data, src.data, size);
 }
 
@@ -408,6 +408,75 @@
   return (part / theStripeSize) % theStripeSize;
 }
 
+// pack/unpack table/index key  XXX support routines, shortcuts
+
+int
+NdbBlob::packKeyValue(const NdbTableImpl* aTable, const Buf& srcBuf)
+{
+  DBUG_ENTER("NdbBlob::packKeyValue");
+  const Uint32* data = (const Uint32*)srcBuf.data;
+  unsigned pos = 0;
+  Uint32* pack_data = (Uint32*)thePackKeyBuf.data;
+  unsigned pack_pos = 0;
+  for (unsigned i = 0; i < aTable->m_columns.size(); i++) {
+    NdbColumnImpl* c = aTable->m_columns[i];
+    assert(c != NULL);
+    if (c->m_pk) {
+      unsigned len = c->m_attrSize * c->m_arraySize;
+      Uint32 pack_len;
+      bool ok = c->get_var_length(&data[pos], pack_len);
+      if (! ok) {
+        setErrorCode(NdbBlobImpl::ErrCorruptPK);
+        DBUG_RETURN(-1);
+      }
+      memcpy(&pack_data[pack_pos], &data[pos], pack_len);
+      while (pack_len % 4 != 0) {
+        char* p = (char*)&pack_data[pack_pos] + pack_len++;
+        *p = 0;
+      }
+      pos += (len + 3) / 4;
+      pack_pos += pack_len / 4;
+    }
+  }
+  assert(4 * pos == srcBuf.size);
+  assert(4 * pack_pos <= thePackKeyBuf.maxsize);
+  thePackKeyBuf.size = 4 * pack_pos;
+  DBUG_RETURN(0);
+}
+
+int
+NdbBlob::unpackKeyValue(const NdbTableImpl* aTable, Buf& dstBuf)
+{
+  DBUG_ENTER("NdbBlob::unpackKeyValue");
+  Uint32* data = (Uint32*)dstBuf.data;
+  unsigned pos = 0;
+  const Uint32* pack_data = (const Uint32*)thePackKeyBuf.data;
+  unsigned pack_pos = 0;
+  for (unsigned i = 0; i < aTable->m_columns.size(); i++) {
+    NdbColumnImpl* c = aTable->m_columns[i];
+    assert(c != NULL);
+    if (c->m_pk) {
+      unsigned len = c->m_attrSize * c->m_arraySize;
+      Uint32 pack_len;
+      bool ok = c->get_var_length(&pack_data[pack_pos], pack_len);
+      if (! ok) {
+        setErrorCode(NdbBlobImpl::ErrCorruptPK);
+        DBUG_RETURN(-1);
+      }
+      memcpy(&data[pos], &pack_data[pack_pos], pack_len);
+      while (pack_len % 4 != 0) {
+        char* p = (char*)&data[pos] + pack_len++;
+        *p = 0;
+      }
+      pos += (len + 3) / 4;
+      pack_pos += pack_len / 4;
+    }
+  }
+  assert(4 * pos == dstBuf.size);
+  assert(4 * pack_pos == thePackKeyBuf.size);
+  DBUG_RETURN(0);
+}
+
 // getters and setters
 
 int
@@ -489,12 +558,10 @@
 NdbBlob::setPartKeyValue(NdbOperation* anOp, Uint32 part)
 {
   DBUG_ENTER("NdbBlob::setPartKeyValue");
-  DBUG_PRINT("info", ("dist=%u part=%u key=", getDistKey(part), part));
-  DBUG_DUMP("info", theKeyBuf.data, 4 * theTable->m_keyLenInWords);
-  //Uint32* data = (Uint32*)theKeyBuf.data;
-  //unsigned size = theTable->m_keyLenInWords;
+  DBUG_PRINT("info", ("dist=%u part=%u packkey=", getDistKey(part), part));
+  DBUG_DUMP("info", thePackKeyBuf.data, 4 * thePackKeyBuf.size);
   // TODO use attr ids after compatibility with 4.1.7 not needed
-  if (anOp->equal("PK", theKeyBuf.data) == -1 ||
+  if (anOp->equal("PK", thePackKeyBuf.data) == -1 ||
       anOp->equal("DIST", getDistKey(part)) == -1 ||
       anOp->equal("PART", part) == -1) {
     setErrorCode(anOp);
@@ -1242,21 +1309,27 @@
   if (isKeyOp()) {
     if (isTableOp()) {
       // get table key
-      Uint32* data = (Uint32*)theKeyBuf.data;
-      unsigned size = theTable->m_keyLenInWords;
+      Uint32* data = (Uint32*)thePackKeyBuf.data;
+      Uint32 size = theTable->m_keyLenInWords; // in-out
       if (theNdbOp->getKeyFromTCREQ(data, size) == -1) {
         setErrorCode(NdbBlobImpl::ErrUsage);
         DBUG_RETURN(-1);
       }
+      thePackKeyBuf.size = 4 * size;
+      if (unpackKeyValue(theTable, theKeyBuf) == -1)
+        DBUG_RETURN(-1);
     }
     if (isIndexOp()) {
       // get index key
-      Uint32* data = (Uint32*)theAccessKeyBuf.data;
-      unsigned size = theAccessTable->m_keyLenInWords;
+      Uint32* data = (Uint32*)thePackKeyBuf.data;
+      Uint32 size = theAccessTable->m_keyLenInWords; // in-out
       if (theNdbOp->getKeyFromTCREQ(data, size) == -1) {
         setErrorCode(NdbBlobImpl::ErrUsage);
         DBUG_RETURN(-1);
       }
+      thePackKeyBuf.size = 4 * size;
+      if (unpackKeyValue(theAccessTable, theAccessKeyBuf) == -1)
+        DBUG_RETURN(-1);
     }
     if (isReadOp()) {
       // add read of head+inline in this op
@@ -1303,6 +1376,7 @@
   theEventOp = anOp;
   theBlobEventOp = aBlobOp;
   theTable = anOp->m_eventImpl->m_tableImpl;
+  theAccessTable = theTable;
   theColumn = aColumn;
   // prepare blob column and table
   if (prepareColumn() == -1)
@@ -1321,7 +1395,7 @@
   if (theBlobEventOp != NULL) {
     if ((theBlobEventPkRecAttr =
            theBlobEventOp->getValue(theBlobTable->getColumn((Uint32)0),
-                                    theKeyBuf.data, version)) == NULL ||
+                                    thePackKeyBuf.data, version)) == NULL ||
         (theBlobEventDistRecAttr =
            theBlobEventOp->getValue(theBlobTable->getColumn((Uint32)1),
                                     (char*)0, version)) == NULL ||
@@ -1380,6 +1454,7 @@
   }
   // these buffers are always used
   theKeyBuf.alloc(theTable->m_keyLenInWords << 2);
+  thePackKeyBuf.alloc(max(theTable->m_keyLenInWords,
theAccessTable->m_keyLenInWords) << 2);
   theHeadInlineBuf.alloc(sizeof(Head) + theInlineSize);
   theHead = (Head*)theHeadInlineBuf.data;
   theInlineData = theHeadInlineBuf.data + sizeof(Head);
@@ -1464,7 +1539,7 @@
         if (tOp == NULL ||
             tOp->readTuple() == -1 ||
             setAccessKeyValue(tOp) == -1 ||
-            tOp->getValue(pkAttrId, theKeyBuf.data) == NULL) {
+            tOp->getValue(pkAttrId, thePackKeyBuf.data) == NULL) {
           setErrorCode(tOp);
           DBUG_RETURN(-1);
         }
@@ -1553,10 +1628,12 @@
   assert(isKeyOp());
   if (isIndexOp()) {
     NdbBlob* tFirstBlob = theNdbOp->theBlobList;
-    if (this != tFirstBlob) {
+    if (this == tFirstBlob) {
+      packKeyValue(theTable, theKeyBuf);
+    } else {
       // copy key from first blob
-      assert(theKeyBuf.size == tFirstBlob->theKeyBuf.size);
-      memcpy(theKeyBuf.data, tFirstBlob->theKeyBuf.data,
tFirstBlob->theKeyBuf.size);
+      theKeyBuf.copyfrom(tFirstBlob->theKeyBuf);
+      thePackKeyBuf.copyfrom(tFirstBlob->thePackKeyBuf);
     }
   }
   if (isReadOp()) {
@@ -1710,12 +1787,16 @@
     DBUG_RETURN(-1);
   assert(isScanOp());
   // get primary key
-  { Uint32* data = (Uint32*)theKeyBuf.data;
-    unsigned size = theTable->m_keyLenInWords;
-    if (((NdbScanOperation*)theNdbOp)->getKeyFromKEYINFO20(data, size) == -1) {
+  { NdbScanOperation* tScanOp = (NdbScanOperation*)theNdbOp;
+    Uint32* data = (Uint32*)thePackKeyBuf.data;
+    unsigned size = theTable->m_keyLenInWords; // in-out
+    if (tScanOp->getKeyFromKEYINFO20(data, size) == -1) {
       setErrorCode(NdbBlobImpl::ErrUsage);
       DBUG_RETURN(-1);
     }
+    thePackKeyBuf.size = 4 * size;
+    if (unpackKeyValue(theTable, theKeyBuf) == -1)
+      DBUG_RETURN(-1);
   }
   getHeadFromRecAttr();
   if (setPos(0) == -1)

--- 1.52/storage/ndb/src/ndbapi/ndberror.c	2006-03-06 11:13:55 +01:00
+++ 1.53/storage/ndb/src/ndbapi/ndberror.c	2006-03-09 13:26:48 +01:00
@@ -599,7 +599,8 @@
   { 4336, DMEC, AE, "Auto-increment value set below current value" },
   { 4271, DMEC, AE, "Invalid index object, not retrieved via getIndex()" },
   { 4272, DMEC, AE, "Table definition has undefined column" },
-  { 4273, DMEC, IE, "No blob table in dict cache" }
+  { 4273, DMEC, IE, "No blob table in dict cache" },
+  { 4274, DMEC, IE, "Corrupted main table PK in blob operation" }
 };
 
 static

--- 1.2/storage/ndb/src/ndbapi/NdbBlobImpl.hpp	2005-04-08 02:44:10 +02:00
+++ 1.3/storage/ndb/src/ndbapi/NdbBlobImpl.hpp	2006-03-09 13:26:48 +01:00
@@ -34,6 +34,8 @@
   STATIC_CONST( ErrAbort = 4268 );
   // "Unknown blob error"
   STATIC_CONST( ErrUnknown = 4269 );
+  // "Corrupted main table PK in blob operation"
+  STATIC_CONST( ErrCorruptPK = 4274 );
 };
 
 #endif