Commit 5aaffe9b authored by unknown's avatar unknown

bug#6775 - ndb - fix bug introduced by bug fix


ndb/src/kernel/blocks/dblqh/DblqhMain.cpp:
  Scan number allocation must be on table even in case of range scan
  Otherwise scan takeover won't work as LQHKEYREQ contains table
parent 98de7e01
...@@ -8573,12 +8573,10 @@ Uint32 Dblqh::initScanrec(const ScanFragReq* scanFragReq) ...@@ -8573,12 +8573,10 @@ Uint32 Dblqh::initScanrec(const ScanFragReq* scanFragReq)
/** /**
* Used for scan take over * Used for scan take over
*/ */
{
FragrecordPtr tFragPtr; FragrecordPtr tFragPtr;
tFragPtr.i = fragptr.p->tableFragptr; tFragPtr.i = fragptr.p->tableFragptr;
ptrCheckGuard(tFragPtr, cfragrecFileSize, fragrecord); ptrCheckGuard(tFragPtr, cfragrecFileSize, fragrecord);
scanptr.p->fragPtrI = fragptr.p->tableFragptr; scanptr.p->fragPtrI = fragptr.p->tableFragptr;
}
/** /**
* !idx uses 1 - (MAX_PARALLEL_SCANS_PER_FRAG - 1) = 1-11 * !idx uses 1 - (MAX_PARALLEL_SCANS_PER_FRAG - 1) = 1-11
...@@ -8587,7 +8585,7 @@ Uint32 Dblqh::initScanrec(const ScanFragReq* scanFragReq) ...@@ -8587,7 +8585,7 @@ Uint32 Dblqh::initScanrec(const ScanFragReq* scanFragReq)
Uint32 start = (idx ? MAX_PARALLEL_SCANS_PER_FRAG : 1 ); Uint32 start = (idx ? MAX_PARALLEL_SCANS_PER_FRAG : 1 );
Uint32 stop = (idx ? MAX_PARALLEL_INDEX_SCANS_PER_FRAG : MAX_PARALLEL_SCANS_PER_FRAG - 1); Uint32 stop = (idx ? MAX_PARALLEL_INDEX_SCANS_PER_FRAG : MAX_PARALLEL_SCANS_PER_FRAG - 1);
stop += start; stop += start;
Uint32 free = fragptr.p->m_scanNumberMask.find(start); Uint32 free = tFragPtr.p->m_scanNumberMask.find(start);
if(free == Fragrecord::ScanNumberMask::NotFound || free >= stop){ if(free == Fragrecord::ScanNumberMask::NotFound || free >= stop){
jam(); jam();
...@@ -8607,9 +8605,8 @@ Uint32 Dblqh::initScanrec(const ScanFragReq* scanFragReq) ...@@ -8607,9 +8605,8 @@ Uint32 Dblqh::initScanrec(const ScanFragReq* scanFragReq)
return ZOK; return ZOK;
} }
scanptr.p->scanNumber = free; scanptr.p->scanNumber = free;
fragptr.p->m_scanNumberMask.clear(free);// Update mask tFragPtr.p->m_scanNumberMask.clear(free);// Update mask
LocalDLList<ScanRecord> active(c_scanRecordPool, fragptr.p->m_activeScans); LocalDLList<ScanRecord> active(c_scanRecordPool, fragptr.p->m_activeScans);
active.add(scanptr); active.add(scanptr);
...@@ -8693,8 +8690,12 @@ void Dblqh::finishScanrec(Signal* signal) ...@@ -8693,8 +8690,12 @@ void Dblqh::finishScanrec(Signal* signal)
LocalDLList<ScanRecord> scans(c_scanRecordPool, fragptr.p->m_activeScans); LocalDLList<ScanRecord> scans(c_scanRecordPool, fragptr.p->m_activeScans);
scans.release(scanptr); scans.release(scanptr);
FragrecordPtr tFragPtr;
tFragPtr.i = scanptr.p->fragPtrI;
ptrCheckGuard(tFragPtr, cfragrecFileSize, fragrecord);
const Uint32 scanNumber = scanptr.p->scanNumber; const Uint32 scanNumber = scanptr.p->scanNumber;
ndbrequire(!fragptr.p->m_scanNumberMask.get(scanNumber)); ndbrequire(!tFragPtr.p->m_scanNumberMask.get(scanNumber));
ScanRecordPtr restart; ScanRecordPtr restart;
/** /**
...@@ -8702,13 +8703,13 @@ void Dblqh::finishScanrec(Signal* signal) ...@@ -8702,13 +8703,13 @@ void Dblqh::finishScanrec(Signal* signal)
*/ */
if(scanNumber == NR_ScanNo || !queue.first(restart)){ if(scanNumber == NR_ScanNo || !queue.first(restart)){
jam(); jam();
fragptr.p->m_scanNumberMask.set(scanNumber); tFragPtr.p->m_scanNumberMask.set(scanNumber);
return; return;
} }
if(ERROR_INSERTED(5034)){ if(ERROR_INSERTED(5034)){
jam(); jam();
fragptr.p->m_scanNumberMask.set(scanNumber); tFragPtr.p->m_scanNumberMask.set(scanNumber);
return; return;
} }
......
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