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

feat(ingest): read connection query history config

Browse source
This commit is contained in:
Andrey Avtomonov 2026-05-13 18:00:52 +02:00
parent 7dd14bb333
commit 03d2d26e71
4 changed files with 126 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

42
packages/context/src/ingest/local-adapters.test.ts
View file

@ -204,6 +204,46 @@ describe('local ingest adapters', () => {
});
});
it('maps connection context.queryHistory to historic-sql pull config', async () => {
const project = projectWithConnections({
warehouse: {
driver: 'postgres',
context: {
queryHistory: {
enabled: true,
windowDays: 45,
minExecutions: 7,
filters: { dropTrivialProbes: true },
},
},
},
});
const adapter = { source: 'historic-sql' } as never;
await expect(localPullConfigForAdapter(project, adapter, 'warehouse')).resolves.toMatchObject({
dialect: 'postgres',
windowDays: 45,
minExecutions: 7,
filters: { dropTrivialProbes: true },
});
});
it('prefers context.queryHistory over legacy historicSql', async () => {
const project = projectWithConnections({
warehouse: {
driver: 'postgres',
historicSql: { enabled: true, dialect: 'postgres', windowDays: 90 },
context: { queryHistory: { enabled: true, windowDays: 30 } },
},
});
const adapter = { source: 'historic-sql' } as never;
await expect(localPullConfigForAdapter(project, adapter, 'warehouse')).resolves.toMatchObject({
dialect: 'postgres',
windowDays: 30,
});
});
it('rejects local historic-sql pulls when the connection has not enabled historic SQL', async () => {
const historicSql = createDefaultLocalIngestAdapters(project, {
historicSql: {
@ -235,7 +275,7 @@ describe('local ingest adapters', () => {
});
await expect(localPullConfigForAdapter(postgresProject, historicSql!, 'warehouse')).rejects.toThrow(
'Connection "warehouse" does not have historicSql.enabled: true',
'Connection "warehouse" does not have context.queryHistory.enabled: true',
);
});

32
packages/context/src/ingest/local-adapters.ts
View file

@ -52,6 +52,7 @@ export interface DefaultLocalIngestAdaptersOptions {
postgresQueryClient?: KtxPostgresQueryClient;
now?: () => Date;
};
historicSqlPullConfigOverride?: Record<string, unknown>;
looker?: {
daemonBaseUrl?: string;
client?: Pick<LookerMappingClient, 'listLookmlModels' | 'getExplore'>;
@ -155,6 +156,28 @@ function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === 'object' && value !== null && !Array.isArray(value);
}
const historicSqlDialectByDriver = new Map<string, 'postgres' | 'bigquery' | 'snowflake'>([
['postgres', 'postgres'],
['postgresql', 'postgres'],
['bigquery', 'bigquery'],
['snowflake', 'snowflake'],
]);
function queryHistoryRecord(connection: unknown): Record<string, unknown> | null {
if (!isRecord(connection)) return null;
const context = isRecord(connection.context) ? connection.context : null;
const queryHistory = isRecord(context?.queryHistory) ? context.queryHistory : null;
return queryHistory;
}
function queryHistoryPullConfig(connection: unknown): Record<string, unknown> | null {
const queryHistory = queryHistoryRecord(connection);
if (queryHistory?.enabled !== true || !isRecord(connection)) return null;
const dialect = historicSqlDialectByDriver.get(String(connection.driver ?? '').toLowerCase());
if (!dialect) return null;
return { ...queryHistory, dialect };
}
function stringField(value: unknown): string | null {
return typeof value === 'string' && value.trim().length > 0 ? value.trim() : null;
}
@ -210,9 +233,16 @@ export async function localPullConfigForAdapter(
}
const connection = project.config.connections[connectionId];
if (adapter.source === HISTORIC_SQL_SOURCE_KEY) {
if (options.historicSqlPullConfigOverride) {
return historicSqlUnifiedPullConfigSchema.parse(options.historicSqlPullConfigOverride);
}
const queryHistory = queryHistoryPullConfig(connection);
if (queryHistory) {
return historicSqlUnifiedPullConfigSchema.parse(queryHistory);
}
const historicSql = isRecord(connection?.historicSql) ? connection.historicSql : null;
if (historicSql?.enabled !== true) {
throw new Error(`Connection "${connectionId}" does not have historicSql.enabled: true`);
throw new Error(`Connection "${connectionId}" does not have context.queryHistory.enabled: true`);
}
return historicSqlUnifiedPullConfigSchema.parse({
...historicSql,