blocks/dbtux/DbtuxSearch.cpp

/*
   Copyright (c) 2004, 2021, Oracle and/or its affiliates.

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License, version 2.0,
   as published by the Free Software Foundation.

   This program is also distributed with certain software (including
   but not limited to OpenSSL) that is licensed under separate terms,
   as designated in a particular file or component or in included license
   documentation.  The authors of MySQL hereby grant you an additional
   permission to link the program and your derivative works with the
   separately licensed software that they have included with MySQL.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License, version 2.0, for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
*/

#define DBTUX_SEARCH_CPP
#include "Dbtux.hpp"

#define JAM_FILE_ID 368


/*
 * Search down non-empty tree for node to update.  Compare search key to
 * each node minimum.  If greater, move to right subtree.  This can
 * overshoot target node.  The last such node is saved.  The search ends
 * at a final node which is a semi-leaf or leaf.  If search key is less
 * than final node minimum then the saved node (if any) is the g.l.b of
 * the final node and we move back to it.
 *
 * Search within the found node is done by caller.  On add, search key
 * may be before minimum or after maximum entry.  On remove, search key
 * is within the node.
 */
void
Dbtux::findNodeToUpdate(TuxCtx& ctx, Frag& frag, const KeyDataC& searchKey, TreeEnt searchEnt, NodeHandle& currNode)
{
  const Index& index = *c_indexPool.getPtr(frag.m_indexId);
  const Uint32 numAttrs = index.m_numAttrs;
  const Uint32 prefAttrs = index.m_prefAttrs;
  const Uint32 prefBytes = index.m_prefBytes;
  KeyData entryKey(index.m_keySpec, false, 0);
  entryKey.set_buf(ctx.c_entryKey, MaxAttrDataSize << 2);
  KeyDataC prefKey(index.m_keySpec, false);
  NodeHandle glbNode(frag);     // potential g.l.b of final node
  while (true) {
    thrjam(ctx.jamBuffer);
    selectNode(currNode, currNode.m_loc);
    prefKey.set_buf(currNode.getPref(), prefBytes, prefAttrs);
    int ret = 0;
    if (prefAttrs > 0) {
      thrjam(ctx.jamBuffer);
      ret = cmpSearchKey(ctx, searchKey, prefKey, prefAttrs);
    }
    if (ret == 0 && prefAttrs < numAttrs) {
      thrjam(ctx.jamBuffer);
      // read and compare all attributes
      readKeyAttrs(ctx, frag, currNode.getEnt(0), entryKey, numAttrs);
      ret = cmpSearchKey(ctx, searchKey, entryKey, numAttrs);
    }
    if (ret == 0) {
      thrjam(ctx.jamBuffer);
      // keys are equal, compare entry values
      ret = searchEnt.cmp(currNode.getEnt(0));
    }
    if (ret < 0) {
      thrjam(ctx.jamBuffer);
      const TupLoc loc = currNode.getLink(0);
      if (loc != NullTupLoc) {
        thrjam(ctx.jamBuffer);
        // continue to left subtree
        currNode.m_loc = loc;
        continue;
      }
      if (! glbNode.isNull()) {
        thrjam(ctx.jamBuffer);
        // move up to the g.l.b
        currNode = glbNode;
      }
      break;
    }
    if (ret > 0) {
      thrjam(ctx.jamBuffer);
      const TupLoc loc = currNode.getLink(1);
      if (loc != NullTupLoc) {
        thrjam(ctx.jamBuffer);
        // save potential g.l.b
        glbNode = currNode;
        // continue to right subtree
        currNode.m_loc = loc;
        continue;
      }
      break;
    }
    // ret == 0
    thrjam(ctx.jamBuffer);
    break;
  }
}

/*
 * Find position within the final node to add entry to.  Use binary
 * search.  Return true if ok i.e. entry to add is not a duplicate.
 */
bool
Dbtux::findPosToAdd(TuxCtx& ctx, Frag& frag, const KeyDataC& searchKey, TreeEnt searchEnt, NodeHandle& currNode, TreePos& treePos)
{
  const Index& index = *c_indexPool.getPtr(frag.m_indexId);
  int lo = -1;
  int hi = (int)currNode.getOccup();
  KeyData entryKey(index.m_keySpec, false, 0);
  entryKey.set_buf(ctx.c_entryKey, MaxAttrDataSize << 2);
  while (hi - lo > 1) {
    thrjam(ctx.jamBuffer);
    // hi - lo > 1 implies lo < j < hi
    int j = (hi + lo) / 2;
    // read and compare all attributes
    readKeyAttrs(ctx, frag, currNode.getEnt(j), entryKey, index.m_numAttrs);
    int ret = cmpSearchKey(ctx, searchKey, entryKey, index.m_numAttrs);
    if (ret == 0) {
      thrjam(ctx.jamBuffer);
      // keys are equal, compare entry values
      ret = searchEnt.cmp(currNode.getEnt(j));
    }
    if (ret < 0) {
      thrjam(ctx.jamBuffer);
      hi = j;
    } else if (ret > 0) {
      thrjam(ctx.jamBuffer);
      lo = j;
    } else {
      treePos.m_pos = j;
      // entry found - error
      return false;
    }
  }
  ndbrequire(hi - lo == 1);
  // return hi pos, see treeAdd() for next step
  treePos.m_pos = hi;
  return true;
}

/*
 * Find position within the final node to remove entry from.  Use linear
 * search.  Return true if ok i.e. the entry was found.
 */
bool
Dbtux::findPosToRemove(TuxCtx& ctx, Frag& frag, const KeyDataC& searchKey, TreeEnt searchEnt, NodeHandle& currNode, TreePos& treePos)
{
  const unsigned occup = currNode.getOccup();
  for (unsigned j = 0; j < occup; j++) {
    thrjam(ctx.jamBuffer);
    // compare only the entry
    if (searchEnt.eq(currNode.getEnt(j))) {
      thrjam(ctx.jamBuffer);
      treePos.m_pos = j;
      return true;
    }
  }
  treePos.m_pos = occup;
  // not found - failed
  return false;
}

/*
 * Search for entry to add.
 */
bool
Dbtux::searchToAdd(TuxCtx& ctx, Frag& frag, const KeyDataC& searchKey, TreeEnt searchEnt, TreePos& treePos)
{
  const TreeHead& tree = frag.m_tree;
  NodeHandle currNode(frag);
  currNode.m_loc = tree.m_root;
  if (unlikely(currNode.m_loc == NullTupLoc)) {
    // empty tree
    thrjam(ctx.jamBuffer);
    return true;
  }
  findNodeToUpdate(ctx, frag, searchKey, searchEnt, currNode);
  treePos.m_loc = currNode.m_loc;
  if (! findPosToAdd(ctx, frag, searchKey, searchEnt, currNode, treePos)) {
    thrjam(ctx.jamBuffer);
    return false;
  }
  return true;
}

/*
 * Search for entry to remove.
 */
bool
Dbtux::searchToRemove(TuxCtx& ctx, Frag& frag, const KeyDataC& searchKey, TreeEnt searchEnt, TreePos& treePos)
{
  const TreeHead& tree = frag.m_tree;
  NodeHandle currNode(frag);
  currNode.m_loc = tree.m_root;
  if (unlikely(currNode.m_loc == NullTupLoc)) {
    // empty tree - failed
    thrjam(ctx.jamBuffer);
    return false;
  }
  findNodeToUpdate(ctx, frag, searchKey, searchEnt, currNode);
  treePos.m_loc = currNode.m_loc;
  if (! findPosToRemove(ctx, frag, searchKey, searchEnt, currNode, treePos)) {
    thrjam(ctx.jamBuffer);
    return false;
  }
  return true;
}

/*
 * Search down non-empty tree for node to start scan from.  Similar to
 * findNodeToUpdate().  Direction is 0-ascending or 1-descending.
 * Search within the found node is done by caller.
 */
void
Dbtux::findNodeToScan(Frag& frag, unsigned idir, const KeyBoundC& searchBound, NodeHandle& currNode)
{
  const int jdir = 1 - 2 * int(idir);
  const Index& index = *c_indexPool.getPtr(frag.m_indexId);
  const Uint32 numAttrs = searchBound.get_data().get_cnt();
  const Uint32 prefAttrs = min(index.m_prefAttrs, numAttrs);
  const Uint32 prefBytes = index.m_prefBytes;
  KeyData entryKey(index.m_keySpec, false, 0);
  entryKey.set_buf(c_ctx.c_entryKey, MaxAttrDataSize << 2);
  KeyDataC prefKey(index.m_keySpec, false);
  NodeHandle glbNode(frag);     // potential g.l.b of final node
  while (true) {
    jam();
    selectNode(currNode, currNode.m_loc);
    prefKey.set_buf(currNode.getPref(), prefBytes, prefAttrs);
    int ret = 0;
    if (numAttrs > 0) {
      if (prefAttrs > 0) {
        jam();
        // compare node prefix - result 0 implies bound is longer
        ret = cmpSearchBound(c_ctx, searchBound, prefKey, prefAttrs);
      }
      if (ret == 0) {
        jam();
        // read and compare all attributes
        readKeyAttrs(c_ctx, frag, currNode.getEnt(0), entryKey, numAttrs);
        ret = cmpSearchBound(c_ctx, searchBound, entryKey, numAttrs);
        ndbrequire(ret != 0);
      }
    } else {
      jam();
      ret = (-1) * jdir;
    }
    if (ret < 0) {
      // bound is left of this node
      jam();
      const TupLoc loc = currNode.getLink(0);
      if (loc != NullTupLoc) {
        jam();
        // continue to left subtree
        currNode.m_loc = loc;
        continue;
      }
      if (! glbNode.isNull()) {
        jam();
        // move up to the g.l.b
        currNode = glbNode;
      }
      break;
    }
    if (ret > 0) {
      // bound is at or right of this node
      jam();
      const TupLoc loc = currNode.getLink(1);
      if (loc != NullTupLoc) {
        jam();
        // save potential g.l.b
        glbNode = currNode;
        // continue to right subtree
        currNode.m_loc = loc;
        continue;
      }
      break;
    }
    // ret == 0 never
    ndbrequire(false);
  }
}

/*
 * Search across final node for position to start scan from.  Use binary
 * search similar to findPosToAdd().
 */
void
Dbtux::findPosToScan(Frag& frag, unsigned idir, const KeyBoundC& searchBound, NodeHandle& currNode, Uint16* pos)
{
  const int jdir = 1 - 2 * int(idir);
  const Index& index = *c_indexPool.getPtr(frag.m_indexId);
  const Uint32 numAttrs = searchBound.get_data().get_cnt();
  int lo = -1;
  int hi = (int)currNode.getOccup();
  KeyData entryKey(index.m_keySpec, false, 0);
  entryKey.set_buf(c_ctx.c_entryKey, MaxAttrDataSize << 2);
  while (hi - lo > 1) {
    jam();
    // hi - lo > 1 implies lo < j < hi
    int j = (hi + lo) / 2;
    int ret = (-1) * jdir;
    if (numAttrs != 0) {
      // read and compare all attributes
      readKeyAttrs(c_ctx, frag, currNode.getEnt(j), entryKey, numAttrs);
      ret = cmpSearchBound(c_ctx, searchBound, entryKey, numAttrs);
      ndbrequire(ret != 0);
    }
    if (ret < 0) {
      jam();
      hi = j;
    } else if (ret > 0) {
      jam();
      lo = j;
    } else {
      // ret == 0 never
      ndbrequire(false);
    }
  }
  // return hi pos, caller handles ascending vs descending
  *pos = hi;
}

/*
 * Search for scan start position.
 */
void
Dbtux::searchToScan(Frag& frag, unsigned idir, const KeyBoundC& searchBound, TreePos& treePos)
{
  const TreeHead& tree = frag.m_tree;
  NodeHandle currNode(frag);
  currNode.m_loc = tree.m_root;
  if (unlikely(currNode.m_loc == NullTupLoc)) {
    // empty tree
    jam();
    return;
  }
  findNodeToScan(frag, idir, searchBound, currNode);
  treePos.m_loc = currNode.m_loc;
  Uint16 pos;
  findPosToScan(frag, idir, searchBound, currNode, &pos);
  const unsigned occup = currNode.getOccup();
  if (idir == 0) {
    if (pos < occup) {
      jam();
      treePos.m_pos = pos;
      treePos.m_dir = 3;
    } else {
      // start scan after node end i.e. proceed to right child
      treePos.m_pos = ZNIL;
      treePos.m_dir = 5;
    }
  } else {
    if (pos > 0) {
      jam();
      // start scan from previous entry
      treePos.m_pos = pos - 1;
      treePos.m_dir = 3;
    } else {
      treePos.m_pos = ZNIL;
      treePos.m_dir = 0;
    }
  }
}