mirror of
https://github.com/Kaelio/ktx.git
synced 2026-07-04 10:52:13 +02:00
feat(scan): apply tableScope during metadata fetch
This commit is contained in:
Andrey Avtomonov
parent
b1a2d4c378
commit
a698389bc9
34 changed files with 1129 additions and 134 deletions
|
|
@ -8,7 +8,14 @@ import type { KtxLlmRuntimePort } from '../../context/llm/runtime-port.js';
|
|||
import { initKtxProject, type KtxLocalProject, loadKtxProject } from '../../context/project/project.js';
|
||||
import { filterSnapshotTables, resolveEnabledTables } from './enabled-tables.js';
|
||||
import { getLocalScanReport, getLocalScanStatus, runLocalScan } from './local-scan.js';
|
||||
import type { KtxQueryResult, KtxReadOnlyQueryInput, KtxSchemaSnapshot, KtxSchemaTable } from './types.js';
|
||||
import { tableRefKey, tableRefSet } from './table-ref.js';
|
||||
import type {
|
||||
KtxQueryResult,
|
||||
KtxReadOnlyQueryInput,
|
||||
KtxScanConnector,
|
||||
KtxSchemaSnapshot,
|
||||
KtxSchemaTable,
|
||||
} from './types.js';
|
||||
|
||||
function relationshipSqlResult(
|
||||
input: KtxReadOnlyQueryInput,
|
||||
|
|
@ -551,6 +558,142 @@ describe('local scan', () => {
|
|||
expect(result.report.warnings).toEqual([]);
|
||||
});
|
||||
|
||||
it('keeps prototype connector methods when enabled_tables is configured', async () => {
|
||||
project.config.connections.warehouse = {
|
||||
...project.config.connections.warehouse,
|
||||
enabled_tables: ['public.customers', 'public.orders'],
|
||||
};
|
||||
const scopedAdapter: SourceAdapter = {
|
||||
source: 'live-database',
|
||||
skillNames: ['live_database_ingest'],
|
||||
async fetch(_pullConfig, stagedDir) {
|
||||
await mkdir(join(stagedDir, 'tables'), { recursive: true });
|
||||
await writeFile(
|
||||
join(stagedDir, 'connection.json'),
|
||||
'{"connectionId":"warehouse","driver":"postgres","scope":{"schemas":["public"]},"metadata":{}}\n',
|
||||
'utf-8',
|
||||
);
|
||||
await writeFile(join(stagedDir, 'foreign-keys.json'), '{"foreignKeys":[]}\n', 'utf-8');
|
||||
await writeFile(
|
||||
join(stagedDir, 'tables', 'customers.json'),
|
||||
'{"name":"customers","catalog":null,"db":"public","kind":"table","comment":null,"estimatedRows":100,"columns":[{"name":"id","nativeType":"integer","normalizedType":"integer","dimensionType":"number","nullable":false,"primaryKey":true,"comment":null}],"foreignKeys":[]}\n',
|
||||
'utf-8',
|
||||
);
|
||||
await writeFile(
|
||||
join(stagedDir, 'tables', 'orders.json'),
|
||||
'{"name":"orders","catalog":null,"db":"public","kind":"table","comment":null,"estimatedRows":1000,"columns":[{"name":"customer_id","nativeType":"integer","normalizedType":"integer","dimensionType":"number","nullable":false,"primaryKey":false,"comment":null}],"foreignKeys":[]}\n',
|
||||
'utf-8',
|
||||
);
|
||||
},
|
||||
async detect() {
|
||||
return true;
|
||||
},
|
||||
async chunk() {
|
||||
return {
|
||||
workUnits: [
|
||||
{
|
||||
unitKey: 'live-database-public-customers',
|
||||
rawFiles: ['tables/customers.json'],
|
||||
dependencyPaths: ['connection.json', 'foreign-keys.json'],
|
||||
peerFileIndex: [],
|
||||
},
|
||||
{
|
||||
unitKey: 'live-database-public-orders',
|
||||
rawFiles: ['tables/orders.json'],
|
||||
dependencyPaths: ['connection.json', 'foreign-keys.json'],
|
||||
peerFileIndex: [],
|
||||
},
|
||||
],
|
||||
};
|
||||
},
|
||||
};
|
||||
class FakeClassConnector implements KtxScanConnector {
|
||||
readonly id = 'test:warehouse';
|
||||
readonly driver = 'postgres' as const;
|
||||
readonly capabilities = {
|
||||
structuralIntrospection: true as const,
|
||||
tableSampling: false,
|
||||
columnSampling: false,
|
||||
columnStats: true,
|
||||
readOnlySql: true,
|
||||
nestedAnalysis: false,
|
||||
eventStreamDiscovery: false,
|
||||
formalForeignKeys: false,
|
||||
estimatedRowCounts: true,
|
||||
};
|
||||
|
||||
async introspect(): Promise<KtxSchemaSnapshot> {
|
||||
return {
|
||||
connectionId: 'warehouse',
|
||||
driver: 'postgres',
|
||||
extractedAt: '2026-05-22T00:00:00.000Z',
|
||||
scope: { schemas: ['public'] },
|
||||
metadata: {},
|
||||
tables: [
|
||||
{
|
||||
catalog: null,
|
||||
db: 'public',
|
||||
name: 'customers',
|
||||
kind: 'table',
|
||||
comment: null,
|
||||
estimatedRows: 100,
|
||||
foreignKeys: [],
|
||||
columns: [
|
||||
{
|
||||
name: 'id',
|
||||
nativeType: 'integer',
|
||||
normalizedType: 'integer',
|
||||
dimensionType: 'number',
|
||||
nullable: false,
|
||||
primaryKey: true,
|
||||
comment: null,
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
catalog: null,
|
||||
db: 'public',
|
||||
name: 'orders',
|
||||
kind: 'table',
|
||||
comment: null,
|
||||
estimatedRows: 1000,
|
||||
foreignKeys: [],
|
||||
columns: [
|
||||
{
|
||||
name: 'customer_id',
|
||||
nativeType: 'integer',
|
||||
normalizedType: 'integer',
|
||||
dimensionType: 'number',
|
||||
nullable: false,
|
||||
primaryKey: false,
|
||||
comment: null,
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
};
|
||||
}
|
||||
|
||||
async executeReadOnly(input: KtxReadOnlyQueryInput): Promise<KtxQueryResult> {
|
||||
return relationshipSqlResult(input);
|
||||
}
|
||||
}
|
||||
|
||||
const result = await runLocalScan({
|
||||
project,
|
||||
adapters: [scopedAdapter],
|
||||
connectionId: 'warehouse',
|
||||
mode: 'relationships',
|
||||
detectRelationships: true,
|
||||
connector: new FakeClassConnector(),
|
||||
jobId: 'scan-prototype-connector-scope',
|
||||
now: () => new Date('2026-05-22T00:00:00.000Z'),
|
||||
});
|
||||
|
||||
expect(result.report.relationships.accepted).toBe(1);
|
||||
expect(result.report.warnings).toEqual([]);
|
||||
});
|
||||
|
||||
it('threads scan relationship settings into relationship-only local scans', async () => {
|
||||
project.config.scan.enrichment = { mode: 'deterministic' };
|
||||
project.config.scan.relationships = {
|
||||
|
|
@ -1512,15 +1655,15 @@ describe('resolveEnabledTables', () => {
|
|||
expect(resolveEnabledTables({ driver: 'postgres', enabled_tables: [] })).toBeNull();
|
||||
});
|
||||
|
||||
it('returns Set of enabled table names', () => {
|
||||
it('returns a canonical set of enabled table refs', () => {
|
||||
const result = resolveEnabledTables({
|
||||
driver: 'postgres',
|
||||
enabled_tables: ['public.users', 'public.orders'],
|
||||
});
|
||||
expect(result).toBeInstanceOf(Set);
|
||||
expect(result!.size).toBe(2);
|
||||
expect(result!.has('public.users')).toBe(true);
|
||||
expect(result!.has('public.orders')).toBe(true);
|
||||
expect(result!.has(tableRefKey({ catalog: null, db: 'public', name: 'users' }))).toBe(true);
|
||||
expect(result!.has(tableRefKey({ catalog: null, db: 'public', name: 'orders' }))).toBe(true);
|
||||
});
|
||||
|
||||
it('returns null for undefined connection', () => {
|
||||
|
|
@ -1557,7 +1700,10 @@ describe('filterSnapshotTables', () => {
|
|||
{ db: 'public', name: 'orders' },
|
||||
{ db: 'public', name: 'logs' },
|
||||
]);
|
||||
const enabled = new Set(['public.users', 'public.orders']);
|
||||
const enabled = tableRefSet([
|
||||
{ catalog: null, db: 'public', name: 'users' },
|
||||
{ catalog: null, db: 'public', name: 'orders' },
|
||||
]);
|
||||
const filtered = filterSnapshotTables(snapshot, enabled);
|
||||
expect(filtered.tables).toHaveLength(2);
|
||||
expect(filtered.tables.map((t) => t.name)).toEqual(['users', 'orders']);
|
||||
|
|
@ -1565,14 +1711,14 @@ describe('filterSnapshotTables', () => {
|
|||
|
||||
it('returns empty tables when none match', () => {
|
||||
const snapshot = makeSnapshot([{ db: 'public', name: 'users' }]);
|
||||
const enabled = new Set(['public.orders']);
|
||||
const enabled = tableRefSet([{ catalog: null, db: 'public', name: 'orders' }]);
|
||||
const filtered = filterSnapshotTables(snapshot, enabled);
|
||||
expect(filtered.tables).toHaveLength(0);
|
||||
});
|
||||
|
||||
it('preserves other snapshot fields', () => {
|
||||
const snapshot = makeSnapshot([{ db: 'public', name: 'users' }]);
|
||||
const enabled = new Set(['public.users']);
|
||||
const enabled = tableRefSet([{ catalog: null, db: 'public', name: 'users' }]);
|
||||
const filtered = filterSnapshotTables(snapshot, enabled);
|
||||
expect(filtered.connectionId).toBe('test');
|
||||
expect(filtered.driver).toBe('postgres');
|
||||
|
|
|
|||
|
|
@ -25,14 +25,11 @@ import type {
|
|||
KtxConnectionDriver,
|
||||
KtxProgressPort,
|
||||
KtxScanConnector,
|
||||
KtxScanContext,
|
||||
KtxScanEnrichmentStateSummary,
|
||||
KtxScanInput,
|
||||
KtxScanMode,
|
||||
KtxScanReport,
|
||||
KtxScanTrigger,
|
||||
KtxScanWarning,
|
||||
KtxSchemaSnapshot,
|
||||
} from './types.js';
|
||||
|
||||
function enrichmentResolutionWarning(
|
||||
|
|
@ -370,17 +367,6 @@ async function readScanReport(
|
|||
}
|
||||
}
|
||||
|
||||
function createFilteredConnector(connector: KtxScanConnector, enabledTables: Set<string>): KtxScanConnector {
|
||||
return {
|
||||
...connector,
|
||||
async introspect(input: KtxScanInput, ctx: KtxScanContext): Promise<KtxSchemaSnapshot> {
|
||||
const snapshot = await connector.introspect(input, ctx);
|
||||
return filterSnapshotTables(snapshot, enabledTables);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
function withInternalLiveDatabaseAdapter(project: KtxLocalProject): KtxLocalProject {
|
||||
if (project.config.ingest.adapters.includes(LIVE_DATABASE_ADAPTER)) {
|
||||
return project;
|
||||
|
|
@ -408,8 +394,8 @@ export async function runLocalScan(options: RunLocalScanOptions): Promise<LocalS
|
|||
throw new Error(`Connection "${options.connectionId}" is not configured in ktx.yaml`);
|
||||
}
|
||||
const driver = normalizeDriver(connection.driver);
|
||||
const enabledTables = resolveEnabledTables(connection);
|
||||
const connector = rawConnector && enabledTables ? createFilteredConnector(rawConnector, enabledTables) : rawConnector;
|
||||
const tableScope = resolveEnabledTables(connection) ?? undefined;
|
||||
const connector = rawConnector;
|
||||
const adapters =
|
||||
options.adapters ??
|
||||
createDefaultLocalIngestAdapters(options.project, { databaseIntrospectionUrl: options.databaseIntrospectionUrl });
|
||||
|
|
@ -476,8 +462,8 @@ export async function runLocalScan(options: RunLocalScanOptions): Promise<LocalS
|
|||
rawSourcesDir: report.artifactPaths.rawSourcesDir,
|
||||
extractedAtFallback: report.createdAt,
|
||||
});
|
||||
const structuralSnapshot = enabledTables ? filterSnapshotTables(rawSnapshot, enabledTables) : rawSnapshot;
|
||||
if (enabledTables && structuralSnapshot.tables.length < rawSnapshot.tables.length) {
|
||||
const structuralSnapshot = tableScope ? filterSnapshotTables(rawSnapshot, tableScope) : rawSnapshot;
|
||||
if (tableScope && structuralSnapshot.tables.length < rawSnapshot.tables.length) {
|
||||
const excluded = rawSnapshot.tables.length - structuralSnapshot.tables.length;
|
||||
let remaining = excluded;
|
||||
const ds = report.diffSummary;
|
||||
|
|
|
|||
|
|
@ -12,10 +12,12 @@ export type KtxTableRefKey = string & { readonly __brand: 'KtxTableRefKey' };
|
|||
|
||||
const SEPARATOR = '\x1f';
|
||||
|
||||
/** @internal */
|
||||
export function tableRefKey(ref: KtxTableRef): KtxTableRefKey {
|
||||
return `${ref.catalog ?? ''}${SEPARATOR}${ref.db ?? ''}${SEPARATOR}${ref.name}` as KtxTableRefKey;
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
export function tableRefFromKey(key: KtxTableRefKey): KtxTableRef {
|
||||
const [catalog = '', db = '', name = ''] = key.split(SEPARATOR);
|
||||
return {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue