Commit 5fc1fd87 authored by unknown's avatar unknown
Browse files

Merge pnousiainen@bk-internal.mysql.com:/home/bk/mysql-5.1-new 

into  mysql.com:/space/pekka/ndb/version/my51
parents 73238e55 1c6d4a8e

storage/ndb/include/ndbapi/NdbBlob.hpp

+4 −0
Original line number Diff line number Diff line
@@ -292,6 +292,7 @@ private: 
  };

  Buf theKeyBuf;

  Buf theAccessKeyBuf;

  Buf thePackKeyBuf;

  Buf theHeadInlineBuf;

  Buf theHeadInlineCopyBuf;     // for writeTuple

  Buf thePartBuf;
@@ -328,6 +329,9 @@ private: 
  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);

storage/ndb/include/ndbapi/NdbOperation.hpp

+1 −1
Original line number Diff line number Diff line
@@ -881,7 +881,7 @@ protected: 
  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.

storage/ndb/include/ndbapi/NdbScanOperation.hpp

+1 −1
Original line number Diff line number Diff line
@@ -240,7 +240,7 @@ protected: 
  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;

storage/ndb/src/ndbapi/NdbBlob.cpp

+99 −18
Original line number Diff line number Diff line
@@ -310,7 +310,7 @@ NdbBlob::Buf::alloc(unsigned n) 
void

NdbBlob::Buf::copyfrom(const NdbBlob::Buf& src)

{

  assert(size == src.size);

  size = src.size;

  memcpy(data, src.data, size);

}
@@ -408,6 +408,75 @@ NdbBlob::getDistKey(Uint32 part) 
  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 @@ int 
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 @@ NdbBlob::atPrepare(NdbTransaction* aCon, NdbOperation* anOp, const NdbColumnImpl 
  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 @@ NdbBlob::atPrepare(NdbEventOperationImpl* anOp, NdbEventOperationImpl* aBlobOp, 
  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 @@ NdbBlob::atPrepare(NdbEventOperationImpl* anOp, NdbEventOperationImpl* aBlobOp, 
  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 @@ NdbBlob::prepareColumn() 
  }

  // 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 @@ NdbBlob::preExecute(NdbTransaction::ExecType anExecType, bool& batch) 
        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 @@ NdbBlob::postExecute(NdbTransaction::ExecType anExecType) 
  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 @@ NdbBlob::atNextResult() 
    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)

storage/ndb/src/ndbapi/NdbBlobImpl.hpp

+2 −0
Original line number Diff line number Diff line
@@ -34,6 +34,8 @@ public: 
  STATIC_CONST( ErrAbort = 4268 );

  // "Unknown blob error"

  STATIC_CONST( ErrUnknown = 4269 );

  // "Corrupted main table PK in blob operation"

  STATIC_CONST( ErrCorruptPK = 4274 );

};



#endif