From d31dc111ae8df2d8a24d843f1065e7dc8d2c3d77 Mon Sep 17 00:00:00 2001 From: Caolan McMahon Date: Tue, 28 Apr 2026 17:28:43 +0100 Subject: [PATCH 1/5] Reducer output stored on tree node pointers --- index.js | 46 ++++++++++++++++ lib/context.js | 4 +- lib/inflate.js | 2 +- lib/tree.js | 7 +-- lib/write.js | 102 +++++++++++++++++++++++++++++++---- reducer-example.mjs | 46 ++++++++++++++++ spec/hyperschema/index.js | 24 ++++++--- spec/hyperschema/schema.json | 6 +++ test/basic.js | 2 +- 9 files changed, 216 insertions(+), 23 deletions(-) create mode 100644 reducer-example.mjs diff --git a/index.js b/index.js index 5d11391..14291f4 100644 --- a/index.js +++ b/index.js @@ -303,6 +303,52 @@ class Hyperbee extends EventEmitter { ptr = v.children.get(i) } } + + 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? + 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 k = v.keys.get(i) + // TODO: test valuePointer? + values.push(k.value) + } + + // 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) + // TODO: avoid recursion? + 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 node + if (!v.reducers) v.reducers = {} + v.reducers[name] = result + ptr.changed = true + + // return result for this (sub)tree + return result + } } module.exports = Hyperbee diff --git a/lib/context.js b/lib/context.js index eeeb840..d38a4fe 100644 --- a/lib/context.js +++ b/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 diff --git a/lib/inflate.js b/lib/inflate.js index b1bf1ce..639df66 100644 --- a/lib/inflate.js +++ b/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) } diff --git a/lib/tree.js b/lib/tree.js index e471063..06d7c28 100644 --- a/lib/tree.js +++ b/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}]` } } diff --git a/lib/write.js b/lib/write.js index 70ab1d6..4c308bc 100644 --- a/lib/write.js +++ b/lib/write.js @@ -78,7 +78,7 @@ module.exports = class WriteBatch { this.hasLock = false } - async flush() { + async flush(reducers) { this.checkIfClosed() await this.lock() @@ -97,7 +97,7 @@ module.exports = class WriteBatch { const value = changed ? new TreeNode([], []) : null this.length = length - this.root = context.createTreeNode(0, seq, 0, changed, value) + this.root = context.createTreeNode(0, seq, 0, changed, value, null, null) for (const op of ops) { if (op.put) op.applied = await this._put(op.key, op.value) @@ -105,7 +105,7 @@ module.exports = class WriteBatch { if (op.applied) this.applied++ } - await this._flush() + await this._flush(reducers) await this.close() if (this.autoUpdate) { @@ -121,6 +121,52 @@ module.exports = class WriteBatch { return this.snapshot.close() } + // Returns: [updated node count, reducer result] + async _materializeReducer(ptr, name, reduce) { + // Already calculated? + const existing = ptr.reducers?.[name] + if (existing !== null && existing !== undefined) return [0, existing] + + const v = await retainAndInflate(ptr, this.snapshot, this.config) + + // Values stored at this node + const values = [] + for (let i = 0, len = v.keys.length; i < len; i++) { + const k = v.keys.get(i) + // TODO: test valuePointer? + values.push(k.value) + } + + // Store results to combine later + const rereduce = [reduce(values, false)] + + let updateCount = 1 + + // Results for child nodes + for (let i = 0, len = v.children.length; i < len; i++) { + const c = v.children.get(i) + // TODO: avoid recursion? + const [count, result] = await this._materializeReducer(c, name, reduce) + updateCount += count + rereduce.push(result) + } + + // Re-reduce if necessary (because this is not a leaf node) + const result = rereduce.length > 1 ? reduce(rereduce, true) : rereduce[0] + + // Store result on node + if (!ptr.reducers) ptr.reducers = {} + // Note: a reducer that returns undefined or null as its result would break + // detection of whether the reducer has been calculated for the current + // node. However, reducers MUST return JSON representable values so they + // can be encoded in the block anyway so undefined is already unsupported. + ptr.reducers[name] = result + ptr.changed = true + + // return result for this (sub)tree + return [updateCount, result] + } + async _put(key, value) { const stack = [] const target = key @@ -149,7 +195,10 @@ module.exports = class WriteBatch { const existing = await inflateValue(m, conf) if (b4a.equals(existing, value)) return false v.setValue(this.tree.context, mid, value) - for (let i = 0; i < stack.length; i++) stack[i].changed = true + for (let i = 0; i < stack.length; i++) { + stack[i].changed = true + delete stack[i].reducers + } return true } @@ -173,7 +222,11 @@ module.exports = class WriteBatch { } ptr.changed = true - for (let i = 0; i < stack.length; i++) stack[i].changed = true + delete ptr.reducers + for (let i = 0; i < stack.length; i++) { + stack[i].changed = true + delete stack[i].reducers + } while (status === NEEDS_SPLIT) { const v = await retainAndInflate(ptr, snap, conf) @@ -185,7 +238,15 @@ module.exports = class WriteBatch { status = p.insertNode(this.tree.context, median, right) ptr = parent } else { - this.root = this.tree.context.createTreeNode(0, 0, 0, true, new TreeNode([], [])) + this.root = this.tree.context.createTreeNode( + 0, + 0, + 0, + true, + new TreeNode([], []), + undefined, + undefined + ) this.root.value.keys.push(median) this.root.value.children.push(ptr) this.root.value.children.push(right) @@ -217,7 +278,10 @@ module.exports = class WriteBatch { if (v.children.ulength) await this._setKeyToNearestLeaf(v, mid, stack) else v.removeKey(mid) // we mark these as changed late, so we don't rewrite them if it is a 404 - for (let i = 0; i < stack.length; i++) stack[i].changed = true + for (let i = 0; i < stack.length; i++) { + stack[i].changed = true + delete stack[i].reducers + } this.root = await this._rebalance(stack) return true } @@ -302,6 +366,7 @@ module.exports = class WriteBatch { // maybe borrow from left sibling? if (l && l.keys.ulength > minKeys) { left.changed = true + delete left.reducers v.keys.unshift(p.keys.uget(index - 1)) if (l.children.ulength) v.children.unshift(l.children.pop()) p.keys.set(index - 1, l.keys.pop()) @@ -313,6 +378,7 @@ module.exports = class WriteBatch { // maybe borrow from right sibling? if (r && r.keys.ulength > minKeys) { right.changed = true + delete right.reducers v.keys.push(p.keys.uget(index)) if (r.children.ulength) v.children.push(r.children.shift()) p.keys.set(index, r.keys.shift()) @@ -330,9 +396,11 @@ module.exports = class WriteBatch { } left.changed = true + delete left.reducers l.merge(r, p.keys.uget(index)) parent.changed = true + delete parent.reducers p.removeKey(index) } @@ -348,7 +416,8 @@ module.exports = class WriteBatch { return k.value.byteLength <= this.inlineValueSize } - async _flush() { + async _flush(reducers) { + // TODO: this.root is always set in flush() so why check if falsy here? if (!this.root || !this.root.changed) { return } @@ -413,8 +482,23 @@ module.exports = class WriteBatch { const length = context.core.length + if (reducers) { + for (const [name, reducer] of Object.entries(reducers)) { + const [updateCount, _result] = await this._materializeReducer(this.root, name, reducer) + // Only consider applied if a subtree changed. If the only change + // being flushed is a recalculation of reducer at root it's probably + // not worth writing a batch for (and we've just recreated root so + // it will always update). + if (updateCount > 1) { + this.applied++ + } + } + } + // if noop and not genesis, bail early - if (this.applied === 0 && length > 0) return + if (this.applied === 0 && length > 0) { + return + } if (minValue > -1 && minValue + update.size < this.preferredBlockSize) { // TODO: repack the value into the block diff --git a/reducer-example.mjs b/reducer-example.mjs new file mode 100644 index 0000000..ad8fbc4 --- /dev/null +++ b/reducer-example.mjs @@ -0,0 +1,46 @@ +import Hyperbee from './index.js' +import Corestore from 'corestore' + +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 b = new Hyperbee(new Corestore('./sandbox/reducer')) + +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') +timeIt(async () => await b.reduce('count', count)) diff --git a/spec/hyperschema/index.js b/spec/hyperschema/index.js index 630a69e..c908845 100644 --- a/spec/hyperschema/index.js +++ b/spec/hyperschema/index.js @@ -44,15 +44,17 @@ const encoding1 = { if (m.core) c.uint.preencode(state, m.core) if (m.seq) c.uint.preencode(state, m.seq) if (m.offset) c.uint.preencode(state, m.offset) + if (m.reducers) c.json.preencode(state, m.reducers) }, encode(state, m) { - const flags = (m.core ? 1 : 0) | (m.seq ? 2 : 0) | (m.offset ? 4 : 0) + const flags = (m.core ? 1 : 0) | (m.seq ? 2 : 0) | (m.offset ? 4 : 0) | (m.reducers ? 8 : 0) c.uint8.encode(state, flags) if (m.core) c.uint.encode(state, m.core) if (m.seq) c.uint.encode(state, m.seq) if (m.offset) c.uint.encode(state, m.offset) + if (m.reducers) c.json.encode(state, m.reducers) }, decode(state) { const flags = c.uint8.decode(state) @@ -60,7 +62,8 @@ const encoding1 = { return { core: (flags & 1) !== 0 ? c.uint.decode(state) : 0, seq: (flags & 2) !== 0 ? c.uint.decode(state) : 0, - offset: (flags & 4) !== 0 ? c.uint.decode(state) : 0 + offset: (flags & 4) !== 0 ? c.uint.decode(state) : 0, + reducers: (flags & 8) !== 0 ? c.json.decode(state) : null } } } @@ -71,11 +74,13 @@ const encoding1_inline = { if (m.core) c.uint.preencode(state, m.core) if (m.seq) c.uint.preencode(state, m.seq) if (m.offset) c.uint.preencode(state, m.offset) + if (m.reducers) c.json.preencode(state, m.reducers) }, encode(state, m) { if (m.core) c.uint.encode(state, m.core) if (m.seq) c.uint.encode(state, m.seq) if (m.offset) c.uint.encode(state, m.offset) + if (m.reducers) c.json.encode(state, m.reducers) }, decode(state, inlining) { const flags = inlining @@ -83,7 +88,8 @@ const encoding1_inline = { return { core: (flags & 1) !== 0 ? c.uint.decode(state) : 0, seq: (flags & 2) !== 0 ? c.uint.decode(state) : 0, - offset: (flags & 4) !== 0 ? c.uint.decode(state) : 0 + offset: (flags & 4) !== 0 ? c.uint.decode(state) : 0, + reducers: (flags & 8) !== 0 ? c.json.decode(state) : null } } } @@ -91,7 +97,7 @@ const encoding1_inline = { // @bee/tree-delta const encoding2 = { preencode(state, m) { - state.end++ // flags are fixed size + state.end += 2 // flags are fixed size if (m.index) c.uint8.preencode(state, m.index) if (m.pointer) encoding1_inline.preencode(state, m.pointer) @@ -99,16 +105,20 @@ const encoding2 = { encode(state, m) { let flags = (m.type & 7) | (m.index ? 8 : 0) | (m.pointer ? 16 : 0) if (m.pointer) { - flags |= (m.pointer.core ? 32 : 0) | (m.pointer.seq ? 64 : 0) | (m.pointer.offset ? 128 : 0) + flags |= + (m.pointer.core ? 32 : 0) | + (m.pointer.seq ? 64 : 0) | + (m.pointer.offset ? 128 : 0) | + (m.pointer.reducers ? 256 : 0) } - c.uint8.encode(state, flags) + c.uint16.encode(state, flags) if (m.index) c.uint8.encode(state, m.index) if (m.pointer) encoding1_inline.encode(state, m.pointer) }, decode(state) { - const flags = c.uint8.decode(state) + const flags = c.uint16.decode(state) return { type: flags & 7, diff --git a/spec/hyperschema/schema.json b/spec/hyperschema/schema.json index 5207ec0..c4e3381 100644 --- a/spec/hyperschema/schema.json +++ b/spec/hyperschema/schema.json @@ -47,6 +47,12 @@ "name": "offset", "type": "uint", "version": 1 + }, + { + "name": "reducers", + "required": false, + "type": "json", + "version": 1 } ] }, diff --git a/test/basic.js b/test/basic.js index c4e567d..4147939 100644 --- a/test/basic.js +++ b/test/basic.js @@ -139,7 +139,7 @@ test('basic delete', async function (t) { const length = db.core.length const w = db.write() w.tryDelete(b4a.from('b')) - await w.flush({ debug: true }) + await w.flush() t.ok(length < db.core.length) } From 53c581d283ade0f059ceb527ddb0ace4e1fb2b4e Mon Sep 17 00:00:00 2001 From: Caolan McMahon Date: Tue, 5 May 2026 14:42:54 +0100 Subject: [PATCH 2/5] Add reduceRange method --- index.js | 72 ++++++++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 68 insertions(+), 4 deletions(-) diff --git a/index.js b/index.js index 14291f4..3e1b3b0 100644 --- a/index.js +++ b/index.js @@ -304,6 +304,71 @@ class Hyperbee extends EventEmitter { } } + 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 current = [] + 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 + current.push(data) + i++ + } + if (current.length) { + subtrees.push(reduce(current, 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) } @@ -341,10 +406,9 @@ class Hyperbee extends EventEmitter { // Re-reduce if necessary (because this is not a leaf node) const result = rereduce.length > 1 ? reduce(rereduce, true) : rereduce[0] - // Store result on node - if (!v.reducers) v.reducers = {} - v.reducers[name] = result - ptr.changed = true + // Store result on tree node pointer + if (!ptr.reducers) ptr.reducers = {} + ptr.reducers[name] = result // return result for this (sub)tree return result From 62bbdf382fae60ccfb1b7abf9c102e687243c234 Mon Sep 17 00:00:00 2001 From: Caolan McMahon Date: Tue, 5 May 2026 15:32:11 +0100 Subject: [PATCH 3/5] Allow temporary unnamed reducers --- index.js | 36 ++++++++++++++++++++---------------- 1 file changed, 20 insertions(+), 16 deletions(-) diff --git a/index.js b/index.js index 3e1b3b0..309dbf8 100644 --- a/index.js +++ b/index.js @@ -317,11 +317,11 @@ class Hyperbee extends EventEmitter { 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++ - } + 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) { @@ -339,11 +339,11 @@ class Hyperbee extends EventEmitter { 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++ - } + 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 @@ -363,7 +363,7 @@ class Hyperbee extends EventEmitter { } // Process last child if not exited outer loop early subtrees.push(await this._reduceRange(v.children.get(i), name, reduce, start, end)) - } while (false); + } while (false) } return reduce(subtrees, true) @@ -378,9 +378,11 @@ class Hyperbee extends EventEmitter { // see if logic can be shared. async _reduce(ptr, name, reduce) { // Already calculated? - const existing = ptr.reducers?.[name] - if (existing !== null && existing !== undefined) { - return existing + 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) @@ -407,8 +409,10 @@ class Hyperbee extends EventEmitter { const result = rereduce.length > 1 ? reduce(rereduce, true) : rereduce[0] // Store result on tree node pointer - if (!ptr.reducers) ptr.reducers = {} - ptr.reducers[name] = result + if (name) { + if (!ptr.reducers) ptr.reducers = {} + ptr.reducers[name] = result + } // return result for this (sub)tree return result From 6328977d86e55333d636b5a0efd387c6f007e16a Mon Sep 17 00:00:00 2001 From: Caolan McMahon Date: Tue, 5 May 2026 16:43:28 +0100 Subject: [PATCH 4/5] Provide entries with key and value properties to reducers --- .../count-reducer.mjs | 16 ++++- example/total-reducer.mjs | 58 +++++++++++++++++++ index.js | 14 ++--- lib/write.js | 5 +- 4 files changed, 79 insertions(+), 14 deletions(-) rename reducer-example.mjs => example/count-reducer.mjs (62%) create mode 100644 example/total-reducer.mjs diff --git a/reducer-example.mjs b/example/count-reducer.mjs similarity index 62% rename from reducer-example.mjs rename to example/count-reducer.mjs index ad8fbc4..f2806e4 100644 --- a/reducer-example.mjs +++ b/example/count-reducer.mjs @@ -1,5 +1,6 @@ -import Hyperbee from './index.js' +import Hyperbee from '../index.js' import Corestore from 'corestore' +import path from 'node:path' const count = (values, rereduce) => { if (rereduce) { @@ -13,7 +14,8 @@ const count = (values, rereduce) => { const reducers = { count } -const b = new Hyperbee(new Corestore('./sandbox/reducer')) +const storePath = path.resolve(import.meta.dirname, '../sandbox/count-reducer') +const b = new Hyperbee(new Corestore(storePath)) await b.ready() @@ -43,4 +45,12 @@ async function timeIt(f) { } console.log('Time query of count reducer') -timeIt(async () => await b.reduce('count', count)) +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)) diff --git a/example/total-reducer.mjs b/example/total-reducer.mjs new file mode 100644 index 0000000..fb86fbf --- /dev/null +++ b/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().substring(1)) + } + } + 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)) diff --git a/index.js b/index.js index 309dbf8..87770bc 100644 --- a/index.js +++ b/index.js @@ -311,7 +311,7 @@ class Hyperbee extends EventEmitter { async _reduceRange(ptr, name, reduce, start, end) { const v = ptr.value ? this.bump(ptr) : await this.inflate(ptr, this.config) - const current = [] + const values = [] const subtrees = [] let i = 0 @@ -327,11 +327,11 @@ class Hyperbee extends EventEmitter { while (i < v.keys.length) { const data = v.keys.get(i) if (end && b4a.compare(data.key, end) >= 0) break - current.push(data) + values.push(data) i++ } - if (current.length) { - subtrees.push(reduce(current, false)) + if (values.length) { + subtrees.push(reduce(values, false)) } if (v.children.length) { @@ -390,9 +390,8 @@ class Hyperbee extends EventEmitter { // Values stored at this node const values = [] for (let i = 0, len = v.keys.length; i < len; i++) { - const k = v.keys.get(i) - // TODO: test valuePointer? - values.push(k.value) + const data = v.keys.get(i) + values.push(data) } // Store results to combine later @@ -401,7 +400,6 @@ class Hyperbee extends EventEmitter { // Results for child nodes for (let i = 0, len = v.children.length; i < len; i++) { const c = v.children.get(i) - // TODO: avoid recursion? rereduce.push(await this._reduce(c, name, reduce)) } diff --git a/lib/write.js b/lib/write.js index 4c308bc..8467101 100644 --- a/lib/write.js +++ b/lib/write.js @@ -132,9 +132,8 @@ module.exports = class WriteBatch { // Values stored at this node const values = [] for (let i = 0, len = v.keys.length; i < len; i++) { - const k = v.keys.get(i) - // TODO: test valuePointer? - values.push(k.value) + const data = v.keys.get(i) + values.push(data) } // Store results to combine later From eb0838717305dc3888bb647c89c95487f69bc78d Mon Sep 17 00:00:00 2001 From: Caolan McMahon Date: Tue, 5 May 2026 16:54:25 +0100 Subject: [PATCH 5/5] Document reducer methods --- README.md | 65 ++++++++++++++++++++++++++++++++++++++- example/total-reducer.mjs | 8 ++--- 2 files changed, 68 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 550599c..b4a03d2 100644 --- a/README.md +++ b/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. diff --git a/example/total-reducer.mjs b/example/total-reducer.mjs index fb86fbf..5d50b55 100644 --- a/example/total-reducer.mjs +++ b/example/total-reducer.mjs @@ -8,7 +8,7 @@ const total = (values, rereduce) => { if (rereduce) { total += v } else { - total += Number(v.value.toString().substring(1)) + total += Number(v.value.toString()) } } return total @@ -26,7 +26,7 @@ if (b.core.length === 0) { const w = b.write() for (let i = 0; i < 1_000_000; i++) { - w.tryPut(Buffer.from('#' + i), Buffer.from('#' + i)) + w.tryPut(Buffer.from('' + i), Buffer.from('' + i)) } await w.flush() @@ -35,7 +35,7 @@ if (b.core.length === 0) { const w = b.write() - w.tryPut(Buffer.from('#500000' + Math.random()), Buffer.from('#500000' + Math.random())) + w.tryPut(Buffer.from('500000' + Math.random()), Buffer.from('500000' + Math.random())) await w.flush(reducers) } @@ -51,7 +51,7 @@ 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')) + async () => await b.reduceRange('total', total, Buffer.from('250000'), Buffer.from('750000')) ) console.log('Time query of temporary reducer')