Reduce method

README.md

@@ -380,6 +380,63 @@ Options:
 }
 ```
 
+#### `await db.reduce(name, reducer)`
+
+Calculates an accumulated value for all entries in the tree.
+
+To cache intermediate results between calls, provide a string as the
+`name` argument. This must be unique to the reducer function and be
+updated if the reducer changes in a backwards-incompatible way.
+
+To perform a one-off temporary reduce, provide `null` as the name.
+
+The `reducer` is a function that will calculate the accumulated value.
+It takes two arguments: `values` and `rereduce`.
+
+When `rereduce` is `false`, the `values` argument will be an Array of
+tree entries with `key` and `value` properties. When the `rereduce`
+argument is `true`, the `values` argument will be an Array of values
+returned from previous `reducer` calls.
+
+The output of a reducer must be a JSON-compatible value.
+
+Note: while entries provided to the reducer are in sorted key order,
+those entries might not be contiguous across reducer calls. For example,
+a reducer might receive `[3,4], [6,7], [9], [5,8]` across 4 calls.
+
+```js
+import Hyperbee from '../index.js'
+import Corestore from 'corestore'
+
+const total = (values, rereduce) => {
+  let total = 0
+  for (const v of values) {
+    if (rereduce) {
+      total += v
+    } else {
+      total += Number(v.value.toString())
+    }
+  }
+  return total
+}
+
+const b = new Hyperbee(new Corestore('./sandbox'))
+await b.ready()
+
+// Calculate total of all number strings
+console.log(await b.reduce('total', total))
+```
+
+#### `await db.reduceRange(name, reducer, start, end)`
+
+Calculates an accumulated value for a range of entries in the tree.
+`start` can be `null` (to begin with the first entry) or a `Buffer` less
+than or equal to the first key to include. `end` can be `null` (to end
+after the last entry), or a `Buffer` greater than the last key to include.
+
+The `name` and `reducer` arguments are described in the documentation for
+`db.reduce()`.
+
 #### `b.on('ready', listener)`
 
 Emitted once the Hyperbee is ready for use.
@@ -481,11 +538,17 @@ does not exist, this method does nothing (and will not throw).
 
 Queues an operation to clear all entries from the tree.
 
-#### `await batch.flush()`
+#### `await batch.flush([reducers])`
 
 Aquires an exclusive write lock and applies the operations queued in this
 batch to the tree, clearing the queue.
 
+The `reducers` argument is an Object with `reducer` functions (as described by
+the documentation for `db.reduce()`) keyed by unique strings. If provided,
+these will be recalculated for all nodes lacking a cached reduce result and
+updated values will be written to Hypercore as part of the batch. This greatly
+improves query time for reducers at the expense of writing a larger batch.
+
 **Warning:** continuing to use the batch after flushing can cause unpredictable
 behavior. Batches applied after the first flush will be 'unapplied' if you
 flush again later. This can lead to accidentally removing data from the tree.

example/count-reducer.mjs

@@ -0,0 +1,56 @@
+import Hyperbee from '../index.js'
+import Corestore from 'corestore'
+import path from 'node:path'
+
+const count = (values, rereduce) => {
+  if (rereduce) {
+    let total = 0
+    for (const v of values) total += v
+    return total
+  } else {
+    return values.length
+  }
+}
+
+const reducers = { count }
+
+const storePath = path.resolve(import.meta.dirname, '../sandbox/count-reducer')
+const b = new Hyperbee(new Corestore(storePath))
+
+await b.ready()
+
+if (b.core.length === 0) {
+  console.log('initial write, no materialized view')
+  const w = b.write()
+
+  for (let i = 0; i < 1_000_000; i++) {
+    w.tryPut(Buffer.from('#' + i), Buffer.from('#' + i))
+  }
+
+  await w.flush()
+} else {
+  console.log('add one more entry and materialize view')
+
+  const w = b.write()
+
+  w.tryPut(Buffer.from('#500000' + Math.random()), Buffer.from('#500000' + Math.random()))
+
+  await w.flush(reducers)
+}
+
+async function timeIt(f) {
+  const t = performance.now()
+  console.log(await f())
+  console.log('Elapsed:', (performance.now() - t).toFixed(3), 'ms')
+}
+
+console.log('Time query of count reducer')
+await timeIt(async () => await b.reduce('count', count))
+
+console.log('Time query of count reducer over range')
+await timeIt(
+  async () => await b.reduceRange('count', count, Buffer.from('#250_000'), Buffer.from('#750_000'))
+)
+
+console.log('Time query of temporary reducer')
+await timeIt(async () => await b.reduce(null, count))

example/total-reducer.mjs

@@ -0,0 +1,58 @@
+import Hyperbee from '../index.js'
+import Corestore from 'corestore'
+import path from 'node:path'
+
+const total = (values, rereduce) => {
+  let total = 0
+  for (const v of values) {
+    if (rereduce) {
+      total += v
+    } else {
+      total += Number(v.value.toString())
+    }
+  }
+  return total
+}
+
+const reducers = { total }
+
+const storePath = path.resolve(import.meta.dirname, '../sandbox/total-reducer')
+const b = new Hyperbee(new Corestore(storePath))
+
+await b.ready()
+
+if (b.core.length === 0) {
+  console.log('initial write, no materialized view')
+  const w = b.write()
+
+  for (let i = 0; i < 1_000_000; i++) {
+    w.tryPut(Buffer.from('' + i), Buffer.from('' + i))
+  }
+
+  await w.flush()
+} else {
+  console.log('add one more entry and materialize view')
+
+  const w = b.write()
+
+  w.tryPut(Buffer.from('500000' + Math.random()), Buffer.from('500000' + Math.random()))
+
+  await w.flush(reducers)
+}
+
+async function timeIt(f) {
+  const t = performance.now()
+  console.log(await f())
+  console.log('Elapsed:', (performance.now() - t).toFixed(3), 'ms')
+}
+
+console.log('Time query of total reducer')
+await timeIt(async () => await b.reduce('total', total))
+
+console.log('Time query of total reducer over range')
+await timeIt(
+  async () => await b.reduceRange('total', total, Buffer.from('250000'), Buffer.from('750000'))
+)
+
+console.log('Time query of temporary reducer')
+await timeIt(async () => await b.reduce(null, total))

index.js

@@ -303,6 +303,118 @@ class Hyperbee extends EventEmitter {
       ptr = v.children.get(i)
     }
   }
+
+  async reduceRange(name, reduce, start, end) {
+    return await this._reduceRange(this.root, name, reduce, start, end)
+  }
+
+  async _reduceRange(ptr, name, reduce, start, end) {
+    const v = ptr.value ? this.bump(ptr) : await this.inflate(ptr, this.config)
+
+    const values = []
+    const subtrees = []
+
+    let i = 0
+    // Skip keys outside of range
+    if (start) {
+      while (i < v.keys.length) {
+        const data = v.keys.get(i)
+        if (b4a.compare(data.key, start) >= 0) break
+        i++
+      }
+    }
+    // Process keys until end or last key reached
+    while (i < v.keys.length) {
+      const data = v.keys.get(i)
+      if (end && b4a.compare(data.key, end) >= 0) break
+      values.push(data)
+      i++
+    }
+    if (values.length) {
+      subtrees.push(reduce(values, false))
+    }
+
+    if (v.children.length) {
+      outer: do {
+        let first = 0
+        // Skip children outside of range
+        if (start) {
+          while (first < v.keys.length) {
+            const data = v.keys.get(first)
+            if (b4a.compare(data.key, start) > 0) break
+            first++
+          }
+        }
+        // Process children until end or last key reached
+        let i = first
+        while (i < v.keys.length) {
+          const data = v.keys.get(i)
+          const isLast = end && b4a.compare(data.key, end) >= 0
+          if (i === first || isLast) {
+            subtrees.push(await this._reduceRange(v.children.get(i), name, reduce, start, end))
+          } else {
+            subtrees.push(await this._reduce(v.children.get(i), name, reduce))
+          }
+          if (isLast) {
+            // This was the last child to process
+            break outer
+          }
+          i++
+        }
+        // Process last child if not exited outer loop early
+        subtrees.push(await this._reduceRange(v.children.get(i), name, reduce, start, end))
+      } while (false)
+    }
+
+    return reduce(subtrees, true)
+  }
+
+  async reduce(name, reduce) {
+    return await this._reduce(this.root, name, reduce)
+  }
+
+  // TODO: avoid recursion?
+  // TODO: this shares a lot of structure with _materializeReducer in write.js,
+  // see if logic can be shared.
+  async _reduce(ptr, name, reduce) {
+    // Already calculated?
+    if (name) {
+      const existing = ptr.reducers?.[name]
+      if (existing !== null && existing !== undefined) {
+        return existing
+      }
+    }
+
+    const v = ptr.value ? this.bump(ptr) : await this.inflate(ptr, this.config)
+
+    // Values stored at this node
+    const values = []
+    for (let i = 0, len = v.keys.length; i < len; i++) {
+      const data = v.keys.get(i)
+      values.push(data)
+    }
+
+    // Store results to combine later
+    const rereduce = [reduce(values, false)]
+
+    // Results for child nodes
+    for (let i = 0, len = v.children.length; i < len; i++) {
+      const c = v.children.get(i)
+      rereduce.push(await this._reduce(c, name, reduce))
+    }
+
+    // Re-reduce if necessary (because this is not a leaf node)
+    const result = rereduce.length > 1 ? reduce(rereduce, true) : rereduce[0]
+
+    // Store result on tree node pointer
+    if (name) {
+      if (!ptr.reducers) ptr.reducers = {}
+      ptr.reducers[name] = result
+    }
+
+    // return result for this (sub)tree
+    return result
+  }
 }
 
 module.exports = Hyperbee

lib/context.js

@@ -74,8 +74,8 @@ class CoreContext {
     this.length = length
   }
 
-  createTreeNode(core, seq, offset, changed, value) {
-    const ptr = new TreeNodePointer(this, core, seq, offset, changed, value)
+  createTreeNode(core, seq, offset, changed, value, reducers) {
+    const ptr = new TreeNodePointer(this, core, seq, offset, changed, value, reducers)
     if (ptr.value || changed) return ptr
     const existing = this.cache.get(ptr)
     if (existing) return existing

lib/inflate.js

@@ -114,7 +114,7 @@ async function inflateChild(context, d, ptr, block, config) {
 
 function inflateChildDelta(context, d, ptr, block, config) {
   const p = d.pointer
-  const c = p && context.createTreeNode(p.core, p.seq, p.offset, false, null)
+  const c = p && context.createTreeNode(p.core, p.seq, p.offset, false, null, p.reducers)
   return new DeltaOp(false, d.type, d.index, c)
 }
 

lib/tree.js

@@ -65,10 +65,11 @@ class KeyPointer extends Pointer {
 }
 
 class TreeNodePointer extends Pointer {
-  constructor(context, core, seq, offset, changed, value) {
+  constructor(context, core, seq, offset, changed, value, reducers) {
     super(context, core, seq, offset, changed)
 
     this.value = value
+    this.reducers = reducers
 
     this.next = null
     this.prev = null
@@ -79,7 +80,7 @@ class TreeNodePointer extends Pointer {
     const value = new TreeNode([], [])
     value.keys = this.value.keys.commit()
     value.children = this.value.children.commit()
-    const ptr = this.context.createTreeNode(this.core, 0, 0, true, value)
+    const ptr = this.context.createTreeNode(this.core, 0, 0, true, value, this.reducers)
     return ptr
   }
 
@@ -94,7 +95,7 @@ class TreeNodePointer extends Pointer {
 
   // TODO: remove, left here for easier debugging for now
   [Symbol.for('nodejs.util.inspect.custom')]() {
-    return `[TreeNodePointer core=${this.core} seq=${this.seq} offset=${this.offset} changed=${this.changed}]`
+    return `[TreeNodePointer core=${this.core} seq=${this.seq} offset=${this.offset} reducers=${JSON.stringify(this.reducers)} changed=${this.changed}]`
   }
 }