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

fix: normalize historic sql probe results

Browse source
This commit is contained in:
Andrey Avtomonov 2026-05-11 19:24:27 +02:00
parent 83b6fd5bac
commit 1b1b32d717
9 changed files with 21 additions and 13 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

@ -33,7 +33,7 @@ describe('BigQueryHistoricSqlQueryHistoryReader', () => {
const client = queryClient([{ headers: ['1'], rows: [[1]], totalRows: 1 }]);
const reader = new BigQueryHistoricSqlQueryHistoryReader({ projectId: 'project-1', region: 'US' });
await expect(reader.probe(client)).resolves.toBeUndefined();
await expect(reader.probe(client)).resolves.toEqual({ warnings: [], info: [] });
expect(client.executeQuery).toHaveBeenCalledWith(
'SELECT 1 FROM `project-1.region-us.INFORMATION_SCHEMA.JOBS_BY_PROJECT` LIMIT 1',

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

@ -195,7 +195,7 @@ export class BigQueryHistoricSqlQueryHistoryReader {
this.viewPath = `\`${projectId}.region-${region}.INFORMATION_SCHEMA.JOBS_BY_PROJECT\``;
}
async probe(client: unknown): Promise<void> {
async probe(client: unknown): Promise<{ warnings: string[]; info: string[] }> {
let result: QueryResultLike;
try {
result = await queryClient(client).executeQuery(`SELECT 1 FROM ${this.viewPath} LIMIT 1`);
@ -205,6 +205,7 @@ export class BigQueryHistoricSqlQueryHistoryReader {
if (result.error) {
throw grantsError(result.error);
}
return { warnings: [], info: [] };
}
async *fetchAggregated(

4
packages/context/src/ingest/adapters/historic-sql/historic-sql.adapter.test.ts
View file

@ -22,7 +22,7 @@ const sqlAnalysis: SqlAnalysisPort = {
const reader: HistoricSqlReader = {
async probe() {
return { warnings: [] };
return { warnings: [], info: [] };
},
async *fetchAggregated() {},
};
@ -42,7 +42,7 @@ describe('HistoricSqlSourceAdapter', () => {
const stagedDir = await tempDir();
const aggregateReader: HistoricSqlReader = {
async probe() {
return { warnings: [] };
return { warnings: [], info: [] };
},
async *fetchAggregated() {
yield {

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

@ -52,6 +52,7 @@ describe('PostgresPgssReader aggregate path', () => {
await expect(reader.probe(client)).resolves.toEqual({
pgServerVersion: 'PostgreSQL 16.4 on x86_64-apple-darwin',
warnings: [],
info: [],
});
expect(executedSql(client, 0)).toContain("current_setting('server_version_num')::int");
@ -159,10 +160,11 @@ describe('PostgresPgssReader aggregate path', () => {
warnings: [
"pg_stat_statements.track is none; set it to top or all in the Postgres parameter group or config",
],
info: [],
});
});
it('warns when pg_stat_statements.max is below the recommended floor', async () => {
it('returns an info note when pg_stat_statements.max is below the recommended floor', async () => {
const client = queryClient([
{
headers: ['server_version_num', 'server_version'],
@ -177,7 +179,8 @@ describe('PostgresPgssReader aggregate path', () => {
await expect(reader.probe(client)).resolves.toEqual({
pgServerVersion: 'PostgreSQL 16.4',
warnings: [
warnings: [],
info: [
'pg_stat_statements.max is 1000; set it to at least 5000 to reduce query-template eviction churn',
],
});

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

@ -247,16 +247,17 @@ export class PostgresPgssReader {
const pgssMax = nullableInteger(value(maxRow, maxHeaders, 'max'));
const warnings: string[] = [];
const info: string[] = [];
if (track === 'none') {
warnings.push('pg_stat_statements.track is none; set it to top or all in the Postgres parameter group or config');
}
if (pgssMax !== null && pgssMax < RECOMMENDED_PGSS_MAX) {
warnings.push(
info.push(
`pg_stat_statements.max is ${pgssMax}; set it to at least ${RECOMMENDED_PGSS_MAX} to reduce query-template eviction churn`,
);
}
return { pgServerVersion, warnings };
return { pgServerVersion, warnings, info };
}
async *fetchAggregated(

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

@ -33,7 +33,7 @@ describe('SnowflakeHistoricSqlQueryHistoryReader', () => {
const client = queryClient([{ headers: ['1'], rows: [[1]], totalRows: 1 }]);
const reader = new SnowflakeHistoricSqlQueryHistoryReader();
await expect(reader.probe(client)).resolves.toBeUndefined();
await expect(reader.probe(client)).resolves.toEqual({ warnings: [], info: [] });
expect(client.executeQuery).toHaveBeenCalledWith(
'SELECT 1 FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY LIMIT 1',

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

@ -169,7 +169,7 @@ function mapAggregatedRow(row: unknown[], indexes: Map<string, number>): Aggrega
}
export class SnowflakeHistoricSqlQueryHistoryReader {
async probe(client: unknown): Promise<void> {
async probe(client: unknown): Promise<{ warnings: string[]; info: string[] }> {
let result: QueryResultLike;
try {
result = await queryClient(client).executeQuery(PROBE_SQL);
@ -179,6 +179,7 @@ export class SnowflakeHistoricSqlQueryHistoryReader {
if (result.error) {
throw grantsError(result.error);
}
return { warnings: [], info: [] };
}
async *fetchAggregated(

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

@ -38,7 +38,7 @@ describe('stageHistoricSqlAggregatedSnapshot', () => {
const stagedDir = await tempDir();
const reader: HistoricSqlReader = {
async probe() {
return { warnings: ['pg_stat_statements.max is low; aggregation still proceeds'] };
return { warnings: ['pg_stat_statements.track is none; aggregation still proceeds'], info: [] };
},
async *fetchAggregated() {
yield aggregate({
@ -123,7 +123,7 @@ describe('stageHistoricSqlAggregatedSnapshot', () => {
touchedTableCount: 2,
parseFailures: 1,
warnings: ['parse_failed:bad-parse'],
probeWarnings: ['pg_stat_statements.max is low; aggregation still proceeds'],
probeWarnings: ['pg_stat_statements.track is none; aggregation still proceeds'],
staleArchiveAfterDays: 90,
});

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

@ -126,6 +126,7 @@ export type StagedManifest = z.infer<typeof stagedManifestSchema>;
export interface HistoricSqlProbeResult {
warnings: string[];
info?: string[];
}
export interface HistoricSqlReader {
@ -146,9 +147,10 @@ export interface KtxPostgresQueryClient {
executeQuery(sql: string, params?: unknown[]): Promise<{ headers: string[]; rows: unknown[][]; totalRows?: number }>;
}
export interface PostgresPgssProbeResult {
export interface PostgresPgssProbeResult extends HistoricSqlProbeResult {
pgServerVersion: string;
warnings: string[];
info: string[];
}
export interface HistoricSqlSourceAdapterDeps {