dblqhmain.cpp

mysql-5.0.22.tar.gz源码包

    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) {
    if (DictTabInfo::isTable(addfragptr.p->tableType) ||
        DictTabInfo::isHashIndex(addfragptr.p->tableType)) {
      jam();
      signal->theData[0] = addfragptr.i;
      signal->theData[1] = cownref;
      signal->theData[2] = 0; /* ADD TABLE */
      signal->theData[3] = addfragptr.p->tabId;
      signal->theData[4] = addfragptr.p->noOfAttr;
      signal->theData[5] =
        addfragptr.p->addfragStatus == AddFragRecord::WAIT_TWO_TUP
        ? addfragptr.p->fragid1 : addfragptr.p->fragid2;
      signal->theData[6] = (addfragptr.p->noOfAllocPages >> 1) + 1;
      signal->theData[7] = addfragptr.p->noOfNull;
      signal->theData[8] = addfragptr.p->schemaVer;
      signal->theData[9] = addfragptr.p->noOfKeyAttr;
      signal->theData[10] = addfragptr.p->noOfNewAttr;
      signal->theData[11] = addfragptr.p->checksumIndicator;
      signal->theData[12] = addfragptr.p->noOfAttributeGroups;
      signal->theData[13] = addfragptr.p->GCPIndicator;
      sendSignal(fragptr.p->tupBlockref, GSN_TUPFRAGREQ,
          signal, TupFragReq::SignalLength, JBB);
      return;
    }
    if (DictTabInfo::isOrderedIndex(addfragptr.p->tableType)) {
      jam();
      signal->theData[0] = addfragptr.i;
      signal->theData[1] = cownref;
      signal->theData[2] = 0; /* ADD TABLE */
      signal->theData[3] = addfragptr.p->tabId;
      signal->theData[4] = 1; /* ordered index: one array attr */
      signal->theData[5] =
        addfragptr.p->addfragStatus == AddFragRecord::WAIT_TWO_TUP
        ? addfragptr.p->fragid1 : addfragptr.p->fragid2;
      signal->theData[6] = (addfragptr.p->noOfAllocPages >> 1) + 1;
      signal->theData[7] = 0; /* ordered index: no nullable */
      signal->theData[8] = addfragptr.p->schemaVer;
      signal->theData[9] = 1; /* ordered index: one key */
      signal->theData[10] = addfragptr.p->noOfNewAttr;
      signal->theData[11] = addfragptr.p->checksumIndicator;
      signal->theData[12] = addfragptr.p->noOfAttributeGroups;
      signal->theData[13] = 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;
    }
    if (DictTabInfo::isOrderedIndex(addfragptr.p->tableType) &&
        primaryAttrId != ZNIL) {
      // this attribute is not for TUP
      jam();
      TupAddAttrConf* tupconf = (TupAddAttrConf*)signal->getDataPtrSend();
      tupconf->userPtr = addfragptr.i;
      tupconf->lastAttr = false;
      sendSignal(reference(), GSN_TUP_ADD_ATTCONF,
          signal, TupAddAttrConf::SignalLength, JBB);
      return;
    }
  }
  if (addfragptr.p->addfragStatus == AddFragRecord::TUX_ATTR_WAIT1 ||
      addfragptr.p->addfragStatus == AddFragRecord::TUX_ATTR_WAIT2) {
    jam();
    if (DictTabInfo::isOrderedIndex(addfragptr.p->tableType) &&
        primaryAttrId != ZNIL) {
      jam();
      TuxAddAttrReq* const tuxreq = (TuxAddAttrReq*)signal->getDataPtrSend();
      tuxreq->tuxConnectPtr =
        addfragptr.p->addfragStatus == AddFragRecord::TUX_ATTR_WAIT1
        ? addfragptr.p->tux1Connectptr : addfragptr.p->tux2Connectptr;
      tuxreq->notused1 = 0;
      tuxreq->attrId = attrId;
      tuxreq->attrDescriptor = entry.attrDescriptor;
      tuxreq->extTypeInfo = entry.extTypeInfo;
      tuxreq->primaryAttrId = primaryAttrId;
      sendSignal(fragptr.p->tuxBlockref, GSN_TUX_ADD_ATTRREQ,
          signal, TuxAddAttrReq::SignalLength, JBB);
      return;
    }
    if (DictTabInfo::isOrderedIndex(addfragptr.p->tableType) &&
        primaryAttrId == ZNIL) {
      // this attribute is not for TUX
      jam();
      TuxAddAttrConf* tuxconf = (TuxAddAttrConf*)signal->getDataPtrSend();
      tuxconf->userPtr = addfragptr.i;
      tuxconf->lastAttr = false;
      sendSignal(reference(), GSN_TUX_ADD_ATTRCONF,
          signal, TuxAddAttrConf::SignalLength, JBB);
      return;
    }
  }
  ndbrequire(false);
}//Dblqh::sendAddAttrReq

/* ************************************************************************>> */
/*  TAB_COMMITREQ: Commit the new table for use in transactions. Sender DICT. */
/* ************************************************************************>> */
void Dblqh::execTAB_COMMITREQ(Signal* signal) 
{
  jamEntry();
  Uint32 dihPtr = signal->theData[0];
  BlockReference dihBlockref = signal->theData[1];
  tabptr.i = signal->theData[2];

  if (tabptr.i >= ctabrecFileSize) {
    jam();
    terrorCode = ZTAB_FILE_SIZE;
    signal->theData[0] = dihPtr;
    signal->theData[1] = cownNodeid;
    signal->theData[2] = tabptr.i;
    signal->theData[3] = terrorCode;
    sendSignal(dihBlockref, GSN_TAB_COMMITREF, signal, 4, JBB);
    return;
  }//if
  ptrAss(tabptr, tablerec);
  if (tabptr.p->tableStatus != Tablerec::ADD_TABLE_ONGOING) {
    jam();
    terrorCode = ZTAB_STATE_ERROR;
    signal->theData[0] = dihPtr;
    signal->theData[1] = cownNodeid;
    signal->theData[2] = tabptr.i;
    signal->theData[3] = terrorCode;
    signal->theData[4] = tabptr.p->tableStatus;
    sendSignal(dihBlockref, GSN_TAB_COMMITREF, signal, 5, JBB);
    ndbrequire(false);
    return;
  }//if
  tabptr.p->usageCount = 0;
  tabptr.p->tableStatus = Tablerec::TABLE_DEFINED;
  signal->theData[0] = dihPtr;
  signal->theData[1] = cownNodeid;
  signal->theData[2] = tabptr.i;
  sendSignal(dihBlockref, GSN_TAB_COMMITCONF, signal, 3, JBB);
  return;
}//Dblqh::execTAB_COMMITREQ()