From 3a184c86ba21762bfd4d39d1c7bfa0d147225250 Mon Sep 17 00:00:00 2001 From: Mayorkun Ayanshina Date: Wed, 17 Jun 2026 10:16:04 +0100 Subject: [PATCH 1/9] feat(redshift): add Amazon Redshift connector Add a redshift connector modeled on the Postgres connector (Redshift speaks the Postgres wire protocol), using Redshift system views for introspection: SVV_TABLES and SVV_TABLE_INFO for tables/row counts, SVV_COLUMNS for columns. Wires the driver into the type union, dialect factory, driver registry, and connection schema. --- .../cli/src/connectors/redshift/connector.ts | 834 ++++++++++++++++++ .../cli/src/connectors/redshift/dialect.ts | 210 +++++ .../redshift/live-database-introspection.ts | 47 + .../cli/src/context/connections/dialects.ts | 2 + .../cli/src/context/connections/drivers.ts | 21 + .../cli/src/context/project/driver-schemas.ts | 2 + packages/cli/src/context/scan/types.ts | 3 +- .../test/context/connections/dialects.test.ts | 32 +- .../test/context/connections/drivers.test.ts | 10 + 9 files changed, 1159 insertions(+), 2 deletions(-) create mode 100644 packages/cli/src/connectors/redshift/connector.ts create mode 100644 packages/cli/src/connectors/redshift/dialect.ts create mode 100644 packages/cli/src/connectors/redshift/live-database-introspection.ts diff --git a/packages/cli/src/connectors/redshift/connector.ts b/packages/cli/src/connectors/redshift/connector.ts new file mode 100644 index 000000000..25698a0d0 --- /dev/null +++ b/packages/cli/src/connectors/redshift/connector.ts @@ -0,0 +1,834 @@ +import { resolveStringReference } from '../shared/string-reference.js'; +import { getDialectForDriver } from '../../context/connections/dialects.js'; +import { assertReadOnlySql, limitSqlForExecution } from '../../context/connections/read-only-sql.js'; +import { tryConstraintQuery } from '../../context/scan/constraint-discovery.js'; +import { scopedTableNames } from '../../context/scan/table-ref.js'; +import { + connectorTestFailure, + createKtxConnectorCapabilities, + type KtxConnectorTestResult, + 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'; + +const REDSHIFT_OID_TYPE_MAP: Record = { + 16: 'boolean', + 20: 'bigint', + 21: 'smallint', + 23: 'integer', + 25: 'text', + 700: 'real', + 701: 'double precision', + 1043: 'varchar', + 1082: 'date', + 1114: 'timestamp', + 1184: 'timestamptz', + 1700: 'numeric', + 2950: 'uuid', + 3802: 'jsonb', + 114: 'json', + 1009: 'text[]', + 1007: 'integer[]', + 1016: 'bigint[]', +}; + +export interface KtxRedshiftConnectionConfig { + driver?: string; + host?: string; + port?: number; + database?: string; + username?: string; + user?: string; + password?: string; + url?: string; + schema?: string; + schemas?: string[]; + ssl?: boolean; + sslmode?: string; + sslMode?: string; + rejectUnauthorized?: boolean; + maxConnections?: number; + [key: string]: unknown; +} + +export interface KtxRedshiftPoolConfig { + host?: string; + port?: number; + database?: string; + user?: string; + password?: string; + connectionString?: string; + max: number; + idleTimeoutMillis: number; + connectionTimeoutMillis: number; + options?: string; + ssl?: { rejectUnauthorized: boolean }; +} + +interface KtxRedshiftQueryResult { + fields?: Array<{ name: string; dataTypeID: number }>; + rows: Record[]; +} + +interface KtxRedshiftClient { + query(sql: string, params?: unknown[]): Promise; + release(): void; +} + +interface KtxRedshiftPool { + connect(): Promise; + end(): Promise; + on?(event: 'error', listener: (error: Error) => void): void; +} + +export interface KtxRedshiftPoolFactory { + createPool(config: KtxRedshiftPoolConfig): KtxRedshiftPool; +} + +interface KtxRedshiftResolvedEndpoint { + host: string; + port: number; + close?: () => Promise; +} + +export interface KtxRedshiftEndpointResolver { + resolve(input: { + host: string; + port: number; + connection: KtxRedshiftConnectionConfig; + }): Promise; +} + +export interface KtxRedshiftScanConnectorOptions { + connectionId: string; + connection: KtxRedshiftConnectionConfig | undefined; + poolFactory?: KtxRedshiftPoolFactory; + endpointResolver?: KtxRedshiftEndpointResolver; + env?: NodeJS.ProcessEnv; + now?: () => Date; +} + +export interface KtxRedshiftReadOnlyQueryInput extends KtxReadOnlyQueryInput { + params?: Record | unknown[]; +} + +export interface KtxRedshiftColumnDistinctValuesOptions { + maxCardinality: number; + limit: number; + sampleSize?: number; +} + +export interface KtxRedshiftColumnDistinctValuesResult { + values: string[] | null; + cardinality: number; +} + +export interface KtxRedshiftColumnStatisticsResult { + cardinalityByColumn: Map; +} + +export interface KtxRedshiftTableSampleResult extends KtxTableSampleResult { + headerTypes?: string[]; +} + +type RedshiftTableRef = Pick & Partial>; + +interface RedshiftTableRow { + table_name: string; + table_kind: string; + row_count: unknown; + table_comment: string | null; +} + +interface RedshiftColumnRow { + table_name: string; + column_name: string; + data_type: string; + is_nullable: boolean; + column_comment: string | null; +} + +interface RedshiftPrimaryKeyRow { + table_name: string; + column_name: string; +} + +interface RedshiftForeignKeyRow { + table_name: string; + column_name: string; + foreign_table_schema: string | null; + foreign_table_name: string; + foreign_column_name: string; + constraint_name: string | null; +} + +interface RedshiftSchemaRow { + schema_name: string; +} + +interface RedshiftTableListRow { + schema_name: string; + table_name: string; + table_kind: string; +} + +interface RedshiftCountRow { + count?: unknown; + cardinality?: unknown; +} + +interface RedshiftDistinctValueRow { + val: unknown; +} + +interface RedshiftStatsRow { + column_name: string; + estimated_cardinality: unknown; +} + +class DefaultRedshiftPoolFactory implements KtxRedshiftPoolFactory { + createPool(config: KtxRedshiftPoolConfig): KtxRedshiftPool { + return new Pool(config); + } +} + +function groupByTable(rows: T[]): Map { + const grouped = new Map(); + for (const row of rows) { + const tableRows = grouped.get(row.table_name) ?? []; + tableRows.push(row); + grouped.set(row.table_name, tableRows); + } + return grouped; +} + +/** @internal */ +export function prepareRedshiftReadOnlyQuery( + sql: string, + params?: Record, +): { sql: string; params?: unknown[] } { + if (!params) { + return { sql, params: undefined }; + } + const paramNames = Object.keys(params); + const values: unknown[] = new Array(paramNames.length); + const paramIndexMap = new Map(); + paramNames.forEach((name, index) => { + paramIndexMap.set(name, index + 1); + values[index] = params[name]; + }); + const sortedKeys = [...paramNames].sort((a, b) => b.length - a.length); + let parameterizedQuery = sql; + for (const name of sortedKeys) { + parameterizedQuery = parameterizedQuery.replace(new RegExp(`:${name}\\b`, 'g'), `$${paramIndexMap.get(name)}`); + } + return { sql: parameterizedQuery, params: values }; +} + +function primaryKeyMap(rows: RedshiftPrimaryKeyRow[]): Map> { + const grouped = new Map>(); + for (const row of rows) { + const columns = grouped.get(row.table_name) ?? new Set(); + columns.add(row.column_name); + grouped.set(row.table_name, columns); + } + 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: KtxRedshiftQueryResult): unknown[][] { + const headers = (result.fields ?? []).map((field) => field.name); + return result.rows.map((row) => headers.map((header) => row[header])); +} + +function finiteNumber(value: unknown): number | null { + const parsed = Number(value); + return Number.isFinite(parsed) ? parsed : null; +} + +function stringConfigValue( + connection: KtxRedshiftConnectionConfig | undefined, + key: keyof KtxRedshiftConnectionConfig, + env: NodeJS.ProcessEnv, +): string | undefined { + const value = connection?.[key]; + return typeof value === 'string' && value.trim().length > 0 ? resolveStringReference(value.trim(), env) : undefined; +} + + +function numberValue(value: unknown): number | undefined { + return typeof value === 'number' && Number.isFinite(value) ? value : undefined; +} + +function positiveIntegerConfigValue(input: { + connection: KtxRedshiftConnectionConfig; + key: keyof KtxRedshiftConnectionConfig; + connectionId: string; + defaultValue: number; +}): number { + const value = input.connection[input.key]; + if (value === undefined) { + return input.defaultValue; + } + const numberValue = Number(value); + if (!Number.isInteger(numberValue) || numberValue < 1) { + throw new Error(`connections.${input.connectionId}.${String(input.key)} must be a positive integer`); + } + return numberValue; +} + +function parseRedshiftUrl(url: string): Partial { + const parsed = new URL(url); + const sslmode = parsed.searchParams.get('sslmode') ?? undefined; + return { + host: parsed.hostname, + port: parsed.port ? Number(parsed.port) : undefined, + database: parsed.pathname.replace(/^\/+/, '') || undefined, + username: parsed.username ? decodeURIComponent(parsed.username) : undefined, + password: parsed.password ? decodeURIComponent(parsed.password) : undefined, + ...(sslmode ? { sslmode } : {}), + }; +} + +function normalizedSslMode(connection: KtxRedshiftConnectionConfig): string | undefined { + const value = connection.sslmode ?? connection.sslMode; + return typeof value === 'string' && value.trim().length > 0 ? value.trim().toLowerCase() : undefined; +} + +function schemasFromConnection(connection: KtxRedshiftConnectionConfig): string[] { + if (Array.isArray(connection.schemas) && connection.schemas.length > 0) { + return connection.schemas.filter((schema): schema is string => typeof schema === 'string' && schema.length > 0); + } + return typeof connection.schema === 'string' && connection.schema.length > 0 ? [connection.schema] : ['public']; +} + +function searchPathSchemasFromConnection(connection: KtxRedshiftConnectionConfig): string[] { + const schemas = schemasFromConnection(connection); + return schemas.includes('public') ? schemas : [...schemas, 'public']; +} + +export function isKtxRedshiftConnectionConfig( + connection: KtxRedshiftConnectionConfig | undefined, +): connection is KtxRedshiftConnectionConfig { + const driver = String(connection?.driver ?? '').toLowerCase(); + return driver === 'redshift'; +} + +/** @internal */ +export function redshiftPoolConfigFromConfig(input: { + connectionId: string; + connection: KtxRedshiftConnectionConfig | undefined; + env?: NodeJS.ProcessEnv; +}): KtxRedshiftPoolConfig { + const inputDriver = input.connection?.driver ?? 'unknown'; + if (!isKtxRedshiftConnectionConfig(input.connection)) { + throw new Error(`Native Redshift connector cannot run driver "${inputDriver}"`); + } + + const env = input.env ?? process.env; + const referencedUrl = stringConfigValue(input.connection, 'url', env); + const urlConfig = referencedUrl ? parseRedshiftUrl(referencedUrl) : {}; + const merged: KtxRedshiftConnectionConfig = { ...urlConfig, ...input.connection }; + const host = stringConfigValue(merged, 'host', env); + const database = stringConfigValue(merged, 'database', env); + const user = stringConfigValue(merged, 'username', env) ?? stringConfigValue(merged, 'user', env); + const password = stringConfigValue(merged, 'password', env); + const sslmode = normalizedSslMode(merged); + const maxConnections = positiveIntegerConfigValue({ + connection: merged, + key: 'maxConnections', + connectionId: input.connectionId, + defaultValue: 10, + }); + + if (!referencedUrl && !host) { + throw new Error(`Native Redshift connector requires connections.${input.connectionId}.host or url`); + } + if (!database && !referencedUrl) { + throw new Error(`Native Redshift connector requires connections.${input.connectionId}.database or url`); + } + if (!user && !referencedUrl) { + throw new Error(`Native Redshift connector requires connections.${input.connectionId}.username, user, or url`); + } + + const config: KtxRedshiftPoolConfig = { + max: maxConnections, + idleTimeoutMillis: 30_000, + connectionTimeoutMillis: 10_000, + ...(referencedUrl && sslmode !== 'prefer' && sslmode !== 'disable' + ? { connectionString: referencedUrl } + : { host, port: numberValue(merged.port) ?? 5439, database, user, password }), + }; + const searchPathSchemas = searchPathSchemasFromConnection(merged); + if (searchPathSchemas.length > 0) { + config.options = `-c search_path=${searchPathSchemas.join(',')}`; + } + if (merged.ssl && sslmode !== 'prefer' && sslmode !== 'disable') { + config.ssl = { rejectUnauthorized: merged.rejectUnauthorized ?? true }; + } + return config; +} + +export class KtxRedshiftScanConnector implements KtxScanConnector { + readonly id: string; + readonly driver = 'redshift' as const; + readonly capabilities = createKtxConnectorCapabilities({ + tableSampling: true, + columnSampling: true, + columnStats: true, + readOnlySql: true, + nestedAnalysis: true, + formalForeignKeys: true, + estimatedRowCounts: true, + }); + + private readonly connectionId: string; + private readonly connection: KtxRedshiftConnectionConfig; + private readonly poolConfig: KtxRedshiftPoolConfig; + private readonly poolFactory: KtxRedshiftPoolFactory; + private readonly endpointResolver?: KtxRedshiftEndpointResolver; + private readonly now: () => Date; + private readonly dialect = getDialectForDriver('redshift'); + private pool: KtxRedshiftPool | null = null; + private lastIdlePoolError: Error | null = null; + private resolvedEndpoint: KtxRedshiftResolvedEndpoint | null = null; + + constructor(options: KtxRedshiftScanConnectorOptions) { + this.connectionId = options.connectionId; + this.connection = options.connection ?? {}; + this.poolConfig = redshiftPoolConfigFromConfig({ + connectionId: options.connectionId, + connection: options.connection, + env: options.env, + }); + this.poolFactory = options.poolFactory ?? new DefaultRedshiftPoolFactory(); + this.endpointResolver = options.endpointResolver; + this.now = options.now ?? (() => new Date()); + this.id = `redshift:${options.connectionId}`; + } + + async testConnection(): Promise { + try { + await this.query('SELECT 1'); + return { success: true }; + } catch (error) { + return connectorTestFailure(error); + } + } + + async introspect(input: KtxScanInput, _ctx: KtxScanContext): Promise { + 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, snapshotWarnings); + allTables.push(...tables); + } + return { + connectionId: this.connectionId, + driver: 'redshift', + extractedAt: this.now().toISOString(), + scope: { schemas }, + metadata: { + database: this.poolConfig.database ?? this.connection.database ?? null, + schemas, + host: this.poolConfig.host ?? this.connection.host ?? null, + table_count: allTables.length, + total_columns: allTables.reduce((sum, table) => sum + table.columns.length, 0), + }, + tables: allTables, + warnings: snapshotWarnings, + }; + } + + async sampleTable(input: KtxTableSampleInput, _ctx: KtxScanContext): Promise { + this.assertConnection(input.connectionId); + const result = await this.query(this.dialect.generateSampleQuery(this.qTableName(input.table), input.limit, input.columns)); + return { + headers: result.headers, + headerTypes: result.headerTypes, + rows: result.rows, + totalRows: result.totalRows, + }; + } + + async sampleColumn(input: KtxColumnSampleInput, _ctx: KtxScanContext): Promise { + this.assertConnection(input.connectionId); + const result = await this.query( + this.dialect.generateColumnSampleQuery(this.qTableName(input.table), input.column, input.limit), + ); + const values = result.rows.filter((row) => row.length > 0 && row[0] !== null).map((row) => row[0]); + return { values, nullCount: null, distinctCount: null }; + } + + async columnStats(input: KtxColumnStatsInput, _ctx: KtxScanContext): Promise { + const stats = await this.getColumnStatistics(input.table); + const value = stats?.cardinalityByColumn.get(input.column); + return value === undefined + ? null + : { min: null, max: null, average: null, nullCount: null, distinctCount: value }; + } + + async executeReadOnly(input: KtxRedshiftReadOnlyQueryInput, _ctx: KtxScanContext): Promise { + this.assertConnection(input.connectionId); + const limitedSql = limitSqlForExecution(assertReadOnlySql(input.sql), input.maxRows); + const prepared = Array.isArray(input.params) + ? { sql: limitedSql, params: input.params } + : prepareRedshiftReadOnlyQuery(limitedSql, input.params); + const result = await this.query(prepared.sql, prepared.params); + return { ...result, rowCount: result.rows.length }; + } + + async getColumnDistinctValues( + table: KtxTableRef, + columnName: string, + options: KtxRedshiftColumnDistinctValuesOptions, + ): Promise { + const sampleSize = options.sampleSize ?? 10000; + const tableName = this.qTableName(table); + const quotedColumn = this.dialect.quoteIdentifier(columnName); + const cardinalityRows = await this.queryRaw( + this.dialect.generateCardinalitySampleQuery(tableName, quotedColumn, sampleSize), + ); + const cardinality = finiteNumber(cardinalityRows[0]?.cardinality); + if (cardinality === null) { + return null; + } + if (cardinality === 0) { + return { values: [], cardinality: 0 }; + } + if (cardinality > options.maxCardinality) { + return { values: null, cardinality }; + } + const valuesRows = await this.queryRaw( + this.dialect.generateDistinctValuesQuery(tableName, quotedColumn, options.limit), + ); + return { + values: valuesRows.filter((row) => row.val !== null).map((row) => String(row.val)), + cardinality, + }; + } + + async getColumnStatistics(table: KtxTableRef): Promise { + const schema = table.db ?? schemasFromConnection(this.connection)[0] ?? 'public'; + const sql = this.dialect.generateColumnStatisticsQuery(schema, table.name); + if (!sql) { + return null; + } + const rows = await this.queryRaw(sql); + const cardinalityByColumn = new Map(); + for (const row of rows) { + const cardinality = finiteNumber(row.estimated_cardinality); + if (cardinality !== null) { + cardinalityByColumn.set(row.column_name, cardinality); + } + } + return cardinalityByColumn.size > 0 ? { cardinalityByColumn } : null; + } + + async getTableRowCount(table: string | RedshiftTableRef): Promise { + const tableRef = + typeof table === 'string' + ? { catalog: null, db: schemasFromConnection(this.connection)[0] ?? 'public', name: table } + : table; + const rows = await this.queryRaw(`SELECT COUNT(*) AS count FROM ${this.qTableName(tableRef)}`); + return finiteNumber(rows[0]?.count) ?? 0; + } + + qTableName(table: RedshiftTableRef): string { + return this.dialect.formatTableName(table); + } + + quoteIdentifier(identifier: string): string { + return this.dialect.quoteIdentifier(identifier); + } + + async listSchemas(): Promise { + const rows = await this.queryRaw(` + SELECT schema_name + FROM information_schema.schemata + WHERE schema_name <> 'information_schema' + AND schema_name NOT LIKE 'pg_%' + ORDER BY schema_name + `); + return rows.map((row) => row.schema_name); + } + + async listTables(schemas?: string[]): Promise { + const filterSchemas = schemas ?? (await this.listSchemas()); + if (filterSchemas.length === 0) return []; + const rows = await this.queryRaw( + ` + SELECT table_schema AS schema_name, table_name AS table_name, + CASE WHEN table_type = 'VIEW' THEN 'v' ELSE 'r' END AS table_kind + FROM svv_tables + WHERE table_schema = ANY($1) + AND table_type IN ('TABLE', 'VIEW', 'EXTERNAL TABLE') + ORDER BY table_schema, table_name + `, + [filterSchemas], + ); + return rows.map((row) => ({ + catalog: null, + schema: row.schema_name, + name: row.table_name, + kind: row.table_kind === 'v' ? ('view' as const) : ('table' as const), + })); + } + + async cleanup(): Promise { + if (this.pool) { + await this.pool.end(); + this.pool = null; + } + if (this.resolvedEndpoint?.close) { + await this.resolvedEndpoint.close(); + this.resolvedEndpoint = null; + } + } + + private async loadSchemaTables( + schema: string, + scopedNames: readonly string[] | null, + snapshotWarnings: KtxScanWarning[], + ): Promise { + if (scopedNames && scopedNames.length === 0) return []; + const svvTableScopeClause = scopedNames ? 'AND t.table_name = ANY($2)' : ''; + const svvColumnScopeClause = scopedNames ? 'AND table_name = ANY($2)' : ''; + const tableConstraintScopeClause = scopedNames ? 'AND tc.table_name = ANY($2)' : ''; + const scopeValues = scopedNames ? [scopedNames] : []; + const tables = await this.queryRaw( + ` + SELECT + t.table_name AS table_name, + CASE WHEN t.table_type = 'VIEW' THEN 'v' ELSE 'r' END AS table_kind, + ti.tbl_rows AS row_count, + t.remarks AS table_comment + FROM svv_tables t + LEFT JOIN svv_table_info ti + ON ti.schema = t.table_schema AND ti.\"table\" = t.table_name + WHERE t.table_schema = $1 + AND t.table_type IN ('TABLE', 'VIEW', 'EXTERNAL TABLE') + ${svvTableScopeClause} + ORDER BY t.table_name + `, + [schema, ...scopeValues], + ); + const columns = await this.queryRaw( + ` + SELECT + table_name AS table_name, + column_name AS column_name, + data_type AS data_type, + CASE WHEN is_nullable = 'YES' THEN TRUE ELSE FALSE END AS is_nullable, + remarks AS column_comment + FROM svv_columns + WHERE table_schema = $1 + ${svvColumnScopeClause} + ORDER BY table_name, ordinal_position + `, + [schema, ...scopeValues], + ); + 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 + ON tc.constraint_name = kcu.constraint_name + AND tc.table_schema = kcu.table_schema + WHERE tc.constraint_type = 'PRIMARY KEY' + AND tc.table_schema = $1 + ${tableConstraintScopeClause} + ORDER BY tc.table_name, kcu.ordinal_position + `, + [schema, ...scopeValues], + ), + ); + 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, + ccu.table_schema AS foreign_table_schema, + ccu.table_name AS foreign_table_name, + ccu.column_name AS foreign_column_name, + tc.constraint_name + FROM information_schema.table_constraints AS tc + JOIN information_schema.key_column_usage AS kcu + ON tc.constraint_name = kcu.constraint_name + AND tc.table_schema = kcu.table_schema + JOIN information_schema.constraint_column_usage AS ccu + ON ccu.constraint_name = tc.constraint_name + AND ccu.table_schema = tc.table_schema + WHERE tc.constraint_type = 'FOREIGN KEY' + AND tc.table_schema = $1 + ${tableConstraintScopeClause} + ORDER BY tc.table_name, kcu.column_name + `, + [schema, ...scopeValues], + ), + ); + const foreignKeys = foreignKeysResult.ok ? foreignKeysResult.value : []; + if (!foreignKeysResult.ok) { + snapshotWarnings.push(foreignKeysResult.warning); + } + + const columnsByTable = groupByTable(columns); + const primaryKeysByTable = primaryKeyMap(primaryKeys); + const foreignKeysByTable = groupByTable(foreignKeys); + return tables.map((table) => + this.toSchemaTable( + schema, + table, + columnsByTable.get(table.table_name) ?? [], + primaryKeysByTable.get(table.table_name) ?? new Set(), + foreignKeysByTable.get(table.table_name) ?? [], + ), + ); + } + + private toSchemaTable( + schema: string, + table: RedshiftTableRow, + columns: RedshiftColumnRow[], + primaryKeys: Set, + foreignKeys: RedshiftForeignKeyRow[], + ): KtxSchemaTable { + const kind = table.table_kind === 'v' ? 'view' : 'table'; + return { + catalog: null, + db: schema, + name: table.table_name, + kind, + comment: table.table_comment || null, + estimatedRows: kind === 'view' ? null : finiteNumber(table.row_count), + columns: columns.map((column) => this.toSchemaColumn(column, primaryKeys)), + foreignKeys: foreignKeys.map((foreignKey) => this.toSchemaForeignKey(foreignKey)), + }; + } + + private toSchemaColumn(column: RedshiftColumnRow, primaryKeys: Set): KtxSchemaColumn { + return { + name: column.column_name, + nativeType: column.data_type, + normalizedType: this.dialect.mapDataType(column.data_type), + dimensionType: this.dialect.mapToDimensionType(column.data_type), + nullable: column.is_nullable, + primaryKey: primaryKeys.has(column.column_name), + comment: column.column_comment || null, + }; + } + + private toSchemaForeignKey(row: RedshiftForeignKeyRow): KtxSchemaForeignKey { + return { + fromColumn: row.column_name, + toCatalog: null, + toDb: row.foreign_table_schema, + toTable: row.foreign_table_name, + toColumn: row.foreign_column_name, + constraintName: row.constraint_name || null, + }; + } + + private async getPool(): Promise { + if (!this.pool) { + let config = { ...this.poolConfig }; + if (this.endpointResolver) { + const endpoint = await this.endpointResolver.resolve({ + host: config.host ?? this.connection.host ?? 'localhost', + port: config.port ?? numberValue(this.connection.port) ?? 5439, + connection: this.connection, + }); + this.resolvedEndpoint = endpoint; + config = { ...config, host: endpoint.host, port: endpoint.port }; + } + this.pool = this.poolFactory.createPool(config); + this.pool.on?.('error', (error) => { + this.lastIdlePoolError = error; + }); + } + return this.pool; + } + + private async queryRaw(sql: string, params?: unknown[]): Promise { + this.throwIdlePoolErrorIfPresent(); + const pool = await this.getPool(); + const client = await pool.connect(); + try { + const result = await client.query(sql, params); + return result.rows as T[]; + } finally { + client.release(); + } + } + + private async query(sql: string, params?: Record | unknown[]): Promise { + this.throwIdlePoolErrorIfPresent(); + const pool = await this.getPool(); + const client = await pool.connect(); + try { + const result = await client.query(assertReadOnlySql(sql), Array.isArray(params) ? params : undefined); + return { + headers: (result.fields ?? []).map((field) => field.name), + headerTypes: (result.fields ?? []).map((field) => REDSHIFT_OID_TYPE_MAP[field.dataTypeID] ?? `oid:${field.dataTypeID}`), + rows: queryRows(result), + totalRows: result.rows.length, + rowCount: result.rows.length, + }; + } finally { + client.release(); + } + } + + private assertConnection(connectionId: string): void { + if (connectionId !== this.connectionId) { + throw new Error(`Redshift connector ${this.connectionId} cannot run scan for ${connectionId}`); + } + } + + private throwIdlePoolErrorIfPresent(): void { + if (!this.lastIdlePoolError) { + return; + } + const error = this.lastIdlePoolError; + this.lastIdlePoolError = null; + throw error; + } +} diff --git a/packages/cli/src/connectors/redshift/dialect.ts b/packages/cli/src/connectors/redshift/dialect.ts new file mode 100644 index 000000000..462269082 --- /dev/null +++ b/packages/cli/src/connectors/redshift/dialect.ts @@ -0,0 +1,210 @@ +import type { KtxDialect } from '../../context/connections/dialects.js'; +import { + columnDisplayPartCount, + formatDialectDisplayRef, + formatDialectTableName, + limitOffsetClause, + parseDialectDisplayRef, +} from '../../context/connections/dialect-helpers.js'; +import type { KtxSchemaDimensionType, KtxTableRef } from '../../context/scan/types.js'; + +type RedshiftTableNameRef = Pick & Partial>; + +/** @internal */ +export class KtxRedshiftDialect implements KtxDialect { + readonly type = 'redshift' as const; + + private readonly typeMappings: Record = { + timestamp: 'time', + 'timestamp without time zone': 'time', + 'timestamp with time zone': 'time', + timestamptz: 'time', + datetime: 'time', + date: 'time', + time: 'time', + integer: 'number', + int: 'number', + int2: 'number', + int4: 'number', + int8: 'number', + bigint: 'number', + smallint: 'number', + decimal: 'number', + numeric: 'number', + float: 'number', + float4: 'number', + float8: 'number', + 'double precision': 'number', + real: 'number', + money: 'number', + text: 'string', + varchar: 'string', + 'character varying': 'string', + char: 'string', + character: 'string', + uuid: 'string', + json: 'string', + jsonb: 'string', + boolean: 'boolean', + bool: 'boolean', + }; + + quoteIdentifier(identifier: string): string { + return `"${identifier.replace(/"/g, '""')}"`; + } + + formatTableName(table: RedshiftTableNameRef): string { + return formatDialectTableName(table, this.quoteIdentifier.bind(this), 'ansi'); + } + + formatDisplayRef(table: RedshiftTableNameRef): string { + return formatDialectDisplayRef(table, 'ansi'); + } + + parseDisplayRef(display: string): KtxTableRef | null { + return parseDialectDisplayRef(display, 'ansi'); + } + + columnDisplayTablePartCount(): 1 | 2 | 3 { + return columnDisplayPartCount('ansi'); + } + + mapDataType(nativeType: string): string { + return nativeType; + } + + mapToDimensionType(nativeType: string): KtxSchemaDimensionType { + if (!nativeType) { + return 'string'; + } + const lower = nativeType.toLowerCase().trim(); + const normalized = lower.includes('(') ? lower.split('(')[0]!.trim() : lower; + if (this.typeMappings[normalized]) { + return this.typeMappings[normalized]; + } + if (normalized.includes('time') || normalized.includes('date')) { + return 'time'; + } + if ( + normalized.includes('int') || + normalized.includes('num') || + normalized.includes('dec') || + normalized.includes('float') || + normalized.includes('double') + ) { + return 'number'; + } + if (normalized.includes('bool')) { + return 'boolean'; + } + return 'string'; + } + + generateSampleQuery(tableName: string, limit: number, columns?: string[]): string { + const columnList = + columns && columns.length > 0 ? columns.map((column) => this.quoteIdentifier(column)).join(', ') : '*'; + return `SELECT ${columnList} FROM ${tableName} LIMIT ${limit}`; + } + + generateColumnSampleQuery(tableName: string, columnName: string, limit: number): string { + const quotedColumn = this.quoteIdentifier(columnName); + return `SELECT ${quotedColumn} FROM ${tableName} WHERE ${quotedColumn} IS NOT NULL AND TRIM(CAST(${quotedColumn} AS TEXT)) != '' LIMIT ${limit}`; + } + + getRandomSampleFilter(samplePct: number): string { + if (samplePct <= 0 || samplePct >= 1) { + return ''; + } + return `RANDOM() < ${samplePct}`; + } + + getTableSampleClause(samplePct: number): string { + if (samplePct <= 0 || samplePct >= 1) { + return ''; + } + return `TABLESAMPLE SYSTEM (${samplePct * 100})`; + } + + getLimitOffsetClause(limit: number, offset?: number): string { + return limitOffsetClause(limit, offset); + } + + getTopClause(_limit: number): string { + return ''; + } + + getNullCountExpression(column: string): string { + return `COUNT(*) FILTER (WHERE ${column} IS NULL)`; + } + + getDistinctCountExpression(column: string): string { + return `COUNT(DISTINCT ${column})`; + } + + textLengthExpression(columnSql: string): string { + return `LENGTH(CAST(${columnSql} AS TEXT))`; + } + + castToText(columnSql: string): string { + return `CAST(${columnSql} AS TEXT)`; + } + + getSampleValueAggregation(innerSql: string): string { + return `(SELECT STRING_AGG(CAST(value AS TEXT), CHR(31)) FROM (${innerSql}) AS relationship_profile_values)`; + } + + generateCardinalitySampleQuery(tableName: string, columnName: string, sampleSize: number): string { + return ` + WITH sampled AS ( + SELECT ${columnName} AS val + FROM ${tableName} + WHERE ${columnName} IS NOT NULL + LIMIT ${sampleSize} + ) + SELECT COUNT(DISTINCT val) AS cardinality + FROM sampled + `; + } + + generateDistinctValuesQuery(tableName: string, columnName: string, limit: number): string { + return ` + SELECT DISTINCT ${columnName}::text AS val + FROM ${tableName} + WHERE ${columnName} IS NOT NULL + ORDER BY val + LIMIT ${limit} + `; + } + + generateColumnStatisticsQuery(schemaName: string, tableName: string): string | null { + return ` + SELECT + s.attname AS column_name, + CASE + WHEN s.n_distinct > 0 THEN s.n_distinct::bigint + WHEN s.n_distinct < 0 THEN (-s.n_distinct * c.reltuples)::bigint + ELSE NULL + END AS estimated_cardinality + FROM pg_stats s + JOIN pg_class c ON c.relname = s.tablename + JOIN pg_namespace n ON c.relnamespace = n.oid AND n.nspname = s.schemaname + WHERE s.schemaname = '${schemaName.replace(/'/g, "''")}' + AND s.tablename = '${tableName.replace(/'/g, "''")}' + AND s.n_distinct IS NOT NULL + `; + } + + generateRandomizedCardinalitySampleQuery(tableName: string, columnName: string, sampleSize: number): string { + return ` + WITH sampled AS ( + SELECT ${columnName} AS val + FROM ${tableName} + WHERE ${columnName} IS NOT NULL + ORDER BY RANDOM() + LIMIT ${sampleSize} + ) + SELECT COUNT(DISTINCT val) AS cardinality + FROM sampled + `; + } +} diff --git a/packages/cli/src/connectors/redshift/live-database-introspection.ts b/packages/cli/src/connectors/redshift/live-database-introspection.ts new file mode 100644 index 000000000..7b8289056 --- /dev/null +++ b/packages/cli/src/connectors/redshift/live-database-introspection.ts @@ -0,0 +1,47 @@ +import type { + LiveDatabaseIntrospectionOptions, + LiveDatabaseIntrospectionPort, +} from '../../context/ingest/adapters/live-database/types.js'; +import type { KtxProjectConnectionConfig } from '../../context/project/config.js'; +import { + KtxRedshiftScanConnector, + type KtxRedshiftConnectionConfig, + type KtxRedshiftEndpointResolver, + type KtxRedshiftPoolFactory, +} from './connector.js'; + +interface CreateRedshiftLiveDatabaseIntrospectionOptions { + connections: Record; + poolFactory?: KtxRedshiftPoolFactory; + endpointResolver?: KtxRedshiftEndpointResolver; + now?: () => Date; +} + +export function createRedshiftLiveDatabaseIntrospection( + options: CreateRedshiftLiveDatabaseIntrospectionOptions, +): LiveDatabaseIntrospectionPort { + return { + async extractSchema(connectionId: string, introspectionOptions?: LiveDatabaseIntrospectionOptions) { + const connection = options.connections[connectionId] as KtxRedshiftConnectionConfig | undefined; + const connector = new KtxRedshiftScanConnector({ + connectionId, + connection, + poolFactory: options.poolFactory, + endpointResolver: options.endpointResolver, + now: options.now, + }); + try { + return await connector.introspect( + { + connectionId, + driver: 'redshift', + ...(introspectionOptions?.tableScope ? { tableScope: introspectionOptions.tableScope } : {}), + }, + { runId: `redshift-${connectionId}` }, + ); + } finally { + await connector.cleanup(); + } + }, + }; +} diff --git a/packages/cli/src/context/connections/dialects.ts b/packages/cli/src/context/connections/dialects.ts index 30608f3e0..ec6f25c18 100644 --- a/packages/cli/src/context/connections/dialects.ts +++ b/packages/cli/src/context/connections/dialects.ts @@ -3,6 +3,7 @@ import { KtxClickHouseDialect } from '../../connectors/clickhouse/dialect.js'; import { KtxMongoDbDialect } from '../../connectors/mongodb/dialect.js'; import { KtxMysqlDialect } from '../../connectors/mysql/dialect.js'; import { KtxPostgresDialect } from '../../connectors/postgres/dialect.js'; +import { KtxRedshiftDialect } from '../../connectors/redshift/dialect.js'; import { KtxSqliteDialect } from '../../connectors/sqlite/dialect.js'; import { KtxSnowflakeDialect } from '../../connectors/snowflake/dialect.js'; import { KtxSqlServerDialect } from '../../connectors/sqlserver/dialect.js'; @@ -57,6 +58,7 @@ const sqlDialectFactories: Record KtxSqlDialect> = { clickhouse: () => new KtxClickHouseDialect(), mysql: () => new KtxMysqlDialect(), postgres: () => new KtxPostgresDialect(), + redshift: () => new KtxRedshiftDialect(), sqlite: () => new KtxSqliteDialect(), snowflake: () => new KtxSnowflakeDialect(), sqlserver: () => new KtxSqlServerDialect(), diff --git a/packages/cli/src/context/connections/drivers.ts b/packages/cli/src/context/connections/drivers.ts index 76ac408ef..4ecf67ae5 100644 --- a/packages/cli/src/context/connections/drivers.ts +++ b/packages/cli/src/context/connections/drivers.ts @@ -131,6 +131,27 @@ export const driverRegistrations: Record { + const m = await import('../../connectors/redshift/connector.js'); + return { + isConnectionConfig: (connection) => { + const typedConnection = connection as Parameters[0]; + return m.isKtxRedshiftConnectionConfig(typedConnection); + }, + createScanConnector: ({ connectionId, connection }) => { + const typedConnection = connection as Parameters[0]; + if (!m.isKtxRedshiftConnectionConfig(typedConnection)) { + throw invalidConnectionConfig('redshift'); + } + return new m.KtxRedshiftScanConnector({ connectionId, connection: typedConnection }); + }, + }; + }, + }, sqlite: { driver: 'sqlite', scopeConfigKey: null, diff --git a/packages/cli/src/context/project/driver-schemas.ts b/packages/cli/src/context/project/driver-schemas.ts index 25fa35079..df8f1ee18 100644 --- a/packages/cli/src/context/project/driver-schemas.ts +++ b/packages/cli/src/context/project/driver-schemas.ts @@ -7,6 +7,7 @@ import { const warehouseDrivers = [ 'postgres', + 'redshift', 'mysql', 'snowflake', 'bigquery', @@ -48,6 +49,7 @@ function warehouseConnectionSchema(driver: const warehouseConnectionSchemas = [ warehouseConnectionSchema('postgres'), + warehouseConnectionSchema('redshift'), warehouseConnectionSchema('mysql'), warehouseConnectionSchema('snowflake'), warehouseConnectionSchema('bigquery'), diff --git a/packages/cli/src/context/scan/types.ts b/packages/cli/src/context/scan/types.ts index 0d269c37b..537298f2f 100644 --- a/packages/cli/src/context/scan/types.ts +++ b/packages/cli/src/context/scan/types.ts @@ -8,7 +8,8 @@ export type KtxConnectionDriver = | 'snowflake' | 'mysql' | 'clickhouse' - | 'mongodb'; + | 'mongodb' + | 'redshift'; export type KtxScanMode = 'structural' | 'relationships' | 'enriched'; diff --git a/packages/cli/test/context/connections/dialects.test.ts b/packages/cli/test/context/connections/dialects.test.ts index cc7eaa59e..a56b3fb33 100644 --- a/packages/cli/test/context/connections/dialects.test.ts +++ b/packages/cli/test/context/connections/dialects.test.ts @@ -65,6 +65,36 @@ const fixtures: DialectFixture[] = [ nativeTypeInput: 'numeric(12,2)', normalizedType: 'numeric(12,2)', }, + { + driver: 'redshift', + table: { catalog: null, db: 'public', name: 'orders' }, + quoteInput: 'order"items', + quotedIdentifier: '"order""items"', + formattedTable: '"public"."orders"', + display: 'public.orders', + invalidDisplay: 'orders', + columnDisplayTablePartCount: 2, + limitClause: 'LIMIT 25 OFFSET 5', + topClause: '', + randomFilter: 'RANDOM() < 0.25', + tableSampleClause: 'TABLESAMPLE SYSTEM (25)', + sampleQuery: 'SELECT "id", "status" FROM "public"."orders" LIMIT 5', + columnSampleContains: 'TRIM(CAST("status" AS TEXT)) != \'\'', + nullCountExpression: 'COUNT(*) FILTER (WHERE "status" IS NULL)', + distinctCountExpression: 'COUNT(DISTINCT "status")', + textLengthExpression: 'LENGTH(CAST("status" AS TEXT))', + castToText: 'CAST("status" AS TEXT)', + sampleValueAggregation: + '(SELECT STRING_AGG(CAST(value AS TEXT), CHR(31)) FROM (SELECT status AS value FROM orders) AS relationship_profile_values)', + cardinalityContains: 'SELECT COUNT(DISTINCT val) AS cardinality', + randomizedCardinalityContains: 'ORDER BY RANDOM()', + distinctValuesContains: 'SELECT DISTINCT "status"::text AS val', + statisticsContains: 'FROM pg_stats s', + dimensionInput: 'timestamp with time zone', + dimensionType: 'time', + nativeTypeInput: 'numeric(12,2)', + normalizedType: 'numeric(12,2)', + }, { driver: 'mysql', table: { catalog: null, db: 'analytics', name: 'orders' }, @@ -305,7 +335,7 @@ describe('getDialectForDriver', () => { it('throws with a supported-driver list for unknown drivers', () => { expect(() => getDialectForDriver('oracle')).toThrow( - 'Unsupported driver "oracle". Supported drivers: bigquery, clickhouse, mongodb, mysql, postgres, snowflake, sqlite, sqlserver', + 'Unsupported driver "oracle". Supported drivers: bigquery, clickhouse, mongodb, mysql, postgres, redshift, snowflake, sqlite, sqlserver', ); }); diff --git a/packages/cli/test/context/connections/drivers.test.ts b/packages/cli/test/context/connections/drivers.test.ts index 65bbca4b8..3740c7ccb 100644 --- a/packages/cli/test/context/connections/drivers.test.ts +++ b/packages/cli/test/context/connections/drivers.test.ts @@ -21,6 +21,15 @@ const connectionFixtures: Record = { url: 'postgresql://reader:secret@localhost:5432/analytics', // pragma: allowlist secret schemas: ['public'], }), + redshift: () => ({ + driver: 'redshift', + host: 'analytics.123456789012.us-east-1.redshift.amazonaws.com', + port: 5439, + database: 'analytics', + username: 'reader', + password: 'secret', // pragma: allowlist secret + schemas: ['public'], + }), sqlite: () => ({ driver: 'sqlite', path: 'warehouse.db' }), mongodb: () => ({ driver: 'mongodb', @@ -104,6 +113,7 @@ describe('driverRegistrations', () => { 'mongodb', 'mysql', 'postgres', + 'redshift', 'snowflake', 'sqlite', 'sqlserver', From e7e7b78cd7dc4d21ab84ad40875305ca9ca32ab8 Mon Sep 17 00:00:00 2001 From: Mayorkun Ayanshina Date: Wed, 17 Jun 2026 11:03:22 +0100 Subject: [PATCH 2/9] test(redshift): add connector unit tests Mirror the Postgres connector test suite with mocked pool factories, adapting SQL mock keys to the Redshift SVV_* introspection queries (svv_tables, svv_columns, svv_table_info) and the 5439 default port. --- .../connectors/redshift/connector.test.ts | 596 ++++++++++++++++++ 1 file changed, 596 insertions(+) create mode 100644 packages/cli/test/connectors/redshift/connector.test.ts diff --git a/packages/cli/test/connectors/redshift/connector.test.ts b/packages/cli/test/connectors/redshift/connector.test.ts new file mode 100644 index 000000000..c37bede4e --- /dev/null +++ b/packages/cli/test/connectors/redshift/connector.test.ts @@ -0,0 +1,596 @@ +import { describe, expect, it, vi } from 'vitest'; +import { createRedshiftLiveDatabaseIntrospection } from '../../../src/connectors/redshift/live-database-introspection.js'; +import { isKtxRedshiftConnectionConfig, KtxRedshiftScanConnector, redshiftPoolConfigFromConfig, prepareRedshiftReadOnlyQuery, type KtxRedshiftConnectionConfig, type KtxRedshiftPoolFactory } from '../../../src/connectors/redshift/connector.js'; +import { tableRefSet } from '../../../src/context/scan/table-ref.js'; + +interface FakeQueryResult { + rows: Record[]; + fields?: Array<{ name: string; dataTypeID: number }>; +} + +type FakeQueryResponse = FakeQueryResult | Error; + +function fakePoolFactory(results: Map): KtxRedshiftPoolFactory { + 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; + } + } + throw new Error(`Unexpected SQL: ${normalized} params=${JSON.stringify(params ?? [])}`); + }); + return { + createPool() { + return { + async connect() { + return { + query, + release: vi.fn(), + }; + }, + end: vi.fn(async () => undefined), + }; + }, + }; +} + +function metadataResults(): Map { + return new Map([ + [ + 'FROM svv_tables t', + { + rows: [ + { schema_name: 'public', table_name: 'customers', table_kind: 'r', row_count: '2', table_comment: 'Customers' }, + { schema_name: 'public', table_name: 'orders', table_kind: 'r', row_count: '3', table_comment: null }, + { schema_name: 'public', table_name: 'recent_orders', table_kind: 'v', row_count: '0', table_comment: 'Recent orders' }, + ], + }, + ], + [ + 'FROM svv_columns', + { + rows: [ + { table_name: 'customers', column_name: 'id', data_type: 'integer', is_nullable: false, column_comment: null }, + { table_name: 'customers', column_name: 'name', data_type: 'text', is_nullable: false, column_comment: 'Name' }, + { table_name: 'orders', column_name: 'id', data_type: 'integer', is_nullable: false, column_comment: null }, + { table_name: 'orders', column_name: 'customer_id', data_type: 'integer', is_nullable: false, column_comment: null }, + { table_name: 'orders', column_name: 'status', data_type: 'text', is_nullable: true, column_comment: null }, + { table_name: 'recent_orders', column_name: 'id', data_type: 'integer', is_nullable: true, column_comment: null }, + ], + }, + ], + [ + "tc.constraint_type = 'FOREIGN KEY'", + { + rows: [ + { + table_name: 'orders', + column_name: 'customer_id', + foreign_table_schema: 'public', + foreign_table_name: 'customers', + foreign_column_name: 'id', + constraint_name: 'orders_customer_id_fkey', + }, + ], + }, + ], + [ + "tc.constraint_type = 'PRIMARY KEY'", + { + rows: [ + { table_name: 'customers', column_name: 'id' }, + { table_name: 'orders', column_name: 'id' }, + ], + }, + ], + ['SELECT "id" FROM "public"."orders" LIMIT 1', { rows: [{ id: 10 }], fields: [{ name: 'id', dataTypeID: 23 }] }], + [ + 'SELECT "status" FROM "public"."orders" WHERE "status" IS NOT NULL', + { rows: [{ status: 'paid' }, { status: 'open' }], fields: [{ name: 'status', dataTypeID: 25 }] }, + ], + ['COUNT(DISTINCT val) AS cardinality', { rows: [{ cardinality: '2' }] }], + ['SELECT DISTINCT "status"::text AS val', { rows: [{ val: 'open' }, { val: 'paid' }] }], + ['SELECT COUNT(*) AS count FROM "public"."orders"', { rows: [{ count: '3' }] }], + ['FROM pg_stats s', { rows: [{ column_name: 'status', estimated_cardinality: '2' }] }], + ['SELECT 1', { rows: [{ '?column?': 1 }], fields: [{ name: '?column?', dataTypeID: 23 }] }], + [ + 'FROM svv_tables WHERE table_schema = ANY($1) AND table_type IN', + { + rows: [ + { schema_name: 'public', table_name: 'customers', table_kind: 'r' }, + { schema_name: 'public', table_name: 'orders', table_kind: 'r' }, + { schema_name: 'public', table_name: 'recent_orders', table_kind: 'v' }, + ], + }, + ], + ['SELECT schema_name FROM information_schema.schemata', { rows: [{ schema_name: 'public' }] }], + ]); +} + +describe('KtxRedshiftScanConnector', () => { + it('prepares read-only SQL parameters with Redshift positional placeholders', () => { + expect( + prepareRedshiftReadOnlyQuery('select * from orders where id = :id and status = :status', { + id: 1, + status: 'paid', + }), + ).toEqual({ + sql: 'select * from orders where id = $1 and status = $2', + params: [1, 'paid'], + }); + expect( + prepareRedshiftReadOnlyQuery('select :Client_Name_10, :Client_Name_1', { + Client_Name_1: 'short', + Client_Name_10: 'long', + }), + ).toEqual({ + sql: 'select $2, $1', + params: ['short', 'long'], + }); + expect(prepareRedshiftReadOnlyQuery('select 1')).toEqual({ sql: 'select 1', params: undefined }); + }); + + it('resolves configuration safely', () => { + expect(isKtxRedshiftConnectionConfig({ driver: 'redshift', url: 'env:DATABASE_URL' })).toBe(true); + expect(isKtxRedshiftConnectionConfig({ driver: 'redshiftql', host: 'db', database: 'analytics' })).toBe(false); + expect(isKtxRedshiftConnectionConfig({ driver: 'mysql', host: 'db' })).toBe(false); + expect( + redshiftPoolConfigFromConfig({ + connectionId: 'warehouse', + connection: { + driver: 'redshift', + host: 'db.example.test', + database: 'analytics', + username: 'reader', + password: 'test-password', // pragma: allowlist secret + schemas: ['analytics', 'public'], + ssl: true, + rejectUnauthorized: false, + }, + }), + ).toMatchObject({ + host: 'db.example.test', + port: 5439, + database: 'analytics', + user: 'reader', + password: 'test-password', // pragma: allowlist secret + options: '-c search_path=analytics,public', + ssl: { rejectUnauthorized: false }, + }); + const libpqPreferConfig = redshiftPoolConfigFromConfig({ + connectionId: 'warehouse', + connection: { + driver: 'redshift', + url: 'env:DEMO_DATABASE_URL', + }, + env: { + DEMO_DATABASE_URL: 'redshift://reader@demo.example.test:5439/demo?sslmode=prefer', + }, + }); + expect(libpqPreferConfig).toMatchObject({ + host: 'demo.example.test', + port: 5439, + database: 'demo', + user: 'reader', + }); + expect(libpqPreferConfig).not.toHaveProperty('connectionString'); + expect(libpqPreferConfig).not.toHaveProperty('ssl'); + expect( + redshiftPoolConfigFromConfig({ + connectionId: 'warehouse', + connection: { driver: 'redshift', host: 'db.example.test', database: 'analytics', username: 'reader' }, + }), + ).toMatchObject({ + host: 'db.example.test', + database: 'analytics', + user: 'reader', + }); + }); + + it('defaults and validates Redshift maxConnections', () => { + const baseConnection: KtxRedshiftConnectionConfig = { + driver: 'redshift', + host: 'db.example.test', + database: 'analytics', + username: 'reader', + password: 'test-password', // pragma: allowlist secret + }; + + expect( + redshiftPoolConfigFromConfig({ + connectionId: 'warehouse', + connection: baseConnection, + }), + ).toMatchObject({ max: 10 }); + + expect( + redshiftPoolConfigFromConfig({ + connectionId: 'warehouse', + connection: { ...baseConnection, maxConnections: 50 }, + }), + ).toMatchObject({ max: 50 }); + + expect( + redshiftPoolConfigFromConfig({ + connectionId: 'warehouse', + connection: { ...baseConnection, maxConnections: '12' as never }, + }), + ).toMatchObject({ max: 12 }); + + for (const maxConnections of [0, -1, 1.5, Number.NaN, 'abc' as never]) { + expect(() => + redshiftPoolConfigFromConfig({ + connectionId: 'warehouse', + connection: { ...baseConnection, maxConnections }, + }), + ).toThrow('connections.warehouse.maxConnections must be a positive integer'); + } + }); + + it('introspects schemas, tables, views, primary keys, comments, row counts, and foreign keys', async () => { + const connector = new KtxRedshiftScanConnector({ + connectionId: 'warehouse', + connection: { + driver: 'redshift', + host: 'db.example.test', + database: 'analytics', + username: 'reader', + password: 'test-password', // pragma: allowlist secret + schema: 'public', + }, + poolFactory: fakePoolFactory(metadataResults()), + now: () => new Date('2026-04-29T10:00:00.000Z'), + }); + + const snapshot = await connector.introspect( + { connectionId: 'warehouse', driver: 'redshift' }, + { runId: 'scan-run-1' }, + ); + + expect(snapshot).toMatchObject({ + connectionId: 'warehouse', + driver: 'redshift', + extractedAt: '2026-04-29T10:00:00.000Z', + scope: { schemas: ['public'] }, + metadata: { + database: 'analytics', + schemas: ['public'], + host: 'db.example.test', + table_count: 3, + total_columns: 6, + }, + }); + expect(snapshot.tables.map((table) => [table.db, table.name, table.kind, table.estimatedRows])).toEqual([ + ['public', 'customers', 'table', 2], + ['public', 'orders', 'table', 3], + ['public', 'recent_orders', 'view', null], + ]); + expect(snapshot.tables.find((table) => table.name === 'customers')?.columns[0]).toMatchObject({ + name: 'id', + nativeType: 'integer', + normalizedType: 'integer', + dimensionType: 'number', + nullable: false, + primaryKey: true, + }); + expect(snapshot.tables.find((table) => table.name === 'orders')?.foreignKeys).toEqual([ + { + fromColumn: 'customer_id', + toCatalog: null, + toDb: 'public', + toTable: 'customers', + toColumn: 'id', + constraintName: 'orders_customer_id_fkey', + }, + ]); + }); + + it('soft-fails denied Redshift 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 KtxRedshiftScanConnector({ + connectionId: 'warehouse', + connection: { + driver: 'redshift', + 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: 'redshift' }, + { 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 Redshift 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 KtxRedshiftScanConnector({ + connectionId: 'warehouse', + connection: { + driver: 'redshift', + 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: 'redshift' }, { runId: 'scan-run-network-error' }), + ).rejects.toBe(resetError); + }); + + it('runs samples, distinct values, statistics, read-only SQL, and schema listing', async () => { + const connector = new KtxRedshiftScanConnector({ + connectionId: 'warehouse', + connection: { + driver: 'redshift', + host: 'db.example.test', + database: 'analytics', + username: 'reader', + password: 'test-password', // pragma: allowlist secret + schema: 'public', + }, + poolFactory: fakePoolFactory(metadataResults()), + }); + + await expect( + connector.sampleTable( + { connectionId: 'warehouse', table: { catalog: null, db: 'public', name: 'orders' }, columns: ['id'], limit: 1 }, + { runId: 'scan-run-1' }, + ), + ).resolves.toEqual({ headers: ['id'], headerTypes: ['integer'], rows: [[10]], totalRows: 1 }); + + await expect( + connector.sampleColumn( + { connectionId: 'warehouse', table: { catalog: null, db: 'public', name: 'orders' }, column: 'status', limit: 5 }, + { runId: 'scan-run-1' }, + ), + ).resolves.toMatchObject({ values: ['paid', 'open'], nullCount: null, distinctCount: null }); + + await expect( + connector.getColumnDistinctValues( + { catalog: null, db: 'public', name: 'orders' }, + 'status', + { maxCardinality: 5, limit: 10, sampleSize: 100 }, + ), + ).resolves.toEqual({ values: ['open', 'paid'], cardinality: 2 }); + + await expect(connector.getColumnStatistics({ catalog: null, db: 'public', name: 'orders' })).resolves.toEqual({ + cardinalityByColumn: new Map([['status', 2]]), + }); + await expect(connector.getTableRowCount({ db: 'public', name: 'orders' })).resolves.toBe(3); + await expect(connector.listSchemas()).resolves.toEqual(['public']); + await expect(connector.listTables(['public'])).resolves.toEqual([ + { catalog: null, schema: 'public', name: 'customers', kind: 'table' }, + { catalog: null, schema: 'public', name: 'orders', kind: 'table' }, + { catalog: null, schema: 'public', name: 'recent_orders', kind: 'view' }, + ]); + await expect(connector.testConnection()).resolves.toEqual({ success: true }); + + await expect( + connector.executeReadOnly({ connectionId: 'warehouse', sql: 'delete from orders' }, { runId: 'scan-run-1' }), + ).rejects.toThrow('Only read-only SELECT/WITH queries can be executed locally'); + }); + + it('limits introspection to tables in tableScope', async () => { + const queries: Array<{ sql: string; params?: unknown[] }> = []; + const poolFactory: KtxRedshiftPoolFactory = { + createPool() { + return { + async connect() { + return { + query: vi.fn(async (sql: string, params?: unknown[]) => { + queries.push({ sql, params }); + if (sql.includes('FROM svv_tables t')) { + return { rows: [{ table_name: 'orders', table_kind: 'r', row_count: '3', table_comment: null }] }; + } + if (sql.includes('FROM svv_columns')) { + return { + rows: [ + { + table_name: 'orders', + column_name: 'id', + data_type: 'integer', + is_nullable: false, + column_comment: null, + }, + ], + }; + } + return { rows: [] }; + }), + release: vi.fn(), + }; + }, + end: vi.fn(async () => undefined), + }; + }, + }; + const connector = new KtxRedshiftScanConnector({ + connectionId: 'warehouse', + connection: { + driver: 'redshift', + host: 'db.example.test', + database: 'analytics', + username: 'reader', + password: 'test-password', // pragma: allowlist secret + schema: 'public', + }, + poolFactory, + }); + const scope = tableRefSet([{ catalog: null, db: 'public', name: 'orders' }]); + const snapshot = await connector.introspect( + { connectionId: 'warehouse', driver: 'redshift', tableScope: scope }, + { runId: 'scope-test' }, + ); + expect(snapshot.tables.map((table) => table.name)).toEqual(['orders']); + const tablesQuery = queries.find((query) => query.sql.includes('FROM svv_tables t')); + expect(tablesQuery?.sql).toMatch(/t\.table_name = ANY\(\$2\)/); + expect(tablesQuery?.params).toEqual(['public', ['orders']]); + }); + + it('adapts native Redshift snapshots to live-database introspection for local ingest', async () => { + const introspection = createRedshiftLiveDatabaseIntrospection({ + connections: { + warehouse: { + driver: 'redshift', + host: 'db.example.test', + database: 'analytics', + username: 'reader', + password: 'test-password', // pragma: allowlist secret + schema: 'public', + }, + }, + poolFactory: fakePoolFactory(metadataResults()), + now: () => new Date('2026-04-29T10:00:00.000Z'), + }); + + const snapshot = await introspection.extractSchema('warehouse'); + + expect(snapshot).toMatchObject({ + connectionId: 'warehouse', + extractedAt: '2026-04-29T10:00:00.000Z', + }); + expect(snapshot.tables.find((table) => table.name === 'customers')).toMatchObject({ + name: 'customers', + catalog: null, + db: 'public', + columns: [ + { + name: 'id', + nativeType: 'integer', + normalizedType: 'integer', + dimensionType: 'number', + nullable: false, + primaryKey: true, + comment: null, + }, + { + name: 'name', + nativeType: 'text', + normalizedType: 'text', + dimensionType: 'string', + nullable: false, + primaryKey: false, + comment: 'Name', + }, + ], + foreignKeys: [], + }); + }); + + it('does not end the pool before introspection completes', async () => { + let endCalled = false; + const endAwarePoolFactory: KtxRedshiftPoolFactory = { + createPool() { + const inner = fakePoolFactory(metadataResults()).createPool({ + max: 1, + idleTimeoutMillis: 1, + connectionTimeoutMillis: 1, + }); + return { + async connect() { + if (endCalled) { + throw new Error('Cannot use a pool after calling end on the pool'); + } + return inner.connect(); + }, + async end() { + endCalled = true; + return inner.end(); + }, + }; + }, + }; + const introspection = createRedshiftLiveDatabaseIntrospection({ + connections: { + warehouse: { + driver: 'redshift', + host: 'db.example.test', + database: 'analytics', + username: 'reader', + password: 'test-password', // pragma: allowlist secret + schema: 'public', + }, + }, + poolFactory: endAwarePoolFactory, + now: () => new Date('2026-04-29T10:00:00.000Z'), + }); + + const snapshot = await introspection.extractSchema('warehouse'); + expect(snapshot.tables.length).toBeGreaterThan(0); + expect(endCalled).toBe(true); + }); + + it('attaches an error listener to the pg pool', async () => { + const on = vi.fn(); + const poolFactory: KtxRedshiftPoolFactory = { + createPool() { + return { + on, + async connect() { + return { + query: vi.fn(async () => ({ rows: [{ '?column?': 1 }], fields: [{ name: '?column?', dataTypeID: 23 }] })), + release: vi.fn(), + }; + }, + end: vi.fn(async () => undefined), + }; + }, + }; + const connector = new KtxRedshiftScanConnector({ + connectionId: 'warehouse', + connection: { + driver: 'redshift', + host: 'db.example.test', + database: 'analytics', + username: 'reader', + password: 'test-password', // pragma: allowlist secret + }, + poolFactory, + }); + + await expect(connector.testConnection()).resolves.toEqual({ success: true }); + + expect(on).toHaveBeenCalledWith('error', expect.any(Function)); + }); +}); From de464ccca95a87cb3a6d6a73006698632f66e8ae Mon Sep 17 00:00:00 2001 From: Mayorkun Ayanshina Date: Wed, 17 Jun 2026 11:32:03 +0100 Subject: [PATCH 3/9] docs(redshift): add integration docs; drop unused live-database introspection Add a Redshift section to primary-sources.mdx with connection-config examples and honest feature/auth scoping (password auth + SVV_* metadata introspection; IAM auth and STL_QUERY history noted as follow-ups). Remove the copied live-database-introspection module and its tests since this connector does not wire the live-database ingest path. --- .../docs/integrations/primary-sources.mdx | 71 +++++++++++++- .../redshift/live-database-introspection.ts | 47 ---------- .../connectors/redshift/connector.test.ts | 94 ------------------- 3 files changed, 70 insertions(+), 142 deletions(-) delete mode 100644 packages/cli/src/connectors/redshift/live-database-introspection.ts diff --git a/docs-site/content/docs/integrations/primary-sources.mdx b/docs-site/content/docs/integrations/primary-sources.mdx index 40af30e09..9f0a410e5 100644 --- a/docs-site/content/docs/integrations/primary-sources.mdx +++ b/docs-site/content/docs/integrations/primary-sources.mdx @@ -26,7 +26,7 @@ Agents should prefer environment or file references over literal secrets. | Field | Required | Applies to | Description | |-------|----------|------------|-------------| -| `driver` | Yes | all connections | Connector driver such as `postgres`, `snowflake`, `bigquery`, `mysql`, `clickhouse`, `sqlserver`, `sqlite`, or `mongodb` | +| `driver` | Yes | all connections | Connector driver such as `postgres`, `redshift`, `snowflake`, `bigquery`, `mysql`, `clickhouse`, `sqlserver`, `sqlite`, or `mongodb` | | `url` | One of the connection methods | URL-style connectors | Database URL, `env:NAME`, or `file:/path/to/secret` | | `host`, `port`, `database`, `username`, `password` | One of the connection methods | PostgreSQL, MySQL, SQL Server | Field-by-field connection values | | `schema` or `schemas` | No | schema-aware warehouses | Single schema or list of schemas to scan | @@ -118,6 +118,75 @@ This helps **ktx** understand how your team actually queries the data. --- +## Amazon Redshift + +Amazon Redshift speaks the PostgreSQL wire protocol, so the connector reuses the +PostgreSQL connection layer (via the `pg` driver) while reading metadata from +Redshift-specific system views for accurate results. Supports schema +introspection, primary/foreign key detection, column statistics, table +sampling, and read-only SQL execution. + +### Connection config + +```yaml title="ktx.yaml" +connections: + my-redshift: + driver: redshift + url: env:REDSHIFT_URL + schema: public +``` + +Or with individual fields: + +```yaml title="ktx.yaml" +connections: + my-redshift: + driver: redshift + host: analytics.123456789012.us-east-1.redshift.amazonaws.com + port: 5439 + database: analytics + username: ktx_reader + password: env:REDSHIFT_PASSWORD + schemas: + - public + - analytics + ssl: true +``` + +### Authentication + +| Method | Config | +|--------|--------| +| Password | `password: env:REDSHIFT_PASSWORD` or `password: file:/path/to/secret` | +| Connection URL | `url: env:REDSHIFT_URL` | +| SSL | `ssl: true`, optionally `rejectUnauthorized: false` for self-signed certs | + +IAM-based authentication (temporary credentials via `GetClusterCredentials`) is +not yet supported and is planned as a follow-up. + +### Features + +| Feature | Supported | Notes | +|---------|-----------|-------| +| Tables & views | Yes | Via `SVV_TABLES` | +| Columns | Yes | Via `SVV_COLUMNS` | +| Primary keys | Yes | Via `information_schema.table_constraints` | +| Foreign keys | Yes | Declared constraints (Redshift does not enforce them) | +| Row count estimates | Yes | Via `SVV_TABLE_INFO.tbl_rows` | +| Column statistics | Yes | Reuses the PostgreSQL-compatible statistics path | +| Table sampling | Yes | `TABLESAMPLE SYSTEM` | +| Query history | No | `STL_QUERY` ingestion is planned as a follow-up | + +### Dialect notes + +- Redshift uses the same SQL compilation as PostgreSQL (`LIMIT/OFFSET` + pagination, positional parameters, `COUNT(*) FILTER (WHERE ...)`). +- Metadata is read from Redshift system views (`SVV_TABLES`, `SVV_COLUMNS`, + `SVV_TABLE_INFO`) rather than `pg_catalog`, because `pg_class.reltuples` is + unreliable on Redshift and the `SVV_*` views report accurate row counts and + cover external (Spectrum) schemas. +- The default port is `5439`. + ## Snowflake Connects via the Snowflake SDK. Supports multi-schema scanning, RSA key authentication, and query-history configuration for Snowflake query history. diff --git a/packages/cli/src/connectors/redshift/live-database-introspection.ts b/packages/cli/src/connectors/redshift/live-database-introspection.ts deleted file mode 100644 index 7b8289056..000000000 --- a/packages/cli/src/connectors/redshift/live-database-introspection.ts +++ /dev/null @@ -1,47 +0,0 @@ -import type { - LiveDatabaseIntrospectionOptions, - LiveDatabaseIntrospectionPort, -} from '../../context/ingest/adapters/live-database/types.js'; -import type { KtxProjectConnectionConfig } from '../../context/project/config.js'; -import { - KtxRedshiftScanConnector, - type KtxRedshiftConnectionConfig, - type KtxRedshiftEndpointResolver, - type KtxRedshiftPoolFactory, -} from './connector.js'; - -interface CreateRedshiftLiveDatabaseIntrospectionOptions { - connections: Record; - poolFactory?: KtxRedshiftPoolFactory; - endpointResolver?: KtxRedshiftEndpointResolver; - now?: () => Date; -} - -export function createRedshiftLiveDatabaseIntrospection( - options: CreateRedshiftLiveDatabaseIntrospectionOptions, -): LiveDatabaseIntrospectionPort { - return { - async extractSchema(connectionId: string, introspectionOptions?: LiveDatabaseIntrospectionOptions) { - const connection = options.connections[connectionId] as KtxRedshiftConnectionConfig | undefined; - const connector = new KtxRedshiftScanConnector({ - connectionId, - connection, - poolFactory: options.poolFactory, - endpointResolver: options.endpointResolver, - now: options.now, - }); - try { - return await connector.introspect( - { - connectionId, - driver: 'redshift', - ...(introspectionOptions?.tableScope ? { tableScope: introspectionOptions.tableScope } : {}), - }, - { runId: `redshift-${connectionId}` }, - ); - } finally { - await connector.cleanup(); - } - }, - }; -} diff --git a/packages/cli/test/connectors/redshift/connector.test.ts b/packages/cli/test/connectors/redshift/connector.test.ts index c37bede4e..2f69fadc1 100644 --- a/packages/cli/test/connectors/redshift/connector.test.ts +++ b/packages/cli/test/connectors/redshift/connector.test.ts @@ -1,5 +1,4 @@ import { describe, expect, it, vi } from 'vitest'; -import { createRedshiftLiveDatabaseIntrospection } from '../../../src/connectors/redshift/live-database-introspection.js'; import { isKtxRedshiftConnectionConfig, KtxRedshiftScanConnector, redshiftPoolConfigFromConfig, prepareRedshiftReadOnlyQuery, type KtxRedshiftConnectionConfig, type KtxRedshiftPoolFactory } from '../../../src/connectors/redshift/connector.js'; import { tableRefSet } from '../../../src/context/scan/table-ref.js'; @@ -468,99 +467,6 @@ describe('KtxRedshiftScanConnector', () => { expect(tablesQuery?.params).toEqual(['public', ['orders']]); }); - it('adapts native Redshift snapshots to live-database introspection for local ingest', async () => { - const introspection = createRedshiftLiveDatabaseIntrospection({ - connections: { - warehouse: { - driver: 'redshift', - host: 'db.example.test', - database: 'analytics', - username: 'reader', - password: 'test-password', // pragma: allowlist secret - schema: 'public', - }, - }, - poolFactory: fakePoolFactory(metadataResults()), - now: () => new Date('2026-04-29T10:00:00.000Z'), - }); - - const snapshot = await introspection.extractSchema('warehouse'); - - expect(snapshot).toMatchObject({ - connectionId: 'warehouse', - extractedAt: '2026-04-29T10:00:00.000Z', - }); - expect(snapshot.tables.find((table) => table.name === 'customers')).toMatchObject({ - name: 'customers', - catalog: null, - db: 'public', - columns: [ - { - name: 'id', - nativeType: 'integer', - normalizedType: 'integer', - dimensionType: 'number', - nullable: false, - primaryKey: true, - comment: null, - }, - { - name: 'name', - nativeType: 'text', - normalizedType: 'text', - dimensionType: 'string', - nullable: false, - primaryKey: false, - comment: 'Name', - }, - ], - foreignKeys: [], - }); - }); - - it('does not end the pool before introspection completes', async () => { - let endCalled = false; - const endAwarePoolFactory: KtxRedshiftPoolFactory = { - createPool() { - const inner = fakePoolFactory(metadataResults()).createPool({ - max: 1, - idleTimeoutMillis: 1, - connectionTimeoutMillis: 1, - }); - return { - async connect() { - if (endCalled) { - throw new Error('Cannot use a pool after calling end on the pool'); - } - return inner.connect(); - }, - async end() { - endCalled = true; - return inner.end(); - }, - }; - }, - }; - const introspection = createRedshiftLiveDatabaseIntrospection({ - connections: { - warehouse: { - driver: 'redshift', - host: 'db.example.test', - database: 'analytics', - username: 'reader', - password: 'test-password', // pragma: allowlist secret - schema: 'public', - }, - }, - poolFactory: endAwarePoolFactory, - now: () => new Date('2026-04-29T10:00:00.000Z'), - }); - - const snapshot = await introspection.extractSchema('warehouse'); - expect(snapshot.tables.length).toBeGreaterThan(0); - expect(endCalled).toBe(true); - }); - it('attaches an error listener to the pg pool', async () => { const on = vi.fn(); const poolFactory: KtxRedshiftPoolFactory = { From d207a8e3f89b8befd49c34bac1057c70a8e1d0eb Mon Sep 17 00:00:00 2001 From: Mayorkun Ayanshina Date: Wed, 8 Jul 2026 16:19:45 +0100 Subject: [PATCH 4/9] fix(redshift): use getSqlDialectForDriver after upstream dialect split Upstream split KtxDialect into base and SQL variants; quoteIdentifier now lives on KtxSqlDialect, so the connector must resolve its dialect via getSqlDialectForDriver like the Postgres connector does. --- packages/cli/src/connectors/redshift/connector.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/cli/src/connectors/redshift/connector.ts b/packages/cli/src/connectors/redshift/connector.ts index 25698a0d0..620d6517c 100644 --- a/packages/cli/src/connectors/redshift/connector.ts +++ b/packages/cli/src/connectors/redshift/connector.ts @@ -1,5 +1,5 @@ import { resolveStringReference } from '../shared/string-reference.js'; -import { getDialectForDriver } from '../../context/connections/dialects.js'; +import { getSqlDialectForDriver } from '../../context/connections/dialects.js'; import { assertReadOnlySql, limitSqlForExecution } from '../../context/connections/read-only-sql.js'; import { tryConstraintQuery } from '../../context/scan/constraint-discovery.js'; import { scopedTableNames } from '../../context/scan/table-ref.js'; @@ -412,7 +412,7 @@ export class KtxRedshiftScanConnector implements KtxScanConnector { private readonly poolFactory: KtxRedshiftPoolFactory; private readonly endpointResolver?: KtxRedshiftEndpointResolver; private readonly now: () => Date; - private readonly dialect = getDialectForDriver('redshift'); + private readonly dialect = getSqlDialectForDriver('redshift'); private pool: KtxRedshiftPool | null = null; private lastIdlePoolError: Error | null = null; private resolvedEndpoint: KtxRedshiftResolvedEndpoint | null = null; From 9d8e51fb55dfc41aad8c7ca50bb95a20d1d09ca8 Mon Sep 17 00:00:00 2001 From: Mayorkun Ayanshina Date: Thu, 9 Jul 2026 15:26:36 +0100 Subject: [PATCH 5/9] refactor(connectors): share the Postgres wire layer with Redshift Extract the pg-protocol plumbing that Postgres and Redshift both need into connectors/shared/pg-wire.ts: pool factory and pool types, pool config resolution (parameterised by default port and connector label), URL parsing, read-only query prep, and query/row helpers. Postgres and Redshift now import this module instead of each carrying their own copy. Each connector keeps its own metadata introspection (Postgres pg_catalog, Redshift SVV_*), driver name, dialect, and default port (5432 / 5439). Postgres behaviour is unchanged: its test suite passes untouched, and its public exports are preserved as thin aliases over the shared types. Redshift picks up the statement_timeout / query-deadline handling it was previously missing, matching Postgres. --- .../cli/src/connectors/postgres/connector.ts | 301 ++---------------- .../cli/src/connectors/redshift/connector.ts | 300 +++-------------- packages/cli/src/connectors/shared/pg-wire.ts | 292 +++++++++++++++++ .../connectors/redshift/connector.test.ts | 2 +- 4 files changed, 369 insertions(+), 526 deletions(-) create mode 100644 packages/cli/src/connectors/shared/pg-wire.ts diff --git a/packages/cli/src/connectors/postgres/connector.ts b/packages/cli/src/connectors/postgres/connector.ts index c2c0f6db2..e2f2cc1b2 100644 --- a/packages/cli/src/connectors/postgres/connector.ts +++ b/packages/cli/src/connectors/postgres/connector.ts @@ -1,6 +1,25 @@ -import { resolveStringReference } from '../shared/string-reference.js'; import { getSqlDialectForDriver } from '../../context/connections/dialects.js'; import { resolveQueryDeadlineMs, queryDeadlineExceededError } from '../../context/connections/query-deadline.js'; +import { + DefaultPgPoolFactory, + PG_OID_TYPE_MAP, + finiteNumber, + groupByTable, + isDeniedError, + isPgTimeoutError, + numberValue, + pgPoolConfigFromConfig, + preparePgReadOnlyQuery, + primaryKeyMap, + queryRows, + schemasFromConnection, + type KtxPgConnectionConfig, + type KtxPgEndpointResolver, + type KtxPgPool, + type KtxPgPoolConfig, + type KtxPgPoolFactory, + type KtxPgResolvedEndpoint, +} from '../shared/pg-wire.js'; import { assertReadOnlySql, limitSqlForExecution } from '../../context/connections/read-only-sql.js'; import { tryConstraintQuery } from '../../context/scan/constraint-discovery.js'; import { scopedTableNames } from '../../context/scan/table-ref.js'; @@ -27,95 +46,13 @@ import { type KtxTableSampleInput, type KtxTableSampleResult, } from '../../context/scan/types.js'; -import { Pool } from 'pg'; - -const PG_OID_TYPE_MAP: Record = { - 16: 'boolean', - 20: 'bigint', - 21: 'smallint', - 23: 'integer', - 25: 'text', - 700: 'real', - 701: 'double precision', - 1043: 'varchar', - 1082: 'date', - 1114: 'timestamp', - 1184: 'timestamptz', - 1700: 'numeric', - 2950: 'uuid', - 3802: 'jsonb', - 114: 'json', - 1009: 'text[]', - 1007: 'integer[]', - 1016: 'bigint[]', -}; - -export interface KtxPostgresConnectionConfig { - driver?: string; - host?: string; - port?: number; - database?: string; - username?: string; - user?: string; - password?: string; - url?: string; - schema?: string; - schemas?: string[]; - ssl?: boolean; - sslmode?: string; - sslMode?: string; - rejectUnauthorized?: boolean; - maxConnections?: number; - [key: string]: unknown; -} -export interface KtxPostgresPoolConfig { - host?: string; - port?: number; - database?: string; - user?: string; - password?: string; - connectionString?: string; - max: number; - idleTimeoutMillis: number; - connectionTimeoutMillis: number; - options?: string; - ssl?: { rejectUnauthorized: boolean }; -} - -interface KtxPostgresQueryResult { - fields?: Array<{ name: string; dataTypeID: number }>; - rows: Record[]; -} - -interface KtxPostgresClient { - query(sql: string, params?: unknown[]): Promise; - release(): void; -} - -interface KtxPostgresPool { - connect(): Promise; - end(): Promise; - on?(event: 'error', listener: (error: Error) => void): void; -} - -export interface KtxPostgresPoolFactory { - createPool(config: KtxPostgresPoolConfig): KtxPostgresPool; -} - -interface KtxPostgresResolvedEndpoint { - host: string; - port: number; - close?: () => Promise; -} - -export interface KtxPostgresEndpointResolver { - resolve(input: { - host: string; - port: number; - connection: KtxPostgresConnectionConfig; - }): Promise; -} +export type KtxPostgresConnectionConfig = KtxPgConnectionConfig; +export type KtxPostgresPoolConfig = KtxPgPoolConfig; +type KtxPostgresPool = KtxPgPool; +export type KtxPostgresPoolFactory = KtxPgPoolFactory; +type KtxPostgresResolvedEndpoint = KtxPgResolvedEndpoint; +export type KtxPostgresEndpointResolver = KtxPgEndpointResolver; export interface KtxPostgresScanConnectorOptions { connectionId: string; @@ -204,138 +141,8 @@ interface PostgresStatsRow { estimated_cardinality: unknown; } -class DefaultPostgresPoolFactory implements KtxPostgresPoolFactory { - createPool(config: KtxPostgresPoolConfig): KtxPostgresPool { - return new Pool(config); - } -} - -function groupByTable(rows: T[]): Map { - const grouped = new Map(); - for (const row of rows) { - const tableRows = grouped.get(row.table_name) ?? []; - tableRows.push(row); - grouped.set(row.table_name, tableRows); - } - return grouped; -} - /** @internal */ -export function preparePostgresReadOnlyQuery( - sql: string, - params?: Record, -): { sql: string; params?: unknown[] } { - if (!params) { - return { sql, params: undefined }; - } - const paramNames = Object.keys(params); - const values: unknown[] = new Array(paramNames.length); - const paramIndexMap = new Map(); - paramNames.forEach((name, index) => { - paramIndexMap.set(name, index + 1); - values[index] = params[name]; - }); - const sortedKeys = [...paramNames].sort((a, b) => b.length - a.length); - let parameterizedQuery = sql; - for (const name of sortedKeys) { - parameterizedQuery = parameterizedQuery.replace(new RegExp(`:${name}\\b`, 'g'), `$${paramIndexMap.get(name)}`); - } - return { sql: parameterizedQuery, params: values }; -} - -function primaryKeyMap(rows: PostgresPrimaryKeyRow[]): Map> { - const grouped = new Map>(); - for (const row of rows) { - const columns = grouped.get(row.table_name) ?? new Set(); - columns.add(row.column_name); - grouped.set(row.table_name, columns); - } - 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'; -} - -// 57014 = query_canceled, which is how statement_timeout surfaces. -function isPostgresTimeoutError(error: unknown): boolean { - return Boolean(error) && typeof error === 'object' && (error as { code?: unknown }).code === '57014'; -} - -function queryRows(result: KtxPostgresQueryResult): unknown[][] { - const headers = (result.fields ?? []).map((field) => field.name); - return result.rows.map((row) => headers.map((header) => row[header])); -} - -function finiteNumber(value: unknown): number | null { - const parsed = Number(value); - return Number.isFinite(parsed) ? parsed : null; -} - -function stringConfigValue( - connection: KtxPostgresConnectionConfig | undefined, - key: keyof KtxPostgresConnectionConfig, - env: NodeJS.ProcessEnv, -): string | undefined { - const value = connection?.[key]; - return typeof value === 'string' && value.trim().length > 0 ? resolveStringReference(value.trim(), env) : undefined; -} - - -function numberValue(value: unknown): number | undefined { - return typeof value === 'number' && Number.isFinite(value) ? value : undefined; -} - -function positiveIntegerConfigValue(input: { - connection: KtxPostgresConnectionConfig; - key: keyof KtxPostgresConnectionConfig; - connectionId: string; - defaultValue: number; -}): number { - const value = input.connection[input.key]; - if (value === undefined) { - return input.defaultValue; - } - const numberValue = Number(value); - if (!Number.isInteger(numberValue) || numberValue < 1) { - throw new Error(`connections.${input.connectionId}.${String(input.key)} must be a positive integer`); - } - return numberValue; -} - -function parsePostgresUrl(url: string): Partial { - const parsed = new URL(url); - const sslmode = parsed.searchParams.get('sslmode') ?? undefined; - return { - host: parsed.hostname, - port: parsed.port ? Number(parsed.port) : undefined, - database: parsed.pathname.replace(/^\/+/, '') || undefined, - username: parsed.username ? decodeURIComponent(parsed.username) : undefined, - password: parsed.password ? decodeURIComponent(parsed.password) : undefined, - ...(sslmode ? { sslmode } : {}), - }; -} - -function normalizedSslMode(connection: KtxPostgresConnectionConfig): string | undefined { - const value = connection.sslmode ?? connection.sslMode; - return typeof value === 'string' && value.trim().length > 0 ? value.trim().toLowerCase() : undefined; -} - -function schemasFromConnection(connection: KtxPostgresConnectionConfig): string[] { - if (Array.isArray(connection.schemas) && connection.schemas.length > 0) { - return connection.schemas.filter((schema): schema is string => typeof schema === 'string' && schema.length > 0); - } - return typeof connection.schema === 'string' && connection.schema.length > 0 ? [connection.schema] : ['public']; -} - -function searchPathSchemasFromConnection(connection: KtxPostgresConnectionConfig): string[] { - const schemas = schemasFromConnection(connection); - return schemas.includes('public') ? schemas : [...schemas, 'public']; -} +export const preparePostgresReadOnlyQuery = preparePgReadOnlyQuery; export function isKtxPostgresConnectionConfig( connection: KtxPostgresConnectionConfig | undefined, @@ -354,53 +161,13 @@ export function postgresPoolConfigFromConfig(input: { if (!isKtxPostgresConnectionConfig(input.connection)) { throw new Error(`Native PostgreSQL connector cannot run driver "${inputDriver}"`); } - - const env = input.env ?? process.env; - const referencedUrl = stringConfigValue(input.connection, 'url', env); - const urlConfig = referencedUrl ? parsePostgresUrl(referencedUrl) : {}; - const merged: KtxPostgresConnectionConfig = { ...urlConfig, ...input.connection }; - const host = stringConfigValue(merged, 'host', env); - const database = stringConfigValue(merged, 'database', env); - const user = stringConfigValue(merged, 'username', env) ?? stringConfigValue(merged, 'user', env); - const password = stringConfigValue(merged, 'password', env); - const sslmode = normalizedSslMode(merged); - const maxConnections = positiveIntegerConfigValue({ - connection: merged, - key: 'maxConnections', + return pgPoolConfigFromConfig({ connectionId: input.connectionId, - defaultValue: 10, + connection: input.connection, + defaultPort: 5432, + connectorLabel: 'Native PostgreSQL connector', + env: input.env, }); - - if (!referencedUrl && !host) { - throw new Error(`Native PostgreSQL connector requires connections.${input.connectionId}.host or url`); - } - if (!database && !referencedUrl) { - throw new Error(`Native PostgreSQL connector requires connections.${input.connectionId}.database or url`); - } - if (!user && !referencedUrl) { - throw new Error(`Native PostgreSQL connector requires connections.${input.connectionId}.username, user, or url`); - } - - const config: KtxPostgresPoolConfig = { - max: maxConnections, - idleTimeoutMillis: 30_000, - connectionTimeoutMillis: 10_000, - ...(referencedUrl && sslmode !== 'prefer' && sslmode !== 'disable' - ? { connectionString: referencedUrl } - : { host, port: numberValue(merged.port) ?? 5432, database, user, password }), - }; - const searchPathSchemas = searchPathSchemasFromConnection(merged); - // statement_timeout (ms) bounds every query on connections from this pool, so - // the server itself aborts a runaway query and frees the connection cleanly. - const serverOptions = [`-c statement_timeout=${resolveQueryDeadlineMs(merged)}`]; - if (searchPathSchemas.length > 0) { - serverOptions.unshift(`-c search_path=${searchPathSchemas.join(',')}`); - } - config.options = serverOptions.join(' '); - if (merged.ssl && sslmode !== 'prefer' && sslmode !== 'disable') { - config.ssl = { rejectUnauthorized: merged.rejectUnauthorized ?? true }; - } - return config; } export class KtxPostgresScanConnector implements KtxScanConnector { @@ -436,7 +203,7 @@ export class KtxPostgresScanConnector implements KtxScanConnector { connection: options.connection, env: options.env, }); - this.poolFactory = options.poolFactory ?? new DefaultPostgresPoolFactory(); + this.poolFactory = options.poolFactory ?? new DefaultPgPoolFactory(); this.endpointResolver = options.endpointResolver; this.now = options.now ?? (() => new Date()); this.deadlineMs = resolveQueryDeadlineMs(this.connection); @@ -832,7 +599,7 @@ export class KtxPostgresScanConnector implements KtxScanConnector { rowCount: result.rows.length, }; } catch (error) { - if (isPostgresTimeoutError(error)) { + if (isPgTimeoutError(error)) { throw queryDeadlineExceededError(this.deadlineMs, { cause: error }); } throw error; diff --git a/packages/cli/src/connectors/redshift/connector.ts b/packages/cli/src/connectors/redshift/connector.ts index 620d6517c..8f09e0cc0 100644 --- a/packages/cli/src/connectors/redshift/connector.ts +++ b/packages/cli/src/connectors/redshift/connector.ts @@ -1,5 +1,25 @@ -import { resolveStringReference } from '../shared/string-reference.js'; import { getSqlDialectForDriver } from '../../context/connections/dialects.js'; +import { resolveQueryDeadlineMs, queryDeadlineExceededError } from '../../context/connections/query-deadline.js'; +import { + DefaultPgPoolFactory, + PG_OID_TYPE_MAP, + finiteNumber, + groupByTable, + isDeniedError, + isPgTimeoutError, + numberValue, + pgPoolConfigFromConfig, + preparePgReadOnlyQuery, + primaryKeyMap, + queryRows, + schemasFromConnection, + type KtxPgConnectionConfig, + type KtxPgEndpointResolver, + type KtxPgPool, + type KtxPgPoolConfig, + type KtxPgPoolFactory, + type KtxPgResolvedEndpoint, +} from '../shared/pg-wire.js'; import { assertReadOnlySql, limitSqlForExecution } from '../../context/connections/read-only-sql.js'; import { tryConstraintQuery } from '../../context/scan/constraint-discovery.js'; import { scopedTableNames } from '../../context/scan/table-ref.js'; @@ -26,95 +46,13 @@ import { type KtxTableSampleInput, type KtxTableSampleResult, } from '../../context/scan/types.js'; -import { Pool } from 'pg'; - -const REDSHIFT_OID_TYPE_MAP: Record = { - 16: 'boolean', - 20: 'bigint', - 21: 'smallint', - 23: 'integer', - 25: 'text', - 700: 'real', - 701: 'double precision', - 1043: 'varchar', - 1082: 'date', - 1114: 'timestamp', - 1184: 'timestamptz', - 1700: 'numeric', - 2950: 'uuid', - 3802: 'jsonb', - 114: 'json', - 1009: 'text[]', - 1007: 'integer[]', - 1016: 'bigint[]', -}; - -export interface KtxRedshiftConnectionConfig { - driver?: string; - host?: string; - port?: number; - database?: string; - username?: string; - user?: string; - password?: string; - url?: string; - schema?: string; - schemas?: string[]; - ssl?: boolean; - sslmode?: string; - sslMode?: string; - rejectUnauthorized?: boolean; - maxConnections?: number; - [key: string]: unknown; -} -export interface KtxRedshiftPoolConfig { - host?: string; - port?: number; - database?: string; - user?: string; - password?: string; - connectionString?: string; - max: number; - idleTimeoutMillis: number; - connectionTimeoutMillis: number; - options?: string; - ssl?: { rejectUnauthorized: boolean }; -} - -interface KtxRedshiftQueryResult { - fields?: Array<{ name: string; dataTypeID: number }>; - rows: Record[]; -} - -interface KtxRedshiftClient { - query(sql: string, params?: unknown[]): Promise; - release(): void; -} - -interface KtxRedshiftPool { - connect(): Promise; - end(): Promise; - on?(event: 'error', listener: (error: Error) => void): void; -} - -export interface KtxRedshiftPoolFactory { - createPool(config: KtxRedshiftPoolConfig): KtxRedshiftPool; -} - -interface KtxRedshiftResolvedEndpoint { - host: string; - port: number; - close?: () => Promise; -} - -export interface KtxRedshiftEndpointResolver { - resolve(input: { - host: string; - port: number; - connection: KtxRedshiftConnectionConfig; - }): Promise; -} +export type KtxRedshiftConnectionConfig = KtxPgConnectionConfig; +export type KtxRedshiftPoolConfig = KtxPgPoolConfig; +type KtxRedshiftPool = KtxPgPool; +export type KtxRedshiftPoolFactory = KtxPgPoolFactory; +type KtxRedshiftResolvedEndpoint = KtxPgResolvedEndpoint; +export type KtxRedshiftEndpointResolver = KtxPgEndpointResolver; export interface KtxRedshiftScanConnectorOptions { connectionId: string; @@ -203,133 +141,8 @@ interface RedshiftStatsRow { estimated_cardinality: unknown; } -class DefaultRedshiftPoolFactory implements KtxRedshiftPoolFactory { - createPool(config: KtxRedshiftPoolConfig): KtxRedshiftPool { - return new Pool(config); - } -} - -function groupByTable(rows: T[]): Map { - const grouped = new Map(); - for (const row of rows) { - const tableRows = grouped.get(row.table_name) ?? []; - tableRows.push(row); - grouped.set(row.table_name, tableRows); - } - return grouped; -} - /** @internal */ -export function prepareRedshiftReadOnlyQuery( - sql: string, - params?: Record, -): { sql: string; params?: unknown[] } { - if (!params) { - return { sql, params: undefined }; - } - const paramNames = Object.keys(params); - const values: unknown[] = new Array(paramNames.length); - const paramIndexMap = new Map(); - paramNames.forEach((name, index) => { - paramIndexMap.set(name, index + 1); - values[index] = params[name]; - }); - const sortedKeys = [...paramNames].sort((a, b) => b.length - a.length); - let parameterizedQuery = sql; - for (const name of sortedKeys) { - parameterizedQuery = parameterizedQuery.replace(new RegExp(`:${name}\\b`, 'g'), `$${paramIndexMap.get(name)}`); - } - return { sql: parameterizedQuery, params: values }; -} - -function primaryKeyMap(rows: RedshiftPrimaryKeyRow[]): Map> { - const grouped = new Map>(); - for (const row of rows) { - const columns = grouped.get(row.table_name) ?? new Set(); - columns.add(row.column_name); - grouped.set(row.table_name, columns); - } - 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: KtxRedshiftQueryResult): unknown[][] { - const headers = (result.fields ?? []).map((field) => field.name); - return result.rows.map((row) => headers.map((header) => row[header])); -} - -function finiteNumber(value: unknown): number | null { - const parsed = Number(value); - return Number.isFinite(parsed) ? parsed : null; -} - -function stringConfigValue( - connection: KtxRedshiftConnectionConfig | undefined, - key: keyof KtxRedshiftConnectionConfig, - env: NodeJS.ProcessEnv, -): string | undefined { - const value = connection?.[key]; - return typeof value === 'string' && value.trim().length > 0 ? resolveStringReference(value.trim(), env) : undefined; -} - - -function numberValue(value: unknown): number | undefined { - return typeof value === 'number' && Number.isFinite(value) ? value : undefined; -} - -function positiveIntegerConfigValue(input: { - connection: KtxRedshiftConnectionConfig; - key: keyof KtxRedshiftConnectionConfig; - connectionId: string; - defaultValue: number; -}): number { - const value = input.connection[input.key]; - if (value === undefined) { - return input.defaultValue; - } - const numberValue = Number(value); - if (!Number.isInteger(numberValue) || numberValue < 1) { - throw new Error(`connections.${input.connectionId}.${String(input.key)} must be a positive integer`); - } - return numberValue; -} - -function parseRedshiftUrl(url: string): Partial { - const parsed = new URL(url); - const sslmode = parsed.searchParams.get('sslmode') ?? undefined; - return { - host: parsed.hostname, - port: parsed.port ? Number(parsed.port) : undefined, - database: parsed.pathname.replace(/^\/+/, '') || undefined, - username: parsed.username ? decodeURIComponent(parsed.username) : undefined, - password: parsed.password ? decodeURIComponent(parsed.password) : undefined, - ...(sslmode ? { sslmode } : {}), - }; -} - -function normalizedSslMode(connection: KtxRedshiftConnectionConfig): string | undefined { - const value = connection.sslmode ?? connection.sslMode; - return typeof value === 'string' && value.trim().length > 0 ? value.trim().toLowerCase() : undefined; -} - -function schemasFromConnection(connection: KtxRedshiftConnectionConfig): string[] { - if (Array.isArray(connection.schemas) && connection.schemas.length > 0) { - return connection.schemas.filter((schema): schema is string => typeof schema === 'string' && schema.length > 0); - } - return typeof connection.schema === 'string' && connection.schema.length > 0 ? [connection.schema] : ['public']; -} - -function searchPathSchemasFromConnection(connection: KtxRedshiftConnectionConfig): string[] { - const schemas = schemasFromConnection(connection); - return schemas.includes('public') ? schemas : [...schemas, 'public']; -} +export const prepareRedshiftReadOnlyQuery = preparePgReadOnlyQuery; export function isKtxRedshiftConnectionConfig( connection: KtxRedshiftConnectionConfig | undefined, @@ -348,49 +161,13 @@ export function redshiftPoolConfigFromConfig(input: { if (!isKtxRedshiftConnectionConfig(input.connection)) { throw new Error(`Native Redshift connector cannot run driver "${inputDriver}"`); } - - const env = input.env ?? process.env; - const referencedUrl = stringConfigValue(input.connection, 'url', env); - const urlConfig = referencedUrl ? parseRedshiftUrl(referencedUrl) : {}; - const merged: KtxRedshiftConnectionConfig = { ...urlConfig, ...input.connection }; - const host = stringConfigValue(merged, 'host', env); - const database = stringConfigValue(merged, 'database', env); - const user = stringConfigValue(merged, 'username', env) ?? stringConfigValue(merged, 'user', env); - const password = stringConfigValue(merged, 'password', env); - const sslmode = normalizedSslMode(merged); - const maxConnections = positiveIntegerConfigValue({ - connection: merged, - key: 'maxConnections', + return pgPoolConfigFromConfig({ connectionId: input.connectionId, - defaultValue: 10, + connection: input.connection, + defaultPort: 5439, + connectorLabel: 'Native Redshift connector', + env: input.env, }); - - if (!referencedUrl && !host) { - throw new Error(`Native Redshift connector requires connections.${input.connectionId}.host or url`); - } - if (!database && !referencedUrl) { - throw new Error(`Native Redshift connector requires connections.${input.connectionId}.database or url`); - } - if (!user && !referencedUrl) { - throw new Error(`Native Redshift connector requires connections.${input.connectionId}.username, user, or url`); - } - - const config: KtxRedshiftPoolConfig = { - max: maxConnections, - idleTimeoutMillis: 30_000, - connectionTimeoutMillis: 10_000, - ...(referencedUrl && sslmode !== 'prefer' && sslmode !== 'disable' - ? { connectionString: referencedUrl } - : { host, port: numberValue(merged.port) ?? 5439, database, user, password }), - }; - const searchPathSchemas = searchPathSchemasFromConnection(merged); - if (searchPathSchemas.length > 0) { - config.options = `-c search_path=${searchPathSchemas.join(',')}`; - } - if (merged.ssl && sslmode !== 'prefer' && sslmode !== 'disable') { - config.ssl = { rejectUnauthorized: merged.rejectUnauthorized ?? true }; - } - return config; } export class KtxRedshiftScanConnector implements KtxScanConnector { @@ -412,6 +189,7 @@ export class KtxRedshiftScanConnector implements KtxScanConnector { private readonly poolFactory: KtxRedshiftPoolFactory; private readonly endpointResolver?: KtxRedshiftEndpointResolver; private readonly now: () => Date; + private readonly deadlineMs: number; private readonly dialect = getSqlDialectForDriver('redshift'); private pool: KtxRedshiftPool | null = null; private lastIdlePoolError: Error | null = null; @@ -425,9 +203,10 @@ export class KtxRedshiftScanConnector implements KtxScanConnector { connection: options.connection, env: options.env, }); - this.poolFactory = options.poolFactory ?? new DefaultRedshiftPoolFactory(); + this.poolFactory = options.poolFactory ?? new DefaultPgPoolFactory(); this.endpointResolver = options.endpointResolver; this.now = options.now ?? (() => new Date()); + this.deadlineMs = resolveQueryDeadlineMs(this.connection); this.id = `redshift:${options.connectionId}`; } @@ -807,11 +586,16 @@ export class KtxRedshiftScanConnector implements KtxScanConnector { const result = await client.query(assertReadOnlySql(sql), Array.isArray(params) ? params : undefined); return { headers: (result.fields ?? []).map((field) => field.name), - headerTypes: (result.fields ?? []).map((field) => REDSHIFT_OID_TYPE_MAP[field.dataTypeID] ?? `oid:${field.dataTypeID}`), + headerTypes: (result.fields ?? []).map((field) => PG_OID_TYPE_MAP[field.dataTypeID] ?? `oid:${field.dataTypeID}`), rows: queryRows(result), totalRows: result.rows.length, rowCount: result.rows.length, }; + } catch (error) { + if (isPgTimeoutError(error)) { + throw queryDeadlineExceededError(this.deadlineMs, { cause: error }); + } + throw error; } finally { client.release(); } diff --git a/packages/cli/src/connectors/shared/pg-wire.ts b/packages/cli/src/connectors/shared/pg-wire.ts new file mode 100644 index 000000000..14d9b96f7 --- /dev/null +++ b/packages/cli/src/connectors/shared/pg-wire.ts @@ -0,0 +1,292 @@ +import { resolveStringReference } from './string-reference.js'; +import { resolveQueryDeadlineMs } from '../../context/connections/query-deadline.js'; +import { Pool } from 'pg'; + +/** + * Shared wire layer for connectors that speak the PostgreSQL wire protocol + * (PostgreSQL and Amazon Redshift). Holds the `pg` pool plumbing, connection + * config resolution, URL parsing, and query helpers so protocol-level behaviour + * stays in one place. Driver-specific metadata introspection lives in each + * connector. + */ + +export const PG_OID_TYPE_MAP: Record = { + 16: 'boolean', + 20: 'bigint', + 21: 'smallint', + 23: 'integer', + 25: 'text', + 700: 'real', + 701: 'double precision', + 1043: 'varchar', + 1082: 'date', + 1114: 'timestamp', + 1184: 'timestamptz', + 1700: 'numeric', + 2950: 'uuid', + 3802: 'jsonb', + 114: 'json', + 1009: 'text[]', + 1007: 'integer[]', + 1016: 'bigint[]', +}; + +export interface KtxPgConnectionConfig { + driver?: string; + host?: string; + port?: number; + database?: string; + username?: string; + user?: string; + password?: string; + url?: string; + schema?: string; + schemas?: string[]; + ssl?: boolean; + sslmode?: string; + sslMode?: string; + rejectUnauthorized?: boolean; + maxConnections?: number; + [key: string]: unknown; +} + +export interface KtxPgPoolConfig { + host?: string; + port?: number; + database?: string; + user?: string; + password?: string; + connectionString?: string; + max: number; + idleTimeoutMillis: number; + connectionTimeoutMillis: number; + options?: string; + ssl?: { rejectUnauthorized: boolean }; +} + +export interface KtxPgQueryResult { + fields?: Array<{ name: string; dataTypeID: number }>; + rows: Record[]; +} + +interface KtxPgClient { + query(sql: string, params?: unknown[]): Promise; + release(): void; +} + +export interface KtxPgPool { + connect(): Promise; + end(): Promise; + on?(event: 'error', listener: (error: Error) => void): void; +} + +export interface KtxPgPoolFactory { + createPool(config: KtxPgPoolConfig): KtxPgPool; +} + +export interface KtxPgResolvedEndpoint { + host: string; + port: number; + close?: () => Promise; +} + +export interface KtxPgEndpointResolver { + resolve(input: { host: string; port: number; connection: KtxPgConnectionConfig }): Promise; +} + +export class DefaultPgPoolFactory implements KtxPgPoolFactory { + createPool(config: KtxPgPoolConfig): KtxPgPool { + return new Pool(config); + } +} + +export function groupByTable(rows: T[]): Map { + const grouped = new Map(); + for (const row of rows) { + const tableRows = grouped.get(row.table_name) ?? []; + tableRows.push(row); + grouped.set(row.table_name, tableRows); + } + return grouped; +} + +/** @internal */ +export function preparePgReadOnlyQuery( + sql: string, + params?: Record, +): { sql: string; params?: unknown[] } { + if (!params) { + return { sql, params: undefined }; + } + const paramNames = Object.keys(params); + const values: unknown[] = new Array(paramNames.length); + const paramIndexMap = new Map(); + paramNames.forEach((name, index) => { + paramIndexMap.set(name, index + 1); + values[index] = params[name]; + }); + const sortedKeys = [...paramNames].sort((a, b) => b.length - a.length); + let parameterizedQuery = sql; + for (const name of sortedKeys) { + parameterizedQuery = parameterizedQuery.replace(new RegExp(`:${name}\\b`, 'g'), `$${paramIndexMap.get(name)}`); + } + return { sql: parameterizedQuery, params: values }; +} + +export function primaryKeyMap( + rows: T[], +): Map> { + const grouped = new Map>(); + for (const row of rows) { + const columns = grouped.get(row.table_name) ?? new Set(); + columns.add(row.column_name); + grouped.set(row.table_name, columns); + } + return grouped; +} + +export function isDeniedError(error: unknown): boolean { + if (!error || typeof error !== 'object') { + return false; + } + const code = (error as { code?: unknown }).code; + return code === '42501' || code === '42P01'; +} + +// 57014 = query_canceled, which is how statement_timeout surfaces. +export function isPgTimeoutError(error: unknown): boolean { + return Boolean(error) && typeof error === 'object' && (error as { code?: unknown }).code === '57014'; +} + +export function queryRows(result: KtxPgQueryResult): unknown[][] { + const headers = (result.fields ?? []).map((field) => field.name); + return result.rows.map((row) => headers.map((header) => row[header])); +} + +export function finiteNumber(value: unknown): number | null { + const parsed = Number(value); + return Number.isFinite(parsed) ? parsed : null; +} + +function stringConfigValue( + connection: KtxPgConnectionConfig | undefined, + key: keyof KtxPgConnectionConfig, + env: NodeJS.ProcessEnv, +): string | undefined { + const value = connection?.[key]; + return typeof value === 'string' && value.trim().length > 0 ? resolveStringReference(value.trim(), env) : undefined; +} + +export function numberValue(value: unknown): number | undefined { + return typeof value === 'number' && Number.isFinite(value) ? value : undefined; +} + +function positiveIntegerConfigValue(input: { + connection: KtxPgConnectionConfig; + key: keyof KtxPgConnectionConfig; + connectionId: string; + defaultValue: number; +}): number { + const value = input.connection[input.key]; + if (value === undefined) { + return input.defaultValue; + } + const numberValue = Number(value); + if (!Number.isInteger(numberValue) || numberValue < 1) { + throw new Error(`connections.${input.connectionId}.${String(input.key)} must be a positive integer`); + } + return numberValue; +} + +function parsePgUrl(url: string): Partial { + const parsed = new URL(url); + const sslmode = parsed.searchParams.get('sslmode') ?? undefined; + return { + host: parsed.hostname, + port: parsed.port ? Number(parsed.port) : undefined, + database: parsed.pathname.replace(/^\/+/, '') || undefined, + username: parsed.username ? decodeURIComponent(parsed.username) : undefined, + password: parsed.password ? decodeURIComponent(parsed.password) : undefined, + ...(sslmode ? { sslmode } : {}), + }; +} + +function normalizedSslMode(connection: KtxPgConnectionConfig): string | undefined { + const value = connection.sslmode ?? connection.sslMode; + return typeof value === 'string' && value.trim().length > 0 ? value.trim().toLowerCase() : undefined; +} + +export function schemasFromConnection(connection: KtxPgConnectionConfig): string[] { + if (Array.isArray(connection.schemas) && connection.schemas.length > 0) { + return connection.schemas.filter((schema): schema is string => typeof schema === 'string' && schema.length > 0); + } + return typeof connection.schema === 'string' && connection.schema.length > 0 ? [connection.schema] : ['public']; +} + +function searchPathSchemasFromConnection(connection: KtxPgConnectionConfig): string[] { + const schemas = schemasFromConnection(connection); + return schemas.includes('public') ? schemas : [...schemas, 'public']; +} + +/** + * Build a `pg` pool config from a ktx connection config. `defaultPort` and + * `connectorLabel` are the only driver-specific inputs: PostgreSQL defaults to + * 5432, Redshift to 5439, and the label prefixes validation errors. + * + * @internal + */ +export function pgPoolConfigFromConfig(input: { + connectionId: string; + connection: KtxPgConnectionConfig; + defaultPort: number; + connectorLabel: string; + env?: NodeJS.ProcessEnv; +}): KtxPgPoolConfig { + const env = input.env ?? process.env; + const { connectorLabel, connectionId } = input; + const referencedUrl = stringConfigValue(input.connection, 'url', env); + const urlConfig = referencedUrl ? parsePgUrl(referencedUrl) : {}; + const merged: KtxPgConnectionConfig = { ...urlConfig, ...input.connection }; + const host = stringConfigValue(merged, 'host', env); + const database = stringConfigValue(merged, 'database', env); + const user = stringConfigValue(merged, 'username', env) ?? stringConfigValue(merged, 'user', env); + const password = stringConfigValue(merged, 'password', env); + const sslmode = normalizedSslMode(merged); + const maxConnections = positiveIntegerConfigValue({ + connection: merged, + key: 'maxConnections', + connectionId, + defaultValue: 10, + }); + + if (!referencedUrl && !host) { + throw new Error(`${connectorLabel} requires connections.${connectionId}.host or url`); + } + if (!database && !referencedUrl) { + throw new Error(`${connectorLabel} requires connections.${connectionId}.database or url`); + } + if (!user && !referencedUrl) { + throw new Error(`${connectorLabel} requires connections.${connectionId}.username, user, or url`); + } + + const config: KtxPgPoolConfig = { + max: maxConnections, + idleTimeoutMillis: 30_000, + connectionTimeoutMillis: 10_000, + ...(referencedUrl && sslmode !== 'prefer' && sslmode !== 'disable' + ? { connectionString: referencedUrl } + : { host, port: numberValue(merged.port) ?? input.defaultPort, database, user, password }), + }; + const searchPathSchemas = searchPathSchemasFromConnection(merged); + // statement_timeout (ms) bounds every query on connections from this pool, so + // the server itself aborts a runaway query and frees the connection cleanly. + const serverOptions = [`-c statement_timeout=${resolveQueryDeadlineMs(merged)}`]; + if (searchPathSchemas.length > 0) { + serverOptions.unshift(`-c search_path=${searchPathSchemas.join(',')}`); + } + config.options = serverOptions.join(' '); + if (merged.ssl && sslmode !== 'prefer' && sslmode !== 'disable') { + config.ssl = { rejectUnauthorized: merged.rejectUnauthorized ?? true }; + } + return config; +} diff --git a/packages/cli/test/connectors/redshift/connector.test.ts b/packages/cli/test/connectors/redshift/connector.test.ts index 2f69fadc1..d064d6bcd 100644 --- a/packages/cli/test/connectors/redshift/connector.test.ts +++ b/packages/cli/test/connectors/redshift/connector.test.ts @@ -157,7 +157,7 @@ describe('KtxRedshiftScanConnector', () => { database: 'analytics', user: 'reader', password: 'test-password', // pragma: allowlist secret - options: '-c search_path=analytics,public', + options: '-c search_path=analytics,public -c statement_timeout=30000', ssl: { rejectUnauthorized: false }, }); const libpqPreferConfig = redshiftPoolConfigFromConfig({ From 0a82c2fdba722cf43d4a71a0cf1d7fb25c49453b Mon Sep 17 00:00:00 2001 From: Mayorkun Ayanshina Date: Thu, 9 Jul 2026 15:57:40 +0100 Subject: [PATCH 6/9] fix(redshift): correct SVV_TABLES filter, scoped predicates, and string aggregate Three Redshift-correctness fixes from review: - SVV_TABLES.table_type reports regular tables as BASE TABLE, not TABLE. The old filter dropped every ordinary table from listTables() and introspection. - Redshift does not accept a bound array for '= ANY($n)'. Scoped predicates now expand to positional IN ($n, $n+1, ...) placeholders, derived from the value count only. - Redshift has no STRING_AGG; the dialect now emits LISTAGG for sample value aggregation used by relationship profiling. Tests assert the generated SQL for each, so the previous behaviour would now fail. --- .../cli/src/connectors/redshift/connector.ts | 29 ++++++++++++++----- .../cli/src/connectors/redshift/dialect.ts | 5 +++- .../connectors/redshift/connector.test.ts | 11 +++++-- .../test/context/connections/dialects.test.ts | 2 +- 4 files changed, 34 insertions(+), 13 deletions(-) diff --git a/packages/cli/src/connectors/redshift/connector.ts b/packages/cli/src/connectors/redshift/connector.ts index 8f09e0cc0..7f5292f12 100644 --- a/packages/cli/src/connectors/redshift/connector.ts +++ b/packages/cli/src/connectors/redshift/connector.ts @@ -144,6 +144,17 @@ interface RedshiftStatsRow { /** @internal */ export const prepareRedshiftReadOnlyQuery = preparePgReadOnlyQuery; +/** + * Build a positional `IN ($n, $n+1, ...)` predicate. Redshift does not accept a + * bound array for `= ANY($1)` the way PostgreSQL does, so scoped predicates are + * expanded into one placeholder per value. Placeholders are derived from the + * value count only, never from the values themselves. + */ +function positionalInClause(values: readonly string[], columnExpression: string, startIndex: number): string { + const placeholders = values.map((_value, index) => `$${startIndex + index}`); + return `${columnExpression} IN (${placeholders.join(', ')})`; +} + export function isKtxRedshiftConnectionConfig( connection: KtxRedshiftConnectionConfig | undefined, ): connection is KtxRedshiftConnectionConfig { @@ -363,16 +374,17 @@ export class KtxRedshiftScanConnector implements KtxScanConnector { async listTables(schemas?: string[]): Promise { const filterSchemas = schemas ?? (await this.listSchemas()); if (filterSchemas.length === 0) return []; + const schemaPredicate = positionalInClause(filterSchemas, 'table_schema', 1); const rows = await this.queryRaw( ` SELECT table_schema AS schema_name, table_name AS table_name, CASE WHEN table_type = 'VIEW' THEN 'v' ELSE 'r' END AS table_kind FROM svv_tables - WHERE table_schema = ANY($1) - AND table_type IN ('TABLE', 'VIEW', 'EXTERNAL TABLE') + WHERE ${schemaPredicate} + AND table_type IN ('BASE TABLE', 'VIEW', 'EXTERNAL TABLE') ORDER BY table_schema, table_name `, - [filterSchemas], + [...filterSchemas], ); return rows.map((row) => ({ catalog: null, @@ -399,10 +411,11 @@ export class KtxRedshiftScanConnector implements KtxScanConnector { snapshotWarnings: KtxScanWarning[], ): Promise { if (scopedNames && scopedNames.length === 0) return []; - const svvTableScopeClause = scopedNames ? 'AND t.table_name = ANY($2)' : ''; - const svvColumnScopeClause = scopedNames ? 'AND table_name = ANY($2)' : ''; - const tableConstraintScopeClause = scopedNames ? 'AND tc.table_name = ANY($2)' : ''; - const scopeValues = scopedNames ? [scopedNames] : []; + // Schema is $1 in every query below, so scoped table names start at $2. + const svvTableScopeClause = scopedNames ? `AND ${positionalInClause(scopedNames, 't.table_name', 2)}` : ''; + const svvColumnScopeClause = scopedNames ? `AND ${positionalInClause(scopedNames, 'table_name', 2)}` : ''; + const tableConstraintScopeClause = scopedNames ? `AND ${positionalInClause(scopedNames, 'tc.table_name', 2)}` : ''; + const scopeValues = scopedNames ? [...scopedNames] : []; const tables = await this.queryRaw( ` SELECT @@ -414,7 +427,7 @@ export class KtxRedshiftScanConnector implements KtxScanConnector { LEFT JOIN svv_table_info ti ON ti.schema = t.table_schema AND ti.\"table\" = t.table_name WHERE t.table_schema = $1 - AND t.table_type IN ('TABLE', 'VIEW', 'EXTERNAL TABLE') + AND t.table_type IN ('BASE TABLE', 'VIEW', 'EXTERNAL TABLE') ${svvTableScopeClause} ORDER BY t.table_name `, diff --git a/packages/cli/src/connectors/redshift/dialect.ts b/packages/cli/src/connectors/redshift/dialect.ts index 462269082..436301bcd 100644 --- a/packages/cli/src/connectors/redshift/dialect.ts +++ b/packages/cli/src/connectors/redshift/dialect.ts @@ -150,7 +150,10 @@ export class KtxRedshiftDialect implements KtxDialect { } getSampleValueAggregation(innerSql: string): string { - return `(SELECT STRING_AGG(CAST(value AS TEXT), CHR(31)) FROM (${innerSql}) AS relationship_profile_values)`; + // Redshift has no STRING_AGG; LISTAGG is its string aggregate. WITHIN GROUP is + // optional here because profiling does not depend on value order. LISTAGG + // returns VARCHAR(MAX) and errors if the concatenation exceeds that limit. + return `(SELECT LISTAGG(CAST(value AS TEXT), CHR(31)) FROM (${innerSql}) AS relationship_profile_values)`; } generateCardinalitySampleQuery(tableName: string, columnName: string, sampleSize: number): string { diff --git a/packages/cli/test/connectors/redshift/connector.test.ts b/packages/cli/test/connectors/redshift/connector.test.ts index d064d6bcd..302b0f7eb 100644 --- a/packages/cli/test/connectors/redshift/connector.test.ts +++ b/packages/cli/test/connectors/redshift/connector.test.ts @@ -97,7 +97,7 @@ function metadataResults(): Map { ['FROM pg_stats s', { rows: [{ column_name: 'status', estimated_cardinality: '2' }] }], ['SELECT 1', { rows: [{ '?column?': 1 }], fields: [{ name: '?column?', dataTypeID: 23 }] }], [ - 'FROM svv_tables WHERE table_schema = ANY($1) AND table_type IN', + 'FROM svv_tables WHERE table_schema IN ($1) AND table_type IN', { rows: [ { schema_name: 'public', table_name: 'customers', table_kind: 'r' }, @@ -463,8 +463,13 @@ describe('KtxRedshiftScanConnector', () => { ); expect(snapshot.tables.map((table) => table.name)).toEqual(['orders']); const tablesQuery = queries.find((query) => query.sql.includes('FROM svv_tables t')); - expect(tablesQuery?.sql).toMatch(/t\.table_name = ANY\(\$2\)/); - expect(tablesQuery?.params).toEqual(['public', ['orders']]); + // Scoped predicates must expand to positional placeholders: Redshift rejects a + // bound array in `= ANY($n)`. + expect(tablesQuery?.sql).toMatch(/t\.table_name IN \(\$2\)/); + expect(tablesQuery?.params).toEqual(['public', 'orders']); + // SVV_TABLES reports regular tables as BASE TABLE, not TABLE. + expect(tablesQuery?.sql).toContain("'BASE TABLE'"); + expect(tablesQuery?.sql).not.toMatch(/IN \('TABLE'/); }); it('attaches an error listener to the pg pool', async () => { diff --git a/packages/cli/test/context/connections/dialects.test.ts b/packages/cli/test/context/connections/dialects.test.ts index 613d1531d..cd7b90482 100644 --- a/packages/cli/test/context/connections/dialects.test.ts +++ b/packages/cli/test/context/connections/dialects.test.ts @@ -85,7 +85,7 @@ const fixtures: DialectFixture[] = [ textLengthExpression: 'LENGTH(CAST("status" AS TEXT))', castToText: 'CAST("status" AS TEXT)', sampleValueAggregation: - '(SELECT STRING_AGG(CAST(value AS TEXT), CHR(31)) FROM (SELECT status AS value FROM orders) AS relationship_profile_values)', + '(SELECT LISTAGG(CAST(value AS TEXT), CHR(31)) FROM (SELECT status AS value FROM orders) AS relationship_profile_values)', cardinalityContains: 'SELECT COUNT(DISTINCT val) AS cardinality', randomizedCardinalityContains: 'ORDER BY RANDOM()', distinctValuesContains: 'SELECT DISTINCT "status"::text AS val', From bd01aa2a613cf6b29fbeb6e4afdeea42cce8a78c Mon Sep 17 00:00:00 2001 From: Mayorkun Ayanshina Date: Thu, 9 Jul 2026 16:00:44 +0100 Subject: [PATCH 7/9] fix(redshift): map redshift to its sqlglot dialect for SQL analysis sqlglot has a first-class redshift dialect that differs from postgres in identifier normalization (case-insensitive) and array index offset, so falling through to the postgres default parsed Redshift SQL under the wrong dialect. --- packages/cli/src/context/sql-analysis/dialect.ts | 1 + packages/cli/test/context/sql-analysis/dialect.test.ts | 1 + 2 files changed, 2 insertions(+) diff --git a/packages/cli/src/context/sql-analysis/dialect.ts b/packages/cli/src/context/sql-analysis/dialect.ts index 830922333..df1813680 100644 --- a/packages/cli/src/context/sql-analysis/dialect.ts +++ b/packages/cli/src/context/sql-analysis/dialect.ts @@ -8,6 +8,7 @@ import type { SqlAnalysisDialect } from './ports.js'; const SQLGLOT_DIALECTS: Record = { postgres: 'postgres', postgresql: 'postgres', + redshift: 'redshift', bigquery: 'bigquery', snowflake: 'snowflake', mysql: 'mysql', diff --git a/packages/cli/test/context/sql-analysis/dialect.test.ts b/packages/cli/test/context/sql-analysis/dialect.test.ts index 0166a1338..0e6349f54 100644 --- a/packages/cli/test/context/sql-analysis/dialect.test.ts +++ b/packages/cli/test/context/sql-analysis/dialect.test.ts @@ -4,6 +4,7 @@ import { sqlAnalysisDialectForDriver } from '../../../src/context/sql-analysis/d describe('sqlAnalysisDialectForDriver', () => { it('maps ktx.yaml driver names to sqlglot dialects', () => { expect(sqlAnalysisDialectForDriver('postgres')).toBe('postgres'); + expect(sqlAnalysisDialectForDriver('redshift')).toBe('redshift'); expect(sqlAnalysisDialectForDriver('bigquery')).toBe('bigquery'); expect(sqlAnalysisDialectForDriver('snowflake')).toBe('snowflake'); expect(sqlAnalysisDialectForDriver('mysql')).toBe('mysql'); From a0cf48911e6a52fd2bb0ff28e85a101afe68d4a0 Mon Sep 17 00:00:00 2001 From: Mayorkun Ayanshina Date: Thu, 9 Jul 2026 16:10:48 +0100 Subject: [PATCH 8/9] feat(redshift): add REDSHIFT warehouse identity Register REDSHIFT as a ConnectionType, map the redshift driver to it in DRIVER_TO_CONNECTION_TYPE, and give it its sqlglot dialect in the connection-type dialect map, so a Redshift connection is treated as a first-class warehouse target by local ingest and semantic-layer compute rather than only being scannable. --- packages/cli/src/context/connections/connection-type-dialect.ts | 1 + packages/cli/src/context/connections/connection-type.ts | 1 + .../cli/src/context/connections/local-warehouse-descriptor.ts | 1 + 3 files changed, 3 insertions(+) diff --git a/packages/cli/src/context/connections/connection-type-dialect.ts b/packages/cli/src/context/connections/connection-type-dialect.ts index f20231de5..72b46cf5c 100644 --- a/packages/cli/src/context/connections/connection-type-dialect.ts +++ b/packages/cli/src/context/connections/connection-type-dialect.ts @@ -2,6 +2,7 @@ import type { ConnectionType } from './connection-type.js'; const CONNECTION_TYPE_TO_SQLGLOT = { POSTGRESQL: 'postgres', + REDSHIFT: 'redshift', SQLITE: 'sqlite', DUCKDB: 'duckdb', SQLSERVER: 'tsql', diff --git a/packages/cli/src/context/connections/connection-type.ts b/packages/cli/src/context/connections/connection-type.ts index 8b12839cc..f7eedbfdb 100644 --- a/packages/cli/src/context/connections/connection-type.ts +++ b/packages/cli/src/context/connections/connection-type.ts @@ -2,6 +2,7 @@ import { z } from 'zod'; export const connectionTypeSchema = z.enum([ 'POSTGRESQL', + 'REDSHIFT', 'SQLITE', 'DUCKDB', 'SQLSERVER', diff --git a/packages/cli/src/context/connections/local-warehouse-descriptor.ts b/packages/cli/src/context/connections/local-warehouse-descriptor.ts index bb4235c2e..d76d7d489 100644 --- a/packages/cli/src/context/connections/local-warehouse-descriptor.ts +++ b/packages/cli/src/context/connections/local-warehouse-descriptor.ts @@ -22,6 +22,7 @@ export interface LocalConnectionInfo { const DRIVER_TO_CONNECTION_TYPE: Record = { postgres: 'POSTGRESQL', + redshift: 'REDSHIFT', sqlite: 'SQLITE', duckdb: 'DUCKDB', sqlserver: 'SQLSERVER', From d24dbaa152e00f82f48f87859656f4647873bdb7 Mon Sep 17 00:00:00 2001 From: Mayorkun Ayanshina Date: Thu, 9 Jul 2026 16:14:07 +0100 Subject: [PATCH 9/9] docs(redshift): note that setup does not manage Redshift connections ktx setup has no Redshift path yet, so the integration docs should not imply it can create one. Redshift is manual ktx.yaml config for now. --- docs-site/content/docs/integrations/primary-sources.mdx | 3 +++ 1 file changed, 3 insertions(+) diff --git a/docs-site/content/docs/integrations/primary-sources.mdx b/docs-site/content/docs/integrations/primary-sources.mdx index 4bc053fc2..7e80b0b4a 100644 --- a/docs-site/content/docs/integrations/primary-sources.mdx +++ b/docs-site/content/docs/integrations/primary-sources.mdx @@ -132,6 +132,9 @@ Redshift-specific system views for accurate results. Supports schema introspection, primary/foreign key detection, column statistics, table sampling, and read-only SQL execution. +Redshift connections are configured manually in `ktx.yaml`. `ktx setup` does not +yet create or manage them interactively. + ### Connection config ```yaml title="ktx.yaml"