dbtupindex.cpp

mysql-5.0.22.tar.gz源码包

    }
  }
  return false;
}

// ordered index build

//#define TIME_MEASUREMENT
#ifdef TIME_MEASUREMENT
  static Uint32 time_events;
  NDB_TICKS tot_time_passed;
  Uint32 number_events;
#endif
void
Dbtup::execBUILDINDXREQ(Signal* signal)
{
  ljamEntry();
#ifdef TIME_MEASUREMENT
  time_events = 0;
  tot_time_passed = 0;
  number_events = 1;
#endif
  // get new operation
  BuildIndexPtr buildPtr;
  if (! c_buildIndexList.seize(buildPtr)) {
    ljam();
    BuildIndexRec buildRec;
    memcpy(buildRec.m_request, signal->theData, sizeof(buildRec.m_request));
    buildRec.m_errorCode = BuildIndxRef::Busy;
    buildIndexReply(signal, &buildRec);
    return;
  }
  memcpy(buildPtr.p->m_request, signal->theData, sizeof(buildPtr.p->m_request));
  // check
  buildPtr.p->m_errorCode = BuildIndxRef::NoError;
  do {
    const BuildIndxReq* buildReq = (const BuildIndxReq*)buildPtr.p->m_request;
    if (buildReq->getTableId() >= cnoOfTablerec) {
      ljam();
      buildPtr.p->m_errorCode = BuildIndxRef::InvalidPrimaryTable;
      break;
    }
    TablerecPtr tablePtr;
    tablePtr.i = buildReq->getTableId();
    ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
    if (tablePtr.p->tableStatus != DEFINED) {
      ljam();
      buildPtr.p->m_errorCode = BuildIndxRef::InvalidPrimaryTable;
      break;
    }
    if (! DictTabInfo::isOrderedIndex(buildReq->getIndexType())) {
      ljam();
      buildPtr.p->m_errorCode = BuildIndxRef::InvalidIndexType;
      break;
    }
    const ArrayList<TupTriggerData>& triggerList = tablePtr.p->tuxCustomTriggers;
    TriggerPtr triggerPtr;
    triggerList.first(triggerPtr);
    while (triggerPtr.i != RNIL) {
      if (triggerPtr.p->indexId == buildReq->getIndexId()) {
        ljam();
        break;
      }
      triggerList.next(triggerPtr);
    }
    if (triggerPtr.i == RNIL) {
      ljam();
      // trigger was not created
      buildPtr.p->m_errorCode = BuildIndxRef::InternalError;
      break;
    }
    buildPtr.p->m_triggerPtrI = triggerPtr.i;
    // set to first tuple position
    buildPtr.p->m_fragNo = 0;
    buildPtr.p->m_pageId = 0;
    buildPtr.p->m_tupleNo = 0;
    // start build
    buildIndex(signal, buildPtr.i);
    return;
  } while (0);
  // check failed
  buildIndexReply(signal, buildPtr.p);
  c_buildIndexList.release(buildPtr);
}

void
Dbtup::buildIndex(Signal* signal, Uint32 buildPtrI)
{
  // get build record
  BuildIndexPtr buildPtr;
  buildPtr.i = buildPtrI;
  c_buildIndexList.getPtr(buildPtr);
  const BuildIndxReq* buildReq = (const BuildIndxReq*)buildPtr.p->m_request;
  // get table
  TablerecPtr tablePtr;
  tablePtr.i = buildReq->getTableId();
  ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
  // get trigger
  TriggerPtr triggerPtr;
  triggerPtr.i = buildPtr.p->m_triggerPtrI;
  c_triggerPool.getPtr(triggerPtr);
  ndbrequire(triggerPtr.p->indexId == buildReq->getIndexId());
#ifdef TIME_MEASUREMENT
  MicroSecondTimer start;
  MicroSecondTimer stop;
  NDB_TICKS time_passed;
#endif
  do {
    // get fragment
    FragrecordPtr fragPtr;
    if (buildPtr.p->m_fragNo == 2 * MAX_FRAG_PER_NODE) {
      ljam();
      // build ready
      buildIndexReply(signal, buildPtr.p);
      c_buildIndexList.release(buildPtr);
      return;
    }
    ndbrequire(buildPtr.p->m_fragNo < 2 * MAX_FRAG_PER_NODE);
    fragPtr.i = tablePtr.p->fragrec[buildPtr.p->m_fragNo];
    if (fragPtr.i == RNIL) {
      ljam();
      buildPtr.p->m_fragNo++;
      buildPtr.p->m_pageId = 0;
      buildPtr.p->m_tupleNo = 0;
      break;
    }
    ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
    // get page
    PagePtr pagePtr;
    if (buildPtr.p->m_pageId >= fragPtr.p->noOfPages) {
      ljam();
      buildPtr.p->m_fragNo++;
      buildPtr.p->m_pageId = 0;
      buildPtr.p->m_tupleNo = 0;
      break;
    }
    Uint32 realPageId = getRealpid(fragPtr.p, buildPtr.p->m_pageId);
    pagePtr.i = realPageId;
    ptrCheckGuard(pagePtr, cnoOfPage, page);
    const Uint32 pageState = pagePtr.p->pageWord[ZPAGE_STATE_POS];
    if (pageState != ZTH_MM_FREE &&
        pageState != ZTH_MM_FREE_COPY &&
        pageState != ZTH_MM_FULL &&
        pageState != ZTH_MM_FULL_COPY) {
      ljam();
      buildPtr.p->m_pageId++;
      buildPtr.p->m_tupleNo = 0;
      break;
    }
    // get tuple
    const Uint32 tupheadsize = tablePtr.p->tupheadsize;
    Uint32 pageOffset = ZPAGE_HEADER_SIZE + buildPtr.p->m_tupleNo * tupheadsize;
    if (pageOffset + tupheadsize > ZWORDS_ON_PAGE) {
      ljam();
      buildPtr.p->m_pageId++;
      buildPtr.p->m_tupleNo = 0;
      break;
    }
    // skip over free tuple
    bool isFree = false;
    if (pageState == ZTH_MM_FREE ||
        pageState == ZTH_MM_FREE_COPY) {
      ljam();
      if ((pagePtr.p->pageWord[pageOffset] >> 16) == tupheadsize) {
        // verify it really is free  XXX far too expensive
        Uint32 nextTuple = pagePtr.p->pageWord[ZFREELIST_HEADER_POS] >> 16;
        ndbrequire(nextTuple != 0);
        while (nextTuple != 0) {
          ljam();
          if (nextTuple == pageOffset) {
            ljam();
            isFree = true;
            break;
          }
          nextTuple = pagePtr.p->pageWord[nextTuple] & 0xffff;
        }
      }
    }
    if (isFree) {
      ljam();
      buildPtr.p->m_tupleNo++;
      break;
    }
    Uint32 tupVersion = pagePtr.p->pageWord[pageOffset + 1];
    OperationrecPtr pageOperPtr;
    pageOperPtr.i = pagePtr.p->pageWord[pageOffset];
    if (pageOperPtr.i != RNIL) {
      /*
      If there is an ongoing operation on the tuple then it is either a
      copy tuple or an original tuple with an ongoing transaction. In
      both cases realPageId and pageOffset refer to the original tuple.
      The tuple address stored in TUX will always be the original tuple
      but with the tuple version of the tuple we found.

      This is necessary to avoid having to update TUX at abort of
      update. If an update aborts then the copy tuple is copied to
      the original tuple. The build will however have found that
      tuple as a copy tuple. The original tuple is stable and is thus
      preferrable to store in TUX.
      */
      ljam();
      ptrCheckGuard(pageOperPtr, cnoOfOprec, operationrec);
      realPageId = pageOperPtr.p->realPageId;
      pageOffset = pageOperPtr.p->pageOffset;
    }//if
#ifdef TIME_MEASUREMENT
    NdbTick_getMicroTimer(&start);
#endif
    // add to index
    TuxMaintReq* const req = (TuxMaintReq*)signal->getDataPtrSend();
    req->errorCode = RNIL;
    req->tableId = tablePtr.i;
    req->indexId = triggerPtr.p->indexId;
    req->fragId = tablePtr.p->fragid[buildPtr.p->m_fragNo];
    req->pageId = realPageId;
    req->pageOffset = pageOffset;
    req->tupVersion = tupVersion;
    req->opInfo = TuxMaintReq::OpAdd;
    EXECUTE_DIRECT(DBTUX, GSN_TUX_MAINT_REQ,
        signal, TuxMaintReq::SignalLength);
    ljamEntry();
    if (req->errorCode != 0) {
      switch (req->errorCode) {
      case TuxMaintReq::NoMemError:
        ljam();
        buildPtr.p->m_errorCode = BuildIndxRef::AllocationFailure;
        break;
      default:
        ndbrequire(false);
        break;
      }
      buildIndexReply(signal, buildPtr.p);
      c_buildIndexList.release(buildPtr);
      return;
    }
#ifdef TIME_MEASUREMENT
    NdbTick_getMicroTimer(&stop);
    time_passed = NdbTick_getMicrosPassed(start, stop);
    if (time_passed < 1000) {
      time_events++;
      tot_time_passed += time_passed;
      if (time_events == number_events) {
        NDB_TICKS mean_time_passed = tot_time_passed / (NDB_TICKS)number_events;
        ndbout << "Number of events = " << number_events;
        ndbout << " Mean time passed = " << mean_time_passed << endl;
        number_events <<= 1;
        tot_time_passed = (NDB_TICKS)0;
        time_events = 0;
      }//if
    }
#endif
    // next tuple
    buildPtr.p->m_tupleNo++;
    break;
  } while (0);
  signal->theData[0] = ZBUILD_INDEX;
  signal->theData[1] = buildPtr.i;
  sendSignal(reference(), GSN_CONTINUEB, signal, 2, JBB);
}

void
Dbtup::buildIndexReply(Signal* signal, const BuildIndexRec* buildPtrP)
{
  const BuildIndxReq* const buildReq = (const BuildIndxReq*)buildPtrP->m_request;
  // conf is subset of ref
  BuildIndxRef* rep = (BuildIndxRef*)signal->getDataPtr();
  rep->setUserRef(buildReq->getUserRef());
  rep->setConnectionPtr(buildReq->getConnectionPtr());
  rep->setRequestType(buildReq->getRequestType());
  rep->setTableId(buildReq->getTableId());
  rep->setIndexType(buildReq->getIndexType());
  rep->setIndexId(buildReq->getIndexId());
  // conf
  if (buildPtrP->m_errorCode == BuildIndxRef::NoError) {
    ljam();
    sendSignal(rep->getUserRef(), GSN_BUILDINDXCONF,
        signal, BuildIndxConf::SignalLength, JBB);
    return;
  }
  // ref
  rep->setErrorCode(buildPtrP->m_errorCode);
  sendSignal(rep->getUserRef(), GSN_BUILDINDXREF,
      signal, BuildIndxRef::SignalLength, JBB);
}