Skip to content

M
mariadb
Commit 46075704 authored by unknown's avatar unknown
Browse files
Options

wl2126 - ndb - add support for batching range scans 

1) removed acc keyinfo
2) introduce handling of multiple bounds


ndb/include/kernel/signaldata/AccScan.hpp:
  Removed keyinfo from AccScan
ndb/include/kernel/signaldata/TuxBound.hpp:
  Add data word
ndb/include/ndbapi/NdbIndexScanOperation.hpp:
  Added new_bound call, which is used when batching range scans
ndb/src/kernel/blocks/dbacc/DbaccMain.cpp:
  removed keyinfo from acc scan
ndb/src/kernel/blocks/dblqh/Dblqh.hpp:
  removed scan keyinfo
ndb/src/kernel/blocks/dblqh/DblqhInit.cpp:
  removed scan keyinfo
ndb/src/kernel/blocks/dblqh/DblqhMain.cpp:
  1) removed add scan,
     let LQH read keyinfo from tup instead
  2) introduce copy_bounds which handles multiple bounds
ndb/src/kernel/blocks/dbtux/Dbtux.hpp:
  removed scan keyinfo
ndb/src/kernel/blocks/dbtux/DbtuxDebug.cpp:
  removed scan keyinfo
ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp:
  removed scan keyinfo
ndb/src/ndbapi/NdbScanOperation.cpp:
  handle multiple bounds
parent 3ef57e28
Hide whitespace changes
Inline Side-by-side
Showing with 265 additions and 263 deletions
ndb/include/kernel/signaldata/AccScan.hpp
View file @ 46075704
...@@ -51,11 +51,9 @@ private: ...@@ -51,11 +51,9 @@ private:
* Previously there where also a scan type * Previously there where also a scan type
*/ */
static Uint32 getLockMode(const Uint32 & requestInfo); static Uint32 getLockMode(const Uint32 & requestInfo);
static Uint32 getKeyinfoFlag(const Uint32 & requestInfo);
static Uint32 getReadCommittedFlag(const Uint32 & requestInfo); static Uint32 getReadCommittedFlag(const Uint32 & requestInfo);
static void setLockMode(Uint32 & requestInfo, Uint32 lockMode); static void setLockMode(Uint32 & requestInfo, Uint32 lockMode);
static void setKeyinfoFlag(Uint32 & requestInfo, Uint32 keyinfo);
static void setReadCommittedFlag(Uint32 & requestInfo, Uint32 readCommitted); static void setReadCommittedFlag(Uint32 & requestInfo, Uint32 readCommitted);
}; };
...@@ -63,17 +61,14 @@ private: ...@@ -63,17 +61,14 @@ private:
* Request Info * Request Info
* *
* l = Lock Mode - 1 Bit 2 * l = Lock Mode - 1 Bit 2
* k = Keyinfo - 1 Bit 4
* h = Read Committed - 1 Bit 5 * h = Read Committed - 1 Bit 5
* *
* 1111111111222222222233 * 1111111111222222222233
* 01234567890123456789012345678901 * 01234567890123456789012345678901
* l kh * l h
*/ */
#define AS_LOCK_MODE_SHIFT (2) #define AS_LOCK_MODE_SHIFT (2)
#define AS_LOCK_MODE_MASK (1) #define AS_LOCK_MODE_MASK (1)
#define AS_KEYINFO_SHIFT (4)
#define AS_READ_COMMITTED_SHIFT (5) #define AS_READ_COMMITTED_SHIFT (5)
inline inline
...@@ -82,12 +77,6 @@ AccScanReq::getLockMode(const Uint32 & requestInfo){ ...@@ -82,12 +77,6 @@ AccScanReq::getLockMode(const Uint32 & requestInfo){
return (requestInfo >> AS_LOCK_MODE_SHIFT) & AS_LOCK_MODE_MASK; return (requestInfo >> AS_LOCK_MODE_SHIFT) & AS_LOCK_MODE_MASK;
} }
inline
Uint32
AccScanReq::getKeyinfoFlag(const Uint32 & requestInfo){
return (requestInfo >> AS_KEYINFO_SHIFT) & 1;
}
inline inline
Uint32 Uint32
AccScanReq::getReadCommittedFlag(const Uint32 & requestInfo){ AccScanReq::getReadCommittedFlag(const Uint32 & requestInfo){
...@@ -101,13 +90,6 @@ AccScanReq::setLockMode(UintR & requestInfo, UintR val){ ...@@ -101,13 +90,6 @@ AccScanReq::setLockMode(UintR & requestInfo, UintR val){
requestInfo |= (val << AS_LOCK_MODE_SHIFT); requestInfo |= (val << AS_LOCK_MODE_SHIFT);
} }
inline
void
AccScanReq::setKeyinfoFlag(UintR & requestInfo, UintR val){
ASSERT_BOOL(val, "AccScanReq::setKeyinfoFlag");
requestInfo |= (val << AS_KEYINFO_SHIFT);
}
inline inline
void void
AccScanReq::setReadCommittedFlag(UintR & requestInfo, UintR val){ AccScanReq::setReadCommittedFlag(UintR & requestInfo, UintR val){
......
ndb/include/kernel/signaldata/TuxBound.hpp
View file @ 46075704
...@@ -50,6 +50,8 @@ private: ...@@ -50,6 +50,8 @@ private:
* Number of words of bound info included after fixed signal data. * Number of words of bound info included after fixed signal data.
*/ */
Uint32 boundAiLength; Uint32 boundAiLength;
Uint32 data[1];
}; };
#endif #endif
ndb/include/ndbapi/NdbIndexScanOperation.hpp
View file @ 46075704
...@@ -114,6 +114,12 @@ public: ...@@ -114,6 +114,12 @@ public:
* sent on next execute * sent on next execute
*/ */
int reset_bounds(); int reset_bounds();
/**
* Set new bound on operation,
* used when batching index reads
*/
int set_new_bound();
bool getSorted() const { return m_ordered; } bool getSorted() const { return m_ordered; }
private: private:
...@@ -132,6 +138,9 @@ private: ...@@ -132,6 +138,9 @@ private:
int compare(Uint32 key, Uint32 cols, const NdbReceiver*, const NdbReceiver*); int compare(Uint32 key, Uint32 cols, const NdbReceiver*, const NdbReceiver*);
Uint32 m_sort_columns; Uint32 m_sort_columns;
Uint32 m_this_bound_start;
Uint32 * m_first_bound_word;
}; };
#endif #endif
ndb/src/kernel/blocks/dbacc/DbaccMain.cpp
View file @ 46075704
...@@ -8949,7 +8949,6 @@ void Dbacc::execACC_SCANREQ(Signal* signal) ...@@ -8949,7 +8949,6 @@ void Dbacc::execACC_SCANREQ(Signal* signal)
rootfragrecptr.p->scan[i] = scanPtr.i; rootfragrecptr.p->scan[i] = scanPtr.i;
scanPtr.p->scanBucketState = ScanRec::FIRST_LAP; scanPtr.p->scanBucketState = ScanRec::FIRST_LAP;
scanPtr.p->scanLockMode = AccScanReq::getLockMode(tscanFlag); scanPtr.p->scanLockMode = AccScanReq::getLockMode(tscanFlag);
scanPtr.p->scanKeyinfoFlag = AccScanReq::getKeyinfoFlag(tscanFlag);
scanPtr.p->scanReadCommittedFlag = AccScanReq::getReadCommittedFlag(tscanFlag); scanPtr.p->scanReadCommittedFlag = AccScanReq::getReadCommittedFlag(tscanFlag);
/* TWELVE BITS OF THE ELEMENT HEAD ARE SCAN */ /* TWELVE BITS OF THE ELEMENT HEAD ARE SCAN */
...@@ -10083,53 +10082,19 @@ void Dbacc::sendNextScanConf(Signal* signal) ...@@ -10083,53 +10082,19 @@ void Dbacc::sendNextScanConf(Signal* signal)
{ {
scanPtr.p->scanTimer = scanPtr.p->scanContinuebCounter; scanPtr.p->scanTimer = scanPtr.p->scanContinuebCounter;
Uint32 blockNo = refToBlock(scanPtr.p->scanUserblockref); Uint32 blockNo = refToBlock(scanPtr.p->scanUserblockref);
if (!scanPtr.p->scanKeyinfoFlag){ jam();
jam(); /** ---------------------------------------------------------------------
/** --------------------------------------------------------------------- * LQH WILL NOT HAVE ANY USE OF THE TUPLE KEY LENGTH IN THIS CASE AND
* LQH WILL NOT HAVE ANY USE OF THE TUPLE KEY LENGTH IN THIS CASE AND * SO WE DO NOT PROVIDE IT. IN THIS CASE THESE VALUES ARE UNDEFINED.
* SO WE DO NOT PROVIDE IT. IN THIS CASE THESE VALUES ARE UNDEFINED. * ---------------------------------------------------------------------- */
* ---------------------------------------------------------------------- */
signal->theData[0] = scanPtr.p->scanUserptr;
signal->theData[1] = operationRecPtr.i;
signal->theData[2] = operationRecPtr.p->fid;
signal->theData[3] = operationRecPtr.p->localdata[0];
signal->theData[4] = operationRecPtr.p->localdata[1];
signal->theData[5] = fragrecptr.p->localkeylen;
EXECUTE_DIRECT(blockNo, GSN_NEXT_SCANCONF, signal, 6);
return;
}//if
fragrecptr.i = operationRecPtr.p->fragptr;
ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
readTablePk(operationRecPtr.p->localdata[0]);
signal->theData[0] = scanPtr.p->scanUserptr; signal->theData[0] = scanPtr.p->scanUserptr;
signal->theData[1] = operationRecPtr.i; signal->theData[1] = operationRecPtr.i;
signal->theData[2] = operationRecPtr.p->fid; signal->theData[2] = operationRecPtr.p->fid;
signal->theData[3] = operationRecPtr.p->localdata[0]; signal->theData[3] = operationRecPtr.p->localdata[0];
signal->theData[4] = operationRecPtr.p->localdata[1]; signal->theData[4] = operationRecPtr.p->localdata[1];
signal->theData[5] = fragrecptr.p->localkeylen; signal->theData[5] = fragrecptr.p->localkeylen;
signal->theData[6] = fragrecptr.p->keyLength; EXECUTE_DIRECT(blockNo, GSN_NEXT_SCANCONF, signal, 6);
signal->theData[7] = ckeys[0]; return;
signal->theData[8] = ckeys[1];
signal->theData[9] = ckeys[2];
signal->theData[10] = ckeys[3];
EXECUTE_DIRECT(blockNo, GSN_NEXT_SCANCONF, signal, 11);
const Uint32 keyLength = fragrecptr.p->keyLength;
Uint32 total = 4;
while (total < keyLength) {
jam();
Uint32 length = keyLength - total;
if (length > 20)
length = 20;
signal->theData[0] = scanPtr.p->scanUserptr;
signal->theData[1] = operationRecPtr.i; // not used by LQH
signal->theData[2] = operationRecPtr.p->fid; // not used by LQH
signal->theData[3] = length;
memcpy(&signal->theData[4], &ckeys[total], length << 2);
EXECUTE_DIRECT(blockNo, GSN_ACC_SCAN_INFO24, signal, 4 + length);
// wl-2066 remove GSN_ACC_SCAN_INFO
total += length;
}//if
}//Dbacc::sendNextScanConf() }//Dbacc::sendNextScanConf()
/*--------------------------------------------------------------------------- /*---------------------------------------------------------------------------
...@@ -11233,13 +11198,12 @@ Dbacc::execDUMP_STATE_ORD(Signal* signal) ...@@ -11233,13 +11198,12 @@ Dbacc::execDUMP_STATE_ORD(Signal* signal)
scanPtr.p->minBucketIndexToRescan, scanPtr.p->minBucketIndexToRescan,
scanPtr.p->maxBucketIndexToRescan); scanPtr.p->maxBucketIndexToRescan);
infoEvent(" scanBucketState=%d, scanLockHeld=%d, userBlockRef=%d, " infoEvent(" scanBucketState=%d, scanLockHeld=%d, userBlockRef=%d, "
"scanMask=%d scanLockMode=%d, keyInfoFlag=%d", "scanMask=%d scanLockMode=%d",
scanPtr.p->scanBucketState, scanPtr.p->scanBucketState,
scanPtr.p->scanLockHeld, scanPtr.p->scanLockHeld,
scanPtr.p->scanUserblockref, scanPtr.p->scanUserblockref,
scanPtr.p->scanMask, scanPtr.p->scanMask,
scanPtr.p->scanLockMode, scanPtr.p->scanLockMode);
scanPtr.p->scanKeyinfoFlag);
return; return;
} }
......
ndb/src/kernel/blocks/dblqh/Dblqh.hpp
View file @ 46075704
...@@ -29,6 +29,9 @@ ...@@ -29,6 +29,9 @@
#include <signaldata/LqhTransConf.hpp> #include <signaldata/LqhTransConf.hpp>
#include <signaldata/LqhFrag.hpp> #include <signaldata/LqhFrag.hpp>
// primary key is stored in TUP
#include <../dbtup/Dbtup.hpp>
#ifdef DBLQH_C #ifdef DBLQH_C
// Constants // Constants
/* ------------------------------------------------------------------------- */ /* ------------------------------------------------------------------------- */
...@@ -500,9 +503,7 @@ public: ...@@ -500,9 +503,7 @@ public:
WAIT_DELETE_STORED_PROC_ID_COPY = 6, WAIT_DELETE_STORED_PROC_ID_COPY = 6,
WAIT_ACC_COPY = 7, WAIT_ACC_COPY = 7,
WAIT_ACC_SCAN = 8, WAIT_ACC_SCAN = 8,
WAIT_SCAN_KEYINFO = 9,
WAIT_SCAN_NEXTREQ = 10, WAIT_SCAN_NEXTREQ = 10,
WAIT_COPY_KEYINFO = 11,
WAIT_CLOSE_SCAN = 12, WAIT_CLOSE_SCAN = 12,
WAIT_CLOSE_COPY = 13, WAIT_CLOSE_COPY = 13,
WAIT_RELEASE_LOCK = 14, WAIT_RELEASE_LOCK = 14,
...@@ -2025,6 +2026,7 @@ public: ...@@ -2025,6 +2026,7 @@ public:
Uint8 opExec; Uint8 opExec;
Uint8 operation; Uint8 operation;
Uint8 reclenAiLqhkey; Uint8 reclenAiLqhkey;
Uint8 m_offset_current_keybuf;
Uint8 replicaType; Uint8 replicaType;
Uint8 simpleRead; Uint8 simpleRead;
Uint8 seqNoReplica; Uint8 seqNoReplica;
...@@ -2128,8 +2130,6 @@ private: ...@@ -2128,8 +2130,6 @@ private:
void execACC_SCANREF(Signal* signal); void execACC_SCANREF(Signal* signal);
void execNEXT_SCANCONF(Signal* signal); void execNEXT_SCANCONF(Signal* signal);
void execNEXT_SCANREF(Signal* signal); void execNEXT_SCANREF(Signal* signal);
void execACC_SCAN_INFO(Signal* signal);
void execACC_SCAN_INFO24(Signal* signal);
void execACC_TO_REF(Signal* signal); void execACC_TO_REF(Signal* signal);
void execSTORED_PROCCONF(Signal* signal); void execSTORED_PROCCONF(Signal* signal);
void execSTORED_PROCREF(Signal* signal); void execSTORED_PROCREF(Signal* signal);
...@@ -2250,7 +2250,7 @@ private: ...@@ -2250,7 +2250,7 @@ private:
void finishScanrec(Signal* signal); void finishScanrec(Signal* signal);
void releaseScanrec(Signal* signal); void releaseScanrec(Signal* signal);
void seizeScanrec(Signal* signal); void seizeScanrec(Signal* signal);
void sendKeyinfo20(Signal* signal, ScanRecord *, TcConnectionrec *); Uint32 sendKeyinfo20(Signal* signal, ScanRecord *, TcConnectionrec *);
void sendScanFragConf(Signal* signal, Uint32 scanCompleted); void sendScanFragConf(Signal* signal, Uint32 scanCompleted);
void initCopyrec(Signal* signal); void initCopyrec(Signal* signal);
void initCopyTc(Signal* signal); void initCopyTc(Signal* signal);
...@@ -2379,6 +2379,8 @@ private: ...@@ -2379,6 +2379,8 @@ private:
void seizeAttrinbuf(Signal* signal); void seizeAttrinbuf(Signal* signal);
Uint32 seize_attrinbuf(); Uint32 seize_attrinbuf();
Uint32 release_attrinbuf(Uint32); Uint32 release_attrinbuf(Uint32);
Uint32 copy_bounds(Uint32 * dst, TcConnectionrec*);
void seizeFragmentrec(Signal* signal); void seizeFragmentrec(Signal* signal);
void seizePageRef(Signal* signal); void seizePageRef(Signal* signal);
void seizeTcrec(); void seizeTcrec();
...@@ -2425,7 +2427,6 @@ private: ...@@ -2425,7 +2427,6 @@ private:
void localCommitLab(Signal* signal); void localCommitLab(Signal* signal);
void abortErrorLab(Signal* signal); void abortErrorLab(Signal* signal);
void continueAfterReceivingAllAiLab(Signal* signal); void continueAfterReceivingAllAiLab(Signal* signal);
void sendScanFragRefLateLab(Signal* signal);
void abortStateHandlerLab(Signal* signal); void abortStateHandlerLab(Signal* signal);
void writeAttrinfoLab(Signal* signal); void writeAttrinfoLab(Signal* signal);
void scanAttrinfoLab(Signal* signal, Uint32* dataPtr, Uint32 length); void scanAttrinfoLab(Signal* signal, Uint32* dataPtr, Uint32 length);
...@@ -2492,7 +2493,7 @@ private: ...@@ -2492,7 +2493,7 @@ private:
void nextScanConfScanLab(Signal* signal); void nextScanConfScanLab(Signal* signal);
void nextScanConfCopyLab(Signal* signal); void nextScanConfCopyLab(Signal* signal);
void continueScanNextReqLab(Signal* signal); void continueScanNextReqLab(Signal* signal);
bool keyinfoLab(Signal* signal, Uint32* dataPtr, Uint32 length); void keyinfoLab(const Uint32 * src, const Uint32 * end);
void copySendTupkeyReqLab(Signal* signal); void copySendTupkeyReqLab(Signal* signal);
void storedProcConfScanLab(Signal* signal); void storedProcConfScanLab(Signal* signal);
void storedProcConfCopyLab(Signal* signal); void storedProcConfCopyLab(Signal* signal);
...@@ -2548,7 +2549,6 @@ private: ...@@ -2548,7 +2549,6 @@ private:
void accScanConfScanLab(Signal* signal); void accScanConfScanLab(Signal* signal);
void accScanConfCopyLab(Signal* signal); void accScanConfCopyLab(Signal* signal);
void scanLockReleasedLab(Signal* signal); void scanLockReleasedLab(Signal* signal);
void accScanInfoEnterLab(Signal* signal, Uint32* dataPtr, Uint32 length);
void openSrFourthNextLab(Signal* signal); void openSrFourthNextLab(Signal* signal);
void closingInitLab(Signal* signal); void closingInitLab(Signal* signal);
void closeExecSrCompletedLab(Signal* signal); void closeExecSrCompletedLab(Signal* signal);
...@@ -2563,6 +2563,8 @@ private: ...@@ -2563,6 +2563,8 @@ private:
void initData(); void initData();
void initRecords(); void initRecords();
Dbtup* c_tup;
Uint32 readPrimaryKeys(ScanRecord*, TcConnectionrec*, Uint32 * dst);
// ---------------------------------------------------------------- // ----------------------------------------------------------------
// These are variables handling the records. For most records one // These are variables handling the records. For most records one
// pointer to the array of structs, one pointer-struct, a file size // pointer to the array of structs, one pointer-struct, a file size
......
ndb/src/kernel/blocks/dblqh/DblqhInit.cpp
View file @ 46075704
...@@ -274,8 +274,6 @@ Dblqh::Dblqh(const class Configuration & conf): ...@@ -274,8 +274,6 @@ Dblqh::Dblqh(const class Configuration & conf):
addRecSignal(GSN_ACC_SCANREF, &Dblqh::execACC_SCANREF); addRecSignal(GSN_ACC_SCANREF, &Dblqh::execACC_SCANREF);
addRecSignal(GSN_NEXT_SCANCONF, &Dblqh::execNEXT_SCANCONF); addRecSignal(GSN_NEXT_SCANCONF, &Dblqh::execNEXT_SCANCONF);
addRecSignal(GSN_NEXT_SCANREF, &Dblqh::execNEXT_SCANREF); addRecSignal(GSN_NEXT_SCANREF, &Dblqh::execNEXT_SCANREF);
addRecSignal(GSN_ACC_SCAN_INFO, &Dblqh::execACC_SCAN_INFO);
addRecSignal(GSN_ACC_SCAN_INFO24, &Dblqh::execACC_SCAN_INFO24);
addRecSignal(GSN_STORED_PROCCONF, &Dblqh::execSTORED_PROCCONF); addRecSignal(GSN_STORED_PROCCONF, &Dblqh::execSTORED_PROCCONF);
addRecSignal(GSN_STORED_PROCREF, &Dblqh::execSTORED_PROCREF); addRecSignal(GSN_STORED_PROCREF, &Dblqh::execSTORED_PROCREF);
addRecSignal(GSN_COPY_FRAGREQ, &Dblqh::execCOPY_FRAGREQ); addRecSignal(GSN_COPY_FRAGREQ, &Dblqh::execCOPY_FRAGREQ);
......
ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
View file @ 46075704
...@@ -490,6 +490,8 @@ void Dblqh::execSTTOR(Signal* signal) ...@@ -490,6 +490,8 @@ void Dblqh::execSTTOR(Signal* signal)
jam(); jam();
cstartPhase = tstartPhase; cstartPhase = tstartPhase;
sttorStartphase1Lab(signal); sttorStartphase1Lab(signal);
c_tup = (Dbtup*)globalData.getBlock(DBTUP);
ndbrequire(c_tup != 0);
return; return;
break; break;
default: default:
...@@ -6830,49 +6832,6 @@ Dblqh::scanMarkers(Signal* signal, ...@@ -6830,49 +6832,6 @@ Dblqh::scanMarkers(Signal* signal,
* ALL TUPLES IN THE FRAGMENT. TUP PERFORMS THE NECESSARY SEARCH CONDITIONS * ALL TUPLES IN THE FRAGMENT. TUP PERFORMS THE NECESSARY SEARCH CONDITIONS
* TO ENSURE THAT ONLY VALID TUPLES ARE RETURNED TO THE APPLICATION. * TO ENSURE THAT ONLY VALID TUPLES ARE RETURNED TO THE APPLICATION.
* ------------------------------------------------------------------------- */ * ------------------------------------------------------------------------- */
void Dblqh::execACC_SCAN_INFO(Signal* signal)
{
jamEntry();
scanptr.i = signal->theData[0];
c_scanRecordPool.getPtr(scanptr);
Uint32 length = signal->theData[3];
ndbrequire(length <= 4);
accScanInfoEnterLab(signal, &signal->theData[4], length);
}//Dblqh::execACC_SCAN_INFO()
void Dblqh::execACC_SCAN_INFO24(Signal* signal)
{
jamEntry();
scanptr.i = signal->theData[0];
c_scanRecordPool.getPtr(scanptr);
Uint32 length = signal->theData[3];
ndbrequire(length <= 20);
accScanInfoEnterLab(signal, &signal->theData[4], length);
}//Dblqh::execACC_SCAN_INFO24()
void Dblqh::accScanInfoEnterLab(Signal* signal,
Uint32* dataPtr,
Uint32 length)
{
ndbrequire(length != 0);
if (scanptr.p->scanState == ScanRecord::WAIT_SCAN_KEYINFO) {
jam();
if (keyinfoLab(signal, dataPtr, length)) {
jam();
nextScanConfLoopLab(signal);
}//if
} else {
ndbrequire(scanptr.p->scanState == ScanRecord::WAIT_COPY_KEYINFO);
jam();
if (keyinfoLab(signal, dataPtr, length)) {
jam();
copySendTupkeyReqLab(signal);
}//if
}//if
}//Dblqh::accScanInfoEnterLab()
/* *************** */ /* *************** */
/* ACC_SCANCONF > */ /* ACC_SCANCONF > */
/* *************** */ /* *************** */
...@@ -7238,7 +7197,6 @@ void Dblqh::closeScanRequestLab(Signal* signal) ...@@ -7238,7 +7197,6 @@ void Dblqh::closeScanRequestLab(Signal* signal)
jam(); jam();
tupScanCloseConfLab(signal); tupScanCloseConfLab(signal);
break; break;
case ScanRecord::WAIT_SCAN_KEYINFO:
case ScanRecord::WAIT_NEXT_SCAN: case ScanRecord::WAIT_NEXT_SCAN:
jam(); jam();
/* ------------------------------------------------------------------- /* -------------------------------------------------------------------
...@@ -7302,6 +7260,7 @@ void Dblqh::closeScanRequestLab(Signal* signal) ...@@ -7302,6 +7260,7 @@ void Dblqh::closeScanRequestLab(Signal* signal)
* WE ARE STILL WAITING FOR THE ATTRIBUTE INFORMATION THAT * WE ARE STILL WAITING FOR THE ATTRIBUTE INFORMATION THAT
* OBVIOUSLY WILL NOT ARRIVE. WE CAN QUIT IMMEDIATELY HERE. * OBVIOUSLY WILL NOT ARRIVE. WE CAN QUIT IMMEDIATELY HERE.
* --------------------------------------------------------------------- */ * --------------------------------------------------------------------- */
//XXX jonas this have to be wrong...
releaseOprec(signal); releaseOprec(signal);
if (tcConnectptr.p->abortState == TcConnectionrec::NEW_FROM_TC) { if (tcConnectptr.p->abortState == TcConnectionrec::NEW_FROM_TC) {
jam(); jam();
...@@ -7665,7 +7624,6 @@ void Dblqh::continueAfterReceivingAllAiLab(Signal* signal) ...@@ -7665,7 +7624,6 @@ void Dblqh::continueAfterReceivingAllAiLab(Signal* signal)
req->fragmentNo = tcConnectptr.p->fragmentid; req->fragmentNo = tcConnectptr.p->fragmentid;
req->requestInfo = 0; req->requestInfo = 0;
AccScanReq::setLockMode(req->requestInfo, scanptr.p->scanLockMode); AccScanReq::setLockMode(req->requestInfo, scanptr.p->scanLockMode);
AccScanReq::setKeyinfoFlag(req->requestInfo, scanptr.p->scanKeyinfoFlag);
AccScanReq::setReadCommittedFlag(req->requestInfo, scanptr.p->readCommitted); AccScanReq::setReadCommittedFlag(req->requestInfo, scanptr.p->readCommitted);
req->transId1 = tcConnectptr.p->transid[0]; req->transId1 = tcConnectptr.p->transid[0];
req->transId2 = tcConnectptr.p->transid[1]; req->transId2 = tcConnectptr.p->transid[1];
...@@ -7763,11 +7721,6 @@ void Dblqh::execSCAN_HBREP(Signal* signal) ...@@ -7763,11 +7721,6 @@ void Dblqh::execSCAN_HBREP(Signal* signal)
} }
} }
void Dblqh::sendScanFragRefLateLab(Signal* signal)
{
}//Dblqh::sendScanFragRefLateLab()
void Dblqh::accScanConfScanLab(Signal* signal) void Dblqh::accScanConfScanLab(Signal* signal)
{ {
AccScanConf * const accScanConf = (AccScanConf *)&signal->theData[0]; AccScanConf * const accScanConf = (AccScanConf *)&signal->theData[0];
...@@ -7786,17 +7739,15 @@ void Dblqh::accScanConfScanLab(Signal* signal) ...@@ -7786,17 +7739,15 @@ void Dblqh::accScanConfScanLab(Signal* signal)
return; return;
}//if }//if
scanptr.p->scanAccPtr = accScanConf->accPtr; scanptr.p->scanAccPtr = accScanConf->accPtr;
Uint32 boundAiLength = tcConnectptr.p->primKeyLen - 4;
if (scanptr.p->rangeScan) { if (scanptr.p->rangeScan) {
jam(); jam();
TuxBoundInfo* const req = (TuxBoundInfo*)signal->getDataPtrSend(); TuxBoundInfo* req = (TuxBoundInfo*)signal->getDataPtrSend();
req->errorCode = RNIL; req->errorCode = RNIL;
req->tuxScanPtrI = scanptr.p->scanAccPtr; req->tuxScanPtrI = scanptr.p->scanAccPtr;
req->boundAiLength = boundAiLength; Uint32 len = req->boundAiLength = copy_bounds(req->data, tcConnectptr.p);
if(boundAiLength > 0) EXECUTE_DIRECT(DBTUX, GSN_TUX_BOUND_INFO, signal,
sendKeyinfoAcc(signal, TuxBoundInfo::SignalLength); TuxBoundInfo::SignalLength + len);
EXECUTE_DIRECT(DBTUX, GSN_TUX_BOUND_INFO,
signal, TuxBoundInfo::SignalLength + boundAiLength);
jamEntry(); jamEntry();
if (req->errorCode != 0) { if (req->errorCode != 0) {
jam(); jam();
...@@ -7820,7 +7771,7 @@ void Dblqh::accScanConfScanLab(Signal* signal) ...@@ -7820,7 +7771,7 @@ void Dblqh::accScanConfScanLab(Signal* signal)
signal->theData[0] = tcConnectptr.p->tupConnectrec; signal->theData[0] = tcConnectptr.p->tupConnectrec;
AttrbufPtr regAttrinbufptr; AttrbufPtr regAttrinbufptr;
regAttrinbufptr.i = tcConnectptr.p->firstAttrinbuf; Uint32 firstAttr = regAttrinbufptr.i = tcConnectptr.p->firstAttrinbuf;
while (regAttrinbufptr.i != RNIL) { while (regAttrinbufptr.i != RNIL) {
ptrCheckGuard(regAttrinbufptr, cattrinbufFileSize, attrbuf); ptrCheckGuard(regAttrinbufptr, cattrinbufFileSize, attrbuf);
jam(); jam();
...@@ -7833,10 +7784,140 @@ void Dblqh::accScanConfScanLab(Signal* signal) ...@@ -7833,10 +7784,140 @@ void Dblqh::accScanConfScanLab(Signal* signal)
sendSignal(tcConnectptr.p->tcTupBlockref, sendSignal(tcConnectptr.p->tcTupBlockref,
GSN_ATTRINFO, signal, dataLen + 3, JBB); GSN_ATTRINFO, signal, dataLen + 3, JBB);
regAttrinbufptr.i = regAttrinbufptr.p->attrbuf[ZINBUF_NEXT]; regAttrinbufptr.i = regAttrinbufptr.p->attrbuf[ZINBUF_NEXT];
c_no_attrinbuf_recs++;
}//while }//while
releaseOprec(signal);
/**
* Release attr info
*/
if(firstAttr != RNIL)
{
regAttrinbufptr.p->attrbuf[ZINBUF_NEXT] = cfirstfreeAttrinbuf;
cfirstfreeAttrinbuf = firstAttr;
tcConnectptr.p->firstAttrinbuf = tcConnectptr.p->lastAttrinbuf = RNIL;
}
}//Dblqh::accScanConfScanLab() }//Dblqh::accScanConfScanLab()
#define print_buf(s,idx,len) {\
printf(s); Uint32 t2=len; DatabufPtr t3; t3.i = idx; \
while(t3.i != RNIL && t2-- > 0){\
ptrCheckGuard(t3, cdatabufFileSize, databuf);\
printf("%d ", t3.i); t3.i= t3.p->nextDatabuf;\
} printf("\n"); }
Uint32
Dblqh::copy_bounds(Uint32 * dst, TcConnectionrec* tcPtrP)
{
/**
* copy_bounds handles multiple bounds by
* in the 16 upper bits of the first words (used to specify bound type)
* setting the length of this specific bound
*
*/
DatabufPtr regDatabufptr;
Uint32 left = 4 - tcPtrP->m_offset_current_keybuf; // left in buf
Uint32 totalLen = tcPtrP->primKeyLen - 4;
regDatabufptr.i = tcPtrP->firstTupkeybuf;
ndbassert(tcPtrP->primKeyLen >= 4);
ndbassert(tcPtrP->m_offset_current_keybuf < 4);
ndbassert(!(totalLen == 0 && regDatabufptr.i != RNIL));
ndbassert(!(totalLen != 0 && regDatabufptr.i == RNIL));
if(totalLen)
{
ptrCheckGuard(regDatabufptr, cdatabufFileSize, databuf);
Uint32 sig0 = regDatabufptr.p->data[0];
Uint32 sig1 = regDatabufptr.p->data[1];
Uint32 sig2 = regDatabufptr.p->data[2];
Uint32 sig3 = regDatabufptr.p->data[3];
switch(left){
case 4:
* dst++ = sig0;
case 3:
* dst++ = sig1;
case 2:
* dst++ = sig2;
case 1:
* dst++ = sig3;
}
Uint32 first = (* (dst - left)); // First word in range
(* (dst - left)) = (first & 0xFFFF); // Remove length (16 upper bits)
// Length of this range
Uint8 offset;
const Uint32 len = (first >> 16) ? (first >> 16) : totalLen;
if(len < left)
{
offset = 4 - len;
}
else
{
Databuf * lastP;
left = (len - left);
regDatabufptr.i = regDatabufptr.p->nextDatabuf;
while(left >= 4)
{
left -= 4;
lastP = regDatabufptr.p;
ptrCheckGuard(regDatabufptr, cdatabufFileSize, databuf);
sig0 = regDatabufptr.p->data[0];
sig1 = regDatabufptr.p->data[1];
sig2 = regDatabufptr.p->data[2];
sig3 = regDatabufptr.p->data[3];
regDatabufptr.i = regDatabufptr.p->nextDatabuf;
* dst++ = sig0;
* dst++ = sig1;
* dst++ = sig2;
* dst++ = sig3;
}
if(left > 0)
{
lastP = regDatabufptr.p;
ptrCheckGuard(regDatabufptr, cdatabufFileSize, databuf);
sig0 = regDatabufptr.p->data[0];
sig1 = regDatabufptr.p->data[1];
sig2 = regDatabufptr.p->data[2];
sig3 = regDatabufptr.p->data[3];
* dst++ = sig0;
* dst++ = sig1;
* dst++ = sig2;
* dst++ = sig3;
}
else
{
lastP = regDatabufptr.p;
}
offset = left & 3;
lastP->nextDatabuf = cfirstfreeDatabuf;
cfirstfreeDatabuf = tcPtrP->firstTupkeybuf;
ndbassert(cfirstfreeDatabuf != RNIL);
}
if(len == totalLen && regDatabufptr.i != RNIL)
{
regDatabufptr.p->nextDatabuf = cfirstfreeDatabuf;
cfirstfreeDatabuf = regDatabufptr.i;
tcPtrP->lastTupkeybuf = regDatabufptr.i = RNIL;
ndbassert(cfirstfreeDatabuf != RNIL);
}
tcPtrP->m_offset_current_keybuf = offset;
tcPtrP->firstTupkeybuf = regDatabufptr.i;
tcPtrP->primKeyLen = 4 + totalLen - len;
return len;
}
return totalLen;
}
/* ------------------------------------------------------------------------- /* -------------------------------------------------------------------------
* ENTER STORED_PROCCONF WITH * ENTER STORED_PROCCONF WITH
* TC_CONNECTPTR, * TC_CONNECTPTR,
...@@ -8070,22 +8151,6 @@ void Dblqh::nextScanConfScanLab(Signal* signal) ...@@ -8070,22 +8151,6 @@ void Dblqh::nextScanConfScanLab(Signal* signal)
scanptr.p->scanLocalref[0] = nextScanConf->localKey[0]; scanptr.p->scanLocalref[0] = nextScanConf->localKey[0];
scanptr.p->scanLocalref[1] = nextScanConf->localKey[1]; scanptr.p->scanLocalref[1] = nextScanConf->localKey[1];
scanptr.p->scanLocalFragid = nextScanConf->fragId; scanptr.p->scanLocalFragid = nextScanConf->fragId;
if (scanptr.p->scanKeyinfoFlag) {
jam();
tcConnectptr.p->primKeyLen = nextScanConf->keyLength;
seizeTupkeybuf(signal);
databufptr.p->data[0] = nextScanConf->key[0];
databufptr.p->data[1] = nextScanConf->key[1];
databufptr.p->data[2] = nextScanConf->key[2];
databufptr.p->data[3] = nextScanConf->key[3];
if (nextScanConf->keyLength > 4) {
jam();
tcConnectptr.p->save1 = 4;
scanptr.p->scanState = ScanRecord::WAIT_SCAN_KEYINFO;
return;
}//if
}//if
jam();
nextScanConfLoopLab(signal); nextScanConfLoopLab(signal);
}//Dblqh::nextScanConfScanLab() }//Dblqh::nextScanConfScanLab()
...@@ -8097,7 +8162,6 @@ void Dblqh::nextScanConfLoopLab(Signal* signal) ...@@ -8097,7 +8162,6 @@ void Dblqh::nextScanConfLoopLab(Signal* signal)
if (scanptr.p->scanCompletedStatus == ZTRUE) { if (scanptr.p->scanCompletedStatus == ZTRUE) {
jam(); jam();
releaseActiveFrag(signal); releaseActiveFrag(signal);
releaseOprec(signal);
if ((scanptr.p->scanLockHold == ZTRUE) && if ((scanptr.p->scanLockHold == ZTRUE) &&
(scanptr.p->m_curr_batch_size_rows > 0)) { (scanptr.p->m_curr_batch_size_rows > 0)) {
jam(); jam();
...@@ -8162,33 +8226,46 @@ void Dblqh::nextScanConfLoopLab(Signal* signal) ...@@ -8162,33 +8226,46 @@ void Dblqh::nextScanConfLoopLab(Signal* signal)
* ------------------------------------------------------------------------- * -------------------------------------------------------------------------
* PRECONDITION: SCAN_STATE = WAIT_SCAN_KEYINFO * PRECONDITION: SCAN_STATE = WAIT_SCAN_KEYINFO
* ------------------------------------------------------------------------- */ * ------------------------------------------------------------------------- */
bool Dblqh::keyinfoLab(Signal* signal, Uint32* dataPtr, Uint32 length) void
Dblqh::keyinfoLab(const Uint32 * src, const Uint32 * end)
{ {
tcConnectptr.i = scanptr.p->scanTcrec;
ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec);
Uint32 index = 0;
do { do {
jam(); jam();
seizeTupkeybuf(signal); seizeTupkeybuf(0);
databufptr.p->data[0] = dataPtr[index]; databufptr.p->data[0] = * src ++;
databufptr.p->data[1] = dataPtr[index + 1]; databufptr.p->data[1] = * src ++;
databufptr.p->data[2] = dataPtr[index + 2]; databufptr.p->data[2] = * src ++;
databufptr.p->data[3] = dataPtr[index + 3]; databufptr.p->data[3] = * src ++;
index += 4; } while (src < end);
tcConnectptr.p->save1 = tcConnectptr.p->save1 + 4;
if (tcConnectptr.p->save1 >= tcConnectptr.p->primKeyLen) {
jam();
return true;
}//if
if (index >= length) {
jam();
return false;
}//if
} while (index < 20);
ndbrequire(false);
return false;
}//Dblqh::keyinfoLab() }//Dblqh::keyinfoLab()
Uint32
Dblqh::readPrimaryKeys(ScanRecord *scanP, TcConnectionrec *tcConP, Uint32 *dst)
{
Uint32 tableId = tcConP->tableref;
Uint32 fragId = scanP->scanLocalFragid;
Uint32 fragPageId = scanP->scanLocalref[0];
Uint32 pageIndex = scanP->scanLocalref[1];
if(scanP->rangeScan)
{
jam();
// for ordered index use primary table
FragrecordPtr tFragPtr;
tFragPtr.i = fragptr.p->tableFragptr;
ptrCheckGuard(tFragPtr, cfragrecFileSize, fragrecord);
tableId = tFragPtr.p->tabRef;
}
int ret = c_tup->accReadPk(tableId, fragId, fragPageId, pageIndex, dst);
if(0)
ndbout_c("readPrimaryKeys(table: %d fragment: %d [ %d %d ] -> %d",
tableId, fragId, fragPageId, pageIndex, ret);
ndbassert(ret > 0);
return ret;
}
/* ------------------------------------------------------------------------- /* -------------------------------------------------------------------------
* ENTER TUPKEYCONF * ENTER TUPKEYCONF
* ------------------------------------------------------------------------- * -------------------------------------------------------------------------
...@@ -8208,7 +8285,6 @@ void Dblqh::scanTupkeyConfLab(Signal* signal) ...@@ -8208,7 +8285,6 @@ void Dblqh::scanTupkeyConfLab(Signal* signal)
/* --------------------------------------------------------------------- /* ---------------------------------------------------------------------
* STOP THE SCAN PROCESS IF THIS HAS BEEN REQUESTED. * STOP THE SCAN PROCESS IF THIS HAS BEEN REQUESTED.
* --------------------------------------------------------------------- */ * --------------------------------------------------------------------- */
releaseOprec(signal);
if ((scanptr.p->scanLockHold == ZTRUE) && if ((scanptr.p->scanLockHold == ZTRUE) &&
(scanptr.p->m_curr_batch_size_rows > 0)) { (scanptr.p->m_curr_batch_size_rows > 0)) {
jam(); jam();
...@@ -8222,10 +8298,8 @@ void Dblqh::scanTupkeyConfLab(Signal* signal) ...@@ -8222,10 +8298,8 @@ void Dblqh::scanTupkeyConfLab(Signal* signal)
}//if }//if
if (scanptr.p->scanKeyinfoFlag) { if (scanptr.p->scanKeyinfoFlag) {
jam(); jam();
sendKeyinfo20(signal, scanptr.p, tcConnectptr.p); // Inform API about keyinfo len aswell
releaseOprec(signal); tdata4 += sendKeyinfo20(signal, scanptr.p, tcConnectptr.p);
tdata4 += tcConnectptr.p->primKeyLen;// Inform API about keyinfo len aswell
}//if }//if
ndbrequire(scanptr.p->m_curr_batch_size_rows < MAX_PARALLEL_OP_PER_SCAN); ndbrequire(scanptr.p->m_curr_batch_size_rows < MAX_PARALLEL_OP_PER_SCAN);
scanptr.p->m_curr_batch_size_bytes+= tdata4; scanptr.p->m_curr_batch_size_bytes+= tdata4;
...@@ -8327,7 +8401,6 @@ void Dblqh::scanTupkeyRefLab(Signal* signal) ...@@ -8327,7 +8401,6 @@ void Dblqh::scanTupkeyRefLab(Signal* signal)
tcConnectptr.p->transactionState = TcConnectionrec::SCAN_STATE_USED; tcConnectptr.p->transactionState = TcConnectionrec::SCAN_STATE_USED;
scanptr.i = tcConnectptr.p->tcScanRec; scanptr.i = tcConnectptr.p->tcScanRec;
releaseActiveFrag(signal); releaseActiveFrag(signal);
releaseOprec(signal);
c_scanRecordPool.getPtr(scanptr); c_scanRecordPool.getPtr(scanptr);
if (scanptr.p->scanCompletedStatus == ZTRUE) { if (scanptr.p->scanCompletedStatus == ZTRUE) {
/* --------------------------------------------------------------------- /* ---------------------------------------------------------------------
...@@ -8449,8 +8522,18 @@ void Dblqh::continueCloseScanAfterBlockedLab(Signal* signal) ...@@ -8449,8 +8522,18 @@ void Dblqh::continueCloseScanAfterBlockedLab(Signal* signal)
void Dblqh::accScanCloseConfLab(Signal* signal) void Dblqh::accScanCloseConfLab(Signal* signal)
{ {
tcConnectptr.i = scanptr.p->scanTcrec; tcConnectptr.i = scanptr.p->scanTcrec;
scanptr.p->scanState = ScanRecord::WAIT_DELETE_STORED_PROC_ID_SCAN;
ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec); ptrCheckGuard(tcConnectptr, ctcConnectrecFileSize, tcConnectionrec);
if((tcConnectptr.p->primKeyLen - 4) > 0 &&
scanptr.p->scanCompletedStatus != ZTRUE)
{
jam();
ndbout_c("Dont close scan");
//sendScanFragConf(signal, 0);
// Don't delete scan in TUP
}
scanptr.p->scanState = ScanRecord::WAIT_DELETE_STORED_PROC_ID_SCAN;
signal->theData[0] = tcConnectptr.p->tupConnectrec; signal->theData[0] = tcConnectptr.p->tupConnectrec;
signal->theData[1] = tcConnectptr.p->tableref; signal->theData[1] = tcConnectptr.p->tableref;
signal->theData[2] = scanptr.p->scanSchemaVersion; signal->theData[2] = scanptr.p->scanSchemaVersion;
...@@ -8639,6 +8722,7 @@ void Dblqh::initScanTc(Signal* signal, ...@@ -8639,6 +8722,7 @@ void Dblqh::initScanTc(Signal* signal,
tcConnectptr.p->operation = ZREAD; tcConnectptr.p->operation = ZREAD;
tcConnectptr.p->listState = TcConnectionrec::NOT_IN_LIST; tcConnectptr.p->listState = TcConnectionrec::NOT_IN_LIST;
tcConnectptr.p->commitAckMarker = RNIL; tcConnectptr.p->commitAckMarker = RNIL;
tcConnectptr.p->m_offset_current_keybuf = 0;
tabptr.p->usageCount++; tabptr.p->usageCount++;
}//Dblqh::initScanTc() }//Dblqh::initScanTc()
...@@ -8744,23 +8828,17 @@ void Dblqh::releaseScanrec(Signal* signal) ...@@ -8744,23 +8828,17 @@ void Dblqh::releaseScanrec(Signal* signal)
* ------- SEND KEYINFO20 TO API ------- * ------- SEND KEYINFO20 TO API -------
* *
* ------------------------------------------------------------------------ */ * ------------------------------------------------------------------------ */
void Dblqh::sendKeyinfo20(Signal* signal, Uint32 Dblqh::sendKeyinfo20(Signal* signal,
ScanRecord * scanP, ScanRecord * scanP,
TcConnectionrec * tcConP) TcConnectionrec * tcConP)
{ {
ndbrequire(scanP->m_curr_batch_size_rows < MAX_PARALLEL_OP_PER_SCAN); ndbrequire(scanP->m_curr_batch_size_rows < MAX_PARALLEL_OP_PER_SCAN);
KeyInfo20 * keyInfo = (KeyInfo20 *)&signal->theData[0]; KeyInfo20 * keyInfo = (KeyInfo20 *)&signal->theData[0];
DatabufPtr TdataBuf;
TdataBuf.i = tcConP->firstTupkeybuf;
Uint32 keyLen = tcConP->primKeyLen;
const Uint32 dataBufSz = cdatabufFileSize;
/** /**
* Note that this code requires signal->theData to be big enough for * Note that this code requires signal->theData to be big enough for
* a entire key * a entire key
*/ */
ndbrequire(keyLen * 4 <= sizeof(signal->theData));
const BlockReference ref = scanP->scanApiBlockref; const BlockReference ref = scanP->scanApiBlockref;
const Uint32 scanOp = scanP->m_curr_batch_size_rows; const Uint32 scanOp = scanP->m_curr_batch_size_rows;
const Uint32 nodeId = refToNode(ref); const Uint32 nodeId = refToNode(ref);
...@@ -8773,19 +8851,7 @@ void Dblqh::sendKeyinfo20(Signal* signal, ...@@ -8773,19 +8851,7 @@ void Dblqh::sendKeyinfo20(Signal* signal,
Uint32 * dst = keyInfo->keyData; Uint32 * dst = keyInfo->keyData;
dst += nodeId == getOwnNodeId() ? 0 : KeyInfo20::DataLength; dst += nodeId == getOwnNodeId() ? 0 : KeyInfo20::DataLength;
/** Uint32 keyLen = readPrimaryKeys(scanP, tcConP, dst);
* Copy keydata from data buffer into signal
*
*/
for(Uint32 i = 0; i < keyLen; i += 4){
ptrCheckGuard(TdataBuf, dataBufSz, databuf);
* dst++ = TdataBuf.p->data[0];
* dst++ = TdataBuf.p->data[1];
* dst++ = TdataBuf.p->data[2];
* dst++ = TdataBuf.p->data[3];
TdataBuf.i = TdataBuf.p->nextDatabuf;
}
Uint32 fragId = tcConP->fragmentid; Uint32 fragId = tcConP->fragmentid;
keyInfo->clientOpPtr = scanP->scanApiOpPtr; keyInfo->clientOpPtr = scanP->scanApiOpPtr;
keyInfo->keyLen = keyLen; keyInfo->keyLen = keyLen;
...@@ -8813,7 +8879,7 @@ void Dblqh::sendKeyinfo20(Signal* signal, ...@@ -8813,7 +8879,7 @@ void Dblqh::sendKeyinfo20(Signal* signal,
MEMCOPY_NO_WORDS(keyInfo->keyData, src, keyLen); MEMCOPY_NO_WORDS(keyInfo->keyData, src, keyLen);
sendSignal(ref, GSN_KEYINFO20, signal, sendSignal(ref, GSN_KEYINFO20, signal,
KeyInfo20::HeaderLength+keyLen, JBB); KeyInfo20::HeaderLength+keyLen, JBB);
return; return keyLen;
} }
LinearSectionPtr ptr[3]; LinearSectionPtr ptr[3];
...@@ -8821,13 +8887,13 @@ void Dblqh::sendKeyinfo20(Signal* signal, ...@@ -8821,13 +8887,13 @@ void Dblqh::sendKeyinfo20(Signal* signal,
ptr[0].sz = keyLen; ptr[0].sz = keyLen;
sendSignal(ref, GSN_KEYINFO20, signal, KeyInfo20::HeaderLength, sendSignal(ref, GSN_KEYINFO20, signal, KeyInfo20::HeaderLength,
JBB, ptr, 1); JBB, ptr, 1);
return; return keyLen;
} }
EXECUTE_DIRECT(refToBlock(ref), GSN_KEYINFO20, signal, EXECUTE_DIRECT(refToBlock(ref), GSN_KEYINFO20, signal,
KeyInfo20::HeaderLength + keyLen); KeyInfo20::HeaderLength + keyLen);
jamEntry(); jamEntry();
return; return keyLen;
} }
/** /**
...@@ -8853,7 +8919,7 @@ void Dblqh::sendKeyinfo20(Signal* signal, ...@@ -8853,7 +8919,7 @@ void Dblqh::sendKeyinfo20(Signal* signal,
keyInfo->keyData[keyLen] = ref; keyInfo->keyData[keyLen] = ref;
sendSignal(routeBlockref, GSN_KEYINFO20_R, signal, sendSignal(routeBlockref, GSN_KEYINFO20_R, signal,
KeyInfo20::HeaderLength+keyLen+1, JBB); KeyInfo20::HeaderLength+keyLen+1, JBB);
return; return keyLen;
} }
keyInfo->keyData[0] = ref; keyInfo->keyData[0] = ref;
...@@ -8862,7 +8928,7 @@ void Dblqh::sendKeyinfo20(Signal* signal, ...@@ -8862,7 +8928,7 @@ void Dblqh::sendKeyinfo20(Signal* signal,
ptr[0].sz = keyLen; ptr[0].sz = keyLen;
sendSignal(routeBlockref, GSN_KEYINFO20_R, signal, sendSignal(routeBlockref, GSN_KEYINFO20_R, signal,
KeyInfo20::HeaderLength+1, JBB, ptr, 1); KeyInfo20::HeaderLength+1, JBB, ptr, 1);
return; return keyLen;
} }
/* ------------------------------------------------------------------------ /* ------------------------------------------------------------------------
...@@ -9038,7 +9104,6 @@ void Dblqh::execCOPY_FRAGREQ(Signal* signal) ...@@ -9038,7 +9104,6 @@ void Dblqh::execCOPY_FRAGREQ(Signal* signal)
req->fragmentNo = fragId; req->fragmentNo = fragId;
req->requestInfo = 0; req->requestInfo = 0;
AccScanReq::setLockMode(req->requestInfo, 0); AccScanReq::setLockMode(req->requestInfo, 0);
AccScanReq::setKeyinfoFlag(req->requestInfo, 1);
AccScanReq::setReadCommittedFlag(req->requestInfo, 0); AccScanReq::setReadCommittedFlag(req->requestInfo, 0);
req->transId1 = tcConnectptr.p->transid[0]; req->transId1 = tcConnectptr.p->transid[0];
req->transId2 = tcConnectptr.p->transid[1]; req->transId2 = tcConnectptr.p->transid[1];
...@@ -9191,12 +9256,6 @@ void Dblqh::nextScanConfCopyLab(Signal* signal) ...@@ -9191,12 +9256,6 @@ void Dblqh::nextScanConfCopyLab(Signal* signal)
set_acc_ptr_in_scan_record(scanptr.p, 0, nextScanConf->accOperationPtr); set_acc_ptr_in_scan_record(scanptr.p, 0, nextScanConf->accOperationPtr);
initCopyTc(signal); initCopyTc(signal);
if (tcConnectptr.p->primKeyLen > 4) {
jam();
tcConnectptr.p->save1 = 4;
scanptr.p->scanState = ScanRecord::WAIT_COPY_KEYINFO;
return;
}//if
copySendTupkeyReqLab(signal); copySendTupkeyReqLab(signal);
return; return;
}//Dblqh::nextScanConfCopyLab() }//Dblqh::nextScanConfCopyLab()
...@@ -9290,6 +9349,7 @@ void Dblqh::copyTupkeyConfLab(Signal* signal) ...@@ -9290,6 +9349,7 @@ void Dblqh::copyTupkeyConfLab(Signal* signal)
scanptr.i = tcConnectptr.p->tcScanRec; scanptr.i = tcConnectptr.p->tcScanRec;
c_scanRecordPool.getPtr(scanptr); c_scanRecordPool.getPtr(scanptr);
ScanRecord* scanP = scanptr.p;
releaseActiveFrag(signal); releaseActiveFrag(signal);
if (tcConnectptr.p->errorCode != 0) { if (tcConnectptr.p->errorCode != 0) {
jam(); jam();
...@@ -9304,9 +9364,23 @@ void Dblqh::copyTupkeyConfLab(Signal* signal) ...@@ -9304,9 +9364,23 @@ void Dblqh::copyTupkeyConfLab(Signal* signal)
closeCopyLab(signal); closeCopyLab(signal);
return; return;
}//if }//if
TcConnectionrec * tcConP = tcConnectptr.p;
tcConnectptr.p->totSendlenAi = readLength; tcConnectptr.p->totSendlenAi = readLength;
tcConnectptr.p->connectState = TcConnectionrec::COPY_CONNECTED; tcConnectptr.p->connectState = TcConnectionrec::COPY_CONNECTED;
calculateHash(signal);
// Read primary keys (used to get here via scan keyinfo)
Uint32* tmp = signal->getDataPtrSend()+24;
Uint32 len= tcConnectptr.p->primKeyLen = readPrimaryKeys(scanP, tcConP, tmp);
// Calculate hash (no need to linearies key)
tcConnectptr.p->hashValue = md5_hash((Uint64*)tmp, len);
// Move into databuffer to make packLqhkeyreqLab happy
memcpy(tcConP->tupkeyData, tmp, 4*4);
if(len > 4)
keyinfoLab(tmp+4, tmp + len);
LqhKeyReq::setKeyLen(tcConP->reqinfo, len);
/*---------------------------------------------------------------------------*/ /*---------------------------------------------------------------------------*/
// To avoid using up to many operation records in ACC we will increase the // To avoid using up to many operation records in ACC we will increase the
// constant to ensure that we never send more than 40 records at a time. // constant to ensure that we never send more than 40 records at a time.
...@@ -9317,7 +9391,7 @@ void Dblqh::copyTupkeyConfLab(Signal* signal) ...@@ -9317,7 +9391,7 @@ void Dblqh::copyTupkeyConfLab(Signal* signal)
// records to ensure that node recovery does not fail because of simultaneous // records to ensure that node recovery does not fail because of simultaneous
// scanning. // scanning.
/*---------------------------------------------------------------------------*/ /*---------------------------------------------------------------------------*/
UintR TnoOfWords = readLength + tcConnectptr.p->primKeyLen; UintR TnoOfWords = readLength + len;
TnoOfWords = TnoOfWords + MAGIC_CONSTANT; TnoOfWords = TnoOfWords + MAGIC_CONSTANT;
TnoOfWords = TnoOfWords + (TnoOfWords >> 2); TnoOfWords = TnoOfWords + (TnoOfWords >> 2);
...@@ -9639,7 +9713,6 @@ void Dblqh::closeCopyRequestLab(Signal* signal) ...@@ -9639,7 +9713,6 @@ void Dblqh::closeCopyRequestLab(Signal* signal)
scanptr.p->scanErrorCounter++; scanptr.p->scanErrorCounter++;
switch (scanptr.p->scanState) { switch (scanptr.p->scanState) {
case ScanRecord::WAIT_TUPKEY_COPY: case ScanRecord::WAIT_TUPKEY_COPY:
case ScanRecord::WAIT_COPY_KEYINFO:
case ScanRecord::WAIT_NEXT_SCAN_COPY: case ScanRecord::WAIT_NEXT_SCAN_COPY:
jam(); jam();
/*---------------------------------------------------------------------------*/ /*---------------------------------------------------------------------------*/
...@@ -9870,11 +9943,6 @@ void Dblqh::execCOPY_STATEREQ(Signal* signal) ...@@ -9870,11 +9943,6 @@ void Dblqh::execCOPY_STATEREQ(Signal* signal)
void Dblqh::initCopyTc(Signal* signal) void Dblqh::initCopyTc(Signal* signal)
{ {
const NextScanConf * const nextScanConf = (NextScanConf *)&signal->theData[0]; const NextScanConf * const nextScanConf = (NextScanConf *)&signal->theData[0];
tcConnectptr.p->primKeyLen = nextScanConf->keyLength;
tcConnectptr.p->tupkeyData[0] = nextScanConf->key[0];
tcConnectptr.p->tupkeyData[1] = nextScanConf->key[1];
tcConnectptr.p->tupkeyData[2] = nextScanConf->key[2];
tcConnectptr.p->tupkeyData[3] = nextScanConf->key[3];
scanptr.p->scanLocalref[0] = nextScanConf->localKey[0]; scanptr.p->scanLocalref[0] = nextScanConf->localKey[0];
scanptr.p->scanLocalref[1] = nextScanConf->localKey[1]; scanptr.p->scanLocalref[1] = nextScanConf->localKey[1];
scanptr.p->scanLocalFragid = nextScanConf->fragId; scanptr.p->scanLocalFragid = nextScanConf->fragId;
...@@ -9883,7 +9951,6 @@ void Dblqh::initCopyTc(Signal* signal) ...@@ -9883,7 +9951,6 @@ void Dblqh::initCopyTc(Signal* signal)
tcConnectptr.p->opExec = 0; /* NOT INTERPRETED MODE */ tcConnectptr.p->opExec = 0; /* NOT INTERPRETED MODE */
tcConnectptr.p->schemaVersion = scanptr.p->scanSchemaVersion; tcConnectptr.p->schemaVersion = scanptr.p->scanSchemaVersion;
Uint32 reqinfo = 0; Uint32 reqinfo = 0;
LqhKeyReq::setKeyLen(reqinfo, nextScanConf->keyLength);
LqhKeyReq::setLockType(reqinfo, ZINSERT); LqhKeyReq::setLockType(reqinfo, ZINSERT);
LqhKeyReq::setDirtyFlag(reqinfo, 1); LqhKeyReq::setDirtyFlag(reqinfo, 1);
LqhKeyReq::setSimpleFlag(reqinfo, 1); LqhKeyReq::setSimpleFlag(reqinfo, 1);
......
ndb/src/kernel/blocks/dbtux/Dbtux.hpp
View file @ 46075704
...@@ -406,7 +406,6 @@ private: ...@@ -406,7 +406,6 @@ private:
Uint32 m_accLockOp; Uint32 m_accLockOp;
Uint8 m_readCommitted; // no locking Uint8 m_readCommitted; // no locking
Uint8 m_lockMode; Uint8 m_lockMode;
Uint8 m_keyInfo;
ScanBound m_boundMin; ScanBound m_boundMin;
ScanBound m_boundMax; ScanBound m_boundMax;
ScanBound* m_bound[2]; // pointers to above 2 ScanBound* m_bound[2]; // pointers to above 2
...@@ -1030,7 +1029,6 @@ Dbtux::ScanOp::ScanOp(ScanBoundPool& scanBoundPool) : ...@@ -1030,7 +1029,6 @@ Dbtux::ScanOp::ScanOp(ScanBoundPool& scanBoundPool) :
m_accLockOp(RNIL), m_accLockOp(RNIL),
m_readCommitted(0), m_readCommitted(0),
m_lockMode(0), m_lockMode(0),
m_keyInfo(0),
m_boundMin(scanBoundPool), m_boundMin(scanBoundPool),
m_boundMax(scanBoundPool), m_boundMax(scanBoundPool),
m_scanPos(), m_scanPos(),
......
ndb/src/kernel/blocks/dbtux/DbtuxDebug.cpp
View file @ 46075704
...@@ -347,7 +347,6 @@ operator<<(NdbOut& out, const Dbtux::ScanOp& scan) ...@@ -347,7 +347,6 @@ operator<<(NdbOut& out, const Dbtux::ScanOp& scan)
out << "]"; out << "]";
out << " [readCommitted " << dec << scan.m_readCommitted << "]"; out << " [readCommitted " << dec << scan.m_readCommitted << "]";
out << " [lockMode " << dec << scan.m_lockMode << "]"; out << " [lockMode " << dec << scan.m_lockMode << "]";
out << " [keyInfo " << dec << scan.m_keyInfo << "]";
out << " [pos " << scan.m_scanPos << "]"; out << " [pos " << scan.m_scanPos << "]";
out << " [ent " << scan.m_scanEnt << "]"; out << " [ent " << scan.m_scanEnt << "]";
for (unsigned i = 0; i <= 1; i++) { for (unsigned i = 0; i <= 1; i++) {
......
ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp
View file @ 46075704
...@@ -73,7 +73,6 @@ Dbtux::execACC_SCANREQ(Signal* signal) ...@@ -73,7 +73,6 @@ Dbtux::execACC_SCANREQ(Signal* signal)
scanPtr.p->m_savePointId = req->savePointId; scanPtr.p->m_savePointId = req->savePointId;
scanPtr.p->m_readCommitted = AccScanReq::getReadCommittedFlag(req->requestInfo); scanPtr.p->m_readCommitted = AccScanReq::getReadCommittedFlag(req->requestInfo);
scanPtr.p->m_lockMode = AccScanReq::getLockMode(req->requestInfo); scanPtr.p->m_lockMode = AccScanReq::getLockMode(req->requestInfo);
scanPtr.p->m_keyInfo = AccScanReq::getKeyinfoFlag(req->requestInfo);
#ifdef VM_TRACE #ifdef VM_TRACE
if (debugFlags & DebugScan) { if (debugFlags & DebugScan) {
debugOut << "Seize scan " << scanPtr.i << " " << *scanPtr.p << endl; debugOut << "Seize scan " << scanPtr.i << " " << *scanPtr.p << endl;
...@@ -473,13 +472,6 @@ Dbtux::execACC_CHECK_SCAN(Signal* signal) ...@@ -473,13 +472,6 @@ Dbtux::execACC_CHECK_SCAN(Signal* signal)
jam(); jam();
// read keys if not already done (uses signal) // read keys if not already done (uses signal)
const TreeEnt ent = scan.m_scanEnt; const TreeEnt ent = scan.m_scanEnt;
if (scan.m_keyInfo) {
jam();
if (pkSize == 0) {
jam();
readTablePk(frag, ent, pkData, pkSize);
}
}
// conf signal // conf signal
NextScanConf* const conf = (NextScanConf*)signal->getDataPtrSend(); NextScanConf* const conf = (NextScanConf*)signal->getDataPtrSend();
conf->scanPtr = scan.m_userPtr; conf->scanPtr = scan.m_userPtr;
...@@ -501,15 +493,6 @@ Dbtux::execACC_CHECK_SCAN(Signal* signal) ...@@ -501,15 +493,6 @@ Dbtux::execACC_CHECK_SCAN(Signal* signal)
conf->localKeyLength = 1; conf->localKeyLength = 1;
unsigned signalLength = 6; unsigned signalLength = 6;
// add key info // add key info
if (scan.m_keyInfo) {
jam();
conf->keyLength = pkSize;
// piggy-back first 4 words of key data
for (unsigned i = 0; i < 4; i++) {
conf->key[i] = i < pkSize ? pkData[i] : 0;
}
signalLength = 11;
}
if (! scan.m_readCommitted) { if (! scan.m_readCommitted) {
sendSignal(scan.m_userRef, GSN_NEXT_SCANCONF, sendSignal(scan.m_userRef, GSN_NEXT_SCANCONF,
signal, signalLength, JBB); signal, signalLength, JBB);
...@@ -517,24 +500,6 @@ Dbtux::execACC_CHECK_SCAN(Signal* signal) ...@@ -517,24 +500,6 @@ Dbtux::execACC_CHECK_SCAN(Signal* signal)
Uint32 blockNo = refToBlock(scan.m_userRef); Uint32 blockNo = refToBlock(scan.m_userRef);
EXECUTE_DIRECT(blockNo, GSN_NEXT_SCANCONF, signal, signalLength); EXECUTE_DIRECT(blockNo, GSN_NEXT_SCANCONF, signal, signalLength);
} }
// send rest of key data
if (scan.m_keyInfo && pkSize > 4) {
unsigned total = 4;
while (total < pkSize) {
jam();
unsigned length = pkSize - total;
if (length > 20)
length = 20;
signal->theData[0] = scan.m_userPtr;
signal->theData[1] = 0;
signal->theData[2] = 0;
signal->theData[3] = length;
memcpy(&signal->theData[4], &pkData[total], length << 2);
sendSignal(scan.m_userRef, GSN_ACC_SCAN_INFO24,
signal, 4 + length, JBB);
total += length;
}
}
// next time look for next entry // next time look for next entry
scan.m_state = ScanOp::Next; scan.m_state = ScanOp::Next;
return; return;
......
ndb/src/ndbapi/NdbScanOperation.cpp
View file @ 46075704
...@@ -129,7 +129,6 @@ NdbResultSet* NdbScanOperation::readTuples(NdbScanOperation::LockMode lm, ...@@ -129,7 +129,6 @@ NdbResultSet* NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
Uint32 parallel) Uint32 parallel)
{ {
m_ordered = 0; m_ordered = 0;
Uint32 fragCount = m_currentTable->m_fragmentCount; Uint32 fragCount = m_currentTable->m_fragmentCount;
if (parallel > fragCount || parallel == 0) { if (parallel > fragCount || parallel == 0) {
...@@ -232,7 +231,7 @@ NdbResultSet* NdbScanOperation::readTuples(NdbScanOperation::LockMode lm, ...@@ -232,7 +231,7 @@ NdbResultSet* NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
tSignal->setSignal(GSN_KEYINFO); tSignal->setSignal(GSN_KEYINFO);
theKEYINFOptr = ((KeyInfo*)tSignal->getDataPtrSend())->keyData; theKEYINFOptr = ((KeyInfo*)tSignal->getDataPtrSend())->keyData;
theTotalNrOfKeyWordInSignal= 0; theTotalNrOfKeyWordInSignal= 0;
getFirstATTRINFOScan(); getFirstATTRINFOScan();
return getResultSet(); return getResultSet();
} }
...@@ -1205,6 +1204,9 @@ NdbIndexScanOperation::readTuples(LockMode lm, ...@@ -1205,6 +1204,9 @@ NdbIndexScanOperation::readTuples(LockMode lm,
#endif #endif
} }
} }
m_this_bound_start = 0;
m_first_bound_word = theKEYINFOptr;
return rs; return rs;
} }
...@@ -1573,7 +1575,8 @@ NdbScanOperation::restart() ...@@ -1573,7 +1575,8 @@ NdbScanOperation::restart()
} }
int int
NdbIndexScanOperation::reset_bounds(){ NdbIndexScanOperation::reset_bounds()
{
int res; int res;
{ {
...@@ -1593,6 +1596,8 @@ NdbIndexScanOperation::reset_bounds(){ ...@@ -1593,6 +1596,8 @@ NdbIndexScanOperation::reset_bounds(){
theTotalNrOfKeyWordInSignal = 0; theTotalNrOfKeyWordInSignal = 0;
theNoOfTupKeyLeft = m_accessTable->m_noOfDistributionKeys; theNoOfTupKeyLeft = m_accessTable->m_noOfDistributionKeys;
theDistrKeyIndicator_ = 0; theDistrKeyIndicator_ = 0;
m_this_bound_start = 0;
m_first_bound_word = theKEYINFOptr;
m_transConnection m_transConnection
->remove_list((NdbOperation*&)m_transConnection->m_firstExecutedScanOp, ->remove_list((NdbOperation*&)m_transConnection->m_firstExecutedScanOp,
this); this);
...@@ -1601,3 +1606,14 @@ NdbIndexScanOperation::reset_bounds(){ ...@@ -1601,3 +1606,14 @@ NdbIndexScanOperation::reset_bounds(){
} }
return res; return res;
} }
int
NdbIndexScanOperation::set_new_bound()
{
Uint32 bound_head = * m_first_bound_word;
bound_head |= (theTupKeyLen - m_this_bound_start) << 16;
* m_first_bound_word = bound_head;
m_first_bound_word = theKEYINFOptr;
m_this_bound_start = theTupKeyLen;
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7