diff --git a/packages/cli/package.json b/packages/cli/package.json
index a5f9e5810..ab0def542 100644
--- a/packages/cli/package.json
+++ b/packages/cli/package.json
@@ -82,6 +82,7 @@
     "semver": "^7.8.1",
     "simple-git": "3.36.0",
     "snowflake-sdk": "^2.4.2",
+    "trino-client": "^0.2.9",
     "yaml": "^2.9.0",
     "zod": "^4.4.3"
   },
diff --git a/packages/cli/src/connection-drivers.ts b/packages/cli/src/connection-drivers.ts
index 4f10e663c..193497669 100644
--- a/packages/cli/src/connection-drivers.ts
+++ b/packages/cli/src/connection-drivers.ts
@@ -8,6 +8,7 @@ const KTX_DATABASE_DRIVER_IDS = new Set([
   'sqlserver',
   'bigquery',
   'snowflake',
+  'trino',
 ]);
 
 export function normalizeConnectionDriver(connection: KtxProjectConnectionConfig): string {
diff --git a/packages/cli/src/connectors/trino/connector.ts b/packages/cli/src/connectors/trino/connector.ts
new file mode 100644
index 000000000..084b25367
--- /dev/null
+++ b/packages/cli/src/connectors/trino/connector.ts
@@ -0,0 +1,504 @@
+import { getDialectForDriver } from '../../context/connections/dialects.js';
+import { assertReadOnlySql, limitSqlForExecution } from '../../context/connections/read-only-sql.js';
+import {
+  connectorTestFailure,
+  createKtxConnectorCapabilities,
+  type KtxColumnSampleInput,
+  type KtxColumnSampleResult,
+  type KtxColumnStatsInput,
+  type KtxColumnStatsResult,
+  type KtxConnectorTestResult,
+  type KtxQueryResult,
+  type KtxReadOnlyQueryInput,
+  type KtxScanConnector,
+  type KtxScanContext,
+  type KtxScanInput,
+  type KtxSchemaColumn,
+  type KtxSchemaSnapshot,
+  type KtxSchemaTable,
+  type KtxTableListEntry,
+  type KtxTableRef,
+  type KtxTableSampleInput,
+  type KtxTableSampleResult,
+} from '../../context/scan/types.js';
+import { scopedTableNames } from '../../context/scan/table-ref.js';
+import { resolveStringReference } from '../shared/string-reference.js';
+
+/**
+ * Catalogs Trino always exposes that hold no user data. Excluded from "all
+ * catalogs" discovery by default; an explicit `catalogs` allowlist overrides
+ * this.
+ */
+const SYSTEM_CATALOGS = new Set(['system', 'jmx', 'tpch', 'tpcds']);
+
+export interface KtxTrinoConnectionConfig {
+  driver?: string;
+  /** e.g. `https://trino.example.com:8443` — overrides host/port/ssl when set. */
+  url?: string;
+  host?: string;
+  port?: number;
+  ssl?: boolean;
+  user?: string;
+  username?: string;
+  password?: string;
+  /** Default catalog for unqualified queries; optional for discovery. */
+  catalog?: string;
+  /** Default schema for unqualified queries. */
+  schema?: string;
+  /**
+   * Restricts discovery to these catalogs. When omitted, the connector
+   * enumerates every catalog from `SHOW CATALOGS` minus {@link SYSTEM_CATALOGS}.
+   */
+  catalogs?: string[];
+  /** Optional allowlist of schemas (applied within every discovered catalog). */
+  schemas?: string[];
+  [key: string]: unknown;
+}
+
+export interface KtxTrinoResolvedClientConfig {
+  server: string;
+  user: string;
+  password?: string;
+  catalog?: string;
+  schema?: string;
+  ssl: boolean;
+}
+
+/** Normalized result returned by {@link KtxTrinoClient.query}. */
+export interface KtxTrinoQueryResult {
+  columns: Array<{ name: string; type: string }>;
+  rows: unknown[][];
+}
+
+/**
+ * Minimal client surface the connector depends on. The default factory drives
+ * the `trino-client` async statement iterator; tests inject a fake.
+ */
+export interface KtxTrinoClient {
+  query(sql: string): Promise<KtxTrinoQueryResult>;
+  close(): Promise<void>;
+}
+
+export interface KtxTrinoClientFactory {
+  createClient(config: KtxTrinoResolvedClientConfig): KtxTrinoClient;
+}
+
+export interface KtxTrinoScanConnectorOptions {
+  connectionId: string;
+  connection: KtxTrinoConnectionConfig | undefined;
+  clientFactory?: KtxTrinoClientFactory;
+  env?: NodeJS.ProcessEnv;
+  now?: () => Date;
+}
+
+interface TrinoTableRow {
+  table_catalog: string;
+  table_schema: string;
+  table_name: string;
+  table_type: string;
+}
+
+interface TrinoColumnRow {
+  table_catalog: string;
+  table_schema: string;
+  table_name: string;
+  column_name: string;
+  data_type: string;
+  is_nullable: string;
+  comment: string | null;
+}
+
+/**
+ * Default factory backed by `trino-client`. Imported lazily so the dependency
+ * is only required when a Trino connection is actually used.
+ */
+class DefaultTrinoClientFactory implements KtxTrinoClientFactory {
+  createClient(config: KtxTrinoResolvedClientConfig): KtxTrinoClient {
+    let trinoPromise: Promise<unknown> | null = null;
+    const getTrino = async (): Promise<{ query(sql: string): Promise<AsyncIterable<unknown>> }> => {
+      if (!trinoPromise) {
+        trinoPromise = (async () => {
+          const mod = (await import('trino-client')) as {
+            Trino: { create(options: unknown): unknown };
+            BasicAuth: new (user: string, password?: string) => unknown;
+          };
+          // `ConnectionOptions` has no `user` field — the username travels on
+          // `BasicAuth` (password is optional, so a no-password connection still
+          // sets the Trino user header).
+          return mod.Trino.create({
+            server: config.server,
+            catalog: config.catalog,
+            schema: config.schema,
+            auth: new mod.BasicAuth(config.user, config.password),
+          });
+        })();
+      }
+      return trinoPromise as Promise<{ query(sql: string): Promise<AsyncIterable<unknown>> }>;
+    };
+
+    return {
+      async query(sql: string): Promise<KtxTrinoQueryResult> {
+        const trino = await getTrino();
+        const iterator = await trino.query(sql);
+        const columns: Array<{ name: string; type: string }> = [];
+        const rows: unknown[][] = [];
+        for await (const page of iterator as AsyncIterable<{
+          columns?: Array<{ name: string; type: string }>;
+          data?: unknown[][];
+          error?: { message?: string };
+        }>) {
+          if (page.error) {
+            throw new Error(page.error.message ?? 'Trino query failed');
+          }
+          if (page.columns && columns.length === 0) {
+            columns.push(...page.columns.map((column) => ({ name: column.name, type: column.type })));
+          }
+          if (page.data) {
+            rows.push(...page.data);
+          }
+        }
+        return { columns, rows };
+      },
+      async close(): Promise<void> {
+        // trino-client is stateless over HTTP; nothing to tear down.
+      },
+    };
+  }
+}
+
+function stringConfigValue(
+  connection: KtxTrinoConnectionConfig | undefined,
+  key: keyof KtxTrinoConnectionConfig,
+  env: NodeJS.ProcessEnv,
+): string | undefined {
+  const value = connection?.[key];
+  return typeof value === 'string' && value.trim().length > 0 ? resolveStringReference(value.trim(), env) : undefined;
+}
+
+function maybeNumber(value: unknown): number | undefined {
+  return typeof value === 'number' && Number.isFinite(value) ? value : undefined;
+}
+
+export function isKtxTrinoConnectionConfig(
+  connection: KtxTrinoConnectionConfig | undefined,
+): connection is KtxTrinoConnectionConfig {
+  return String(connection?.driver ?? '').toLowerCase() === 'trino';
+}
+
+/** @internal */
+export function trinoClientConfigFromConfig(input: {
+  connectionId: string;
+  connection: KtxTrinoConnectionConfig | undefined;
+  env?: NodeJS.ProcessEnv;
+}): KtxTrinoResolvedClientConfig {
+  const inputDriver = input.connection?.driver ?? 'unknown';
+  if (!isKtxTrinoConnectionConfig(input.connection)) {
+    throw new Error(`Trino connector cannot run driver "${inputDriver}"`);
+  }
+
+  const env = input.env ?? process.env;
+  const referencedUrl = stringConfigValue(input.connection, 'url', env);
+  let server = referencedUrl;
+  let ssl = input.connection.ssl === true;
+  if (!server) {
+    const host = stringConfigValue(input.connection, 'host', env);
+    if (!host) {
+      throw new Error(`Trino connector requires connections.${input.connectionId}.url or host`);
+    }
+    const port = maybeNumber(input.connection.port) ?? (ssl ? 8443 : 8080);
+    server = `${ssl ? 'https' : 'http'}://${host}:${port}`;
+  } else {
+    ssl = server.startsWith('https://');
+  }
+
+  const user = stringConfigValue(input.connection, 'user', env) ?? stringConfigValue(input.connection, 'username', env);
+  if (!user) {
+    throw new Error(`Trino connector requires connections.${input.connectionId}.user`);
+  }
+
+  return {
+    server,
+    user,
+    password: stringConfigValue(input.connection, 'password', env),
+    catalog: stringConfigValue(input.connection, 'catalog', env),
+    schema: stringConfigValue(input.connection, 'schema', env),
+    ssl,
+  };
+}
+
+function trinoTableKey(catalog: string, schema: string, table: string): string {
+  return `${catalog}.${schema}.${table}`;
+}
+
+export class KtxTrinoScanConnector implements KtxScanConnector {
+  readonly id: string;
+  readonly driver = 'trino' as const;
+  readonly capabilities = createKtxConnectorCapabilities({
+    tableSampling: true,
+    columnSampling: true,
+    columnStats: false,
+    readOnlySql: true,
+    nestedAnalysis: false,
+    formalForeignKeys: false,
+    estimatedRowCounts: false,
+  });
+
+  private readonly connectionId: string;
+  private readonly connection: KtxTrinoConnectionConfig;
+  private readonly clientConfig: KtxTrinoResolvedClientConfig;
+  private readonly clientFactory: KtxTrinoClientFactory;
+  private readonly now: () => Date;
+  private readonly dialect = getDialectForDriver('trino');
+  private client: KtxTrinoClient | null = null;
+
+  constructor(options: KtxTrinoScanConnectorOptions) {
+    this.connectionId = options.connectionId;
+    this.connection = options.connection ?? {};
+    this.clientConfig = trinoClientConfigFromConfig({
+      connectionId: options.connectionId,
+      connection: options.connection,
+      env: options.env,
+    });
+    this.clientFactory = options.clientFactory ?? new DefaultTrinoClientFactory();
+    this.now = options.now ?? (() => new Date());
+    this.id = `trino:${options.connectionId}`;
+  }
+
+  async testConnection(): Promise<KtxConnectorTestResult> {
+    try {
+      await this.query('SELECT 1');
+      return { success: true };
+    } catch (error) {
+      return connectorTestFailure(error);
+    }
+  }
+
+  /** Resolve which catalogs to introspect: explicit allowlist, else all non-system catalogs. */
+  private async resolveCatalogs(): Promise<string[]> {
+    const configured = (this.connection.catalogs ?? [])
+      .filter((catalog): catalog is string => typeof catalog === 'string' && catalog.trim().length > 0)
+      .map((catalog) => catalog.trim());
+    if (configured.length > 0) {
+      return [...new Set(configured)];
+    }
+    // `SHOW CATALOGS` is not a SELECT, so it would be rejected by the read-only
+    // gate; Trino exposes the same data as a queryable system table.
+    const rows = await this.queryRows<{ catalog_name: string }>(
+      'SELECT catalog_name FROM system.metadata.catalogs ORDER BY catalog_name',
+    );
+    return rows
+      .map((row) => String(row.catalog_name))
+      .filter((catalog) => !SYSTEM_CATALOGS.has(catalog.toLowerCase()));
+  }
+
+  async introspect(input: KtxScanInput, _ctx: KtxScanContext): Promise<KtxSchemaSnapshot> {
+    this.assertConnection(input.connectionId);
+    const catalogs = await this.resolveCatalogs();
+    const schemaTables: KtxSchemaTable[] = [];
+
+    for (const catalog of catalogs) {
+      const quotedCatalog = this.dialect.quoteIdentifier(catalog);
+      const schemaFilter =
+        this.connection.schemas && this.connection.schemas.length > 0
+          ? ` AND table_schema IN (${this.connection.schemas.map((schema) => this.quoteLiteral(schema)).join(', ')})`
+          : " AND table_schema <> 'information_schema'";
+
+      const tables = await this.queryRows<TrinoTableRow>(
+        `SELECT table_catalog, table_schema, table_name, table_type
+         FROM ${quotedCatalog}.information_schema.tables
+         WHERE 1 = 1${schemaFilter}
+         ORDER BY table_schema, table_name`,
+      );
+      if (tables.length === 0) {
+        continue;
+      }
+
+      const columns = await this.queryRows<TrinoColumnRow>(
+        `SELECT table_catalog, table_schema, table_name, column_name, data_type, is_nullable, comment
+         FROM ${quotedCatalog}.information_schema.columns
+         WHERE 1 = 1${schemaFilter}
+         ORDER BY table_schema, table_name, ordinal_position`,
+      );
+      const columnsByTable = new Map<string, TrinoColumnRow[]>();
+      for (const column of columns) {
+        const key = trinoTableKey(column.table_catalog, column.table_schema, column.table_name);
+        columnsByTable.set(key, [...(columnsByTable.get(key) ?? []), column]);
+      }
+
+      // Honor tableScope by post-filtering against the (catalog, schema) namespaces.
+      for (const table of tables) {
+        if (input.tableScope) {
+          const allowed = scopedTableNames(input.tableScope, { catalog, db: table.table_schema });
+          if (!allowed.includes(table.table_name)) {
+            continue;
+          }
+        }
+        const key = trinoTableKey(table.table_catalog, table.table_schema, table.table_name);
+        schemaTables.push(this.toSchemaTable(table, columnsByTable.get(key) ?? []));
+      }
+    }
+
+    return {
+      connectionId: this.connectionId,
+      driver: 'trino',
+      extractedAt: this.now().toISOString(),
+      scope: { catalogs },
+      metadata: {
+        server: this.clientConfig.server,
+        catalogs,
+        table_count: schemaTables.length,
+        total_columns: schemaTables.reduce((sum, table) => sum + table.columns.length, 0),
+      },
+      tables: schemaTables,
+    };
+  }
+
+  async sampleTable(input: KtxTableSampleInput, _ctx: KtxScanContext): Promise<KtxTableSampleResult> {
+    this.assertConnection(input.connectionId);
+    const result = await this.query(
+      this.dialect.generateSampleQuery(this.qTableName(input.table), input.limit, input.columns),
+    );
+    return { headers: result.headers, 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> {
+    return null;
+  }
+
+  async executeReadOnly(input: KtxReadOnlyQueryInput, _ctx: KtxScanContext): Promise<KtxQueryResult> {
+    this.assertConnection(input.connectionId);
+    const limitedSql = limitSqlForExecution(assertReadOnlySql(input.sql), input.maxRows);
+    const result = await this.query(limitedSql);
+    return { ...result, rowCount: result.rows.length };
+  }
+
+  async listSchemas(): Promise<string[]> {
+    const catalogs = await this.resolveCatalogs();
+    const schemas = new Set<string>();
+    for (const catalog of catalogs) {
+      const quotedCatalog = this.dialect.quoteIdentifier(catalog);
+      const rows = await this.queryRows<{ schema_name: string }>(
+        `SELECT schema_name FROM ${quotedCatalog}.information_schema.schemata
+         WHERE schema_name <> 'information_schema'
+         ORDER BY schema_name`,
+      );
+      // Schemas are namespaced by catalog to stay unambiguous across catalogs.
+      for (const row of rows) {
+        schemas.add(`${catalog}.${row.schema_name}`);
+      }
+    }
+    return [...schemas];
+  }
+
+  async listTables(schemas?: string[]): Promise<KtxTableListEntry[]> {
+    const catalogs = await this.resolveCatalogs();
+    const wanted = schemas ? new Set(schemas) : null;
+    const entries: KtxTableListEntry[] = [];
+    for (const catalog of catalogs) {
+      const quotedCatalog = this.dialect.quoteIdentifier(catalog);
+      const rows = await this.queryRows<TrinoTableRow>(
+        `SELECT table_catalog, table_schema, table_name, table_type
+         FROM ${quotedCatalog}.information_schema.tables
+         WHERE table_schema <> 'information_schema'
+         ORDER BY table_schema, table_name`,
+      );
+      for (const row of rows) {
+        if (wanted && !wanted.has(`${catalog}.${row.table_schema}`) && !wanted.has(row.table_schema)) {
+          continue;
+        }
+        entries.push({
+          catalog: row.table_catalog,
+          schema: row.table_schema,
+          name: row.table_name,
+          kind: row.table_type.toUpperCase().includes('VIEW') ? 'view' : 'table',
+        });
+      }
+    }
+    return entries;
+  }
+
+  async cleanup(): Promise<void> {
+    if (this.client) {
+      await this.client.close();
+      this.client = null;
+    }
+  }
+
+  qTableName(table: Pick<KtxTableRef, 'name'> & Partial<Pick<KtxTableRef, 'catalog' | 'db'>>): string {
+    return this.dialect.formatTableName(table);
+  }
+
+  private toSchemaTable(table: TrinoTableRow, columns: TrinoColumnRow[]): KtxSchemaTable {
+    const kind = table.table_type.toUpperCase().includes('VIEW') ? 'view' : 'table';
+    return {
+      catalog: table.table_catalog,
+      db: table.table_schema,
+      name: table.table_name,
+      kind,
+      comment: null,
+      estimatedRows: null,
+      columns: columns.map((column) => this.toSchemaColumn(column)),
+      foreignKeys: [],
+    };
+  }
+
+  private toSchemaColumn(column: TrinoColumnRow): 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: String(column.is_nullable).toUpperCase() !== 'NO',
+      primaryKey: false,
+      comment: column.comment ?? null,
+    };
+  }
+
+  private quoteLiteral(value: string): string {
+    return `'${value.replace(/'/g, "''")}'`;
+  }
+
+  private async clientForQuery(): Promise<KtxTrinoClient> {
+    if (!this.client) {
+      this.client = this.clientFactory.createClient(this.clientConfig);
+    }
+    return this.client;
+  }
+
+  private async queryRows<T>(sql: string): Promise<T[]> {
+    const result = await this.query(sql);
+    return result.rows.map((row) => {
+      const record: Record<string, unknown> = {};
+      result.headers.forEach((header, index) => {
+        record[header] = row[index];
+      });
+      return record as T;
+    });
+  }
+
+  private async query(sql: string): Promise<Omit<KtxQueryResult, 'rowCount'>> {
+    const client = await this.clientForQuery();
+    const result = await client.query(assertReadOnlySql(sql));
+    return {
+      headers: result.columns.map((column) => column.name),
+      headerTypes: result.columns.map((column) => column.type),
+      rows: result.rows,
+      totalRows: result.rows.length,
+    };
+  }
+
+  private assertConnection(connectionId: string): void {
+    if (connectionId !== this.connectionId) {
+      throw new Error(`ktx Trino connector ${this.id} cannot serve connection ${connectionId}`);
+    }
+  }
+}
diff --git a/packages/cli/src/connectors/trino/dialect.ts b/packages/cli/src/connectors/trino/dialect.ts
new file mode 100644
index 000000000..158d774c0
--- /dev/null
+++ b/packages/cli/src/connectors/trino/dialect.ts
@@ -0,0 +1,182 @@
+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 TrinoTableNameRef = Pick<KtxTableRef, 'name'> & Partial<Pick<KtxTableRef, 'catalog' | 'db'>>;
+
+/**
+ * Trino is fully qualified as `catalog.schema.table`, so it uses the
+ * `three-part` identifier shape (unlike Snowflake/BigQuery which surface as
+ * two-part display refs). Identifiers are ANSI double-quoted; SQL is ANSI
+ * standard with `TABLESAMPLE BERNOULLI` for sampling.
+ *
+ * @internal
+ */
+export class KtxTrinoDialect implements KtxDialect {
+  readonly type = 'trino' as const;
+
+  private readonly typeMappings: Record<string, KtxSchemaDimensionType> = {
+    boolean: 'boolean',
+    tinyint: 'number',
+    smallint: 'number',
+    integer: 'number',
+    int: 'number',
+    bigint: 'number',
+    real: 'number',
+    double: 'number',
+    decimal: 'number',
+    varchar: 'string',
+    char: 'string',
+    varbinary: 'string',
+    json: 'string',
+    uuid: 'string',
+    ipaddress: 'string',
+    date: 'time',
+    time: 'time',
+    timestamp: 'time',
+  };
+
+  quoteIdentifier(identifier: string): string {
+    return `"${identifier.replace(/"/g, '""')}"`;
+  }
+
+  formatTableName(table: TrinoTableNameRef): string {
+    return formatDialectTableName(table, this.quoteIdentifier.bind(this), 'three-part');
+  }
+
+  formatDisplayRef(table: TrinoTableNameRef): string {
+    return formatDialectDisplayRef(table, 'three-part');
+  }
+
+  parseDisplayRef(display: string): KtxTableRef | null {
+    return parseDialectDisplayRef(display, 'three-part');
+  }
+
+  columnDisplayTablePartCount(): 1 | 2 | 3 {
+    return columnDisplayPartCount('three-part');
+  }
+
+  mapDataType(nativeType: string): string {
+    return nativeType;
+  }
+
+  mapToDimensionType(nativeType: string): KtxSchemaDimensionType {
+    if (!nativeType) {
+      return 'string';
+    }
+    // Strip parameters/array/row wrappers: `decimal(10,2)` -> `decimal`,
+    // `array(varchar)` -> `array`, `timestamp(3) with time zone` -> `timestamp`.
+    const base = nativeType.toLowerCase().trim().split('(')[0]?.split(' ')[0] ?? '';
+    if (this.typeMappings[base]) {
+      return this.typeMappings[base];
+    }
+    if (base.includes('timestamp') || base.includes('date') || base.includes('time')) {
+      return 'time';
+    }
+    if (base.includes('int') || base.includes('decimal') || base === 'double' || base === 'real') {
+      return 'number';
+    }
+    if (base === 'boolean') {
+      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 LIMIT ${limit}`;
+  }
+
+  getRandomSampleFilter(samplePct: number): string {
+    if (samplePct <= 0 || samplePct >= 1) {
+      return '';
+    }
+    return `rand() < ${samplePct}`;
+  }
+
+  getTableSampleClause(samplePct: number): string {
+    if (samplePct <= 0 || samplePct >= 1) {
+      return '';
+    }
+    return `TABLESAMPLE BERNOULLI (${Math.min(100, Math.max(0, samplePct * 100))})`;
+  }
+
+  getLimitOffsetClause(limit: number, offset?: number): string {
+    return limitOffsetClause(limit, offset);
+  }
+
+  getTopClause(_limit: number): string {
+    return '';
+  }
+
+  getNullCountExpression(column: string): string {
+    return `count_if(${column} IS NULL)`;
+  }
+
+  getDistinctCountExpression(column: string): string {
+    return `COUNT(DISTINCT ${column})`;
+  }
+
+  textLengthExpression(columnSql: string): string {
+    return `length(CAST(${columnSql} AS VARCHAR))`;
+  }
+
+  castToText(columnSql: string): string {
+    return `CAST(${columnSql} AS VARCHAR)`;
+  }
+
+  getSampleValueAggregation(innerSql: string): string {
+    return `(SELECT array_join(array_agg(CAST(value AS VARCHAR)), '\\x1F') FROM (${innerSql}) AS relationship_profile_values)`;
+  }
+
+  generateCardinalitySampleQuery(tableName: string, columnName: string, sampleSize: number): string {
+    return `
+      SELECT COUNT(DISTINCT val) AS cardinality
+      FROM (
+        SELECT ${columnName} AS val
+        FROM ${tableName}
+        WHERE ${columnName} IS NOT NULL
+        LIMIT ${sampleSize}
+      )
+    `;
+  }
+
+  generateRandomizedCardinalitySampleQuery(tableName: string, columnName: string, sampleSize: number): string {
+    return `
+      SELECT COUNT(DISTINCT val) AS cardinality
+      FROM (
+        SELECT ${columnName} AS val
+        FROM ${tableName}
+        WHERE ${columnName} IS NOT NULL
+        ORDER BY rand()
+        LIMIT ${sampleSize}
+      )
+    `;
+  }
+
+  generateDistinctValuesQuery(tableName: string, columnName: string, limit: number): string {
+    return `
+      SELECT DISTINCT CAST(${columnName} AS VARCHAR) AS val
+      FROM ${tableName}
+      WHERE ${columnName} IS NOT NULL
+      ORDER BY val
+      LIMIT ${limit}
+    `;
+  }
+
+  generateColumnStatisticsQuery(_schemaName: string, _tableName: string): string | null {
+    return null;
+  }
+}
diff --git a/packages/cli/src/context/connections/dialects.ts b/packages/cli/src/context/connections/dialects.ts
index c7929cea6..0ebf7f582 100644
--- a/packages/cli/src/context/connections/dialects.ts
+++ b/packages/cli/src/context/connections/dialects.ts
@@ -5,6 +5,7 @@ import { KtxPostgresDialect } from '../../connectors/postgres/dialect.js';
 import { KtxSqliteDialect } from '../../connectors/sqlite/dialect.js';
 import { KtxSnowflakeDialect } from '../../connectors/snowflake/dialect.js';
 import { KtxSqlServerDialect } from '../../connectors/sqlserver/dialect.js';
+import { KtxTrinoDialect } from '../../connectors/trino/dialect.js';
 import type { KtxConnectionDriver, KtxSchemaDimensionType, KtxTableRef } from '../scan/types.js';
 import type { KtxDialectTableRef } from './dialect-helpers.js';
 
@@ -42,6 +43,7 @@ const supportedDrivers: KtxConnectionDriver[] = [
   'sqlite',
   'snowflake',
   'sqlserver',
+  'trino',
 ];
 
 const dialectFactories: Record<KtxConnectionDriver, () => KtxDialect> = {
@@ -52,6 +54,7 @@ const dialectFactories: Record<KtxConnectionDriver, () => KtxDialect> = {
   sqlite: () => new KtxSqliteDialect(),
   snowflake: () => new KtxSnowflakeDialect(),
   sqlserver: () => new KtxSqlServerDialect(),
+  trino: () => new KtxTrinoDialect(),
 };
 
 export function getDialectForDriver(driver: string): KtxDialect {
diff --git a/packages/cli/src/context/connections/drivers.ts b/packages/cli/src/context/connections/drivers.ts
index 3fbeb058c..1b8d1953c 100644
--- a/packages/cli/src/context/connections/drivers.ts
+++ b/packages/cli/src/context/connections/drivers.ts
@@ -1,7 +1,7 @@
 import type { KtxConnectionDriver, KtxScanConnector } from '../scan/types.js';
 
 /** @internal */
-export type KtxScopeConfigKey = 'dataset_ids' | 'databases' | 'schemas' | 'schema_names';
+export type KtxScopeConfigKey = 'dataset_ids' | 'databases' | 'schemas' | 'schema_names' | 'catalogs';
 
 /** @internal */
 export interface KtxDriverConnectorModule {
@@ -173,6 +173,27 @@ export const driverRegistrations: Record<KtxConnectionDriver, KtxDriverRegistrat
       };
     },
   },
+  trino: {
+    driver: 'trino',
+    scopeConfigKey: 'catalogs',
+    hasHistoricSqlReader: false,
+    load: async () => {
+      const m = await import('../../connectors/trino/connector.js');
+      return {
+        isConnectionConfig: (connection) => {
+          const typedConnection = connection as Parameters<typeof m.isKtxTrinoConnectionConfig>[0];
+          return m.isKtxTrinoConnectionConfig(typedConnection);
+        },
+        createScanConnector: ({ connectionId, connection }) => {
+          const typedConnection = connection as Parameters<typeof m.isKtxTrinoConnectionConfig>[0];
+          if (!m.isKtxTrinoConnectionConfig(typedConnection)) {
+            throw invalidConnectionConfig('trino');
+          }
+          return new m.KtxTrinoScanConnector({ connectionId, connection: typedConnection });
+        },
+      };
+    },
+  },
 };
 
 const supportedDrivers = Object.keys(driverRegistrations).sort() as KtxConnectionDriver[];
diff --git a/packages/cli/src/context/project/driver-schemas.ts b/packages/cli/src/context/project/driver-schemas.ts
index 1b2434c6b..7520fe68e 100644
--- a/packages/cli/src/context/project/driver-schemas.ts
+++ b/packages/cli/src/context/project/driver-schemas.ts
@@ -13,6 +13,7 @@ const warehouseDrivers = [
   'sqlite',
   'clickhouse',
   'sqlserver',
+  'trino',
 ] as const;
 
 type WarehouseDriver = (typeof warehouseDrivers)[number];
@@ -46,6 +47,7 @@ const warehouseConnectionSchemas = [
   warehouseConnectionSchema('sqlite'),
   warehouseConnectionSchema('clickhouse'),
   warehouseConnectionSchema('sqlserver'),
+  warehouseConnectionSchema('trino'),
 ] as const;
 
 const positiveIntKeyMessage = (field: string) => `${field} keys must be positive-integer strings (e.g. "1", "42")`;
diff --git a/packages/cli/src/context/scan/types.ts b/packages/cli/src/context/scan/types.ts
index fc445b5ec..0474fa0f2 100644
--- a/packages/cli/src/context/scan/types.ts
+++ b/packages/cli/src/context/scan/types.ts
@@ -7,7 +7,8 @@ export type KtxConnectionDriver =
   | 'bigquery'
   | 'snowflake'
   | 'mysql'
-  | 'clickhouse';
+  | 'clickhouse'
+  | 'trino';
 
 export type KtxScanMode = 'structural' | 'relationships' | 'enriched';
 
diff --git a/packages/cli/src/context/sql-analysis/dialect.ts b/packages/cli/src/context/sql-analysis/dialect.ts
index 54bc95256..22c44b95e 100644
--- a/packages/cli/src/context/sql-analysis/dialect.ts
+++ b/packages/cli/src/context/sql-analysis/dialect.ts
@@ -15,6 +15,7 @@ const SQLGLOT_DIALECTS: Record<string, SqlAnalysisDialect> = {
   sqlite: 'sqlite',
   duckdb: 'duckdb',
   clickhouse: 'clickhouse',
+  trino: 'trino',
   databricks: 'databricks',
 };
 
diff --git a/packages/cli/src/context/sql-analysis/ports.ts b/packages/cli/src/context/sql-analysis/ports.ts
index 898fca386..43398d1ed 100644
--- a/packages/cli/src/context/sql-analysis/ports.ts
+++ b/packages/cli/src/context/sql-analysis/ports.ts
@@ -9,6 +9,7 @@ export type SqlAnalysisDialect =
   | 'sqlite'
   | 'tsql'
   | 'clickhouse'
+  | 'trino'
   | (string & {});
 
 export type SqlAnalysisLiteralSlotType = 'string' | 'number' | 'timestamp' | 'date' | 'boolean' | 'null' | 'unknown';
diff --git a/packages/cli/src/setup-databases.ts b/packages/cli/src/setup-databases.ts
index 9f9f828db..1c3e1dfcd 100644
--- a/packages/cli/src/setup-databases.ts
+++ b/packages/cli/src/setup-databases.ts
@@ -69,7 +69,8 @@ export type KtxSetupDatabaseDriver =
   | 'clickhouse'
   | 'sqlserver'
   | 'bigquery'
-  | 'snowflake';
+  | 'snowflake'
+  | 'trino';
 
 export interface KtxSetupDatabasesArgs {
   projectDir: string;
@@ -156,6 +157,7 @@ const DRIVER_OPTIONS: Array<{ value: KtxSetupDatabaseDriver; label: string }> =
   { value: 'mysql', label: 'MySQL' },
   { value: 'clickhouse', label: 'ClickHouse' },
   { value: 'sqlserver', label: 'SQL Server' },
+  { value: 'trino', label: 'Trino' },
   { value: 'sqlite', label: 'SQLite' },
 ];
 
@@ -178,6 +180,7 @@ const DEFAULT_CONNECTION_IDS: Record<KtxSetupDatabaseDriver, string> = {
   sqlserver: 'sqlserver-warehouse',
   bigquery: 'bigquery-warehouse',
   snowflake: 'snowflake-warehouse',
+  trino: 'trino-warehouse',
 };
 
 interface ScopeDiscoverySpec {
@@ -255,15 +258,23 @@ const SCOPE_DISCOVERY_SPECS: Partial<Record<KtxSetupDatabaseDriver, ScopeDiscove
     configSingleField: 'schema_name',
     suggest: defaultSuggest,
   },
+  trino: {
+    noun: 'catalog',
+    nounPlural: 'catalogs',
+    promptLabel: 'Trino catalogs',
+    configArrayField: 'catalogs',
+    suggest: defaultSuggest,
+  },
 };
 
-type UrlDriverType = Extract<KtxSetupDatabaseDriver, 'postgres' | 'mysql' | 'clickhouse' | 'sqlserver'>;
+type UrlDriverType = Extract<KtxSetupDatabaseDriver, 'postgres' | 'mysql' | 'clickhouse' | 'sqlserver' | 'trino'>;
 
 const DRIVER_CONNECTION_DEFAULTS: Record<UrlDriverType, { port: string }> = {
   postgres: { port: '5432' },
   mysql: { port: '3306' },
   clickhouse: { port: '8123' },
   sqlserver: { port: '1433' },
+  trino: { port: '8080' },
 };
 
 function driverLabel(driver: KtxSetupDatabaseDriver): string {
diff --git a/packages/cli/test/connectors/trino/connector.test.ts b/packages/cli/test/connectors/trino/connector.test.ts
new file mode 100644
index 000000000..297a5224a
--- /dev/null
+++ b/packages/cli/test/connectors/trino/connector.test.ts
@@ -0,0 +1,200 @@
+import { describe, expect, it, vi } from 'vitest';
+import {
+  isKtxTrinoConnectionConfig,
+  KtxTrinoScanConnector,
+  trinoClientConfigFromConfig,
+  type KtxTrinoClient,
+  type KtxTrinoClientFactory,
+  type KtxTrinoConnectionConfig,
+  type KtxTrinoQueryResult,
+} from '../../../src/connectors/trino/connector.js';
+import { tableRefSet } from '../../../src/context/scan/table-ref.js';
+import type { KtxScanContext } from '../../../src/context/scan/types.js';
+
+const ctx: KtxScanContext = { runId: 'test-run' };
+
+function column(name: string, type = 'varchar'): { name: string; type: string } {
+  return { name, type };
+}
+
+/**
+ * Fake Trino client that answers each query shape the connector issues by
+ * matching on the SQL it generates. Returns the normalized
+ * {@link KtxTrinoQueryResult} (columns + rows) the real HTTP client would.
+ */
+function fakeClientFactory(): KtxTrinoClientFactory {
+  const query = vi.fn(async (sql: string): Promise<KtxTrinoQueryResult> => {
+    if (sql.trim() === 'SELECT 1') {
+      return { columns: [column('_col0', 'integer')], rows: [[1]] };
+    }
+    if (sql.includes('system.metadata.catalogs')) {
+      // `system` and `tpch` are filtered out by SYSTEM_CATALOGS.
+      return {
+        columns: [column('catalog_name')],
+        rows: [['hive'], ['system'], ['tpch']],
+      };
+    }
+    if (sql.includes('information_schema.schemata')) {
+      return { columns: [column('schema_name')], rows: [['analytics']] };
+    }
+    if (sql.includes('information_schema.tables')) {
+      return {
+        columns: [column('table_catalog'), column('table_schema'), column('table_name'), column('table_type')],
+        rows: [
+          ['hive', 'analytics', 'orders', 'BASE TABLE'],
+          ['hive', 'analytics', 'order_summary', 'VIEW'],
+        ],
+      };
+    }
+    if (sql.includes('information_schema.columns')) {
+      return {
+        columns: [
+          column('table_catalog'),
+          column('table_schema'),
+          column('table_name'),
+          column('column_name'),
+          column('data_type'),
+          column('is_nullable'),
+          column('comment'),
+        ],
+        rows: [
+          ['hive', 'analytics', 'orders', 'id', 'bigint', 'NO', 'Primary key'],
+          ['hive', 'analytics', 'orders', 'status', 'varchar', 'YES', null],
+          ['hive', 'analytics', 'order_summary', 'status', 'varchar', 'YES', null],
+        ],
+      };
+    }
+    if (sql.includes('FROM "hive"."analytics"."orders"') && sql.includes('"id"')) {
+      return { columns: [column('id', 'bigint'), column('status')], rows: [[1, 'paid']] };
+    }
+    if (sql.includes('FROM "hive"."analytics"."orders"')) {
+      return { columns: [column('status')], rows: [['paid'], ['open']] };
+    }
+    if (sql.includes('SELECT id, status FROM orders')) {
+      return { columns: [column('id', 'bigint'), column('status')], rows: [[1, 'paid']] };
+    }
+    throw new Error(`Unexpected SQL: ${sql}`);
+  });
+
+  const client: KtxTrinoClient = { query, close: vi.fn(async () => undefined) };
+  return { createClient: () => client };
+}
+
+function connector(connection: Partial<KtxTrinoConnectionConfig> = {}): KtxTrinoScanConnector {
+  return new KtxTrinoScanConnector({
+    connectionId: 'trino-warehouse',
+    connection: { driver: 'trino', host: 'trino.example.com', port: 8080, user: 'analyst', ...connection },
+    clientFactory: fakeClientFactory(),
+  });
+}
+
+describe('isKtxTrinoConnectionConfig', () => {
+  it('accepts trino configs case-insensitively and rejects others', () => {
+    expect(isKtxTrinoConnectionConfig({ driver: 'trino' })).toBe(true);
+    expect(isKtxTrinoConnectionConfig({ driver: 'Trino' })).toBe(true);
+    expect(isKtxTrinoConnectionConfig({ driver: 'snowflake' })).toBe(false);
+    expect(isKtxTrinoConnectionConfig(undefined)).toBe(false);
+  });
+});
+
+describe('trinoClientConfigFromConfig', () => {
+  it('builds an http server URL from host/port and requires a user', () => {
+    const config = trinoClientConfigFromConfig({
+      connectionId: 'trino-warehouse',
+      connection: { driver: 'trino', host: 'trino.example.com', port: 8080, user: 'analyst' },
+    });
+    expect(config.server).toBe('http://trino.example.com:8080');
+    expect(config.user).toBe('analyst');
+    expect(config.ssl).toBe(false);
+  });
+
+  it('derives ssl from an https url', () => {
+    const config = trinoClientConfigFromConfig({
+      connectionId: 'trino-warehouse',
+      connection: { driver: 'trino', url: 'https://trino.example.com:8443', user: 'analyst' },
+    });
+    expect(config.server).toBe('https://trino.example.com:8443');
+    expect(config.ssl).toBe(true);
+  });
+
+  it('throws when no user is provided', () => {
+    expect(() =>
+      trinoClientConfigFromConfig({
+        connectionId: 'trino-warehouse',
+        connection: { driver: 'trino', host: 'trino.example.com' },
+      }),
+    ).toThrow(/requires connections.trino-warehouse.user/);
+  });
+});
+
+describe('KtxTrinoScanConnector', () => {
+  it('introspects every non-system catalog into catalog-qualified tables', async () => {
+    const snapshot = await connector().introspect({ connectionId: 'trino-warehouse', driver: 'trino' }, ctx);
+
+    expect(snapshot.driver).toBe('trino');
+    expect(snapshot.scope.catalogs).toEqual(['hive']);
+    expect(snapshot.tables).toHaveLength(2);
+
+    const orders = snapshot.tables.find((table) => table.name === 'orders');
+    expect(orders).toMatchObject({ catalog: 'hive', db: 'analytics', kind: 'table' });
+    expect(orders?.columns.map((c) => c.name)).toEqual(['id', 'status']);
+    expect(orders?.columns[0]).toMatchObject({ nativeType: 'bigint', dimensionType: 'number', nullable: false });
+    expect(orders?.columns[1]).toMatchObject({ dimensionType: 'string', nullable: true });
+
+    const view = snapshot.tables.find((table) => table.name === 'order_summary');
+    expect(view?.kind).toBe('view');
+  });
+
+  it('honors an explicit catalogs allowlist without querying system.metadata.catalogs', async () => {
+    const snapshot = await connector({ catalogs: ['hive'] }).introspect(
+      { connectionId: 'trino-warehouse', driver: 'trino' },
+      ctx,
+    );
+    expect(snapshot.scope.catalogs).toEqual(['hive']);
+  });
+
+  it('restricts introspection to tableScope', async () => {
+    const tableScope = tableRefSet([{ catalog: 'hive', db: 'analytics', name: 'orders' }]);
+    const snapshot = await connector().introspect(
+      { connectionId: 'trino-warehouse', driver: 'trino', tableScope },
+      ctx,
+    );
+    expect(snapshot.tables.map((table) => table.name)).toEqual(['orders']);
+  });
+
+  it('namespaces schemas and tables by catalog', async () => {
+    const conn = connector();
+    expect(await conn.listSchemas()).toEqual(['hive.analytics']);
+    const tables = await conn.listTables();
+    expect(tables).toContainEqual({ catalog: 'hive', schema: 'analytics', name: 'orders', kind: 'table' });
+  });
+
+  it('samples tables and columns', async () => {
+    const conn = connector();
+    const table = await conn.sampleTable(
+      { connectionId: 'trino-warehouse', table: { catalog: 'hive', db: 'analytics', name: 'orders' }, columns: ['id', 'status'], limit: 5 },
+      ctx,
+    );
+    expect(table.headers).toEqual(['id', 'status']);
+    expect(table.rows).toEqual([[1, 'paid']]);
+
+    const col = await conn.sampleColumn(
+      { connectionId: 'trino-warehouse', table: { catalog: 'hive', db: 'analytics', name: 'orders' }, column: 'status', limit: 10 },
+      ctx,
+    );
+    expect(col.values).toEqual(['paid', 'open']);
+  });
+
+  it('executes read-only SQL and rejects mutations', async () => {
+    const conn = connector();
+    const result = await conn.executeReadOnly({ connectionId: 'trino-warehouse', sql: 'SELECT id, status FROM orders' }, ctx);
+    expect(result.rowCount).toBe(1);
+    await expect(
+      conn.executeReadOnly({ connectionId: 'trino-warehouse', sql: 'DELETE FROM orders' }, ctx),
+    ).rejects.toThrow();
+  });
+
+  it('reports a successful connection test', async () => {
+    expect(await connector().testConnection()).toEqual({ success: true });
+  });
+});
diff --git a/packages/cli/test/connectors/trino/dialect.test.ts b/packages/cli/test/connectors/trino/dialect.test.ts
new file mode 100644
index 000000000..3869c76c4
--- /dev/null
+++ b/packages/cli/test/connectors/trino/dialect.test.ts
@@ -0,0 +1,64 @@
+import { describe, expect, it } from 'vitest';
+import { KtxTrinoDialect } from '../../../src/connectors/trino/dialect.js';
+
+describe('KtxTrinoDialect', () => {
+  const dialect = new KtxTrinoDialect();
+
+  it('quotes identifiers and formats catalog.schema.table names (three-part)', () => {
+    expect(dialect.quoteIdentifier('order"items')).toBe('"order""items"');
+    expect(dialect.formatTableName({ catalog: 'hive', db: 'analytics', name: 'orders' })).toBe(
+      '"hive"."analytics"."orders"',
+    );
+    // Trino is fully qualified; a two-part ref omits only the absent catalog.
+    expect(dialect.formatTableName({ db: 'analytics', name: 'orders' })).toBe('"analytics"."orders"');
+    expect(dialect.formatTableName({ name: 'orders' })).toBe('"orders"');
+  });
+
+  it('parses three-part display refs and rejects two-part ones', () => {
+    expect(dialect.parseDisplayRef('hive.analytics.orders')).toEqual({
+      catalog: 'hive',
+      db: 'analytics',
+      name: 'orders',
+    });
+    expect(dialect.parseDisplayRef('analytics.orders')).toBeNull();
+    expect(dialect.columnDisplayTablePartCount()).toBe(3);
+  });
+
+  it('maps native Trino types to scan dimensions, unwrapping parameters', () => {
+    expect(dialect.mapDataType('decimal(10,2)')).toBe('decimal(10,2)');
+    expect(dialect.mapToDimensionType('timestamp(3) with time zone')).toBe('time');
+    expect(dialect.mapToDimensionType('date')).toBe('time');
+    expect(dialect.mapToDimensionType('bigint')).toBe('number');
+    expect(dialect.mapToDimensionType('decimal(38,0)')).toBe('number');
+    expect(dialect.mapToDimensionType('double')).toBe('number');
+    expect(dialect.mapToDimensionType('boolean')).toBe('boolean');
+    expect(dialect.mapToDimensionType('varchar(255)')).toBe('string');
+    expect(dialect.mapToDimensionType('json')).toBe('string');
+    expect(dialect.mapToDimensionType('array(varchar)')).toBe('string');
+  });
+
+  it('generates ANSI sampling and dictionary SQL', () => {
+    expect(dialect.generateSampleQuery('"hive"."analytics"."orders"', 5, ['id', 'status'])).toBe(
+      'SELECT "id", "status" FROM "hive"."analytics"."orders" LIMIT 5',
+    );
+    expect(dialect.generateColumnSampleQuery('"hive"."analytics"."orders"', 'status', 10)).toBe(
+      'SELECT "status" FROM "hive"."analytics"."orders" WHERE "status" IS NOT NULL LIMIT 10',
+    );
+    expect(dialect.generateCardinalitySampleQuery('"t"', '"status"', 100)).toContain(
+      'SELECT COUNT(DISTINCT val) AS cardinality',
+    );
+    expect(dialect.generateDistinctValuesQuery('"t"', '"status"', 20)).toContain(
+      'SELECT DISTINCT CAST("status" AS VARCHAR) AS val',
+    );
+  });
+
+  it('produces a BERNOULLI table-sample clause only for in-range fractions', () => {
+    expect(dialect.getTableSampleClause(0.1)).toBe('TABLESAMPLE BERNOULLI (10)');
+    expect(dialect.getTableSampleClause(0)).toBe('');
+    expect(dialect.getTableSampleClause(1)).toBe('');
+  });
+
+  it('keeps unsupported statistics explicit', () => {
+    expect(dialect.generateColumnStatisticsQuery('analytics', 'orders')).toBeNull();
+  });
+});
diff --git a/packages/cli/test/context/connections/drivers.test.ts b/packages/cli/test/context/connections/drivers.test.ts
index 5d59db4e4..055fcfa36 100644
--- a/packages/cli/test/context/connections/drivers.test.ts
+++ b/packages/cli/test/context/connections/drivers.test.ts
@@ -64,9 +64,16 @@ const connectionFixtures: Record<KtxConnectionDriver, FixtureFactory> = {
     database: 'ANALYTICS',
     schema: 'PUBLIC',
   }),
+  trino: () => ({
+    driver: 'trino',
+    host: 'localhost',
+    port: 8080,
+    user: 'reader',
+    catalogs: ['hive'],
+  }),
 };
 
-const allowedScopeKeys = new Set(['dataset_ids', 'databases', 'schemas', 'schema_names']);
+const allowedScopeKeys = new Set(['dataset_ids', 'databases', 'schemas', 'schema_names', 'catalogs']);
 const historicSqlReaderDrivers = new Set<KtxConnectionDriver>(['postgres', 'bigquery', 'snowflake']);
 
 function assertExportedRegistryBoundaryTypes(input: {
@@ -101,6 +108,7 @@ describe('driverRegistrations', () => {
       'snowflake',
       'sqlite',
       'sqlserver',
+      'trino',
     ]);
   });
 
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index 2eea99969..3410fae60 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -224,6 +224,9 @@ importers:
       snowflake-sdk:
         specifier: ^2.4.2
         version: 2.4.2(asn1.js@5.4.1)
+      trino-client:
+        specifier: ^0.2.9
+        version: 0.2.9
       yaml:
         specifier: ^2.9.0
         version: 2.9.0
@@ -2772,6 +2775,9 @@ packages:
     resolution: {integrity: sha512-NZKeq9AfyQvEeNlN0zSYAaWrmBffJh3IELMZfRpJVWgrpEbtEpnjvzqBPf+mxoI287JohRDoa+/nsfqqiZmF6g==}
     engines: {node: '>= 6.0.0'}
 
+  axios@1.13.2:
+    resolution: {integrity: sha512-VPk9ebNqPcy5lRGuSlKx752IlDatOjT9paPlm8A7yOuW2Fbvp4X3JznJtT4f0GzGLLiWE9W8onz51SqLYwzGaA==}
+
   axios@1.16.1:
     resolution: {integrity: sha512-caYkukvroVPO8KrzuJEb50Hm07KwfBZPEC3VeFHTsqWHvKTsy54hjJz9BS/cdaypROE2rH6xvm9mHX4fgWkr3A==}
 
@@ -5061,6 +5067,9 @@ packages:
     resolution: {integrity: sha512-llQsMLSUDUPT44jdrU/O37qlnifitDP+ZwrmmZcoSKyLKvtZxpyV0n2/bD/N4tBAAZ/gJEdZU7KMraoK1+XYAg==}
     engines: {node: '>= 0.10'}
 
+  proxy-from-env@1.1.0:
+    resolution: {integrity: sha512-D+zkORCbA9f1tdWRK0RaCR3GPv50cMxcrz4X8k5LTSUD1Dkw47mKJEZQNunItRTkWwgtaUSo1RVFRIG9ZXiFYg==}
+
   proxy-from-env@2.1.0:
     resolution: {integrity: sha512-cJ+oHTW1VAEa8cJslgmUZrc+sjRKgAKl3Zyse6+PV38hZe/V6Z14TbCuXcan9F9ghlz4QrFr2c92TNF82UkYHA==}
     engines: {node: '>=10'}
@@ -5641,6 +5650,9 @@ packages:
   trim-lines@3.0.1:
     resolution: {integrity: sha512-kRj8B+YHZCc9kQYdWfJB2/oUl9rA99qbowYYBtr4ui4mZyAQ2JpvVBd/6U2YloATfqBhBTSMhTpgBHtU0Mf3Rg==}
 
+  trino-client@0.2.9:
+    resolution: {integrity: sha512-bINcQxDH5v9DR1zqXtoNqnRJ5leYMyYzRuFZLsoqg4irWYPDh/4wBndC8c2x+QfNIOr6c35BtHKOwotelqSATg==}
+
   triple-beam@1.4.1:
     resolution: {integrity: sha512-aZbgViZrg1QNcG+LULa7nhZpJTZSLm/mXnHXnbAbjmN5aSa0y7V+wvv6+4WaBtpISJzThKy+PIPxc1Nq1EJ9mg==}
     engines: {node: '>= 14.0.0'}
@@ -8652,6 +8664,14 @@ snapshots:
 
   aws-ssl-profiles@1.1.2: {}
 
+  axios@1.13.2:
+    dependencies:
+      follow-redirects: 1.16.0
+      form-data: 4.0.5
+      proxy-from-env: 1.1.0
+    transitivePeerDependencies:
+      - debug
+
   axios@1.16.1:
     dependencies:
       follow-redirects: 1.16.0
@@ -11212,6 +11232,8 @@ snapshots:
       forwarded: 0.2.0
       ipaddr.js: 1.9.1
 
+  proxy-from-env@1.1.0: {}
+
   proxy-from-env@2.1.0: {}
 
   pump@3.0.4:
@@ -11986,6 +12008,12 @@ snapshots:
 
   trim-lines@3.0.1: {}
 
+  trino-client@0.2.9:
+    dependencies:
+      axios: 1.13.2
+    transitivePeerDependencies:
+      - debug
+
   triple-beam@1.4.1: {}
 
   trough@2.2.0: {}