ndboperationsearch.cpp

mysql-5.0.22.tar.gz源码包

  DBUG_RETURN(-1);

 equal_error2:
  setErrorCodeAbort(4206);
  DBUG_RETURN(-1);

 equal_error3:
  setErrorCodeAbort(4209);
  DBUG_RETURN(-1);
}

/******************************************************************************
 * int insertKEYINFO(const char* aValue, aStartPosition, 
 *                   anAttrSizeInWords, Uint32 anAttrBitsInLastWord);
 *
 * Return Value:   Return 0 : insertKEYINFO was succesful.
 *                 Return -1: In all other case.   
 * Parameters:     aValue: the data to insert into KEYINFO.
 *    		   aStartPosition : Start position for Tuplekey in 
 *                                  KEYINFO (TCKEYREQ).
 *                 aKeyLenInByte : Length of tuplekey or part of tuplekey
 *                 anAttrBitsInLastWord : Nr of bits in last word. 
 * Remark:         Puts the the data into either TCKEYREQ signal 
 *                 or KEYINFO signal.
 *****************************************************************************/
int
NdbOperation::insertKEYINFO(const char* aValue,
			    register Uint32 aStartPosition,
			    register Uint32 anAttrSizeInWords)
{
  NdbApiSignal* tSignal;
  NdbApiSignal* tCurrentKEYINFO;
  //register NdbApiSignal* tTCREQ = theTCREQ;
  register Uint32 tAttrPos;
  Uint32 tPosition;
  Uint32 tEndPos;
  Uint32 tPos;
  Uint32 signalCounter;
  Uint32 tData;

/*****************************************************************************
 *	Calculate the end position of the attribute in the key information.  *
 *	Since the first attribute starts at position one we need to subtract *
 *	one to get the correct end position.				     *
 *	We must also remember the last word with only partial information.   *
 *****************************************************************************/
  tEndPos = aStartPosition + anAttrSizeInWords - 1;

  if ((tEndPos < 9)) {
    register Uint32 tkeyData = *(Uint32*)aValue;
    //TcKeyReq* tcKeyReq = CAST_PTR(TcKeyReq, tTCREQ->getDataPtrSend());
    register Uint32* tDataPtr = (Uint32*)aValue;
    tAttrPos = 1;
    register Uint32* tkeyDataPtr = theKEYINFOptr + aStartPosition - 1;
    // (Uint32*)&tcKeyReq->keyInfo[aStartPosition - 1];
    do {
      tDataPtr++;
      *tkeyDataPtr = tkeyData;
      if (tAttrPos < anAttrSizeInWords) {
        ;
      } else {
        return 0;
      }//if
      tkeyData = *tDataPtr;
      tkeyDataPtr++;
      tAttrPos++;
    } while (1);
    return 0;
  }//if
/*****************************************************************************
 *	Allocate all the KEYINFO signals needed for this key before starting *
 *	to fill the signals with data. This simplifies error handling and    *
 *      avoids duplication of code.					     *
 *****************************************************************************/
  tAttrPos = 0;
  signalCounter = 1;
  while(tEndPos > theTotalNrOfKeyWordInSignal)
  {
    tSignal = theNdb->getSignal();
    if (tSignal == NULL)
    {
      setErrorCodeAbort(4000);
      return -1;
    }
    if (tSignal->setSignal(m_keyInfoGSN) == -1)
    {
      setErrorCodeAbort(4001);
      return -1;
    }
    if (theTCREQ->next() != NULL)
       theLastKEYINFO->next(tSignal);
    else
      theTCREQ->next(tSignal);

    theLastKEYINFO = tSignal;
    theLastKEYINFO->next(NULL);
    theTotalNrOfKeyWordInSignal += 20;
  }

/*****************************************************************************
 *	Change to variable tPosition for more appropriate naming of rest of  *
 *	the code. We must set up current KEYINFO already here if the last    *
 *	word is a word which is set at LastWordLabel and at the same time    *
 *	this is the first word in a KEYINFO signal.			     *
 *****************************************************************************/
  tPosition = aStartPosition;
  tCurrentKEYINFO = theTCREQ->next();
 
/*****************************************************************************
 *	Start by filling up Key information in the 8 words allocated in the  *
 *	TC[KEY/INDX]REQ signal.						     *
 *****************************************************************************/
  while (tPosition < 9)
  {
    theKEYINFOptr[tPosition-1] = * (Uint32*)(aValue + (tAttrPos << 2));
    tAttrPos++;
    if (anAttrSizeInWords == tAttrPos)
      goto LastWordLabel;
    tPosition++;
  }

/*****************************************************************************
 *	We must set up the start position of the writing of Key information  *
 *	before we start the writing of KEYINFO signals. If the start is not  *
 *	the first word of the first KEYINFO signals then we must step forward*
 *	to the proper KEYINFO signal and set the signalCounter properly.     *
 *	signalCounter is set to the actual position in the signal ( = 4 for  *
 *	first key word in KEYINFO signal.				     *
 *****************************************************************************/
  tPos = 8;
  while ((tPosition - tPos) > 20)
  {
    tCurrentKEYINFO = tCurrentKEYINFO->next();
    tPos += 20;
  }
  signalCounter = tPosition - tPos + 3;    

/*****************************************************************************
 *	The loop that actually fills in the Key information into the KEYINFO *
 *	signals. Can be optimised by writing larger chunks than 4 bytes at a *
 *	time.								     *
 *****************************************************************************/
  do
  {
    if (signalCounter > 23)
    {
      tCurrentKEYINFO = tCurrentKEYINFO->next();
      signalCounter = 4;
    }
    tCurrentKEYINFO->setData(*(Uint32*)(aValue + (tAttrPos << 2)), 
			     signalCounter);
    tAttrPos++;
    if (anAttrSizeInWords == tAttrPos)
      goto LastWordLabel;
    tPosition++;
    signalCounter++;
  } while (1);

LastWordLabel:
  return 0;
}

int
NdbOperation::getKeyFromTCREQ(Uint32* data, unsigned size)
{
  assert(m_accessTable != 0 && m_accessTable->m_keyLenInWords != 0);
  assert(m_accessTable->m_keyLenInWords == size);
  unsigned pos = 0;
  while (pos < 8 && pos < size) {
    data[pos] = theKEYINFOptr[pos];
    pos++;
  }
  NdbApiSignal* tSignal = theTCREQ->next();
  unsigned n = 0;
  while (pos < size) {
    if (n == 20) {
      tSignal = tSignal->next();
      n = 0;
    }
    data[pos++] = tSignal->getDataPtrSend()[3 + n++];
  }
  return 0;
}

int
NdbOperation::handle_distribution_key(const Uint64* value, Uint32 len)
{
  if(theDistrKeyIndicator_ == 1 || 
     (theNoOfTupKeyLeft > 0 && m_accessTable->m_noOfDistributionKeys > 1))
  {
    return 0;
  }
  
  if(m_accessTable->m_noOfDistributionKeys == 1)
  {
    setPartitionHash(value, len);
  }
  else if(theTCREQ->readSignalNumber() == GSN_TCKEYREQ)
  {
    // No support for combined distribution key and scan

    /**
     * Copy distribution key to linear memory
     */
    NdbColumnImpl* const * cols = m_accessTable->m_columns.getBase();
    Uint32 len = 0;
    Uint64 tmp[1000];

    Uint32 chunk = 8;
    Uint32* dst = (Uint32*)tmp;
    NdbApiSignal* tSignal = theTCREQ;
    Uint32* src = ((TcKeyReq*)tSignal->getDataPtrSend())->keyInfo;
    if(tSignal->readSignalNumber() == GSN_SCAN_TABREQ)
    {
      tSignal = tSignal->next();
      src = ((KeyInfo*)tSignal->getDataPtrSend())->keyData;
      chunk = KeyInfo::DataLength;
    }

    for(unsigned i = m_accessTable->m_columns.size(); i>0; cols++, i--)
    {
      if (!(* cols)->getPrimaryKey())
	continue;
      
      NdbColumnImpl* tAttrInfo = * cols;
      Uint32 sizeInBytes = tAttrInfo->m_attrSize * tAttrInfo->m_arraySize;
      Uint32 currLen = (sizeInBytes + 3) >> 2;
      if (tAttrInfo->getDistributionKey())
      {
	while (currLen >= chunk)
	{
	  memcpy(dst, src, 4*chunk);
	  dst += chunk;
	  tSignal = tSignal->next();
	  src = ((KeyInfo*)tSignal->getDataPtrSend())->keyData;
	  currLen -= chunk;
	  chunk = KeyInfo::DataLength;
	}

	memcpy(dst, src, 4*currLen);
	dst += currLen;
	src += currLen;
	chunk -= currLen;
      }
      else
      {
	while (currLen >= chunk)
	{
	  tSignal = tSignal->next();
	  src = ((KeyInfo*)tSignal->getDataPtrSend())->keyData;
	  currLen -= chunk;
	  chunk = KeyInfo::DataLength;
	}
	
	src += currLen;
	chunk -= currLen;
      }
    }
    setPartitionHash(tmp, dst - (Uint32*)tmp);
  }
  return 0;
}

void
NdbOperation::setPartitionHash(Uint32 value)
{
  union {
    Uint32 tmp32;
    Uint64 tmp64;
  };

  tmp32 = value;
  setPartitionHash(&tmp64, 1);
}

void
NdbOperation::setPartitionHash(const Uint64* value, Uint32 len)
{
  Uint32 buf[4];
  md5_hash(buf, value, len);
  setPartitionId(buf[1]);
}

void
NdbOperation::setPartitionId(Uint32 value)
{
  theDistributionKey = value;
  theDistrKeyIndicator_ = 1;
}

Uint32
NdbOperation::getPartitionId() const 
{
  return theDistributionKey;
}