diff --git a/docs-site/content/docs/integrations/primary-sources.mdx b/docs-site/content/docs/integrations/primary-sources.mdx
index b553a3a7e..7e80b0b4a 100644
--- a/docs-site/content/docs/integrations/primary-sources.mdx
+++ b/docs-site/content/docs/integrations/primary-sources.mdx
@@ -1,6 +1,6 @@
 ---
 title: Primary Sources
-description: Connect ktx to PostgreSQL, Snowflake, BigQuery, MySQL, ClickHouse, SQL Server, SQLite, DuckDB, MongoDB, or Amazon Athena.
+description: Connect ktx to PostgreSQL, Amazon Redshift, Snowflake, BigQuery, MySQL, ClickHouse, SQL Server, SQLite, DuckDB, MongoDB, or Amazon Athena.
 ---
 
 **ktx** connects to your data warehouse or database to build schema context,
@@ -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`, `duckdb`, `mongodb`, or `athena` |
+| `driver` | Yes | all connections | Connector driver such as `postgres`, `redshift`, `snowflake`, `bigquery`, `mysql`, `clickhouse`, `sqlserver`, `sqlite`, `duckdb`, `mongodb`, or `athena` |
 | `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 |
@@ -124,6 +124,78 @@ 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.
+
+Redshift connections are configured manually in `ktx.yaml`. `ktx setup` does not
+yet create or manage them interactively.
+
+### 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/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<number, string> = {
-  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<string, unknown>[];
-}
-
-interface KtxPostgresClient {
-  query(sql: string, params?: unknown[]): Promise<KtxPostgresQueryResult>;
-  release(): void;
-}
-
-interface KtxPostgresPool {
-  connect(): Promise<KtxPostgresClient>;
-  end(): Promise<void>;
-  on?(event: 'error', listener: (error: Error) => void): void;
-}
-
-export interface KtxPostgresPoolFactory {
-  createPool(config: KtxPostgresPoolConfig): KtxPostgresPool;
-}
-
-interface KtxPostgresResolvedEndpoint {
-  host: string;
-  port: number;
-  close?: () => Promise<void>;
-}
-
-export interface KtxPostgresEndpointResolver {
-  resolve(input: {
-    host: string;
-    port: number;
-    connection: KtxPostgresConnectionConfig;
-  }): Promise<KtxPostgresResolvedEndpoint>;
-}
+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<T extends { table_name: string }>(rows: T[]): Map<string, T[]> {
-  const grouped = new Map<string, T[]>();
-  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<string, unknown>,
-): { 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<string, number>();
-  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<string, Set<string>> {
-  const grouped = new Map<string, Set<string>>();
-  for (const row of rows) {
-    const columns = grouped.get(row.table_name) ?? new Set<string>();
-    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<KtxPostgresConnectionConfig> {
-  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
new file mode 100644
index 000000000..7f5292f12
--- /dev/null
+++ b/packages/cli/src/connectors/redshift/connector.ts
@@ -0,0 +1,631 @@
+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';
+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';
+
+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;
+  connection: KtxRedshiftConnectionConfig | undefined;
+  poolFactory?: KtxRedshiftPoolFactory;
+  endpointResolver?: KtxRedshiftEndpointResolver;
+  env?: NodeJS.ProcessEnv;
+  now?: () => Date;
+}
+
+export interface KtxRedshiftReadOnlyQueryInput extends KtxReadOnlyQueryInput {
+  params?: Record<string, unknown> | unknown[];
+}
+
+export interface KtxRedshiftColumnDistinctValuesOptions {
+  maxCardinality: number;
+  limit: number;
+  sampleSize?: number;
+}
+
+export interface KtxRedshiftColumnDistinctValuesResult {
+  values: string[] | null;
+  cardinality: number;
+}
+
+export interface KtxRedshiftColumnStatisticsResult {
+  cardinalityByColumn: Map<string, number>;
+}
+
+export interface KtxRedshiftTableSampleResult extends KtxTableSampleResult {
+  headerTypes?: string[];
+}
+
+type RedshiftTableRef = Pick<KtxTableRef, 'name'> & Partial<Pick<KtxTableRef, 'catalog' | 'db'>>;
+
+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;
+}
+
+/** @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 {
+  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}"`);
+  }
+  return pgPoolConfigFromConfig({
+    connectionId: input.connectionId,
+    connection: input.connection,
+    defaultPort: 5439,
+    connectorLabel: 'Native Redshift connector',
+    env: input.env,
+  });
+}
+
+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 deadlineMs: number;
+  private readonly dialect = getSqlDialectForDriver('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 DefaultPgPoolFactory();
+    this.endpointResolver = options.endpointResolver;
+    this.now = options.now ?? (() => new Date());
+    this.deadlineMs = resolveQueryDeadlineMs(this.connection);
+    this.id = `redshift:${options.connectionId}`;
+  }
+
+  async testConnection(): Promise<KtxConnectorTestResult> {
+    try {
+      await this.query('SELECT 1');
+      return { success: true };
+    } catch (error) {
+      return connectorTestFailure(error);
+    }
+  }
+
+  async introspect(input: KtxScanInput, _ctx: KtxScanContext): Promise<KtxSchemaSnapshot> {
+    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<KtxRedshiftTableSampleResult> {
+    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<KtxColumnSampleResult> {
+    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<KtxColumnStatsResult | null> {
+    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<KtxQueryResult> {
+    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<KtxRedshiftColumnDistinctValuesResult | null> {
+    const sampleSize = options.sampleSize ?? 10000;
+    const tableName = this.qTableName(table);
+    const quotedColumn = this.dialect.quoteIdentifier(columnName);
+    const cardinalityRows = await this.queryRaw<RedshiftCountRow>(
+      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<RedshiftDistinctValueRow>(
+      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<KtxRedshiftColumnStatisticsResult | null> {
+    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<RedshiftStatsRow>(sql);
+    const cardinalityByColumn = new Map<string, number>();
+    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<number> {
+    const tableRef =
+      typeof table === 'string'
+        ? { catalog: null, db: schemasFromConnection(this.connection)[0] ?? 'public', name: table }
+        : table;
+    const rows = await this.queryRaw<RedshiftCountRow>(`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<string[]> {
+    const rows = await this.queryRaw<RedshiftSchemaRow>(`
+      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<KtxTableListEntry[]> {
+    const filterSchemas = schemas ?? (await this.listSchemas());
+    if (filterSchemas.length === 0) return [];
+    const schemaPredicate = positionalInClause(filterSchemas, 'table_schema', 1);
+    const rows = await this.queryRaw<RedshiftTableListRow>(
+      `
+      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 ${schemaPredicate}
+        AND table_type IN ('BASE 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<void> {
+    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<KtxSchemaTable[]> {
+    if (scopedNames && scopedNames.length === 0) return [];
+    // 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<RedshiftTableRow>(
+      `
+      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 ('BASE TABLE', 'VIEW', 'EXTERNAL TABLE')
+        ${svvTableScopeClause}
+      ORDER BY t.table_name
+      `,
+      [schema, ...scopeValues],
+    );
+    const columns = await this.queryRaw<RedshiftColumnRow>(
+      `
+      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<RedshiftPrimaryKeyRow>(
+          `
+      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<RedshiftForeignKeyRow>(
+          `
+      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<string>(),
+        foreignKeysByTable.get(table.table_name) ?? [],
+      ),
+    );
+  }
+
+  private toSchemaTable(
+    schema: string,
+    table: RedshiftTableRow,
+    columns: RedshiftColumnRow[],
+    primaryKeys: Set<string>,
+    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<string>): 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<KtxRedshiftPool> {
+    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<T>(sql: string, params?: unknown[]): Promise<T[]> {
+    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<string, unknown> | unknown[]): Promise<KtxQueryResult> {
+    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) => 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();
+    }
+  }
+
+  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..436301bcd
--- /dev/null
+++ b/packages/cli/src/connectors/redshift/dialect.ts
@@ -0,0 +1,213 @@
+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<KtxTableRef, 'name'> & Partial<Pick<KtxTableRef, 'catalog' | 'db'>>;
+
+/** @internal */
+export class KtxRedshiftDialect implements KtxDialect {
+  readonly type = 'redshift' as const;
+
+  private readonly typeMappings: Record<string, KtxSchemaDimensionType> = {
+    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 {
+    // 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 {
+    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/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<number, string> = {
+  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<string, unknown>[];
+}
+
+interface KtxPgClient {
+  query(sql: string, params?: unknown[]): Promise<KtxPgQueryResult>;
+  release(): void;
+}
+
+export interface KtxPgPool {
+  connect(): Promise<KtxPgClient>;
+  end(): Promise<void>;
+  on?(event: 'error', listener: (error: Error) => void): void;
+}
+
+export interface KtxPgPoolFactory {
+  createPool(config: KtxPgPoolConfig): KtxPgPool;
+}
+
+export interface KtxPgResolvedEndpoint {
+  host: string;
+  port: number;
+  close?: () => Promise<void>;
+}
+
+export interface KtxPgEndpointResolver {
+  resolve(input: { host: string; port: number; connection: KtxPgConnectionConfig }): Promise<KtxPgResolvedEndpoint>;
+}
+
+export class DefaultPgPoolFactory implements KtxPgPoolFactory {
+  createPool(config: KtxPgPoolConfig): KtxPgPool {
+    return new Pool(config);
+  }
+}
+
+export function groupByTable<T extends { table_name: string }>(rows: T[]): Map<string, T[]> {
+  const grouped = new Map<string, T[]>();
+  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<string, unknown>,
+): { 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<string, number>();
+  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<T extends { table_name: string; column_name: string }>(
+  rows: T[],
+): Map<string, Set<string>> {
+  const grouped = new Map<string, Set<string>>();
+  for (const row of rows) {
+    const columns = grouped.get(row.table_name) ?? new Set<string>();
+    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<KtxPgConnectionConfig> {
+  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/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/dialects.ts b/packages/cli/src/context/connections/dialects.ts
index 8512ed577..76d44d2dc 100644
--- a/packages/cli/src/context/connections/dialects.ts
+++ b/packages/cli/src/context/connections/dialects.ts
@@ -5,6 +5,7 @@ import { KtxDuckDbDialect } from '../../connectors/duckdb/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';
@@ -61,6 +62,7 @@ const sqlDialectFactories: Record<KtxSqlDriver, () => KtxSqlDialect> = {
   duckdb: () => new KtxDuckDbDialect(),
   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 6a2d45890..183327731 100644
--- a/packages/cli/src/context/connections/drivers.ts
+++ b/packages/cli/src/context/connections/drivers.ts
@@ -169,6 +169,27 @@ export const driverRegistrations: Record<KtxConnectionDriver, KtxDriverRegistrat
       };
     },
   },
+  redshift: {
+    driver: 'redshift',
+    scopeConfigKey: 'schemas',
+    hasHistoricSqlReader: false,
+    load: async () => {
+      const m = await import('../../connectors/redshift/connector.js');
+      return {
+        isConnectionConfig: (connection) => {
+          const typedConnection = connection as Parameters<typeof m.isKtxRedshiftConnectionConfig>[0];
+          return m.isKtxRedshiftConnectionConfig(typedConnection);
+        },
+        createScanConnector: ({ connectionId, connection }) => {
+          const typedConnection = connection as Parameters<typeof m.isKtxRedshiftConnectionConfig>[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/connections/local-warehouse-descriptor.ts b/packages/cli/src/context/connections/local-warehouse-descriptor.ts
index 39cdddbf5..e51863493 100644
--- a/packages/cli/src/context/connections/local-warehouse-descriptor.ts
+++ b/packages/cli/src/context/connections/local-warehouse-descriptor.ts
@@ -24,6 +24,7 @@ export interface LocalConnectionInfo {
 
 const DRIVER_TO_CONNECTION_TYPE: Record<string, ConnectionType> = {
   postgres: 'POSTGRESQL',
+  redshift: 'REDSHIFT',
   sqlite: 'SQLITE',
   duckdb: 'DUCKDB',
   sqlserver: 'SQLSERVER',
diff --git a/packages/cli/src/context/project/driver-schemas.ts b/packages/cli/src/context/project/driver-schemas.ts
index 72d1a1335..e67e0afc2 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',
@@ -56,6 +57,7 @@ function warehouseConnectionSchema<const Driver extends WarehouseDriver>(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 bf72558c4..a72371327 100644
--- a/packages/cli/src/context/scan/types.ts
+++ b/packages/cli/src/context/scan/types.ts
@@ -10,7 +10,8 @@ export type KtxConnectionDriver =
   | 'mysql'
   | 'clickhouse'
   | 'athena'
-  | 'mongodb';
+  | 'mongodb'
+  | 'redshift';
 
 /** Canonical scan-mode registry. Runtime validation derives its allowlist here. */
 export const KTX_SCAN_MODES = ['structural', 'relationships', 'enriched'] as const;
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<string, SqlAnalysisDialect> = {
   postgres: 'postgres',
   postgresql: 'postgres',
+  redshift: 'redshift',
   bigquery: 'bigquery',
   snowflake: 'snowflake',
   mysql: 'mysql',
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..302b0f7eb
--- /dev/null
+++ b/packages/cli/test/connectors/redshift/connector.test.ts
@@ -0,0 +1,507 @@
+import { describe, expect, it, vi } from 'vitest';
+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<string, unknown>[];
+  fields?: Array<{ name: string; dataTypeID: number }>;
+}
+
+type FakeQueryResponse = FakeQueryResult | Error;
+
+function fakePoolFactory(results: Map<string, FakeQueryResponse>): 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<string, FakeQueryResponse> {
+  return new Map<string, FakeQueryResponse>([
+    [
+      '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 IN ($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 -c statement_timeout=30000',
+      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'));
+    // 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 () => {
+    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));
+  });
+});
diff --git a/packages/cli/test/context/connections/dialects.test.ts b/packages/cli/test/context/connections/dialects.test.ts
index fad193ed4..cd7b90482 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 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',
+    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: athena, bigquery, clickhouse, duckdb, mongodb, mysql, postgres, snowflake, sqlite, sqlserver',
+      'Unsupported driver "oracle". Supported drivers: athena, bigquery, clickhouse, duckdb, 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 69ac8464b..e139e03d8 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<KtxConnectionDriver, FixtureFactory> = {
     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' }),
   duckdb: (projectDir) => ({ driver: 'duckdb', path: join(projectDir, 'warehouse.duckdb') }),
   mongodb: () => ({
@@ -112,6 +121,7 @@ describe('driverRegistrations', () => {
       'mongodb',
       'mysql',
       'postgres',
+      'redshift',
       'snowflake',
       'sqlite',
       'sqlserver',
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');