support map

index.js

@@ -226,7 +226,7 @@ class Hyperbee extends EventEmitter {
 
   async reindex(
     until,
-    { timeout = this.config.timeout, wait = this.config.wait, prefetch = 128 } = {}
+    { timeout = this.config.timeout, wait = this.config.wait, prefetch = 128, map = null } = {}
   ) {
     if (!this.writable) throw new Error('Not writable')
 
@@ -246,7 +246,7 @@ class Hyperbee extends EventEmitter {
     await local.lock.lock()
 
     try {
-      await copyChanges(this, changes, config)
+      await copyChanges(this, changes, config, map)
     } finally {
       local.lock.unlock()
     }

lib/copy.js

@@ -1,7 +1,7 @@
 const b4a = require('b4a')
 const { encodeBlock, TYPE_COMPAT } = require('./encoding.js')
 
-module.exports = async function copyChanges(tree, changes, config) {
+module.exports = async function copyChanges(tree, changes, config, map) {
   const local = tree.context.getLocalContext()
 
   await local.core.ready()
@@ -23,6 +23,8 @@ module.exports = async function copyChanges(tree, changes, config) {
 
     for (let j = 0; j < batch.length; j++) copied.set(srcHex + ':' + (first + j), start + j)
 
+    if (map !== null) await mapValues(batch, first, change, map)
+
     for (let j = 0; j < batch.length; j++) {
       const blk = batch[j]
 
@@ -99,3 +101,34 @@ module.exports = async function copyChanges(tree, changes, config) {
     return { core, seq: tail.length - 1 }
   }
 }
+
+// values live in the same batch as their key
+async function mapValues(batch, first, change, map) {
+  for (const blk of batch) {
+    if (blk.type === TYPE_COMPAT) throw new Error('Cannot index compat blocks')
+    if (blk.keys === null) continue
+
+    for (const k of blk.keys) {
+      const ptr = k.value === null ? k.valuePointer : null
+      if (k.value === null && ptr === null) continue
+
+      const vals = ptr === null ? null : getValues(batch, first, ptr)
+      const value = ptr === null ? k.value : vals[ptr.offset]
+      const next = await map(k.key, value, change)
+
+      if (!next || next === value) continue
+
+      if (ptr === null) k.value = next
+      else vals[ptr.offset] = next
+    }
+  }
+}
+
+function getValues(batch, first, ptr) {
+  const i = ptr.seq - first
+  const blk = ptr.core === 0 && ptr.split === 0 && i >= 0 && i < batch.length ? batch[i] : null
+  if (blk === null || blk.values === null || ptr.offset >= blk.values.length) {
+    throw new Error('Cannot map a value stored outside its batch')
+  }
+  return blk.values
+}

test/basic.js

@@ -1475,3 +1475,65 @@ test('undo with timeout and wait options', async function (t) {
   await t.exception(noWait.get(b4a.from('a')), /BLOCK_NOT_AVAILABLE/)
   await noWait.close()
 })
+
+test('reindex maps values as they are copied', async function (t) {
+  const [a, b] = await createMultiple(t, 2)
+  await a.ready()
+  await b.ready()
+
+  const big = b4a.alloc(4096).fill('x')
+  const huge = b4a.alloc(3 * 4096).fill('y') // spans several value blocks
+  const heads = []
+
+  for (let i = 0; i < 3; i++) {
+    const w = b.write()
+    w.tryPut(b4a.from('inline' + i), b4a.from('val' + i))
+    w.tryPut(b4a.from('big' + i), big)
+    w.tryPut(b4a.from('huge' + i), huge)
+    w.tryPut(b4a.from('keep' + i), b4a.from('keep' + i))
+    await w.flush()
+    heads.push(b.head())
+  }
+
+  a.move(b.head())
+
+  const seen = []
+
+  const n = await a.reindex(() => false, {
+    map: (key, value, change) => {
+      seen.push([b4a.toString(key), value.byteLength, change.head.length])
+      const name = b4a.toString(key)
+      if (name.startsWith('inline')) return b4a.from('mapped' + name.slice(6))
+      if (name.startsWith('big')) return b4a.concat([b4a.from('mapped'), value])
+      if (name.startsWith('huge')) return b4a.concat([b4a.from('mapped'), value])
+      return null
+    }
+  })
+
+  t.is(n, 3)
+  t.is(seen.length, 12, 'every copied key is offered to map')
+  t.ok(
+    seen.some(([name, size]) => name === 'big0' && size === big.byteLength),
+    'pointed values are inflated for map'
+  )
+
+  a.cache.empty()
+
+  for (let i = 0; i < 3; i++) {
+    t.alike((await a.get(b4a.from('inline' + i))).value, b4a.from('mapped' + i))
+    t.alike((await a.get(b4a.from('big' + i))).value, b4a.concat([b4a.from('mapped'), big]))
+    t.alike((await a.get(b4a.from('huge' + i))).value, b4a.concat([b4a.from('mapped'), huge]))
+    t.alike((await a.get(b4a.from('keep' + i))).value, b4a.from('keep' + i))
+  }
+
+  const copied = []
+  for await (const { head } of a.createChangesStream()) copied.unshift(head)
+  t.is(copied.length, 3)
+
+  for (let i = 0; i < 3; i++) {
+    const c = a.checkout(copied[i])
+    t.alike((await c.get(b4a.from('inline' + i))).value, b4a.from('mapped' + i), 'version ' + i)
+    t.is(await c.get(b4a.from('inline' + (i + 1))), null)
+    await c.close()
+  }
+})