Revert to old index implementation (#76)

* Revert back to map implementation of index

* Modify index tests to no longer assert uncompacted form

* Remove obsolete comments

* changeset

* Refactor reduce

* Update changeset

Author: kevin-dp

.changeset/evil-pianos-hear.md

@@ -0,0 +1,5 @@
+---
+'@electric-sql/d2mini': patch
+---
+
+Modify index implementation to keep a map of consolidated values and their multiplicities. This improves efficiency to get a value's multiplicity since it's already precomputed. Also modify reduce operator to emit a single diff instead of 2 diffs (1 that is -oldMultiplicity and 1 that is +newMultiplicity).

packages/d2mini/src/indexes.ts

@@ -7,24 +7,32 @@ import { DefaultMap, hash } from './utils.js'
  * exploit the key-value structure of the data to run efficiently.
  */
 export class Index<K, V> {
-  #inner: DefaultMap<K, [V, number][]>
-  #changedKeys: Set<K>
+  #inner: DefaultMap<K, DefaultMap<string, [V, number]>>
 
   constructor() {
-    this.#inner = new DefaultMap<K, [V, number][]>(() => [])
-    this.#changedKeys = new Set<K>()
+    this.#inner = new DefaultMap<K, DefaultMap<string, [V, number]>>(
+      () =>
+        new DefaultMap<string, [V, number]>(() => [undefined as any as V, 0]),
+    )
+    // #inner is as map of:
+    // {
+    //   [key]: {
+    //     [hash(value)]: [value, multiplicity]
+    //   }
+    // }
   }
 
   toString(indent = false): string {
     return `Index(${JSON.stringify(
-      [...this.#inner],
+      [...this.#inner].map(([k, valueMap]) => [k, [...valueMap]]),
       undefined,
       indent ? '  ' : undefined,
     )})`
   }
 
   get(key: K): [V, number][] {
-    return this.#inner.get(key)
+    const valueMap = this.#inner.get(key)
+    return [...valueMap.values()]
   }
 
   entries() {
@@ -36,78 +44,44 @@ export class Index<K, V> {
   }
 
   has(key: K): boolean {
-    return this.#inner.has(key) && this.#inner.get(key).length > 0
+    return this.#inner.has(key)
   }
 
   get size(): number {
-    let count = 0
-    for (const [, values] of this.#inner.entries()) {
-      if (values.length > 0) {
-        count++
-      }
-    }
-    return count
+    return this.#inner.size
   }
 
   addValue(key: K, value: [V, number]): void {
-    const values = this.#inner.get(key)
-    values.push(value)
-    this.#changedKeys.add(key)
-  }
-
-  append(other: Index<K, V>): void {
-    for (const [key, otherValues] of other.entries()) {
-      const thisValues = this.#inner.get(key)
-      for (const value of otherValues) {
-        thisValues.push(value)
-      }
-      this.#changedKeys.add(key)
-    }
-  }
-
-  compact(keys: K[] = []): void {
-    // If no keys specified, use the changed keys
-    const keysToProcess = keys.length === 0 ? [...this.#changedKeys] : keys
-
-    for (const key of keysToProcess) {
-      if (!this.#inner.has(key)) continue
-
-      const values = this.#inner.get(key)
-      const consolidated = this.consolidateValues(values)
-
-      // Remove the key entirely and re-add only if there are non-zero values
-      this.#inner.delete(key)
-      if (consolidated.length > 0) {
-        this.#inner.get(key).push(...consolidated)
-      }
-    }
-
-    // Clear the changed keys after compaction
-    if (keys.length === 0) {
-      this.#changedKeys.clear()
-    } else {
-      // Only remove the keys that were explicitly compacted
-      for (const key of keys) {
-        this.#changedKeys.delete(key)
+    const [val, multiplicity] = value
+    const valueMap = this.#inner.get(key)
+    const valueHash = hash(val)
+    const [, existingMultiplicity] = valueMap.get(valueHash)
+    const newMultiplicity = existingMultiplicity + multiplicity
+    if (multiplicity !== 0) {
+      if (newMultiplicity === 0) {
+        valueMap.delete(valueHash)
+      } else {
+        valueMap.set(valueHash, [val, newMultiplicity])
       }
     }
   }
 
-  private consolidateValues(values: [V, number][]): [V, number][] {
-    const consolidated = new Map<string, { value: V; multiplicity: number }>()
-
-    for (const [value, multiplicity] of values) {
-      const valueHash = hash(value)
-      if (consolidated.has(valueHash)) {
-        consolidated.get(valueHash)!.multiplicity += multiplicity
-      } else {
-        consolidated.set(valueHash, { value, multiplicity })
+  append(other: Index<K, V>): void {
+    for (const [key, otherValueMap] of other.entries()) {
+      const thisValueMap = this.#inner.get(key)
+      for (const [
+        valueHash,
+        [value, multiplicity],
+      ] of otherValueMap.entries()) {
+        const [, existingMultiplicity] = thisValueMap.get(valueHash)
+        const newMultiplicity = existingMultiplicity + multiplicity
+        if (newMultiplicity === 0) {
+          thisValueMap.delete(valueHash)
+        } else {
+          thisValueMap.set(valueHash, [value, newMultiplicity])
+        }
       }
     }
-
-    return [...consolidated.values()]
-      .filter(({ multiplicity }) => multiplicity !== 0)
-      .map(({ value, multiplicity }) => [value, multiplicity])
   }
 
   join<V2>(other: Index<K, V2>): MultiSet<[K, [V, V2]]> {
@@ -116,23 +90,23 @@ export class Index<K, V> {
     // We want to iterate over the smaller of the two indexes to reduce the
     // number of operations we need to do.
     if (this.size <= other.size) {
-      for (const [key, values1] of this.entries()) {
+      for (const [key, valueMap] of this.entries()) {
         if (!other.has(key)) continue
-        const values2 = other.get(key)
-        for (const [val1, mul1] of values1) {
-          for (const [val2, mul2] of values2) {
+        const otherValues = other.get(key)
+        for (const [val1, mul1] of valueMap.values()) {
+          for (const [val2, mul2] of otherValues) {
             if (mul1 !== 0 && mul2 !== 0) {
               result.push([[key, [val1, val2]], mul1 * mul2])
             }
           }
         }
       }
     } else {
-      for (const [key, values2] of other.entries()) {
+      for (const [key, otherValueMap] of other.entries()) {
         if (!this.has(key)) continue
-        const values1 = this.get(key)
-        for (const [val2, mul2] of values2) {
-          for (const [val1, mul1] of values1) {
+        const values = this.get(key)
+        for (const [val2, mul2] of otherValueMap.values()) {
+          for (const [val1, mul1] of values) {
             if (mul1 !== 0 && mul2 !== 0) {
               result.push([[key, [val1, val2]], mul1 * mul2])
             }

packages/d2mini/src/operators/join.ts

@@ -77,13 +77,6 @@ export class JoinOperator<K, V1, V2> extends BinaryOperator<
 
     // Append deltaB to indexB
     this.#indexB.append(deltaB)
-
-    // Compact both indexes to consolidate values and remove zero-multiplicity entries
-    // Only compact changed keys for efficiency
-    deltaA.compact()
-    deltaB.compact()
-    this.#indexA.compact()
-    this.#indexB.compact()
   }
 }
 

packages/d2mini/src/operators/reduce copy.ts

@@ -0,0 +1,103 @@
+import { IStreamBuilder, KeyValue } from '../types.js'
+import {
+  DifferenceStreamReader,
+  DifferenceStreamWriter,
+  UnaryOperator,
+} from '../graph.js'
+import { StreamBuilder } from '../d2.js'
+import { MultiSet } from '../multiset.js'
+import { Index } from '../indexes.js'
+import { hash } from '../utils.js'
+
+/**
+ * Base operator for reduction operations (version-free)
+ */
+export class ReduceOperator<K, V1, V2> extends UnaryOperator<[K, V1], [K, V2]> {
+  #index = new Index<K, V1>()
+  #indexOut = new Index<K, V2>()
+  #f: (values: [V1, number][]) => [V2, number][]
+
+  constructor(
+    id: number,
+    inputA: DifferenceStreamReader<[K, V1]>,
+    output: DifferenceStreamWriter<[K, V2]>,
+    f: (values: [V1, number][]) => [V2, number][],
+  ) {
+    super(id, inputA, output)
+    this.#f = f
+  }
+
+  run(): void {
+    const keysTodo = new Set<K>()
+
+    // Collect all input messages and update the index
+    for (const message of this.inputMessages()) {
+      for (const [item, multiplicity] of message.getInner()) {
+        const [key, value] = item
+        this.#index.addValue(key, [value, multiplicity])
+        keysTodo.add(key)
+      }
+    }
+
+    // For each key, compute the reduction and delta
+    const result: [[K, V2], number][] = []
+    for (const key of keysTodo) {
+      const curr = this.#index.get(key)
+      const currOut = this.#indexOut.get(key)
+      const out = this.#f(curr)
+
+      // Calculate delta between current and previous output
+      const delta = new Map<string, number>()
+      const values = new Map<string, V2>()
+      for (const [value, multiplicity] of out) {
+        const valueKey = hash(value)
+        values.set(valueKey, value)
+        delta.set(valueKey, (delta.get(valueKey) || 0) + multiplicity)
+      }
+      for (const [value, multiplicity] of currOut) {
+        const valueKey = hash(value)
+        values.set(valueKey, value)
+        delta.set(valueKey, (delta.get(valueKey) || 0) - multiplicity)
+      }
+
+      // Add non-zero deltas to result
+      for (const [valueKey, multiplicity] of delta) {
+        const value = values.get(valueKey)!
+        if (multiplicity !== 0) {
+          result.push([[key, value], multiplicity])
+          this.#indexOut.addValue(key, [value, multiplicity])
+        }
+      }
+    }
+
+    if (result.length > 0) {
+      this.output.sendData(new MultiSet(result))
+    }
+  }
+}
+
+/**
+ * Reduces the elements in the stream by key (version-free)
+ */
+export function reduce<
+  K extends T extends KeyValue<infer K, infer _V> ? K : never,
+  V1 extends T extends KeyValue<K, infer V> ? V : never,
+  R,
+  T,
+>(f: (values: [V1, number][]) => [R, number][]) {
+  return (stream: IStreamBuilder<T>): IStreamBuilder<KeyValue<K, R>> => {
+    const output = new StreamBuilder<KeyValue<K, R>>(
+      stream.graph,
+      new DifferenceStreamWriter<KeyValue<K, R>>(),
+    )
+    const operator = new ReduceOperator<K, V1, R>(
+      stream.graph.getNextOperatorId(),
+      stream.connectReader() as DifferenceStreamReader<KeyValue<K, V1>>,
+      output.writer,
+      f,
+    )
+    stream.graph.addOperator(operator)
+    stream.graph.addStream(output.connectReader())
+    return output
+  }
+}

packages/d2mini/src/operators/reduce.ts

@@ -75,37 +75,50 @@ export class ReduceOperator<K, V1, V2> extends UnaryOperator<[K, V1], [K, V2]> {
         }
       }
 
-      // First, emit removals for old values that are no longer present or have changed
+      const commonKeys = new Set<string>()
+
+      // First, emit removals for old values that are no longer present
       for (const [valueKey, { value, multiplicity }] of oldOutputMap) {
         const newEntry = newOutputMap.get(valueKey)
-        if (!newEntry || newEntry.multiplicity !== multiplicity) {
+        if (!newEntry) {
           // Remove the old value entirely
           result.push([[key, value], -multiplicity])
           this.#indexOut.addValue(key, [value, -multiplicity])
+        } else {
+          commonKeys.add(valueKey)
         }
       }
 
-      // Then, emit additions for new values that are not present in old or have changed
+      // Then, emit additions for new values that are not present in old
       for (const [valueKey, { value, multiplicity }] of newOutputMap) {
         const oldEntry = oldOutputMap.get(valueKey)
-        if (!oldEntry || oldEntry.multiplicity !== multiplicity) {
+        if (!oldEntry) {
           // Add the new value only if it has non-zero multiplicity
           if (multiplicity !== 0) {
             result.push([[key, value], multiplicity])
             this.#indexOut.addValue(key, [value, multiplicity])
           }
+        } else {
+          commonKeys.add(valueKey)
+        }
+      }
+
+      // Then, emit multiplicity changes for values that were present and are still present
+      for (const valueKey of commonKeys) {
+        const newEntry = newOutputMap.get(valueKey)
+        const oldEntry = oldOutputMap.get(valueKey)
+        const delta = newEntry!.multiplicity - oldEntry!.multiplicity
+        // Only emit actual changes, i.e. non-zero deltas
+        if (delta !== 0) {
+          result.push([[key, newEntry!.value], delta])
+          this.#indexOut.addValue(key, [newEntry!.value, delta])
         }
       }
     }
 
     if (result.length > 0) {
       this.output.sendData(new MultiSet(result))
     }
-
-    // Compact both indexes to consolidate values and remove zero-multiplicity entries
-    // Only compact changed keys for efficiency
-    this.#index.compact()
-    this.#indexOut.compact()
   }
 }
 

packages/d2mini/src/operators/topKWithFractionalIndex.ts

@@ -293,11 +293,6 @@ export class TopKWithFractionalIndexOperator<K, V1> extends UnaryOperator<
     if (result.length > 0) {
       this.output.sendData(new MultiSet(result))
     }
-
-    // Compact both indexes to consolidate values and remove zero-multiplicity entries
-    // Only compact changed keys for efficiency
-    this.#index.compact()
-    this.#indexOut.compact()
   }
 }