dblqhmain.cpp

来自「MySQL源码文件5.X系列, 可自已编译到服务器」· C++ 代码 · 共 1,915 行 · 第 1/5 页

  }
  ndbrequire(false);
}//Dblqh::execLQHFRAGREQ()

/* *************** */
/*  ACCFRAGCONF  > */
/* *************** */
void Dblqh::execACCFRAGCONF(Signal* signal) 
{
  jamEntry();
  addfragptr.i = signal->theData[0];
  Uint32 taccConnectptr = signal->theData[1];
  Uint32 fragId1 = signal->theData[2];
  Uint32 fragId2 = signal->theData[3];
  Uint32 accFragPtr1 = signal->theData[4];
  Uint32 accFragPtr2 = signal->theData[5];
  ptrCheckGuard(addfragptr, caddfragrecFileSize, addFragRecord);
  ndbrequire(addfragptr.p->addfragStatus == AddFragRecord::ACC_ADDFRAG);

  addfragptr.p->accConnectptr = taccConnectptr;
  addfragptr.p->fragid1 = fragId1;
  addfragptr.p->fragid2 = fragId2;
  fragptr.i = addfragptr.p->fragmentPtr;
  ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
  fragptr.p->accFragptr[0] = accFragPtr1;
  fragptr.p->accFragptr[1] = accFragPtr2;

  addfragptr.p->addfragStatus = AddFragRecord::WAIT_TWO_TUP;
  sendAddFragReq(signal);
}//Dblqh::execACCFRAGCONF()

/* *************** */
/*  TUPFRAGCONF  > */
/* *************** */
void Dblqh::execTUPFRAGCONF(Signal* signal) 
{
  jamEntry();
  addfragptr.i = signal->theData[0];
  Uint32 tupConnectptr = signal->theData[1];
  Uint32 tupFragPtr = signal->theData[2];  /* TUP FRAGMENT POINTER */
  Uint32 localFragId = signal->theData[3];  /* LOCAL FRAGMENT ID    */
  ptrCheckGuard(addfragptr, caddfragrecFileSize, addFragRecord);
  fragptr.i = addfragptr.p->fragmentPtr;
  ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
  if (localFragId == addfragptr.p->fragid1) {
    jam();
    fragptr.p->tupFragptr[0] = tupFragPtr;
  } else if (localFragId == addfragptr.p->fragid2) {
    jam();
    fragptr.p->tupFragptr[1] = tupFragPtr;
  } else {
    ndbrequire(false);
    return;
  }//if
  switch (addfragptr.p->addfragStatus) {
  case AddFragRecord::WAIT_TWO_TUP:
    jam();
    fragptr.p->tupFragptr[0] = tupFragPtr;
    addfragptr.p->tup1Connectptr = tupConnectptr;
    addfragptr.p->addfragStatus = AddFragRecord::WAIT_ONE_TUP;
    sendAddFragReq(signal);
    break;
  case AddFragRecord::WAIT_ONE_TUP:
    jam();
    fragptr.p->tupFragptr[1] = tupFragPtr;
    addfragptr.p->tup2Connectptr = tupConnectptr;
    if (DictTabInfo::isOrderedIndex(addfragptr.p->tableType)) {
      addfragptr.p->addfragStatus = AddFragRecord::WAIT_TWO_TUX;
      sendAddFragReq(signal);
      break;
    }
    goto done_with_frag;
    break;
  case AddFragRecord::WAIT_TWO_TUX:
    jam();
    fragptr.p->tuxFragptr[0] = tupFragPtr;
    addfragptr.p->tux1Connectptr = tupConnectptr;
    addfragptr.p->addfragStatus = AddFragRecord::WAIT_ONE_TUX;
    sendAddFragReq(signal);
    break;
  case AddFragRecord::WAIT_ONE_TUX:
    jam();
    fragptr.p->tuxFragptr[1] = tupFragPtr;
    addfragptr.p->tux2Connectptr = tupConnectptr;
    goto done_with_frag;
    break;
  done_with_frag:
    /* ---------------------------------------------------------------- */
    /* Finished create of fragments. Now ready for creating attributes. */
    /* ---------------------------------------------------------------- */
    addfragptr.p->addfragStatus = AddFragRecord::WAIT_ADD_ATTR;
    {
      LqhFragConf* conf = (LqhFragConf*)signal->getDataPtrSend();
      conf->senderData = addfragptr.p->dictConnectptr;
      conf->lqhFragPtr = addfragptr.i;
      sendSignal(addfragptr.p->dictBlockref, GSN_LQHFRAGCONF,
          signal, LqhFragConf::SignalLength, JBB);
    }
    break;
  default:
    ndbrequire(false);
    break;
  }
}//Dblqh::execTUPFRAGCONF()

/* *************** */
/*  TUXFRAGCONF  > */
/* *************** */
void Dblqh::execTUXFRAGCONF(Signal* signal) 
{
  jamEntry();
  execTUPFRAGCONF(signal);
}//Dblqh::execTUXFRAGCONF

/*
 * Add fragment in TUP or TUX.  Called up to 4 times.
 */
void
Dblqh::sendAddFragReq(Signal* signal)
{
  fragptr.i = addfragptr.p->fragmentPtr;
  ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
  if (addfragptr.p->addfragStatus == AddFragRecord::WAIT_TWO_TUP ||
      addfragptr.p->addfragStatus == AddFragRecord::WAIT_ONE_TUP) {
    TupFragReq* const tupFragReq = (TupFragReq*)signal->getDataPtrSend();
    if (DictTabInfo::isTable(addfragptr.p->tableType) ||
        DictTabInfo::isHashIndex(addfragptr.p->tableType)) {
      jam();
      tupFragReq->userPtr = addfragptr.i;
      tupFragReq->userRef = cownref;
      tupFragReq->reqInfo = 0; /* ADD TABLE */
      tupFragReq->tableId = addfragptr.p->tabId;
      tupFragReq->noOfAttr = addfragptr.p->noOfAttr;
      tupFragReq->fragId =
        addfragptr.p->addfragStatus == AddFragRecord::WAIT_TWO_TUP
        ? addfragptr.p->fragid1 : addfragptr.p->fragid2;
      tupFragReq->maxRowsLow = addfragptr.p->maxRowsLow;
      tupFragReq->maxRowsHigh = addfragptr.p->maxRowsHigh;
      tupFragReq->minRowsLow = addfragptr.p->minRowsLow;
      tupFragReq->minRowsHigh = addfragptr.p->minRowsHigh;
      tupFragReq->noOfNullAttr = addfragptr.p->noOfNull;
      tupFragReq->schemaVersion = addfragptr.p->schemaVer;
      tupFragReq->noOfKeyAttr = addfragptr.p->noOfKeyAttr;
      tupFragReq->noOfNewAttr = addfragptr.p->noOfNewAttr;
      tupFragReq->noOfCharsets = addfragptr.p->noOfCharsets;
      tupFragReq->checksumIndicator = addfragptr.p->checksumIndicator;
      tupFragReq->noOfAttributeGroups = addfragptr.p->noOfAttributeGroups;
      tupFragReq->globalCheckpointIdIndicator = addfragptr.p->GCPIndicator;
      sendSignal(fragptr.p->tupBlockref, GSN_TUPFRAGREQ,
          signal, TupFragReq::SignalLength, JBB);
      return;
    }
    if (DictTabInfo::isOrderedIndex(addfragptr.p->tableType)) {
      jam();
      tupFragReq->userPtr = addfragptr.i;
      tupFragReq->userRef = cownref;
      tupFragReq->reqInfo = 0; /* ADD TABLE */
      tupFragReq->tableId = addfragptr.p->tabId;
      tupFragReq->noOfAttr = 1; /* ordered index: one array attr */
      tupFragReq->fragId =
        addfragptr.p->addfragStatus == AddFragRecord::WAIT_TWO_TUP
        ? addfragptr.p->fragid1 : addfragptr.p->fragid2;
      tupFragReq->maxRowsLow = addfragptr.p->maxRowsLow;
      tupFragReq->maxRowsHigh = addfragptr.p->maxRowsHigh;
      tupFragReq->minRowsLow = addfragptr.p->minRowsLow;
      tupFragReq->minRowsHigh = addfragptr.p->minRowsHigh;
      tupFragReq->noOfNullAttr = 0; /* ordered index: no nullable */
      tupFragReq->schemaVersion = addfragptr.p->schemaVer;
      tupFragReq->noOfKeyAttr = 1; /* ordered index: one key */
      tupFragReq->noOfNewAttr = addfragptr.p->noOfNewAttr;
      tupFragReq->noOfCharsets = addfragptr.p->noOfCharsets;
      tupFragReq->checksumIndicator = addfragptr.p->checksumIndicator;
      tupFragReq->noOfAttributeGroups = addfragptr.p->noOfAttributeGroups;
      tupFragReq->globalCheckpointIdIndicator = addfragptr.p->GCPIndicator;
      sendSignal(fragptr.p->tupBlockref, GSN_TUPFRAGREQ,
          signal, TupFragReq::SignalLength, JBB);
      return;
    }
  }
  if (addfragptr.p->addfragStatus == AddFragRecord::WAIT_TWO_TUX ||
      addfragptr.p->addfragStatus == AddFragRecord::WAIT_ONE_TUX) {
    if (DictTabInfo::isOrderedIndex(addfragptr.p->tableType)) {
      jam();
      TuxFragReq* const tuxreq = (TuxFragReq*)signal->getDataPtrSend();
      tuxreq->userPtr = addfragptr.i;
      tuxreq->userRef = cownref;
      tuxreq->reqInfo = 0; /* ADD TABLE */
      tuxreq->tableId = addfragptr.p->tabId;
      ndbrequire(addfragptr.p->noOfAttr >= 2);
      tuxreq->noOfAttr = addfragptr.p->noOfAttr - 1; /* skip NDB$TNODE */
      tuxreq->fragId =
        addfragptr.p->addfragStatus == AddFragRecord::WAIT_TWO_TUX
        ? addfragptr.p->fragid1: addfragptr.p->fragid2;
      tuxreq->fragOff = addfragptr.p->lh3DistrBits;
      tuxreq->tableType = addfragptr.p->tableType;
      tuxreq->primaryTableId = addfragptr.p->primaryTableId;
      // pointer to index fragment in TUP
      tuxreq->tupIndexFragPtrI =
        addfragptr.p->addfragStatus == AddFragRecord::WAIT_TWO_TUX ?
        fragptr.p->tupFragptr[0] : fragptr.p->tupFragptr[1];
      // pointers to table fragments in TUP and ACC
      FragrecordPtr tFragPtr;
      tFragPtr.i = fragptr.p->tableFragptr;
      ptrCheckGuard(tFragPtr, cfragrecFileSize, fragrecord);
      tuxreq->tupTableFragPtrI[0] = tFragPtr.p->tupFragptr[0];
      tuxreq->tupTableFragPtrI[1] = tFragPtr.p->tupFragptr[1];
      tuxreq->accTableFragPtrI[0] = tFragPtr.p->accFragptr[0];
      tuxreq->accTableFragPtrI[1] = tFragPtr.p->accFragptr[1];
      sendSignal(fragptr.p->tuxBlockref, GSN_TUXFRAGREQ,
          signal, TuxFragReq::SignalLength, JBB);
      return;
    }
  }
  ndbrequire(false);
}//Dblqh::sendAddFragReq

/* ************************************************************************> */
/*  LQHADDATTRREQ: Request from DICT to create attributes for the new table. */
/* ************************************************************************> */
void Dblqh::execLQHADDATTREQ(Signal* signal) 
{
  jamEntry();
  LqhAddAttrReq * const req = (LqhAddAttrReq*)signal->getDataPtr();
  
  addfragptr.i = req->lqhFragPtr;
  const Uint32 tnoOfAttr = req->noOfAttributes;
  const Uint32 senderData = req->senderData;
  const Uint32 senderAttrPtr = req->senderAttrPtr;

  ptrCheckGuard(addfragptr, caddfragrecFileSize, addFragRecord);
  ndbrequire(addfragptr.p->addfragStatus == AddFragRecord::WAIT_ADD_ATTR);
  ndbrequire((tnoOfAttr != 0) && (tnoOfAttr <= LqhAddAttrReq::MAX_ATTRIBUTES));
  addfragptr.p->totalAttrReceived += tnoOfAttr;
  ndbrequire(addfragptr.p->totalAttrReceived <= addfragptr.p->noOfAttr);

  addfragptr.p->attrReceived = tnoOfAttr;
  for (Uint32 i = 0; i < tnoOfAttr; i++) {
    addfragptr.p->attributes[i] = req->attributes[i];
  }//for
  addfragptr.p->attrSentToTup = 0;
  ndbrequire(addfragptr.p->dictConnectptr == senderData);
  addfragptr.p->m_senderAttrPtr = senderAttrPtr;
  addfragptr.p->addfragStatus = AddFragRecord::TUP_ATTR_WAIT1;
  sendAddAttrReq(signal);
}//Dblqh::execLQHADDATTREQ()

/* *********************>> */
/*  TUP_ADD_ATTCONF      > */
/* *********************>> */
void Dblqh::execTUP_ADD_ATTCONF(Signal* signal) 
{
  jamEntry();
  addfragptr.i = signal->theData[0];
  // implies that operation was released on the other side
  const bool lastAttr = signal->theData[1];
  ptrCheckGuard(addfragptr, caddfragrecFileSize, addFragRecord);
  switch (addfragptr.p->addfragStatus) {
  case AddFragRecord::TUP_ATTR_WAIT1:
    jam();
    if (lastAttr)
      addfragptr.p->tup1Connectptr = RNIL;
    addfragptr.p->addfragStatus = AddFragRecord::TUP_ATTR_WAIT2;
    sendAddAttrReq(signal);
    break;
  case AddFragRecord::TUP_ATTR_WAIT2:
    jam();
    if (lastAttr)
      addfragptr.p->tup2Connectptr = RNIL;
    if (DictTabInfo::isOrderedIndex(addfragptr.p->tableType)) {
      addfragptr.p->addfragStatus = AddFragRecord::TUX_ATTR_WAIT1;
      sendAddAttrReq(signal);
      break;
    }
    goto done_with_attr;
    break;
  case AddFragRecord::TUX_ATTR_WAIT1:
    jam();
    if (lastAttr)
      addfragptr.p->tux1Connectptr = RNIL;
    addfragptr.p->addfragStatus = AddFragRecord::TUX_ATTR_WAIT2;
    sendAddAttrReq(signal);
    break;
  case AddFragRecord::TUX_ATTR_WAIT2:
    jam();
    if (lastAttr)
      addfragptr.p->tux2Connectptr = RNIL;
    goto done_with_attr;
    break;
  done_with_attr:
    addfragptr.p->attrSentToTup = addfragptr.p->attrSentToTup + 1;
    ndbrequire(addfragptr.p->attrSentToTup <= addfragptr.p->attrReceived);
    ndbrequire(addfragptr.p->totalAttrReceived <= addfragptr.p->noOfAttr);
    if (addfragptr.p->attrSentToTup < addfragptr.p->attrReceived) {
      // more in this batch
      jam();
      addfragptr.p->addfragStatus = AddFragRecord::TUP_ATTR_WAIT1;
      sendAddAttrReq(signal);
    } else if (addfragptr.p->totalAttrReceived < addfragptr.p->noOfAttr) {
      // more batches to receive
      jam();
      addfragptr.p->addfragStatus = AddFragRecord::WAIT_ADD_ATTR;
      LqhAddAttrConf *const conf = (LqhAddAttrConf*)signal->getDataPtrSend();
      conf->senderData = addfragptr.p->dictConnectptr;
      conf->senderAttrPtr = addfragptr.p->m_senderAttrPtr;
      conf->fragId = addfragptr.p->addFragid;
      sendSignal(addfragptr.p->dictBlockref, GSN_LQHADDATTCONF,
          signal, LqhAddAttrConf::SignalLength, JBB);
    } else {
      fragptr.i = addfragptr.p->fragmentPtr;
      ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
      /* ------------------------------------------------------------------ 
       * WE HAVE NOW COMPLETED ADDING THIS FRAGMENT. WE NOW NEED TO SET THE 
       * PROPER STATE IN FRAG_STATUS DEPENDENT ON IF WE ARE CREATING A NEW 
       * REPLICA OR IF WE ARE CREATING A TABLE. FOR FRAGMENTS IN COPY
       * PROCESS WE DO NOT WANT LOGGING ACTIVATED.      
       * ----------------------------------------------------------------- */
      if (addfragptr.p->fragCopyCreation == 1) {
        jam();
        if (! DictTabInfo::isOrderedIndex(addfragptr.p->tableType))
          fragptr.p->fragStatus = Fragrecord::ACTIVE_CREATION;
        else
          fragptr.p->fragStatus = Fragrecord::FSACTIVE;
        fragptr.p->logFlag = Fragrecord::STATE_FALSE;
      } else {
        jam();
        fragptr.p->fragStatus = Fragrecord::FSACTIVE;
      }//if
      LqhAddAttrConf *const conf = (LqhAddAttrConf*)signal->getDataPtrSend();
      conf->senderData = addfragptr.p->dictConnectptr;
      conf->senderAttrPtr = addfragptr.p->m_senderAttrPtr;
      conf->fragId = addfragptr.p->addFragid;
      sendSignal(addfragptr.p->dictBlockref, GSN_LQHADDATTCONF, signal, 
                 LqhAddAttrConf::SignalLength, JBB);
      releaseAddfragrec(signal);
    }//if
    break;
  default:
    ndbrequire(false);
    break;
  }
}

/* **********************>> */
/*  TUX_ADD_ATTRCONF      > */
/* **********************>> */
void Dblqh::execTUX_ADD_ATTRCONF(Signal* signal) 
{
  jamEntry();
  execTUP_ADD_ATTCONF(signal);
}//Dblqh::execTUX_ADD_ATTRCONF

/*
 * Add attribute in TUP or TUX.  Called up to 4 times.
 */
void
Dblqh::sendAddAttrReq(Signal* signal)
{
  arrGuard(addfragptr.p->attrSentToTup, LqhAddAttrReq::MAX_ATTRIBUTES);
  LqhAddAttrReq::Entry& entry =
    addfragptr.p->attributes[addfragptr.p->attrSentToTup];
  const Uint32 attrId = entry.attrId & 0xffff;
  const Uint32 primaryAttrId = entry.attrId >> 16;
  fragptr.i = addfragptr.p->fragmentPtr;
  ptrCheckGuard(fragptr, cfragrecFileSize, fragrecord);
  if (addfragptr.p->addfragStatus == AddFragRecord::TUP_ATTR_WAIT1 ||
      addfragptr.p->addfragStatus == AddFragRecord::TUP_ATTR_WAIT2) {
    if (DictTabInfo::isTable(addfragptr.p->tableType) ||
        DictTabInfo::isHashIndex(addfragptr.p->tableType) ||
        (DictTabInfo::isOrderedIndex(addfragptr.p->tableType) &&
         primaryAttrId == ZNIL)) {
      jam();
      TupAddAttrReq* const tupreq = (TupAddAttrReq*)signal->getDataPtrSend();
      tupreq->tupConnectPtr =
        addfragptr.p->addfragStatus == AddFragRecord::TUP_ATTR_WAIT1
        ? addfragptr.p->tup1Connectptr : addfragptr.p->tup2Connectptr;
      tupreq->notused1 = 0;
      tupreq->attrId = attrId;
      tupreq->attrDescriptor = entry.attrDescriptor;
      tupreq->extTypeInfo = entry.extTypeInfo;
      sendSignal(fragptr.p->tupBlockref, GSN_TUP_ADD_ATTRREQ,
          signal, TupAddAttrReq::SignalLength, JBB);
      return;
    }