Fix memory leak

.changeset/empty-poets-wait.md

@@ -0,0 +1,5 @@
+---
+"@tanstack/db-ivm": patch
+---
+
+Fix memory leak that results in linear memory growth with incremental changes over time. Thanks to @sorenbs for finding and fixing this.

packages/db-ivm/src/d2.ts

@@ -8,7 +8,6 @@ import type { MultiSet, MultiSetArray } from "./multiset.js"
 import type { ID2, IStreamBuilder, PipedOperator } from "./types.js"
 
 export class D2 implements ID2 {
-  #streams: Array<DifferenceStreamReader<any>> = []
   #operators: Array<UnaryOperator<any> | BinaryOperator<any>> = []
   #nextOperatorId = 0
   #finalized = false
@@ -31,7 +30,6 @@ export class D2 implements ID2 {
     const writer = new DifferenceStreamWriter<T>()
     // Use the root stream builder that exposes the sendData and sendFrontier methods
     const streamBuilder = new RootStreamBuilder<T>(this, writer)
-    this.#streams.push(streamBuilder.connectReader())
     return streamBuilder
   }
 
@@ -40,11 +38,6 @@ export class D2 implements ID2 {
     this.#operators.push(operator)
   }
 
-  addStream(stream: DifferenceStreamReader<any>): void {
-    this.#checkNotFinalized()
-    this.#streams.push(stream)
-  }
-
   finalize() {
     this.#checkNotFinalized()
     this.#finalized = true

packages/db-ivm/src/operators/concat.ts

@@ -39,7 +39,6 @@ export function concat<T, T2>(
       output.writer
     )
     stream.graph.addOperator(operator)
-    stream.graph.addStream(output.connectReader())
     return output
   }
 }

packages/db-ivm/src/operators/consolidate.ts

@@ -44,7 +44,6 @@ export function consolidate<T>(): PipedOperator<T, T> {
       output.writer
     )
     stream.graph.addOperator(operator)
-    stream.graph.addStream(output.connectReader())
     return output
   }
 }

packages/db-ivm/src/operators/count.ts

@@ -46,7 +46,6 @@ export function count<
       output.writer
     )
     stream.graph.addOperator(operator)
-    stream.graph.addStream(output.connectReader())
     return output
   }
 }

packages/db-ivm/src/operators/debug.ts

@@ -52,7 +52,6 @@ export function debug<T>(
       indent
     )
     stream.graph.addOperator(operator)
-    stream.graph.addStream(output.connectReader())
     return output
   }
 }

packages/db-ivm/src/operators/distinct.ts

@@ -92,7 +92,6 @@ export function distinct<T>(by: (value: T) => any = (value: T) => value) {
       by
     )
     stream.graph.addOperator(operator)
-    stream.graph.addStream(output.connectReader())
     return output
   }
 }

packages/db-ivm/src/operators/filter.ts

@@ -42,7 +42,6 @@ export function filter<T>(f: (data: T) => boolean): PipedOperator<T, T> {
       f
     )
     stream.graph.addOperator(operator)
-    stream.graph.addStream(output.connectReader())
     return output
   }
 }

packages/db-ivm/src/operators/join.ts

@@ -149,7 +149,6 @@ export function innerJoin<
       output.writer
     )
     stream.graph.addOperator(operator)
-    stream.graph.addStream(output.connectReader())
     return output
   }
 }

packages/db-ivm/src/operators/map.ts

@@ -42,7 +42,6 @@ export function map<T, O>(f: (data: T) => O): PipedOperator<T, O> {
       f
     )
     stream.graph.addOperator(operator)
-    stream.graph.addStream(output.connectReader())
     return output
   }
 }

packages/db-ivm/src/operators/negate.ts

@@ -27,7 +27,6 @@ export function negate<T>(): PipedOperator<T, T> {
       output.writer
     )
     stream.graph.addOperator(operator)
-    stream.graph.addStream(output.connectReader())
     return output
   }
 }

packages/db-ivm/src/operators/output.ts

@@ -47,7 +47,6 @@ export function output<T>(
       fn
     )
     stream.graph.addOperator(operator)
-    stream.graph.addStream(outputStream.connectReader())
     return outputStream
   }
 }

packages/db-ivm/src/operators/reduce.ts

@@ -118,7 +118,6 @@ export function reduce<
       f
     )
     stream.graph.addOperator(operator)
-    stream.graph.addStream(output.connectReader())
     return output
   }
 }

packages/db-ivm/src/operators/tap.ts

@@ -45,7 +45,6 @@ export function tap<T>(f: (data: MultiSet<T>) => void): PipedOperator<T, T> {
       f
     )
     stream.graph.addOperator(operator)
-    stream.graph.addStream(output.connectReader())
     return output
   }
 }

packages/db-ivm/src/operators/topKWithFractionalIndex.ts

@@ -327,7 +327,6 @@ export function topKWithFractionalIndex<KType, T>(
       opts
     )
     stream.graph.addOperator(operator)
-    stream.graph.addStream(output.connectReader())
     return output
   }
 }

packages/db-ivm/src/operators/topKWithFractionalIndexBTree.ts

@@ -302,7 +302,6 @@ export function topKWithFractionalIndexBTree<KType, T>(
       opts
     )
     stream.graph.addOperator(operator)
-    stream.graph.addStream(output.connectReader())
     return output
   }
 }

packages/db-ivm/src/types.ts

@@ -27,7 +27,6 @@ export interface ID2 {
   getNextOperatorId: () => number
   newInput: <T>() => IStreamBuilder<T>
   addOperator: (operator: UnaryOperator<any> | BinaryOperator<any>) => void
-  addStream: (stream: DifferenceStreamReader<any>) => void
   finalize: () => void
   step: () => void
 }