Commit 64077103 authored by pekka@mysql.com's avatar pekka@mysql.com

tux optim 9 - use TUP method to read search key

parent 26d2d453
...@@ -1003,17 +1003,22 @@ public: ...@@ -1003,17 +1003,22 @@ public:
/* /*
* TUX index in TUP has single Uint32 array attribute which stores an * TUX index in TUP has single Uint32 array attribute which stores an
* index node. TUX uses following methods. * index node. TUX reads and writes the node directly via pointer.
*/ */
int tuxAllocNode(Signal* signal, Uint32 fragPtrI, Uint32& pageId, Uint32& pageOffset, Uint32*& node); int tuxAllocNode(Signal* signal, Uint32 fragPtrI, Uint32& pageId, Uint32& pageOffset, Uint32*& node);
void tuxFreeNode(Signal* signal, Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32* node); void tuxFreeNode(Signal* signal, Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32* node);
void tuxGetNode(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32*& node); void tuxGetNode(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32*& node);
/* /*
* TUX reads primary table attributes for 1) index key 2) primary key * TUX reads primary table attributes for index keys. Input is
* when returning keyinfo. TUX uses following methods. * attribute ids in AttributeHeader format. Output is pointers to
* attribute data within tuple or 0 for NULL value.
*/
void tuxReadAttrs(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tupVersion, Uint32 numAttrs, const Uint32* attrIds, const Uint32** attrData);
/*
* TUX reads primary key for md5 summing and when returning keyinfo.
*/ */
void tuxReadAttrs(); // under construction
void tuxReadKeys(); // under construction void tuxReadKeys(); // under construction
private: private:
......
...@@ -33,6 +33,7 @@ ...@@ -33,6 +33,7 @@
void void
Dbtup::tuxGetTupAddr(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32& tupAddr) Dbtup::tuxGetTupAddr(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32& tupAddr)
{ {
ljamEntry();
FragrecordPtr fragPtr; FragrecordPtr fragPtr;
fragPtr.i = fragPtrI; fragPtr.i = fragPtrI;
ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord); ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
...@@ -54,6 +55,7 @@ Dbtup::tuxGetTupAddr(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32& ...@@ -54,6 +55,7 @@ Dbtup::tuxGetTupAddr(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32&
int int
Dbtup::tuxAllocNode(Signal* signal, Uint32 fragPtrI, Uint32& pageId, Uint32& pageOffset, Uint32*& node) Dbtup::tuxAllocNode(Signal* signal, Uint32 fragPtrI, Uint32& pageId, Uint32& pageOffset, Uint32*& node)
{ {
ljamEntry();
FragrecordPtr fragPtr; FragrecordPtr fragPtr;
fragPtr.i = fragPtrI; fragPtr.i = fragPtrI;
ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord); ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
...@@ -63,7 +65,7 @@ Dbtup::tuxAllocNode(Signal* signal, Uint32 fragPtrI, Uint32& pageId, Uint32& pag ...@@ -63,7 +65,7 @@ Dbtup::tuxAllocNode(Signal* signal, Uint32 fragPtrI, Uint32& pageId, Uint32& pag
PagePtr pagePtr; PagePtr pagePtr;
terrorCode = 0; terrorCode = 0;
if (! allocTh(fragPtr.p, tablePtr.p, NORMAL_PAGE, signal, pageOffset, pagePtr)) { if (! allocTh(fragPtr.p, tablePtr.p, NORMAL_PAGE, signal, pageOffset, pagePtr)) {
jam(); ljam();
ndbrequire(terrorCode != 0); ndbrequire(terrorCode != 0);
return terrorCode; return terrorCode;
} }
...@@ -77,6 +79,7 @@ Dbtup::tuxAllocNode(Signal* signal, Uint32 fragPtrI, Uint32& pageId, Uint32& pag ...@@ -77,6 +79,7 @@ Dbtup::tuxAllocNode(Signal* signal, Uint32 fragPtrI, Uint32& pageId, Uint32& pag
void void
Dbtup::tuxFreeNode(Signal* signal, Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32* node) Dbtup::tuxFreeNode(Signal* signal, Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32* node)
{ {
ljamEntry();
FragrecordPtr fragPtr; FragrecordPtr fragPtr;
fragPtr.i = fragPtrI; fragPtr.i = fragPtrI;
ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord); ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
...@@ -95,6 +98,7 @@ Dbtup::tuxFreeNode(Signal* signal, Uint32 fragPtrI, Uint32 pageId, Uint32 pageOf ...@@ -95,6 +98,7 @@ Dbtup::tuxFreeNode(Signal* signal, Uint32 fragPtrI, Uint32 pageId, Uint32 pageOf
void void
Dbtup::tuxGetNode(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32*& node) Dbtup::tuxGetNode(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32*& node)
{ {
ljamEntry();
FragrecordPtr fragPtr; FragrecordPtr fragPtr;
fragPtr.i = fragPtrI; fragPtr.i = fragPtrI;
ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord); ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
...@@ -109,9 +113,62 @@ Dbtup::tuxGetNode(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32*& no ...@@ -109,9 +113,62 @@ Dbtup::tuxGetNode(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32*& no
node = &pagePtr.p->pageWord[pageOffset] + attrDataOffset; node = &pagePtr.p->pageWord[pageOffset] + attrDataOffset;
} }
void // under construction void
Dbtup::tuxReadAttrs() Dbtup::tuxReadAttrs(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tupVersion, Uint32 numAttrs, const Uint32* attrIds, const Uint32** attrData)
{ {
ljamEntry();
FragrecordPtr fragPtr;
fragPtr.i = fragPtrI;
ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
TablerecPtr tablePtr;
tablePtr.i = fragPtr.p->fragTableId;
ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
PagePtr pagePtr;
pagePtr.i = pageId;
ptrCheckGuard(pagePtr, cnoOfPage, page);
// search for tuple version if not original
if (pagePtr.p->pageWord[pageOffset + 1] != tupVersion) {
ljam();
OperationrecPtr opPtr;
opPtr.i = pagePtr.p->pageWord[pageOffset];
Uint32 loopGuard = 0;
while (true) {
ptrCheckGuard(opPtr, cnoOfOprec, operationrec);
if (opPtr.p->realPageIdC != RNIL) {
pagePtr.i = opPtr.p->realPageIdC;
pageOffset = opPtr.p->pageOffsetC;
ptrCheckGuard(pagePtr, cnoOfPage, page);
if (pagePtr.p->pageWord[pageOffset + 1] == tupVersion) {
ljam();
break;
}
}
ljam();
opPtr.i = opPtr.p->nextActiveOp;
ndbrequire(++loopGuard < (1 << ZTUP_VERSION_BITS));
}
}
const Uint32 tabDescriptor = tablePtr.p->tabDescriptor;
const Uint32* tupleHeader = &pagePtr.p->pageWord[pageOffset];
for (Uint32 i = 0; i < numAttrs; i++) {
AttributeHeader ah(attrIds[i]);
Uint32 attrId = ah.getAttributeId();
Uint32 index = tabDescriptor + (attrId << ZAD_LOG_SIZE);
Uint32 desc1 = tableDescriptor[index].tabDescr;
Uint32 desc2 = tableDescriptor[index + 1].tabDescr;
if (AttributeDescriptor::getNullable(desc1)) {
Uint32 offset = AttributeOffset::getNullFlagOffset(desc2);
ndbrequire(offset < tablePtr.p->tupNullWords);
offset += tablePtr.p->tupNullIndex;
ndbrequire(offset < tablePtr.p->tupheadsize);
if (AttributeOffset::isNULL(tupleHeader[offset], desc2)) {
ljam();
attrData[i] = 0;
continue;
}
}
attrData[i] = tupleHeader + AttributeOffset::getOffset(desc2);
}
} }
void // under construction void // under construction
...@@ -259,10 +316,10 @@ Dbtup::execTUP_QUERY_TH(Signal* signal) ...@@ -259,10 +316,10 @@ Dbtup::execTUP_QUERY_TH(Signal* signal)
for this transaction and savepoint id. If its tuple version equals for this transaction and savepoint id. If its tuple version equals
the requested then we have a visible tuple otherwise not. the requested then we have a visible tuple otherwise not.
*/ */
jam(); ljam();
Uint32 read_tupVersion = pagePtr.p->pageWord[tempOp.pageOffset + 1]; Uint32 read_tupVersion = pagePtr.p->pageWord[tempOp.pageOffset + 1];
if (read_tupVersion == req_tupVersion) { if (read_tupVersion == req_tupVersion) {
jam(); ljam();
ret_result = 1; ret_result = 1;
} }
} }
...@@ -580,7 +637,7 @@ Dbtup::buildIndex(Signal* signal, Uint32 buildPtrI) ...@@ -580,7 +637,7 @@ Dbtup::buildIndex(Signal* signal, Uint32 buildPtrI)
tuple as a copy tuple. The original tuple is stable and is thus tuple as a copy tuple. The original tuple is stable and is thus
preferrable to store in TUX. preferrable to store in TUX.
*/ */
jam(); ljam();
ptrCheckGuard(pageOperPtr, cnoOfOprec, operationrec); ptrCheckGuard(pageOperPtr, cnoOfOprec, operationrec);
realPageId = pageOperPtr.p->realPageId; realPageId = pageOperPtr.p->realPageId;
pageOffset = pageOperPtr.p->pageOffset; pageOffset = pageOperPtr.p->pageOffset;
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include <new> #include <new>
#include <ndb_limits.h> #include <ndb_limits.h>
#include <SimulatedBlock.hpp> #include <SimulatedBlock.hpp>
#include <AttributeDescriptor.hpp>
#include <AttributeHeader.hpp> #include <AttributeHeader.hpp>
#include <ArrayPool.hpp> #include <ArrayPool.hpp>
#include <DataBuffer.hpp> #include <DataBuffer.hpp>
...@@ -84,6 +85,10 @@ ...@@ -84,6 +85,10 @@
#define jam() jamLine(90000 + __LINE__) #define jam() jamLine(90000 + __LINE__)
#define jamEntry() jamEntryLine(90000 + __LINE__) #define jamEntry() jamEntryLine(90000 + __LINE__)
#endif #endif
#ifndef jam
#define jam() jamLine(__LINE__)
#define jamEntry() jamEntryLine(__LINE__)
#endif
#undef max #undef max
#undef min #undef min
...@@ -115,7 +120,7 @@ private: ...@@ -115,7 +120,7 @@ private:
struct DescEnt; struct DescEnt;
/* /*
* Pointer to Uint32 data. Interpretation is context dependent. * Pointer to array of Uint32.
*/ */
struct Data { struct Data {
private: private:
...@@ -131,7 +136,7 @@ private: ...@@ -131,7 +136,7 @@ private:
friend class Data; friend class Data;
/* /*
* Pointer to constant Uint32 data. * Pointer to array of constant Uint32.
*/ */
struct ConstData; struct ConstData;
friend struct ConstData; friend struct ConstData;
...@@ -153,6 +158,11 @@ private: ...@@ -153,6 +158,11 @@ private:
// AttributeHeader size is assumed to be 1 word // AttributeHeader size is assumed to be 1 word
static const unsigned AttributeHeaderSize = 1; static const unsigned AttributeHeaderSize = 1;
/*
* Array of pointers to TUP table attributes. Always read-on|y.
*/
typedef const Uint32** TableData;
/* /*
* Logical tuple address, "local key". Identifies table tuples. * Logical tuple address, "local key". Identifies table tuples.
*/ */
...@@ -560,25 +570,11 @@ private: ...@@ -560,25 +570,11 @@ private:
struct SearchPar; struct SearchPar;
friend struct SearchPar; friend struct SearchPar;
struct SearchPar { struct SearchPar {
ConstData m_data; // input index key values TableData m_data; // input index key values
TreeEnt m_ent; // input tuple and version TreeEnt m_ent; // input tuple and version
SearchPar(); SearchPar();
}; };
/*
* Attribute data comparison.
*/
struct CmpPar;
friend struct CmpPar;
struct CmpPar {
ConstData m_data1; // full search key
ConstData m_data2; // full or prefix data
unsigned m_len2; // words in data2 buffer
unsigned m_first; // first attribute
unsigned m_numEq; // number of initial equal attributes
CmpPar();
};
/* /*
* Scan bound comparison. * Scan bound comparison.
*/ */
...@@ -672,7 +668,7 @@ private: ...@@ -672,7 +668,7 @@ private:
/* /*
* DbtuxCmp.cpp * DbtuxCmp.cpp
*/ */
int cmpTreeAttrs(const Frag& frag, CmpPar& cmpPar); int cmpSearchKey(const Frag& frag, unsigned& start, TableData data1, ConstData data2, unsigned size2 = MaxAttrDataSize);
int cmpScanBound(const Frag& frag, const BoundPar boundPar); int cmpScanBound(const Frag& frag, const BoundPar boundPar);
/* /*
...@@ -716,12 +712,23 @@ private: ...@@ -716,12 +712,23 @@ private:
Uint32 c_internalStartPhase; Uint32 c_internalStartPhase;
Uint32 c_typeOfStart; Uint32 c_typeOfStart;
// buffers // buffer for scan bounds and keyinfo (primary key)
Data c_keyBuffer; // search key or scan bound Data c_dataBuffer;
// array of index key attribute ids in AttributeHeader format
Data c_keyAttrs;
// search key data as pointers to TUP storage
TableData c_searchKey;
// current entry key data as pointers to TUP storage
TableData c_entryKey;
// inlined utils // inlined utils
DescEnt& getDescEnt(Uint32 descPage, Uint32 descOff); DescEnt& getDescEnt(Uint32 descPage, Uint32 descOff);
Uint32 getTupAddr(const Frag& frag, const TreeEnt ent); Uint32 getTupAddr(const Frag& frag, TreeEnt ent);
void setKeyAttrs(const Frag& frag, Data keyAttrs);
void readKeyAttrs(const Frag& frag, TreeEnt ent, unsigned start, ConstData keyAttrs, TableData keyData);
static unsigned min(unsigned x, unsigned y); static unsigned min(unsigned x, unsigned y);
static unsigned max(unsigned x, unsigned y); static unsigned max(unsigned x, unsigned y);
}; };
...@@ -1218,16 +1225,6 @@ Dbtux::SearchPar::SearchPar() : ...@@ -1218,16 +1225,6 @@ Dbtux::SearchPar::SearchPar() :
{ {
} }
inline
Dbtux::CmpPar::CmpPar() :
m_data1(0),
m_data2(0),
m_len2(0),
m_first(0),
m_numEq(0)
{
}
inline inline
Dbtux::BoundPar::BoundPar() : Dbtux::BoundPar::BoundPar() :
m_data1(0), m_data1(0),
...@@ -1267,15 +1264,43 @@ Dbtux::getDescEnt(Uint32 descPage, Uint32 descOff) ...@@ -1267,15 +1264,43 @@ Dbtux::getDescEnt(Uint32 descPage, Uint32 descOff)
} }
inline Uint32 inline Uint32
Dbtux::getTupAddr(const Frag& frag, const TreeEnt ent) Dbtux::getTupAddr(const Frag& frag, TreeEnt ent)
{ {
const Uint32 tableFragPtrI = frag.m_tupTableFragPtrI[ent.m_fragBit]; const Uint32 tableFragPtrI = frag.m_tupTableFragPtrI[ent.m_fragBit];
const TupLoc tupLoc = ent.m_tupLoc; const TupLoc tupLoc = ent.m_tupLoc;
Uint32 tupAddr = NullTupAddr; Uint32 tupAddr = NullTupAddr;
c_tup->tuxGetTupAddr(tableFragPtrI, tupLoc.m_pageId, tupLoc.m_pageOffset, tupAddr); c_tup->tuxGetTupAddr(tableFragPtrI, tupLoc.m_pageId, tupLoc.m_pageOffset, tupAddr);
jamEntry();
return tupAddr; return tupAddr;
} }
inline void
Dbtux::setKeyAttrs(const Frag& frag, Data keyAttrs)
{
const unsigned numAttrs = frag.m_numAttrs;
const DescEnt& descEnt = getDescEnt(frag.m_descPage, frag.m_descOff);
for (unsigned i = 0; i < numAttrs; i++) {
const DescAttr& descAttr = descEnt.m_descAttr[i];
Uint32 size = AttributeDescriptor::getSizeInWords(descAttr.m_attrDesc);
keyAttrs.ah() = AttributeHeader(descAttr.m_primaryAttrId, size);
keyAttrs += 1;
}
}
inline void
Dbtux::readKeyAttrs(const Frag& frag, TreeEnt ent, unsigned start, ConstData keyAttrs, TableData keyData)
{
const Uint32 tableFragPtrI = frag.m_tupTableFragPtrI[ent.m_fragBit];
const TupLoc tupLoc = ent.m_tupLoc;
const Uint32 tupVersion = ent.m_tupVersion;
ndbrequire(start < frag.m_numAttrs);
const unsigned numAttrs = frag.m_numAttrs - start;
keyAttrs += start;
keyData += start;
c_tup->tuxReadAttrs(tableFragPtrI, tupLoc.m_pageId, tupLoc.m_pageOffset, tupVersion, numAttrs, keyAttrs, keyData);
jamEntry();
}
inline unsigned inline unsigned
Dbtux::min(unsigned x, unsigned y) Dbtux::min(unsigned x, unsigned y)
{ {
......
...@@ -25,43 +25,35 @@ ...@@ -25,43 +25,35 @@
* many (additional) initial attributes were equal. * many (additional) initial attributes were equal.
*/ */
int int
Dbtux::cmpTreeAttrs(const Frag& frag, CmpPar& cmpPar) Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData data1, ConstData data2, unsigned size2)
{ {
const unsigned numAttrs = frag.m_numAttrs;
const DescEnt& descEnt = getDescEnt(frag.m_descPage, frag.m_descOff); const DescEnt& descEnt = getDescEnt(frag.m_descPage, frag.m_descOff);
ConstData data1 = cmpPar.m_data1;
ConstData data2 = cmpPar.m_data2;
// number of words of attribute data left // number of words of attribute data left
unsigned len2 = cmpPar.m_len2; unsigned len2 = size2;
const unsigned numAttrs = frag.m_numAttrs; // skip to right position in search key
unsigned index = cmpPar.m_first; data1 += start;
ndbrequire(index < numAttrs);
// skip to right position in search key XXX do it before the call
for (unsigned i = 0; i < index; i++) {
jam();
data1 += AttributeHeaderSize + data1.ah().getDataSize();
}
unsigned numEq = 0;
int ret = 0; int ret = 0;
while (index < numAttrs) { while (start < numAttrs) {
if (len2 < AttributeHeaderSize) { if (len2 < AttributeHeaderSize) {
jam(); jam();
ret = NdbSqlUtil::CmpUnknown; ret = NdbSqlUtil::CmpUnknown;
break; break;
} }
len2 -= AttributeHeaderSize; len2 -= AttributeHeaderSize;
if (! data1.ah().isNULL()) { if (*data1 != 0) {
if (! data2.ah().isNULL()) { if (! data2.ah().isNULL()) {
jam(); jam();
// current attribute // current attribute
const DescAttr& descAttr = descEnt.m_descAttr[index]; const DescAttr& descAttr = descEnt.m_descAttr[start];
const unsigned typeId = descAttr.m_typeId; const unsigned typeId = descAttr.m_typeId;
// full data size // full data size
const unsigned size1 = data1.ah().getDataSize(); const unsigned size1 = AttributeDescriptor::getSizeInWords(descAttr.m_attrDesc);
ndbrequire(size1 != 0 && size1 == data2.ah().getDataSize()); ndbrequire(size1 != 0 && size1 == data2.ah().getDataSize());
const unsigned size2 = min(size1, len2); const unsigned size2 = min(size1, len2);
len2 -= size2; len2 -= size2;
// compare // compare
const Uint32* const p1 = &data1[AttributeHeaderSize]; const Uint32* const p1 = *data1;
const Uint32* const p2 = &data2[AttributeHeaderSize]; const Uint32* const p2 = &data2[AttributeHeaderSize];
ret = NdbSqlUtil::cmp(typeId, p1, p2, size1, size2); ret = NdbSqlUtil::cmp(typeId, p1, p2, size1, size2);
if (ret != 0) { if (ret != 0) {
...@@ -82,14 +74,12 @@ Dbtux::cmpTreeAttrs(const Frag& frag, CmpPar& cmpPar) ...@@ -82,14 +74,12 @@ Dbtux::cmpTreeAttrs(const Frag& frag, CmpPar& cmpPar)
break; break;
} }
} }
data1 += AttributeHeaderSize + data1.ah().getDataSize(); data1 += 1;
data2 += AttributeHeaderSize + data2.ah().getDataSize(); data2 += AttributeHeaderSize + data2.ah().getDataSize();
numEq++; start++;
index++;
} }
// XXX until data format errors are handled // XXX until data format errors are handled
ndbrequire(ret != NdbSqlUtil::CmpError); ndbrequire(ret != NdbSqlUtil::CmpError);
cmpPar.m_numEq += numEq; // add to previous count
return ret; return ret;
} }
......
...@@ -30,7 +30,7 @@ Dbtux::Dbtux(const Configuration& conf) : ...@@ -30,7 +30,7 @@ Dbtux::Dbtux(const Configuration& conf) :
#endif #endif
c_internalStartPhase(0), c_internalStartPhase(0),
c_typeOfStart(NodeState::ST_ILLEGAL_TYPE), c_typeOfStart(NodeState::ST_ILLEGAL_TYPE),
c_keyBuffer(0) c_dataBuffer(0)
{ {
BLOCK_CONSTRUCTOR(Dbtux); BLOCK_CONSTRUCTOR(Dbtux);
// verify size assumptions (also when release-compiled) // verify size assumptions (also when release-compiled)
...@@ -195,7 +195,10 @@ Dbtux::execREAD_CONFIG_REQ(Signal* signal) ...@@ -195,7 +195,10 @@ Dbtux::execREAD_CONFIG_REQ(Signal* signal)
new (indexPtr.p) Index(); new (indexPtr.p) Index();
} }
// allocate buffers // allocate buffers
c_keyBuffer = (Uint32*)allocRecord("c_keyBuffer", sizeof(Uint64), (MaxAttrDataSize + 1) >> 1); c_dataBuffer = (Uint32*)allocRecord("c_dataBuffer", sizeof(Uint64), (MaxAttrDataSize + 1) >> 1);
c_keyAttrs = (Uint32*)allocRecord("c_keyAttrs", sizeof(Uint32), MaxIndexAttributes);
c_searchKey = (TableData)allocRecord("c_searchKey", sizeof(Uint32*), MaxIndexAttributes);
c_entryKey = (TableData)allocRecord("c_entryKey", sizeof(Uint32*), MaxIndexAttributes);
// ack // ack
ReadConfigConf * conf = (ReadConfigConf*)signal->getDataPtrSend(); ReadConfigConf * conf = (ReadConfigConf*)signal->getDataPtrSend();
conf->senderRef = reference(); conf->senderRef = reference();
......
...@@ -73,30 +73,25 @@ Dbtux::execTUX_MAINT_REQ(Signal* signal) ...@@ -73,30 +73,25 @@ Dbtux::execTUX_MAINT_REQ(Signal* signal)
} }
ndbrequire(fragPtr.i != RNIL); ndbrequire(fragPtr.i != RNIL);
Frag& frag = *fragPtr.p; Frag& frag = *fragPtr.p;
// set up index entry // set up index keys for this operation
setKeyAttrs(frag, c_keyAttrs);
// set up search entry
TreeEnt ent; TreeEnt ent;
ent.m_tupLoc = TupLoc(req->pageId, req->pageOffset); ent.m_tupLoc = TupLoc(req->pageId, req->pageOffset);
ent.m_tupVersion = req->tupVersion; ent.m_tupVersion = req->tupVersion;
ent.m_fragBit = fragBit; ent.m_fragBit = fragBit;
// read search key // read search key
ReadPar readPar; readKeyAttrs(frag, ent, 0, c_keyAttrs, c_searchKey);
readPar.m_ent = ent;
readPar.m_first = 0;
readPar.m_count = frag.m_numAttrs;
// output goes here
readPar.m_data = c_keyBuffer;
tupReadAttrs(signal, frag, readPar);
// check if all keys are null // check if all keys are null
{ {
const unsigned numAttrs = frag.m_numAttrs;
bool allNull = true; bool allNull = true;
ConstData data = readPar.m_data; for (unsigned i = 0; i < numAttrs; i++) {
for (unsigned i = 0; i < frag.m_numAttrs; i++) { if (c_searchKey[i] != 0) {
if (! data.ah().isNULL()) {
jam(); jam();
allNull = false; allNull = false;
break; break;
} }
data += AttributeHeaderSize + data.ah().getDataSize();
} }
if (allNull) { if (allNull) {
jam(); jam();
...@@ -107,7 +102,7 @@ Dbtux::execTUX_MAINT_REQ(Signal* signal) ...@@ -107,7 +102,7 @@ Dbtux::execTUX_MAINT_REQ(Signal* signal)
} }
// find position in tree // find position in tree
SearchPar searchPar; SearchPar searchPar;
searchPar.m_data = c_keyBuffer; searchPar.m_data = c_searchKey;
searchPar.m_ent = ent; searchPar.m_ent = ent;
TreePos treePos; TreePos treePos;
#ifdef VM_TRACE #ifdef VM_TRACE
......
...@@ -28,6 +28,7 @@ Dbtux::allocNode(Signal* signal, NodeHandle& node) ...@@ -28,6 +28,7 @@ Dbtux::allocNode(Signal* signal, NodeHandle& node)
Uint32 pageOffset = NullTupLoc.m_pageOffset; Uint32 pageOffset = NullTupLoc.m_pageOffset;
Uint32* node32 = 0; Uint32* node32 = 0;
int errorCode = c_tup->tuxAllocNode(signal, frag.m_tupIndexFragPtrI, pageId, pageOffset, node32); int errorCode = c_tup->tuxAllocNode(signal, frag.m_tupIndexFragPtrI, pageId, pageOffset, node32);
jamEntry();
if (errorCode == 0) { if (errorCode == 0) {
jam(); jam();
node.m_loc = TupLoc(pageId, pageOffset); node.m_loc = TupLoc(pageId, pageOffset);
...@@ -63,6 +64,7 @@ Dbtux::selectNode(Signal* signal, NodeHandle& node, TupLoc loc, AccSize acc) ...@@ -63,6 +64,7 @@ Dbtux::selectNode(Signal* signal, NodeHandle& node, TupLoc loc, AccSize acc)
Uint32 pageOffset = loc.m_pageOffset; Uint32 pageOffset = loc.m_pageOffset;
Uint32* node32 = 0; Uint32* node32 = 0;
c_tup->tuxGetNode(frag.m_tupIndexFragPtrI, pageId, pageOffset, node32); c_tup->tuxGetNode(frag.m_tupIndexFragPtrI, pageId, pageOffset, node32);
jamEntry();
node.m_loc = loc; node.m_loc = loc;
node.m_node = reinterpret_cast<TreeNode*>(node32); node.m_node = reinterpret_cast<TreeNode*>(node32);
node.m_acc = AccNone; node.m_acc = AccNone;
...@@ -103,6 +105,7 @@ Dbtux::deleteNode(Signal* signal, NodeHandle& node) ...@@ -103,6 +105,7 @@ Dbtux::deleteNode(Signal* signal, NodeHandle& node)
Uint32 pageOffset = loc.m_pageOffset; Uint32 pageOffset = loc.m_pageOffset;
Uint32* node32 = reinterpret_cast<Uint32*>(node.m_node); Uint32* node32 = reinterpret_cast<Uint32*>(node.m_node);
c_tup->tuxFreeNode(signal, frag.m_tupIndexFragPtrI, pageId, pageOffset, node32); c_tup->tuxFreeNode(signal, frag.m_tupIndexFragPtrI, pageId, pageOffset, node32);
jamEntry();
// invalidate handle and storage // invalidate handle and storage
node.m_loc = NullTupLoc; node.m_loc = NullTupLoc;
node.m_node = 0; node.m_node = 0;
......
...@@ -390,7 +390,7 @@ Dbtux::execACC_CHECK_SCAN(Signal* signal) ...@@ -390,7 +390,7 @@ Dbtux::execACC_CHECK_SCAN(Signal* signal)
const TreeEnt ent = scan.m_scanPos.m_ent; const TreeEnt ent = scan.m_scanPos.m_ent;
// read tuple key // read tuple key
keyPar.m_ent = ent; keyPar.m_ent = ent;
keyPar.m_data = c_keyBuffer; keyPar.m_data = c_dataBuffer;
tupReadKeys(signal, frag, keyPar); tupReadKeys(signal, frag, keyPar);
// get read lock or exclusive lock // get read lock or exclusive lock
AccLockReq* const lockReq = (AccLockReq*)signal->getDataPtrSend(); AccLockReq* const lockReq = (AccLockReq*)signal->getDataPtrSend();
...@@ -483,7 +483,7 @@ Dbtux::execACC_CHECK_SCAN(Signal* signal) ...@@ -483,7 +483,7 @@ Dbtux::execACC_CHECK_SCAN(Signal* signal)
if (keyPar.m_data == 0) { if (keyPar.m_data == 0) {
jam(); jam();
keyPar.m_ent = ent; keyPar.m_ent = ent;
keyPar.m_data = c_keyBuffer; keyPar.m_data = c_dataBuffer;
tupReadKeys(signal, frag, keyPar); tupReadKeys(signal, frag, keyPar);
} }
} }
...@@ -704,12 +704,12 @@ Dbtux::scanFirst(Signal* signal, ScanOpPtr scanPtr) ...@@ -704,12 +704,12 @@ Dbtux::scanFirst(Signal* signal, ScanOpPtr scanPtr)
bound.first(iter); bound.first(iter);
for (unsigned j = 0; j < bound.getSize(); j++) { for (unsigned j = 0; j < bound.getSize(); j++) {
jam(); jam();
c_keyBuffer[j] = *iter.data; c_dataBuffer[j] = *iter.data;
bound.next(iter); bound.next(iter);
} }
// comparison parameters // comparison parameters
BoundPar boundPar; BoundPar boundPar;
boundPar.m_data1 = c_keyBuffer; boundPar.m_data1 = c_dataBuffer;
boundPar.m_count1 = scan.m_boundCnt[0]; boundPar.m_count1 = scan.m_boundCnt[0];
boundPar.m_dir = 0; boundPar.m_dir = 0;
loop: { loop: {
...@@ -847,12 +847,12 @@ Dbtux::scanNext(Signal* signal, ScanOpPtr scanPtr) ...@@ -847,12 +847,12 @@ Dbtux::scanNext(Signal* signal, ScanOpPtr scanPtr)
bound.first(iter); bound.first(iter);
for (unsigned j = 0; j < bound.getSize(); j++) { for (unsigned j = 0; j < bound.getSize(); j++) {
jam(); jam();
c_keyBuffer[j] = *iter.data; c_dataBuffer[j] = *iter.data;
bound.next(iter); bound.next(iter);
} }
// comparison parameters // comparison parameters
BoundPar boundPar; BoundPar boundPar;
boundPar.m_data1 = c_keyBuffer; boundPar.m_data1 = c_dataBuffer;
boundPar.m_count1 = scan.m_boundCnt[1]; boundPar.m_count1 = scan.m_boundCnt[1];
boundPar.m_dir = 1; boundPar.m_dir = 1;
// use copy of position // use copy of position
......
...@@ -45,36 +45,28 @@ loop: { ...@@ -45,36 +45,28 @@ loop: {
const unsigned occup = node.getOccup(); const unsigned occup = node.getOccup();
ndbrequire(occup != 0); ndbrequire(occup != 0);
// number of equal initial attributes in bounding node // number of equal initial attributes in bounding node
unsigned numEq = ZNIL; unsigned start = ZNIL;
for (unsigned i = 0; i <= 1; i++) { for (unsigned i = 0; i <= 1; i++) {
jam(); jam();
unsigned start1 = 0;
// compare prefix // compare prefix
CmpPar cmpPar; int ret = cmpSearchKey(frag, start1, searchPar.m_data, node.getPref(i), tree.m_prefSize);
cmpPar.m_data1 = searchPar.m_data;
cmpPar.m_data2 = node.getPref(i);
cmpPar.m_len2 = tree.m_prefSize;
cmpPar.m_first = 0;
cmpPar.m_numEq = 0;
int ret = cmpTreeAttrs(frag, cmpPar);
if (ret == NdbSqlUtil::CmpUnknown) { if (ret == NdbSqlUtil::CmpUnknown) {
jam(); jam();
// read full value // read full value
ReadPar readPar; ReadPar readPar;
readPar.m_ent = node.getMinMax(i); readPar.m_ent = node.getMinMax(i);
ndbrequire(cmpPar.m_numEq < numAttrs); ndbrequire(start1 < numAttrs);
readPar.m_first = cmpPar.m_numEq; readPar.m_first = start1;
readPar.m_count = numAttrs - cmpPar.m_numEq; readPar.m_count = numAttrs - start1;
readPar.m_data = 0; // leave in signal data readPar.m_data = 0; // leave in signal data
tupReadAttrs(signal, frag, readPar); tupReadAttrs(signal, frag, readPar);
// compare full value // compare full value
cmpPar.m_data2 = readPar.m_data; ret = cmpSearchKey(frag, start1, searchPar.m_data, readPar.m_data);
cmpPar.m_len2 = ZNIL; // big
cmpPar.m_first = readPar.m_first;
ret = cmpTreeAttrs(frag, cmpPar);
ndbrequire(ret != NdbSqlUtil::CmpUnknown); ndbrequire(ret != NdbSqlUtil::CmpUnknown);
} }
if (numEq > cmpPar.m_numEq) if (start > start1)
numEq = cmpPar.m_numEq; start = start1;
if (ret == 0) { if (ret == 0) {
jam(); jam();
// keys are equal, compare entry values // keys are equal, compare entry values
...@@ -111,21 +103,17 @@ loop: { ...@@ -111,21 +103,17 @@ loop: {
jam(); jam();
int ret = 0; int ret = 0;
// compare remaining attributes // compare remaining attributes
if (numEq < numAttrs) { if (start < numAttrs) {
jam(); jam();
ReadPar readPar; ReadPar readPar;
readPar.m_ent = node.getEnt(j); readPar.m_ent = node.getEnt(j);
readPar.m_first = numEq; readPar.m_first = start;
readPar.m_count = numAttrs - numEq; readPar.m_count = numAttrs - start;
readPar.m_data = 0; // leave in signal data readPar.m_data = 0; // leave in signal data
tupReadAttrs(signal, frag, readPar); tupReadAttrs(signal, frag, readPar);
// compare // compare
CmpPar cmpPar; unsigned start1 = start;
cmpPar.m_data1 = searchPar.m_data; ret = cmpSearchKey(frag, start1, searchPar.m_data, readPar.m_data);
cmpPar.m_data2 = readPar.m_data;
cmpPar.m_len2 = ZNIL; // big
cmpPar.m_first = readPar.m_first;
ret = cmpTreeAttrs(frag, cmpPar);
ndbrequire(ret != NdbSqlUtil::CmpUnknown); ndbrequire(ret != NdbSqlUtil::CmpUnknown);
} }
if (ret == 0) { if (ret == 0) {
......
...@@ -40,5 +40,7 @@ optim 7 mc02/a 42 ms 69 ms 61 pct ...@@ -40,5 +40,7 @@ optim 7 mc02/a 42 ms 69 ms 61 pct
optim 8 mc02/a 42 ms 69 ms 62 pct optim 8 mc02/a 42 ms 69 ms 62 pct
mc02/b 54 ms 104 ms 92 pct mc02/b 54 ms 104 ms 92 pct
optim 9 mc02/a 43 ms 67 ms 54 pct
mc02/b 53 ms 102 ms 91 pct
vim: set et: vim: set et:
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