dbtupexecquery.cpp

mysql-5.0.22.tar.gz源码包

bool
Dbtup::getPage(PagePtr& pagePtr,
               Operationrec* const regOperPtr,
               Fragrecord* const regFragPtr,
               Tablerec* const regTabPtr)
{
/* ------------------------------------------------------------------------- */
// GET THE REFERENCE TO THE TUPLE HEADER BY TRANSLATING THE FRAGMENT PAGE ID
// INTO A REAL PAGE ID AND BY USING THE PAGE INDEX TO DERIVE THE PROPER INDEX
// IN THE REAL PAGE.
/* ------------------------------------------------------------------------- */
  pagePtr.i = getRealpid(regFragPtr, regOperPtr->fragPageId);
  regOperPtr->realPageId = pagePtr.i;
  Uint32 RpageIndex = regOperPtr->pageIndex;
  Uint32 Rtupheadsize = regTabPtr->tupheadsize;
  ptrCheckGuard(pagePtr, cnoOfPage, page);
  Uint32 RpageIndexScaled = RpageIndex >> 1;
  ndbrequire((RpageIndex & 1) == 0);
  regOperPtr->pageOffset = ZPAGE_HEADER_SIZE + 
                           (Rtupheadsize * RpageIndexScaled);

  OperationrecPtr leaderOpPtr;
  ndbrequire(regOperPtr->pageOffset < ZWORDS_ON_PAGE);
  leaderOpPtr.i = pagePtr.p->pageWord[regOperPtr->pageOffset];
  if (leaderOpPtr.i == RNIL) {
    return true;
  }//if
  ptrCheckGuard(leaderOpPtr, cnoOfOprec, operationrec);
  bool dirtyRead = ((regOperPtr->optype == ZREAD) &&
                    (regOperPtr->dirtyOp == 1));
  if (dirtyRead) {
    bool sameTrans = ((regOperPtr->transid1 == leaderOpPtr.p->transid1) &&
                      (regOperPtr->transid2 == leaderOpPtr.p->transid2));
    if (!sameTrans) {
      if (!getPageLastCommitted(regOperPtr, leaderOpPtr.p)) {
        return false;
      }//if
      pagePtr.i = regOperPtr->realPageId;
      ptrCheckGuard(pagePtr, cnoOfPage, page);
      return true;
    }//if
  }//if
  if (regOperPtr->optype == ZREAD) {
    /*
    Read uses savepoint id's to find the correct tuple version.
    */
    if (getPageThroughSavePoint(regOperPtr, leaderOpPtr.p)) {
      jam();
      pagePtr.i = regOperPtr->realPageId;
      ptrCheckGuard(pagePtr, cnoOfPage, page);
      return true;
    }
    return false;
  }
//----------------------------------------------------------------------
// Check that no other operation is already active on the tuple. Also
// that abort or commit is not ongoing.
//----------------------------------------------------------------------
  if (leaderOpPtr.p->tupleState == NO_OTHER_OP) {
    jam();
    if ((leaderOpPtr.p->optype == ZDELETE) &&
        (regOperPtr->optype != ZINSERT)) {
      jam();
      terrorCode = ZTUPLE_DELETED_ERROR;
      return false;
    }//if
    return true;
  } else if (leaderOpPtr.p->tupleState == ALREADY_ABORTED) {
    jam();
    terrorCode = ZMUST_BE_ABORTED_ERROR;
    return false;
  } else {
    ndbrequire(false);
  }//if
  return true;
}//Dbtup::getPage()

bool
Dbtup::getPageThroughSavePoint(Operationrec* regOperPtr,
                               Operationrec* leaderOpPtr)
{
  bool found = false;
  OperationrecPtr loopOpPtr;
  loopOpPtr.p = leaderOpPtr;
  while(true) {
    if (regOperPtr->savePointId > loopOpPtr.p->savePointId) {
      jam();
      found = true;
      break;
    }
    if (loopOpPtr.p->nextActiveOp == RNIL) {
      break;
    }
    loopOpPtr.i = loopOpPtr.p->nextActiveOp;
    ptrCheckGuard(loopOpPtr, cnoOfOprec, operationrec);
    jam();
  }
  if (!found) {
    return getPageLastCommitted(regOperPtr, loopOpPtr.p);
  } else {
    if (loopOpPtr.p->optype == ZDELETE) {
      jam();
      terrorCode = ZTUPLE_DELETED_ERROR;
      return false;
    }
    if (loopOpPtr.p->tupleState == ALREADY_ABORTED) {
      /*
      Requested tuple version has already been aborted
      */
      jam();
      terrorCode = ZMUST_BE_ABORTED_ERROR;
      return false;
    }
    bool use_copy;
    if (loopOpPtr.p->prevActiveOp == RNIL) {
      jam();
      /*
      Use original tuple since we are reading from the last written tuple.
      We are the 
      */
      use_copy = false;
    } else {
      /*
      Go forward in time to find a copy of the tuple which this operation
      produced
      */
      loopOpPtr.i = loopOpPtr.p->prevActiveOp;
      ptrCheckGuard(loopOpPtr, cnoOfOprec, operationrec);
      if (loopOpPtr.p->optype == ZDELETE) {
        /*
        This operation was a Delete and thus have no copy tuple attached to
        it. We will move forward to the next that either doesn't exist in
        which case we will return the original tuple of any operation and
        otherwise it must be an insert which contains a copy record.
        */
        if (loopOpPtr.p->prevActiveOp == RNIL) {
          jam();
          use_copy = false;
        } else {
          jam();
          loopOpPtr.i = loopOpPtr.p->prevActiveOp;
          ptrCheckGuard(loopOpPtr, cnoOfOprec, operationrec);
          ndbrequire(loopOpPtr.p->optype == ZINSERT);
          use_copy = true;
        }
      } else if (loopOpPtr.p->optype == ZUPDATE) {
        jam();
        /*
        This operation which was the next in time have a copy which was the
        result of the previous operation which we want to use. Thus use
        the copy tuple of this operation.
        */
        use_copy = true;
      } else {
        /*
        This operation was an insert that happened after an insert or update.
        This is not a possible case.
        */
        ndbrequire(false);
        return false;
      }
    }
    if (use_copy) {
      regOperPtr->realPageId = loopOpPtr.p->realPageIdC;
      regOperPtr->fragPageId = loopOpPtr.p->fragPageIdC;
      regOperPtr->pageIndex = loopOpPtr.p->pageIndexC;
      regOperPtr->pageOffset = loopOpPtr.p->pageOffsetC;
    } else {
      regOperPtr->realPageId = loopOpPtr.p->realPageId;
      regOperPtr->fragPageId = loopOpPtr.p->fragPageId;
      regOperPtr->pageIndex = loopOpPtr.p->pageIndex;
      regOperPtr->pageOffset = loopOpPtr.p->pageOffset;
    }
    return true;
  }
}

bool
Dbtup::getPageLastCommitted(Operationrec* const regOperPtr,
                            Operationrec* const leaderOpPtr)
{
//----------------------------------------------------------------------
// Dirty reads wants to read the latest committed tuple. The latest
// tuple value could be not existing or else we have to find the copy
// tuple. Start by finding the end of the list to find the first operation
// on the record in the ongoing transaction.
//----------------------------------------------------------------------
  jam();
  OperationrecPtr loopOpPtr;
  loopOpPtr.p = leaderOpPtr;
  while (loopOpPtr.p->nextActiveOp != RNIL) {
    jam();
    loopOpPtr.i = loopOpPtr.p->nextActiveOp;
    ptrCheckGuard(loopOpPtr, cnoOfOprec, operationrec);
  }//while
  if (loopOpPtr.p->optype == ZINSERT) {
    jam();
//----------------------------------------------------------------------
// With an insert in the start of the list we know that the tuple did not
// exist before this transaction was started. We don't care if the current
// transaction is in the commit phase since the commit is not really
// completed until the operation is gone from TUP.
//----------------------------------------------------------------------
    terrorCode = ZTUPLE_DELETED_ERROR;
    return false;
  } else {
//----------------------------------------------------------------------
// A successful update and delete as first in the queue means that a tuple
// exist in the committed world. We need to find it.
//----------------------------------------------------------------------
    if (loopOpPtr.p->optype == ZUPDATE) {
      jam();
//----------------------------------------------------------------------
// The first operation was a delete we set our tuple reference to the
// copy tuple of this operation.
//----------------------------------------------------------------------
      regOperPtr->realPageId = loopOpPtr.p->realPageIdC;
      regOperPtr->fragPageId = loopOpPtr.p->fragPageIdC;
      regOperPtr->pageIndex  = loopOpPtr.p->pageIndexC;
      regOperPtr->pageOffset = loopOpPtr.p->pageOffsetC;
    } else if ((loopOpPtr.p->optype == ZDELETE) &&
               (loopOpPtr.p->prevActiveOp == RNIL)) {
      jam();
//----------------------------------------------------------------------
// There was only a delete. The original tuple still is ok.
//----------------------------------------------------------------------
    } else {
      jam();
//----------------------------------------------------------------------
// There was another operation after the delete, this must be an insert
// and we have found our copy tuple there.
//----------------------------------------------------------------------
      loopOpPtr.i = loopOpPtr.p->prevActiveOp;
      ptrCheckGuard(loopOpPtr, cnoOfOprec, operationrec);
      ndbrequire(loopOpPtr.p->optype == ZINSERT);
      regOperPtr->realPageId = loopOpPtr.p->realPageIdC;
      regOperPtr->fragPageId = loopOpPtr.p->fragPageIdC;
      regOperPtr->pageIndex  = loopOpPtr.p->pageIndexC;
      regOperPtr->pageOffset = loopOpPtr.p->pageOffsetC;
    }//if
  }//if
  return true;
}//Dbtup::getPageLastCommitted()

void Dbtup::execTUPKEYREQ(Signal* signal) 
{
  TupKeyReq * const tupKeyReq = (TupKeyReq *)signal->getDataPtr();
  Uint32 RoperPtr = tupKeyReq->connectPtr;
  Uint32 Rtabptr = tupKeyReq->tableRef;
  Uint32 RfragId = tupKeyReq->fragId;
  Uint32 Rstoredid = tupKeyReq->storedProcedure;
  Uint32 Rfragptr = tupKeyReq->fragPtr;

  Uint32 RnoOfOprec = cnoOfOprec;
  Uint32 RnoOfTablerec = cnoOfTablerec;
  Uint32 RnoOfFragrec = cnoOfFragrec;

  operPtr.i = RoperPtr;
  fragptr.i = Rfragptr;
  tabptr.i = Rtabptr;
  jamEntry();

  ndbrequire(((RoperPtr < RnoOfOprec) &&
        (Rtabptr < RnoOfTablerec) &&
        (Rfragptr < RnoOfFragrec)));
  ptrAss(operPtr, operationrec);
  Operationrec * const regOperPtr = operPtr.p;
  ptrAss(fragptr, fragrecord);
  Fragrecord * const regFragPtr = fragptr.p;
  ptrAss(tabptr, tablerec);
  Tablerec* const regTabPtr = tabptr.p;

  Uint32 TrequestInfo = tupKeyReq->request;

  if (regOperPtr->transstate != IDLE) {
    TUPKEY_abort(signal, 39);
    return;
  }//if
/* ----------------------------------------------------------------- */
// Operation is ZREAD when we arrive here so no need to worry about the
// abort process.
/* ----------------------------------------------------------------- */
/* -----------    INITIATE THE OPERATION RECORD       -------------- */
/* ----------------------------------------------------------------- */
  regOperPtr->fragmentPtr = Rfragptr;
  regOperPtr->dirtyOp = TrequestInfo & 1;
  regOperPtr->opSimple = (TrequestInfo >> 1) & 1;
  regOperPtr->interpretedExec = (TrequestInfo >> 10) & 1;
  regOperPtr->optype = (TrequestInfo >> 6) & 0xf;

  // Attributes needed by trigger execution
  regOperPtr->noFiredTriggers = 0;
  regOperPtr->tableRef = Rtabptr;
  regOperPtr->tcOperationPtr = tupKeyReq->opRef;
  regOperPtr->primaryReplica = tupKeyReq->primaryReplica;
  regOperPtr->coordinatorTC = tupKeyReq->coordinatorTC;
  regOperPtr->tcOpIndex = tupKeyReq->tcOpIndex;
  regOperPtr->savePointId = tupKeyReq->savePointId;

  regOperPtr->fragId = RfragId;

  regOperPtr->fragPageId = tupKeyReq->keyRef1;
  regOperPtr->pageIndex = tupKeyReq->keyRef2;
  regOperPtr->attrinbufLen = regOperPtr->logSize = tupKeyReq->attrBufLen;
  regOperPtr->recBlockref = tupKeyReq->applRef;

// Schema Version in tupKeyReq->schemaVersion not used in this version
  regOperPtr->storedProcedureId = Rstoredid;
  regOperPtr->transid1 = tupKeyReq->transId1;
  regOperPtr->transid2 = tupKeyReq->transId2;

  regOperPtr->attroutbufLen = 0;
/* ----------------------------------------------------------------------- */
// INITIALISE TO DEFAULT VALUE
// INIT THE COPY REFERENCE RECORDS TO RNIL TO ENSURE THAT THEIR VALUES
// ARE VALID IF THEY EXISTS
// NO PENDING CHECKPOINT WHEN COPY CREATED (DEFAULT)
// NO TUPLE HAS BEEN ALLOCATED YET
// NO COPY HAS BEEN CREATED YET
/* ----------------------------------------------------------------------- */
  regOperPtr->undoLogged = false;
  regOperPtr->realPageId = RNIL;
  regOperPtr->realPageIdC = RNIL;
  regOperPtr->fragPageIdC = RNIL;

  regOperPtr->pageOffset = ZNIL;
  regOperPtr->pageOffsetC = ZNIL;

  regOperPtr->pageIndexC = ZNIL;

  // version not yet known
  regOperPtr->tupVersion = ZNIL;
  regOperPtr->deleteInsertFlag = 0;

  regOperPtr->tupleState = TUPLE_BLOCKED;
  regOperPtr->changeMask.clear();
  
  if (Rstoredid != ZNIL) {
    ndbrequire(initStoredOperationrec(regOperPtr, Rstoredid) == ZOK);
  }//if
  copyAttrinfo(signal, regOperPtr, &cinBuffer[0]);

  PagePtr pagePtr;
  if (!getPage(pagePtr, regOperPtr, regFragPtr, regTabPtr)) {
    tupkeyErrorLab(signal);
    return;
  }//if

  Uint32 Roptype = regOperPtr->optype;
  if (Roptype == ZREAD) {
    jam();
    if (handleReadReq(signal, regOperPtr, regTabPtr, pagePtr.p) != -1) {
      sendTUPKEYCONF(signal, regOperPtr, 0);
/* ------------------------------------------------------------------------- */
// Read Operations need not to be taken out of any lists. We also do not
// need to wait for commit since there is no changes to commit. Thus we
// prepare the operation record already now for the next operation.
// Write operations have set the state to STARTED above indicating that
// they are waiting for the Commit or Abort decision.
/* ------------------------------------------------------------------------- */
      regOperPtr->transstate = IDLE;
      regOperPtr->currentAttrinbufLen = 0;
    }//if
    return;
  }//if
  linkOpIntoFragList(operPtr, regFragPtr);
  insertActiveOpList(signal,
                     operPtr,
                     pagePtr.p,
                     regOperPtr->pageOffset);