dbtupindex.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>
#include <signaldata/TuxMaint.hpp>

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

// methods used by ordered index

void
Dbtup::tuxGetTupAddr(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32& tupAddr)
{
  ljamEntry();
  FragrecordPtr fragPtr;
  fragPtr.i = fragPtrI;
  ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
  TablerecPtr tablePtr;
  tablePtr.i = fragPtr.p->fragTableId;
  ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
  PagePtr pagePtr;
  pagePtr.i = pageId;
  ptrCheckGuard(pagePtr, cnoOfPage, page);
  Uint32 fragPageId = pagePtr.p->pageWord[ZPAGE_FRAG_PAGE_ID_POS];
  Uint32 tupheadsize = tablePtr.p->tupheadsize;
  ndbrequire(pageOffset >= ZPAGE_HEADER_SIZE);
  Uint32 offset = pageOffset - ZPAGE_HEADER_SIZE;
  ndbrequire(offset % tupheadsize == 0);
  Uint32 pageIndex = (offset / tupheadsize) << 1;
  tupAddr = (fragPageId << MAX_TUPLES_BITS) | pageIndex;
}

int
Dbtup::tuxAllocNode(Signal* signal, Uint32 fragPtrI, Uint32& pageId, Uint32& pageOffset, Uint32*& node)
{
  ljamEntry();
  FragrecordPtr fragPtr;
  fragPtr.i = fragPtrI;
  ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
  TablerecPtr tablePtr;
  tablePtr.i = fragPtr.p->fragTableId;
  ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
  PagePtr pagePtr;
  terrorCode = 0;
  if (! allocTh(fragPtr.p, tablePtr.p, NORMAL_PAGE, signal, pageOffset, pagePtr)) {
    ljam();
    ndbrequire(terrorCode != 0);
    return terrorCode;
  }
  pageId = pagePtr.i;
  Uint32 attrDescIndex = tablePtr.p->tabDescriptor + (0 << ZAD_LOG_SIZE);
  Uint32 attrDataOffset = AttributeOffset::getOffset(tableDescriptor[attrDescIndex + 1].tabDescr);
  node = &pagePtr.p->pageWord[pageOffset] + attrDataOffset;
  return 0;
}

void
Dbtup::tuxFreeNode(Signal* signal, Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32* node)
{
  ljamEntry();
  FragrecordPtr fragPtr;
  fragPtr.i = fragPtrI;
  ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
  TablerecPtr tablePtr;
  tablePtr.i = fragPtr.p->fragTableId;
  ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
  PagePtr pagePtr;
  pagePtr.i = pageId;
  ptrCheckGuard(pagePtr, cnoOfPage, page);
  Uint32 attrDescIndex = tablePtr.p->tabDescriptor + (0 << ZAD_LOG_SIZE);
  Uint32 attrDataOffset = AttributeOffset::getOffset(tableDescriptor[attrDescIndex + 1].tabDescr);
  ndbrequire(node == &pagePtr.p->pageWord[pageOffset] + attrDataOffset);
  freeTh(fragPtr.p, tablePtr.p, signal, pagePtr.p, pageOffset);
}

void
Dbtup::tuxGetNode(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32*& node)
{
  ljamEntry();
  FragrecordPtr fragPtr;
  fragPtr.i = fragPtrI;
  ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
  TablerecPtr tablePtr;
  tablePtr.i = fragPtr.p->fragTableId;
  ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
  PagePtr pagePtr;
  pagePtr.i = pageId;
  ptrCheckGuard(pagePtr, cnoOfPage, page);
  Uint32 attrDescIndex = tablePtr.p->tabDescriptor + (0 << ZAD_LOG_SIZE);
  Uint32 attrDataOffset = AttributeOffset::getOffset(tableDescriptor[attrDescIndex + 1].tabDescr);
  node = &pagePtr.p->pageWord[pageOffset] + attrDataOffset;
}

int
Dbtup::tuxReadAttrs(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tupVersion, const Uint32* attrIds, Uint32 numAttrs, Uint32* dataOut)
{
  ljamEntry();
  // use own variables instead of globals
  FragrecordPtr fragPtr;
  fragPtr.i = fragPtrI;
  ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
  TablerecPtr tablePtr;
  tablePtr.i = fragPtr.p->fragTableId;
  ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
  PagePtr pagePtr;
  pagePtr.i = pageId;
  ptrCheckGuard(pagePtr, cnoOfPage, page);
  // search for tuple version if not original
  if (pagePtr.p->pageWord[pageOffset + 1] != tupVersion) {
    ljam();
    OperationrecPtr opPtr;
    opPtr.i = pagePtr.p->pageWord[pageOffset];
    Uint32 loopGuard = 0;
    while (true) {
      ptrCheckGuard(opPtr, cnoOfOprec, operationrec);
      if (opPtr.p->realPageIdC != RNIL) {
        // update page and offset
        pagePtr.i = opPtr.p->realPageIdC;
        pageOffset = opPtr.p->pageOffsetC;
        ptrCheckGuard(pagePtr, cnoOfPage, page);
        if (pagePtr.p->pageWord[pageOffset + 1] == tupVersion) {
          ljam();
          break;
        }
      }
      ljam();
      opPtr.i = opPtr.p->nextActiveOp;
      ndbrequire(++loopGuard < (1 << ZTUP_VERSION_BITS));
    }
  }
  // read key attributes from found tuple version
  // save globals
  TablerecPtr tabptr_old = tabptr;
  FragrecordPtr fragptr_old = fragptr;
  OperationrecPtr operPtr_old = operPtr;
  // new globals
  tabptr = tablePtr;
  fragptr = fragPtr;
  operPtr.i = RNIL;
  operPtr.p = NULL;
  // do it
  int ret = readAttributes(pagePtr.p, pageOffset, attrIds, numAttrs, dataOut, ZNIL, true);
  // restore globals
  tabptr = tabptr_old;
  fragptr = fragptr_old;
  operPtr = operPtr_old;
  // done
  if (ret == -1) {
    ret = terrorCode ? (-(int)terrorCode) : -1;
  }
  return ret;
}

int
Dbtup::tuxReadPk(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32* dataOut, bool xfrmFlag)
{
  ljamEntry();
  // use own variables instead of globals
  FragrecordPtr fragPtr;
  fragPtr.i = fragPtrI;
  ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
  TablerecPtr tablePtr;
  tablePtr.i = fragPtr.p->fragTableId;
  ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
  PagePtr pagePtr;
  pagePtr.i = pageId;
  ptrCheckGuard(pagePtr, cnoOfPage, page);
  const Uint32 tabDescriptor = tablePtr.p->tabDescriptor;
  const Uint32* attrIds = &tableDescriptor[tablePtr.p->readKeyArray].tabDescr;
  const Uint32 numAttrs = tablePtr.p->noOfKeyAttr;
  // read pk attributes from original tuple
  // save globals
  TablerecPtr tabptr_old = tabptr;
  FragrecordPtr fragptr_old = fragptr;
  OperationrecPtr operPtr_old = operPtr;
  // new globals
  tabptr = tablePtr;
  fragptr = fragPtr;
  operPtr.i = RNIL;
  operPtr.p = NULL;
  // do it
  int ret = readAttributes(pagePtr.p, pageOffset, attrIds, numAttrs, dataOut, ZNIL, xfrmFlag);
  // restore globals
  tabptr = tabptr_old;
  fragptr = fragptr_old;
  operPtr = operPtr_old;
  // done
  if (ret != -1) {
    // remove headers
    Uint32 n = 0;
    Uint32 i = 0;
    while (n < numAttrs) {
      const AttributeHeader ah(dataOut[i]);
      Uint32 size = ah.getDataSize();
      ndbrequire(size != 0);
      for (Uint32 j = 0; j < size; j++) {
        dataOut[i + j - n] = dataOut[i + j + 1];
      }
      n += 1;
      i += 1 + size;
    }
    ndbrequire((int)i == ret);
    ret -= numAttrs;
  } else {
    ret = terrorCode ? (-(int)terrorCode) : -1;
  }
  return ret;
}

int
Dbtup::accReadPk(Uint32 tableId, Uint32 fragId, Uint32 fragPageId, Uint32 pageIndex, Uint32* dataOut, bool xfrmFlag)
{
  ljamEntry();
  // get table
  TablerecPtr tablePtr;
  tablePtr.i = tableId;
  ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
  // get fragment
  FragrecordPtr fragPtr;
  getFragmentrec(fragPtr, fragId, tablePtr.p);
  // get real page id and tuple offset
  PagePtr pagePtr;
  Uint32 pageId = getRealpid(fragPtr.p, fragPageId);
  ndbrequire((pageIndex & 0x1) == 0);
  Uint32 pageOffset = ZPAGE_HEADER_SIZE + (pageIndex >> 1) * tablePtr.p->tupheadsize;
  // use TUX routine - optimize later
  int ret = tuxReadPk(fragPtr.i, pageId, pageOffset, dataOut, xfrmFlag);
  return ret;
}

bool
Dbtup::tuxQueryTh(Uint32 fragPtrI, Uint32 tupAddr, Uint32 tupVersion, Uint32 transId1, Uint32 transId2, Uint32 savePointId)
{
  ljamEntry();
  FragrecordPtr fragPtr;
  fragPtr.i = fragPtrI;
  ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
  TablerecPtr tablePtr;
  tablePtr.i = fragPtr.p->fragTableId;
  ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
  // get page
  PagePtr pagePtr;
  Uint32 fragPageId = tupAddr >> MAX_TUPLES_BITS;
  Uint32 pageIndex = tupAddr & ((1 << MAX_TUPLES_BITS ) - 1);
  // use temp op rec
  Operationrec tempOp;
  tempOp.fragPageId = fragPageId;
  tempOp.pageIndex = pageIndex;
  tempOp.transid1 = transId1;
  tempOp.transid2 = transId2;
  tempOp.savePointId = savePointId;
  tempOp.optype = ZREAD;
  tempOp.dirtyOp = 1;
  if (getPage(pagePtr, &tempOp, fragPtr.p, tablePtr.p)) {
    /*
    * We use the normal getPage which will return the tuple to be used
    * for this transaction and savepoint id.  If its tuple version
    * equals the requested then we have a visible tuple otherwise not.
    */
    ljam();
    Uint32 read_tupVersion = pagePtr.p->pageWord[tempOp.pageOffset + 1];
    if (read_tupVersion == tupVersion) {
      ljam();
      return true;