From f5f440757c25c58387db075ee4513528f9a7274b Mon Sep 17 00:00:00 2001 From: Konstantin Wohlwend Date: Fri, 3 Jul 2026 15:46:14 -0700 Subject: [PATCH] Left join table no longer has quadratic behavior --- .../src/databases/bulldozer/index.ts | 8 +- .../augmented-tree-map.test.ts | 76 +++++++++++++++++++ .../data-structures/augmented-tree-map.ts | 43 +++++++++-- .../bulldozer-js/src/payments/schema/index.ts | 2 +- 4 files changed, 120 insertions(+), 9 deletions(-) diff --git a/apps/bulldozer-js/src/databases/bulldozer/index.ts b/apps/bulldozer-js/src/databases/bulldozer/index.ts index fb26338f5..776caeca9 100644 --- a/apps/bulldozer-js/src/databases/bulldozer/index.ts +++ b/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)) { diff --git a/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.test.ts b/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.test.ts index b2f20e747..1ffcc3a8e 100644 --- a/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.test.ts +++ b/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.test.ts @@ -162,6 +162,34 @@ function withHeapCounters(tree: AugmentedTreeMap, arity }; } +function withMultiMapHeapCounters(tree: AugmentedTreeMultiMap, multiMapOptions: ConstructorParameters>[0]) { + let gets = 0; + const seen = new WeakMap(); + + 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({ ...serialized, tree: { ...serialized.tree, root: wrapChild(serialized.tree.root) } }, multiMapOptions), + reset: () => { gets = 0; }, + gets: () => gets, + }; +} + async function collectRefs(tree: AugmentedTreeMap) { const refs = new Set(); 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(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)); diff --git a/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts b/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts index dd37a2659..6177fb0da 100644 --- a/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts +++ b/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts @@ -105,6 +105,15 @@ export class AugmentedTreeMultiMap { + 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> = {}): AsyncIterable<[MultiKey, Value]> { let yielded = 0; + const lowerBound = this.tighterLowerBound(options.gte, options.gt); + const upperBound = this.tighterUpperBound(options.lte, options.lt); const isInRange = (key: MultiKey) => (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= 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, StoredValue>[], key: MultiKey) { + const low = this.lowerBoundIndex(entries, key); + return { index: low, found: low < entries.length && this.compareKeys(entries[low][0], key) === 0 }; + } + + private lowerBoundIndex(entries: Entry, StoredValue>[], key: MultiKey) { 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 | undefined, b: MultiKey | undefined) { + if (a === undefined) return b; + if (b === undefined) return a; + return this.compareKeys(a, b) >= 0 ? a : b; + } + + private tighterUpperBound(a: MultiKey | undefined, b: MultiKey | 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, Augmentation>, split?: Split, Value, Augmentation> }) { diff --git a/apps/bulldozer-js/src/payments/schema/index.ts b/apps/bulldozer-js/src/payments/schema/index.ts index 8b99c6313..87231cac0 100644 --- a/apps/bulldozer-js/src/payments/schema/index.ts +++ b/apps/bulldozer-js/src/payments/schema/index.ts @@ -668,7 +668,7 @@ export function createPaymentsSchema() { const subscription = rowObject(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 => {