From 75e61486b5a02fd0afb86da88330294f5de3d249 Mon Sep 17 00:00:00 2001 From: Andrey Avtomonov Date: Sun, 24 May 2026 01:09:45 +0200 Subject: [PATCH] feat(postgres): soft-fail denied constraint discovery --- .../src/connectors/postgres/connector.test.ts | 80 ++++++++++++++++++- .../cli/src/connectors/postgres/connector.ts | 69 +++++++++++++--- 2 files changed, 137 insertions(+), 12 deletions(-) diff --git a/packages/cli/src/connectors/postgres/connector.test.ts b/packages/cli/src/connectors/postgres/connector.test.ts index 052a589e..6b5a74e0 100644 --- a/packages/cli/src/connectors/postgres/connector.test.ts +++ b/packages/cli/src/connectors/postgres/connector.test.ts @@ -8,11 +8,16 @@ interface FakeQueryResult { fields?: Array<{ name: string; dataTypeID: number }>; } -function fakePoolFactory(results: Map): KtxPostgresPoolFactory { +type FakeQueryResponse = FakeQueryResult | Error; + +function fakePoolFactory(results: Map): KtxPostgresPoolFactory { const query = vi.fn(async (sql: string, params?: unknown[]) => { const normalized = sql.replace(/\s+/g, ' ').trim(); for (const [key, value] of results.entries()) { if (normalized.includes(key)) { + if (value instanceof Error) { + throw value; + } return value; } } @@ -33,8 +38,8 @@ function fakePoolFactory(results: Map): KtxPostgresPool }; } -function metadataResults(): Map { - return new Map([ +function metadataResults(): Map { + return new Map([ [ 'FROM pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n', { @@ -252,6 +257,75 @@ describe('KtxPostgresScanConnector', () => { ]); }); + it('soft-fails denied Postgres constraint discovery with scan warnings', async () => { + const results = metadataResults(); + results.set( + "tc.constraint_type = 'PRIMARY KEY'", + Object.assign(new Error('permission denied for information_schema'), { code: '42501' }), + ); + results.set( + "tc.constraint_type = 'FOREIGN KEY'", + Object.assign(new Error('relation information_schema.key_column_usage does not exist'), { code: '42P01' }), + ); + const connector = new KtxPostgresScanConnector({ + connectionId: 'warehouse', + connection: { + driver: 'postgres', + host: 'db.example.test', + database: 'analytics', + username: 'reader', + password: 'test-password', // pragma: allowlist secret + schema: 'public', + }, + poolFactory: fakePoolFactory(results), + now: () => new Date('2026-04-29T10:00:00.000Z'), + }); + + const snapshot = await connector.introspect( + { connectionId: 'warehouse', driver: 'postgres' }, + { runId: 'scan-run-denied-constraints' }, + ); + + expect(snapshot.warnings).toEqual([ + { + code: 'constraint_discovery_unauthorized', + message: 'Skipped primary-key discovery in public (insufficient grants on system catalogs)', + recoverable: true, + metadata: { schema: 'public', kind: 'primary_key' }, + }, + { + code: 'constraint_discovery_unauthorized', + message: 'Skipped foreign-key discovery in public (insufficient grants on system catalogs)', + recoverable: true, + metadata: { schema: 'public', kind: 'foreign_key' }, + }, + ]); + expect(snapshot.tables.every((table) => table.columns.every((column) => column.primaryKey === false))).toBe(true); + expect(snapshot.tables.every((table) => table.foreignKeys.length === 0)).toBe(true); + }); + + it('propagates non-denial Postgres constraint discovery errors', async () => { + const results = metadataResults(); + const resetError = Object.assign(new Error('connection reset'), { code: 'ECONNRESET' }); + results.set("tc.constraint_type = 'PRIMARY KEY'", resetError); + const connector = new KtxPostgresScanConnector({ + connectionId: 'warehouse', + connection: { + driver: 'postgres', + host: 'db.example.test', + database: 'analytics', + username: 'reader', + password: 'test-password', // pragma: allowlist secret + schema: 'public', + }, + poolFactory: fakePoolFactory(results), + }); + + await expect( + connector.introspect({ connectionId: 'warehouse', driver: 'postgres' }, { runId: 'scan-run-network-error' }), + ).rejects.toBe(resetError); + }); + it('runs samples, distinct values, statistics, read-only SQL, and schema listing', async () => { const connector = new KtxPostgresScanConnector({ connectionId: 'warehouse', diff --git a/packages/cli/src/connectors/postgres/connector.ts b/packages/cli/src/connectors/postgres/connector.ts index 9cf35799..f70cd8ce 100644 --- a/packages/cli/src/connectors/postgres/connector.ts +++ b/packages/cli/src/connectors/postgres/connector.ts @@ -2,8 +2,29 @@ import { readFileSync } from 'node:fs'; import { homedir } from 'node:os'; import { resolve } from 'node:path'; import { assertReadOnlySql, limitSqlForExecution } from '../../context/connections/read-only-sql.js'; -import { createKtxConnectorCapabilities, type KtxColumnSampleInput, type KtxColumnSampleResult, type KtxColumnStatsInput, type KtxColumnStatsResult, type KtxQueryResult, type KtxReadOnlyQueryInput, type KtxScanConnector, type KtxScanContext, type KtxScanInput, type KtxSchemaColumn, type KtxSchemaForeignKey, type KtxSchemaSnapshot, type KtxSchemaTable, type KtxTableListEntry, type KtxTableRef, type KtxTableSampleInput, type KtxTableSampleResult } from '../../context/scan/types.js'; +import { tryConstraintQuery } from '../../context/scan/constraint-discovery.js'; import { scopedTableNames } from '../../context/scan/table-ref.js'; +import { + createKtxConnectorCapabilities, + type KtxColumnSampleInput, + type KtxColumnSampleResult, + type KtxColumnStatsInput, + type KtxColumnStatsResult, + type KtxQueryResult, + type KtxReadOnlyQueryInput, + type KtxScanConnector, + type KtxScanContext, + type KtxScanInput, + type KtxScanWarning, + type KtxSchemaColumn, + type KtxSchemaForeignKey, + type KtxSchemaSnapshot, + type KtxSchemaTable, + type KtxTableListEntry, + type KtxTableRef, + type KtxTableSampleInput, + type KtxTableSampleResult, +} from '../../context/scan/types.js'; import { Pool } from 'pg'; import { KtxPostgresDialect } from './dialect.js'; @@ -208,6 +229,14 @@ function primaryKeyMap(rows: PostgresPrimaryKeyRow[]): Map> return grouped; } +function isDeniedError(error: unknown): boolean { + if (!error || typeof error !== 'object') { + return false; + } + const code = (error as { code?: unknown }).code; + return code === '42501' || code === '42P01'; +} + function queryRows(result: KtxPostgresQueryResult): unknown[][] { const headers = (result.fields ?? []).map((field) => field.name); return result.rows.map((row) => headers.map((header) => row[header])); @@ -403,10 +432,11 @@ export class KtxPostgresScanConnector implements KtxScanConnector { this.assertConnection(input.connectionId); const schemas = schemasFromConnection(this.connection); const allTables: KtxSchemaTable[] = []; + const snapshotWarnings: KtxScanWarning[] = []; for (const schema of schemas) { const scopedNames = input.tableScope ? scopedTableNames(input.tableScope, { catalog: null, db: schema }) : null; if (scopedNames && scopedNames.length === 0) continue; - const tables = await this.loadSchemaTables(schema, scopedNames); + const tables = await this.loadSchemaTables(schema, scopedNames, snapshotWarnings); allTables.push(...tables); } return { @@ -422,6 +452,7 @@ export class KtxPostgresScanConnector implements KtxScanConnector { total_columns: allTables.reduce((sum, table) => sum + table.columns.length, 0), }, tables: allTables, + warnings: snapshotWarnings, }; } @@ -570,7 +601,11 @@ export class KtxPostgresScanConnector implements KtxScanConnector { } } - private async loadSchemaTables(schema: string, scopedNames: readonly string[] | null): Promise { + private async loadSchemaTables( + schema: string, + scopedNames: readonly string[] | null, + snapshotWarnings: KtxScanWarning[], + ): Promise { if (scopedNames && scopedNames.length === 0) return []; const pgCatalogScopeClause = scopedNames ? 'AND c.relname = ANY($2)' : ''; const tableConstraintScopeClause = scopedNames ? 'AND tc.table_name = ANY($2)' : ''; @@ -615,8 +650,11 @@ export class KtxPostgresScanConnector implements KtxScanConnector { `, [schema, ...scopeValues], ); - const primaryKeys = await this.queryRaw( - ` + const primaryKeysResult = await tryConstraintQuery( + { schema, kind: 'primary_key', isDeniedError }, + () => + this.queryRaw( + ` SELECT tc.table_name, kcu.column_name FROM information_schema.table_constraints tc JOIN information_schema.key_column_usage kcu @@ -627,10 +665,18 @@ export class KtxPostgresScanConnector implements KtxScanConnector { ${tableConstraintScopeClause} ORDER BY tc.table_name, kcu.ordinal_position `, - [schema, ...scopeValues], + [schema, ...scopeValues], + ), ); - const foreignKeys = await this.queryRaw( - ` + const primaryKeys = primaryKeysResult.ok ? primaryKeysResult.value : []; + if (!primaryKeysResult.ok) { + snapshotWarnings.push(primaryKeysResult.warning); + } + const foreignKeysResult = await tryConstraintQuery( + { schema, kind: 'foreign_key', isDeniedError }, + () => + this.queryRaw( + ` SELECT tc.table_name, kcu.column_name, @@ -650,8 +696,13 @@ export class KtxPostgresScanConnector implements KtxScanConnector { ${tableConstraintScopeClause} ORDER BY tc.table_name, kcu.column_name `, - [schema, ...scopeValues], + [schema, ...scopeValues], + ), ); + const foreignKeys = foreignKeysResult.ok ? foreignKeysResult.value : []; + if (!foreignKeysResult.ok) { + snapshotWarnings.push(foreignKeysResult.warning); + } const columnsByTable = groupByTable(columns); const primaryKeysByTable = primaryKeyMap(primaryKeys);