Change isValid() method on BTreeRoot to also check aggregation.

6 files changed, 69 insertions, 30 deletions

diff --git a/searchlib/src/vespa/searchlib/btree/btreeaggregator.h b/searchlib/src/vespa/searchlib/btree/btreeaggregator.h
index 9da0096e093..95c03b11ef8 100644
--- a/searchlib/src/vespa/searchlib/btree/btreeaggregator.h
+++ b/searchlib/src/vespa/searchlib/btree/btreeaggregator.h
@@ -31,6 +31,9 @@ public:
     LeafNodeType;
     typedef AggrT AggregatedType;
 
+    static AggrT aggregate(const LeafNodeType &node, AggrCalcT aggrCalc);
+    static AggrT aggregate(const InternalNodeType &node, const NodeAllocatorType &allocator, AggrCalcT aggrCalc);
+
     static void
     recalc(LeafNodeType &node, const AggrCalcT &aggrCalc);
 
diff --git a/searchlib/src/vespa/searchlib/btree/btreeaggregator.hpp b/searchlib/src/vespa/searchlib/btree/btreeaggregator.hpp
index ef8589dd49a..5a424fb55ff 100644
--- a/searchlib/src/vespa/searchlib/btree/btreeaggregator.hpp
+++ b/searchlib/src/vespa/searchlib/btree/btreeaggregator.hpp
@@ -12,25 +12,20 @@ namespace btree
 
 template <typename KeyT, typename DataT, typename AggrT,
           size_t INTERNAL_SLOTS, size_t LEAF_SLOTS, class AggrCalcT>
-void
-BTreeAggregator<KeyT, DataT, AggrT, INTERNAL_SLOTS, LEAF_SLOTS, AggrCalcT>::
-recalc(LeafNodeType &node, const AggrCalcT &aggrCalc)
+AggrT
+BTreeAggregator<KeyT, DataT, AggrT, INTERNAL_SLOTS, LEAF_SLOTS, AggrCalcT>::aggregate(const LeafNodeType &node, AggrCalcT aggrCalc)
 {
     AggrT a;
     for (uint32_t i = 0, ie = node.validSlots(); i < ie; ++i) {
         aggrCalc.add(a, aggrCalc.getVal(node.getData(i)));
     }
-    node.getAggregated() = a;
+    return a;
 }
 
-
 template <typename KeyT, typename DataT, typename AggrT,
           size_t INTERNAL_SLOTS, size_t LEAF_SLOTS, class AggrCalcT>
-void
-BTreeAggregator<KeyT, DataT, AggrT, INTERNAL_SLOTS, LEAF_SLOTS, AggrCalcT>::
-recalc(InternalNodeType &node,
-       const NodeAllocatorType &allocator,
-       const AggrCalcT &aggrCalc)
+AggrT
+BTreeAggregator<KeyT, DataT, AggrT, INTERNAL_SLOTS, LEAF_SLOTS, AggrCalcT>::aggregate(const InternalNodeType &node, const NodeAllocatorType &allocator, AggrCalcT aggrCalc)
 {
     AggrT a;
     for (uint32_t i = 0, ie = node.validSlots(); i < ie; ++i) {
@@ -38,7 +33,27 @@ recalc(InternalNodeType &node,
         const AggrT &ca(allocator.getAggregated(childRef));
         aggrCalc.add(a, ca);
     }
-    node.getAggregated() = a;
+    return a;
+}
+
+template <typename KeyT, typename DataT, typename AggrT,
+          size_t INTERNAL_SLOTS, size_t LEAF_SLOTS, class AggrCalcT>
+void
+BTreeAggregator<KeyT, DataT, AggrT, INTERNAL_SLOTS, LEAF_SLOTS, AggrCalcT>::
+recalc(LeafNodeType &node, const AggrCalcT &aggrCalc)
+{
+    node.getAggregated() = aggregate(node, aggrCalc);
+}
+
+template <typename KeyT, typename DataT, typename AggrT,
+          size_t INTERNAL_SLOTS, size_t LEAF_SLOTS, class AggrCalcT>
+void
+BTreeAggregator<KeyT, DataT, AggrT, INTERNAL_SLOTS, LEAF_SLOTS, AggrCalcT>::
+recalc(InternalNodeType &node,
+       const NodeAllocatorType &allocator,
+       const AggrCalcT &aggrCalc)
+{
+    node.getAggregated() = aggregate(node, allocator, aggrCalc);
 }
 
 
diff --git a/searchlib/src/vespa/searchlib/btree/btreeroot.h b/searchlib/src/vespa/searchlib/btree/btreeroot.h
index d7b58341ce6..86dd47ca5fa 100644
--- a/searchlib/src/vespa/searchlib/btree/btreeroot.h
+++ b/searchlib/src/vespa/searchlib/btree/btreeroot.h
@@ -54,9 +54,6 @@ protected:
     using ParentType::isFrozen;
 
     vespalib::string toString(BTreeNode::Ref node, const NodeAllocatorType &allocator) const;
-    bool isValid(BTreeNode::Ref node, bool ignoreMinSlots, uint32_t level,
-                 const NodeAllocatorType &allocator, CompareT comp) const;
-
 public:
     /**
      * Read view of the frozen version of the tree.
@@ -157,12 +154,6 @@ public:
 
     vespalib::string toString(const NodeAllocatorType &allocator) const;
 
-    bool
-    isValid(const NodeAllocatorType &allocator, CompareT comp = CompareT()) const;
-
-    bool
-    isValidFrozen(const NodeAllocatorType &allocator, CompareT comp = CompareT()) const;
-
     size_t
     bitSize(const NodeAllocatorType &allocator) const;
 
@@ -208,6 +199,10 @@ public:
     using Parent2Type::getFrozenRootRelaxed;
     using Parent2Type::isFrozen;
 
+protected:
+    bool isValid(BTreeNode::Ref node, bool ignoreMinSlots, uint32_t level,
+                 const NodeAllocatorType &allocator, CompareT comp, AggrCalcT aggrCalc) const;
+
 public:
     /**
      * Create a tree from a tree builder.  This is a destructive
@@ -235,6 +230,10 @@ public:
     void
     remove(Iterator &itr,
            const AggrCalcT &aggrCalc = AggrCalcT());
+
+    bool isValid(const NodeAllocatorType &allocator, CompareT comp = CompareT()) const;
+
+    bool isValidFrozen(const NodeAllocatorType &allocator, CompareT comp = CompareT()) const;
 };
 
 
diff --git a/searchlib/src/vespa/searchlib/btree/btreeroot.hpp b/searchlib/src/vespa/searchlib/btree/btreeroot.hpp
index a42d9e82122..365a52aa8de 100644
--- a/searchlib/src/vespa/searchlib/btree/btreeroot.hpp
+++ b/searchlib/src/vespa/searchlib/btree/btreeroot.hpp
@@ -40,12 +40,12 @@ toString(BTreeNode::Ref node,
 }
 
 template <typename KeyT, typename DataT, typename AggrT, typename CompareT,
-          typename TraitsT>
+          typename TraitsT, class AggrCalcT>
 bool
-BTreeRootT<KeyT, DataT, AggrT, CompareT, TraitsT>::
+BTreeRoot<KeyT, DataT, AggrT, CompareT, TraitsT, AggrCalcT>::
 isValid(BTreeNode::Ref node,
         bool ignoreMinSlots, uint32_t level, const NodeAllocatorType &allocator,
-        CompareT comp) const
+        CompareT comp, AggrCalcT aggrCalc) const
 {
     if (allocator.isLeafRef(node)) {
         if (level != 0) {
@@ -64,6 +64,12 @@ isValid(BTreeNode::Ref node,
                 return false;
             }
         }
+        if (AggrCalcT::hasAggregated()) {
+            AggrT aggregated = Aggregator::aggregate(*lnode, aggrCalc);
+            if (aggregated != lnode->getAggregated()) {
+                return false;
+            }
+        }
     } else {
         if (level == 0) {
             return false;
@@ -98,7 +104,7 @@ isValid(BTreeNode::Ref node,
             if (comp(allocator.getLastKey(childRef), inode->getKey(i))) {
                 return false;
             }
-            if (!isValid(childRef, false, level - 1, allocator, comp)) {
+            if (!isValid(childRef, false, level - 1, allocator, comp, aggrCalc)) {
                 return false;
             }
         }
@@ -108,6 +114,12 @@ isValid(BTreeNode::Ref node,
         if (lChildren < inode->validSlots() && iChildren < inode->validSlots()) {
             return false;
         }
+        if (AggrCalcT::hasAggregated()) {
+            AggrT aggregated = Aggregator::aggregate(*inode, allocator, aggrCalc);
+            if (aggregated != inode->getAggregated()) {
+                return false;
+            }
+        }
     }
     return true;
 }
@@ -320,31 +332,31 @@ toString(const NodeAllocatorType &allocator) const
 }
 
 template <typename KeyT, typename DataT, typename AggrT, typename CompareT,
-          typename TraitsT>
+          typename TraitsT, class AggrCalcT>
 bool
-BTreeRootT<KeyT, DataT, AggrT, CompareT, TraitsT>::
+BTreeRoot<KeyT, DataT, AggrT, CompareT, TraitsT, AggrCalcT>::
 isValid(const NodeAllocatorType &allocator,
         CompareT comp) const
 {
     if (NodeAllocatorType::isValidRef(_root)) {
         uint32_t level  = allocator.getLevel(_root);
-        return isValid(_root, true, level, allocator, comp);
+        return isValid(_root, true, level, allocator, comp, AggrCalcT());
     }
     return true;
 }
 
 
 template <typename KeyT, typename DataT, typename AggrT, typename CompareT,
-          typename TraitsT>
+          typename TraitsT, class AggrCalcT>
 bool
-BTreeRootT<KeyT, DataT, AggrT, CompareT, TraitsT>::
+BTreeRoot<KeyT, DataT, AggrT, CompareT, TraitsT, AggrCalcT>::
 isValidFrozen(const NodeAllocatorType &allocator,
               CompareT comp) const
 {
     BTreeNode::Ref frozenRoot = getFrozenRoot();
     if (NodeAllocatorType::isValidRef(frozenRoot)) {
         uint32_t level  = allocator.getLevel(frozenRoot);
-        return isValid(frozenRoot, true, level, allocator, comp);
+        return isValid(frozenRoot, true, level, allocator, comp, AggrCalcT());
     }
     return true;
 }
diff --git a/searchlib/src/vespa/searchlib/btree/minmaxaggregated.h b/searchlib/src/vespa/searchlib/btree/minmaxaggregated.h
index e968e3bd122..65dd4839020 100644
--- a/searchlib/src/vespa/searchlib/btree/minmaxaggregated.h
+++ b/searchlib/src/vespa/searchlib/btree/minmaxaggregated.h
@@ -27,6 +27,14 @@ public:
     int32_t getMin() const { return _min; }
     int32_t getMax() const { return _max; }
 
+    bool operator==(const MinMaxAggregated &rhs) const {
+        return ((_min == rhs._min) && (_max == rhs._max));
+    }
+
+    bool operator!=(const MinMaxAggregated &rhs) const {
+        return ((_min != rhs._min) || (_max != rhs._max));
+    }
+
     void
     add(int32_t val)
     {
diff --git a/searchlib/src/vespa/searchlib/btree/noaggregated.h b/searchlib/src/vespa/searchlib/btree/noaggregated.h
index ae153d41f3a..33533f18014 100644
--- a/searchlib/src/vespa/searchlib/btree/noaggregated.h
+++ b/searchlib/src/vespa/searchlib/btree/noaggregated.h
@@ -11,6 +11,8 @@ class NoAggregated
 {
 public:
     NoAggregated() { }
+    bool operator==(const NoAggregated &) const { return true; }
+    bool operator!=(const NoAggregated &) const { return false; }
 };