From 882b383e29daf1346ca39755170a04a21ec8b1d9 Mon Sep 17 00:00:00 2001 From: Tyagiquamar Date: Mon, 21 Sep 2026 19:32:12 +0530 Subject: [PATCH] fix(dataset): include tombstone xact_ids in Dataset.version() via audit_log BTQL query Previously, Dataset.version() delegated to super.version() (ObjectFetcher), which queried BTQL without audit_log: true. This meant deleted records' transaction IDs were invisible to the version query, so calling dataset.createSnapshot() after deletions would use a stale version that predated those deletions. Override Dataset.version() to issue its own BTQL query with audit_log: true, scanning all rows including tombstones, and returning the highest _xact_id seen. Add tests covering: - the audit-log BTQL query structure and max-version extraction - createSnapshot() picking up the correct post-deletion version --- js/src/logger.test.ts | 148 ++++++++++++++++++++++++++++++++++++++++++ js/src/logger.ts | 60 ++++++++++++++++- 2 files changed, 206 insertions(+), 2 deletions(-) diff --git a/js/src/logger.test.ts b/js/src/logger.test.ts index 84781b5b8..bf23a2265 100644 --- a/js/src/logger.test.ts +++ b/js/src/logger.test.ts @@ -1408,6 +1408,154 @@ test("dataset.version preserves pinned-version fast path", async () => { vi.restoreAllMocks(); }); +test("dataset.version queries BTQL audit log and includes tombstone transaction IDs", async () => { + const state = await _exportsForTestingOnly.simulateLoginForTests(); + const login = vi.spyOn(state, "login").mockResolvedValue(state as any); + const postJson = vi.spyOn(state.appConn(), "post_json").mockResolvedValue({ + project: { + id: "00000000-0000-0000-0000-000000000001", + name: "test-project", + }, + dataset: { + id: "00000000-0000-0000-0000-000000000002", + name: "test-dataset", + }, + }); + + let btqlBody: unknown; + const postApi = vi + .spyOn(state.apiConn(), "post") + .mockImplementation(async (_path, body) => { + btqlBody = body; + return new Response( + JSON.stringify({ + data: [ + { _xact_id: "1000197874935204590" }, + { _xact_id: "1000197874935204592" }, + { _xact_id: "1000197874946873171" }, + ], + }), + { + status: 200, + headers: { "Content-Type": "application/json" }, + }, + ); + }); + + try { + const dataset = initDataset({ + project: "test-project", + dataset: "test-dataset", + state, + }); + + const version = await dataset.version(); + expect(version).toBe("1000197874946873171"); + expect(btqlBody).toEqual( + expect.objectContaining({ + audit_log: true, + query: expect.objectContaining({ + from: { + op: "function", + name: { + op: "ident", + name: ["dataset"], + }, + args: [ + { + op: "literal", + value: "00000000-0000-0000-0000-000000000002", + }, + ], + }, + }), + }), + ); + } finally { + postJson.mockRestore(); + postApi.mockRestore(); + login.mockRestore(); + _exportsForTestingOnly.simulateLogoutForTests(); + vi.restoreAllMocks(); + } +}); + +test("dataset.createSnapshot uses latest transaction id from audit log after deletes", async () => { + const state = await _exportsForTestingOnly.simulateLoginForTests(); + const login = vi.spyOn(state, "login").mockResolvedValue(state as any); + const postJson = vi + .spyOn(state.appConn(), "post_json") + .mockResolvedValueOnce({ + project: { + id: "00000000-0000-0000-0000-000000000001", + name: "test-project", + }, + dataset: { + id: "00000000-0000-0000-0000-000000000002", + name: "test-dataset", + }, + }) + .mockResolvedValueOnce({ + dataset_snapshot: { + id: "00000000-0000-0000-0000-000000000005", + dataset_id: "00000000-0000-0000-0000-000000000002", + name: "after-delete-snapshot", + description: null, + xact_id: "1000197874946873171", + created: "2026-03-31T00:00:00.000Z", + }, + found_existing: false, + }); + + const postApi = vi + .spyOn(state.apiConn(), "post") + .mockImplementation(async () => { + return new Response( + JSON.stringify({ + data: [ + { _xact_id: "1000197874935204590" }, + { _xact_id: "1000197874946873171" }, + ], + }), + { + status: 200, + headers: { "Content-Type": "application/json" }, + }, + ); + }); + + try { + const dataset = initDataset({ + project: "test-project", + dataset: "test-dataset", + state, + }); + + await expect( + dataset.createSnapshot({ + name: "after-delete-snapshot", + }), + ).resolves.toMatchObject({ + id: "00000000-0000-0000-0000-000000000005", + xact_id: "1000197874946873171", + }); + + expect(postJson).toHaveBeenNthCalledWith(2, "api/dataset_snapshot/register", { + dataset_id: "00000000-0000-0000-0000-000000000002", + dataset_snapshot_name: "after-delete-snapshot", + description: undefined, + xact_id: "1000197874946873171", + update: undefined, + }); + } finally { + postJson.mockRestore(); + postApi.mockRestore(); + login.mockRestore(); + _exportsForTestingOnly.simulateLogoutForTests(); + vi.restoreAllMocks(); + } +}); + test("dataset.createSnapshot forwards update when requested", async () => { const state = await _exportsForTestingOnly.simulateLoginForTests(); vi.spyOn(state, "login").mockResolvedValue(state as any); diff --git a/js/src/logger.ts b/js/src/logger.ts index 5eb438ba9..8ad47c181 100644 --- a/js/src/logger.ts +++ b/js/src/logger.ts @@ -8782,8 +8782,64 @@ export class Dataset< if (pinnedVersion !== undefined) { return pinnedVersion; } - await this.getState(); - return await super.version(options); + const state = await this.getState(); + const objectId = await this.id; + const batchLimit = options?.batchSize ?? DEFAULT_FETCH_BATCH_SIZE; + let cursor = undefined; + let maxVersion: string | undefined = undefined; + let iterations = 0; + while (true) { + const resp = await state.apiConn().post( + `btql`, + { + query: { + select: [ + { + op: "ident", + name: [TRANSACTION_ID_FIELD], + }, + ], + from: { + op: "function", + name: { + op: "ident", + name: ["dataset"], + }, + args: [ + { + op: "literal", + value: objectId, + }, + ], + }, + cursor, + limit: batchLimit, + }, + audit_log: true, + use_columnstore: false, + brainstore_realtime: true, + query_source: `js_sdk_dataset_version`, + }, + { headers: { "Accept-Encoding": "gzip" } }, + BTQL_HTTP_RETRIES, + ); + const respJson = await resp.json(); + for (const record of respJson.data ?? []) { + const xactId = String(record[TRANSACTION_ID_FIELD] ?? "0"); + if (maxVersion === undefined || xactId > maxVersion) { + maxVersion = xactId; + } + } + if (!respJson.cursor) { + break; + } + cursor = respJson.cursor; + iterations++; + if (iterations > MAX_BTQL_ITERATIONS) { + throw new Error("Too many BTQL iterations"); + } + } + return maxVersion; } private validateEvent({