dbtuproutines.cpp

mysql-5.0.22.tar.gz源码包

/* Copyright (C) 2003 MySQL AB

   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; either version 2 of the License, or
   (at your option) any later version.

   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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */


#define DBTUP_C
#include "Dbtup.hpp"
#include <RefConvert.hpp>
#include <ndb_limits.h>
#include <pc.hpp>
#include <AttributeDescriptor.hpp>
#include "AttributeOffset.hpp"
#include <AttributeHeader.hpp>

#define ljam() { jamLine(3000 + __LINE__); }
#define ljamEntry() { jamEntryLine(3000 + __LINE__); }

void
Dbtup::setUpQueryRoutines(Tablerec* const regTabPtr)
{
  Uint32 startDescriptor = regTabPtr->tabDescriptor;
  ndbrequire((startDescriptor + (regTabPtr->noOfAttr << ZAD_LOG_SIZE)) <= cnoOfTabDescrRec);
  for (Uint32 i = 0; i < regTabPtr->noOfAttr; i++) {
    Uint32 attrDescriptorStart = startDescriptor + (i << ZAD_LOG_SIZE);
    Uint32 attrDescriptor = tableDescriptor[attrDescriptorStart].tabDescr;
    Uint32 attrOffset = tableDescriptor[attrDescriptorStart + 1].tabDescr;
    if (!AttributeDescriptor::getDynamic(attrDescriptor)) {
      if ((AttributeDescriptor::getArrayType(attrDescriptor) == ZNON_ARRAY) ||
          (AttributeDescriptor::getArrayType(attrDescriptor) == ZFIXED_ARRAY)) {
        if (!AttributeDescriptor::getNullable(attrDescriptor)) {
	  if (AttributeDescriptor::getSize(attrDescriptor) == 0){
	    ljam();
            regTabPtr->readFunctionArray[i] = &Dbtup::readBitsNotNULL;
            regTabPtr->updateFunctionArray[i] = &Dbtup::updateBitsNotNULL;
	  } else if (AttributeDescriptor::getSizeInWords(attrDescriptor) == 1){
            ljam();
            regTabPtr->readFunctionArray[i] = &Dbtup::readFixedSizeTHOneWordNotNULL;
            regTabPtr->updateFunctionArray[i] = &Dbtup::updateFixedSizeTHOneWordNotNULL;
          } else if (AttributeDescriptor::getSizeInWords(attrDescriptor) == 2) {
            ljam();
            regTabPtr->readFunctionArray[i] = &Dbtup::readFixedSizeTHTwoWordNotNULL;
            regTabPtr->updateFunctionArray[i] = &Dbtup::updateFixedSizeTHTwoWordNotNULL;
          } else if (AttributeDescriptor::getSizeInWords(attrDescriptor) > 2) {
            ljam();
            regTabPtr->readFunctionArray[i] = &Dbtup::readFixedSizeTHManyWordNotNULL;
            regTabPtr->updateFunctionArray[i] = &Dbtup::updateFixedSizeTHManyWordNotNULL;
          } else {
            ndbrequire(false);
          }//if
          // replace functions for char attribute
          if (AttributeOffset::getCharsetFlag(attrOffset)) {
            ljam();
            regTabPtr->readFunctionArray[i] = &Dbtup::readFixedSizeTHManyWordNotNULL;
            regTabPtr->updateFunctionArray[i] = &Dbtup::updateFixedSizeTHManyWordNotNULL;
          }
        } else {
	  if (AttributeDescriptor::getSize(attrDescriptor) == 0){
            ljam();
            regTabPtr->readFunctionArray[i] = &Dbtup::readBitsNULLable;
            regTabPtr->updateFunctionArray[i] = &Dbtup::updateBitsNULLable;
	  } else if (AttributeDescriptor::getSizeInWords(attrDescriptor) == 1){
            ljam();
            regTabPtr->readFunctionArray[i] = &Dbtup::readFixedSizeTHOneWordNULLable;
            regTabPtr->updateFunctionArray[i] = &Dbtup::updateFixedSizeTHManyWordNULLable;
          } else if (AttributeDescriptor::getSizeInWords(attrDescriptor) == 2) {
            ljam();
            regTabPtr->readFunctionArray[i] = &Dbtup::readFixedSizeTHTwoWordNULLable;
            regTabPtr->updateFunctionArray[i] = &Dbtup::updateFixedSizeTHManyWordNULLable;
          } else if (AttributeDescriptor::getSizeInWords(attrDescriptor) > 2) {
            ljam();
            regTabPtr->readFunctionArray[i] = &Dbtup::readFixedSizeTHManyWordNULLable;
            regTabPtr->updateFunctionArray[i] = &Dbtup::updateFixedSizeTHManyWordNULLable;
          } else {
            ljam();
            regTabPtr->readFunctionArray[i] = &Dbtup::readFixedSizeTHZeroWordNULLable;
            regTabPtr->updateFunctionArray[i] = &Dbtup::updateFixedSizeTHManyWordNULLable;
          }//if
          // replace functions for char attribute
          if (AttributeOffset::getCharsetFlag(attrOffset)) {
            ljam();
            regTabPtr->readFunctionArray[i] = &Dbtup::readFixedSizeTHManyWordNULLable;
            regTabPtr->updateFunctionArray[i] = &Dbtup::updateFixedSizeTHManyWordNULLable;
          }
        }//if
      } else if (AttributeDescriptor::getArrayType(attrDescriptor) == ZVAR_ARRAY) {
        if (!AttributeDescriptor::getNullable(attrDescriptor)) {
          if (AttributeDescriptor::getArraySize(attrDescriptor) == 0) {
            ljam();
            regTabPtr->readFunctionArray[i] = &Dbtup::readVarSizeUnlimitedNotNULL;
            regTabPtr->updateFunctionArray[i] = &Dbtup::updateVarSizeUnlimitedNotNULL;
          } else if (AttributeDescriptor::getArraySize(attrDescriptor) > ZMAX_SMALL_VAR_ARRAY) {
            ljam();
            regTabPtr->readFunctionArray[i] = &Dbtup::readBigVarSizeNotNULL;
            regTabPtr->updateFunctionArray[i] = &Dbtup::updateBigVarSizeNotNULL;
          } else {
            ljam();
            regTabPtr->readFunctionArray[i] = &Dbtup::readSmallVarSizeNotNULL;
            regTabPtr->updateFunctionArray[i] = &Dbtup::updateSmallVarSizeNotNULL;
          }//if
        } else {
          if (AttributeDescriptor::getArraySize(attrDescriptor) == 0) {
            ljam();
            regTabPtr->readFunctionArray[i] = &Dbtup::readVarSizeUnlimitedNULLable;
            regTabPtr->updateFunctionArray[i] = &Dbtup::updateVarSizeUnlimitedNULLable;
          } else if (AttributeDescriptor::getArraySize(attrDescriptor) > ZMAX_SMALL_VAR_ARRAY) {
            ljam();
            regTabPtr->readFunctionArray[i] = &Dbtup::readBigVarSizeNULLable;
            regTabPtr->updateFunctionArray[i] = &Dbtup::updateBigVarSizeNULLable;
          } else {
            ljam();
            regTabPtr->readFunctionArray[i] = &Dbtup::readSmallVarSizeNULLable;
            regTabPtr->updateFunctionArray[i] = &Dbtup::updateSmallVarSizeNULLable;
          }//if
        }//if
      } else {
        ndbrequire(false);
      }//if
    } else {
      if ((AttributeDescriptor::getArrayType(attrDescriptor) == ZNON_ARRAY) ||
          (AttributeDescriptor::getArrayType(attrDescriptor) == ZFIXED_ARRAY)) {
        ljam();
        regTabPtr->readFunctionArray[i] = &Dbtup::readDynFixedSize;
        regTabPtr->updateFunctionArray[i] = &Dbtup::updateDynFixedSize;
      } else if (AttributeDescriptor::getType(attrDescriptor) == ZVAR_ARRAY) {
        if (AttributeDescriptor::getArraySize(attrDescriptor) == 0) {
          ljam();
          regTabPtr->readFunctionArray[i] = &Dbtup::readDynVarSizeUnlimited;
          regTabPtr->updateFunctionArray[i] = &Dbtup::updateDynVarSizeUnlimited;
        } else if (AttributeDescriptor::getArraySize(attrDescriptor) > ZMAX_SMALL_VAR_ARRAY) {
          ljam();
          regTabPtr->readFunctionArray[i] = &Dbtup::readDynBigVarSize;
          regTabPtr->updateFunctionArray[i] = &Dbtup::updateDynBigVarSize;
        } else {
          ljam();
          regTabPtr->readFunctionArray[i] = &Dbtup::readDynSmallVarSize;
          regTabPtr->updateFunctionArray[i] = &Dbtup::updateDynSmallVarSize;
        }//if
      } else {
        ndbrequire(false);
      }//if
    }//if
  }//for
}//Dbtup::setUpQueryRoutines()

/* ---------------------------------------------------------------- */
/*       THIS ROUTINE IS USED TO READ A NUMBER OF ATTRIBUTES IN THE */
/*       DATABASE AND PLACE THE RESULT IN ATTRINFO RECORDS.         */
//
// In addition to the parameters used in the call it also relies on the
// following variables set-up properly.
//
// operPtr.p      Operation record pointer
// fragptr.p      Fragment record pointer
// tabptr.p       Table record pointer
/* ---------------------------------------------------------------- */
int Dbtup::readAttributes(Page* const pagePtr,
			  Uint32  tupHeadOffset,
			  const Uint32* inBuffer,
			  Uint32  inBufLen,
			  Uint32* outBuffer,
			  Uint32  maxRead,
			  bool    xfrmFlag)
{
  Tablerec* const regTabPtr =  tabptr.p;
  Uint32 numAttributes = regTabPtr->noOfAttr;
  Uint32 attrDescriptorStart = regTabPtr->tabDescriptor;
  Uint32 inBufIndex = 0;

  ndbrequire(attrDescriptorStart + (numAttributes << ZAD_LOG_SIZE) <= cnoOfTabDescrRec);

  tOutBufIndex = 0;
  tCheckOffset = regTabPtr->tupheadsize;
  tMaxRead = maxRead;
  tTupleHeader = &pagePtr->pageWord[tupHeadOffset];
  tXfrmFlag = xfrmFlag;

  ndbrequire(tupHeadOffset + tCheckOffset <= ZWORDS_ON_PAGE);
  while (inBufIndex < inBufLen) {
    Uint32 tmpAttrBufIndex = tOutBufIndex;
    AttributeHeader ahIn(inBuffer[inBufIndex]);
    inBufIndex++;
    Uint32 attributeId = ahIn.getAttributeId();
    Uint32 attrDescriptorIndex = attrDescriptorStart + (attributeId << ZAD_LOG_SIZE);
    ljam();

    AttributeHeader::init(&outBuffer[tmpAttrBufIndex], attributeId, 0);
    AttributeHeader* ahOut = (AttributeHeader*)&outBuffer[tmpAttrBufIndex];
    tOutBufIndex = tmpAttrBufIndex + 1;
    if (attributeId < numAttributes) {
      Uint32 attributeDescriptor = tableDescriptor[attrDescriptorIndex].tabDescr;
      Uint32 attributeOffset = tableDescriptor[attrDescriptorIndex + 1].tabDescr;
      ReadFunction f = regTabPtr->readFunctionArray[attributeId];
      if ((this->*f)(outBuffer,
                     ahOut,
                     attributeDescriptor,
                     attributeOffset)) {
        continue;
      } else {
        return -1;
      }//if
    } else if(attributeId & AttributeHeader::PSUEDO){
      Uint32 sz = read_psuedo(attributeId, 
			      outBuffer+tmpAttrBufIndex+1);
      AttributeHeader::init(&outBuffer[tmpAttrBufIndex], attributeId, sz);
      tOutBufIndex = tmpAttrBufIndex + 1 + sz;
    } else {
      terrorCode = ZATTRIBUTE_ID_ERROR;
      return -1;
    }//if
  }//while
  return tOutBufIndex;
}//Dbtup::readAttributes()

#if 0
int Dbtup::readAttributesWithoutHeader(Page* const pagePtr,
                                       Uint32  tupHeadOffset,
                                       Uint32* inBuffer,
                                       Uint32  inBufLen,
                                       Uint32* outBuffer,
                                       Uint32* attrBuffer,
                                       Uint32  maxRead)
{
  Tablerec* const regTabPtr =  tabptr.p;
  Uint32 numAttributes = regTabPtr->noOfAttr;
  Uint32 attrDescriptorStart = regTabPtr->tabDescriptor;
  Uint32 inBufIndex = 0;
  Uint32 attrBufIndex = 0;

  ndbrequire(attrDescriptorStart + (numAttributes << ZAD_LOG_SIZE) <= cnoOfTabDescrRec);

  tOutBufIndex = 0;
  tCheckOffset = regTabPtr->tupheadsize;
  tMaxRead = maxRead;
  tTupleHeader = &pagePtr->pageWord[tupHeadOffset];

  ndbrequire(tupHeadOffset + tCheckOffset <= ZWORDS_ON_PAGE);
  while (inBufIndex < inBufLen) {
    AttributeHeader ahIn(inBuffer[inBufIndex]);
    inBufIndex++;
    Uint32 attributeId = ahIn.getAttributeId();
    Uint32 attrDescriptorIndex = attrDescriptorStart + (attributeId << ZAD_LOG_SIZE);
    ljam();

    AttributeHeader::init(&attrBuffer[attrBufIndex], attributeId, 0);
    AttributeHeader* ahOut = (AttributeHeader*)&attrBuffer[attrBufIndex];
    attrBufIndex++;
    if (attributeId < numAttributes) {
      Uint32 attributeDescriptor = tableDescriptor[attrDescriptorIndex].tabDescr;
      Uint32 attributeOffset = tableDescriptor[attrDescriptorIndex + 1].tabDescr;
      ReadFunction f = regTabPtr->readFunctionArray[attributeId];
      if ((this->*f)(outBuffer,
                     ahOut,
                     attributeDescriptor,
                     attributeOffset)) {
        continue;
      } else {
        return -1;
      }//if
    } else {
      terrorCode = ZATTRIBUTE_ID_ERROR;
      return -1;
    }//if
  }//while
  ndbrequire(attrBufIndex == inBufLen);
  return tOutBufIndex;
}//Dbtup::readAttributes()
#endif

bool
Dbtup::readFixedSizeTHOneWordNotNULL(Uint32* outBuffer,
                                     AttributeHeader* ahOut,
                                     Uint32  attrDescriptor,
                                     Uint32  attrDes2)
{
  Uint32 indexBuf = tOutBufIndex;
  Uint32 readOffset = AttributeOffset::getOffset(attrDes2);
  Uint32 const wordRead = tTupleHeader[readOffset];
  Uint32 newIndexBuf = indexBuf + 1;
  Uint32 maxRead = tMaxRead;

  ndbrequire(readOffset < tCheckOffset);
  if (newIndexBuf <= maxRead) {
    ljam();
    outBuffer[indexBuf] = wordRead;
    ahOut->setDataSize(1);
    tOutBufIndex = newIndexBuf;
    return true;
  } else {
    ljam();
    terrorCode = ZTRY_TO_READ_TOO_MUCH_ERROR;
    return false;
  }//if
}//Dbtup::readFixedSizeTHOneWordNotNULL()

bool
Dbtup::readFixedSizeTHTwoWordNotNULL(Uint32* outBuffer,
                                     AttributeHeader* ahOut,
                                     Uint32  attrDescriptor,
                                     Uint32  attrDes2)
{
  Uint32 indexBuf = tOutBufIndex;
  Uint32 readOffset = AttributeOffset::getOffset(attrDes2);
  Uint32 const wordReadFirst = tTupleHeader[readOffset];
  Uint32 const wordReadSecond = tTupleHeader[readOffset + 1];
  Uint32 newIndexBuf = indexBuf + 2;
  Uint32 maxRead = tMaxRead;

  ndbrequire(readOffset + 1 < tCheckOffset);
  if (newIndexBuf <= maxRead) {
    ljam();
    ahOut->setDataSize(2);
    outBuffer[indexBuf] = wordReadFirst;
    outBuffer[indexBuf + 1] = wordReadSecond;
    tOutBufIndex = newIndexBuf;
    return true;
  } else {
    ljam();
    terrorCode = ZTRY_TO_READ_TOO_MUCH_ERROR;
    return false;
  }//if
}//Dbtup::readFixedSizeTHTwoWordNotNULL()

bool
Dbtup::readFixedSizeTHManyWordNotNULL(Uint32* outBuffer,
                                      AttributeHeader* ahOut,
                                      Uint32  attrDescriptor,
                                      Uint32  attrDes2)
{
  Uint32 indexBuf = tOutBufIndex;
  Uint32 charsetFlag = AttributeOffset::getCharsetFlag(attrDes2);
  Uint32 readOffset = AttributeOffset::getOffset(attrDes2);
  Uint32 attrNoOfWords = AttributeDescriptor::getSizeInWords(attrDescriptor);
  Uint32 maxRead = tMaxRead;

  ndbrequire((readOffset + attrNoOfWords - 1) < tCheckOffset);
  if (! charsetFlag || ! tXfrmFlag) {
    Uint32 newIndexBuf = indexBuf + attrNoOfWords;
    if (newIndexBuf <= maxRead) {
      ljam();
      ahOut->setDataSize(attrNoOfWords);
      MEMCOPY_NO_WORDS(&outBuffer[indexBuf],
                       &tTupleHeader[readOffset],
                       attrNoOfWords);
      tOutBufIndex = newIndexBuf;
      return true;
    } else {
      ljam();
      terrorCode = ZTRY_TO_READ_TOO_MUCH_ERROR;
    }//if
  } else {
    ljam();
    Tablerec* regTabPtr = tabptr.p;
    Uint32 srcBytes = AttributeDescriptor::getSizeInBytes(attrDescriptor);
    uchar* dstPtr = (uchar*)&outBuffer[indexBuf];
    const uchar* srcPtr = (uchar*)&tTupleHeader[readOffset];
    Uint32 i = AttributeOffset::getCharsetPos(attrDes2);
    ndbrequire(i < regTabPtr->noOfCharsets);
    CHARSET_INFO* cs = regTabPtr->charsetArray[i];
    Uint32 typeId = AttributeDescriptor::getType(attrDescriptor);
    Uint32 lb, len;
    bool ok = NdbSqlUtil::get_var_length(typeId, srcPtr, srcBytes, lb, len);
    if (ok) {
      Uint32 xmul = cs->strxfrm_multiply;
      if (xmul == 0)
        xmul = 1;
      // see comment in DbtcMain.cpp
      Uint32 dstLen = xmul * (srcBytes - lb);
      Uint32 maxIndexBuf = indexBuf + (dstLen >> 2);
      if (maxIndexBuf <= maxRead) {
        ljam();
        int n = NdbSqlUtil::strnxfrm_bug7284(cs, dstPtr, dstLen, srcPtr + lb, len);
        ndbrequire(n != -1);
        while ((n & 3) != 0) {
          dstPtr[n++] = 0;
        }
        Uint32 dstWords = (n >> 2);
        ahOut->setDataSize(dstWords);
        Uint32 newIndexBuf = indexBuf + dstWords;
        ndbrequire(newIndexBuf <= maxRead);
        tOutBufIndex = newIndexBuf;
        return true;
      } else {
        ljam();
        terrorCode = ZTRY_TO_READ_TOO_MUCH_ERROR;
      }
    } else {
      ljam();
      terrorCode = ZTUPLE_CORRUPTED_ERROR;
    }
  }
  return false;
}//Dbtup::readFixedSizeTHManyWordNotNULL()