Redshift Client

/**
 * Amazon Redshift warehouse client with SSH tunnel support
 * Built on PostgreSQL wire protocol with Redshift-specific SSL configuration
 * Extends PostgresClient to inherit all PostgreSQL functionality
 */
class RedshiftWarehouseClient extends PostgresClient<CreateRedshiftCredentials> {
  /**
   * Initialize Redshift client with credentials
   * Configures SSL with AWS Redshift CA certificates and overrides SQL builder with RedshiftSqlBuilder
   * @param credentials - Redshift credentials with host, port, database, and SSL config
   */
  constructor(credentials: CreateRedshiftCredentials);

  /**
   * Test the database connection
   * Executes a simple SELECT 1 query to verify connectivity and credentials
   * @returns Promise that resolves if connection is successful
   * @throws {WarehouseConnectionError} If connection fails
   */
  test(): Promise<void>;

  /**
   * Stream query results using server-side cursors
   * Executes query with DECLARE CURSOR and fetches in batches
   * @param sql - SQL query string with optional $1, $2 parameter placeholders for positional parameters
   * @param streamCallback - Callback invoked for each batch of results
   * @param options - Query options with parameter values, tags, and timezone
   * @returns Promise resolving when all results are streamed
   */
  streamQuery(
    sql: string,
    streamCallback: (data: WarehouseResults) => void,
    options: {
      values?: AnyType[];
      tags?: Record<string, string>;
      timezone?: string;
    }
  ): Promise<void>;

  /**
   * Get all BASE TABLEs in the database
   * Queries information_schema.tables for BASE TABLE type only
   * Note: Materialized views are only included in getCatalog(), not getAllTables()
   * Inherited from WarehouseClient interface - optional parameters are accepted but ignored by Redshift implementation
   * @param schema - Optional schema name (ignored by Redshift - returns all schemas)
   * @param tags - Optional tags for query tracking (ignored by Redshift)
   * @returns Promise resolving to array of table metadata
   */
  getAllTables(
    schema?: string,
    tags?: Record<string, string>
  ): Promise<Array<{ database: string; schema: string; table: string }>>;

  /**
   * Get catalog metadata for specific tables
   * Queries information_schema for columns, data types, and constraints
   * Inherited from PostgresClient
   * @param requests - Array of table identifiers (database, schema, table)
   * @returns Promise resolving to nested catalog structure
   */
  getCatalog(
    requests: Array<{ database: string; schema: string; table: string }>
  ): Promise<WarehouseCatalog>;

  /**
   * Get field metadata for specific table
   * Inherited from PostgresClient
   * @param tableName - Name of the table
   * @param schema - Optional schema name (uses credentials schema if omitted)
   * @param database - Optional database name (catalog)
   * @param tags - Optional tags for query tracking
   * @returns Promise resolving to catalog with field information
   */
  getFields(
    tableName: string,
    schema?: string,
    database?: string,
    tags?: Record<string, string>
  ): Promise<WarehouseCatalog>;

  /**
   * Run query and return all results at once
   * @deprecated Use streamQuery() instead to avoid loading all results into memory
   * Inherited from WarehouseBaseClient via PostgresClient
   * @param sql - SQL query string
   * @param tags - Optional tags for query tracking
   * @param timezone - Optional timezone for query execution
   * @param values - Optional positional parameter values
   * @param queryParams - Optional named query parameters
   * @returns Promise resolving to complete query results
   */
  runQuery(
    sql: string,
    tags?: Record<string, string>,
    timezone?: string,
    values?: AnyType[],
    queryParams?: Record<string, AnyType>
  ): Promise<WarehouseResults>;

  /**
   * Execute async query using streaming fallback
   * Inherited from WarehouseBaseClient via PostgresClient
   * @param args - Async query arguments (sql, tags, timezone, values)
   * @param resultsStreamCallback - Callback for streaming results (REQUIRED)
   * @returns Promise resolving to query metadata (queryId will be null)
   */
  executeAsyncQuery(
    args: WarehouseExecuteAsyncQueryArgs,
    resultsStreamCallback: (
      rows: WarehouseResults["rows"],
      fields: WarehouseResults["fields"]
    ) => void
  ): Promise<WarehouseExecuteAsyncQuery>;

  /**
   * Parse PostgreSQL/Redshift database errors into WarehouseQueryError
   * Inherited from PostgresClient
   * @param error - PostgreSQL/Redshift error from pg driver
   * @param query - Optional SQL query for enhanced error messages
   * @returns Enhanced error with line numbers and context
   */
  parseError(error: pg.DatabaseError, query?: string): WarehouseQueryError;

  /**
   * Parse warehouse catalog from raw query results
   * Inherited from WarehouseBaseClient via PostgresClient
   * @param rows - Raw rows from catalog query
   * @param mapFieldType - Function to map warehouse-specific types to DimensionType
   * @returns Parsed warehouse catalog structure
   */
  parseWarehouseCatalog(
    rows: Record<string, AnyType>[],
    mapFieldType: (type: string) => DimensionType
  ): WarehouseCatalog;
}

import { RedshiftWarehouseClient } from "@lightdash/warehouses";

// Basic Redshift connection with SSL
const client = new RedshiftWarehouseClient({
  type: "redshift",
  host: "my-cluster.abc123.us-east-1.redshift.amazonaws.com",
  port: 5439,
  user: "admin",
  password: "secure-password",
  dbname: "analytics",
  schema: "public",
  sslmode: "require",
  startOfWeek: null,
});

// With SSH tunnel for private Redshift cluster
const tunnelClient = new RedshiftWarehouseClient({
  type: "redshift",
  host: "internal-redshift.local",
  port: 5439,
  user: "dbuser",
  password: "dbpass",
  dbname: "warehouse",
  schema: "reporting",
  sslmode: "disable",
  startOfWeek: null,
  useSshTunnel: true,
  sshTunnelHost: "bastion.example.com",
  sshTunnelPort: 22,
  sshTunnelUser: "sshuser",
  sshTunnelPrivateKey: "-----BEGIN RSA PRIVATE KEY-----\n...",
});

// Test connection
await client.test();

// Stream query results
await client.streamQuery(
  `
    SELECT
      product_category,
      SUM(sales_amount) as total_sales,
      COUNT(DISTINCT customer_id) as unique_customers
    FROM sales
    WHERE sale_date >= $1
    GROUP BY product_category
    ORDER BY total_sales DESC
  `,
  (data) => {
    console.log("Received batch:", data.rows.length, "rows");
    data.rows.forEach((row) => {
      console.log(
        `${row.product_category}: $${row.total_sales} (${row.unique_customers} customers)`
      );
    });
  },
  {
    values: ["2023-01-01"],
    tags: { source: "lightdash", report: "sales_summary" },
  }
);

// Run query with positional parameters ($1, $2, etc.)
const results = await client.runQuery(
  `
    SELECT COUNT(*) as total
    FROM events
    WHERE event_type = $1
    AND created_at BETWEEN $2 AND $3
  `,
  { source: "analytics" },
  "UTC",
  ["purchase", "2023-01-01", "2023-12-31"]
);
console.log("Total events:", results.rows[0].total);

// Get all tables (filters out system schemas automatically)
const tables = await client.getAllTables();
console.log("Tables:", tables.map((t) => `${t.schema}.${t.table}`).join(", "));

// Get table fields
const catalog = await client.getFields("customer_orders", "analytics");
const database = Object.keys(catalog)[0];
const columns = catalog[database]["analytics"]["customer_orders"];
Object.entries(columns).forEach(([col, type]) => {
  console.log(`${col}: ${type}`);
});

/**
 * Redshift SQL builder (inherits PostgreSQL syntax)
 * Uses PostgreSQL-compatible SQL with Redshift warehouse type identifier
 */
class RedshiftSqlBuilder extends PostgresSqlBuilder {
  /** Warehouse type identifier */
  type = WarehouseTypes.REDSHIFT;

  /**
   * Get DBT adapter type for Redshift
   * @returns SupportedDbtAdapter.REDSHIFT
   */
  getAdapterType(): SupportedDbtAdapter;
}

import { RedshiftSqlBuilder } from "@lightdash/warehouses";

const builder = new RedshiftSqlBuilder();

// Get adapter type
const adapter = builder.getAdapterType(); // 'redshift'

// Get quote characters (inherited from PostgreSQL)
const fieldQuote = builder.getFieldQuoteChar(); // '"'
const fieldName = `${fieldQuote}customer_id${fieldQuote}`; // "customer_id"

// Escape string with single quotes (inherited from PostgreSQL)
const escaped = builder.escapeString("Company's Name");
// Result: 'Company''s Name'

// Generate concatenation SQL (inherited from PostgreSQL)
const concatSql = builder.concatString('"first_name"', "' '", '"last_name"');
// Result: "first_name" || ' ' || "last_name"

// Generate metric SQL (inherited from PostgreSQL)
const medianSql = builder.getMetricSql("revenue", {
  type: MetricType.MEDIAN,
  sql: "revenue",
});
// Result: 'PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY revenue)'

const avgSql = builder.getMetricSql("amount", {
  type: MetricType.AVERAGE,
  sql: "amount",
});
// Result: 'AVG(amount)'

/**
 * Redshift warehouse credentials
 * Identical structure to PostgreSQL credentials with different type discriminator
 */
interface CreateRedshiftCredentials extends CreatePostgresLikeCredentials {
  /** Warehouse type discriminator */
  type: "redshift";
  /** Redshift cluster endpoint hostname */
  host: string;
  /** Redshift port (default: 5439) */
  port: number;
  /** Database username */
  user: string;
  /** Database password */
  password: string;
  /** Database name */
  dbname: string;
  /** Default schema for queries */
  schema: string;
  /** SSL mode (optional, defaults to 'prefer'): disable, no-verify, allow, prefer, require, verify-ca, verify-full */
  sslmode?: string;
  /** Enable SSH tunnel (optional) */
  useSshTunnel?: boolean;
  /** SSH tunnel hostname (optional) */
  sshTunnelHost?: string;
  /** SSH tunnel port (default: 22) (optional) */
  sshTunnelPort?: number;
  /** SSH tunnel username (optional) */
  sshTunnelUser?: string;
  /** SSH tunnel public key (if using key-based auth) (optional) */
  sshTunnelPublicKey?: string;
  /** SSH tunnel private key (PEM format, required if useSshTunnel is true) (optional) */
  sshTunnelPrivateKey?: string;
  /** Week start day configuration (optional) */
  startOfWeek?: WeekDay | null;
  /** Require user credentials for authentication (optional) */
  requireUserCredentials?: boolean;
  /** Number of threads for parallel operations (optional) */
  threads?: number;
  /** TCP keepalive idle interval in seconds (optional) */
  keepalivesIdle?: number;
  /** Enable RA3 node optimizations (optional) */
  ra3Node?: boolean;
  /** Query timeout in seconds (optional) */
  timeoutSeconds?: number;
}

/**
 * Union type for PostgreSQL-like credentials
 * Includes both PostgreSQL and Redshift credentials
 */
type CreatePostgresLikeCredentials = CreateRedshiftCredentials | CreatePostgresCredentials;