NdbOperationSearch.cpp

/*
   Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
*/


/******************************************************************************
Name:          NdbOperationSearch.C
Include:
Link:
Author:        UABMNST Mona Natterkvist UAB/B/SD                         
Date:          970829
Version:       0.1
Description:   Interface between TIS and NDB
Documentation:
Adjust:  971022  UABMNST   First version.
         971206  UABRONM
 *****************************************************************************/
#include "API.hpp"


#include <AttributeHeader.hpp>
#include <signaldata/TcKeyReq.hpp>
#include <signaldata/KeyInfo.hpp>

/******************************************************************************
CondIdType equal(const char* anAttrName, char* aValue, Uint32 aVarKeylen);

Return Value    Return 0 : Equal was successful.
                Return -1: In all other case. 
Parameters:     anAttrName : Attribute name for search condition..
                aValue : Referense to the search value.
                aVariableKeylen : The length of key in bytes  
Remark:         Defines search condition with equality anAttrName.
******************************************************************************/
int
NdbOperation::equal_impl(const NdbColumnImpl* tAttrInfo, 
                         const char* aValuePassed)
{
  DBUG_ENTER("NdbOperation::equal_impl");
  DBUG_PRINT("enter", ("col: %s  op: %d  val: 0x%lx",
                       (tAttrInfo == NULL) ? "NULL" :
                       tAttrInfo->m_name.c_str(), 
                       theOperationType,
                       (long) aValuePassed));
  
  const char* aValue = aValuePassed;
  const Uint32 MaxKeyLenInLongWords= (NDB_MAX_KEY_SIZE + 7)/8;
  Uint64 tempData[ MaxKeyLenInLongWords ];

  if ((theStatus == OperationDefined) &&
      (aValue != NULL) &&
      (tAttrInfo != NULL )) {
/******************************************************************************
 *      Start by checking that the attribute is a tuple key. 
 *      This value is also the word order in the tuple key of this 
 *      tuple key attribute. 
 *      Then check that this tuple key has not already been defined. 
 *      Finally check if all tuple key attributes have been defined. If
 *      this is true then set Operation state to tuple key defined.
 *****************************************************************************/

    /*
     * For each call theTupleKeyDefined stores 3 items:
     *
     * [0] = m_column_no (external column id)
     * [1] = 1-based index of first word of accumulating keyinfo
     * [2] = number of words of keyinfo
     *
     * This is used to re-order keyinfo if not in m_attrId order.
     *
     * Note: No point to "clean up" this code.  The upcoming
     * record-based ndb api makes it obsolete.
     */

    Uint32 tAttrId = tAttrInfo->m_column_no; // not m_attrId;
    Uint32 i = 0;
    if (tAttrInfo->m_pk) {
      Uint32 tKeyDefined = theTupleKeyDefined[0][2];
      Uint32 tKeyAttrId = theTupleKeyDefined[0][0];
      do {
        if (tKeyDefined == false) {
          goto keyEntryFound;
        } else {
          if (tKeyAttrId != tAttrId) {
            /******************************************************************
             * We read the key defined variable in advance. 
             * It could potentially read outside its area when 
             * i = MAXNROFTUPLEKEY - 1, 
             * it is not a problem as long as the variable 
             * theTupleKeyDefined is defined
             * in the middle of the object. 
             * Reading wrong data and not using it causes no problems.
             *****************************************************************/
            i++;
            tKeyAttrId = theTupleKeyDefined[i][0];
            tKeyDefined = theTupleKeyDefined[i][2];
            continue;
          } else {
            goto equal_error2;
          }//if
        }//if
      } while (i < NDB_MAX_NO_OF_ATTRIBUTES_IN_KEY);
      goto equal_error2;
    } else {
      goto equal_error1;
    }
    /*************************************************************************
     *  Now it is time to retrieve the tuple key data from the pointer supplied
     *      by the application. 
     *      We have to retrieve the size of the attribute in words and bits.
     *************************************************************************/
  keyEntryFound:
    Uint32 sizeInBytes;
    if (! tAttrInfo->get_var_length(aValue, sizeInBytes)) {
      setErrorCodeAbort(4209);
      DBUG_RETURN(-1);
    }

    Uint32 tKeyInfoPosition =
      i == 0 ? 1 : theTupleKeyDefined[i-1][1] + theTupleKeyDefined[i-1][2];
    theTupleKeyDefined[i][0] = tAttrId;
    theTupleKeyDefined[i][1] = tKeyInfoPosition; 
    theTupleKeyDefined[i][2] = (sizeInBytes + 3) / 4;

    {
      /************************************************************************
       * Check if the pointer of the value passed is aligned on a 4 byte 
       * boundary. If so only assign the pointer to the internal variable 
       * aValue. If it is not aligned then we start by copying the value to 
       * tempData and use this as aValue instead.
       ***********************************************************************/
      const bool tDistrKey = tAttrInfo->m_distributionKey;
      const int attributeSize = sizeInBytes;
      const int slack = sizeInBytes & 3;
      const int align = Uint32(UintPtr(aValue)) & 7;

      if (((align & 3) != 0) || (slack != 0) || (tDistrKey && (align != 0)))
      {
        ((Uint32*)tempData)[attributeSize >> 2] = 0;
        memcpy(&tempData[0], aValue, attributeSize);
        aValue = (char*)&tempData[0];
      }//if
    }

    Uint32 totalSizeInWords = (sizeInBytes + 3)/4; // Inc. bits in last word
    theTupKeyLen += totalSizeInWords;
#if 0
    else {
      /************************************************************************
       * The attribute is a variable array. We need to use the length parameter
       * to know the size of this attribute in the key information and 
       * variable area. A key is however not allowed to be larger than 4 
       * kBytes and this is checked for variable array attributes
       * used as keys.
       ************************************************************************/
      Uint32 tMaxVariableKeyLenInWord = (MAXTUPLEKEYLENOFATTERIBUTEINWORD -
                                         tKeyInfoPosition);
      tAttrSizeInBits = aVariableKeyLen << 3;
      tAttrSizeInWords = tAttrSizeInBits >> 5;
      tAttrBitsInLastWord = tAttrSizeInBits - (tAttrSizeInWords << 5);
      tAttrLenInWords = ((tAttrSizeInBits + 31) >> 5);
      if (tAttrLenInWords > tMaxVariableKeyLenInWord) {
        setErrorCodeAbort(4207);
        return -1;
      }//if
      theTupKeyLen = theTupKeyLen + tAttrLenInWords;
    }//if
#endif

    /**************************************************************************
     *  If the operation is an insert request and the attribute is stored then
     *      we also set the value in the stored part through putting the 
     *      information in the ATTRINFO signals.
     *************************************************************************/
    OperationType tOpType = theOperationType;
    if ((tOpType == InsertRequest) ||
        (tOpType == WriteRequest)) {
      Uint32 ahValue;

      if(m_accessTable == m_currentTable) {
        AttributeHeader::init(&ahValue, tAttrInfo->m_attrId, sizeInBytes);
      } else {
        assert(tOpType == WriteRequest && m_accessTable->m_index);
        // use attrId of primary table column
        int column_no_current_table = 
          m_accessTable->m_index->m_columns[tAttrId]->m_keyInfoPos;
        int attr_id_current_table =
          m_currentTable->m_columns[column_no_current_table]->m_attrId;
        AttributeHeader::init(&ahValue, attr_id_current_table, sizeInBytes);
      }
      
      insertATTRINFO( ahValue );
      insertATTRINFOloop((Uint32*)aValue, totalSizeInWords);
    }//if
    
    /**************************************************************************
     *  Store the Key information in the TCKEYREQ and KEYINFO signals. 
     *************************************************************************/
    if (insertKEYINFO(aValue, tKeyInfoPosition, totalSizeInWords) != -1) {
      /************************************************************************
       * Add one to number of tuple key attributes defined. 
       * If all have been defined then set the operation state to indicate 
       * that tuple key is defined. 
       * Thereby no more search conditions are allowed in this version.
       ***********************************************************************/
      Uint32 tNoKeysDef = theNoOfTupKeyLeft - 1;
      Uint32 tErrorLine = theErrorLine;
      unsigned char tInterpretInd = theInterpretIndicator;
      theNoOfTupKeyLeft = tNoKeysDef;
      tErrorLine++;
      theErrorLine = tErrorLine;
      
      if (tNoKeysDef == 0) {    

        // re-order keyinfo if not entered in order
        if (m_accessTable->m_noOfKeys != 1) {
          for (Uint32 i = 0; i < m_accessTable->m_noOfKeys; i++) {
            Uint32 k = theTupleKeyDefined[i][0]; // column_no
            if (m_accessTable->m_columns[k]->m_keyInfoPos != i) {
              DBUG_PRINT("info", ("key disorder at %d", i));
              reorderKEYINFO();
              break;
            }
          }
        }

        if (tOpType == UpdateRequest) {
          if (tInterpretInd == 1) {
            theStatus = GetValue;
          } else {
            theStatus = SetValue;
          }//if
          DBUG_RETURN(0);
        } else if ((tOpType == ReadRequest) || (tOpType == DeleteRequest) ||
                   (tOpType == ReadExclusive)) {
          theStatus = GetValue;
          // create blob handles automatically
          if (tOpType == DeleteRequest && m_currentTable->m_noOfBlobs != 0) {
            for (unsigned i = 0; i < m_currentTable->m_columns.size(); i++) {
              NdbColumnImpl* c = m_currentTable->m_columns[i];
              assert(c != 0);
              if (c->getBlobType()) {
                if (getBlobHandle(theNdbCon, c) == NULL)
                  DBUG_RETURN(-1);
              }
            }
          }
          DBUG_RETURN(0);
        } else if ((tOpType == InsertRequest) || (tOpType == WriteRequest)) {
          theStatus = SetValue;
          DBUG_RETURN(0);
        } else {
          setErrorCodeAbort(4005);
          DBUG_RETURN(-1);
        }//if
        DBUG_RETURN(0);
      }//if
    } else {
      DBUG_RETURN(-1);
    }//if
    DBUG_RETURN(0);
  }
  
  if (aValue == NULL) {
    // NULL value in primary key
    setErrorCodeAbort(4505);
    DBUG_RETURN(-1);
  }//if
  
  if ( tAttrInfo == NULL ) {      
    // Attribute name not found in table
    setErrorCodeAbort(4004);
    DBUG_RETURN(-1);
  }//if

  if (theStatus == GetValue || theStatus == SetValue){
    // All pk's defined
    setErrorCodeAbort(4225);
    DBUG_RETURN(-1);
  }//if

  ndbout_c("theStatus: %d", theStatus);
  
  // If we come here, set a general errorcode
  // and exit
  setErrorCodeAbort(4200);
  DBUG_RETURN(-1);

 equal_error1:
  setErrorCodeAbort(4205);
  DBUG_RETURN(-1);

 equal_error2:
  setErrorCodeAbort(4206);
  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;

/*****************************************************************************
 *      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, refToBlock(theNdbCon->m_tcRef)) == -1)
    {
      setErrorCodeAbort(4001);
      return -1;
    }
    tSignal->setLength(KeyInfo::MaxSignalLength);
    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;
}

void
NdbOperation::reorderKEYINFO()
{
  Uint32 data[ NDB_MAX_KEYSIZE_IN_WORDS ];
  Uint32 size = NDB_MAX_KEYSIZE_IN_WORDS;
  int rc = getKeyFromTCREQ(data, size);
  assert(rc == 0);
  Uint32 pos = 1;
  Uint32 k;
  for (k = 0; k < m_accessTable->m_noOfKeys; k++) {
    Uint32 i;
    for (i = 0; i < m_accessTable->m_columns.size(); i++) {
      NdbColumnImpl* col = m_accessTable->m_columns[i];
      if (col->m_pk && col->m_keyInfoPos == k) {
        Uint32 j;
        for (j = 0; j < m_accessTable->m_noOfKeys; j++) {
          if (theTupleKeyDefined[j][0] == i) {
            Uint32 off = theTupleKeyDefined[j][1] - 1;
            Uint32 len = theTupleKeyDefined[j][2];
            assert(off < NDB_MAX_KEYSIZE_IN_WORDS && 
                   off + len <= NDB_MAX_KEYSIZE_IN_WORDS);
            int ret = insertKEYINFO((char*)&data[off], pos, len);
            assert(ret == 0);
            pos += len;
            break;
          }
        }
        assert(j < m_accessTable->m_columns.size());
        break;
      }
    }
    assert(i < m_accessTable->m_columns.size());
  }
}

int
NdbOperation::getKeyFromTCREQ(Uint32* data, Uint32 & size)
{
  /* Check that we can correctly return a valid key */
  if ((size < theTupKeyLen) || (theTupKeyLen == 0))
    return -1;
  
  size = theTupKeyLen;
  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 == KeyInfo::DataLength) {
      tSignal = tSignal->next();
      n = 0;
    }
    data[pos++] = 
      tSignal->getDataPtrSend()[KeyInfo::HeaderLength + n++];
  }
  return 0;
}

void
NdbOperation::setPartitionId(Uint32 value)
{
  if (theStatus == UseNdbRecord)
  {
    /* Method not allowed for NdbRecord, use OperationOptions or 
       ScanOptions structure instead */
    setErrorCodeAbort(4515);
    return; // TODO : Consider adding int rc for error
  }

  /* We only allow setPartitionId() for :
   *   PrimaryKey ops on a UserDefined partitioned table
   *   Ordered index scans
   *   Table scans
   *
   * It is not allowed on :
   *   Primary key access to Natively partitioned tables
   *   Any unique key access
   */
  assert(((m_type == PrimaryKeyAccess) && 
          (m_currentTable->getFragmentType() ==
           NdbDictionary::Object::UserDefined)) ||
         (m_type == OrderedIndexScan) ||
         (m_type == TableScan));
  
  theDistributionKey = value;
  theDistrKeyIndicator_ = 1;
  DBUG_PRINT("info", ("NdbOperation::setPartitionId: %u",
                       theDistributionKey));
}

Uint32
NdbOperation::getPartitionId() const 
{
  DBUG_PRINT("info", ("NdbOperation::getPartitionId: %u ind=%d",
                      theDistributionKey, theDistrKeyIndicator_));
  return theDistributionKey;
}