dbtupcommit.cpp

mysql-5.0.22.tar.gz源码包

  if (regOperPtr.p->tupleState == NO_OTHER_OP) {
    if ((regOperPtr.p->prevActiveOp == RNIL) &&
        (regOperPtr.p->nextActiveOp == RNIL)) {
      ljam();
/* ---------------------------------------------------------- */
// We handle the simple case separately as an optimisation
/* ---------------------------------------------------------- */
      commitSimple(signal,
                   regOperPtr.p,
                   regFragPtr.p,
                   regTabPtr.p);
    } else {
/* ---------------------------------------------------------- */
// This is the first commit message of this record in this
// transaction. We will commit this record completely for this
// transaction. If there are other operations they will be
// responsible to release their own resources. Also commit of
// a delete is postponed until the last operation is committed
// on the tuple.
//
// As part of this commitRecord we will also handle detached
// triggers and release of resources for this operation.
/* ---------------------------------------------------------- */
      ljam();
      commitRecord(signal,
                   regOperPtr.p,
                   regFragPtr.p,
                   regTabPtr.p);
      removeActiveOpList(regOperPtr.p);
    }//if
  } else {
    ljam();
/* ---------------------------------------------------------- */
// Release any copy tuples
/* ---------------------------------------------------------- */
    ndbrequire(regOperPtr.p->tupleState == TO_BE_COMMITTED);
    commitUpdate(signal, regOperPtr.p, regFragPtr.p, regTabPtr.p);
    removeActiveOpList(regOperPtr.p);
  }//if
  initOpConnection(regOperPtr.p, regFragPtr.p);
}//execTUP_COMMITREQ()

void
Dbtup::updateGcpId(Signal* signal,
                   Operationrec* const regOperPtr,
                   Fragrecord* const regFragPtr,
                   Tablerec* const regTabPtr)
{
  PagePtr pagePtr;
  ljam();
//--------------------------------------------------------------------
// Is this code safe for UNDO logging. Not sure currently. RONM
//--------------------------------------------------------------------
  pagePtr.i = regOperPtr->realPageId;
  ptrCheckGuard(pagePtr, cnoOfPage, page);
  Uint32 temp = regOperPtr->pageOffset + regTabPtr->tupGCPIndex;
  ndbrequire((temp < ZWORDS_ON_PAGE) &&
             (regTabPtr->tupGCPIndex < regTabPtr->tupheadsize));
  if (isUndoLoggingNeeded(regFragPtr, regOperPtr->fragPageId)) {
    Uint32 prevGCI = pagePtr.p->pageWord[temp];
    ljam();
    cprAddUndoLogRecord(signal,
                        ZLCPR_TYPE_UPDATE_GCI,
                        regOperPtr->fragPageId,
                        regOperPtr->pageIndex,
                        regOperPtr->tableRef,
                        regOperPtr->fragId,
                        regFragPtr->checkpointVersion);
    cprAddGCIUpdate(signal,
                    prevGCI,
                    regFragPtr);
  }//if
  pagePtr.p->pageWord[temp] = regOperPtr->gci;
  if (regTabPtr->checksumIndicator) {
    ljam();
    setChecksum(pagePtr.p, regOperPtr->pageOffset, regTabPtr->tupheadsize);
  }//if
}//Dbtup::updateGcpId()

void
Dbtup::commitRecord(Signal* signal,
                    Operationrec* const regOperPtr,
                    Fragrecord* const regFragPtr,
                    Tablerec* const regTabPtr)
{
  Uint32 opType;
  OperationrecPtr firstOpPtr;
  PagePtr pagePtr;

  pagePtr.i = regOperPtr->realPageId;
  ptrCheckGuard(pagePtr, cnoOfPage, page);

  setTupleStatesSetOpType(regOperPtr, pagePtr.p, opType, firstOpPtr);

  fragptr.p = regFragPtr;
  tabptr.p = regTabPtr;

  if (opType == ZINSERT_DELETE) {
    ljam();
//--------------------------------------------------------------------
// We started by inserting the tuple and ended by deleting. Seen from
// transactions point of view no changes were made.
//--------------------------------------------------------------------
    commitUpdate(signal, regOperPtr, regFragPtr, regTabPtr);
    return;
  } else if (opType == ZINSERT) {
    ljam();
//--------------------------------------------------------------------
// We started by inserting whereafter we made several changes to the
// tuple that could include updates, deletes and new inserts. The final
// state of the tuple is the original tuple. This is reached from this
// operation. We change the optype on this operation to ZINSERT to
// ensure proper operation of the detached trigger.
// We restore the optype after executing triggers although not really
// needed.
//--------------------------------------------------------------------
    Uint32 saveOpType = regOperPtr->optype;
    regOperPtr->optype = ZINSERT;
    operPtr.p = regOperPtr;

    checkDetachedTriggers(signal,
                          regOperPtr,
                          regTabPtr);

    regOperPtr->optype = saveOpType;
  } else if (opType == ZUPDATE) {
    ljam();
//--------------------------------------------------------------------
// We want to use the first operation which contains a copy tuple
// reference. This operation contains the before value of this record
// for this transaction. Then this operation is used for executing
// triggers with optype set to update.
//--------------------------------------------------------------------
    OperationrecPtr befOpPtr;
    findBeforeValueOperation(befOpPtr, firstOpPtr);

    Uint32 saveOpType = befOpPtr.p->optype;
    Bitmask<MAXNROFATTRIBUTESINWORDS> attributeMask;
    Bitmask<MAXNROFATTRIBUTESINWORDS> saveAttributeMask;

    calculateChangeMask(pagePtr.p,
                        regTabPtr,
                        befOpPtr.p->pageOffset,
                        attributeMask);

    saveAttributeMask.clear();
    saveAttributeMask.bitOR(befOpPtr.p->changeMask);
    befOpPtr.p->changeMask.clear();
    befOpPtr.p->changeMask.bitOR(attributeMask);
    befOpPtr.p->gci = regOperPtr->gci;
    
    befOpPtr.p->optype = opType;
    operPtr.p = befOpPtr.p;
    checkDetachedTriggers(signal,
                          befOpPtr.p,
                          regTabPtr);

    befOpPtr.p->changeMask.clear();
    befOpPtr.p->changeMask.bitOR(saveAttributeMask);

    befOpPtr.p->optype = saveOpType;
  } else if (opType == ZDELETE) {
    ljam();
//--------------------------------------------------------------------
// We want to use the first operation which contains a copy tuple.
// We benefit from the fact that we know that it cannot be a simple
// delete and it cannot be an insert followed by a delete. Thus there
// must either be an update or a insert following a delete. In both
// cases we will find a before value in a copy tuple.
//
// An added complexity is that the trigger handling assumes that the
// before value is located in the original tuple so we have to move the
// copy tuple reference to the original tuple reference and afterwards
// restore it again.
//--------------------------------------------------------------------
    OperationrecPtr befOpPtr;
    findBeforeValueOperation(befOpPtr, firstOpPtr);
    Uint32 saveOpType = befOpPtr.p->optype;

    Uint32 realPageId = befOpPtr.p->realPageId;
    Uint32 pageOffset = befOpPtr.p->pageOffset;
    Uint32 fragPageId = befOpPtr.p->fragPageId;
    Uint32 pageIndex  = befOpPtr.p->pageIndex;

    befOpPtr.p->realPageId = befOpPtr.p->realPageIdC;
    befOpPtr.p->pageOffset = befOpPtr.p->pageOffsetC;
    befOpPtr.p->fragPageId = befOpPtr.p->fragPageIdC;
    befOpPtr.p->pageIndex  = befOpPtr.p->pageIndexC;
    befOpPtr.p->gci = regOperPtr->gci;

    befOpPtr.p->optype = opType;
    operPtr.p = befOpPtr.p;
    checkDetachedTriggers(signal,
                          befOpPtr.p,
                          regTabPtr);

    befOpPtr.p->realPageId = realPageId;
    befOpPtr.p->pageOffset = pageOffset;
    befOpPtr.p->fragPageId = fragPageId;
    befOpPtr.p->pageIndex  = pageIndex;
    befOpPtr.p->optype     = saveOpType;
  } else {
    ndbrequire(false);
  }//if

  commitUpdate(signal, regOperPtr, regFragPtr, regTabPtr);
  if (regTabPtr->GCPIndicator) {
    updateGcpId(signal, regOperPtr, regFragPtr, regTabPtr);
  }//if
}//Dbtup::commitRecord()

void
Dbtup::setTupleStatesSetOpType(Operationrec* const regOperPtr,
                               Page* const pagePtr,
                               Uint32& opType,
                               OperationrecPtr& firstOpPtr)
{
  OperationrecPtr loopOpPtr;
  OperationrecPtr lastOpPtr;

  ndbrequire(regOperPtr->pageOffset < ZWORDS_ON_PAGE);
  loopOpPtr.i = pagePtr->pageWord[regOperPtr->pageOffset];
  ptrCheckGuard(loopOpPtr, cnoOfOprec, operationrec);
  lastOpPtr = loopOpPtr;
  if (loopOpPtr.p->optype == ZDELETE) {
    ljam();
    opType = ZDELETE;
  } else {
    ljam();
    opType = ZUPDATE;
  }//if
  do {
    ljam();
    ptrCheckGuard(loopOpPtr, cnoOfOprec, operationrec);
    firstOpPtr = loopOpPtr;
    loopOpPtr.p->tupleState = TO_BE_COMMITTED;
    loopOpPtr.i = loopOpPtr.p->nextActiveOp;
  } while (loopOpPtr.i != RNIL);
  if (opType == ZDELETE) {
    ljam();
    if (firstOpPtr.p->optype == ZINSERT) {
      ljam();
      opType = ZINSERT_DELETE;
    }//if
  } else {
    ljam();
    if (firstOpPtr.p->optype == ZINSERT) {
      ljam();
      opType = ZINSERT;
    }//if
  }///if
}//Dbtup::setTupleStatesSetOpType()

void Dbtup::findBeforeValueOperation(OperationrecPtr& befOpPtr,
                                     OperationrecPtr firstOpPtr)
{
  befOpPtr = firstOpPtr;
  if (befOpPtr.p->realPageIdC != RNIL) {
    ljam();
    return;
  } else {
    ljam();
    befOpPtr.i = befOpPtr.p->prevActiveOp;
    ptrCheckGuard(befOpPtr, cnoOfOprec, operationrec);
    ndbrequire(befOpPtr.p->realPageIdC != RNIL);
  }//if
}//Dbtup::findBeforeValueOperation()

void
Dbtup::calculateChangeMask(Page* const pagePtr,
                           Tablerec* const regTabPtr,
                           Uint32 pageOffset,
                           Bitmask<MAXNROFATTRIBUTESINWORDS>& attributeMask)
{
  OperationrecPtr loopOpPtr;

  attributeMask.clear();
  ndbrequire(pageOffset < ZWORDS_ON_PAGE);
  loopOpPtr.i = pagePtr->pageWord[pageOffset];
  do {
    ptrCheckGuard(loopOpPtr, cnoOfOprec, operationrec);
    if (loopOpPtr.p->optype == ZUPDATE) {
      ljam();
      attributeMask.bitOR(loopOpPtr.p->changeMask);
    } else if (loopOpPtr.p->optype == ZINSERT) {
      ljam();
      attributeMask.set();
      return;
    } else {
      ndbrequire(loopOpPtr.p->optype == ZDELETE);
    }//if
    loopOpPtr.i = loopOpPtr.p->nextActiveOp;
  } while (loopOpPtr.i != RNIL);
}//Dbtup::calculateChangeMask()