diff --git a/.changeset/search-cursor-pagination.md b/.changeset/search-cursor-pagination.md new file mode 100644 index 0000000000..2ccaae4af8 --- /dev/null +++ b/.changeset/search-cursor-pagination.md @@ -0,0 +1,6 @@ +--- +'@hyperdx/api': minor +'@hyperdx/common-utils': patch +--- + +Add cursor pagination to the external API v2 search endpoint: send `cursor: "start"`, then pass back each `nextCursor` until it is `null`. Cursors are signed with `EXPRESS_SESSION_SECRET`: replicas need a shared value, and rotating it invalidates outstanding cursors. diff --git a/packages/api/openapi.json b/packages/api/openapi.json index f6ea5912f6..9036893bc9 100644 --- a/packages/api/openapi.json +++ b/packages/api/openapi.json @@ -4812,7 +4812,13 @@ "minimum": 0, "maximum": 10000, "default": 0, - "description": "Number of rows to skip (best-effort offset pagination). Default is\n0, max is 10000. Offset pagination is non-deterministic when\nmultiple rows share the same timestamp; for reliable deep paging\nfilter by the last Timestamp value returned in the previous page\ninstead of using a large offset.\n" + "description": "Number of rows to skip (best-effort offset pagination). Default is\n0, max is 10000. Offset pagination is non-deterministic when\nmultiple rows share the same timestamp. Prefer cursor for paging.\n" + }, + "cursor": { + "type": "string", + "maxLength": 1024, + "example": "start", + "description": "Send the literal string \"start\" to begin cursor paging, then pass\nback the nextCursor from each response. When set, offset is\nignored; omit cursor entirely to use offset-based paging.\nAs with offset, rows that share a timestamp can be skipped or\nrepeated between pages unless orderBy breaks the tie.\n\nKeep requesting pages until nextCursor is null. A page that is\nempty or returns fewer rows than maxResults does NOT mean the\nresults are exhausted.\n" } } }, @@ -4834,6 +4840,12 @@ "rows": { "type": "integer", "description": "Number of rows in this response (not total matching rows)." + }, + "nextCursor": { + "type": "string", + "nullable": true, + "example": "eyJ2IjoxLCJ3IjoyLCJvIjo1MDAsImYiOiJhMWIyYzNkNGU1ZjZhN2I4In0", + "description": "Cursor for the next page, or null when the walk is complete or the request did not include a cursor. Pass it back as cursor in the next request to fetch the next page. This is an opaque value; no guarantees are made about its content.\n" } } }, diff --git a/packages/api/src/routers/external-api/__tests__/search.int.test.ts b/packages/api/src/routers/external-api/__tests__/search.int.test.ts index 29add43ec0..f9f349c121 100644 --- a/packages/api/src/routers/external-api/__tests__/search.int.test.ts +++ b/packages/api/src/routers/external-api/__tests__/search.int.test.ts @@ -1,3 +1,4 @@ +import type { QuerySettings } from '@hyperdx/common-utils/dist/types'; import { SourceKind } from '@hyperdx/common-utils/dist/types'; import mongoose from 'mongoose'; import request from 'supertest'; @@ -14,6 +15,11 @@ import Connection from '@/models/connection'; import type { ISource } from '@/models/source'; import { Source } from '@/models/source'; import type { IUser } from '@/models/user'; +import { + encodeSearchCursor, + fingerprintQuery, +} from '@/routers/external-api/v2/utils/searchCursor'; +import logger from '@/utils/logger'; const DEFAULT_END_TIME = Math.floor(Date.now() / 60000) * 60000; const DEFAULT_START_TIME = DEFAULT_END_TIME - 3600 * 1000; @@ -557,4 +563,614 @@ describe('External API v2 Search', () => { expect(res.status).toBe(404); expect(res.body).toHaveProperty('message'); }); + + // Each test seeds its own rows: afterEach calls server.clearDBs(). + const seedCursorRows = async () => { + // 40 rows a minute apart, plus 3 on the newest 15m window's boundary. + const boundary = DEFAULT_END_TIME - 15 * 60 * 1000; + await bulkInsertLogs([ + ...Array.from({ length: 40 }, (_, i) => ({ + ServiceName: 'cursor-test', + Timestamp: new Date(DEFAULT_END_TIME - i * 60 * 1000), + SeverityText: 'INFO', + Body: `row-${i}`, + })), + ...Array.from({ length: 3 }, (_, i) => ({ + ServiceName: 'cursor-test', + Timestamp: new Date(boundary), + SeverityText: 'INFO', + Body: `boundary-${i}`, + })), + ]); + }; + + const cursorOffset = (cursor: string) => + JSON.parse(Buffer.from(cursor.split('.')[0], 'base64url').toString()).o; + + const cursorBody = () => ({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + where: 'ServiceName:"cursor-test"', + select: 'Timestamp,Body', + maxResults: 5, + }); + + it('walks every row exactly once with cursor pagination', async () => { + await seedCursorRows(); + + const seen: string[] = []; + let cursor: string | null | undefined; + let pages = 0; + for (; pages < 50; pages++) { + const res = await search({ + ...cursorBody(), + cursor: cursor ?? 'start', + }).expect(200); + seen.push(...res.body.data.map((r: Record) => r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + } + + expect(pages).toBeLessThan(49); + expect(cursor).toBeNull(); + expect(seen).toHaveLength(43); + expect(new Set(seen).size).toBe(43); + }); + + it.each([ + ['omitted', {}], + ['empty', { endTime: '' }], + ])( + 'pages with a defaulted endTime (%s), which resolves to a fresh now each call', + async (_label, endTimeField) => { + await seedCursorRows(); + + // An omitted or empty endTime resolves to a fresh `now` on each request. + const body = { + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + ...endTimeField, + where: 'ServiceName:"cursor-test"', + select: 'Timestamp,Body', + maxResults: 5, + }; + + const seen: string[] = []; + let cursor: string | null | undefined; + let pages = 0; + for (; pages < 50; pages++) { + const res = await search({ ...body, cursor: cursor ?? 'start' }).expect( + 200, + ); + seen.push(...res.body.data.map((r: Record) => r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + } + + expect(cursor).toBeNull(); + expect(seen).toHaveLength(43); + expect(new Set(seen).size).toBe(43); + }, + ); + + it.each([ + ['on a whole second', 0], + ['mid-second', 500], + ])( + 'returns each row once around a window boundary %s with a second-precision timestamp column', + async (_label, boundaryMs) => { + const coarseSource = await Source.create({ + kind: SourceKind.Log, + team: logSource.team, + from: logSource.from, + defaultTableSelectExpression: 'Timestamp, SeverityText, Body', + timestampValueExpression: 'toDateTime(Timestamp), Timestamp', + connection: logSource.connection, + name: 'Logs (second-precision)', + }); + + // The first descending window boundary sits 15 minutes before endTime. + const endTime = DEFAULT_END_TIME + boundaryMs; + const boundary = endTime - 15 * 60 * 1000; + await bulkInsertLogs( + [-300, 0, 300].map(deltaMs => ({ + ServiceName: 'cursor-boundary-test', + Timestamp: new Date(boundary + deltaMs), + SeverityText: 'INFO', + Body: `at-${deltaMs}`, + })), + ); + + const body = { + sourceId: coarseSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(endTime), + where: 'ServiceName:"cursor-boundary-test"', + select: 'Timestamp,Body', + orderBy: 'Timestamp DESC', + maxResults: 5, + }; + + const seen: string[] = []; + let cursor: string | null | undefined; + for (let pages = 0; pages < 50; pages++) { + const res = await search({ ...body, cursor: cursor ?? 'start' }).expect( + 200, + ); + seen.push(...res.body.data.map((r: Record) => r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + } + + expect(cursor).toBeNull(); + expect(seen.sort()).toEqual(['at--300', 'at-0', 'at-300']); + }, + ); + + it('starts a cursor walk over a range longer than a year', async () => { + await seedCursorRows(); + + const res = await search({ + ...cursorBody(), + startTime: iso(DEFAULT_END_TIME - 400 * 24 * 60 * 60 * 1000), + cursor: 'start', + }).expect(200); + expect(res.body.data).toHaveLength(5); + expect(res.body.nextCursor).toBeTruthy(); + }); + + it('pages past 10,000 rows in a range shorter than the first window', async () => { + const rangeStart = DEFAULT_END_TIME - 10 * 60 * 1000; + await bulkInsertLogs( + Array.from({ length: 10_050 }, (_, i) => ({ + ServiceName: 'cursor-dense-test', + Timestamp: new Date(rangeStart + i * 50), + SeverityText: 'INFO', + Body: `dense-${i}`, + })), + ); + const body = { + ...cursorBody(), + startTime: iso(rangeStart), + where: 'ServiceName:"cursor-dense-test"', + maxResults: 2000, + }; + + const seen = new Set(); + const offsets: number[] = []; + let cursor: string | null | undefined; + for (let pages = 0; pages < 20; pages++) { + const res = await search({ ...body, cursor: cursor ?? 'start' }).expect( + 200, + ); + res.body.data.forEach((r: Record) => seen.add(r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + offsets.push(cursorOffset(cursor)); + } + expect(cursor).toBeNull(); + expect(seen.size).toBe(10_050); + // Rows are 50ms apart, so a second holds 20 of them. + expect(Math.max(...offsets)).toBeLessThanOrEqual(20); + }); + + it.each(['Timestamp DESC', 'Timestamp ASC'])( + 'returns every row once when one second holds more rows than a page (%s)', + async orderBy => { + const second = DEFAULT_END_TIME - 5 * 60 * 1000; + await bulkInsertLogs( + Array.from({ length: 900 }, (_, i) => ({ + ServiceName: 'cursor-burst-test', + Timestamp: new Date(second + i), + SeverityText: 'INFO', + Body: `burst-${i}`, + })), + ); + const body = { + ...cursorBody(), + where: 'ServiceName:"cursor-burst-test"', + orderBy, + maxResults: 300, + }; + + const seen: string[] = []; + let cursor: string | null | undefined; + for (let pages = 0; pages < 20; pages++) { + const res = await search({ ...body, cursor: cursor ?? 'start' }).expect( + 200, + ); + seen.push(...res.body.data.map((r: Record) => r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + } + expect(cursor).toBeNull(); + expect(seen).toHaveLength(900); + expect(new Set(seen).size).toBe(900); + }, + ); + + it('walks every row exactly once in ascending order', async () => { + await seedCursorRows(); + // seedCursorRows puts four rows on one Timestamp; Body breaks the tie. + + const seen: string[] = []; + let cursor: string | null | undefined; + for (let pages = 0; pages < 50; pages++) { + const res = await search({ + ...cursorBody(), + orderBy: 'Timestamp ASC, Body ASC', + cursor: cursor ?? 'start', + }).expect(200); + seen.push(...res.body.data.map((r: Record) => r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + } + expect(cursor).toBeNull(); + expect(seen).toHaveLength(43); + expect(new Set(seen).size).toBe(43); + }); + + it.each([ + ['omitted', {}], + ['without a timestamp', { select: 'Body' }], + ])( + 'never returns the cursor timestamp column when select is %s', + async (_label, selectField) => { + await seedCursorRows(); + const { select: _omit, ...rest } = cursorBody(); + const res = await search({ + ...rest, + ...selectField, + cursor: 'start', + }).expect(200); + expect(res.body.data.length).toBeGreaterThan(0); + for (const row of res.body.data) { + expect(row).not.toHaveProperty('__hdx_cursor_ts'); + } + }, + ); + + const createLimitedSource = (querySettings: QuerySettings) => + Source.create({ + kind: SourceKind.Log, + team: logSource.team, + from: logSource.from, + defaultTableSelectExpression: 'Timestamp, SeverityText, Body', + timestampValueExpression: 'Timestamp', + connection: logSource.connection, + name: 'Logs (limited)', + querySettings, + }); + + it('fails a cursor walk instead of truncating when source settings say break', async () => { + await seedCursorRows(); + const limited = await createLimitedSource([ + { setting: 'max_rows_to_read', value: '1' }, + { setting: 'read_overflow_mode', value: 'break' }, + ]); + const errorSpy = jest.spyOn(logger, 'error'); + + const { maxResults: _max, ...body } = cursorBody(); + // Offset paging keeps the source's 'break' setting; a cursor walk overrides + // it and fails instead of truncating. + await search({ ...body, sourceId: limited.id.toString() }).expect(200); + expect(errorSpy).not.toHaveBeenCalled(); + const res = await search({ + ...body, + sourceId: limited.id.toString(), + cursor: 'start', + }); + expect(res.status).toBe(500); + expect(errorSpy).toHaveBeenCalledWith( + expect.objectContaining({ chType: 'TOO_MANY_ROWS' }), + '[search] ClickHouse query error', + ); + errorSpy.mockRestore(); + }); + + it.each([5, 100])( + 'walks every row when maxResults %i is at or above the source max_result_rows', + async maxResults => { + await seedCursorRows(); + const limited = await createLimitedSource([ + { setting: 'max_result_rows', value: '5' }, + { setting: 'result_overflow_mode', value: 'break' }, + ]); + + const seen: string[] = []; + let cursor: string | null | undefined; + for (let pages = 0; pages < 50; pages++) { + const res = await search({ + ...cursorBody(), + sourceId: limited.id.toString(), + maxResults, + cursor: cursor ?? 'start', + }).expect(200); + expect(res.body.data.length).toBeLessThanOrEqual(5); + seen.push(...res.body.data.map((r: Record) => r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + } + + expect(cursor).toBeNull(); + expect(seen).toHaveLength(43); + expect(new Set(seen).size).toBe(43); + }, + ); + + it('keeps a DISTINCT select distinct during a cursor walk', async () => { + const now = DEFAULT_END_TIME - 2 * 60 * 1000; + await bulkInsertLogs( + Array.from({ length: 30 }, (_, i) => ({ + ServiceName: 'cursor-distinct-test', + // Half the rows are older than the first 15m window's start. + Timestamp: new Date(now - (i < 15 ? i : 30 * 60 + i) * 1000), + SeverityText: 'INFO', + Body: `dup-${i % 3}`, + })), + ); + + const seen: string[] = []; + let cursor: string | null | undefined; + for (let pages = 0; pages < 20; pages++) { + const res = await search({ + ...cursorBody(), + where: 'ServiceName:"cursor-distinct-test"', + select: 'DISTINCT Body', + maxResults: 2, + cursor: cursor ?? 'start', + }).expect(200); + seen.push(...res.body.data.map((r: Record) => r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + } + expect(seen.sort()).toEqual(['dup-0', 'dup-1', 'dup-2']); + }); + + it.each([ + [ + 'a rounded leading key', + 'toStartOfMinute(Timestamp), Timestamp', + 'toStartOfMinute(Timestamp) DESC, Body ASC', + ], + [ + 'an exact leading key on a compound expression', + 'toDateTime(Timestamp), Timestamp', + 'Timestamp DESC', + ], + ])( + 'returns every row once when the walk is ordered by %s', + async (_label, timestampValueExpression, orderBy) => { + const compound = await Source.create({ + kind: SourceKind.Log, + team: logSource.team, + from: logSource.from, + defaultTableSelectExpression: 'Timestamp, SeverityText, Body', + timestampValueExpression, + connection: logSource.connection, + name: 'Logs (compound)', + }); + const minute = + Math.floor((DEFAULT_END_TIME - 3 * 60 * 1000) / 60_000) * 60_000; + // Body order is unrelated to time, so a cut on the wrong key loses rows. + await bulkInsertLogs( + Array.from({ length: 120 }, (_, i) => ({ + ServiceName: 'cursor-order-test', + Timestamp: new Date(minute + i * 400), + SeverityText: 'INFO', + Body: `row-${String((i * 37) % 120).padStart(3, '0')}`, + })), + ); + + const seen: string[] = []; + let cursor: string | null | undefined; + for (let pages = 0; pages < 50; pages++) { + const res = await search({ + ...cursorBody(), + sourceId: compound.id.toString(), + where: 'ServiceName:"cursor-order-test"', + orderBy, + maxResults: 10, + cursor: cursor ?? 'start', + }).expect(200); + seen.push(...res.body.data.map((r: Record) => r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + } + expect(cursor).toBeNull(); + expect(seen).toHaveLength(120); + expect(new Set(seen).size).toBe(120); + }, + ); + + it('pages a minute-rounded sort in order with minute-aligned windows and cuts', async () => { + const minuteSource = await Source.create({ + kind: SourceKind.Log, + team: logSource.team, + from: logSource.from, + defaultTableSelectExpression: 'Timestamp, SeverityText, Body', + timestampValueExpression: 'toStartOfMinute(Timestamp), Timestamp', + connection: logSource.connection, + name: 'Logs (minute key)', + }); + // 20 rows a minute for 30 minutes, so the walk crosses the first window's + // boundary; Body order is unrelated to time within each minute. + const base = DEFAULT_END_TIME - 30 * 60 * 1000; + await bulkInsertLogs( + Array.from({ length: 600 }, (_, i) => ({ + ServiceName: 'cursor-minute-test', + Timestamp: new Date(base + i * 3000), + SeverityText: 'INFO', + Body: `row-${String((i * 37) % 600).padStart(3, '0')}`, + })), + ); + + const rows: { minute: number; body: string }[] = []; + const cursors: { + ws: string; + we: string; + o: number; + s: string; + e: string; + }[] = []; + let cursor: string | null | undefined; + for (let pages = 0; pages < 100; pages++) { + const res = await search({ + ...cursorBody(), + sourceId: minuteSource.id.toString(), + where: 'ServiceName:"cursor-minute-test"', + orderBy: 'toStartOfMinute(Timestamp) DESC, Body ASC', + maxResults: 50, + cursor: cursor ?? 'start', + }).expect(200); + for (const r of res.body.data) { + const minute = Math.floor(Date.parse(r.Timestamp) / 60_000); + rows.push({ minute, body: r.Body }); + } + cursor = res.body.nextCursor; + if (cursor == null) break; + cursors.push( + JSON.parse(Buffer.from(cursor.split('.')[0], 'base64url').toString()), + ); + } + + expect(cursor).toBeNull(); + expect(new Set(rows.map(r => r.body)).size).toBe(600); + rows.slice(1).forEach((r, i) => { + const prev = rows.at(i); + expect(prev).toBeDefined(); + if (prev && r.minute === prev.minute) + expect(r.body > prev.body).toBe(true); + else if (prev) expect(r.minute < prev.minute).toBe(true); + }); + for (const c of cursors) { + expect(c.o).toBeLessThanOrEqual(20); + for (const bound of [c.ws, c.we]) { + if (bound !== c.s && bound !== c.e) { + expect(Date.parse(bound) % 60_000).toBe(0); + } + } + } + }); + + it('rejects a select that uses the reserved cursor column name', async () => { + await search({ + ...cursorBody(), + select: 'Timestamp AS __hdx_cursor_ts', + cursor: 'start', + }).expect(400); + }); + + it('rejects a cursor whose contents were edited', async () => { + await seedCursorRows(); + + const first = await search({ ...cursorBody(), cursor: 'start' }).expect( + 200, + ); + const [payload, signature] = String(first.body.nextCursor).split('.'); + const edited = JSON.parse(Buffer.from(payload, 'base64url').toString()); + edited.o += 1; + const forged = `${Buffer.from(JSON.stringify(edited)).toString('base64url')}.${signature}`; + + const res = await search({ ...cursorBody(), cursor: forged }).expect(400); + expect(res.body.message).toMatch(/invalid or expired/); + }); + + const forgeCursor = ( + body: ReturnType & { orderBy: string }, + offset: number, + ) => + encodeSearchCursor( + { + window: { + ordinal: 0, + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + }, + offset, + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + }, + fingerprintQuery({ + sourceId: body.sourceId, + where: body.where, + whereLanguage: 'lucene', + select: body.select, + orderBy: body.orderBy, + }), + ); + + it('rejects a single-window cursor at the offset cap', async () => { + const body = { ...cursorBody(), orderBy: 'Body DESC' }; + const res = await search({ + ...body, + cursor: forgeCursor(body, 10_000), + }).expect(400); + expect(res.body.message).toMatch(/Too many matching rows/); + }); + + it('serves a short last page just under the offset cap', async () => { + const body = { ...cursorBody(), orderBy: 'Body DESC' }; + const res = await search({ + ...body, + cursor: forgeCursor(body, 9_998), + }).expect(200); + expect(res.body.nextCursor).toBeNull(); + }); + + it('rejects a cursor replayed with a different explicit range', async () => { + await seedCursorRows(); + + const first = await search({ ...cursorBody(), cursor: 'start' }).expect( + 200, + ); + expect(first.body.nextCursor).toBeTruthy(); + + await search({ + ...cursorBody(), + startTime: iso(DEFAULT_START_TIME + 60 * 1000), + cursor: first.body.nextCursor, + }).expect(400); + }); + + it('rejects a cursor reused against a different query', async () => { + await seedCursorRows(); + + const first = await search({ ...cursorBody(), cursor: 'start' }).expect( + 200, + ); + expect(first.body.nextCursor).toBeTruthy(); + + await search({ + ...cursorBody(), + where: 'ServiceName:"other-cursor-svc"', + cursor: first.body.nextCursor, + }).expect(400); + }); + + it('leaves offset pagination behaving as before', async () => { + await seedCursorRows(); + + // Rows span ~40 minutes, so a large offset must reach past the newest 15m window. + const first = await search({ ...cursorBody(), maxResults: 5 }).expect(200); + const deep = await search({ + ...cursorBody(), + maxResults: 5, + offset: 20, + }).expect(200); + + expect(first.body.data).toHaveLength(5); + expect(deep.body.data).toHaveLength(5); + // Offset-only callers are not driving a cursor walk. + expect(first.body.nextCursor).toBeNull(); + expect(deep.body.nextCursor).toBeNull(); + const firstBodies = first.body.data.map( + (r: Record) => r.Body, + ); + const deepBodies = deep.body.data.map( + (r: Record) => r.Body, + ); + expect(deepBodies).not.toEqual(firstBodies); + }); }); diff --git a/packages/api/src/routers/external-api/v2/search.ts b/packages/api/src/routers/external-api/v2/search.ts index ede7ed6676..dfba2068eb 100644 --- a/packages/api/src/routers/external-api/v2/search.ts +++ b/packages/api/src/routers/external-api/v2/search.ts @@ -19,6 +19,15 @@ import logger from '@/utils/logger'; import { externalDashboardSearchRequestSchema } from '@/utils/zod'; import { runSearchConfig, type SearchErrorCode } from './utils/search'; +import { + CURSOR_START, + CURSOR_TS_COLUMN, + decodeSearchCursor, + encodeSearchCursor, + fingerprintQuery, + MAX_SINGLE_WINDOW_OFFSET, + type SearchWalk, +} from './utils/searchCursor'; const searchQueryDuration = getHistogram('hyperdx.search.query.duration_ms', { description: 'Duration of external API v2 search queries against ClickHouse.', @@ -132,9 +141,21 @@ const CH_USER_INPUT_ERRORS = new Set([ * description: | * Number of rows to skip (best-effort offset pagination). Default is * 0, max is 10000. Offset pagination is non-deterministic when - * multiple rows share the same timestamp; for reliable deep paging - * filter by the last Timestamp value returned in the previous page - * instead of using a large offset. + * multiple rows share the same timestamp. Prefer cursor for paging. + * cursor: + * type: string + * maxLength: 1024 + * example: "start" + * description: | + * Send the literal string "start" to begin cursor paging, then pass + * back the nextCursor from each response. When set, offset is + * ignored; omit cursor entirely to use offset-based paging. + * As with offset, rows that share a timestamp can be skipped or + * repeated between pages unless orderBy breaks the tie. + * + * Keep requesting pages until nextCursor is null. A page that is + * empty or returns fewer rows than maxResults does NOT mean the + * results are exhausted. * * SearchRow: * type: object @@ -155,6 +176,15 @@ const CH_USER_INPUT_ERRORS = new Set([ * rows: * type: integer * description: Number of rows in this response (not total matching rows). + * nextCursor: + * type: string + * nullable: true + * example: "eyJ2IjoxLCJ3IjoyLCJvIjo1MDAsImYiOiJhMWIyYzNkNGU1ZjZhN2I4In0" + * description: > + * Cursor for the next page, or null when the walk is complete or + * the request did not include a cursor. Pass it back as cursor in + * the next request to fetch the next page. This is an opaque + * value; no guarantees are made about its content. */ // Rejects semicolons and SELECT subqueries in column expressions. @@ -210,6 +240,9 @@ const searchRequestSchema = z.object({ 'select must not contain semicolons or SELECT subqueries; ' + 'use column references, map lookups, or scalar functions only', }) + .refine(value => !value.includes(CURSOR_TS_COLUMN), { + message: `select must not use the reserved name ${CURSOR_TS_COLUMN}`, + }) .describe( 'Comma-separated list of column expressions to return. ' + 'When omitted the source default columns are used. ' + @@ -237,12 +270,21 @@ const searchRequestSchema = z.object({ .number() .int() .min(0) - .max(10_000) + .max(MAX_SINGLE_WINDOW_OFFSET) .optional() .default(0) .describe( 'Number of rows to skip for pagination (0-10000). Default: 0. ' + - 'Prefer timestamp-cursor pagination for large datasets.', + 'Prefer cursor for paging large datasets.', + ), + cursor: z + .string() + .max(1024) + .optional() + .describe( + 'Set to "start" to begin cursor paging, then pass back the nextCursor ' + + 'from each response. When set, offset is ignored. Keep requesting ' + + 'pages until nextCursor is null.', ), }); @@ -357,6 +399,8 @@ function codeToStatus(code: SearchErrorCode): number { case 'SOURCE_NOT_FOUND': case 'CONNECTION_NOT_FOUND': return 404; + case 'CURSOR_WINDOW_TOO_DENSE': + return 400; default: { const _exhaustive: never = code; void _exhaustive; @@ -383,6 +427,7 @@ router.post( orderBy, maxResults, offset, + cursor, } = req.body; const timeRange = parseTimeRange(startTime, endTime); @@ -391,6 +436,49 @@ router.post( } const { startDate, endDate } = timeRange; + const fingerprint = fingerprintQuery({ + sourceId, + where, + whereLanguage, + select, + orderBy: orderBy ?? '', + }); + + // The cursor pins this request's resolved range, so a defaulted endTime + // stays the same on later pages. + let cursorState: SearchWalk | undefined; + if (cursor === CURSOR_START) { + cursorState = { + window: null, + offset: 0, + startTime: startDate.toISOString(), + endTime: endDate.toISOString(), + }; + } else if (cursor != null) { + const decoded = decodeSearchCursor(cursor, fingerprint); + if ('error' in decoded) { + searchQueryErrors.add(1, { error_type: decoded.error }); + return res.status(400).json({ + message: + decoded.error === 'CURSOR_QUERY_MISMATCH' + ? 'Cursor does not match this query. Start a new search.' + : 'Cursor is invalid or expired. Start a new search with cursor "start".', + }); + } + // Reject an explicitly changed bound; omitted or empty bounds default + // to `now` and are not compared. + const rangeChanged = + (!!startTime && startDate.toISOString() !== decoded.startTime) || + (!!endTime && endDate.toISOString() !== decoded.endTime); + if (rangeChanged) { + searchQueryErrors.add(1, { error_type: 'CURSOR_QUERY_MISMATCH' }); + return res.status(400).json({ + message: 'Cursor does not match this query. Start a new search.', + }); + } + cursorState = decoded; + } + const config = externalDashboardSearchRequestSchema.parse({ displayType: 'search' as const, sourceId, @@ -408,6 +496,7 @@ router.post( endDate, maxResults, offset, + cursorState, }), ); @@ -418,7 +507,14 @@ router.post( .json({ message: result.message }); } - return res.json({ data: result.data, rows: result.data.length }); + return res.json({ + data: result.data, + rows: result.data.length, + nextCursor: + result.nextCursorState == null + ? null + : encodeSearchCursor(result.nextCursorState, fingerprint), + }); } catch (err) { if (err instanceof ClickHouseQueryError) { const chType = ((err.cause as Record | undefined) diff --git a/packages/api/src/routers/external-api/v2/utils/__tests__/searchCursor.test.ts b/packages/api/src/routers/external-api/v2/utils/__tests__/searchCursor.test.ts new file mode 100644 index 0000000000..9b1be5fdbd --- /dev/null +++ b/packages/api/src/routers/external-api/v2/utils/__tests__/searchCursor.test.ts @@ -0,0 +1,142 @@ +import { + decodeSearchCursor, + encodeSearchCursor, + fingerprintQuery, + type SearchCursorState, +} from '@/routers/external-api/v2/utils/searchCursor'; + +const FP_INPUT = { + sourceId: '69b46cb0d964ce2d0b9506a8', + where: 'SeverityText:ERROR', + whereLanguage: 'lucene', + select: 'Timestamp,Body', + orderBy: 'Timestamp DESC', +}; + +const RANGE = { + startTime: '2026-05-10T00:00:00.000Z', + endTime: '2026-05-10T01:00:00.000Z', +}; + +const state = ( + overrides: Partial = {}, +): SearchCursorState => ({ + window: { + ordinal: 1, + startTime: '2026-05-10T00:30:00.000Z', + endTime: '2026-05-10T00:45:00.000Z', + }, + offset: 500, + ...RANGE, + ...overrides, +}); + +// Rewrites the payload half of a cursor while keeping its signature. +const tamper = (cursor: string, edit: (p: Record) => void) => { + const [payload, sig] = cursor.split('.'); + const parsed = JSON.parse(Buffer.from(payload, 'base64url').toString('utf8')); + edit(parsed); + return `${Buffer.from(JSON.stringify(parsed)).toString('base64url')}.${sig}`; +}; + +describe('searchCursor', () => { + it('round-trips the window, offset and range', () => { + const fp = fingerprintQuery(FP_INPUT); + const cursor = encodeSearchCursor(state(), fp); + expect(decodeSearchCursor(cursor, fp)).toEqual(state()); + }); + + it('encodes the same state to the same cursor', () => { + const fp = fingerprintQuery(FP_INPUT); + expect(encodeSearchCursor(state(), fp)).toBe( + encodeSearchCursor(state(), fp), + ); + }); + + it('rejects a cursor minted for a different query', () => { + const fp = fingerprintQuery(FP_INPUT); + const cursor = encodeSearchCursor(state(), fp); + const otherFp = fingerprintQuery({ + ...FP_INPUT, + where: 'SeverityText:WARN', + }); + expect(decodeSearchCursor(cursor, otherFp)).toEqual({ + error: 'CURSOR_QUERY_MISMATCH', + }); + }); + + it.each([ + ['the offset', (p: Record) => (p.o = 999_999)], + [ + 'the window start', + (p: Record) => (p.ws = RANGE.startTime), + ], + [ + 'the range end', + (p: Record) => (p.e = '2030-01-01T00:00:00.000Z'), + ], + ])('rejects a cursor whose %s was edited', (_, edit) => { + const fp = fingerprintQuery(FP_INPUT); + const edited = tamper(encodeSearchCursor(state(), fp), edit); + expect(decodeSearchCursor(edited, fp)).toEqual({ error: 'INVALID_CURSOR' }); + }); + + it('rejects a cursor signed with a different secret', () => { + const fp = fingerprintQuery(FP_INPUT); + const cursor = encodeSearchCursor(state(), fp, 'secret-a'); + expect(decodeSearchCursor(cursor, fp, 'secret-b')).toEqual({ + error: 'INVALID_CURSOR', + }); + expect(decodeSearchCursor(cursor, fp, 'secret-a')).toEqual(state()); + }); + + it('rejects malformed, unsigned and unknown-version cursors', () => { + const fp = fingerprintQuery(FP_INPUT); + expect(decodeSearchCursor('not-base64!!', fp)).toEqual({ + error: 'INVALID_CURSOR', + }); + const [payload] = encodeSearchCursor(state(), fp).split('.'); + expect(decodeSearchCursor(payload, fp)).toEqual({ + error: 'INVALID_CURSOR', + }); + const future = tamper(encodeSearchCursor(state(), fp), p => (p.v = 99)); + expect(decodeSearchCursor(future, fp)).toEqual({ error: 'INVALID_CURSOR' }); + }); + + it('rejects a window outside the pinned range', () => { + const fp = fingerprintQuery(FP_INPUT); + const outside = encodeSearchCursor( + state({ + window: { + ordinal: 0, + startTime: '2026-05-10T00:45:00.000Z', + endTime: '2026-05-10T01:15:00.000Z', + }, + }), + fp, + ); + expect(decodeSearchCursor(outside, fp)).toEqual({ + error: 'INVALID_CURSOR', + }); + }); + + it('accepts in-window offsets past the request field cap', () => { + // A walk through a dense window passes the request field's 10,000 cap. + const fp = fingerprintQuery(FP_INPUT); + const deep = encodeSearchCursor(state({ offset: 50_000 }), fp); + expect(decodeSearchCursor(deep, fp)).toMatchObject({ offset: 50_000 }); + }); + + it('changes the fingerprint for every field that changes the result set', () => { + const base = fingerprintQuery(FP_INPUT); + for (const patch of [ + { sourceId: 'aaaaaaaaaaaaaaaaaaaaaaaa' }, + { where: 'x:1' }, + { whereLanguage: 'sql' }, + { select: 'Body' }, + { orderBy: 'Timestamp ASC' }, + ]) { + expect(fingerprintQuery({ ...FP_INPUT, ...patch })).not.toEqual(base); + } + }); +}); diff --git a/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindowCut.test.ts b/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindowCut.test.ts new file mode 100644 index 0000000000..18d1ffc2e4 --- /dev/null +++ b/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindowCut.test.ts @@ -0,0 +1,96 @@ +import { cutWindowAtBucket } from '@/routers/external-api/v2/utils/searchWindows'; + +const window = { + ordinal: 0, + startTime: '2026-10-25T11:45:00.000Z', + endTime: '2026-10-25T12:00:00.400Z', +}; +const ms = (iso: string) => Date.parse(iso); + +describe('cutWindowAtBucket', () => { + it('ends a descending window just after the last row second', () => { + expect( + cutWindowAtBucket({ + window, + direction: 'DESC', + offset: 0, + pageLength: 2000, + lastBucketMs: ms('2026-10-25T11:58:31.000Z'), + bucketMs: 1000, + trailingRunLength: 12, + }), + ).toEqual({ + window: { ...window, endTime: '2026-10-25T11:58:32.000Z' }, + offset: 12, + }); + }); + + it('keeps the range end when the last row second reaches it', () => { + expect( + cutWindowAtBucket({ + window, + direction: 'DESC', + offset: 0, + pageLength: 2000, + lastBucketMs: ms('2026-10-25T12:00:00.000Z'), + bucketMs: 1000, + trailingRunLength: 2000, + }), + ).toEqual({ window, offset: 2000 }); + }); + + it('starts an ascending window at the last row second', () => { + expect( + cutWindowAtBucket({ + window, + direction: 'ASC', + offset: 0, + pageLength: 2000, + lastBucketMs: ms('2026-10-25T11:46:10.000Z'), + bucketMs: 1000, + trailingRunLength: 7, + }), + ).toEqual({ + window: { ...window, startTime: '2026-10-25T11:46:10.000Z' }, + offset: 7, + }); + }); + + it.each([ + ['DESC', { ...window, endTime: '2026-10-25T11:59:00.000Z' }], + ['ASC', { ...window, startTime: '2026-10-25T11:58:00.000Z' }], + ] as const)('cuts a %s window at a minute bucket', (direction, expected) => { + expect( + cutWindowAtBucket({ + window, + direction, + offset: 0, + pageLength: 2000, + lastBucketMs: ms('2026-10-25T11:58:00.000Z'), + bucketMs: 60_000, + trailingRunLength: 40, + }), + ).toEqual({ window: expected, offset: 40 }); + }); + + it.each(['DESC', 'ASC'] as const)( + 'adds the page to the offset when the whole %s page was one second', + direction => { + const cut = + direction === 'DESC' + ? { ...window, endTime: '2026-10-25T11:58:32.000Z' } + : { ...window, startTime: '2026-10-25T11:58:31.000Z' }; + expect( + cutWindowAtBucket({ + window: cut, + direction, + offset: 300, + pageLength: 1000, + lastBucketMs: ms('2026-10-25T11:58:31.000Z'), + bucketMs: 1000, + trailingRunLength: 1000, + }), + ).toEqual({ window: cut, offset: 1300 }); + }, + ); +}); diff --git a/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindows.test.ts b/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindows.test.ts new file mode 100644 index 0000000000..a31cb846ed --- /dev/null +++ b/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindows.test.ts @@ -0,0 +1,479 @@ +import { + DEFAULT_TIME_WINDOWS_SECONDS, + generateTimeWindowsAscending, + generateTimeWindowsDescending, +} from '@hyperdx/common-utils/dist/core/searchWindows'; + +import { + nextSearchWindow, + resolveFirstSearchWindow, + windowBounds, +} from '@/routers/external-api/v2/utils/searchWindows'; + +type Opts = Parameters[4]; + +// Every window of a walk, in the order pages visit them. +async function resolveSearchWindows( + source: { timestampValueExpression?: string }, + orderBy: string, + start: Date, + end: Date, + opts?: Opts, +) { + const first = await resolveFirstSearchWindow( + source, + orderBy, + start, + end, + opts, + ); + const range = { start, end }; + const windows: { + startTime: Date; + endTime: Date; + windowIndex: number; + direction: 'ASC' | 'DESC'; + }[] = []; + for ( + let w: typeof first.window | null = first.window; + w != null; + w = nextSearchWindow(w, range, first.direction, first.cut?.bucketMs) + ) { + windows.push({ + startTime: new Date(w.startTime), + endTime: new Date(w.endTime), + windowIndex: w.ordinal, + direction: first.direction, + }); + } + return windows; +} + +// Reference windows: build every window, floor the boundaries between +// them to the second, drop empties. Stepping must reproduce it exactly. +function referenceWindows( + start: Date, + end: Date, + direction: 'ASC' | 'DESC', + alignMs = 1000, +) { + const windows = + direction === 'ASC' + ? generateTimeWindowsAscending(start, end) + : generateTimeWindowsDescending(start, end); + if (windows.length <= 1) return windows.map(w => [w.startTime, w.endTime]); + const snap = (t: Date) => + t.getTime() === start.getTime() || t.getTime() === end.getTime() + ? t.getTime() + : Math.max(Math.floor(t.getTime() / alignMs) * alignMs, start.getTime()); + return windows + .map(w => [snap(w.startTime), snap(w.endTime)]) + .filter(([a, b]) => a < b) + .map(([a, b]) => [new Date(a), new Date(b)]); +} + +const source = { timestampValueExpression: 'Timestamp' }; + +const START = new Date('2026-05-10T00:00:00.000Z'); +const END = new Date('2026-05-11T00:00:00.000Z'); + +describe('search window stepping', () => { + it('windows a timestamp-led descending order', async () => { + const windows = await resolveSearchWindows( + source, + 'Timestamp DESC', + START, + END, + ); + expect(windows.length).toBeGreaterThan(1); + expect(windows[0].direction).toBe('DESC'); + expect(windows[0].endTime).toEqual(END); + }); + + it('windows a timestamp-led ascending order in the other direction', async () => { + const windows = await resolveSearchWindows( + source, + 'Timestamp ASC', + START, + END, + ); + expect(windows[0].direction).toBe('ASC'); + expect(windows[0].startTime).toEqual(START); + }); + + it('disengages windowing when the order does not lead with the timestamp', async () => { + const windows = await resolveSearchWindows( + source, + 'SeverityText DESC', + START, + END, + ); + expect(windows).toHaveLength(1); + expect(windows[0].startTime).toEqual(START); + expect(windows[0].endTime).toEqual(END); + }); + + it('returns exactly one window when start equals end', async () => { + const windows = await resolveSearchWindows( + source, + 'Timestamp DESC', + START, + START, + ); + expect(windows).toHaveLength(1); + }); + + it('covers the full range with no gaps between adjacent windows', async () => { + const windows = await resolveSearchWindows( + source, + 'Timestamp DESC', + START, + END, + ); + windows + .slice(0, -1) + .forEach((w, i) => + expect(w.startTime).toEqual(windows.at(i + 1)?.endTime), + ); + expect(windows[windows.length - 1].startTime).toEqual(START); + expect(windows[0].endTime).toEqual(END); + }); + + it.each([ + ['descending', 'Timestamp DESC'], + ['ascending', 'Timestamp ASC'], + ])( + 'snaps %s interior boundaries to whole seconds and keeps the caller bounds exact', + async (_, orderBy) => { + const start = new Date('2026-05-10T00:00:00.250Z'); + const end = new Date('2026-05-11T00:00:00.750Z'); + const windows = await resolveSearchWindows(source, orderBy, start, end); + expect(windows.length).toBeGreaterThan(1); + + const bounds = windows.flatMap(w => [w.startTime, w.endTime]); + const interior = bounds.filter( + t => t.getTime() !== start.getTime() && t.getTime() !== end.getTime(), + ); + expect(interior.length).toBeGreaterThan(0); + expect(interior.every(t => t.getTime() % 1000 === 0)).toBe(true); + expect(bounds.map(t => t.getTime())).toContain(start.getTime()); + expect(bounds.map(t => t.getTime())).toContain(end.getTime()); + windows.forEach((w, i) => expect(w.windowIndex).toBe(i)); + }, + ); + + it('drops a window that snapping leaves empty and keeps the range covered', async () => { + const start = new Date('2026-05-10T00:00:00.400Z'); + const end = new Date('2026-05-10T00:15:00.700Z'); + const windows = await resolveSearchWindows( + source, + 'Timestamp DESC', + start, + end, + ); + + expect(windows).toEqual([ + { startTime: start, endTime: end, windowIndex: 0, direction: 'DESC' }, + ]); + }); + + it.each([ + [ + 'a compound toStartOf part next to the raw column', + 'toStartOfMinute(Timestamp), Timestamp', + undefined, + ], + [ + 'a Date column next to a DateTime64 column', + 'EventDate, Timestamp', + 'Date', + ], + ])('windows %s', async (_, timestampValueExpression, eventDateType) => { + const windows = await resolveSearchWindows( + { timestampValueExpression }, + 'Timestamp DESC', + START, + END, + { + columnType: async column => + column === 'EventDate' ? eventDateType : 'DateTime64(9)', + }, + ); + expect(windows.length).toBeGreaterThan(1); + }); + + it.each([ + [ + 'a lone toStartOf expression', + 'toStartOfMinute(Timestamp)', + 'toStartOfMinute(Timestamp) DESC', + undefined, + ], + ['a bare Date column', 'EventDate', 'EventDate DESC', 'Date'], + [ + 'a Nullable(Date32) column', + 'EventDate', + 'EventDate DESC', + 'Nullable(Date32)', + ], + ])( + 'uses one window for %s, whose bounds are inclusive on both ends', + async (_, timestampValueExpression, orderBy, columnType) => { + const windows = await resolveSearchWindows( + { timestampValueExpression }, + orderBy, + START, + END, + { columnType: async () => columnType }, + ); + expect(windows).toHaveLength(1); + }, + ); + + it.each([ + ['DESC', 'Timestamp', 'Timestamp', 1000], + ['ASC', 'Timestamp', 'Timestamp', 1000], + [ + 'DESC', + 'toStartOfMinute(Timestamp), Timestamp', + 'toStartOfMinute(Timestamp)', + 60_000, + ], + [ + 'ASC', + 'toStartOfMinute(Timestamp), Timestamp', + 'toStartOfMinute(Timestamp)', + 60_000, + ], + ] as const)( + 'steps through exactly the reference windows (%s, %s)', + async (direction, timestampValueExpression, sortKey, alignMs) => { + const walkSource = { timestampValueExpression }; + // Deterministic pseudo-random ranges: 0 to 30 days, millisecond precision. + let seed = 42; + const rand = () => + (seed = (seed * 1_103_515_245 + 12_345) % 2 ** 31) / 2 ** 31; + // Plus edge ranges: a mid-second start, with lengths within a second of + // each cumulative window boundary, where flooring can empty a window. + const edgeStart = new Date('2026-05-10T00:00:00.700Z'); + let cumulative = 0; + for (const size of [...DEFAULT_TIME_WINDOWS_SECONDS, 86_400, 86_400]) { + cumulative += size * 1000; + for (const delta of [-61_300, -1300, -300, 0, 300, 1300, 61_300]) { + const end = new Date(edgeStart.getTime() + cumulative + delta); + const walked = await resolveSearchWindows( + walkSource, + `${sortKey} ${direction}`, + edgeStart, + end, + ); + expect(walked.map(w => [w.startTime, w.endTime])).toEqual( + referenceWindows(edgeStart, end, direction, alignMs), + ); + } + } + for (let i = 0; i < 200; i++) { + const start = new Date( + Date.UTC(2026, 4, 1) + Math.floor(rand() * 86_400_000), + ); + const end = new Date( + start.getTime() + Math.floor(rand() * 30 * 86_400_000), + ); + const walked = await resolveSearchWindows( + walkSource, + `${sortKey} ${direction}`, + start, + end, + ); + expect(walked.map(w => [w.startTime, w.endTime])).toEqual( + referenceWindows(start, end, direction, alignMs), + ); + } + }, + ); + + it('steps through windows sized by the default table, then 24h repeating', async () => { + const windows = await resolveSearchWindows( + source, + 'Timestamp DESC', + START, + new Date(START.getTime() + 3 * 86_400_000), + ); + const sizes = windows.map( + w => (w.endTime.getTime() - w.startTime.getTime()) / 1000, + ); + expect(sizes.slice(0, DEFAULT_TIME_WINDOWS_SECONDS.length)).toEqual( + DEFAULT_TIME_WINDOWS_SECONDS, + ); + expect( + sizes + .slice(DEFAULT_TIME_WINDOWS_SECONDS.length, -1) + .every(s => s === 86_400), + ).toBe(true); + }); + + it.each([ + ['EventDate, toDateTime(ts)', 'toDateTime(ts) DESC', 'toDateTime(ts)'], + [ + 'TimestampTime, Timestamp', + '(TimestampTime, Timestamp) DESC', + 'TimestampTime', + ], + ['TimestampTime, Timestamp', 'Timestamp DESC, Body', 'Timestamp'], + ['toStartOfMinute(Timestamp), Timestamp', 'Timestamp DESC', 'Timestamp'], + ])( + 'cuts %s ordered by %s on %s', + async (timestampValueExpression, orderBy, cutPart) => { + const first = await resolveFirstSearchWindow( + { timestampValueExpression }, + orderBy, + START, + END, + { + columnType: async column => + column === 'EventDate' + ? 'Date' + : column === 'TimestampTime' || column === 'EventTime' + ? 'DateTime' + : 'DateTime64(9)', + }, + ); + expect(first.windowed).toBe(true); + expect(first.cut).toEqual({ expression: cutPart, bucketMs: 1000 }); + }, + ); + + it.each([ + ['toStartOfSecond(Timestamp)', 1000], + ['toStartOfMinute(Timestamp)', 60_000], + ['toStartOfFiveMinutes(Timestamp)', 300_000], + ['toStartOfFiveMinute(Timestamp)', 300_000], + ['toStartOfFifteenMinutes(Timestamp)', 900_000], + ])( + 'cuts on a rounded leading key %s in %ims buckets', + async (key, bucketMs) => { + const first = await resolveFirstSearchWindow( + { timestampValueExpression: `${key}, Timestamp` }, + `${key} DESC, Body`, + START, + END, + { columnType: async () => 'DateTime64(9)' }, + ); + expect(first.windowed).toBe(true); + expect(first.cut).toEqual({ expression: key, bucketMs }); + }, + ); + + it.each([ + // Ten-minute buckets don't align to UTC in +x:45 time zones. + [ + 'toStartOfTenMinutes(Timestamp), Timestamp', + 'toStartOfTenMinutes(Timestamp) DESC, Body', + ], + [ + 'toStartOfHour(Timestamp), Timestamp', + 'toStartOfHour(Timestamp) DESC, Body', + ], + ['EventDate, EventTime', 'EventDate DESC, Body'], + // Without an exact part, rounded bounds are inclusive on both ends. + ['toStartOfMinute(Timestamp)', 'toStartOfMinute(Timestamp) DESC, Body'], + ])( + 'uses one window for %s ordered by %s', + async (timestampValueExpression, orderBy) => { + const first = await resolveFirstSearchWindow( + { timestampValueExpression }, + orderBy, + START, + END, + { + columnType: async column => + column === 'EventDate' ? 'Date' : 'DateTime64(9)', + }, + ); + expect(first.windowed).toBe(false); + expect(first.cut).toBeNull(); + }, + ); + + it('reports windowing for a range shorter than the first window', async () => { + const short = await resolveFirstSearchWindow( + source, + 'Timestamp DESC', + START, + new Date(START.getTime() + 10 * 60 * 1000), + ); + expect(short.windowed).toBe(true); + const unordered = await resolveFirstSearchWindow( + source, + 'Body DESC', + START, + END, + ); + expect(unordered.windowed).toBe(false); + const offsetOnly = await resolveFirstSearchWindow( + source, + 'Timestamp DESC', + START, + END, + { + forceSingleWindow: true, + }, + ); + expect(offsetOnly.windowed).toBe(false); + }); + + it('includes a window end only when it is the range end', () => { + const range = { start: START, end: END }; + const newest = { + ordinal: 0, + startTime: '2026-05-10T23:45:00.000Z', + endTime: END.toISOString(), + }; + const older = { + ordinal: 1, + startTime: '2026-05-10T17:45:00.000Z', + endTime: '2026-05-10T23:45:00.000Z', + }; + expect(windowBounds(newest, range)).toEqual({ + dateRange: [new Date(newest.startTime), END], + dateRangeStartInclusive: true, + dateRangeEndInclusive: true, + }); + expect(windowBounds(older, range)).toMatchObject({ + dateRangeStartInclusive: true, + dateRangeEndInclusive: false, + }); + }); + + it('windows a default tuple order led by the timestamp', async () => { + const tupleSource = { timestampValueExpression: 'TimestampTime' }; + const windows = await resolveSearchWindows( + tupleSource, + '(TimestampTime, Timestamp) DESC', + START, + END, + ); + expect(windows.length).toBeGreaterThan(1); + expect(windows[0].direction).toBe('DESC'); + }); + + it('uses one window when a tuple order is not led by the timestamp', async () => { + const windows = await resolveSearchWindows( + source, + '(SeverityText, Timestamp) DESC', + START, + END, + ); + expect(windows).toHaveLength(1); + }); + + it('uses one window when the first sort key only derives from the timestamp', async () => { + const windows = await resolveSearchWindows( + source, + 'toHour(Timestamp) DESC', + START, + END, + ); + expect(windows).toHaveLength(1); + }); +}); diff --git a/packages/api/src/routers/external-api/v2/utils/search.ts b/packages/api/src/routers/external-api/v2/utils/search.ts index 552fdb14be..a744412462 100644 --- a/packages/api/src/routers/external-api/v2/utils/search.ts +++ b/packages/api/src/routers/external-api/v2/utils/search.ts @@ -1,11 +1,15 @@ import { getMetadata } from '@hyperdx/common-utils/dist/core/metadata'; -import { buildSearchChartConfig } from '@hyperdx/common-utils/dist/core/searchChartConfig'; +import { + buildSearchChartConfig, + resolveSelect, +} from '@hyperdx/common-utils/dist/core/searchChartConfig'; import { getFirstTimestampValueExpression, splitAndTrimWithBracket, } from '@hyperdx/common-utils/dist/core/utils'; import type { ChartConfigWithDateRange, + QuerySettings, TSource, } from '@hyperdx/common-utils/dist/types'; import { @@ -19,7 +23,55 @@ import { getConnectionById } from '@/controllers/connection'; import { getSource } from '@/controllers/sources'; import type { ExternalDashboardSearchRequestConfig } from '@/utils/zod'; -export type SearchErrorCode = 'SOURCE_NOT_FOUND' | 'CONNECTION_NOT_FOUND'; +import { + CURSOR_TS_COLUMN, + MAX_SINGLE_WINDOW_OFFSET, + type SearchCursorState, + type SearchWalk, +} from './searchCursor'; +import { + cutWindowAtBucket, + nextSearchWindow, + resolveFirstSearchWindow, + windowBounds, +} from './searchWindows'; + +const OVERFLOW_MODE_SETTING = /_overflow_mode(_leaf)?$/; +const DEFAULT_OVERFLOW_MODE_SETTINGS = [ + 'result_overflow_mode', + 'read_overflow_mode', + 'timeout_overflow_mode', +]; + +// A limit set to 'break' returns a partial page with no error, which a walk +// would read as the end of its window. Cursor walks fail loudly instead. +function withThrowingOverflow( + querySettings: QuerySettings | undefined, +): QuerySettings { + const settings = querySettings ?? []; + const overflowModes = new Set([ + ...DEFAULT_OVERFLOW_MODE_SETTINGS, + ...settings + .map(({ setting }) => setting) + .filter(setting => OVERFLOW_MODE_SETTING.test(setting)), + ]); + return [ + ...settings.filter(({ setting }) => !overflowModes.has(setting)), + ...[...overflowModes].map(setting => ({ setting, value: 'throw' })), + ]; +} + +function maxResultRows(querySettings: QuerySettings | undefined): number { + const value = Number( + querySettings?.find(s => s.setting === 'max_result_rows')?.value, + ); + return value > 0 ? value : Infinity; +} + +export type SearchErrorCode = + | 'SOURCE_NOT_FOUND' + | 'CONNECTION_NOT_FOUND' + | 'CURSOR_WINDOW_TOO_DENSE'; type SearchError = { isError: true; @@ -30,6 +82,7 @@ type SearchError = { type SearchResults = { isError: false; data: Record[]; + nextCursorState: SearchCursorState | null; }; // Mirrors `optimizeDefaultOrderBy` + `useDefaultOrderBy` from DBSearchPage.tsx. @@ -85,6 +138,7 @@ export async function runSearchConfig({ endDate, maxResults, offset, + cursorState, }: { teamId: string; config: ExternalDashboardSearchRequestConfig; @@ -92,6 +146,7 @@ export async function runSearchConfig({ endDate: Date; maxResults: number; offset: number; + cursorState?: SearchWalk; }): Promise { const source = await getSource(teamId, config.sourceId); if (!source) { @@ -133,26 +188,93 @@ export async function runSearchConfig({ ...(requestTimeout != null ? { requestTimeout } : {}), }); + const effectiveOrderBy = + config.orderBy?.trim() || resolveSearchOrderBy(source); + + // Windows come from the range pinned in the cursor, so boundaries stay put + // between pages. + const walkStart = cursorState ? new Date(cursorState.startTime) : startDate; + const walkEnd = cursorState ? new Date(cursorState.endTime) : endDate; + + const metadata = getMetadata(clickhouseClient); + + const baseSelect = resolveSelect(source, config.select); + // Windows are cut via a hidden select column, which would change what + // DISTINCT deduplicates; DISTINCT is also per window. Those walks, and + // offset-only requests, use one window over the whole range. + const canWindow = + typeof baseSelect === 'string' && !/^\s*DISTINCT\b/i.test(baseSelect); + const first = await resolveFirstSearchWindow( + source, + effectiveOrderBy, + walkStart, + walkEnd, + { + forceSingleWindow: !cursorState || !canWindow, + columnType: async column => + ( + await metadata.getColumn({ + databaseName: source.from.databaseName, + tableName: source.from.tableName, + column, + connectionId: source.connection.toString(), + }) + )?.type, + }, + ); + const range = { start: walkStart, end: walkEnd }; + const window = cursorState?.window ?? first.window; + const effectiveOffset = cursorState ? cursorState.offset : offset; + const cut = cursorState ? first.cut : null; + // A walk that can't cut its windows pages with plain offsets, so it keeps + // the `offset` field's cap. Erroring, rather than ending the walk, keeps a + // truncated result from looking complete. + const requestedLimit = + cursorState && !cut + ? Math.min(maxResults, MAX_SINGLE_WINDOW_OFFSET - effectiveOffset) + : maxResults; + // A cursor walk reads one extra row to learn whether the window continues, + // unless the source's max_result_rows leaves no room for it; then a full + // page counts as "more in the window". + const rowCap = maxResultRows(source.querySettings); + const pageLimit = cursorState + ? Math.min(requestedLimit, rowCap) + : requestedLimit; + const queryLimit = cursorState + ? Math.min(pageLimit + 1, rowCap) + : requestedLimit; + if (cursorState && pageLimit <= 0) { + return { + isError: true, + code: 'CURSOR_WINDOW_TOO_DENSE', + message: + 'Too many matching rows to page further. Narrow the time range or the filter.', + }; + } + const searchBase = buildSearchChartConfig(source, { where: typeof config.where === 'string' ? config.where : '', whereLanguage: config.whereLanguage ?? 'lucene', - select: config.select ?? null, + select: cut + ? `${baseSelect || '*'}, toUnixTimestamp64Milli(toDateTime64(${cut.expression}, 3)) AS ${CURSOR_TS_COLUMN}` + : (config.select ?? null), displayType: DisplayType.Search, - orderBy: config.orderBy?.trim() || resolveSearchOrderBy(source), - dateRange: [startDate, endDate], + orderBy: effectiveOrderBy, + ...windowBounds(window, range), }); const chartConfig: ChartConfigWithDateRange = { ...searchBase, connection: source.connection.toString(), - limit: { limit: maxResults, offset }, + limit: { limit: queryLimit, offset: effectiveOffset }, } as ChartConfigWithDateRange; - const metadata = getMetadata(clickhouseClient); const result = await clickhouseClient.queryChartConfig({ config: chartConfig, metadata, - querySettings: source.querySettings, + querySettings: cursorState + ? withThrowingOverflow(source.querySettings) + : source.querySettings, }); if ( @@ -164,8 +286,70 @@ export async function runSearchConfig({ throw new Error('Unexpected ClickHouse response shape: missing data array'); } - return { - isError: false, - data: (result as { data: unknown[] }).data as Record[], - }; + const rows = (result as { data: unknown[] }).data as Record< + string, + unknown + >[]; + const hasMoreInWindow = + cursorState != null && + (rows.length > pageLimit || + (queryLimit === pageLimit && rows.length === pageLimit)); + const page = cursorState ? rows.slice(0, pageLimit) : rows; + const split = cut + ? page.map(({ [CURSOR_TS_COLUMN]: ms, ...row }) => { + const bucket = Math.floor(Number(ms) / cut.bucketMs) * cut.bucketMs; + if (!Number.isFinite(bucket)) { + throw new Error(`Search result is missing ${CURSOR_TS_COLUMN}`); + } + return { bucket, row }; + }) + : null; + const buckets = split?.map(r => r.bucket) ?? []; + const data = split?.map(r => r.row) ?? page; + + // Only a cursor walk gets a next cursor; offset-only callers are unchanged. + let nextCursorState: SearchCursorState | null = null; + if (cursorState) { + const pinned = { + startTime: cursorState.startTime, + endTime: cursorState.endTime, + }; + if (hasMoreInWindow && cut) { + const lastBucket = buckets[buckets.length - 1]; + let trailingRunLength = 0; + while (buckets[buckets.length - 1 - trailingRunLength] === lastBucket) { + trailingRunLength++; + } + nextCursorState = { + ...cutWindowAtBucket({ + window, + direction: first.direction, + offset: effectiveOffset, + pageLength: data.length, + lastBucketMs: lastBucket, + bucketMs: cut.bucketMs, + trailingRunLength, + }), + ...pinned, + }; + } else if (hasMoreInWindow) { + nextCursorState = { + window, + offset: effectiveOffset + data.length, + ...pinned, + }; + } else { + const nextWindow = nextSearchWindow( + window, + range, + first.direction, + first.cut?.bucketMs, + ); + if (nextWindow) { + nextCursorState = { window: nextWindow, offset: 0, ...pinned }; + } + } + } + + return { isError: false, data, nextCursorState }; } diff --git a/packages/api/src/routers/external-api/v2/utils/searchCursor.ts b/packages/api/src/routers/external-api/v2/utils/searchCursor.ts new file mode 100644 index 0000000000..26973af190 --- /dev/null +++ b/packages/api/src/routers/external-api/v2/utils/searchCursor.ts @@ -0,0 +1,156 @@ +import { createHash, createHmac, timingSafeEqual } from 'crypto'; + +import { EXPRESS_SESSION_SECRET } from '@/config'; + +const CURSOR_VERSION = 1; +const FINGERPRINT_LENGTH = 16; +// Also the `offset` request field's cap. +export const MAX_SINGLE_WINDOW_OFFSET = 10_000; +const SIGNATURE_CONTEXT = 'hdx-search-cursor:'; +/** Hidden column carrying each row's second; reserved in `select`. */ +export const CURSOR_TS_COLUMN = '__hdx_cursor_ts'; + +/** Literal `cursor` value that starts a windowed cursor walk. */ +export const CURSOR_START = 'start'; + +export type CursorWindow = { + ordinal: number; + startTime: string; + endTime: string; +}; + +export type SearchCursorState = { + window: CursorWindow; + offset: number; + /** Time range pinned by the walk, so a defaulted endTime stays stable. */ + startTime: string; + endTime: string; +}; + +/** A walk in progress; `window` is null on its first page. */ +export type SearchWalk = Omit & { + window: CursorWindow | null; +}; + +export type CursorFingerprintInput = { + sourceId: string; + where: string; + whereLanguage: string; + select: string; + orderBy: string; +}; + +export type CursorDecodeError = { + error: 'INVALID_CURSOR' | 'CURSOR_QUERY_MISMATCH'; +}; + +/** + * Ties a cursor to the query that produced it, so it can't be reused after + * `where`, `select`, or `orderBy` change. The time range travels in the cursor + * instead, so a defaulted `endTime` doesn't invalidate it. + */ +export function fingerprintQuery(input: CursorFingerprintInput): string { + // A JSON array keeps field boundaries unambiguous. + const canonical = JSON.stringify([ + input.sourceId, + input.where, + input.whereLanguage, + input.select, + input.orderBy, + ]); + return createHash('sha256') + .update(canonical) + .digest('hex') + .slice(0, FINGERPRINT_LENGTH); +} + +// Signed with the session secret, so a cursor can only come from this server +// and is valid wherever its sessions are. +function sign(payload: string, secret: string): Buffer { + return createHmac('sha256', secret) + .update(SIGNATURE_CONTEXT + payload) + .digest(); +} + +export function encodeSearchCursor( + state: SearchCursorState, + fingerprint: string, + secret: string = EXPRESS_SESSION_SECRET, +): string { + const payload = Buffer.from( + JSON.stringify({ + v: CURSOR_VERSION, + k: state.window.ordinal, + ws: state.window.startTime, + we: state.window.endTime, + o: state.offset, + f: fingerprint, + s: state.startTime, + e: state.endTime, + }), + ).toString('base64url'); + return `${payload}.${sign(payload, secret).toString('base64url')}`; +} + +export function decodeSearchCursor( + cursor: string, + fingerprint: string, + secret: string = EXPRESS_SESSION_SECRET, +): SearchCursorState | CursorDecodeError { + const [payload, signature, ...rest] = cursor.split('.'); + if (!payload || !signature || rest.length > 0) { + return { error: 'INVALID_CURSOR' }; + } + const expected = sign(payload, secret); + const actual = Buffer.from(signature, 'base64url'); + if (actual.length !== expected.length || !timingSafeEqual(actual, expected)) { + return { error: 'INVALID_CURSOR' }; + } + + let parsed: unknown; + try { + parsed = JSON.parse(Buffer.from(payload, 'base64url').toString('utf8')); + } catch { + return { error: 'INVALID_CURSOR' }; + } + + if (parsed == null || typeof parsed !== 'object') { + return { error: 'INVALID_CURSOR' }; + } + const fields: Record = { ...parsed }; + const { v, k, ws, we, o, f, s: start, e: end } = fields; + if ( + v !== CURSOR_VERSION || + typeof k !== 'number' || + typeof o !== 'number' || + !Number.isInteger(k) || + !Number.isInteger(o) || + k < 0 || + o < 0 || + typeof f !== 'string' || + typeof ws !== 'string' || + typeof we !== 'string' || + typeof start !== 'string' || + typeof end !== 'string' + ) { + return { error: 'INVALID_CURSOR' }; + } + const [wsMs, weMs, startMs, endMs] = [ws, we, start, end].map(Date.parse); + if ( + [wsMs, weMs, startMs, endMs].some(Number.isNaN) || + wsMs < startMs || + weMs > endMs || + wsMs > weMs + ) { + return { error: 'INVALID_CURSOR' }; + } + if (f !== fingerprint) { + return { error: 'CURSOR_QUERY_MISMATCH' }; + } + return { + window: { ordinal: k, startTime: ws, endTime: we }, + offset: o, + startTime: start, + endTime: end, + }; +} diff --git a/packages/api/src/routers/external-api/v2/utils/searchWindows.ts b/packages/api/src/routers/external-api/v2/utils/searchWindows.ts new file mode 100644 index 0000000000..6157a4eb2a --- /dev/null +++ b/packages/api/src/routers/external-api/v2/utils/searchWindows.ts @@ -0,0 +1,252 @@ +import { DEFAULT_TIME_WINDOWS_SECONDS } from '@hyperdx/common-utils/dist/core/searchWindows'; +import { + getFirstOrderingItem, + isFirstOrderByAscending, + isRoundedTimestampPart, + parseToStartOfFunction, + removeTrailingDirection, + splitAndTrimWithBracket, +} from '@hyperdx/common-utils/dist/core/utils'; + +import type { CursorWindow } from './searchCursor'; + +// Windows keep the requested order only when the first sort key is the +// timestamp itself, not a derived key such as toHour(Timestamp). Stricter than +// isTimestampExpressionInFirstOrderBy, which accepts any key containing it. +function leadingTimestampPart( + orderBy: string, + timestampValueExpression: string | undefined, +): string | null { + const first = getFirstOrderingItem(orderBy); + if (typeof first !== 'string' || timestampValueExpression == null) { + return null; + } + const expression = removeTrailingDirection(first); + // A tuple order such as (TimestampTime, Timestamp) is led by its first element. + const leading = /^\(.*\)$/s.test(expression) + ? splitAndTrimWithBracket(expression.slice(1, -1))[0] + : expression; + return splitAndTrimWithBracket(timestampValueExpression).includes(leading) + ? leading + : null; +} + +// toStartOf*, toDate and Date-typed parts get inclusive bounds on both ends. +async function isExactTimestampPart( + part: string, + columnType: (column: string) => Promise, +): Promise { + if (isRoundedTimestampPart(part, undefined)) return false; + return !isRoundedTimestampPart(part, await columnType(part)); +} + +export type SearchDirection = 'ASC' | 'DESC'; +type Range = { start: Date; end: Date }; + +/** The leading sort key windows are cut on, and the length of its buckets. */ +export type SearchCut = { expression: string; bucketMs: number }; + +// Rounded leading keys whose buckets divide 15 minutes. Every current UTC +// offset is a multiple of 15 minutes, so these buckets align to UTC in any +// time zone and windows can be floored to the bucket length. +const ROUNDED_KEY_BUCKET_MS: Record = { + toStartOfSecond: 1000, + toStartOfMinute: 60_000, + toStartOfFiveMinute: 300_000, + toStartOfFiveMinutes: 300_000, + toStartOfFifteenMinutes: 900_000, +}; + +async function resolveCut( + leading: string, + timestampValueExpression: string, + columnType: (column: string) => Promise, +): Promise { + if (await isExactTimestampPart(leading, columnType)) { + return { expression: leading, bucketMs: 1000 }; + } + const bucketMs = + ROUNDED_KEY_BUCKET_MS[parseToStartOfFunction(leading)?.function ?? '']; + if (bucketMs == null) return null; + // Rounded bounds are inclusive on both ends; an exact part's strict bound + // is what keeps windows half-open. + for (const part of splitAndTrimWithBracket(timestampValueExpression)) { + if (await isExactTimestampPart(part, columnType)) { + return { expression: leading, bucketMs }; + } + } + return null; +} + +export async function resolveFirstSearchWindow( + source: { timestampValueExpression?: string }, + orderBy: string, + startDate: Date, + endDate: Date, + { + forceSingleWindow = false, + columnType = async () => undefined, + }: { + forceSingleWindow?: boolean; + columnType?: (column: string) => Promise; + } = {}, +): Promise<{ + window: CursorWindow; + windowed: boolean; + cut: SearchCut | null; + direction: SearchDirection; +}> { + const direction = isFirstOrderByAscending(orderBy) ? 'ASC' : 'DESC'; + const wholeRange = { + ordinal: 0, + startTime: startDate.toISOString(), + endTime: endDate.toISOString(), + }; + const leading = leadingTimestampPart( + orderBy, + source.timestampValueExpression, + ); + // Window boundaries are floored to the leading key's bucket, so one bucket's + // rows, which the sort keeps together, never span two windows. + const cut = + !forceSingleWindow && startDate < endDate && leading != null + ? await resolveCut( + leading, + source.timestampValueExpression ?? '', + columnType, + ) + : null; + if (cut == null) { + return { window: wholeRange, windowed: false, cut, direction }; + } + + // Step from an empty window at the edge the walk starts from. + const edge = (direction === 'DESC' ? endDate : startDate).toISOString(); + const first = nextSearchWindow( + { ordinal: -1, startTime: edge, endTime: edge }, + { start: startDate, end: endDate }, + direction, + cut.bucketMs, + ); + return { window: first ?? wholeRange, windowed: true, cut, direction }; +} + +const WINDOW_SIZES_MS = DEFAULT_TIME_WINDOWS_SECONDS.map(sec => sec * 1000); +const LARGEST_WINDOW_MS = WINDOW_SIZES_MS[WINDOW_SIZES_MS.length - 1]; + +// Total length of the first `count` windows; the largest size repeats. +function cumulativeWindowMs(count: number): number { + const listed = WINDOW_SIZES_MS.slice(0, count).reduce((a, b) => a + b, 0); + return ( + listed + Math.max(0, count - WINDOW_SIZES_MS.length) * LARGEST_WINDOW_MS + ); +} + +// Boundaries between windows are floored to the bucket (a second for an exact +// key), so a second-precision column such as TimestampTime doesn't drop rows +// at a boundary. The caller's bounds stay exact; a window that flooring +// empties is skipped. +const floorToBucket = (ms: number, startMs: number, bucketMs: number) => + Math.max(Math.floor(ms / bucketMs) * bucketMs, startMs); + +/** + * The window after `current`, or null when the walk has reached the range + * edge. Its near edge is the current window's far edge, so windows never gap + * or overlap; its far edge is measured from the range edge, so flooring + * doesn't accumulate. + */ +export function nextSearchWindow( + current: CursorWindow, + range: Range, + direction: SearchDirection, + bucketMs = 1000, +): CursorWindow | null { + const startMs = range.start.getTime(); + const endMs = range.end.getTime(); + let { ordinal } = current; + let ws = Date.parse(current.startTime); + let we = Date.parse(current.endTime); + for (;;) { + ordinal += 1; + if (direction === 'DESC') { + if (ws <= startMs) return null; + we = ws; + const raw = endMs - cumulativeWindowMs(ordinal + 1); + ws = raw <= startMs ? startMs : floorToBucket(raw, startMs, bucketMs); + } else { + if (we >= endMs) return null; + ws = we; + const raw = startMs + cumulativeWindowMs(ordinal + 1); + we = raw >= endMs ? endMs : floorToBucket(raw, startMs, bucketMs); + } + if (ws < we) { + return { + ordinal, + startTime: new Date(ws).toISOString(), + endTime: new Date(we).toISOString(), + }; + } + } +} + +/** Windows are [start, end); the one holding the range end also includes it. */ +export function windowBounds( + window: CursorWindow, + range: Range, +): { + dateRange: [Date, Date]; + dateRangeStartInclusive: true; + dateRangeEndInclusive: boolean; +} { + const end = new Date(window.endTime); + return { + dateRange: [new Date(window.startTime), end], + dateRangeStartInclusive: true, + dateRangeEndInclusive: end.getTime() === range.end.getTime(), + }; +} + +/** + * Shrinks a window to the rows not yet returned, cut at the last row's bucket, + * so the next page's offset only skips that bucket's returned rows. The far + * edge is untouched, so nextSearchWindow is unaffected. + */ +export function cutWindowAtBucket({ + window, + direction, + offset, + pageLength, + lastBucketMs, + bucketMs, + trailingRunLength, +}: { + window: CursorWindow; + direction: SearchDirection; + offset: number; + pageLength: number; + lastBucketMs: number; + bucketMs: number; + trailingRunLength: number; +}): { window: CursorWindow; offset: number } { + const startMs = Date.parse(window.startTime); + const endMs = Date.parse(window.endTime); + const cut = + direction === 'DESC' + ? { + ...window, + endTime: new Date( + Math.min(lastBucketMs + bucketMs, endMs), + ).toISOString(), + } + : { + ...window, + startTime: new Date(Math.max(lastBucketMs, startMs)).toISOString(), + }; + // An unchanged window means the whole page was that one second. + const unchanged = + cut.startTime === window.startTime && cut.endTime === window.endTime; + return { + window: unchanged ? window : cut, + offset: unchanged ? offset + pageLength : trailingRunLength, + }; +} diff --git a/packages/app/src/hooks/__tests__/useChartConfig.test.tsx b/packages/app/src/hooks/__tests__/useChartConfig.test.tsx index a20788a176..d2cc3a3e72 100644 --- a/packages/app/src/hooks/__tests__/useChartConfig.test.tsx +++ b/packages/app/src/hooks/__tests__/useChartConfig.test.tsx @@ -27,8 +27,10 @@ import { import { useMVOptimizationExplanation } from '@/hooks/useMVOptimizationExplanation'; // Mock DEFAULT_TIME_WINDOWS_SECONDS to remove the 15m window -jest.mock('@/utils/searchWindows', () => { - const original = jest.requireActual('@/utils/searchWindows'); +jest.mock('@hyperdx/common-utils/dist/core/searchWindows', () => { + const original = jest.requireActual( + '@hyperdx/common-utils/dist/core/searchWindows', + ); const mockWindows = [ 6 * 60 * 60, // 6h 6 * 60 * 60, // 6h diff --git a/packages/app/src/hooks/useChartConfig.tsx b/packages/app/src/hooks/useChartConfig.tsx index ed367f0c27..fd30ab4597 100644 --- a/packages/app/src/hooks/useChartConfig.tsx +++ b/packages/app/src/hooks/useChartConfig.tsx @@ -16,6 +16,7 @@ import { isUsingGranularity, renderChartConfig, } from '@hyperdx/common-utils/dist/core/renderChartConfig'; +import { generateTimeWindowsDescending } from '@hyperdx/common-utils/dist/core/searchWindows'; import { convertDateRangeToGranularityString, convertGranularityToSeconds, @@ -55,7 +56,6 @@ import { queryPromqlChartConfig, reduceBucketRows, } from '@/utils/promqlChartQuery'; -import { generateTimeWindowsDescending } from '@/utils/searchWindows'; import { useMVOptimizationExplanation } from './useMVOptimizationExplanation'; diff --git a/packages/app/src/hooks/useOffsetPaginatedQuery.tsx b/packages/app/src/hooks/useOffsetPaginatedQuery.tsx index e8b5b573f6..5e08f9910d 100644 --- a/packages/app/src/hooks/useOffsetPaginatedQuery.tsx +++ b/packages/app/src/hooks/useOffsetPaginatedQuery.tsx @@ -14,6 +14,13 @@ import { } from '@hyperdx/common-utils/dist/clickhouse'; import { Metadata } from '@hyperdx/common-utils/dist/core/metadata'; import { renderChartConfig } from '@hyperdx/common-utils/dist/core/renderChartConfig'; +import { + DEFAULT_TIME_WINDOWS_SECONDS, + generateTimeWindowsAscending, + generateTimeWindowsDescending, + ONE_MIN_WINDOW, + TimeWindow, +} from '@hyperdx/common-utils/dist/core/searchWindows'; import { isFirstOrderByAscending, isTimestampExpressionInFirstOrderBy, @@ -42,13 +49,6 @@ import { useMVOptimizationExplanation } from '@/hooks/useMVOptimizationExplanati import { useQueryAttribution } from '@/queryAttribution'; import { useSource } from '@/source'; import { queryPromqlChartConfig } from '@/utils/promqlChartQuery'; -import { - DEFAULT_TIME_WINDOWS_SECONDS, - generateTimeWindowsAscending, - generateTimeWindowsDescending, - ONE_MIN_WINDOW, - TimeWindow, -} from '@/utils/searchWindows'; type TQueryKey = readonly [ string, diff --git a/packages/common-utils/src/__tests__/utils.test.ts b/packages/common-utils/src/__tests__/utils.test.ts index ee0ae6eae0..3eb16e0551 100644 --- a/packages/common-utils/src/__tests__/utils.test.ts +++ b/packages/common-utils/src/__tests__/utils.test.ts @@ -19,6 +19,7 @@ import { hasPositiveSeriesLimit, isFirstOrderByAscending, isJsonExpression, + isRoundedTimestampPart, isTimestampExpressionInFirstOrderBy, joinQuerySettings, optimizeTimestampValueExpression, @@ -2550,6 +2551,22 @@ describe('utils', () => { }); }); + describe('isRoundedTimestampPart', () => { + it.each([ + ['toStartOfMinute(Timestamp)', undefined, true], + ['toDate(Timestamp)', undefined, true], + ['EventDate', 'Date', true], + ['EventDate', 'LowCardinality(Date)', true], + ['EventDate', 'Nullable(Date32)', true], + ['TimestampTime', 'DateTime', false], + ['Timestamp', 'DateTime64(9)', false], + ['toDateTime(Timestamp)', undefined, false], + ['TODATE(Timestamp)', undefined, false], + ])('%s (%s) -> %s', (part, columnType, expected) => { + expect(isRoundedTimestampPart(part, columnType)).toBe(expected); + }); + }); + describe('parseToStartOfFunction', () => { it.each([ { diff --git a/packages/common-utils/src/core/renderChartConfig.ts b/packages/common-utils/src/core/renderChartConfig.ts index 8de7551337..0c3979984f 100644 --- a/packages/common-utils/src/core/renderChartConfig.ts +++ b/packages/common-utils/src/core/renderChartConfig.ts @@ -18,6 +18,7 @@ import { extractSettingsClauseFromEnd, getFirstTimestampValueExpression, hasPositiveSeriesLimit, + isRoundedTimestampPart, joinQuerySettings, optimizeTimestampValueExpression, parseToStartOfFunction, @@ -1065,15 +1066,12 @@ export async function timeFilterExpr({ const isDateType = /^Date(?:32)?$/i.test(columnType) || isToDateExpr; // toStartOf* and Date filters must stay inclusive — strict < on a rounded value drops a whole interval + const rounded = isRoundedTimestampPart(col, columnType); const startOp = - dateRangeStartInclusive || - toStartOf || - isDateType || - flooredPrecision != null + dateRangeStartInclusive || rounded || flooredPrecision != null ? '>=' : '>'; - const endOp = - dateRangeEndInclusive || toStartOf || isDateType ? '<=' : '<'; + const endOp = dateRangeEndInclusive || rounded ? '<=' : '<'; if (isDateType) { return chSql`(${unsafeTimestampValueExpression} ${startOp} toDate(${startTimeCond}) AND ${unsafeTimestampValueExpression} ${endOp} toDate(${endTimeCond}))`; diff --git a/packages/common-utils/src/core/searchChartConfig.ts b/packages/common-utils/src/core/searchChartConfig.ts index 37144db8ab..1e019ef697 100644 --- a/packages/common-utils/src/core/searchChartConfig.ts +++ b/packages/common-utils/src/core/searchChartConfig.ts @@ -91,7 +91,7 @@ export type SearchChartConfigInput = { * so a single non-empty check covers "skip empty `''` strings and skip empty * `[]` arrays" symmetrically — matching the app's historical behavior. */ -function resolveSelect( +export function resolveSelect( source: TSource, select: SelectList | null | undefined, ): BuilderChartConfig['select'] { diff --git a/packages/app/src/utils/searchWindows.ts b/packages/common-utils/src/core/searchWindows.ts similarity index 96% rename from packages/app/src/utils/searchWindows.ts rename to packages/common-utils/src/core/searchWindows.ts index 4dd4fa61f7..d7940d051e 100644 --- a/packages/app/src/utils/searchWindows.ts +++ b/packages/common-utils/src/core/searchWindows.ts @@ -37,7 +37,7 @@ export function generateTimeWindowsDescending( while (currentEnd > startDate) { const windowSizeSeconds = - windowDurationsSeconds[windowIndex] || + windowDurationsSeconds.at(windowIndex) ?? windowDurationsSeconds[windowDurationsSeconds.length - 1]; // use largest window size const windowSizeMs = windowSizeSeconds * 1000; const windowStart = new Date( @@ -81,7 +81,7 @@ export function generateTimeWindowsAscending( while (currentStart < endDate) { const windowSizeSeconds = - windowDurationsSeconds[windowIndex] || + windowDurationsSeconds.at(windowIndex) ?? windowDurationsSeconds[windowDurationsSeconds.length - 1]; // use largest window size const windowSizeMs = windowSizeSeconds * 1000; const windowEnd = new Date( diff --git a/packages/common-utils/src/core/utils.ts b/packages/common-utils/src/core/utils.ts index 0042b9da81..bd28d49eaf 100644 --- a/packages/common-utils/src/core/utils.ts +++ b/packages/common-utils/src/core/utils.ts @@ -4,6 +4,8 @@ import { formatInTimeZone, zonedTimeToUtc } from 'date-fns-tz'; import { omit } from 'lodash'; import { z } from 'zod'; +import { stripTypeWrappers } from '@/core/eventDeltas'; + export { default as objectHash } from 'object-hash'; import { @@ -1164,6 +1166,21 @@ export const isFirstOrderByAscending = ( return !isDescending; }; +/** + * Whether timeFilterExpr filters this part of a timestamp expression with + * inclusive bounds on both ends: toStartOf*, toDate, or a Date/Date32 column. + */ +export function isRoundedTimestampPart( + part: string, + columnType: string | undefined, +): boolean { + return ( + parseToStartOfFunction(part) != null || + /^toDate\s*\(/.test(part) || + /^Date(?:32)?$/i.test(stripTypeWrappers(columnType ?? '')) + ); +} + /** * Parses a single expression of the form * `toStartOf(column[, timezone])` or `toStartOfInterval(column[, interval[, origin[, timezone]]])`.