Clickhouse Client

/**
 * Clickhouse warehouse client with HTTP/HTTPS protocol and named parameters
 *
 * IMPORTANT: Clickhouse uses database as the primary namespace (not catalog+schema).
 * In this implementation, the 'schema' credential field maps to Clickhouse's 'database'.
 * Named parameters use {paramName: Type} syntax in SQL queries.
 */
class ClickhouseWarehouseClient implements WarehouseClient {
  /** Clickhouse client instance (from '@clickhouse/client' package) */
  client: ClickHouseClient;

  /**
   * Initialize Clickhouse client with credentials
   * Uses HTTP or HTTPS protocol based on secure flag
   * @param credentials - Clickhouse credentials with host, port, database, and auth
   */
  constructor(credentials: CreateClickhouseCredentials);

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

  /**
   * Stream query results with named parameters and timezone support
   * Uses JSONCompactEachRowWithNamesAndTypes format for efficient streaming
   * @param sql - SQL query string with optional named parameters {paramName: Type}
   * @param streamCallback - Callback invoked for each batch of results
   * @param options - Query execution options with named parameters, tags, and timezone
   * @returns Promise that resolves when streaming completes
   * @note Clickhouse REQUIRES named parameters format: {paramName: Type} in SQL
   * @note Example: SELECT * FROM users WHERE id = {userId: UInt32}
   */
  streamQuery(
    sql: string,
    streamCallback: (data: WarehouseResults) => void,
    options: {
      queryParams?: Record<string, AnyType>;
      tags?: Record<string, string>;
      timezone?: string;
    }
  ): Promise<void>;

  /**
   * Execute query and return all results
   * Inherited from WarehouseBaseClient
   * @param sql - SQL query string to execute
   * @param tags - Optional tags for query tracking
   * @param timezone - Optional timezone for query execution
   * @param values - NOT SUPPORTED for Clickhouse (use queryParams instead)
   * @param queryParams - Named parameters for query (format: {paramName: value})
   * @returns Promise resolving to query results with fields and rows
   * @deprecated Use streamQuery() instead to avoid loading all results into memory
   */
  runQuery(
    sql: string,
    tags?: Record<string, string>,
    timezone?: string,
    values?: AnyType[],
    queryParams?: Record<string, AnyType>
  ): Promise<WarehouseResults>;

  /**
   * Get catalog metadata for specific tables
   * Queries system.columns table for column metadata
   * @param requests - Array of table identifiers (database in schema field, table)
   * @returns Promise resolving to nested catalog structure
   * @note In Clickhouse, 'schema' parameter maps to 'database'
   */
  getCatalog(
    requests: Array<{ database: string; schema?: string; table: string }>
  ): Promise<WarehouseCatalog>;

  /**
   * Get all tables in the configured database
   * Queries system.tables table
   * @returns Promise resolving to array of table metadata
   * @note WarehouseTables is { database: string; schema: string; table: string; partitionColumn?: PartitionColumn }[]
   */
  getAllTables(): Promise<WarehouseTables>;

  /**
   * Get tables in specific database
   * @param schema - Optional database name (uses credentials schema/database if omitted)
   * @param tags - Optional tags for query tracking
   * @returns Promise resolving to catalog with tables in database
   * @note In Clickhouse, 'schema' parameter maps to 'database'
   */
  getTables(
    schema?: string,
    tags?: Record<string, string>
  ): Promise<WarehouseCatalog>;

  /**
   * Get field metadata for specific table
   * Queries system.columns for column definitions
   * @param tableName - Name of the table
   * @param schema - Optional database name (uses credentials schema/database if omitted)
   * @param database - Always empty string for Clickhouse (uses schema parameter)
   * @param tags - Optional tags for query tracking
   * @returns Promise resolving to catalog with table's column definitions
   * @note In Clickhouse, 'schema' parameter maps to 'database'
   */
  getFields(
    tableName: string,
    schema?: string,
    database?: string,
    tags?: Record<string, string>
  ): Promise<WarehouseCatalog>;

  /**
   * Parse warehouse errors into WarehouseQueryError
   * Uses inherited base implementation from WarehouseBaseClient
   * @param error - Error object from Clickhouse
   * @returns WarehouseQueryError with enhanced error message
   */
  parseError(error: Error): Error;

  /**
   * Execute query asynchronously and optionally stream results
   * Inherited from WarehouseBaseClient
   * @param args - Async query execution arguments
   * @param resultsStreamCallback - Optional callback for streaming results
   * @returns Promise resolving to async query execution details
   * @note Clickhouse implementation uses streamQuery() internally (no native async support)
   */
  executeAsyncQuery(
    args: WarehouseExecuteAsyncQueryArgs,
    resultsStreamCallback?: (rows: WarehouseResults['rows'], fields: WarehouseResults['fields']) => void
  ): Promise<WarehouseExecuteAsyncQuery>;
}

import { warehouseClientFromCredentials } from '@lightdash/warehouses';
import { WarehouseTypes } from '@lightdash/common';

// Create Clickhouse client (only via factory function - not directly exported)
const client = warehouseClientFromCredentials({
  type: WarehouseTypes.CLICKHOUSE,
  host: 'clickhouse.example.com',
  port: 8123,
  user: 'analyst',
  password: 'secure-password',
  schema: 'analytics', // Maps to Clickhouse 'database'
  secure: true, // Use HTTPS
  timeoutSeconds: 30,
  startOfWeek: null,
});

// Test connection
await client.test();

// Stream query with named parameters and timezone
await client.streamQuery(
  `
    SELECT
      customer_id,
      SUM(order_amount) as total_spent,
      COUNT(*) as order_count
    FROM orders
    WHERE order_date >= {startDate: Date}
      AND customer_region = {region: String}
    GROUP BY customer_id
    ORDER BY total_spent DESC
    LIMIT 100
  `,
  (data) => {
    console.log('Received batch:', data.rows.length, 'rows');
    data.rows.forEach(row => {
      console.log(`Customer ${row.customer_id}: $${row.total_spent} (${row.order_count} orders)`);
    });
  },
  {
    queryParams: {
      startDate: '2023-01-01',
      region: 'US-WEST',
    },
    timezone: 'America/Los_Angeles',
    tags: { source: 'lightdash', report: 'top_customers' },
  }
);

// Get table metadata
const tables = await client.getAllTables();
console.log('Tables:', tables.map(t => t.table));

// Get column definitions
const schema = await client.getFields('orders', 'analytics');
console.log('Columns:', Object.keys(schema['']['analytics']['orders']));

/**
 * SQL builder for Clickhouse with quantile/median aggregations
 * Handles single-quote escaping and SQL comment removal
 */
class ClickhouseSqlBuilder implements WarehouseSqlBuilder {
  /** Warehouse type identifier */
  readonly type: WarehouseTypes.CLICKHOUSE;

  /**
   * Get the DBT adapter type for Clickhouse
   * @returns SupportedDbtAdapter.CLICKHOUSE
   */
  getAdapterType(): SupportedDbtAdapter;

  /**
   * Get the character used for string quoting
   * @returns Single quote (')
   */
  getStringQuoteChar(): string;

  /**
   * Get the character used for escaping string quotes
   * @returns Single quote (') - Clickhouse uses quote doubling
   */
  getEscapeStringQuoteChar(): string;

  /**
   * Get the character used for field/column quoting
   * @returns Double quote (") - inherited from base
   */
  getFieldQuoteChar(): string;

  /**
   * Get the floating point type for Clickhouse
   * @returns 'FLOAT' - inherited from base
   */
  getFloatingType(): string;

  /**
   * Get the start of week day
   * @returns WeekDay value or null/undefined
   */
  getStartOfWeek(): WeekDay | null | undefined;

  /**
   * Generate Clickhouse-specific metric SQL
   * Handles percentile and median aggregations
   * @param sql - Base SQL expression (column or calculation)
   * @param metric - Metric definition with type and parameters
   * @returns SQL string for the metric aggregation
   * @note Uses quantile(fraction)(expr) for percentiles
   * @note Uses median(expr) for median
   * @note Falls back to base implementation for other metric types
   */
  getMetricSql(sql: string, metric: Metric): string;

  /**
   * Concatenate strings using Clickhouse concat function
   * @param args - Variable number of string expressions
   * @returns SQL concat expression - inherited from base
   */
  concatString(...args: string[]): string;

  /**
   * Escape string value for safe SQL interpolation
   * Doubles single quotes, escapes backslashes, removes SQL comments and null bytes
   * @param value - String value to escape
   * @returns Escaped string safe for SQL
   */
  escapeString(value: string): string;
}

import { warehouseSqlBuilderFromType } from '@lightdash/warehouses';
import { SupportedDbtAdapter, MetricType, FieldType } from '@lightdash/common';

// Create Clickhouse SQL builder (only via factory - ClickhouseSqlBuilder not exported)
const builder = warehouseSqlBuilderFromType(SupportedDbtAdapter.CLICKHOUSE);

// Generate percentile SQL (90th percentile)
const p90Sql = builder.getMetricSql('response_time', {
  fieldType: FieldType.METRIC,
  type: MetricType.PERCENTILE,
  percentile: 90,
  sql: 'response_time',
  name: 'p90_response_time',
  label: 'P90 Response Time',
  table: 'requests',
  tableLabel: 'Requests',
  hidden: false,
});
// Result: quantile(0.9)(response_time)

// Generate median SQL
const medianSql = builder.getMetricSql('revenue', {
  fieldType: FieldType.METRIC,
  type: MetricType.MEDIAN,
  sql: 'revenue',
  name: 'median_revenue',
  label: 'Median Revenue',
  table: 'orders',
  tableLabel: 'Orders',
  hidden: false,
});
// Result: median(revenue)

// Escape string values
const escaped = builder.escapeString("O'Reilly");
// Result: O''Reilly

// Get field quote character
const fieldQuote = builder.getFieldQuoteChar();
// Result: "

/**
 * Clickhouse warehouse credentials
 * Note: CreateClickhouseCredentials is re-exported from @lightdash/common
 */
interface CreateClickhouseCredentials {
  /** Warehouse type discriminator */
  type: WarehouseTypes.CLICKHOUSE;
  /** Clickhouse server hostname */
  host: string;
  /** Clickhouse server port (default: 8123 for HTTP, 8443 for HTTPS) */
  port: number;
  /** Username for authentication */
  user: string;
  /** Password for authentication */
  password: string;
  /** Database name (referred to as 'schema' in this implementation) */
  schema: string;
  /** Use HTTPS instead of HTTP (optional, default: false) */
  secure?: boolean;
  /** Request timeout in seconds (optional, default: 30) */
  timeoutSeconds?: number;
  /** Week start day configuration (optional) */
  startOfWeek?: WeekDay | null;
  /** Require user credentials flag (optional) */
  requireUserCredentials?: boolean;
}

/**
 * Clickhouse native data types
 * Note: This enum is NOT directly exported from @lightdash/warehouses
 * Only accessible when using ClickhouseWarehouseClient via factory function
 */
enum ClickhouseTypes {
  UINT8 = 'UInt8',
  UINT16 = 'UInt16',
  UINT32 = 'UInt32',
  UINT64 = 'UInt64',
  INT8 = 'Int8',
  INT16 = 'Int16',
  INT32 = 'Int32',
  INT64 = 'Int64',
  FLOAT32 = 'Float32',
  FLOAT64 = 'Float64',
  DECIMAL = 'Decimal',
  DECIMAL32 = 'Decimal32',
  DECIMAL64 = 'Decimal64',
  DECIMAL128 = 'Decimal128',
  DECIMAL256 = 'Decimal256',
  BOOL = 'Bool',
  STRING = 'String',
  FIXEDSTRING = 'FixedString',
  UUID = 'UUID',
  DATE = 'Date',
  DATE32 = 'Date32',
  DATETIME = 'DateTime',
  DATETIME64 = 'DateTime64',
  ENUM8 = 'Enum8',
  ENUM16 = 'Enum16',
  ARRAY = 'Array',
  TUPLE = 'Tuple',
  MAP = 'Map',
  NULLABLE = 'Nullable',
  LOWCARDINALITY = 'LowCardinality',
  IPV4 = 'IPv4',
  IPV6 = 'IPv6',
}

{
  type: WarehouseTypes.CLICKHOUSE,
  host: 'clickhouse.example.com',
  port: 8123, // HTTP: 8123, HTTPS: 8443 (default Clickhouse ports)
  user: 'username',
  password: 'password',
  schema: 'default', // Clickhouse database name
  secure: true, // Use HTTPS (recommended for production)
  timeoutSeconds: 30, // Request timeout (default: 30 seconds)
}

// Correct named parameter syntax
const query = `
  SELECT * FROM users
  WHERE user_id = {userId: UInt32}
    AND email = {email: String}
    AND created_at >= {startDate: Date}
`;

await client.streamQuery(query, callback, {
  queryParams: {
    userId: 12345,
    email: 'user@example.com',
    startDate: '2023-01-01',
  },
});

// Clickhouse structure:
// database.table (no separate catalog)

// In this implementation:
// - credentials.schema maps to Clickhouse 'database'
// - database parameter is ignored (always empty string)
// - schema parameter in methods maps to Clickhouse 'database'

// Example:
await client.getFields(
  'orders',      // table name
  'analytics',   // schema (maps to Clickhouse database)
  '',            // database (ignored)
);

// ❌ WRONG - ClickhouseWarehouseClient is NOT exported from index.ts
import { ClickhouseWarehouseClient } from '@lightdash/warehouses';

// ✅ CORRECT - Use factory function
import { warehouseClientFromCredentials } from '@lightdash/warehouses';
import { WarehouseTypes } from '@lightdash/common';

const client = warehouseClientFromCredentials({
  type: WarehouseTypes.CLICKHOUSE,
  // ... credentials
});

// ❌ WRONG - Missing type annotation
const query = `SELECT * FROM users WHERE id = {userId}`;

// ❌ WRONG - Positional parameters not supported
const query = `SELECT * FROM users WHERE id = $1`;

// ✅ CORRECT - Named parameter with type
const query = `SELECT * FROM users WHERE id = {userId: UInt32}`;

// ❌ WRONG - Treating schema as namespace
const client = warehouseClientFromCredentials({
  type: WarehouseTypes.CLICKHOUSE,
  schema: 'public', // This maps to Clickhouse database 'public'
});

// ✅ CORRECT - Use actual Clickhouse database name
const client = warehouseClientFromCredentials({
  type: WarehouseTypes.CLICKHOUSE,
  schema: 'analytics', // Clickhouse database name
});

// ❌ WRONG - Insecure connection in production
const client = warehouseClientFromCredentials({
  type: WarehouseTypes.CLICKHOUSE,
  secure: false, // Uses HTTP
  port: 8123,
  // ...
});

// ✅ CORRECT - Use HTTPS in production
const client = warehouseClientFromCredentials({
  type: WarehouseTypes.CLICKHOUSE,
  secure: true, // Uses HTTPS
  port: 8443, // Default HTTPS port for Clickhouse
  // ...
});

// Clickhouse column type: Nullable(String)
// Mapped to DimensionType: STRING

// Clickhouse column type: LowCardinality(String)
// Mapped to DimensionType: STRING

// Clickhouse column type: Nullable(LowCardinality(String))
// Mapped to DimensionType: STRING

// Get table information
const tables = await client.runQuery(`
  SELECT database, name, engine, total_rows, total_bytes
  FROM system.tables
  WHERE database = {db: String}
`, {}, undefined, undefined, { db: 'analytics' });

// Get column information
const columns = await client.runQuery(`
  SELECT database, table, name, type, position
  FROM system.columns
  WHERE database = {db: String}
    AND table = {tbl: String}
  ORDER BY position
`, {}, undefined, undefined, { db: 'analytics', tbl: 'orders' });

await client.streamQuery(
  'SELECT now() as current_time, toDateTime(order_timestamp) as order_time FROM orders',
  (data) => console.log(data.rows),
  {
    timezone: 'America/New_York', // Results converted to this timezone
  }
);