Skip to content

fix(dataset): include tombstone xact_ids in Dataset.version() via audit_log BTQL query #2508

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Mohd Quamar Tyagi (Tyagiquamar) wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(dataset): include tombstone xact_ids in Dataset.version() via audit_log BTQL query #2508

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
148 changes: 148 additions & 0 deletions js/src/logger.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
60 changes: 58 additions & 2 deletions js/src/logger.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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({
Expand Down