diff --git a/.changeset/green-indexes-rest.md b/.changeset/green-indexes-rest.md new file mode 100644 index 0000000..62a6a64 --- /dev/null +++ b/.changeset/green-indexes-rest.md @@ -0,0 +1,11 @@ +--- +"@will-be-done/hyperdb": minor +--- + +Add `PreloadedHybridDB`, which preloads every declared index with ID-only leaves +and batch-hydrates missing entity rows through the built-in `byId` `uniqhash`. +Secondary `uniqhash` indexes are preloaded as value-to-ID pointers, while shared +hash-index transactions provide copy-on-write commit and rollback behavior. +Add `externalStorageMergeTrait` for changesets already persisted by another +runtime sharing the primary. Their normal merge operations update the preloaded +snapshot, notify subscribers, and persist external inserts idempotently. diff --git a/README.md b/README.md index 70d2b2c..a7723dd 100644 --- a/README.md +++ b/README.md @@ -37,6 +37,9 @@ to strain: possible and load missing ranges from the primary store on demand. Writes update the cache first so the UI can respond immediately, then flush to the primary store in order. +- **Preloaded indexes without preloaded rows.** `PreloadedHybridDB` loads every + declared index into memory as key-to-id entries at startup, then batch-loads + only the entity rows selected by a scan and caches them by id. - **Run the same logic on the backend.** Because a table index is just a B-tree, the same schema, selectors, and actions run against a persistent store on the server (SQLite today, pg/mongodb in future). The runtime reads only the rows a @@ -176,6 +179,20 @@ export async function createAppDB() { `AsyncSqlDriver` is also exercised against Turso Database's browser WASM engine in the shared driver conformance suite. +If all index keys fit in memory but all entity rows do not, use +`new SubscribableDB(new PreloadedHybridDB(primary))` instead. Its `loadTables` +call automatically preloads every index on every loaded table; scans resolve +ordered IDs in memory, including non-ID `uniqhash` value-to-ID pointers, and +batch-fetch only unresolved rows through the built-in `byId` entity index. No +explicit `preloadTables` call or extra B-tree `byIds` index is needed. +Repeated `loadTables` calls are incremental: previously loaded tables remain +available while supplied table definitions are added or refreshed. +Exact `byId` misses reconcile rows added by another connected runtime. +Apply changesets already persisted by another runtime sharing the primary with +`externalStorageMergeTrait`. Their normal merge operations update the preloaded +snapshot and invalidate subscribers, while external inserts persist +idempotently instead of failing as duplicates. + If your whole app state can be loaded into memory at startup, you may not need `HybridDB`. A plain `new SubscribableDB(new DB(new BptreeInmemDriver()))` keeps reads and writes synchronous, so you can use `useSyncSelector`, `useSyncDispatch`, @@ -256,15 +273,15 @@ internally, so the same async subscription behavior is available without React. ## Entry points -| Import path | Contents | -| ---------------------------------------- | ------------------------------------------------------------------------------------ | -| `@will-be-done/hyperdb` | Core: `defineTable`, `v`, `selectFrom`, builders, `DB`, `HybridDB`, `SubscribableDB` | -| `@will-be-done/hyperdb/react` | React hooks and `DBProvider` | -| `@will-be-done/hyperdb/tracing` | Tracing store and tracer configuration | -| `@will-be-done/hyperdb/drivers/inmemory` | `BptreeInmemDriver` | -| `@will-be-done/hyperdb/drivers/sqlite` | `SqlDriver`, `AsyncSqlDriver` | -| `@will-be-done/hyperdb/drivers/idb` | `openIndexedDBDriver`, `IdbDriver` | -| `@will-be-done/hyperdb-devtool/react` | `HyperDBDevtools`, `HyperDBDevtoolsPanel` (separate package) | +| Import path | Contents | +| ---------------------------------------- | --------------------------------------------------------------------------------------------------------- | +| `@will-be-done/hyperdb` | Core: `defineTable`, `v`, `selectFrom`, builders, `DB`, `HybridDB`, `PreloadedHybridDB`, `SubscribableDB` | +| `@will-be-done/hyperdb/react` | React hooks and `DBProvider` | +| `@will-be-done/hyperdb/tracing` | Tracing store and tracer configuration | +| `@will-be-done/hyperdb/drivers/inmemory` | `BptreeInmemDriver` | +| `@will-be-done/hyperdb/drivers/sqlite` | `SqlDriver`, `AsyncSqlDriver` | +| `@will-be-done/hyperdb/drivers/idb` | `openIndexedDBDriver`, `IdbDriver` | +| `@will-be-done/hyperdb-devtool/react` | `HyperDBDevtools`, `HyperDBDevtoolsPanel` (separate package) | ## Learn more diff --git a/packages/hyperdb-doc/src/content/docs/runtime/db.md b/packages/hyperdb-doc/src/content/docs/runtime/db.md index 611bdda..7de0057 100644 --- a/packages/hyperdb-doc/src/content/docs/runtime/db.md +++ b/packages/hyperdb-doc/src/content/docs/runtime/db.md @@ -1,6 +1,6 @@ --- title: The DB Runtime -description: DB, SubscribableDB, HybridDB, transactions, lifecycle hooks, and traits. +description: DB, SubscribableDB, HybridDB, PreloadedHybridDB, transactions, lifecycle hooks, and traits. sidebar: order: 1 --- @@ -10,12 +10,13 @@ commands. ## Which runtime should I use? -| Runtime shape | Use when | Tradeoff | -| ------------------------------- | ------------------------------------------------------------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `DB` | You only need `selectSync`, `insert`, `upsert`, `deleteRows`, and transactions. | Lowest overhead. No subscriptions, reactive selector cache, revisions, or lifecycle hooks. | -| `SubscribableDB` + sync driver | Your reactive app can load its working state into memory. | Best interactive path: selectors and actions can stay synchronous with `useSyncSelector`, `useSyncDispatch`, `selectSync`, and `syncDispatch`. | -| `SubscribableDB` + async driver | Your reactive app should keep memory low and read directly from IndexedDB or async SQLite. | Uses async selectors/actions. Simpler than `HybridDB`, but every read follows the async driver path. | -| `SubscribableDB` + `HybridDB` | Local-first browser apps that want persistent storage plus fast reads for hot data. | Uses async APIs, but reads check the in-memory cache first. Missing index ranges fall through to the primary store, then get cached for next time. Writes update the cache first for immediate UI response, then flush to the primary store. | +| Runtime shape | Use when | Tradeoff | +| -------------------------------------- | ------------------------------------------------------------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `DB` | You only need `selectSync`, `insert`, `upsert`, `deleteRows`, and transactions. | Lowest overhead. No subscriptions, reactive selector cache, revisions, or lifecycle hooks. | +| `SubscribableDB` + sync driver | Your reactive app can load its working state into memory. | Best interactive path: selectors and actions can stay synchronous with `useSyncSelector`, `useSyncDispatch`, `selectSync`, and `syncDispatch`. | +| `SubscribableDB` + async driver | Your reactive app should keep memory low and read directly from IndexedDB or async SQLite. | Uses async selectors/actions. Simpler than `HybridDB`, but every read follows the async driver path. | +| `SubscribableDB` + `HybridDB` | Local-first browser apps that want persistent storage plus fast reads for hot data. | Uses async APIs, but reads check the in-memory cache first. Missing index ranges fall through to the primary store, then get cached for next time. Writes update the cache first for immediate UI response, then flush to the primary store. | +| `SubscribableDB` + `PreloadedHybridDB` | Index keys fit in memory, but retaining every entity row would be too expensive. | `loadTables` eagerly reads all tables once to build ID-only indexes. Scans never need persistent index work, but the first access to an entity row is asynchronous. Writes are persisted before the in-memory indexes are published. | ## `DB` @@ -236,6 +237,78 @@ when a delete did not know the old row. That lets exact unique reads return from memory while broader uncached scans still wait when a pending write could affect their interval. +## `PreloadedHybridDB` + +`PreloadedHybridDB` is the middle ground between range-cached `HybridDB` and a +fully resident `BptreeInmemDriver`. On `loadTables`, it preloads every declared +index on every loaded table. Each B-tree or hash leaf stores only the entity ID; +the startup rows used to build those keys are then released. + +```ts +import { + DB, + PreloadedHybridDB, + SubscribableDB, + externalStorageMergeTrait, + execAsync, +} from "@will-be-done/hyperdb"; +import { openIndexedDBDriver } from "@will-be-done/hyperdb/drivers/idb"; + +const primary = new DB(await openIndexedDBDriver("my-app")); +const db = new SubscribableDB(new PreloadedHybridDB(primary)); + +// Automatically preloads all indexes on both tables, including byId. +await execAsync(db.loadTables([tasksTable, projectsTable])); +``` + +`loadTables` is incremental. A later call keeps previously loaded tables and +adds or refreshes only the table definitions passed to that call. + +A scan first reads its bounds from the in-memory ID-only index. This includes +non-ID `uniqhash` indexes, which are preloaded as unique value-to-ID pointers. +The built-in `byId` `uniqhash` is the canonical entity store: it begins with +unresolved ID entries, loads missing rows from the primary in batches, and +replaces those entries with hydrated rows. The result is returned in the order +produced by the scanned index. Repeating the scan—or reaching the same entities +through another index—reuses the hydrated `byId` entries. + +An exact `byId` miss checks the primary before returning no row. This lets an +independently connected runtime, such as another browser tab, add a row after +the preload snapshot; discovering that row also adds its secondary index +pointers locally. + +When merging a changeset that another runtime has already committed to the same +primary storage, add `externalStorageMergeTrait` alongside the trait that +suppresses local change tracking: + +```ts +await asyncDispatch( + db.withTraits({ type: "skip-sync" }, externalStorageMergeTrait), + mergeChanges(args), +); +``` + +In this mode the merge compares against the preloaded snapshot. A row absent +from that snapshot follows the normal insert path, while persistence uses an +idempotent upsert because the external writer may already have stored it. The +committed insert/upsert/delete operations therefore update every preloaded +index and notify `SubscribableDB` subscribers normally. Use this trait only for +already-persisted external changesets; ordinary inserts retain duplicate-ID +checking. + +`preloadTables` is unnecessary in this mode because `loadTables` always covers +all indexes, and tables do not need an extra B-tree full-scan index. Inserts, +upserts, deletes, and transactions update both the durable primary and the +ID-only indexes. Unlike `HybridDB`'s optimistic write path, these writes wait for +the primary operation before publishing the new in-memory index state. +Transactional hash-index changes use copy-on-write buckets and are published or +discarded together with the transaction. + +Use this runtime when index keys are substantially smaller than complete rows +and startup can afford one bulk read of each table. Use regular `HybridDB` when +startup should touch only queried ranges, or a plain in-memory driver when all +rows comfortably fit in memory and reads must remain synchronous. + ## Executing commands Selectors and actions are generators. The dispatch and select helpers run them diff --git a/packages/hyperdb-doc/src/content/docs/runtime/drivers.md b/packages/hyperdb-doc/src/content/docs/runtime/drivers.md index f1c3e39..cb173e9 100644 --- a/packages/hyperdb-doc/src/content/docs/runtime/drivers.md +++ b/packages/hyperdb-doc/src/content/docs/runtime/drivers.md @@ -8,8 +8,10 @@ sidebar: A driver is the actual storage backend behind a `DB`. The same selectors and actions run unchanged against any driver, and in any environment. You can use a single driver directly, or combine a persistent primary driver with an in-memory -cache through [`HybridDB`](/runtime/db/#hybriddb). You also choose whether to use -the sync or async runtime helpers, which depends on the storage path. +cache through [`HybridDB`](/runtime/db/#hybriddb), or preload ID-only indexes +through [`PreloadedHybridDB`](/runtime/db/#preloadedhybriddb). You also choose +whether to use the sync or async runtime helpers, which depends on the storage +path. ## Choosing a driver @@ -21,9 +23,9 @@ the sync or async runtime helpers, which depends on the storage path. | `AsyncSqlDriver` | `.../drivers/sqlite` | async | both | Async SQLite, including Turso WASM as a `HybridDB` primary | Sync drivers work with `execSync` / `syncDispatch` / `selectSync`. Async drivers -require `execAsync` / `asyncDispatch` / `selectAsync`. `HybridDB` also uses the -async helpers, because a read may miss the memory cache and fall through to the -primary store. +require `execAsync` / `asyncDispatch` / `selectAsync`. `HybridDB` and +`PreloadedHybridDB` also use the async helpers, because a read may need entity +rows from the primary store. A typical local-first browser setup uses `HybridDB` with IndexedDB or async SQLite as the primary store and `BptreeInmemDriver` as the cache. If your whole @@ -31,6 +33,10 @@ working set can be loaded eagerly, a plain `SubscribableDB` over `BptreeInmemDriver` keeps the UI path fully synchronous. On the server, use a native `SqlDriver` while running the _same_ schema, selectors, and actions. +All built-in drivers support the bulk table read used by +`PreloadedHybridDB.loadTables`. This path does not require a user-declared +full-scan index. + ## In-memory The simplest driver: a set of in-memory B+trees. Construct it with no arguments. diff --git a/packages/hyperdb-doc/src/content/docs/start/introduction.md b/packages/hyperdb-doc/src/content/docs/start/introduction.md index 30c8cf1..8157369 100644 --- a/packages/hyperdb-doc/src/content/docs/start/introduction.md +++ b/packages/hyperdb-doc/src/content/docs/start/introduction.md @@ -41,6 +41,9 @@ both the client and server. cache. Reads use cached index ranges when possible and fall through to the primary store only for missing ranges. Writes update the cache first for immediate UI feedback, then flush to the primary store. +- Index-preloaded reads: `PreloadedHybridDB` keeps every declared index in + memory with ID-only leaves, then batch-loads and caches entity rows only when + scans select them. - JavaScript selectors and actions: selectors and actions are ordinary JS, with loops, conditionals, and function calls. HyperDB gives you fast indexed lookups and inserts underneath, not a query language to learn, and the same mental model on @@ -93,14 +96,14 @@ npm install react react-dom The core package ships several entry points: -| Import path | Contents | -| ---------------------------------------- | ----------------------------------------------------------------------------------------------------- | -| `@will-be-done/hyperdb` | Core: `defineTable`, `v`, `selectFrom`, builders, `DB`, `HybridDB`, `SubscribableDB`, runtime helpers | -| `@will-be-done/hyperdb/react` | React hooks and `DBProvider` | -| `@will-be-done/hyperdb/tracing` | Tracing store and tracer configuration | -| `@will-be-done/hyperdb/drivers/inmemory` | `BptreeInmemDriver` | -| `@will-be-done/hyperdb/drivers/sqlite` | `SqlDriver`, `AsyncSqlDriver` | -| `@will-be-done/hyperdb/drivers/idb` | `openIndexedDBDriver`, `IdbDriver` | +| Import path | Contents | +| ---------------------------------------- | -------------------------------------------------------------------------------------------------------------------------- | +| `@will-be-done/hyperdb` | Core: `defineTable`, `v`, `selectFrom`, builders, `DB`, `HybridDB`, `PreloadedHybridDB`, `SubscribableDB`, runtime helpers | +| `@will-be-done/hyperdb/react` | React hooks and `DBProvider` | +| `@will-be-done/hyperdb/tracing` | Tracing store and tracer configuration | +| `@will-be-done/hyperdb/drivers/inmemory` | `BptreeInmemDriver` | +| `@will-be-done/hyperdb/drivers/sqlite` | `SqlDriver`, `AsyncSqlDriver` | +| `@will-be-done/hyperdb/drivers/idb` | `openIndexedDBDriver`, `IdbDriver` | The React devtool ships as a separate package, `@will-be-done/hyperdb-devtool`, exposing `HyperDBDevtools` from `@will-be-done/hyperdb-devtool/react`. diff --git a/packages/hyperdb-doc/src/content/docs/start/llm-cheat-sheet.md b/packages/hyperdb-doc/src/content/docs/start/llm-cheat-sheet.md index e3e4fc5..86a8db0 100644 --- a/packages/hyperdb-doc/src/content/docs/start/llm-cheat-sheet.md +++ b/packages/hyperdb-doc/src/content/docs/start/llm-cheat-sheet.md @@ -55,6 +55,7 @@ and views that should re-run only when the exact index ranges they read change. import { DB, HybridDB, + PreloadedHybridDB, SubscribableDB, asyncDispatch, createAction, @@ -319,6 +320,39 @@ read, write, or transaction (including cache-only reads) throws `HybridDBCrashedError` with the persistence error as its `cause`. Recover by creating a new `HybridDB` and reloading tables. +For datasets where every index key fits in memory but every full row does not, +use `PreloadedHybridDB`: + +```ts +import { + DB, + PreloadedHybridDB, + SubscribableDB, + externalStorageMergeTrait, + execAsync, +} from "@will-be-done/hyperdb"; + +const db = new SubscribableDB(new PreloadedHybridDB(primary)); +await execAsync(db.loadTables([tasksTable, projectsTable])); +``` + +`loadTables` automatically preloads all declared indexes with entity IDs as +their leaves, including non-ID `uniqhash` value-to-ID pointers. Calls are +incremental: existing tables remain loaded while supplied definitions are added +or refreshed. A bounded scan resolves IDs from memory and dereferences them +through the built-in `byId` `uniqhash`. Unresolved `byId` entries are batch-loaded +from the primary and then reused by reads through every index. Do not add a +`byIds` B-tree or call `preloadTables` for this runtime. Reads remain async, and +writes wait for primary persistence before publishing copy-on-write index +changes. +Exact `byId` misses check the primary so rows added by another connected runtime +can be discovered and incorporated into the preloaded indexes. +For a changeset already persisted by another runtime sharing the primary, run +its merge with `externalStorageMergeTrait` and the app's change-tracking +suppression trait. The merge then updates the preloaded snapshot through normal +transaction operations, subscribers receive normal invalidations, and an +external insert is persisted idempotently rather than failing as a duplicate. + ## React Pattern ```tsx diff --git a/packages/hyperdb-doc/src/content/docs/start/why.md b/packages/hyperdb-doc/src/content/docs/start/why.md index 5ffc4b5..f884099 100644 --- a/packages/hyperdb-doc/src/content/docs/start/why.md +++ b/packages/hyperdb-doc/src/content/docs/start/why.md @@ -125,6 +125,13 @@ can fall through to storage, while startup stays quick and memory stays low. Writes update the cache first for an immediate UI response, then flush to the primary store in order. +`PreloadedHybridDB` offers a different memory/startup tradeoff. It reads each +table once at startup to build every index with entity IDs as leaves, but does +not retain the rows themselves. Scans therefore resolve bounds entirely in +memory and batch-load only entity IDs that have not already been cached. This is +useful when the index projection fits in memory while the complete dataset does +not. + ## Composable app logic HyperDB selectors and actions compose like ordinary code. A selector can call diff --git a/packages/hyperdb/README.md b/packages/hyperdb/README.md index eeb6260..de868bb 100644 --- a/packages/hyperdb/README.md +++ b/packages/hyperdb/README.md @@ -33,6 +33,9 @@ to strain: possible and load missing ranges from the primary store on demand. Writes update the cache first so the UI can respond immediately, then flush to the primary store in order. +- **Preloaded indexes without preloaded rows.** `PreloadedHybridDB` loads every + declared index into memory as key-to-id entries at startup, then batch-loads + only the entity rows selected by a scan and caches them by id. - **Run the same logic on the backend.** Because a table index is just a B-tree, the same schema, selectors, and actions run against a persistent store on the server (SQLite today, pg/mongodb in future). The runtime reads only the rows a @@ -174,6 +177,20 @@ export async function createAppDB() { for successful index scans. `HybridDB` keeps the persistent store durable while serving cached index ranges from memory. +If all index keys fit in memory but all entity rows do not, wrap +`new PreloadedHybridDB(primary)` instead of `HybridDB`. `loadTables` +automatically preloads every declared index as key-to-id entries. A scan first +resolves ordered IDs in memory, including through non-ID `uniqhash` pointers. +It then uses the built-in `byId` `uniqhash` as the canonical entity cache and +batch-loads only unresolved IDs. This mode does not need an explicit +`preloadTables` call or a separate B-tree `byIds` index. +Exact `byId` misses check the primary, allowing rows added by another connected +runtime to be incorporated after startup. +Apply changesets already persisted by another runtime sharing the primary with +`externalStorageMergeTrait`. Their merge updates the preloaded snapshot through +normal transaction operations and produces normal subscriber invalidations; +external inserts persist idempotently instead of failing as duplicates. + ```tsx import { DBProvider, @@ -235,15 +252,15 @@ that same promise resolves. ## Entry points -| Import path | Contents | -| ---------------------------------------- | ------------------------------------------------------------------------------------ | -| `@will-be-done/hyperdb` | Core: `defineTable`, `v`, `selectFrom`, builders, `DB`, `HybridDB`, `SubscribableDB` | -| `@will-be-done/hyperdb/react` | React hooks and `DBProvider` | -| `@will-be-done/hyperdb/tracing` | Tracing store and tracer configuration | -| `@will-be-done/hyperdb/drivers/inmemory` | `BptreeInmemDriver` | -| `@will-be-done/hyperdb/drivers/sqlite` | `SqlDriver`, `AsyncSqlDriver` | -| `@will-be-done/hyperdb/drivers/idb` | `openIndexedDBDriver`, `IdbDriver` | -| `@will-be-done/hyperdb-devtool/react` | `HyperDBDevtools`, `HyperDBDevtoolsPanel` (separate package) | +| Import path | Contents | +| ---------------------------------------- | --------------------------------------------------------------------------------------------------------- | +| `@will-be-done/hyperdb` | Core: `defineTable`, `v`, `selectFrom`, builders, `DB`, `HybridDB`, `PreloadedHybridDB`, `SubscribableDB` | +| `@will-be-done/hyperdb/react` | React hooks and `DBProvider` | +| `@will-be-done/hyperdb/tracing` | Tracing store and tracer configuration | +| `@will-be-done/hyperdb/drivers/inmemory` | `BptreeInmemDriver` | +| `@will-be-done/hyperdb/drivers/sqlite` | `SqlDriver`, `AsyncSqlDriver` | +| `@will-be-done/hyperdb/drivers/idb` | `openIndexedDBDriver`, `IdbDriver` | +| `@will-be-done/hyperdb-devtool/react` | `HyperDBDevtools`, `HyperDBDevtoolsPanel` (separate package) | ## Learn more diff --git a/packages/hyperdb/src/hyperdb/core/driver.ts b/packages/hyperdb/src/hyperdb/core/driver.ts index a21e6f7..6c38935 100644 --- a/packages/hyperdb/src/hyperdb/core/driver.ts +++ b/packages/hyperdb/src/hyperdb/core/driver.ts @@ -43,6 +43,15 @@ export type BaseDBDriverOperations = { export interface DBDriver extends BaseDBDriverOperations { loadTables(table: TableDefinition[]): Generator; + /** + * Internal bulk-read capability used by runtimes that build derived indexes + * during startup. Unlike `intervalScan`, this does not require a logical + * full-scan index on the table. + */ + scanAll?( + tableName: string, + options?: DBDriverOperationOptions, + ): Generator; beginTx( mode?: DBTransactionMode, options?: DBDriverOperationOptions, diff --git a/packages/hyperdb/src/hyperdb/drivers/idb/idb-driver.scan-all.test.ts b/packages/hyperdb/src/hyperdb/drivers/idb/idb-driver.scan-all.test.ts new file mode 100644 index 0000000..2511a5c --- /dev/null +++ b/packages/hyperdb/src/hyperdb/drivers/idb/idb-driver.scan-all.test.ts @@ -0,0 +1,48 @@ +import { describe, expect, it } from "vitest"; +import { execAsync } from "../../core/executor"; +import { DB } from "../../runtime/db"; +import { defineTable } from "../../schema/table"; +import { v } from "../../schema/values"; +import { openIndexedDBDriver } from "./idb-driver"; + +const bulkRowsTable = defineTable("idbScanAllBulkRows", { + id: v.string(), + value: v.number(), +}); + +let databaseCounter = 0; + +function deleteDatabase(databaseName: string): Promise { + return new Promise((resolve, reject) => { + const request = indexedDB.deleteDatabase(databaseName); + request.onsuccess = () => resolve(); + request.onerror = () => + reject(request.error ?? new Error("Failed to delete test database")); + request.onblocked = () => + reject(new Error("IndexedDB delete request was blocked")); + }); +} + +describe("IdbDriver scanAll", () => { + it("reads at least 1,001 decoded rows in key order", async () => { + databaseCounter += 1; + const databaseName = `hyperdb-idb-scan-all-${Date.now().toString(36)}-${databaseCounter}`; + await deleteDatabase(databaseName); + const driver = await openIndexedDBDriver(databaseName); + const db = new DB(driver); + const rows = Array.from({ length: 1_001 }, (_, index) => ({ + id: `row-${String(index).padStart(4, "0")}`, + value: index, + })); + + try { + await execAsync(db.loadTables([bulkRowsTable])); + await execAsync(db.insert(bulkRowsTable, rows)); + + await expect(execAsync(db.scanAll(bulkRowsTable))).resolves.toEqual(rows); + } finally { + driver.close(); + await deleteDatabase(databaseName); + } + }); +}); diff --git a/packages/hyperdb/src/hyperdb/drivers/idb/idb-driver.ts b/packages/hyperdb/src/hyperdb/drivers/idb/idb-driver.ts index a0dd612..3d3c76e 100644 --- a/packages/hyperdb/src/hyperdb/drivers/idb/idb-driver.ts +++ b/packages/hyperdb/src/hyperdb/drivers/idb/idb-driver.ts @@ -195,6 +195,26 @@ function requestToPromise(request: IDBRequest): Promise { }); } +function readAllWithCursor(source: IDBObjectStore): Promise { + return new Promise((resolve, reject) => { + const results: T[] = []; + const request = source.openCursor(); + + request.onsuccess = () => { + const cursor = request.result; + if (!cursor) { + resolve(results); + return; + } + + results.push(cursor.value as T); + cursor.continue(); + }; + request.onerror = () => + reject(request.error ?? new Error("IDB cursor request failed")); + }); +} + function txDone(tx: IDBTransaction): Promise { return new Promise((resolve, reject) => { tx.oncomplete = () => resolve(); @@ -1639,6 +1659,23 @@ export class IdbDriver implements DBDriver { } } + *scanAll( + tableName: string, + options: DBDriverOperationOptions = {}, + ): Generator { + this.getTableDefinition(tableName); + return yield* this.withTransaction( + "readonly", + [tableStoreName(tableName)], + async (tx) => { + const store = tx.objectStore(tableStoreName(tableName)); + const records = await readAllWithCursor(store); + return records.map(decodeStoredRecord); + }, + options, + ); + } + private async ensureSchema( tableDefinitions: TableDefinition[], ): Promise { diff --git a/packages/hyperdb/src/hyperdb/drivers/inmemory/bptree-inmem-driver.ts b/packages/hyperdb/src/hyperdb/drivers/inmemory/bptree-inmem-driver.ts index 2b016bd..d84f218 100644 --- a/packages/hyperdb/src/hyperdb/drivers/inmemory/bptree-inmem-driver.ts +++ b/packages/hyperdb/src/hyperdb/drivers/inmemory/bptree-inmem-driver.ts @@ -13,17 +13,23 @@ import { InMemoryBinaryPlusTree } from "../../structures/bptree"; import { compareStoredTuple, compareTuple } from "../../core/query/tuple"; import { convertWhereToBound } from "../../core/query/bounds"; import type { DBCmd } from "../../commands/async"; +import { + HashIndex as HashIndexStore, + HashIndexTx as HashIndexStoreTx, + hashIndexKey, + type HashIndexEntry, +} from "../../structures/hash-index"; type TableData = { tableDef: TableDefinition; indexes: Map; - idIndex: HashIndex; + idIndex: DriverHashIndex; }; type TxTableData = { tableDef: TableDefinition; indexes: Map; - idIndex: HashIndexTx; + idIndex: DriverHashIndexTx; }; type BtreeIndexDef = { @@ -86,30 +92,6 @@ const getHashIndexValue = (row: Row, column: string): Value | undefined => { return value === undefined ? null : (value as Value); }; -function bytesOfHashValue(value: ArrayBuffer | ArrayBufferView): number[] { - if (value instanceof ArrayBuffer) { - return Array.from(new Uint8Array(value)); - } - return Array.from( - new Uint8Array(value.buffer, value.byteOffset, value.byteLength), - ); -} - -function toHex(value: number): string { - return value.toString(16).padStart(2, "0"); -} - -function hashIndexKey(value: Value): HashColumnKey { - if (value === null) return "null:"; - if (typeof value === "string") return `string:${value}`; - if (typeof value === "number") { - return `number:${Object.is(value, -0) ? 0 : value}`; - } - if (typeof value === "bigint") return `bigint:${value.toString()}`; - if (typeof value === "boolean") return `boolean:${value ? "1" : "0"}`; - return `bytes:${bytesOfHashValue(value).map(toHex).join("")}`; -} - function stringifyWithBigInt(value: unknown): string { return JSON.stringify(value, (_key, item) => typeof item === "bigint" ? item.toString() : item, @@ -392,7 +374,7 @@ const getColumnValuesFromBounds = ( indexDef: HashIndexDef, tupleBounds: TupleScanOptions[], ) => { - const idxValues = new Set(); + const idxValues = new Map(); for (const bound of tupleBounds) { if ( @@ -436,24 +418,35 @@ const getColumnValuesFromBounds = ( ); } - idxValues.add(lteKey); + idxValues.set(lteKey, bound.lte[0] as Value); } - return idxValues; + return idxValues.values(); }; -type HashColumnKey = string; +function hashEntries( + indexDef: HashIndexDef, + values: readonly Row[], +): HashIndexEntry[] { + return values.flatMap((record) => { + const key = getHashIndexValue(record, indexDef.column); + return key === undefined ? [] : [{ key, id: record.id, value: record }]; + }); +} -class HashIndex implements Index { +class DriverHashIndex implements Index { get type(): "hash" | "uniqhash" { return this.indexDef.unique ? "uniqhash" : "hash"; } - indexDef: HashIndexDef; - records: Map> = new Map(); - rowKeys: Map = new Map(); + readonly indexDef: HashIndexDef; + readonly store: HashIndexStore; constructor(indexDef: HashIndexDef) { this.indexDef = indexDef; + this.store = new HashIndexStore({ + name: indexDef.name, + unique: indexDef.unique, + }); } cols(): string[] { @@ -461,112 +454,43 @@ class HashIndex implements Index { } scan(tupleBounds: TupleScanOptions[], selectOptions: SelectOptions): Row[] { - if (selectOptions.limit !== undefined && selectOptions.limit <= 0) - return []; - - const idxValues = getColumnValuesFromBounds(this.indexDef, tupleBounds); - - const results: Row[] = []; - - for (const idxValue of idxValues) { - const rows = this.records.get(idxValue); - - if (!rows) continue; - - for (const row of rows.values()) { - results.push(row); - - if ( - selectOptions.limit !== undefined && - results.length >= selectOptions.limit - ) { - return results; - } - } - } - - return results; + return this.store.scan( + getColumnValuesFromBounds(this.indexDef, tupleBounds), + selectOptions, + ); } validateInsert(values: Row[], replacingIds: Set): void { - if (!this.indexDef.unique) return; - - const batchKeys = new Map(); - for (const record of values) { - const colValue = getHashIndexValue(record, this.indexDef.column); - if (colValue === undefined) continue; - const key = hashIndexKey(colValue); - const batchId = batchKeys.get(key); - if (batchId !== undefined && batchId !== record.id) { - throw new Error( - `Unique hash index ${this.indexDef.name} already has value for record ${batchId}`, - ); - } - batchKeys.set(key, record.id); - - const existingRows = this.records.get(key); - if (!existingRows) continue; - - for (const existingId of existingRows.keys()) { - if (existingId === record.id || replacingIds.has(existingId)) continue; - throw new Error( - `Unique hash index ${this.indexDef.name} already has value for record ${existingId}`, - ); - } - } + this.store.validateInsert(hashEntries(this.indexDef, values), replacingIds); } insert(values: Row[]): void { - for (const record of values) { - const colValue = getHashIndexValue(record, this.indexDef.column); - if (colValue === undefined) continue; - const key = hashIndexKey(colValue); - this.rowKeys.set(record.id, key); - - const rows = this.records.get(key); - - if (!rows) { - const m = new Map(); - m.set(record.id, record); - this.records.set(key, m); - } else { - rows.set(record.id, record); - } - } + this.store.insert(hashEntries(this.indexDef, values)); } delete(values: Row[]): void { - for (const record of values) { - const key = this.rowKeys.get(record.id); - if (key === undefined) continue; - - const rows = this.records.get(key); - - if (!rows) continue; - - rows.delete(record.id); - this.rowKeys.delete(record.id); - } + this.store.delete(values.map((record) => record.id)); } tx(): IndexTx { - return new HashIndexTx(this); + return new DriverHashIndexTx(this, this.store.tx()); + } + + values(): Row[] { + return this.store.values(); } } -type RowId = string; -type ColumnValue = HashColumnKey; -class HashIndexTx implements IndexTx { +class DriverHashIndexTx implements IndexTx { get type(): "hash" | "uniqhash" { return this.originalIndex.type; } - originalIndex: HashIndex; - private txBuckets = new Map>(); - private txRowKeys = new Map(); - isCommitted = false; + readonly originalIndex: DriverHashIndex; + private readonly store: HashIndexStoreTx; - constructor(index: HashIndex) { + constructor(index: DriverHashIndex, store: HashIndexStoreTx) { this.originalIndex = index; + this.store = store; } cols(): string[] { @@ -574,147 +498,33 @@ class HashIndexTx implements IndexTx { } commit(): void { - if (this.isCommitted) throw new Error("Can't commit after commit"); - - this.isCommitted = true; - for (const [columnValue, rows] of this.txBuckets) { - if (rows.size === 0) { - this.originalIndex.records.delete(columnValue); - } else { - this.originalIndex.records.set(columnValue, rows); - } - } - for (const [rowId, key] of this.txRowKeys) { - if (key === undefined) { - this.originalIndex.rowKeys.delete(rowId); - } else { - this.originalIndex.rowKeys.set(rowId, key); - } - } + this.store.commit(); } rollback(): void { - if (this.isCommitted) throw new Error("Can't rollback after commit"); - - this.isCommitted = true; + this.store.rollback(); } scan(tupleBounds: TupleScanOptions[], selectOptions: SelectOptions): Row[] { - if (this.isCommitted) throw new Error("Can't scan after commit"); - if (selectOptions.limit !== undefined && selectOptions.limit <= 0) - return []; - - const idxValues = getColumnValuesFromBounds( - this.originalIndex.indexDef, - tupleBounds, + return this.store.scan( + getColumnValuesFromBounds(this.originalIndex.indexDef, tupleBounds), + selectOptions, ); - - const results: Row[] = []; - - for (const idxValue of idxValues) { - const rows = this.txBuckets.has(idxValue) - ? this.txBuckets.get(idxValue) - : this.originalIndex.records.get(idxValue); - - if (!rows) continue; - - for (const row of rows.values()) { - results.push(row); - - if ( - selectOptions.limit !== undefined && - results.length >= selectOptions.limit - ) { - return results; - } - } - } - - return results; - } - - private writableRows(columnValue: ColumnValue): Map | undefined { - const txRows = this.txBuckets.get(columnValue); - if (txRows) return txRows; - - const rows = this.originalIndex.records.get(columnValue); - if (!rows) return undefined; - - const copiedRows = new Map(rows); - this.txBuckets.set(columnValue, copiedRows); - return copiedRows; } validateInsert(values: Row[], replacingIds: Set): void { - if (!this.originalIndex.indexDef.unique) return; - - const batchKeys = new Map(); - for (const record of values) { - const colValue = getHashIndexValue( - record, - this.originalIndex.indexDef.column, - ); - if (colValue === undefined) continue; - const key = hashIndexKey(colValue); - const batchId = batchKeys.get(key); - if (batchId !== undefined && batchId !== record.id) { - throw new Error( - `Unique hash index ${this.originalIndex.indexDef.name} already has value for record ${batchId}`, - ); - } - batchKeys.set(key, record.id); - - const existingRows = this.txBuckets.has(key) - ? this.txBuckets.get(key) - : this.originalIndex.records.get(key); - if (!existingRows) continue; - - for (const existingId of existingRows.keys()) { - if (existingId === record.id || replacingIds.has(existingId)) continue; - throw new Error( - `Unique hash index ${this.originalIndex.indexDef.name} already has value for record ${existingId}`, - ); - } - } + this.store.validateInsert( + hashEntries(this.originalIndex.indexDef, values), + replacingIds, + ); } insert(values: Row[]): void { - if (this.isCommitted) throw new Error("Can't insert after commit"); - - for (const record of values) { - const colValue = getHashIndexValue( - record, - this.originalIndex.indexDef.column, - ); - if (colValue === undefined) continue; - const key = hashIndexKey(colValue); - this.txRowKeys.set(record.id, key); - - const rows = this.writableRows(key); - - if (!rows) { - const m = new Map(); - m.set(record.id, record); - this.txBuckets.set(key, m); - } else { - rows.set(record.id, record); - } - } + this.store.insert(hashEntries(this.originalIndex.indexDef, values)); } delete(values: Row[]): void { - if (this.isCommitted) throw new Error("Can't delete after commit"); - - for (const record of values) { - const key = this.txRowKeys.has(record.id) - ? this.txRowKeys.get(record.id) - : this.originalIndex.rowKeys.get(record.id); - if (key === undefined) continue; - - const rows = this.writableRows(key); - rows?.delete(record.id); - this.txRowKeys.set(record.id, undefined); - } + this.store.delete(values.map((record) => record.id)); } } @@ -954,12 +764,12 @@ export class BptreeInmemDriverTx implements DBDriverTX { const indexes = new Map(); - let idTxIndex: HashIndexTx | undefined; + let idTxIndex: DriverHashIndexTx | undefined; for (const [name, index] of noTxTableData.indexes) { const txIndex = index.tx(); if (index === noTxTableData.idIndex) { - idTxIndex = txIndex as HashIndexTx; + idTxIndex = txIndex as DriverHashIndexTx; } indexes.set(name, txIndex); @@ -1033,7 +843,7 @@ export class BptreeInmemDriver implements DBDriver { indexes.set( indexName, - new HashIndex({ + new DriverHashIndex({ name: indexName, column: indexDef.cols[0] as string, unique: indexDef.type === "uniqhash", @@ -1044,9 +854,12 @@ export class BptreeInmemDriver implements DBDriver { } } - let idIndex: HashIndex | undefined; + let idIndex: DriverHashIndex | undefined; for (const index of indexes.values()) { - if (index instanceof HashIndex && index.indexDef.column === "id") { + if ( + index instanceof DriverHashIndex && + index.indexDef.column === "id" + ) { idIndex = index; break; } @@ -1066,6 +879,19 @@ export class BptreeInmemDriver implements DBDriver { } } + *scanAll(tableName: string): Generator { + if (this.isInTransaction) { + throw new Error("can't run while transaction is in progress"); + } + + const tableData = this.tblDatas.get(tableName); + if (!tableData) { + throw new Error(`Table ${tableName} not found`); + } + + return tableData.idIndex.values(); + } + *upsert(tableName: string, values: Row[]): Generator { if (this.isInTransaction) { throw new Error("can't run while transaction is in progress"); diff --git a/packages/hyperdb/src/hyperdb/drivers/sqlite/async-sql-driver.ts b/packages/hyperdb/src/hyperdb/drivers/sqlite/async-sql-driver.ts index d64636e..7461c80 100644 --- a/packages/hyperdb/src/hyperdb/drivers/sqlite/async-sql-driver.ts +++ b/packages/hyperdb/src/hyperdb/drivers/sqlite/async-sql-driver.ts @@ -346,6 +346,42 @@ function* performAsyncScanOperation( }); } +function* performAsyncScanAllOperation( + db: AsyncSQLiteDB, + tableName: string, + debug?: AsyncSqlDriverDebug, +): Generator { + return yield* unwrapCb(async () => { + const sql = `SELECT data FROM ${tableName}`; + const result: Row[] = []; + const startedAt = debug ? nowMs() : 0; + const statement = await db.prepare(sql); + + try { + for (const row of await statement.values([])) { + result.push(parseSqliteStoredRow(row[0] as string)); + } + emitAsyncSqlDebug(debug, "scan", sql, startedAt, () => ({ + tableName, + rowCount: result.length, + })); + return result; + } catch (error) { + emitAsyncSqlDebug( + debug, + "scan", + sql, + startedAt, + () => ({ tableName, rowCount: result.length }), + error, + ); + throw error; + } finally { + await statement.finalize(); + } + }); +} + class AsyncSqlDriverTx implements DBDriverTX { private db: AsyncSQLiteDB; private tableDefinitions: Map; @@ -678,6 +714,23 @@ export class AsyncSqlDriver implements DBDriver { } } + *scanAll(tableName: string): Generator { + yield* unwrapCb(async () => { + await this.txAndQueryLock.acquireAsync(); + }); + + try { + this.getTableDefinition(tableName); + return yield* performAsyncScanAllOperation( + this.db, + tableName, + this.debug, + ); + } finally { + this.txAndQueryLock.release(); + } + } + *loadTables( tableDefinitions: TableDefinition[], ): Generator { diff --git a/packages/hyperdb/src/hyperdb/drivers/sqlite/sql-driver.ts b/packages/hyperdb/src/hyperdb/drivers/sqlite/sql-driver.ts index e76df2f..b14dc9e 100644 --- a/packages/hyperdb/src/hyperdb/drivers/sqlite/sql-driver.ts +++ b/packages/hyperdb/src/hyperdb/drivers/sqlite/sql-driver.ts @@ -143,6 +143,17 @@ function performScanOperation( } } +function performScanAllOperation(db: SQLiteDB, tableName: string): Row[] { + const statement = db.prepare(`SELECT data FROM ${tableName}`); + try { + return statement + .values([]) + .map((row) => parseSqliteStoredRow(row[0] as string)); + } finally { + statement.finalize(); + } +} + function rollbackQuietly(db: SQLiteDB): void { try { db.exec("ROLLBACK"); @@ -341,6 +352,16 @@ export class SqlDriver implements DBDriver { ); } + *scanAll(tableName: string): Generator { + if (this.isInTransaction) { + throw new Error("can't run while transaction is in progress"); + } + if (!this.tableDefinitions.has(tableName)) { + throw new Error(`Table ${tableName} not found`); + } + return performScanAllOperation(this.db, tableName); + } + *loadTables( tableDefinitions: TableDefinition[], ): Generator { diff --git a/packages/hyperdb/src/hyperdb/index.ts b/packages/hyperdb/src/hyperdb/index.ts index 408ece7..640e112 100644 --- a/packages/hyperdb/src/hyperdb/index.ts +++ b/packages/hyperdb/src/hyperdb/index.ts @@ -14,6 +14,7 @@ export type { export * from "./core/query/bounds"; export * from "./runtime/subscribable-db"; export * from "./runtime/hybrid-db"; +export * from "./runtime/preloaded-hybrid-db"; export * from "./schema/table"; export * from "./schema/values"; export * from "./tracing"; diff --git a/packages/hyperdb/src/hyperdb/runtime/db.test.ts b/packages/hyperdb/src/hyperdb/runtime/db.test.ts index 846b6a3..9533c78 100644 --- a/packages/hyperdb/src/hyperdb/runtime/db.test.ts +++ b/packages/hyperdb/src/hyperdb/runtime/db.test.ts @@ -7,6 +7,7 @@ import { defineTable } from "../schema/table"; import { v } from "../schema/values"; import { AsyncDB } from "../test-utils/async-db"; import { createDriverFactories } from "../test-utils/driver-factories"; +import { execAsync } from "../core/executor"; export const fractionalCompare = ( item1: T, @@ -96,6 +97,23 @@ describe("db", async () => { ).resolves.toBeUndefined(); }); + it( + "bulk-reads a table without requiring a full-scan index - " + driverName, + async () => { + const db = new DB(await createDriver()); + await execAsync(db.loadTables([writeSemanticsTable])); + const rows = [ + { id: "a", value: "A" }, + { id: "b", value: "B", optionalValue: "present" }, + ]; + await execAsync(db.insert(writeSemanticsTable, rows)); + + await expect( + execAsync(db.scanAll(writeSemanticsTable)), + ).resolves.toEqual(rows); + }, + ); + it( "queries matching uniqhash and B-tree logical indexes - " + driverName, async () => { diff --git a/packages/hyperdb/src/hyperdb/runtime/db.ts b/packages/hyperdb/src/hyperdb/runtime/db.ts index e3a9221..cd48b33 100644 --- a/packages/hyperdb/src/hyperdb/runtime/db.ts +++ b/packages/hyperdb/src/hyperdb/runtime/db.ts @@ -223,6 +223,22 @@ export class DB implements HyperDB { _specs: TSpecs & ValidateHybridPreloadTableSpecs, ): Generator {} + /** @internal Bulk source for runtimes that preload derived indexes. */ + *scanAll( + table: TTable, + ): Generator[]> { + if (!this.driver.scanAll) { + throw new Error( + `Driver does not support startup index preloading for table: ${table.tableName}`, + ); + } + + const records = yield* this.driver.scanAll(table.tableName, { + traceContext: getDriverTraceContextForDB(this), + }); + return validateRecordsFromDriver(table, records, this.options); + } + *beginTx(mode: DBTransactionMode = "readwrite"): Generator { const tx = yield* this.driver.beginTx(mode, { traceContext: getDriverTraceContextForDB(this), diff --git a/packages/hyperdb/src/hyperdb/runtime/preloaded-hybrid-db-indexes.ts b/packages/hyperdb/src/hyperdb/runtime/preloaded-hybrid-db-indexes.ts new file mode 100644 index 0000000..c06d9ef --- /dev/null +++ b/packages/hyperdb/src/hyperdb/runtime/preloaded-hybrid-db-indexes.ts @@ -0,0 +1,416 @@ +import { convertWhereToBound } from "../core/query/bounds"; +import { compareStoredTuple, compareTuple } from "../core/query/tuple"; +import type { + Row, + ScanValue, + SelectOptions, + Value, + WhereClause, +} from "../core/primitives"; +import type { TableDefinition } from "../schema/table"; +import { InMemoryBinaryPlusTree } from "../structures/bptree"; +import { + HashIndex, + HashIndexTx, + hashIndexKey, + type HashIndexEntry, + type HashIndexView, +} from "../structures/hash-index"; + +type IndexDefinition = TableDefinition["indexes"][string]; +type IndexEntry = { key: ScanValue[]; value: string }; + +const isSchemalessTable = (table: TableDefinition): boolean => + !table.schemaValidator; + +function rowValue(row: Row, column: string): Value | undefined { + if (!Object.prototype.hasOwnProperty.call(row, column)) return undefined; + const value = row[column]; + return value === undefined ? null : (value as Value); +} + +function btreeKey( + row: Row, + columns: readonly string[], + includeMissing: boolean, +): ScanValue[] | undefined { + const values: ScanValue[] = []; + for (const column of columns) { + if (!Object.prototype.hasOwnProperty.call(row, column)) { + if (!includeMissing) return undefined; + values.push(undefined as unknown as ScanValue); + continue; + } + const value = row[column]; + values.push((value === undefined ? null : value) as ScanValue); + } + return values; +} + +class IdHeap { + private values: T[] = []; + private readonly compare: (left: T, right: T) => number; + + constructor(compare: (left: T, right: T) => number) { + this.compare = compare; + } + + push(value: T): void { + this.values.push(value); + let index = this.values.length - 1; + while (index > 0) { + const parent = (index - 1) >> 1; + if (this.compare(this.values[index]!, this.values[parent]!) >= 0) break; + [this.values[index], this.values[parent]] = [ + this.values[parent]!, + this.values[index]!, + ]; + index = parent; + } + } + + pop(): T | undefined { + const first = this.values[0]; + const last = this.values.pop(); + if (last === undefined || this.values.length === 0) return first; + this.values[0] = last; + + let index = 0; + while (true) { + const left = index * 2 + 1; + const right = left + 1; + let smallest = index; + if ( + left < this.values.length && + this.compare(this.values[left]!, this.values[smallest]!) < 0 + ) { + smallest = left; + } + if ( + right < this.values.length && + this.compare(this.values[right]!, this.values[smallest]!) < 0 + ) { + smallest = right; + } + if (smallest === index) break; + [this.values[index], this.values[smallest]] = [ + this.values[smallest]!, + this.values[index]!, + ]; + index = smallest; + } + return first; + } + + get size(): number { + return this.values.length; + } +} + +interface IdOnlyIndex { + scan(clauses: WhereClause[], options: SelectOptions): string[]; + validateUpsert(rows: Row[]): void; + upsert(rows: Row[]): void; + delete(ids: string[]): void; + fork(): IdOnlyIndex; + materializeFork(): void; + discardFork(): void; +} + +class IdBtreeIndex implements IdOnlyIndex { + private tree: InMemoryBinaryPlusTree; + private readonly keysById: Map; + private readonly logicalColumns: string[]; + private readonly storedColumns: string[]; + private readonly includeMissing: boolean; + private readonly compareKey: ( + left: ScanValue[], + right: ScanValue[], + ) => number; + + constructor( + definition: IndexDefinition, + includeMissing: boolean, + tree?: InMemoryBinaryPlusTree, + keysById?: Map, + ) { + this.logicalColumns = definition.cols.map(String); + this.storedColumns = [...this.logicalColumns]; + if (this.storedColumns[this.storedColumns.length - 1] !== "id") { + this.storedColumns.push("id"); + } + this.includeMissing = includeMissing; + this.compareKey = this.includeMissing + ? (compareStoredTuple as ( + left: ScanValue[], + right: ScanValue[], + ) => number) + : compareTuple; + this.tree = + tree ?? + new InMemoryBinaryPlusTree(64, 128, this.compareKey); + this.keysById = keysById ?? new Map(); + } + + scan(clauses: WhereClause[], options: SelectOptions): string[] { + if (options.limit !== undefined && options.limit <= 0) return []; + const bounds = convertWhereToBound(this.storedColumns, clauses); + const iterators = bounds.map((bound) => + this.tree.iterate({ + ...bound, + reverse: options.order === "desc", + }), + ); + type Cursor = { + iterator: IterableIterator; + current: IndexEntry; + sequence: number; + }; + const reverse = options.order === "desc"; + const heap = new IdHeap((left, right) => { + const compared = this.compareKey(left.current.key, right.current.key); + if (compared !== 0) return reverse ? -compared : compared; + return left.sequence - right.sequence; + }); + + iterators.forEach((iterator, sequence) => { + const next = iterator.next(); + if (!next.done) heap.push({ iterator, current: next.value, sequence }); + }); + + const result: string[] = []; + const seen = new Set(); + while (heap.size > 0) { + const cursor = heap.pop()!; + if (!seen.has(cursor.current.value)) { + seen.add(cursor.current.value); + result.push(cursor.current.value); + if (options.limit !== undefined && result.length >= options.limit) { + return result; + } + } + const next = cursor.iterator.next(); + if (!next.done) { + cursor.current = next.value; + heap.push(cursor); + } + } + return result; + } + + upsert(rows: Row[]): void { + for (const row of rows) { + const oldKey = this.keysById.get(row.id); + if (oldKey) this.tree.delete(oldKey); + + const key = btreeKey(row, this.storedColumns, this.includeMissing); + if (!key) { + this.keysById.delete(row.id); + continue; + } + this.keysById.set(row.id, key); + this.tree.set(key, row.id); + } + } + + validateUpsert(_rows: Row[]): void {} + + delete(ids: string[]): void { + for (const id of ids) { + const key = this.keysById.get(id); + if (key) this.tree.delete(key); + this.keysById.delete(id); + } + } + + fork(): IdOnlyIndex { + return new IdBtreeIndex( + { type: "btree", cols: this.logicalColumns }, + this.includeMissing, + this.tree.fork(), + new Map(this.keysById), + ); + } + + materializeFork(): void { + this.tree = this.tree.materializeFork(); + } + + discardFork(): void { + this.tree.discardFork(); + } +} + +class IdHashIndex implements IdOnlyIndex { + private readonly name: string; + private readonly column: string; + private readonly unique: boolean; + private index: HashIndex | HashIndexTx; + + constructor( + name: string, + definition: IndexDefinition, + index?: HashIndex | HashIndexTx, + ) { + this.name = name; + this.column = String(definition.cols[0]); + this.unique = definition.type === "uniqhash"; + this.index = index ?? new HashIndex({ name, unique: this.unique }); + } + + scan(clauses: WhereClause[], options: SelectOptions): string[] { + return scanHashIndex(this.index, this.column, clauses, options); + } + + private entries(rows: readonly Row[]): HashIndexEntry[] { + return rows.flatMap((row) => { + const value = rowValue(row, this.column); + return value === undefined + ? [] + : [{ key: value, id: row.id, value: row.id }]; + }); + } + + validateUpsert(rows: Row[]): void { + const entries = this.entries(rows); + this.index.validateInsert(entries, new Set(rows.map((row) => row.id))); + } + + upsert(rows: Row[]): void { + this.index.delete(rows.map((row) => row.id)); + this.index.insert(this.entries(rows)); + } + + delete(ids: string[]): void { + this.index.delete(ids); + } + + fork(): IdOnlyIndex { + if (!(this.index instanceof HashIndex)) { + throw new Error(`Cannot fork hash index transaction ${this.name}`); + } + return new IdHashIndex( + this.name, + { + type: this.unique ? "uniqhash" : "hash", + cols: [this.column], + }, + this.index.tx(), + ); + } + + materializeFork(): void { + if (this.index instanceof HashIndexTx) this.index = this.index.commit(); + } + + discardFork(): void { + if (this.index instanceof HashIndexTx) this.index = this.index.rollback(); + } +} + +export function hashScanValues( + column: string, + clauses: WhereClause[], +): Value[] { + const bounds = convertWhereToBound([column], clauses); + const values: Value[] = []; + + for (const bound of bounds) { + if ( + bound.gt || + bound.lt || + !bound.gte || + !bound.lte || + bound.gte.length !== 1 || + bound.lte.length !== 1 || + typeof bound.gte[0] === "symbol" || + typeof bound.lte[0] === "symbol" || + hashIndexKey(bound.gte[0]) !== hashIndexKey(bound.lte[0]) + ) { + throw new Error( + `Hash index should have exactly one equality condition for column '${column}'`, + ); + } + values.push(bound.gte[0]); + } + + return values; +} + +export function scanHashIndex( + index: HashIndexView, + column: string, + clauses: WhereClause[], + options: SelectOptions = {}, +): T[] { + return index.scan(hashScanValues(column, clauses), options); +} + +export class PreloadedTableIndexes { + private readonly indexes = new Map(); + readonly table: TableDefinition; + + private static empty(table: TableDefinition): PreloadedTableIndexes { + return new PreloadedTableIndexes(table, [], false); + } + + constructor( + table: TableDefinition, + rows: Row[] = [], + initializeIndexes = true, + ) { + this.table = table; + if (!initializeIndexes) return; + + for (const [indexName, definition] of Object.entries(table.indexes)) { + if (indexName === table.idIndexName) continue; + this.indexes.set( + indexName, + definition.type === "btree" + ? new IdBtreeIndex(definition, isSchemalessTable(table)) + : new IdHashIndex(indexName, definition), + ); + } + this.upsert(rows); + } + + scan( + indexName: string, + clauses: WhereClause[], + options: SelectOptions = {}, + ): string[] { + if (clauses.length === 0) throw new Error("scan clauses must be provided"); + const index = this.indexes.get(indexName); + if (!index) { + throw new Error( + `Index not found: ${indexName} for table: ${this.table.tableName}`, + ); + } + return index.scan(clauses, options); + } + + upsert(rows: Row[]): void { + for (const index of this.indexes.values()) index.validateUpsert(rows); + for (const index of this.indexes.values()) index.upsert(rows); + } + + delete(ids: string[]): void { + for (const index of this.indexes.values()) index.delete(ids); + } + + fork(): PreloadedTableIndexes { + const fork = PreloadedTableIndexes.empty(this.table); + for (const [name, index] of this.indexes) { + fork.indexes.set(name, index.fork()); + } + return fork; + } + + materializeFork(): void { + for (const index of this.indexes.values()) index.materializeFork(); + } + + discardFork(): void { + for (const index of this.indexes.values()) index.discardFork(); + } +} diff --git a/packages/hyperdb/src/hyperdb/runtime/preloaded-hybrid-db.test.ts b/packages/hyperdb/src/hyperdb/runtime/preloaded-hybrid-db.test.ts new file mode 100644 index 0000000..9234010 --- /dev/null +++ b/packages/hyperdb/src/hyperdb/runtime/preloaded-hybrid-db.test.ts @@ -0,0 +1,433 @@ +import { describe, expect, it, vi } from "vitest"; +import { DB } from "./db"; +import { + externalStorageMergeTrait, + PreloadedHybridDB, +} from "./preloaded-hybrid-db"; +import { SubscribableDB, type Op } from "./subscribable-db"; +import { PreloadedTableIndexes } from "./preloaded-hybrid-db-indexes"; +import { AsyncDB } from "../test-utils/async-db"; +import { createSqlJsDriver } from "../test-utils/sql-js-driver"; +import { defineTable, type TableDefinition } from "../schema/table"; +import { v } from "../schema/values"; +import { execAsync } from "../core/executor"; + +const tasksTable = defineTable("preloadedHybridTasks", { + id: v.string(), + title: v.string(), + value: v.number(), + slug: v.string(), +}) + .index("byValue", ["value"]) + .index("byTitle", ["title"], { type: "hash" }) + .index("bySlug", ["slug"], { type: "uniqhash" }); + +const projectsTable = defineTable("preloadedHybridProjects", { + id: v.string(), + title: v.string(), +}); + +const schemalessHashTable = { + tableName: "preloadedHybridSchemalessHash", + schema: {}, + indexes: { + byId: { type: "uniqhash", cols: ["id"] }, + byTag: { type: "hash", cols: ["tag"] }, + }, + idIndexName: "byId", +} as unknown as TableDefinition; + +type Task = { + id: string; + title: string; + value: number; + slug: string; +}; + +const task = (value: number, title = `Task ${value}`): Task => ({ + id: String(value).padStart(3, "0"), + title, + value, + slug: `task-${value}`, +}); + +async function createRuntime(rows: Task[]) { + const primary = new DB(await createSqlJsDriver()); + const primaryDB = new AsyncDB(primary); + await primaryDB.loadTables([tasksTable, projectsTable]); + await primaryDB.insert(tasksTable, rows); + await primaryDB.insert(projectsTable, [{ id: "p1", title: "Project" }]); + + const scanAllSpy = vi.spyOn(primary, "scanAll"); + const intervalScanSpy = vi.spyOn(primary, "intervalScan"); + const runtime = new PreloadedHybridDB(primary); + const db = new AsyncDB(runtime); + await db.loadTables([tasksTable, projectsTable]); + + return { db, runtime, primary, scanAllSpy, intervalScanSpy }; +} + +describe("PreloadedHybridDB", () => { + it("preloads every table index without retaining entity rows", async () => { + const rows = [task(1, "same"), task(2, "same"), task(3, "other")]; + const { db, scanAllSpy, intervalScanSpy } = await createRuntime(rows); + + expect(scanAllSpy).toHaveBeenCalledTimes(2); + expect(intervalScanSpy).not.toHaveBeenCalled(); + + await expect( + db.intervalScan(tasksTable, "byValue", [ + { gte: [{ col: "value", val: 1 }], lte: [{ col: "value", val: 2 }] }, + ]), + ).resolves.toEqual(rows.slice(0, 2)); + expect(intervalScanSpy).toHaveBeenCalledTimes(1); + expect(intervalScanSpy.mock.calls[0]?.[1]).toBe("byId"); + expect(intervalScanSpy.mock.calls[0]?.[2]).toEqual([ + { eq: [{ col: "id", val: "001" }] }, + { eq: [{ col: "id", val: "002" }] }, + ]); + + intervalScanSpy.mockClear(); + await expect( + db.intervalScan(tasksTable, "byTitle", [ + { eq: [{ col: "title", val: "same" }] }, + ]), + ).resolves.toEqual(rows.slice(0, 2)); + await expect( + db.intervalScan(tasksTable, "bySlug", [ + { eq: [{ col: "slug", val: "task-1" }] }, + ]), + ).resolves.toEqual([rows[0]]); + expect(intervalScanSpy).not.toHaveBeenCalled(); + + await expect( + db.intervalScan(tasksTable, "byValue", [ + { gte: [{ col: "value", val: 1 }], lte: [{ col: "value", val: 3 }] }, + ]), + ).resolves.toEqual(rows); + expect(intervalScanSpy).toHaveBeenCalledTimes(1); + expect(intervalScanSpy.mock.calls[0]?.[2]).toEqual([ + { eq: [{ col: "id", val: "003" }] }, + ]); + }); + + it("supports tables whose only declared index is the built-in byId", async () => { + const { db, intervalScanSpy } = await createRuntime([task(1)]); + + await expect( + db.intervalScan(projectsTable, "byId", [ + { eq: [{ col: "id", val: "p1" }] }, + ]), + ).resolves.toEqual([{ id: "p1", title: "Project" }]); + expect(intervalScanSpy).toHaveBeenCalledTimes(1); + expect(intervalScanSpy.mock.calls[0]?.[1]).toBe("byId"); + }); + + it("preserves tables loaded by earlier loadTables calls", async () => { + const primary = new DB(await createSqlJsDriver()); + const primaryDB = new AsyncDB(primary); + const row = task(1); + await primaryDB.loadTables([tasksTable, projectsTable]); + await primaryDB.insert(tasksTable, [row]); + await primaryDB.insert(projectsTable, [{ id: "p1", title: "Project" }]); + + const db = new AsyncDB(new PreloadedHybridDB(primary)); + await db.loadTables([tasksTable]); + await db.loadTables([projectsTable]); + + await expect( + db.intervalScan(tasksTable, "byValue", [ + { eq: [{ col: "value", val: row.value }] }, + ]), + ).resolves.toEqual([row]); + await expect( + db.intervalScan(projectsTable, "byId", [ + { eq: [{ col: "id", val: "p1" }] }, + ]), + ).resolves.toEqual([{ id: "p1", title: "Project" }]); + }); + + it("uses the built-in byId hash index as the canonical entity cache", async () => { + const first = task(1); + const second = task(2); + const { db, runtime } = await createRuntime([first, second]); + const internal = runtime as unknown as { + state: { + data: { + tables: Map< + string, + { + byId: { + values(): Array< + | { id: string; loaded: false } + | { id: string; loaded: true; row: Task } + >; + }; + } + >; + }; + }; + }; + const byId = internal.state.data.tables.get(tasksTable.tableName)?.byId; + + expect( + byId?.values().sort((left, right) => left.id.localeCompare(right.id)), + ).toEqual([ + { id: first.id, loaded: false }, + { id: second.id, loaded: false }, + ]); + + await db.intervalScan(tasksTable, "bySlug", [ + { eq: [{ col: "slug", val: first.slug }] }, + ]); + + expect( + byId?.values().sort((left, right) => left.id.localeCompare(right.id)), + ).toEqual([ + { id: first.id, loaded: true, row: first }, + { id: second.id, loaded: false }, + ]); + }); + + it("reconciles exact byId misses written by another runtime", async () => { + const { db, primary, intervalScanSpy } = await createRuntime([]); + const external = task(7, "external"); + await execAsync(primary.insert(tasksTable, [external])); + intervalScanSpy.mockClear(); + + await expect( + db.intervalScan(tasksTable, "byId", [ + { eq: [{ col: "id", val: external.id }] }, + ]), + ).resolves.toEqual([external]); + expect(intervalScanSpy).toHaveBeenCalledTimes(1); + + intervalScanSpy.mockClear(); + await expect( + db.intervalScan(tasksTable, "bySlug", [ + { eq: [{ col: "slug", val: external.slug }] }, + ]), + ).resolves.toEqual([external]); + expect(intervalScanSpy).not.toHaveBeenCalled(); + }); + + it("naturally publishes an external insert already present in storage", async () => { + const { runtime, primary, intervalScanSpy } = await createRuntime([]); + const subscribable = new SubscribableDB(runtime); + const external = task(8, "external merge"); + await execAsync(primary.insert(tasksTable, [external])); + intervalScanSpy.mockClear(); + + const notifications: { operations: Op[]; traits: string[] }[] = []; + subscribable.subscribe((operations, traits) => { + notifications.push({ + operations, + traits: traits.map((trait) => trait.type), + }); + }); + + const externalMergeDB = new AsyncDB( + subscribable.withTraits({ type: "skip-sync" }, externalStorageMergeTrait), + ); + + await expect( + externalMergeDB.intervalScan(tasksTable, "byId", [ + { eq: [{ col: "id", val: external.id }] }, + ]), + ).resolves.toEqual([]); + expect(intervalScanSpy).not.toHaveBeenCalled(); + + await externalMergeDB.insert(tasksTable, [external]); + + expect(notifications).toEqual([ + { + operations: [{ type: "insert", table: tasksTable, newValue: external }], + traits: ["skip-sync", externalStorageMergeTrait.type], + }, + ]); + await expect( + new AsyncDB(subscribable).intervalScan(tasksTable, "bySlug", [ + { eq: [{ col: "slug", val: external.slug }] }, + ]), + ).resolves.toEqual([external]); + await expect( + new AsyncDB(primary).intervalScan(tasksTable, "byId", [ + { eq: [{ col: "id", val: external.id }] }, + ]), + ).resolves.toEqual([external]); + + await expect( + new AsyncDB(subscribable).insert(tasksTable, [external]), + ).rejects.toThrow(); + }); + + it("keeps B-tree ordering and limits while hydrating only selected IDs", async () => { + const rows = [task(1), task(2), task(3), task(4)]; + const { db, intervalScanSpy } = await createRuntime(rows); + + await expect( + db.intervalScan(tasksTable, "byValue", [{}], { + order: "desc", + limit: 2, + }), + ).resolves.toEqual([rows[3], rows[2]]); + expect(intervalScanSpy.mock.calls[0]?.[2]).toEqual([ + { eq: [{ col: "id", val: "004" }] }, + { eq: [{ col: "id", val: "003" }] }, + ]); + }); + + it("applies byId limits before hydrating entities", async () => { + const rows = [task(1), task(2), task(3)]; + const { db, intervalScanSpy } = await createRuntime(rows); + const clauses = rows.map((row) => ({ + eq: [{ col: "id", val: row.id }], + })); + + await expect( + db.intervalScan(tasksTable, "byId", clauses, { limit: 2 }), + ).resolves.toEqual(rows.slice(0, 2)); + expect(intervalScanSpy.mock.calls[0]?.[2]).toEqual(clauses.slice(0, 2)); + + intervalScanSpy.mockClear(); + await expect( + db.intervalScan(tasksTable, "byId", clauses, { limit: 0 }), + ).resolves.toEqual([]); + expect(intervalScanSpy).not.toHaveBeenCalled(); + }); + + it("updates preloaded indexes and the entity cache on writes", async () => { + const original = task(1); + const { db, intervalScanSpy } = await createRuntime([original]); + intervalScanSpy.mockClear(); + + const inserted = task(2, "new"); + await db.insert(tasksTable, [inserted]); + await expect( + db.intervalScan(tasksTable, "byTitle", [ + { eq: [{ col: "title", val: "new" }] }, + ]), + ).resolves.toEqual([inserted]); + + const updated = { ...original, title: "updated", value: 10 }; + await db.upsert(tasksTable, [updated]); + await expect( + db.intervalScan(tasksTable, "byValue", [ + { eq: [{ col: "value", val: 10 }] }, + ]), + ).resolves.toEqual([updated]); + await expect( + db.intervalScan(tasksTable, "byValue", [ + { eq: [{ col: "value", val: 1 }] }, + ]), + ).resolves.toEqual([]); + + await db.delete(tasksTable, [inserted.id]); + await expect( + db.intervalScan(tasksTable, "byId", [ + { eq: [{ col: "id", val: inserted.id }] }, + ]), + ).resolves.toEqual([]); + expect(intervalScanSpy).toHaveBeenCalledTimes(1); + }); + + it("publishes committed transaction index changes and discards rollbacks", async () => { + const original = task(1); + const { db } = await createRuntime([original]); + + const rollbackTx = await db.beginTx(); + await rollbackTx.upsert(tasksTable, [{ ...original, value: 2 }]); + await expect( + rollbackTx.intervalScan(tasksTable, "byValue", [ + { eq: [{ col: "value", val: 2 }] }, + ]), + ).resolves.toEqual([{ ...original, value: 2 }]); + await rollbackTx.rollback(); + await expect( + db.intervalScan(tasksTable, "byValue", [ + { eq: [{ col: "value", val: 1 }] }, + ]), + ).resolves.toEqual([original]); + + const commitTx = await db.beginTx(); + const committed = { ...original, value: 3 }; + await commitTx.upsert(tasksTable, [committed]); + await commitTx.commit(); + await expect( + db.intervalScan(tasksTable, "byValue", [ + { eq: [{ col: "value", val: 3 }] }, + ]), + ).resolves.toEqual([committed]); + }); + + it("rejects writes through readonly transactions", async () => { + const original = task(1); + const { runtime } = await createRuntime([original]); + const tx = await execAsync(runtime.beginTx("readonly")); + try { + await expect( + execAsync(tx.upsert(tasksTable, [{ ...original, value: 2 }])), + ).rejects.toThrow("Cannot write through a readonly transaction"); + } finally { + await execAsync(tx.rollback()); + } + }); + + it("supports swapping unique values in one upsert batch", async () => { + const first = task(1); + const second = task(2); + const { db } = await createRuntime([first, second]); + const swapped = [ + { ...first, slug: second.slug }, + { ...second, slug: first.slug }, + ]; + + await db.upsert(tasksTable, swapped); + await expect( + db.intervalScan(tasksTable, "bySlug", [ + { eq: [{ col: "slug", val: first.slug }] }, + ]), + ).resolves.toEqual([swapped[1]]); + }); + + it("removes stale hash entries when a schemaless row loses the column", () => { + const indexes = new PreloadedTableIndexes(schemalessHashTable, [ + { id: "row-1", tag: "old" }, + ]); + + indexes.upsert([{ id: "row-1" }]); + + expect( + indexes.scan("byTag", [{ eq: [{ col: "tag", val: "old" }] }]), + ).toEqual([]); + }); + + it("stores entity IDs, not rows, in B-tree and hash index leaves", () => { + const indexes = new PreloadedTableIndexes(tasksTable, [ + task(1, "same"), + task(2, "same"), + ]); + const internal = indexes as unknown as { + indexes: Map< + string, + { + tree?: { nodes: Map }; + index?: { buckets: Map> }; + } + >; + }; + + const btree = internal.indexes.get("byValue")?.tree; + const btreeLeafValues = Array.from(btree?.nodes.values() ?? []) + .filter((node) => node.leaf) + .flatMap((node) => node.values ?? []) as { value: unknown }[]; + expect(btreeLeafValues.map((entry) => entry.value)).toEqual(["001", "002"]); + + const hashBuckets = internal.indexes.get("byTitle")?.index?.buckets; + expect( + Array.from(hashBuckets?.values() ?? []).flatMap((bucket) => [ + ...bucket.values(), + ]), + ).toEqual(["001", "002"]); + }); +}); diff --git a/packages/hyperdb/src/hyperdb/runtime/preloaded-hybrid-db.ts b/packages/hyperdb/src/hyperdb/runtime/preloaded-hybrid-db.ts new file mode 100644 index 0000000..0903377 --- /dev/null +++ b/packages/hyperdb/src/hyperdb/runtime/preloaded-hybrid-db.ts @@ -0,0 +1,623 @@ +import type { DBCmd } from "../commands/async"; +import { unwrap } from "../commands/async"; +import type { + HyperDB, + HyperDBTx, + HybridPreloadTableSpecInput, + ValidateHybridPreloadTableSpecs, +} from "../core/contracts"; +import type { DBTransactionMode } from "../core/driver"; +import type { + Row, + SelectOptions, + Trait, + WhereClause, +} from "../core/primitives"; +import { + getCurrentSelectEventForDB, + getDriverTraceContextForDB, + type HyperDBTracerOption, + withDriverTraceContextTrait, +} from "../core/tracer"; +import type { + ExtractIndexes, + ExtractSchema, + TableDefinition, +} from "../schema/table"; +import { DEFAULT_CODEC_OPTIONS, type CodecOptions } from "../storage/codec"; +import { + HashIndex, + HashIndexTx, + type HashIndexEntry, + type HashIndexView, +} from "../structures/hash-index"; +import { refVar, type RefVar } from "../utils"; +import AwaitLock from "../utils/await-lock"; +import type { DB } from "./db"; +import { + hashScanValues, + PreloadedTableIndexes, +} from "./preloaded-hybrid-db-indexes"; + +const entityLoadBatchSize = 500; + +export const externalStorageMergeTrait = { + type: "external-storage-merge", +} as const satisfies Trait; + +const usesExternalStorageMerge = (db: HyperDB): boolean => + db.getTraits().some((trait) => trait.type === externalStorageMergeTrait.type); + +type EntityEntry = + | { id: string; loaded: false } + | { id: string; loaded: true; row: Row }; + +type EntityHashIndex = HashIndex | HashIndexTx; + +function entityPointers(rows: readonly Row[]): HashIndexEntry[] { + return rows.map((row) => ({ + key: row.id, + id: row.id, + value: { id: row.id, loaded: false }, + })); +} + +function loadedEntities(rows: readonly Row[]): HashIndexEntry[] { + return rows.map((row) => ({ + key: row.id, + id: row.id, + value: { id: row.id, loaded: true, row }, + })); +} + +function entityById( + index: HashIndexView, + id: string, +): EntityEntry | undefined { + return index.scan([id], { limit: 1 })[0]; +} + +function forkEntityHashIndex( + index: EntityHashIndex, + tableName: string, +): HashIndexTx { + if (!(index instanceof HashIndex)) { + throw new Error(`Cannot fork table transaction ${tableName}`); + } + return index.tx(); +} + +type PreloadedTableState = { + indexes: PreloadedTableIndexes; + byId: EntityHashIndex; +}; + +class PreloadedHybridData { + readonly tables = new Map(); + + get(table: TableDefinition): PreloadedTableState { + const state = this.tables.get(table.tableName); + if (!state) throw new Error(`Table ${table.tableName} not found`); + return state; + } + + fork(): PreloadedHybridData { + const fork = new PreloadedHybridData(); + for (const [tableName, state] of this.tables) { + fork.tables.set(tableName, { + indexes: state.indexes.fork(), + byId: forkEntityHashIndex(state.byId, tableName), + }); + } + return fork; + } + + materializeFork(): void { + for (const state of this.tables.values()) { + state.indexes.materializeFork(); + if (state.byId instanceof HashIndexTx) state.byId = state.byId.commit(); + } + } + + discardFork(): void { + for (const state of this.tables.values()) { + state.indexes.discardFork(); + if (state.byId instanceof HashIndexTx) state.byId = state.byId.rollback(); + } + } +} + +type PreloadedHybridDBState = { + data: PreloadedHybridData; + lock: AwaitLock; +}; + +const createState = (): PreloadedHybridDBState => ({ + data: new PreloadedHybridData(), + lock: new AwaitLock(), +}); + +export type PreloadedHybridDBOptions = { + traits?: Trait[]; +}; + +function* acquireLock(lock: AwaitLock): Generator void> { + if (!lock.tryAcquire()) yield* unwrap(lock.acquireAsync()); + let released = false; + return () => { + if (released) return; + released = true; + lock.release(); + }; +} + +function chunks(values: T[], size: number): T[][] { + const result: T[][] = []; + for (let index = 0; index < values.length; index += size) { + result.push(values.slice(index, index + size)); + } + return result; +} + +function* scanPreloaded( + owner: HyperDB, + primary: HyperDB, + data: PreloadedHybridData, + table: TTable, + indexName: keyof ExtractIndexes, + clauses: WhereClause[], + selectOptions?: SelectOptions, +): Generator[]> { + const tableState = data.get(table); + const limit = selectOptions?.limit; + if (limit !== undefined && limit <= 0) return []; + + const scansById = String(indexName) === table.idIndexName; + let ids: string[]; + if (scansById) { + const seen = new Set(); + ids = hashScanValues("id", clauses).flatMap((value) => { + if (typeof value !== "string") { + throw new Error("Primary-key index byId requires string IDs"); + } + if (seen.has(value)) return []; + seen.add(value); + return [value]; + }); + } else { + ids = tableState.indexes.scan(String(indexName), clauses, selectOptions); + } + if (limit !== undefined) ids = ids.slice(0, limit); + + const missingIds = ids.filter((id) => { + const entry = entityById(tableState.byId, id); + return entry === undefined + ? !usesExternalStorageMerge(owner) + : !entry.loaded; + }); + const selectEvent = getCurrentSelectEventForDB(owner); + + if (selectEvent) { + selectEvent.source = missingIds.length === 0 ? "in-mem" : "persist"; + } + + if (missingIds.length > 0) { + const loadedRows: Row[] = []; + for (const batch of chunks(missingIds, entityLoadBatchSize)) { + loadedRows.push( + ...(yield* primary.intervalScan( + table, + table.idIndexName, + batch.map((id) => ({ eq: [{ col: "id", val: id }] })), + )), + ); + } + + tableState.indexes.upsert(loadedRows); + tableState.byId.upsert(loadedEntities(loadedRows)); + const loadedIds = new Set(loadedRows.map((row) => row.id)); + const staleIds = missingIds.filter((id) => !loadedIds.has(id)); + if (staleIds.length > 0) { + tableState.indexes.delete(staleIds); + tableState.byId.delete(staleIds); + } + } + + const rows = ids.flatMap((id) => { + const entry = entityById(tableState.byId, id); + return entry?.loaded ? [entry.row as ExtractSchema] : []; + }); + return rows; +} + +/** + * A persistent/in-memory runtime that preloads every table index as + * key-to-entity-id entries while loading entity rows only when a scan needs + * them. Hydrated rows are retained in a unique id cache and shared by every + * index. + */ +export class PreloadedHybridDB implements HyperDB { + readonly primary: DB; + traits: Trait[]; + private state: PreloadedHybridDBState; + + constructor(primary: DB, options: PreloadedHybridDBOptions = {}) { + this.primary = primary; + this.traits = options.traits ?? []; + this.state = createState(); + } + + withTraits(...traits: Trait[]): HyperDB { + const db = new PreloadedHybridDB(this.primary, { + traits: [...this.traits, ...traits], + }); + db.state = this.state; + return db; + } + + getTraits(): Trait[] { + return [...this.traits, ...this.primary.getTraits()]; + } + + getId(): string { + return this.primary.getId(); + } + + getDBName(): string | undefined { + return this.primary.getDBName?.(); + } + + getTracer(): HyperDBTracerOption | undefined { + return this.primary.getTracer?.(); + } + + getOptions(): CodecOptions { + return this.primary.getOptions?.() ?? DEFAULT_CODEC_OPTIONS; + } + + canUseReadonlyTransactionsForSelectors(): boolean { + return this.primary.canUseReadonlyTransactionsForSelectors(); + } + + private delegatePrimary(): HyperDB { + return this.traits.length > 0 + ? this.primary.withTraits(...this.traits) + : this.primary; + } + + *loadTables(tables: TableDefinition[]): Generator { + const release = yield* acquireLock(this.state.lock); + try { + yield* this.delegatePrimary().loadTables(tables); + const data = this.state.data; + for (const table of tables) { + const rows = yield* this.primary.scanAll(table); + const byId = new HashIndex({ + name: table.idIndexName, + unique: true, + }); + byId.insert(entityPointers(rows)); + data.tables.set(table.tableName, { + indexes: new PreloadedTableIndexes(table, rows as Row[]), + byId, + }); + } + } finally { + release(); + } + } + + *preloadTables( + specs: TSpecs & ValidateHybridPreloadTableSpecs, + ): Generator { + for (const spec of specs) this.state.data.get(spec.table); + } + + *intervalScan< + TTable extends TableDefinition, + K extends keyof ExtractIndexes, + >( + table: TTable, + indexName: K, + clauses: WhereClause[], + selectOptions?: SelectOptions, + ): Generator[]> { + const release = yield* acquireLock(this.state.lock); + try { + const primary = withDriverTraceContextTrait( + this.delegatePrimary(), + getDriverTraceContextForDB(this), + ); + return yield* scanPreloaded( + this, + primary, + this.state.data, + table, + indexName, + clauses, + selectOptions, + ); + } finally { + release(); + } + } + + *insert( + table: TTable, + records: ExtractSchema[], + ): Generator { + const release = yield* acquireLock(this.state.lock); + try { + const primary = this.delegatePrimary(); + if (usesExternalStorageMerge(this)) { + yield* primary.upsert(table, records); + } else { + yield* primary.insert(table, records); + } + const state = this.state.data.get(table); + state.indexes.upsert(records as Row[]); + state.byId.upsert(loadedEntities(records as Row[])); + } finally { + release(); + } + } + + *upsert( + table: TTable, + records: ExtractSchema[], + ): Generator { + const release = yield* acquireLock(this.state.lock); + try { + yield* this.delegatePrimary().upsert(table, records); + const state = this.state.data.get(table); + state.indexes.upsert(records as Row[]); + state.byId.upsert(loadedEntities(records as Row[])); + } finally { + release(); + } + } + + *delete( + table: TTable, + ids: string[], + ): Generator { + const release = yield* acquireLock(this.state.lock); + try { + yield* this.delegatePrimary().delete(table, ids); + const state = this.state.data.get(table); + state.indexes.delete(ids); + state.byId.delete(ids); + } finally { + release(); + } + } + + *beginTx(mode: DBTransactionMode = "readwrite"): Generator { + const release = yield* acquireLock(this.state.lock); + try { + const primary = withDriverTraceContextTrait( + this.delegatePrimary(), + getDriverTraceContextForDB(this), + ); + const primaryTx = yield* primary.beginTx(mode); + const data = + mode === "readonly" ? this.state.data : this.state.data.fork(); + return new PreloadedHybridDBTx(this, primaryTx, data, mode, release); + } catch (error) { + release(); + throw error; + } + } + + commitData(data: PreloadedHybridData): void { + data.materializeFork(); + this.state.data = data; + } +} + +type PreloadedHybridTxState = { + committed: RefVar; + rollbacked: RefVar; + counter: RefVar; + release: () => void; +}; + +class PreloadedHybridDBTx implements HyperDBTx { + private readonly state: PreloadedHybridTxState; + private readonly owner: PreloadedHybridDB; + private readonly primaryTx: HyperDBTx; + private readonly data: PreloadedHybridData; + private readonly mode: DBTransactionMode; + private readonly traits: Trait[]; + + constructor( + owner: PreloadedHybridDB, + primaryTx: HyperDBTx, + data: PreloadedHybridData, + mode: DBTransactionMode, + release: () => void, + traits: Trait[] = [], + state?: PreloadedHybridTxState, + ) { + this.owner = owner; + this.primaryTx = primaryTx; + this.data = data; + this.mode = mode; + this.traits = traits; + this.state = state ?? { + committed: refVar(false), + rollbacked: refVar(false), + counter: refVar(1), + release, + }; + } + + withTraits(...traits: Trait[]): HyperDBTx { + return new PreloadedHybridDBTx( + this.owner, + this.primaryTx, + this.data, + this.mode, + this.state.release, + [...this.traits, ...traits], + this.state, + ); + } + + getTraits(): Trait[] { + return [...this.traits, ...this.owner.getTraits()]; + } + + getId(): string { + return this.owner.getId(); + } + + getDBName(): string | undefined { + return this.owner.getDBName?.(); + } + + getTracer(): HyperDBTracerOption | undefined { + return this.owner.getTracer?.(); + } + + getOptions(): CodecOptions { + return this.owner.getOptions(); + } + + canUseReadonlyTransactionsForSelectors(): boolean { + return false; + } + + private delegatePrimary(): HyperDBTx { + return this.traits.length > 0 + ? (this.primaryTx.withTraits(...this.traits) as HyperDBTx) + : this.primaryTx; + } + + *loadTables(): Generator { + throw new Error("Not supported"); + } + + *preloadTables( + _specs: TSpecs & ValidateHybridPreloadTableSpecs, + ): Generator { + throw new Error( + "preloadTables is not supported inside PreloadedHybridDB transactions", + ); + } + + *beginTx(): Generator { + this.throwIfDone(); + this.state.counter.val++; + return this; + } + + *intervalScan< + TTable extends TableDefinition, + K extends keyof ExtractIndexes, + >( + table: TTable, + indexName: K, + clauses: WhereClause[], + selectOptions?: SelectOptions, + ): Generator[]> { + this.throwIfDone(); + return yield* scanPreloaded( + this, + this.delegatePrimary(), + this.data, + table, + indexName, + clauses, + selectOptions, + ); + } + + *insert( + table: TTable, + records: ExtractSchema[], + ): Generator { + this.throwIfDone(); + this.throwIfReadonly(); + const primary = this.delegatePrimary(); + if (usesExternalStorageMerge(this)) { + yield* primary.upsert(table, records); + } else { + yield* primary.insert(table, records); + } + const state = this.data.get(table); + state.indexes.upsert(records as Row[]); + state.byId.upsert(loadedEntities(records as Row[])); + } + + *upsert( + table: TTable, + records: ExtractSchema[], + ): Generator { + this.throwIfDone(); + this.throwIfReadonly(); + yield* this.delegatePrimary().upsert(table, records); + const state = this.data.get(table); + state.indexes.upsert(records as Row[]); + state.byId.upsert(loadedEntities(records as Row[])); + } + + *delete( + table: TTable, + ids: string[], + ): Generator { + this.throwIfDone(); + this.throwIfReadonly(); + yield* this.delegatePrimary().delete(table, ids); + const state = this.data.get(table); + state.indexes.delete(ids); + state.byId.delete(ids); + } + + *commit(): Generator { + this.throwIfDone(); + this.state.counter.val--; + if (this.state.counter.val > 0) return; + + try { + yield* this.delegatePrimary().commit(); + if (this.mode === "readwrite") this.owner.commitData(this.data); + this.state.committed.val = true; + } catch (error) { + if (this.mode === "readwrite") this.data.discardFork(); + this.state.rollbacked.val = true; + throw error; + } finally { + this.state.release(); + } + } + + *rollback(): Generator { + this.throwIfDone(); + let rollbackFailed = false; + let rollbackError: unknown; + try { + yield* this.delegatePrimary().rollback(); + } catch (error) { + rollbackFailed = true; + rollbackError = error; + } finally { + if (this.mode === "readwrite") this.data.discardFork(); + this.state.rollbacked.val = true; + this.state.release(); + } + if (rollbackFailed) throw rollbackError; + } + + private throwIfDone(): void { + if (this.state.committed.val || this.state.rollbacked.val) { + throw new Error("Transaction already finished"); + } + } + + private throwIfReadonly(): void { + if (this.mode === "readonly") { + throw new Error("Cannot write through a readonly transaction"); + } + } +} diff --git a/packages/hyperdb/src/hyperdb/runtime/subscribable-db.ts b/packages/hyperdb/src/hyperdb/runtime/subscribable-db.ts index 4740b35..55ccf15 100644 --- a/packages/hyperdb/src/hyperdb/runtime/subscribable-db.ts +++ b/packages/hyperdb/src/hyperdb/runtime/subscribable-db.ts @@ -36,6 +36,21 @@ import type { InsertOp, UpsertOp, DeleteOp, Op } from "./ops"; type Subscriber = (op: Op[], traits: Trait[], revision: number) => void; +const notifySubscribers = ( + subscribers: Subscriber[], + operations: Op[], + traits: Trait[], + revision: number, +) => { + for (const subscriber of [...subscribers]) { + try { + subscriber(operations, traits, revision); + } catch (error) { + console.error(error); + } + } +}; + type AfterInsertSub = ( db: HyperDB, table: TableDefinition, @@ -546,14 +561,12 @@ export class SubscribableDBTx implements HyperDBTx { const traits = this.getTraits(); const revision = this.subDb.incrementRevision(); - const subscribers = [...this.subDb.subscribers]; - for (const subscriber of subscribers) { - try { - subscriber(this.operations, traits, revision); - } catch (error) { - console.error(error); - } - } + notifySubscribers( + this.subDb.subscribers, + this.operations, + traits, + revision, + ); } throwIfDone() { diff --git a/packages/hyperdb/src/hyperdb/structures/hash-index.test.ts b/packages/hyperdb/src/hyperdb/structures/hash-index.test.ts new file mode 100644 index 0000000..747b6df --- /dev/null +++ b/packages/hyperdb/src/hyperdb/structures/hash-index.test.ts @@ -0,0 +1,59 @@ +import { describe, expect, it } from "vitest"; +import { HashIndex } from "./hash-index"; + +describe("HashIndex", () => { + it("stores arbitrary leaf values and enforces uniqueness", () => { + const index = new HashIndex({ name: "bySlug", unique: true }); + index.insert([{ key: "first", id: "1", value: "1" }]); + + expect(index.scan(["first"])).toEqual(["1"]); + expect(() => index.upsert([{ key: "first", id: "2", value: "2" }])).toThrow( + "Unique hash index bySlug already has value for record 1", + ); + }); + + it("supports non-unique buckets", () => { + const index = new HashIndex({ name: "byTitle", unique: false }); + index.insert([ + { key: "same", id: "1", value: "1" }, + { key: "same", id: "2", value: "2" }, + ]); + + expect(index.scan(["same"])).toEqual(["1", "2"]); + }); + + it("commits and rolls back copy-on-write transactions", () => { + const index = new HashIndex<{ id: string; value: number }>({ + name: "byId", + unique: true, + }); + index.insert([{ key: "1", id: "1", value: { id: "1", value: 1 } }]); + + const rollback = index.tx(); + rollback.upsert([{ key: "1", id: "1", value: { id: "1", value: 2 } }]); + expect(rollback.scan(["1"])).toEqual([{ id: "1", value: 2 }]); + rollback.rollback(); + expect(index.scan(["1"])).toEqual([{ id: "1", value: 1 }]); + + const commit = index.tx(); + commit.upsert([{ key: "1", id: "1", value: { id: "1", value: 3 } }]); + expect(commit.commit()).toBe(index); + expect(index.scan(["1"])).toEqual([{ id: "1", value: 3 }]); + }); + + it("supports swapping unique keys in one upsert", () => { + const index = new HashIndex({ name: "bySlug", unique: true }); + index.insert([ + { key: "first", id: "1", value: "1" }, + { key: "second", id: "2", value: "2" }, + ]); + + index.upsert([ + { key: "second", id: "1", value: "1" }, + { key: "first", id: "2", value: "2" }, + ]); + + expect(index.scan(["first"])).toEqual(["2"]); + expect(index.scan(["second"])).toEqual(["1"]); + }); +}); diff --git a/packages/hyperdb/src/hyperdb/structures/hash-index.ts b/packages/hyperdb/src/hyperdb/structures/hash-index.ts new file mode 100644 index 0000000..11a1613 --- /dev/null +++ b/packages/hyperdb/src/hyperdb/structures/hash-index.ts @@ -0,0 +1,279 @@ +import type { SelectOptions, Value } from "../core/primitives"; + +export type HashIndexEntry = { + key: Value; + id: string; + value: T; +}; + +export type HashIndexOptions = { + name: string; + unique: boolean; +}; + +type HashKey = string; + +function bytesOfHashValue(value: ArrayBuffer | ArrayBufferView): number[] { + if (value instanceof ArrayBuffer) { + return Array.from(new Uint8Array(value)); + } + return Array.from( + new Uint8Array(value.buffer, value.byteOffset, value.byteLength), + ); +} + +export function hashIndexKey(value: Value): HashKey { + if (value === null) return "null:"; + if (typeof value === "string") return `string:${value}`; + if (typeof value === "number") { + return `number:${Object.is(value, -0) ? 0 : value}`; + } + if (typeof value === "bigint") return `bigint:${value.toString()}`; + if (typeof value === "boolean") return `boolean:${value ? "1" : "0"}`; + return `bytes:${bytesOfHashValue(value) + .map((byte) => byte.toString(16).padStart(2, "0")) + .join("")}`; +} + +export interface HashIndexView { + readonly name: string; + readonly unique: boolean; + + scan(keys: Iterable, options?: SelectOptions): T[]; + validateInsert( + entries: readonly HashIndexEntry[], + replacingIds: ReadonlySet, + ): void; + insert(entries: readonly HashIndexEntry[]): void; + upsert(entries: readonly HashIndexEntry[]): void; + delete(ids: readonly string[]): void; +} + +function validateBatch( + index: HashIndexView, + entries: readonly HashIndexEntry[], + replacingIds: ReadonlySet, + bucket: (key: HashKey) => ReadonlyMap | undefined, +): void { + const keysById = new Map(); + const idsByKey = new Map(); + + for (const entry of entries) { + const key = hashIndexKey(entry.key); + const previousKey = keysById.get(entry.id); + if (previousKey !== undefined && previousKey !== key) { + throw new Error( + `Hash index ${index.name} received duplicate id ${entry.id}`, + ); + } + keysById.set(entry.id, key); + + if (!index.unique) continue; + + const batchId = idsByKey.get(key); + if (batchId !== undefined && batchId !== entry.id) { + throw new Error( + `Unique hash index ${index.name} already has value for record ${batchId}`, + ); + } + idsByKey.set(key, entry.id); + + const existingValues = bucket(key); + if (!existingValues) continue; + for (const existingId of existingValues.keys()) { + if (existingId === entry.id || replacingIds.has(existingId)) continue; + throw new Error( + `Unique hash index ${index.name} already has value for record ${existingId}`, + ); + } + } +} + +function scanBuckets( + keys: Iterable, + options: SelectOptions, + bucket: (key: HashKey) => ReadonlyMap | undefined, +): T[] { + if (options.limit !== undefined && options.limit <= 0) return []; + + const results: T[] = []; + const seenKeys = new Set(); + for (const value of keys) { + const key = hashIndexKey(value); + if (seenKeys.has(key)) continue; + seenKeys.add(key); + + const values = bucket(key); + if (!values) continue; + for (const item of values.values()) { + results.push(item); + if (options.limit !== undefined && results.length >= options.limit) { + return results; + } + } + } + return results; +} + +export class HashIndex implements HashIndexView { + readonly name: string; + readonly unique: boolean; + readonly buckets = new Map>(); + readonly keysById = new Map(); + + constructor(options: HashIndexOptions) { + this.name = options.name; + this.unique = options.unique; + } + + scan(keys: Iterable, options: SelectOptions = {}): T[] { + return scanBuckets(keys, options, (key) => this.buckets.get(key)); + } + + values(): T[] { + return Array.from(this.buckets.values()).flatMap((bucket) => [ + ...bucket.values(), + ]); + } + + validateInsert( + entries: readonly HashIndexEntry[], + replacingIds: ReadonlySet, + ): void { + validateBatch(this, entries, replacingIds, (key) => this.buckets.get(key)); + } + + insert(entries: readonly HashIndexEntry[]): void { + for (const entry of entries) { + const key = hashIndexKey(entry.key); + const bucket = this.buckets.get(key) ?? new Map(); + bucket.set(entry.id, entry.value); + this.buckets.set(key, bucket); + this.keysById.set(entry.id, key); + } + } + + upsert(entries: readonly HashIndexEntry[]): void { + const replacingIds = new Set(entries.map((entry) => entry.id)); + this.validateInsert(entries, replacingIds); + this.delete([...replacingIds]); + this.insert(entries); + } + + delete(ids: readonly string[]): void { + for (const id of ids) { + const key = this.keysById.get(id); + if (key === undefined) continue; + const bucket = this.buckets.get(key); + bucket?.delete(id); + if (bucket?.size === 0) this.buckets.delete(key); + this.keysById.delete(id); + } + } + + tx(): HashIndexTx { + return new HashIndexTx(this); + } +} + +export class HashIndexTx implements HashIndexView { + readonly name: string; + readonly unique: boolean; + private readonly original: HashIndex; + private readonly txBuckets = new Map>(); + private readonly txKeysById = new Map(); + private finished = false; + + constructor(original: HashIndex) { + this.original = original; + this.name = original.name; + this.unique = original.unique; + } + + scan(keys: Iterable, options: SelectOptions = {}): T[] { + this.throwIfFinished("scan"); + return scanBuckets(keys, options, (key) => this.currentBucket(key)); + } + + validateInsert( + entries: readonly HashIndexEntry[], + replacingIds: ReadonlySet, + ): void { + this.throwIfFinished("validate inserts"); + validateBatch(this, entries, replacingIds, (key) => + this.currentBucket(key), + ); + } + + insert(entries: readonly HashIndexEntry[]): void { + this.throwIfFinished("insert"); + for (const entry of entries) { + const key = hashIndexKey(entry.key); + const bucket = this.writableBucket(key); + bucket.set(entry.id, entry.value); + this.txKeysById.set(entry.id, key); + } + } + + upsert(entries: readonly HashIndexEntry[]): void { + const replacingIds = new Set(entries.map((entry) => entry.id)); + this.validateInsert(entries, replacingIds); + this.delete([...replacingIds]); + this.insert(entries); + } + + delete(ids: readonly string[]): void { + this.throwIfFinished("delete"); + for (const id of ids) { + const key = this.txKeysById.has(id) + ? this.txKeysById.get(id) + : this.original.keysById.get(id); + if (key === undefined) continue; + this.writableBucket(key).delete(id); + this.txKeysById.set(id, undefined); + } + } + + commit(): HashIndex { + this.throwIfFinished("commit"); + this.finished = true; + for (const [key, bucket] of this.txBuckets) { + if (bucket.size === 0) this.original.buckets.delete(key); + else this.original.buckets.set(key, bucket); + } + for (const [id, key] of this.txKeysById) { + if (key === undefined) this.original.keysById.delete(id); + else this.original.keysById.set(id, key); + } + return this.original; + } + + rollback(): HashIndex { + this.throwIfFinished("rollback"); + this.finished = true; + return this.original; + } + + private currentBucket(key: HashKey): ReadonlyMap | undefined { + return this.txBuckets.has(key) + ? this.txBuckets.get(key) + : this.original.buckets.get(key); + } + + private writableBucket(key: HashKey): Map { + const existing = this.txBuckets.get(key); + if (existing) return existing; + + const bucket = new Map(this.original.buckets.get(key)); + this.txBuckets.set(key, bucket); + return bucket; + } + + private throwIfFinished(operation: string): void { + if (this.finished) { + throw new Error( + `Can't ${operation} after hash index transaction finished`, + ); + } + } +}