Fix severe bugs in BTree::findNext

Author: komozoi

include/fs/BTree.h

@@ -190,29 +190,49 @@ class BTreeBase {
 	 * @return true if a match was found, false if not
 	 */
 	bool findNext(T& val) {
-		btree_node_t<T, N> node;
-		uint64_t offset;
-
-		int i = findNearest(val, node, offset);
-		if (i > 0)
-			i--;
+		btree_node_t<T, N> node = getRootNode();
+		uint64_t offset = getRootOffset();
+		bool found = false;
+		T best;
 
 		while (true) {
-			if (i < node.header.nElements && compare(node.elements[i], val) >= 0) {
-				// Found a valid next value (either equal or next greater)
-				val = node.elements[i];
-				return true;
+			int low = 0;
+			int high = node.header.nElements - 1;
+			int midIdx = -1;
+
+			while (low <= high) {
+				int mid = (low + high) / 2;
+				int c = compare(node.elements[mid], val);
+				if (c < 0) {
+					low = mid + 1;
+				} else if (c == 0) {
+					val = node.elements[mid];
+					return true;
+				} else {
+					midIdx = mid;
+					high = mid - 1;
+				}
+			}
+
+			if (midIdx != -1) {
+				if (!found || compare(node.elements[midIdx], best) < 0) {
+					best = node.elements[midIdx];
+					found = true;
+				}
 			}
 
-			// If this node has no children, there is no next element
 			if (node.header.nChildren == 0)
-				return false;
+				break;
 
-			// Go to the next child
-			offset = node.header.childOffsets[i];
+			offset = node.header.childOffsets[low];
 			node = getNode(offset);
-			i = 0;
 		}
+
+		if (found) {
+			val = best;
+			return true;
+		}
+		return false;
 	}
 
 	/**
@@ -279,11 +299,11 @@ class BTreeBase {
 				int c = compare(node.elements[mid], val);
 
 				if (c < 0)
-					high = mid - 1;
+					low = mid + 1;
 				else if (c == 0) {
 					return mid;
 				} else
-					low = mid + 1;
+					high = mid - 1;
 			}
 
 			if (node.header.nChildren == 0)
@@ -323,18 +343,18 @@ class BTreeBase {
 			overwriteNode(offset, node);
 
 		} else {
-			btree_node_t<T, N> child = getNode(node.childOffsets[i]);
+			btree_node_t<T, N> child = getNode(node.header.childOffsets[i]);
 
 			if (child.header.nElements == N) {
 				uint64_t newChildOffset;
-				btree_node_t<T, N> newChild = splitChild(node, child, node.childOffsets[i], newChildOffset);
+				btree_node_t<T, N> newChild = splitChild(node, child, node.header.childOffsets[i], newChildOffset);
 				if (compare(node.elements[i], element) <= 0) {
 					child = newChild;
 					i++;
 				}
 			}
 
-			insertNonFull(child, node.childOffsets[i], element);
+			insertNonFull(child, node.header.childOffsets[i], element);
 		}
 
 	}
@@ -452,7 +472,7 @@ class BTreeBase {
 	int scanNode(const btree_node_t<T, N>& node, const T& element) {
 		int i = 0;
 		for (; i < node.header.nElements; i++) {
-			if (compare(node.elements[i], element) <= 0)
+			if (compare(node.elements[i], element) >= 0)
 				break;
 		}
 

src/fs/FdHandle.cpp

@@ -551,11 +551,6 @@ std::lock_guard<std::mutex> FdHandle::getLock() const {
 	return std::lock_guard<std::mutex>(getHandleData(fd).mutex);
 }
 
-FdHandle::~FdHandle() {
-	if (fd >= 0)
-		getHandleData(fd).decRef();
-}
-
 void FdHandle::markToClose() const {
 	getHandleData(fd).markToClose();
 }
@@ -609,6 +604,11 @@ int FdHandle::numReferences() const {
 	return getHandleData(fd).numReferences();
 }
 
+FdHandle::~FdHandle() {
+	if (fd >= 0)
+		getHandleData(fd).decRef();
+}
+
 FdTransaction::FdTransaction(const FdHandle& handle) : guard(getHandleData(handle.getFd()).mutex), handleData(getHandleData(handle.getFd())) {
 }
 

src/tests/test_BTree.cpp

@@ -320,3 +320,66 @@ TEST_F(BTreeBaseTest, FindNext) {
 	EXPECT_FALSE(tree.findNext(val));
 	EXPECT_EQ(10000, val);
 }
+
+TEST(BTreeFindNext, FindsClosestHigherInChildNode) {
+	RAMBTree<int, 3> tree(compareInts);
+
+	// Insert enough elements to force multiple splits and a multi-level tree.
+	// Using even numbers so we can probe with odd "gap" queries.
+	const int numElements = 200;
+	for (int i = 0; i < numElements; ++i)
+		tree.insert(i * 2);
+
+	// Sanity check: every inserted value must be retrievable via find().
+	for (int i = 0; i < numElements; ++i) {
+		int val = i * 2;
+		ASSERT_TRUE(tree.find(val)) << "find missing for " << (i * 2);
+		ASSERT_EQ(i * 2, val);
+	}
+
+	// For every odd value in range, findNext must return the next even value.
+	// Under the previous buggy findNext, the descent skipped the correct child
+	// and returned an unrelated greater value (or returned false), making this
+	// test fail.
+	for (int q = 1; q < numElements * 2 - 1; q += 2) {
+		int val = q;
+		EXPECT_TRUE(tree.findNext(val)) << "query=" << q;
+		EXPECT_EQ(q + 1, val) << "query=" << q;
+	}
+
+	// Exact-match probes must still work.
+	for (int q = 0; q < numElements * 2; q += 2) {
+		int val = q;
+		EXPECT_TRUE(tree.findNext(val)) << "query=" << q;
+		EXPECT_EQ(q, val) << "query=" << q;
+	}
+
+	// A value strictly larger than every element should not be found.
+	int val = numElements * 2 + 10;
+	EXPECT_FALSE(tree.findNext(val));
+	EXPECT_EQ(numElements * 2 + 10, val);
+}
+
+
+// Regression test: findNext with a sparse outlier in a deep tree.  Searching
+// in the gap between the densely packed range and the outlier must return
+// the outlier (which lives in a different subtree than the descent path of
+// the buggy implementation).
+TEST(BTreeFindNext, FindsOutlierAcrossSubtrees) {
+	RAMBTree<int, 3> tree(compareInts);
+
+	for (int i = 0; i < 100; ++i)
+		tree.insert(i * 2);
+	tree.insert(100000);
+
+	int val = 500;
+	EXPECT_TRUE(tree.findNext(val));
+	EXPECT_EQ(100000, val);
+
+	val = 99999;
+	EXPECT_TRUE(tree.findNext(val));
+	EXPECT_EQ(100000, val);
+
+	val = 100001;
+	EXPECT_FALSE(tree.findNext(val));
+}