Left join table no longer has quadratic behavior

apps/bulldozer-js/src/databases/bulldozer/index.ts

@@ -2368,7 +2368,11 @@ export function declareLeftJoinTable(options: {
       const addRight = async (row: InputRow) => {
         const right: StoredRow = { ...row, joinKey: await options.rightJoinKeyExtractor(row) };
         const lefts = await leftMatches(right.joinKey);
-        if (!(await state.rightByJoinKey.getAll(right.joinKey)).length) for (const left of lefts) await deleteOutput(left, null);
+        // hasAny (O(depth)) instead of getAll().length: many rows can share one join key (e.g.
+        // a null-ish key on most rows), and materializing all of them per inserted row makes
+        // bulk ingestion O(n^2). The check is only needed when there are matching lefts whose
+        // left-with-null output row must be retracted.
+        if (lefts.length > 0 && !(await state.rightByJoinKey.hasAny(right.joinKey))) for (const left of lefts) await deleteOutput(left, null);
         state = { ...state, rightRows: await state.rightRows.set(right.rowIdentifier, right), rightByJoinKey: await state.rightByJoinKey.add(right.joinKey, right.rowIdentifier, right.rowIdentifier) };
         for (const left of lefts) await addOutput(left, right);
       };
@@ -2378,7 +2382,7 @@ export function declareLeftJoinTable(options: {
         const lefts = await leftMatches(right.joinKey);
         for (const left of lefts) await deleteOutput(left, right);
         state = { ...state, rightRows: await state.rightRows.delete(right.rowIdentifier), rightByJoinKey: await state.rightByJoinKey.delete(right.joinKey, right.rowIdentifier) };
-        if (!(await state.rightByJoinKey.getAll(right.joinKey)).length) for (const left of lefts) await addOutput(left, null);
+        if (lefts.length > 0 && !(await state.rightByJoinKey.hasAny(right.joinKey))) for (const left of lefts) await addOutput(left, null);
       };
 
       for (const row of changedRowsFromTableChanges(changes.left)) {

apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.test.ts

@@ -162,6 +162,34 @@ function withHeapCounters(tree: AugmentedTreeMap<number, number, number>, arity
   };
 }
 
+function withMultiMapHeapCounters(tree: AugmentedTreeMultiMap<number, number, number, string>, multiMapOptions: ConstructorParameters<typeof AugmentedTreeMultiMap<number, number, number, string>>[0]) {
+  let gets = 0;
+  const seen = new WeakMap<PiledriverHeapObject, PiledriverHeapObject>();
+
+  const wrapRef = (ref: PiledriverHeapObject): PiledriverHeapObject => {
+    const cached = seen.get(ref);
+    if (cached) return cached;
+    const wrapped = {
+      async get() {
+        gets++;
+        const node: any = await ref.get();
+        return isNode(node) ? { ...node, children: node.children.map(wrapChild) } : node;
+      },
+      [isPiledriverHeapObjectSymbol]: true as const,
+    } as PiledriverHeapObject;
+    seen.set(ref, wrapped);
+    return wrapped;
+  };
+  const wrapChild = (child: any) => child ? { ...child, ref: wrapRef(child.ref) } : child;
+  const serialized = tree.toPiledriverObject() as { type: string, tree: { type: string, root: any } };
+
+  return {
+    tree: AugmentedTreeMultiMap.fromPiledriverObject<number, number, number, string>({ ...serialized, tree: { ...serialized.tree, root: wrapChild(serialized.tree.root) } }, multiMapOptions),
+    reset: () => { gets = 0; },
+    gets: () => gets,
+  };
+}
+
 async function collectRefs(tree: AugmentedTreeMap<number, number, number>) {
   const refs = new Set<PiledriverHeapObject>();
   const visit = async (child: Serialized["root"]) => {
@@ -295,6 +323,54 @@ describe("AugmentedTreeMap", () => {
     expect(await arrayFrom(tree.entries())).toEqual([[10, "b", 2], [11, "a", 4]]);
   });
 
+  it("answers hasAny without materializing all entries for a key", async () => {
+    const multiMapOptions = {
+      ...options(8),
+      entryIdComparator: stringCompare,
+    };
+    let tree = new AugmentedTreeMultiMap(multiMapOptions);
+
+    expect(await tree.hasAny(10)).toBe(false);
+    for (let i = 0; i < 1000; i++) tree = await tree.add(10, `id-${i}`, i);
+    tree = await tree.add(11, "a", 1);
+
+    expect(await tree.hasAny(10)).toBe(true);
+    expect(await tree.hasAny(11)).toBe(true);
+    expect(await tree.hasAny(12)).toBe(false);
+
+    // hasAny must stay O(depth) even when many entries share the key; getAll here would load
+    // all 1000 entries (hundreds of node reads).
+    const counted = withMultiMapHeapCounters(tree, multiMapOptions);
+    counted.reset();
+    await counted.tree.hasAny(10);
+    expect(counted.gets()).toBeLessThanOrEqual(8);
+  });
+
+  it("uses binary search for bounded range scans inside large nodes", async () => {
+    let comparisons = 0;
+    const countedOptions = {
+      arity: 2048,
+      comparator: (a: number, b: number) => {
+        comparisons++;
+        return a - b;
+      },
+      initialAugmentation: 0,
+      extractAugmentation: (value: number) => value,
+      mergeAugmentations: (...values: number[]) => values.reduce((sum, value) => sum + value, 0),
+      entryIdComparator: stringCompare,
+    };
+    let tree = new AugmentedTreeMultiMap<number, number, number, string>(countedOptions);
+    for (let i = 0; i < 1000; i++) tree = await tree.add(i, "", i);
+
+    comparisons = 0;
+    expect(await arrayFrom(tree.entries({ gte: 990, limit: 1 }))).toEqual([[990, "", 990]]);
+    expect(comparisons).toBeLessThan(80);
+
+    comparisons = 0;
+    expect(await arrayFrom(tree.entries({ lte: 10, reverse: true, limit: 1 }))).toEqual([[10, "", 10]]);
+    expect(comparisons).toBeLessThan(80);
+  });
+
   it("keeps heap reads logarithmic for point operations", async () => {
     for (const size of [100, 1_000, 10_000]) {
       const counted = withHeapCounters(await build(size));

apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts

@@ -105,6 +105,15 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
     return result;
   }
 
+  // O(depth) existence check. Callers that only need to know whether *any* entry exists for a
+  // key must use this instead of `getAll(key).length`: getAll materializes every entry, which
+  // is O(n) when many entries share one key (e.g. null-ish join keys) and turns bulk ingestion
+  // into O(n^2).
+  async hasAny(key: Key): Promise<boolean> {
+    for await (const _entry of this.entries({ gte: key, lte: key, limit: 1 })) return true;
+    return false;
+  }
+
   async add(key: Key, entryId: EntryId, value: Value) {
     return await this.insertRaw({ key, id: entryId }, value);
   }
@@ -197,6 +206,8 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
 
   private async *rawEntries(options: EntryOptions<MultiKey<Key, EntryId>> = {}): AsyncIterable<[MultiKey<Key, EntryId>, Value]> {
     let yielded = 0;
+    const lowerBound = this.tighterLowerBound(options.gte, options.gt);
+    const upperBound = this.tighterUpperBound(options.lte, options.lt);
     const isInRange = (key: MultiKey<Key, EntryId>) =>
       (options.gte === undefined || this.compareKeys(key, options.gte) >= 0)
       && (options.gt === undefined || this.compareKeys(key, options.gt) > 0)
@@ -211,18 +222,21 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
         if (isInRange(entry[0]) && (options.limit === undefined || yielded++ < options.limit)) yield [entry[0], await tree.loadValue(entry[1])];
       };
 
+      const startIndex = lowerBound === undefined ? 0 : tree.lowerBoundIndex(node.entries, lowerBound);
+      const endIndex = upperBound === undefined ? node.entries.length : tree.lowerBoundIndex(node.entries, upperBound);
+
       if (options.reverse) {
-        for (let i = node.entries.length - 1; i >= 0; i--) {
-          yield* walk(tree, node.children[i + 1] ?? null);
+        yield* walk(tree, node.children[endIndex] ?? null);
+        for (let i = endIndex - 1; i >= startIndex; i--) {
           yield* visitEntry(node.entries[i]);
+          yield* walk(tree, node.children[i] ?? null);
         }
-        yield* walk(tree, node.children[0] ?? null);
       } else {
-        for (let i = 0; i < node.entries.length; i++) {
+        for (let i = startIndex; i < endIndex; i++) {
           yield* walk(tree, node.children[i] ?? null);
           yield* visitEntry(node.entries[i]);
         }
-        yield* walk(tree, node.children[node.entries.length] ?? null);
+        yield* walk(tree, node.children[endIndex] ?? null);
       }
     };
 
@@ -333,13 +347,30 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
   }
 
   private search(entries: Entry<MultiKey<Key, EntryId>, StoredValue<Value>>[], key: MultiKey<Key, EntryId>) {
+    const low = this.lowerBoundIndex(entries, key);
+    return { index: low, found: low < entries.length && this.compareKeys(entries[low][0], key) === 0 };
+  }
+
+  private lowerBoundIndex(entries: Entry<MultiKey<Key, EntryId>, StoredValue<Value>>[], key: MultiKey<Key, EntryId>) {
     let low = 0, high = entries.length;
     while (low < high) {
       const mid = (low + high) >> 1;
       if (this.compareKeys(entries[mid][0], key) < 0) low = mid + 1;
       else high = mid;
     }
-    return { index: low, found: low < entries.length && this.compareKeys(entries[low][0], key) === 0 };
+    return low;
+  }
+
+  private tighterLowerBound(a: MultiKey<Key, EntryId> | undefined, b: MultiKey<Key, EntryId> | undefined) {
+    if (a === undefined) return b;
+    if (b === undefined) return a;
+    return this.compareKeys(a, b) >= 0 ? a : b;
+  }
+
+  private tighterUpperBound(a: MultiKey<Key, EntryId> | undefined, b: MultiKey<Key, EntryId> | undefined) {
+    if (a === undefined) return b;
+    if (b === undefined) return a;
+    return this.compareKeys(a, b) <= 0 ? a : b;
   }
 
   private async finishUpsert(result: { root: Child<MultiKey<Key, EntryId>, Augmentation>, split?: Split<MultiKey<Key, EntryId>, Value, Augmentation> }) {

apps/bulldozer-js/src/payments/schema/index.ts

@@ -668,7 +668,7 @@ export function createPaymentsSchema() {
         const subscription = rowObject<SubscriptionRow>(row.rowData);
         return { tenancyId: subscription.tenancyId, stripeSubscriptionId: subscription.stripeSubscriptionId };
       },
-      joinKeyComparator: compareJson,
+      joinKeyComparator: (a: any, b: any) => stringCompare(a.tenancyId, b.tenancyId) || stringCompare(a.stripeSubscriptionId ?? "", b.stripeSubscriptionId ?? "") || (a.stripeSubscriptionId === b.stripeSubscriptionId ? 0 : a.stripeSubscriptionId === null ? -1 : 1),
       joiner: async (left, right) => ({ leftRowData: left.rowData, rightRowData: right?.rowData ?? null }),
     }), { left: "payments-subscription-invoices", right: "payments-subscriptions" }),
     table("payments-renewal-invoice-rows", defineFilterTable(row => {