diff --git a/ndb/src/kernel/blocks/dbtux/Dbtux.hpp b/ndb/src/kernel/blocks/dbtux/Dbtux.hpp index 911b6f84838..feb0ae30e4c 100644 --- a/ndb/src/kernel/blocks/dbtux/Dbtux.hpp +++ b/ndb/src/kernel/blocks/dbtux/Dbtux.hpp @@ -564,17 +564,6 @@ private: ReadPar(); }; - /* - * Tree search for entry. - */ - struct SearchPar; - friend struct SearchPar; - struct SearchPar { - TableData m_data; // input index key values - TreeEnt m_ent; // input tuple and version - SearchPar(); - }; - /* * Scan bound comparison. */ @@ -641,7 +630,7 @@ private: /* * DbtuxTree.cpp */ - void treeSearch(Signal* signal, Frag& frag, SearchPar searchPar, TreePos& treePos); + void treeSearch(Signal* signal, Frag& frag, TableData searchKey, TreeEnt searchEnt, TreePos& treePos); void treeAdd(Signal* signal, Frag& frag, TreePos treePos, TreeEnt ent); void treeRemove(Signal* signal, Frag& frag, TreePos treePos); void treeRotateSingle(Signal* signal, Frag& frag, NodeHandle& node, unsigned i); @@ -669,6 +658,7 @@ private: * DbtuxCmp.cpp */ int cmpSearchKey(const Frag& frag, unsigned& start, TableData data1, ConstData data2, unsigned size2 = MaxAttrDataSize); + int cmpSearchKey(const Frag& frag, unsigned& start, TableData data1, TableData data2); int cmpScanBound(const Frag& frag, const BoundPar boundPar); /* @@ -1218,13 +1208,6 @@ Dbtux::ReadPar::ReadPar() : { } -inline -Dbtux::SearchPar::SearchPar() : - m_data(0), - m_ent() -{ -} - inline Dbtux::BoundPar::BoundPar() : m_data1(0), @@ -1295,6 +1278,7 @@ Dbtux::readKeyAttrs(const Frag& frag, TreeEnt ent, unsigned start, ConstData key const Uint32 tupVersion = ent.m_tupVersion; ndbrequire(start < frag.m_numAttrs); const unsigned numAttrs = frag.m_numAttrs - start; + // start applies to both keys and output data keyAttrs += start; keyData += start; c_tup->tuxReadAttrs(tableFragPtrI, tupLoc.m_pageId, tupLoc.m_pageOffset, tupVersion, numAttrs, keyAttrs, keyData); diff --git a/ndb/src/kernel/blocks/dbtux/DbtuxCmp.cpp b/ndb/src/kernel/blocks/dbtux/DbtuxCmp.cpp index 71e8a3fbed3..cd0d4722207 100644 --- a/ndb/src/kernel/blocks/dbtux/DbtuxCmp.cpp +++ b/ndb/src/kernel/blocks/dbtux/DbtuxCmp.cpp @@ -18,11 +18,11 @@ #include "Dbtux.hpp" /* - * Search key vs tree entry. + * Search key vs node prefix. * - * Compare search key and index attribute data. The attribute data may - * be partial in which case CmpUnknown may be returned. Also counts how - * many (additional) initial attributes were equal. + * The comparison starts at given attribute position (in fact 0). The + * position is updated by number of equal initial attributes found. The + * prefix may be partial in which case CmpUnknown may be returned. */ int Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData data1, ConstData data2, unsigned size2) @@ -83,6 +83,61 @@ Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData data1, ConstDat return ret; } +/* + * Search key vs tree entry. + * + * Start position is updated as in previous routine. + */ +int +Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData data1, TableData data2) +{ + const unsigned numAttrs = frag.m_numAttrs; + const DescEnt& descEnt = getDescEnt(frag.m_descPage, frag.m_descOff); + // skip to right position + data1 += start; + data2 += start; + int ret = 0; + while (start < numAttrs) { + if (*data1 != 0) { + if (*data2 != 0) { + jam(); + // current attribute + const DescAttr& descAttr = descEnt.m_descAttr[start]; + const unsigned typeId = descAttr.m_typeId; + // full data size + const unsigned size1 = AttributeDescriptor::getSizeInWords(descAttr.m_attrDesc); + // compare + const Uint32* const p1 = *data1; + const Uint32* const p2 = *data2; + ret = NdbSqlUtil::cmp(typeId, p1, p2, size1, size1); + if (ret != 0) { + jam(); + break; + } + } else { + jam(); + // not NULL < NULL + ret = -1; + break; + } + } else { + if (*data2 != 0) { + jam(); + // NULL > not NULL + ret = +1; + break; + } + } + data1 += 1; + data2 += 1; + start++; + } + // XXX until data format errors are handled + ndbrequire(ret != NdbSqlUtil::CmpError); + return ret; +} + + /* * Scan bound vs tree entry. * diff --git a/ndb/src/kernel/blocks/dbtux/DbtuxMaint.cpp b/ndb/src/kernel/blocks/dbtux/DbtuxMaint.cpp index 337bd63314f..5001016f89c 100644 --- a/ndb/src/kernel/blocks/dbtux/DbtuxMaint.cpp +++ b/ndb/src/kernel/blocks/dbtux/DbtuxMaint.cpp @@ -100,11 +100,6 @@ Dbtux::execTUX_MAINT_REQ(Signal* signal) return; } } - // find position in tree - SearchPar searchPar; - searchPar.m_data = c_searchKey; - searchPar.m_ent = ent; - TreePos treePos; #ifdef VM_TRACE if (debugFlags & DebugMaint) { debugOut << "opCode=" << dec << opCode; @@ -116,7 +111,9 @@ Dbtux::execTUX_MAINT_REQ(Signal* signal) debugOut << endl; } #endif - treeSearch(signal, frag, searchPar, treePos); + // find position in tree + TreePos treePos; + treeSearch(signal, frag, c_searchKey, ent, treePos); #ifdef VM_TRACE if (debugFlags & DebugMaint) { debugOut << treePos << endl; diff --git a/ndb/src/kernel/blocks/dbtux/DbtuxTree.cpp b/ndb/src/kernel/blocks/dbtux/DbtuxTree.cpp index 57283bb0c9f..6e7fbad5940 100644 --- a/ndb/src/kernel/blocks/dbtux/DbtuxTree.cpp +++ b/ndb/src/kernel/blocks/dbtux/DbtuxTree.cpp @@ -26,7 +26,7 @@ * same in min/max need not be checked. */ void -Dbtux::treeSearch(Signal* signal, Frag& frag, SearchPar searchPar, TreePos& treePos) +Dbtux::treeSearch(Signal* signal, Frag& frag, TableData searchKey, TreeEnt searchEnt, TreePos& treePos) { const TreeHead& tree = frag.m_tree; const unsigned numAttrs = frag.m_numAttrs; @@ -50,19 +50,12 @@ loop: { jam(); unsigned start1 = 0; // compare prefix - int ret = cmpSearchKey(frag, start1, searchPar.m_data, node.getPref(i), tree.m_prefSize); + int ret = cmpSearchKey(frag, start1, searchKey, node.getPref(i), tree.m_prefSize); if (ret == NdbSqlUtil::CmpUnknown) { jam(); - // read full value - ReadPar readPar; - readPar.m_ent = node.getMinMax(i); - ndbrequire(start1 < numAttrs); - readPar.m_first = start1; - readPar.m_count = numAttrs - start1; - readPar.m_data = 0; // leave in signal data - tupReadAttrs(signal, frag, readPar); - // compare full value - ret = cmpSearchKey(frag, start1, searchPar.m_data, readPar.m_data); + // read and compare remaining attributes + readKeyAttrs(frag, node.getMinMax(i), start1, c_keyAttrs, c_entryKey); + ret = cmpSearchKey(frag, start1, searchKey, c_entryKey); ndbrequire(ret != NdbSqlUtil::CmpUnknown); } if (start > start1) @@ -70,7 +63,7 @@ loop: { if (ret == 0) { jam(); // keys are equal, compare entry values - ret = searchPar.m_ent.cmp(node.getMinMax(i)); + ret = searchEnt.cmp(node.getMinMax(i)); } if (i == 0 ? (ret < 0) : (ret > 0)) { jam(); @@ -102,24 +95,18 @@ loop: { for (unsigned j = 1; j <= numWithin; j++) { jam(); int ret = 0; - // compare remaining attributes if (start < numAttrs) { jam(); - ReadPar readPar; - readPar.m_ent = node.getEnt(j); - readPar.m_first = start; - readPar.m_count = numAttrs - start; - readPar.m_data = 0; // leave in signal data - tupReadAttrs(signal, frag, readPar); - // compare + // read and compare remaining attributes unsigned start1 = start; - ret = cmpSearchKey(frag, start1, searchPar.m_data, readPar.m_data); + readKeyAttrs(frag, node.getEnt(j), start1, c_keyAttrs, c_entryKey); + ret = cmpSearchKey(frag, start1, searchKey, c_entryKey); ndbrequire(ret != NdbSqlUtil::CmpUnknown); } if (ret == 0) { jam(); // keys are equal, compare entry values - ret = searchPar.m_ent.cmp(node.getEnt(j)); + ret = searchEnt.cmp(node.getEnt(j)); } if (ret <= 0) { jam(); diff --git a/ndb/src/kernel/blocks/dbtux/Times.txt b/ndb/src/kernel/blocks/dbtux/Times.txt index ad4073f5d10..735d3568717 100644 --- a/ndb/src/kernel/blocks/dbtux/Times.txt +++ b/ndb/src/kernel/blocks/dbtux/Times.txt @@ -43,4 +43,7 @@ optim 8 mc02/a 42 ms 69 ms 62 pct optim 9 mc02/a 43 ms 67 ms 54 pct mc02/b 53 ms 102 ms 91 pct +optim 10 mc02/a 44 ms 65 ms 46 pct + mc02/b 53 ms 88 ms 66 pct + vim: set et: