apunkt/ktx
1
0
Fork 0
mirror of https://github.com/Kaelio/ktx.git synced 2026-07-16 11:31:02 +02:00
Code Issues Projects Releases Packages Wiki Activity

test: verify historic sql unified hot path

Browse source
This commit is contained in:
Andrey Avtomonov 2026-05-11 18:18:35 +02:00
parent 7549872d0f
commit 80fffcd8a1
6 changed files with 6 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

2
packages/context/src/ingest/adapters/historic-sql/bigquery-query-history-reader.test.ts
View file

@ -219,7 +219,7 @@ describe('BigQueryHistoricSqlQueryHistoryReader', () => {
for await (const row of reader.fetchAggregated( for await (const row of reader.fetchAggregated(
client, client,
{ start: new Date('2026-02-10T00:00:00.000Z'), end: new Date('2026-05-11T00:00:00.000Z') }, { start: new Date('2026-02-10T00:00:00.000Z'), end: new Date('2026-05-11T00:00:00.000Z') },
{ dialect: 'bigquery', minExecutions: 5, windowDays: 90, concurrency: 12, filters: {}, redactionPatterns: [], staleArchiveAfterDays: 90 }, { dialect: 'bigquery', minExecutions: 5, windowDays: 90, concurrency: 12, filters: { dropTrivialProbes: true }, redactionPatterns: [], staleArchiveAfterDays: 90 },
)) { )) {
rows.push(row); rows.push(row);
} }

2
packages/context/src/ingest/adapters/historic-sql/postgres-pgss-reader.test.ts
View file

@ -31,7 +31,7 @@ describe('PostgresPgssReader aggregate path', () => {
for await (const row of reader.fetchAggregated( for await (const row of reader.fetchAggregated(
{ executeQuery }, { executeQuery },
{ start: new Date('2026-02-10T00:00:00.000Z'), end: new Date('2026-05-11T00:00:00.000Z') }, { start: new Date('2026-02-10T00:00:00.000Z'), end: new Date('2026-05-11T00:00:00.000Z') },
{ dialect: 'postgres', minExecutions: 5, windowDays: 90, concurrency: 12, filters: {}, redactionPatterns: [], staleArchiveAfterDays: 90 }, { dialect: 'postgres', minExecutions: 5, windowDays: 90, concurrency: 12, filters: { dropTrivialProbes: true }, redactionPatterns: [], staleArchiveAfterDays: 90 },
)) { )) {
rows.push(row); rows.push(row);
} }

2
packages/context/src/ingest/adapters/historic-sql/snowflake-query-history-reader.test.ts
View file

@ -221,7 +221,7 @@ describe('SnowflakeHistoricSqlQueryHistoryReader', () => {
for await (const row of reader.fetchAggregated( for await (const row of reader.fetchAggregated(
client, client,
{ start: new Date('2026-02-10T00:00:00.000Z'), end: new Date('2026-05-11T00:00:00.000Z') }, { start: new Date('2026-02-10T00:00:00.000Z'), end: new Date('2026-05-11T00:00:00.000Z') },
{ dialect: 'snowflake', minExecutions: 5, windowDays: 90, concurrency: 12, filters: {}, redactionPatterns: [], staleArchiveAfterDays: 90 }, { dialect: 'snowflake', minExecutions: 5, windowDays: 90, concurrency: 12, filters: { dropTrivialProbes: true }, redactionPatterns: [], staleArchiveAfterDays: 90 },
)) { )) {
rows.push(row); rows.push(row);
} }

4
packages/context/src/ingest/adapters/historic-sql/stage-unified.ts
View file

@ -152,14 +152,14 @@ function addTemplate(acc: TableAccumulator, parsed: ParsedTemplate): void {
function toStagedTable(acc: TableAccumulator, now: Date): StagedTableInput { function toStagedTable(acc: TableAccumulator, now: Date): StagedTableInput {
const errorRate = acc.executions > 0 ? acc.errorRateNumerator / acc.executions : 0; const errorRate = acc.executions > 0 ? acc.errorRateNumerator / acc.executions : 0;
const columnsByClause = Object.fromEntries( const columnsByClause: Record<string, Array<[string, string]>> = Object.fromEntries(
[...acc.columnsByClause.entries()] [...acc.columnsByClause.entries()]
.sort(([left], [right]) => left.localeCompare(right)) .sort(([left], [right]) => left.localeCompare(right))
.map(([clause, counts]) => [ .map(([clause, counts]) => [
clause, clause,
[...counts.entries()] [...counts.entries()]
.sort((left, right) => right[1] - left[1] || left[0].localeCompare(right[0])) .sort((left, right) => right[1] - left[1] || left[0].localeCompare(right[0]))
.map(([column, count]) => [column, bucketFrequency(count, acc.executions)]), .map(([column, count]) => [column, bucketFrequency(count, acc.executions)] as [string, string]),
]), ]),
); );
const observedJoins = [...acc.observedJoins.entries()] const observedJoins = [...acc.observedJoins.entries()]

2
packages/context/src/ingest/adapters/historic-sql/types.ts
View file

@ -50,7 +50,7 @@ export const historicSqlUnifiedPullConfigSchema = z.preprocess((value) => {
errorRate: z.number().min(0).max(1), errorRate: z.number().min(0).max(1),
executions: z.number().int().nonnegative(), executions: z.number().int().nonnegative(),
}).optional(), }).optional(),
}).default({}), }).default({ dropTrivialProbes: true }),
redactionPatterns: z.array(z.string()).default([]), redactionPatterns: z.array(z.string()).default([]),
staleArchiveAfterDays: z.number().int().positive().default(90), staleArchiveAfterDays: z.number().int().positive().default(90),
})); }));

1
packages/context/src/ingest/index.ts
View file

@ -367,7 +367,6 @@ export type {
KtxPostgresQueryClient, KtxPostgresQueryClient,
PostgresPgssAggregateRow, PostgresPgssAggregateRow,
PostgresPgssProbeResult, PostgresPgssProbeResult,
PostgresPgssReader,
PostgresPgssRow, PostgresPgssRow,
PostgresPgssSnapshot, PostgresPgssSnapshot,
StagedManifest, StagedManifest,