Below is the list of changes that have just been committed into a local
5.1 repository of tomas. When tomas does a push these changes will
be propagated to the main repository and, within 24 hours after the
push, to the public repository.
For information on how to access the public repository
see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html
ChangeSet@stripped, 2007-03-25 16:44:44+02:00, tomas@stripped +1 -0
Merge whalegate.ndb.mysql.com:/home/tomas/mysql-5.0-ndb
into whalegate.ndb.mysql.com:/home/tomas/mysql-5.1-new-ndb
MERGE: 1.1810.2374.92
storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp@stripped, 2007-03-25 16:44:40+02:00, tomas@stripped +0 -0
Auto merged
MERGE: 1.73.46.2
storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp@stripped, 2007-03-25 16:44:40+02:00, tomas@stripped +0 -0
Merge rename: ndb/src/kernel/blocks/dbtc/DbtcMain.cpp -> storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp
# This is a BitKeeper patch. What follows are the unified diffs for the
# set of deltas contained in the patch. The rest of the patch, the part
# that BitKeeper cares about, is below these diffs.
# User: tomas
# Host: whalegate.ndb.mysql.com
# Root: /home/tomas/mysql-5.1-new-ndb/RESYNC
--- 1.73.46.1/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp 2007-03-25 16:44:03 +02:00
+++ 1.146/storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp 2007-03-25 16:44:40 +02:00
@@ -37,6 +37,7 @@
#include <signaldata/TcContinueB.hpp>
#include <signaldata/TcKeyFailConf.hpp>
#include <signaldata/AbortAll.hpp>
+#include <signaldata/DihFragCount.hpp>
#include <signaldata/ScanFrag.hpp>
#include <signaldata/ScanTab.hpp>
#include <signaldata/PrepDropTab.hpp>
@@ -70,6 +71,8 @@
#include <NdbOut.hpp>
#include <DebuggerNames.hpp>
+#include <signaldata/RouteOrd.hpp>
+
// Use DEBUG to print messages that should be
// seen only when we debug the product
#ifdef VM_TRACE
@@ -345,7 +348,7 @@
tabptr.p->noOfKeyAttr = desc->noOfKeyAttr;
tabptr.p->hasCharAttr = desc->hasCharAttr;
tabptr.p->noOfDistrKeys = desc->noOfDistrKeys;
-
+ tabptr.p->hasVarKeys = desc->noOfVarKeys > 0;
signal->theData[0] = tabptr.i;
signal->theData[1] = retPtr;
sendSignal(retRef, GSN_TC_SCHVERCONF, signal, 2, JBB);
@@ -611,7 +614,7 @@
jamEntry();
const ndb_mgm_configuration_iterator * p =
- theConfiguration.getOwnConfigIterator();
+ m_ctx.m_config.getOwnConfigIterator();
ndbrequire(p != 0);
initData();
@@ -1297,6 +1300,7 @@
apiConnectptr.p->firstTcConnect == RNIL))
{
jam(); /* JUST REPLY OK */
+ apiConnectptr.p->m_transaction_nodes.clear();
releaseApiCon(signal, apiConnectptr.i);
signal->theData[0] = tuserpointer;
sendSignal(tapiBlockref,
@@ -2318,14 +2322,15 @@
{
Uint64 Tmp[MAX_KEY_SIZE_IN_WORDS * MAX_XFRM_MULTIPLY];
const TableRecord* tabPtrP = &tableRecord[tabPtrI];
+ const bool hasVarKeys = tabPtrP->hasVarKeys;
const bool hasCharAttr = tabPtrP->hasCharAttr;
- const bool hasDistKeys = tabPtrP->noOfDistrKeys > 0;
+ const bool compute_distkey = distr && (tabPtrP->noOfDistrKeys > 0);
Uint32 *dst = (Uint32*)Tmp;
Uint32 dstPos = 0;
Uint32 keyPartLen[MAX_ATTRIBUTES_IN_INDEX];
Uint32 * keyPartLenPtr;
- if(hasCharAttr)
+ if(hasCharAttr || (compute_distkey && hasVarKeys))
{
keyPartLenPtr = keyPartLen;
dstPos = xfrm_key(tabPtrI, src, dst, sizeof(Tmp) >> 2, keyPartLenPtr);
@@ -2343,7 +2348,7 @@
md5_hash(dstHash, (Uint64*)dst, dstPos);
- if(distr && hasDistKeys)
+ if(compute_distkey)
{
jam();
@@ -2542,7 +2547,7 @@
ApiConnectRecord * const regApiPtr = &localApiConnectRecord[TapiIndex];
apiConnectptr.p = regApiPtr;
- Uint32 TstartFlag = tcKeyReq->getStartFlag(Treqinfo);
+ Uint32 TstartFlag = TcKeyReq::getStartFlag(Treqinfo);
Uint32 TexecFlag = TcKeyReq::getExecuteFlag(Treqinfo);
Uint8 isIndexOp = regApiPtr->isIndexOp;
@@ -2723,14 +2728,14 @@
/* */
/* ---------------------------------------------------------------------- */
- UintR TapiVersionNo = tcKeyReq->getAPIVersion(tcKeyReq->attrLen);
+ UintR TapiVersionNo = TcKeyReq::getAPIVersion(tcKeyReq->attrLen);
UintR Tlqhkeyreqrec = regApiPtr->lqhkeyreqrec;
regApiPtr->lqhkeyreqrec = Tlqhkeyreqrec + 1;
regCachePtr->apiVersionNo = TapiVersionNo;
UintR TapiConnectptrIndex = apiConnectptr.i;
UintR TsenderData = tcKeyReq->senderData;
- UintR TattrLen = tcKeyReq->getAttrinfoLen(tcKeyReq->attrLen);
+ UintR TattrLen = TcKeyReq::getAttrinfoLen(tcKeyReq->attrLen);
UintR TattrinfoCount = c_counters.cattrinfoCount;
regTcPtr->apiConnect = TapiConnectptrIndex;
@@ -2758,21 +2763,23 @@
UintR TtabptrIndex = localTabptr.i;
UintR TtableSchemaVersion = tcKeyReq->tableSchemaVersion;
- Uint8 TOperationType = tcKeyReq->getOperationType(Treqinfo);
+ Uint8 TOperationType = TcKeyReq::getOperationType(Treqinfo);
regCachePtr->tableref = TtabptrIndex;
regCachePtr->schemaVersion = TtableSchemaVersion;
regTcPtr->operation = TOperationType;
- Uint8 TSimpleFlag = tcKeyReq->getSimpleFlag(Treqinfo);
- Uint8 TDirtyFlag = tcKeyReq->getDirtyFlag(Treqinfo);
- Uint8 TInterpretedFlag = tcKeyReq->getInterpretedFlag(Treqinfo);
- Uint8 TDistrKeyFlag = tcKeyReq->getDistributionKeyFlag(Treqinfo);
+ Uint8 TSimpleFlag = TcKeyReq::getSimpleFlag(Treqinfo);
+ Uint8 TDirtyFlag = TcKeyReq::getDirtyFlag(Treqinfo);
+ Uint8 TInterpretedFlag = TcKeyReq::getInterpretedFlag(Treqinfo);
+ Uint8 TDistrKeyFlag = TcKeyReq::getDistributionKeyFlag(Treqinfo);
+ Uint8 TNoDiskFlag = TcKeyReq::getNoDiskFlag(Treqinfo);
Uint8 TexecuteFlag = TexecFlag;
regCachePtr->opSimple = TSimpleFlag;
regCachePtr->opExec = TInterpretedFlag;
regTcPtr->dirtyOp = TDirtyFlag;
regCachePtr->distributionKeyIndicator = TDistrKeyFlag;
+ regCachePtr->m_no_disk_flag = TNoDiskFlag;
//-------------------------------------------------------------
// The next step is to read the upto three conditional words.
@@ -2780,10 +2787,10 @@
Uint32 TkeyIndex;
Uint32* TOptionalDataPtr = (Uint32*)&tcKeyReq->scanInfo;
{
- Uint32 TDistrGHIndex = tcKeyReq->getScanIndFlag(Treqinfo);
+ Uint32 TDistrGHIndex = TcKeyReq::getScanIndFlag(Treqinfo);
Uint32 TDistrKeyIndex = TDistrGHIndex;
- Uint32 TscanInfo = tcKeyReq->getTakeOverScanInfo(TOptionalDataPtr[0]);
+ Uint32 TscanInfo = TcKeyReq::getTakeOverScanInfo(TOptionalDataPtr[0]);
regCachePtr->scanTakeOverInd = TDistrGHIndex;
regCachePtr->scanInfo = TscanInfo;
@@ -2805,7 +2812,7 @@
regCachePtr->keydata[2] = Tdata3;
regCachePtr->keydata[3] = Tdata4;
- TkeyLength = tcKeyReq->getKeyLength(Treqinfo);
+ TkeyLength = TcKeyReq::getKeyLength(Treqinfo);
Uint32 TAIDataIndex;
if (TkeyLength > 8) {
TAIDataIndex = TkeyIndex + 8;
@@ -2818,7 +2825,7 @@
}//if
Uint32* TAIDataPtr = &TOptionalDataPtr[TAIDataIndex];
- titcLenAiInTckeyreq = tcKeyReq->getAIInTcKeyReq(Treqinfo);
+ titcLenAiInTckeyreq = TcKeyReq::getAIInTcKeyReq(Treqinfo);
regCachePtr->keylen = TkeyLength;
regCachePtr->lenAiInTckeyreq = titcLenAiInTckeyreq;
regCachePtr->currReclenAi = titcLenAiInTckeyreq;
@@ -2836,17 +2843,9 @@
regCachePtr->attrinfo15[2] = Tdata4;
regCachePtr->attrinfo15[3] = Tdata5;
- if (TOperationType == ZREAD) {
- Uint32 TreadCount = c_counters.creadCount;
- jam();
- regCachePtr->opLock = 0;
- c_counters.creadCount = TreadCount + 1;
- } else if(TOperationType == ZREAD_EX){
+ if (TOperationType == ZREAD || TOperationType == ZREAD_EX) {
Uint32 TreadCount = c_counters.creadCount;
jam();
- TOperationType = ZREAD;
- regTcPtr->operation = ZREAD;
- regCachePtr->opLock = ZUPDATE;
c_counters.creadCount = TreadCount + 1;
} else {
if(regApiPtr->commitAckMarker == RNIL){
@@ -2863,6 +2862,12 @@
tmp.p->apiNodeId = refToNode(regApiPtr->ndbapiBlockref);
tmp.p->apiConnectPtr = TapiIndex;
tmp.p->noOfLqhs = 0;
+#if defined VM_TRACE || defined ERROR_INSERT
+ {
+ CommitAckMarkerPtr check;
+ ndbrequire(!m_commitAckMarkerHash.find(check, *tmp.p));
+ }
+#endif
m_commitAckMarkerHash.add(tmp);
}
}
@@ -2880,24 +2885,10 @@
c_counters.cwriteCount = TwriteCount + 1;
switch (TOperationType) {
case ZUPDATE:
- jam();
- if (TattrLen == 0) {
- //TCKEY_abort(signal, 5);
- //return;
- }//if
- /*---------------------------------------------------------------------*/
- // The missing break is intentional since we also want to set the opLock
- // variable also for updates
- /*---------------------------------------------------------------------*/
case ZINSERT:
case ZDELETE:
- jam();
- regCachePtr->opLock = TOperationType;
- break;
case ZWRITE:
jam();
- // A write operation is originally an insert operation.
- regCachePtr->opLock = ZINSERT;
break;
default:
TCKEY_abort(signal, 9);
@@ -2905,14 +2896,14 @@
}//switch
}//if
- Uint32 TabortOption = tcKeyReq->getAbortOption(Treqinfo);
+ Uint32 TabortOption = TcKeyReq::getAbortOption(Treqinfo);
regTcPtr->m_execAbortOption = TabortOption;
/*-------------------------------------------------------------------------
* Check error handling per operation
* If CommitFlag is set state accordingly and check for early abort
*------------------------------------------------------------------------*/
- if (tcKeyReq->getCommitFlag(Treqinfo) == 1) {
+ if (TcKeyReq::getCommitFlag(Treqinfo) == 1) {
ndbrequire(TexecuteFlag);
regApiPtr->apiConnectstate = CS_REC_COMMITTING;
} else {
@@ -3072,7 +3063,7 @@
tnoOfStandby = (tnodeinfo >> 8) & 3;
regCachePtr->fragmentDistributionKey = (tnodeinfo >> 16) & 255;
- if (Toperation == ZREAD) {
+ if (Toperation == ZREAD || Toperation == ZREAD_EX) {
if (Tdirty == 1) {
jam();
/*-------------------------------------------------------------*/
@@ -3105,28 +3096,7 @@
}//if
}//for
}
-
- if (regTcPtr->tcNodedata[0] != getOwnNodeId())
- {
- jam();
- for (Uint32 i = 0; i < tnoOfBackup + 1; i++)
- {
- HostRecordPtr hostPtr;
- hostPtr.i = regTcPtr->tcNodedata[i];
- ptrCheckGuard(hostPtr, chostFilesize, hostRecord);
- if (hostPtr.p->m_nf_bits & HostRecord::NF_STARTED)
- {
- jam();
- if (i != 0)
- {
- jam();
- regTcPtr->tcNodedata[0] = hostPtr.i;
- }
- break;
- }
- }
- }//if
- }
+ }//if
jam();
regTcPtr->lastReplicaNo = 0;
regTcPtr->noOfNodes = 1;
@@ -3222,6 +3192,7 @@
TcConnectRecord * const regTcPtr = tcConnectptr.p;
ApiConnectRecord * const regApiPtr = apiConnectptr.p;
CacheRecord * const regCachePtr = cachePtr.p;
+ Uint32 version = getNodeInfo(refToNode(TBRef)).m_version;
UintR sig0, sig1, sig2, sig3, sig4, sig5, sig6;
#ifdef ERROR_INSERT
if (ERROR_INSERTED(8002)) {
@@ -3265,7 +3236,12 @@
bool simpleRead = (sig1 == ZREAD && sig0 == ZTRUE);
LqhKeyReq::setKeyLen(Tdata10, regCachePtr->keylen);
LqhKeyReq::setLastReplicaNo(Tdata10, regTcPtr->lastReplicaNo);
- LqhKeyReq::setLockType(Tdata10, regCachePtr->opLock);
+ if (unlikely(version < NDBD_ROWID_VERSION))
+ {
+ Uint32 op = regTcPtr->operation;
+ Uint32 lock = (Operation_t) op == ZREAD_EX ? ZUPDATE : (Operation_t) op == ZWRITE ? ZINSERT : (Operation_t) op;
+ LqhKeyReq::setLockType(Tdata10, lock);
+ }
/* ---------------------------------------------------------------------- */
// Indicate Application Reference is present in bit 15
/* ---------------------------------------------------------------------- */
@@ -3274,6 +3250,8 @@
LqhKeyReq::setInterpretedFlag(Tdata10, regCachePtr->opExec);
LqhKeyReq::setSimpleFlag(Tdata10, sig0);
LqhKeyReq::setOperation(Tdata10, sig1);
+ LqhKeyReq::setNoDiskFlag(Tdata10, regCachePtr->m_no_disk_flag);
+
/* -----------------------------------------------------------------------
* Sequential Number of first LQH = 0, bit 22-23
* IF ATTRIBUTE INFORMATION IS SENT IN TCKEYREQ,
@@ -3991,7 +3969,7 @@
const UintR TopWords = (UintR)regApiPtr->tckeyrec;
localHostptr.i = refToNode(regApiPtr->ndbapiBlockref);
const Uint32 type = getNodeInfo(localHostptr.i).m_type;
- const bool is_api = (type >= NodeInfo::API && type <= NodeInfo::REP);
+ const bool is_api = (type >= NodeInfo::API && type <= NodeInfo::MGM);
const BlockNumber TblockNum = refToBlock(regApiPtr->ndbapiBlockref);
const Uint32 Tmarker = (regApiPtr->commitAckMarker == RNIL) ? 0 : 1;
ptrAss(localHostptr, hostRecord);
@@ -4673,7 +4651,8 @@
commitConf->transId1 = regApiPtr->transid[0];
commitConf->transId2 = regApiPtr->transid[1];
commitConf->gci = regApiPtr->globalcheckpointid;
- sendSignal(regApiPtr->ndbapiBlockref, GSN_TC_COMMITCONF, signal,
+
+ sendSignal(regApiPtr->ndbapiBlockref, GSN_TC_COMMITCONF, signal,
TcCommitConf::SignalLength, JBB);
} else if (regApiPtr->returnsignal == RS_NO_RETURN) {
jam();
@@ -4865,13 +4844,14 @@
key.transid2 = signal->theData[1];
CommitAckMarkerPtr removedMarker;
- m_commitAckMarkerHash.release(removedMarker, key);
+ m_commitAckMarkerHash.remove(removedMarker, key);
if (removedMarker.i == RNIL) {
jam();
warningHandlerLab(signal, __LINE__);
return;
}//if
sendRemoveMarkers(signal, removedMarker.p);
+ m_commitAckMarkerPool.release(removedMarker);
}
void
@@ -5216,6 +5196,19 @@
return;
}
+ /* Only ref in certain situations */
+ {
+ const Uint32 opType = regTcPtr->operation;
+ if ( (opType == ZDELETE && errCode != ZNOT_FOUND)
+ || (opType == ZINSERT && errCode != ZALREADYEXIST)
+ || (opType == ZUPDATE && errCode != ZNOT_FOUND)
+ || (opType == ZWRITE && errCode != 839 && errCode != 840))
+ {
+ TCKEY_abort(signal, 49);
+ return;
+ }
+ }
+
/* *************** */
/* TCKEYREF < */
/* *************** */
@@ -7130,19 +7123,6 @@
}//Dbtc::execNODE_FAILREP()
void
-Dbtc::execNODE_START_REP(Signal* signal)
-{
- Uint32 nodeId = signal->theData[0];
- hostptr.i = nodeId;
- ptrCheckGuard(hostptr, chostFilesize, hostRecord);
- if (hostptr.p->m_nf_bits == 0)
- {
- jam();
- hostptr.p->m_nf_bits |= HostRecord::NF_STARTED;
- }
-}
-
-void
Dbtc::checkNodeFailComplete(Signal* signal,
Uint32 failedNodeId,
Uint32 bit)
@@ -7233,7 +7213,7 @@
if (transPtr.p->m_transaction_nodes.get(failedNodeId))
{
jam();
-
+
// Force timeout regardless of state
c_appl_timeout_value = 1;
setApiConTimer(transPtr.i, TtcTimer - 2, __LINE__);
@@ -8241,6 +8221,13 @@
tmp.p->noOfLqhs = 1;
tmp.p->lqhNodeId[0] = tnodeid;
tmp.p->apiConnectPtr = apiConnectptr.i;
+
+#if defined VM_TRACE || defined ERROR_INSERT
+ {
+ CommitAckMarkerPtr check;
+ ndbrequire(!m_commitAckMarkerHash.find(check, *tmp.p));
+ }
+#endif
m_commitAckMarkerHash.add(tmp);
}
}//Dbtc::initApiConnectFail()
@@ -8397,6 +8384,12 @@
tmp.p->noOfLqhs = 1;
tmp.p->lqhNodeId[0] = tnodeid;
tmp.p->apiConnectPtr = apiConnectptr.i;
+#if defined VM_TRACE || defined ERROR_INSERT
+ {
+ CommitAckMarkerPtr check;
+ ndbrequire(!m_commitAckMarkerHash.find(check, *tmp.p));
+ }
+#endif
m_commitAckMarkerHash.add(tmp);
} else {
jam();
@@ -8941,6 +8934,7 @@
ScanFragReq::setDescendingFlag(tmp, ScanTabReq::getDescendingFlag(ri));
ScanFragReq::setTupScanFlag(tmp, ScanTabReq::getTupScanFlag(ri));
ScanFragReq::setAttrLen(tmp, scanTabReq->attrLenKeyLen & 0xFFFF);
+ ScanFragReq::setNoDiskFlag(tmp, ScanTabReq::getNoDiskFlag(ri));
scanptr.p->scanRequestInfo = tmp;
scanptr.p->scanStoredProcId = scanTabReq->storedProcId;
@@ -9062,9 +9056,11 @@
* THE FIRST STEP TO RECEIVE IS SUCCESSFULLY COMPLETED.
* WE MUST FIRST GET THE NUMBER OF FRAGMENTS IN THE TABLE.
***************************************************/
- signal->theData[0] = tcConnectptr.p->dihConnectptr;
- signal->theData[1] = scanptr.p->scanTableref;
- sendSignal(cdihblockref, GSN_DI_FCOUNTREQ, signal, 2, JBB);
+ DihFragCountReq * const req = (DihFragCountReq*)signal->getDataPtrSend();
+ req->m_connectionData = tcConnectptr.p->dihConnectptr;
+ req->m_tableRef = scanptr.p->scanTableref;
+ sendSignal(cdihblockref, GSN_DI_FCOUNTREQ, signal,
+ DihFragCountReq::SignalLength, JBB);
}
else
{
@@ -9075,17 +9071,18 @@
UintR TerrorIndicator = signal->theData[0];
jamEntry();
if (TerrorIndicator != 0) {
- signal->theData[0] = tcConnectptr.i;
- //signal->theData[1] Contains error
+ DihFragCountRef * const ref = (DihFragCountRef*)signal->getDataPtr();
+ ref->m_connectionData = tcConnectptr.i;
+ ref->m_error = signal->theData[1];
execDI_FCOUNTREF(signal);
return;
}
UintR Tdata1 = signal->theData[1];
scanptr.p->scanNextFragId = Tdata1;
-
- signal->theData[0] = tcConnectptr.i;
- signal->theData[1] = 1; // Frag count
+ DihFragCountConf * const conf = (DihFragCountConf*)signal->getDataPtr();
+ conf->m_connectionData = tcConnectptr.i;
+ conf->m_fragmentCount = 1; // Frag count
execDI_FCOUNTCONF(signal);
}
return;
@@ -9103,8 +9100,9 @@
void Dbtc::execDI_FCOUNTCONF(Signal* signal)
{
jamEntry();
- tcConnectptr.i = signal->theData[0];
- Uint32 tfragCount = signal->theData[1];
+ DihFragCountConf * const conf = (DihFragCountConf*)signal->getDataPtr();
+ tcConnectptr.i = conf->m_connectionData;
+ Uint32 tfragCount = conf->m_fragmentCount;
ptrCheckGuard(tcConnectptr, ctcConnectFilesize, tcConnectRecord);
apiConnectptr.i = tcConnectptr.p->apiConnect;
ptrCheckGuard(apiConnectptr, capiConnectFilesize, apiConnectRecord);
@@ -9188,9 +9186,10 @@
void Dbtc::execDI_FCOUNTREF(Signal* signal)
{
jamEntry();
- tcConnectptr.i = signal->theData[0];
+ DihFragCountRef * const ref = (DihFragCountRef*)signal->getDataPtr();
+ tcConnectptr.i = ref->m_connectionData;
ptrCheckGuard(tcConnectptr, ctcConnectFilesize, tcConnectRecord);
- const Uint32 errCode = signal->theData[1];
+ const Uint32 errCode = ref->m_error;
apiConnectptr.i = tcConnectptr.p->apiConnect;
ptrCheckGuard(apiConnectptr, capiConnectFilesize, apiConnectRecord);
ScanRecordPtr scanptr;
@@ -10350,6 +10349,7 @@
tabptr.p->noOfKeyAttr = 0;
tabptr.p->hasCharAttr = 0;
tabptr.p->noOfDistrKeys = 0;
+ tabptr.p->hasVarKeys = 0;
}//for
}//Dbtc::initTable()
@@ -11385,7 +11385,6 @@
ApiConnectRecordPtr transPtr;
TcConnectRecord *localTcConnectRecord = tcConnectRecord;
TcConnectRecordPtr opPtr;
-
/**
* TODO
* Check transid,
@@ -11399,6 +11398,7 @@
c_firedTriggerHash.remove(trigPtr);
+ trigPtr.p->fragId= fireOrd->fragId;
bool ok = trigPtr.p->keyValues.getSize() == fireOrd->m_noPrimKeyWords;
ok &= trigPtr.p->afterValues.getSize() == fireOrd->m_noAfterValueWords;
ok &= trigPtr.p->beforeValues.getSize() == fireOrd->m_noBeforeValueWords;
@@ -11587,7 +11587,7 @@
// If operation is readTupleExclusive or updateTuple then read index
// table with exclusive lock
Uint32 indexLength = TcKeyReq::getKeyLength(tcIndxRequestInfo);
- Uint32 attrLength = tcIndxReq->attrLen;
+ Uint32 attrLength = TcKeyReq::getAttrinfoLen(tcIndxReq->attrLen);
indexOp->expectedKeyInfo = indexLength;
Uint32 includedIndexLength = MIN(indexLength, indexBufSize);
indexOp->expectedAttrInfo = attrLength;
@@ -11621,7 +11621,7 @@
const UintR TopWords = (UintR)regApiPtr->tcindxrec;
localHostptr.i = refToNode(regApiPtr->ndbapiBlockref);
const Uint32 type = getNodeInfo(localHostptr.i).m_type;
- const bool is_api = (type >= NodeInfo::API && type <= NodeInfo::REP);
+ const bool is_api = (type >= NodeInfo::API && type <= NodeInfo::MGM);
const BlockNumber TblockNum = refToBlock(regApiPtr->ndbapiBlockref);
const Uint32 Tmarker = (regApiPtr->commitAckMarker == RNIL ? 0 : 1);
ptrAss(localHostptr, hostRecord);
@@ -12318,7 +12318,11 @@
Uint32 dataPos = 0;
TcKeyReq * const tcIndxReq = &indexOp->tcIndxReq;
TcKeyReq * const tcKeyReq = (TcKeyReq *)signal->getDataPtrSend();
- Uint32 * dataPtr = &tcKeyReq->scanInfo;
+ /*
+ Data points to distrGroupHashValue since scanInfo is used to send
+ fragment id of receiving fragment
+ */
+ Uint32 * dataPtr = &tcKeyReq->distrGroupHashValue;
Uint32 tcKeyLength = TcKeyReq::StaticLength;
Uint32 tcKeyRequestInfo = tcIndxReq->requestInfo;
TcIndexData* indexData;
@@ -12357,11 +12361,16 @@
regApiPtr->executingIndexOp = indexOp->indexOpId;;
regApiPtr->noIndexOp++; // Increase count
- // Filter out AttributeHeader:s since this should not be in key
+ /*
+ Filter out AttributeHeader:s since this should not be in key.
+ Also filter out fragment id from primary key and handle that
+ separately by setting it as Distribution Key and set indicator.
+ */
+
AttributeHeader* attrHeader = (AttributeHeader *) aiIter.data;
Uint32 headerSize = attrHeader->getHeaderSize();
- Uint32 keySize = attrHeader->getDataSize();
+ Uint32 keySize = attrHeader->getDataSize() - 1;
TcKeyReq::setKeyLength(tcKeyRequestInfo, keySize);
// Skip header
if (headerSize == 1) {
@@ -12371,6 +12380,9 @@
jam();
moreKeyData = indexOp->transIdAI.next(aiIter, headerSize - 1);
}//if
+ tcKeyReq->scanInfo = *aiIter.data; //Fragment Id
+ moreKeyData = indexOp->transIdAI.next(aiIter);
+ TcKeyReq::setDistributionKeyFlag(tcKeyRequestInfo, 1U);
while(// If we have not read complete key
(keySize != 0) &&
(dataPos < keyBufSize)) {
@@ -12636,7 +12648,7 @@
tmp2.release();
LocalDataBuffer<11> tmp3(pool, trigPtr.p->afterValues);
tmp3.release();
- regApiPtr->theFiredTriggers.release(trigPtr.i);
+ regApiPtr->theFiredTriggers.release(trigPtr);
}
trigPtr = nextTrigPtr;
}
@@ -12753,7 +12765,7 @@
AttributeBuffer::DataBufferIterator iter;
Uint32 attrId = 0;
Uint32 keyLength = 0;
- Uint32 totalPrimaryKeyLength = 0;
+ Uint32 totalPrimaryKeyLength = 1; // fragment length
Uint32 hops;
indexTabPtr.i = indexData->indexId;
@@ -12806,11 +12818,12 @@
hops = attrHeader->getHeaderSize() + attrHeader->getDataSize();
moreAttrData = keyValues.next(iter, hops);
}
- AttributeHeader pkAttrHeader(attrId, totalPrimaryKeyLength);
+ AttributeHeader pkAttrHeader(attrId, totalPrimaryKeyLength << 2);
+ Uint32 attributesLength = afterValues.getSize() +
+ pkAttrHeader.getHeaderSize() + pkAttrHeader.getDataSize();
TcKeyReq::setKeyLength(tcKeyRequestInfo, keyLength);
- tcKeyReq->attrLen = afterValues.getSize() +
- pkAttrHeader.getHeaderSize() + pkAttrHeader.getDataSize();
+ tcKeyReq->attrLen = attributesLength;
tcKeyReq->tableId = indexData->indexId;
TcKeyReq::setOperationType(tcKeyRequestInfo, ZINSERT);
TcKeyReq::setExecutingTrigger(tcKeyRequestInfo, true);
@@ -12860,8 +12873,11 @@
}
tcKeyLength += dataPos;
- Uint32 attributesLength = afterValues.getSize() +
- pkAttrHeader.getHeaderSize() + pkAttrHeader.getDataSize();
+ /*
+ Size of attrinfo is unique index attributes one by one, header for each
+ of them (all contained in the afterValues data structure), plus a header,
+ the primary key (compacted) and the fragment id before the primary key
+ */
if (attributesLength <= attrBufSize) {
jam();
// ATTRINFO fits in TCKEYREQ
@@ -12878,6 +12894,10 @@
// as one attribute
pkAttrHeader.insertHeader(dataPtr);
dataPtr += pkAttrHeader.getHeaderSize();
+ /*
+ Insert fragment id before primary key as part of reference to tuple
+ */
+ *dataPtr++ = firedTriggerData->fragId;
moreAttrData = keyValues.first(iter);
while(moreAttrData) {
jam();
@@ -13073,6 +13093,29 @@
pkAttrHeader.insertHeader(dataPtr);
dataPtr += pkAttrHeader.getHeaderSize();
attrInfoPos += pkAttrHeader.getHeaderSize();
+ /*
+ Add fragment id before primary key
+ TODO: This code really needs to be made into a long signal
+ to remove this messy code.
+ */
+ if (attrInfoPos == AttrInfo::DataLength)
+ {
+ jam();
+ // Flush ATTRINFO
+#if INTERNAL_TRIGGER_TCKEYREQ_JBA
+ sendSignal(reference(), GSN_ATTRINFO, signal,
+ AttrInfo::HeaderLength + AttrInfo::DataLength, JBA);
+#else
+ EXECUTE_DIRECT(DBTC, GSN_ATTRINFO, signal,
+ AttrInfo::HeaderLength + AttrInfo::DataLength);
+ jamEntry();
+#endif
+ dataPtr = (Uint32 *) &attrInfo->attrData;
+ attrInfoPos = 0;
+ }
+ attrInfoPos++;
+ *dataPtr++ = firedTriggerData->fragId;
+
moreAttrData = keyValues.first(iter);
while(moreAttrData) {
jam();
@@ -13373,3 +13416,55 @@
return 0;
}
+void
+Dbtc::execROUTE_ORD(Signal* signal)
+{
+ jamEntry();
+ if(!assembleFragments(signal)){
+ jam();
+ return;
+ }
+
+ RouteOrd* ord = (RouteOrd*)signal->getDataPtr();
+ Uint32 dstRef = ord->dstRef;
+ Uint32 srcRef = ord->srcRef;
+ Uint32 gsn = ord->gsn;
+
+ if (likely(getNodeInfo(refToNode(dstRef)).m_connected))
+ {
+ jam();
+ Uint32 secCount = signal->getNoOfSections();
+ SegmentedSectionPtr ptr[3];
+ ndbrequire(secCount >= 1 && secCount <= 3);
+
+ jamLine(secCount);
+ for (Uint32 i = 0; i<secCount; i++)
+ signal->getSection(ptr[i], i);
+
+ /**
+ * Put section 0 in signal->theData
+ */
+ ndbrequire(ptr[0].sz <= 25);
+ copy(signal->theData, ptr[0]);
+
+ signal->header.m_noOfSections = 0;
+
+ /**
+ * Shift rest of sections
+ */
+ for(Uint32 i = 1; i<secCount; i++)
+ {
+ signal->setSection(ptr[i], i - 1);
+ }
+
+ sendSignal(dstRef, gsn, signal, ptr[0].sz, JBB);
+
+ signal->header.m_noOfSections = 0;
+ signal->setSection(ptr[0], 0);
+ releaseSections(signal);
+ return ;
+ }
+
+ warningEvent("Unable to route GSN: %d from %x to %x",
+ gsn, srcRef, dstRef);
+}
| Thread |
|---|
| • bk commit into 5.1 tree (tomas:1.2518) | tomas | 25 Mar |
