tux optim 10 - use TUP method to read entry keys

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);

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.
  *

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;

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();

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: