fix(schema): batch DuckDB introspection and bound indexing

packages/drivers/src/duckdb.ts

@@ -3,7 +3,7 @@
  */
 
 import { assertStoreExists, requireStorePath } from "./file-store"
-import type { ConnectionConfig, Connector, ConnectorResult, ExecuteOptions, SchemaColumn } from "./types"
+import type { ConnectionConfig, Connector, ConnectorResult, ExecuteOptions, SchemaColumn, SchemaMetadata } from "./types"
 import { loadOptionalDriver } from "./resolve"
 
 // altimate_change start — configurable, generous open budget
@@ -101,15 +101,34 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
   }
   // altimate_change end
 
-  function query(sql: string): Promise<any[]> {
+  function query(sql: string, signal?: AbortSignal): Promise<any[]> {
     return new Promise((resolve, reject) => {
-      connection.all(sql, (err: Error | null, rows: any[]) => {
-        if (err) reject(wrapDuckDBError(err))
-        else resolve(rows ?? [])
-      })
+      signal?.throwIfAborted()
+      // The legacy DuckDB SDK's interrupt() is a no-op. Stop waiting and avoid
+      // subsequent work on abort; the native query still finishes in the SDK.
+      const onAbort = () => reject(signal!.reason)
+      signal?.addEventListener("abort", onAbort, { once: true })
+      try {
+        connection.all(sql, (err: Error | null, rows: any[]) => {
+          signal?.removeEventListener("abort", onAbort)
+          if (err) reject(wrapDuckDBError(err))
+          else resolve(rows ?? [])
+        })
+      } catch (err) {
+        signal?.removeEventListener("abort", onAbort)
+        reject(err)
+      }
     })
   }
 
+  async function schemaNames(signal?: AbortSignal): Promise<string[]> {
+    const rows = await query(
+      "SELECT DISTINCT schema_name FROM information_schema.schemata WHERE catalog_name = current_database() ORDER BY schema_name",
+      signal,
+    )
+    return rows.map((r) => r.schema_name as string)
+  }
+
   function queryWithParams(sql: string, params: any[]): Promise<any[]> {
     return new Promise((resolve, reject) => {
       connection.all(sql, ...params, (err: Error | null, rows: any[]) => {
@@ -282,17 +301,14 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
     },
 
     async listSchemas(): Promise<string[]> {
-      const rows = await query(
-        "SELECT schema_name FROM information_schema.schemata ORDER BY schema_name",
-      )
-      return rows.map((r) => r.schema_name as string)
+      return schemaNames()
     },
 
     async listTables(
       schema: string,
     ): Promise<Array<{ name: string; type: string }>> {
       const rows = await queryWithParams(
-        `SELECT table_name, table_type FROM information_schema.tables WHERE table_schema = $1 ORDER BY table_name`,
+        `SELECT table_name, table_type FROM information_schema.tables WHERE table_catalog = current_database() AND table_schema = $1 ORDER BY table_name`,
         [schema],
       )
       return rows.map((r) => ({
@@ -306,7 +322,7 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
       table: string,
     ): Promise<SchemaColumn[]> {
       const rows = await queryWithParams(
-        `SELECT column_name, data_type, is_nullable FROM information_schema.columns WHERE table_schema = $1 AND table_name = $2 ORDER BY ordinal_position`,
+        `SELECT column_name, data_type, is_nullable FROM information_schema.columns WHERE table_catalog = current_database() AND table_schema = $1 AND table_name = $2 ORDER BY ordinal_position`,
         [schema, table],
       )
       return rows.map((r) => ({
@@ -316,6 +332,43 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
       }))
     },
 
+    async introspectSchema(signal?: AbortSignal): Promise<SchemaMetadata> {
+      const schemas = await schemaNames(signal)
+      const rows = await query(
+        `SELECT t.table_schema, t.table_name, t.table_type,
+                c.column_name, c.data_type, c.is_nullable
+         FROM information_schema.tables t
+         LEFT JOIN information_schema.columns c
+           ON c.table_catalog = t.table_catalog
+          AND c.table_schema = t.table_schema
+          AND c.table_name = t.table_name
+         WHERE t.table_catalog = current_database()
+         ORDER BY t.table_schema, t.table_name, c.ordinal_position`,
+        signal,
+      )
+      const tables: SchemaMetadata["tables"] = []
+      let table: SchemaMetadata["tables"][number] | undefined
+      for (const row of rows) {
+        if (!table || table.schema !== row.table_schema || table.name !== row.table_name) {
+          table = {
+            schema: row.table_schema,
+            name: row.table_name,
+            type: row.table_type === "VIEW" ? "view" : "table",
+            columns: [],
+          }
+          tables.push(table)
+        }
+        if (row.column_name !== null) {
+          table.columns.push({
+            name: row.column_name,
+            data_type: row.data_type,
+            nullable: row.is_nullable === "YES",
+          })
+        }
+      }
+      return { schemas, tables }
+    },
+
     async close() {
       if (db) {
         await new Promise<void>((resolve) => {

packages/drivers/src/index.ts

@@ -1,5 +1,5 @@
 // Re-export types
-export type { Connector, ConnectorResult, SchemaColumn, ConnectionConfig } from "./types"
+export type { Connector, ConnectorResult, SchemaColumn, SchemaMetadata, ConnectionConfig } from "./types"
 
 // Re-export config normalization
 export { normalizeConfig, sanitizeConnectionString } from "./normalize"

packages/drivers/src/types.ts

@@ -20,6 +20,11 @@ export interface SchemaColumn {
   nullable: boolean
 }
 
+export interface SchemaMetadata {
+  schemas: string[]
+  tables: Array<{ schema: string; name: string; type: string; columns: SchemaColumn[] }>
+}
+
 export interface ExecuteOptions {
   /** Skip the default LIMIT injection and post-truncation. Use when the caller
    *  needs the complete, untruncated result set (e.g. data-diff pipelines). */
@@ -32,5 +37,6 @@ export interface Connector {
   listSchemas(): Promise<string[]>
   listTables(schema: string): Promise<Array<{ name: string; type: string }>>
   describeTable(schema: string, table: string): Promise<SchemaColumn[]>
+  introspectSchema?(signal?: AbortSignal): Promise<SchemaMetadata>
   close(): Promise<void>
 }

packages/drivers/test/duckdb-schema.test.ts

@@ -0,0 +1,113 @@
+import { afterEach, describe, expect, spyOn, test } from "bun:test"
+import { connect } from "../src/duckdb"
+import type { Connector } from "../src/types"
+
+// Run in a separate process because other suites mock the DuckDB module.
+const duckdbTest = process.env.ALTIMATE_DUCKDB_E2E === "1" ? test : test.skip
+
+let connector: Connector | undefined
+let allSpy: ReturnType<typeof spyOn> | undefined
+let duckdb: typeof import("duckdb")
+
+afterEach(async () => {
+  allSpy?.mockRestore()
+  allSpy = undefined
+  await connector?.close()
+  connector = undefined
+})
+
+async function open() {
+  duckdb = await import("duckdb").then((module) => module.default)
+  connector = await connect({ type: "duckdb", path: ":memory:" })
+  await connector.connect()
+  return connector
+}
+
+describe("DuckDB schema introspection", () => {
+  duckdbTest("lists each schema once and scopes tables and columns to the current catalog", async () => {
+    const driver = await open()
+    await driver.execute("CREATE TABLE shared (local_id INTEGER NOT NULL)")
+    await driver.execute("ATTACH ':memory:' AS other")
+    await driver.execute("CREATE TABLE other.main.shared (foreign_id VARCHAR)")
+    await driver.execute("CREATE TABLE other.main.foreign_only (id INTEGER)")
+
+    expect(await driver.listSchemas()).toEqual(["main"])
+    expect(await driver.listTables("main")).toEqual([{ name: "shared", type: "table" }])
+    expect(await driver.describeTable("main", "shared")).toEqual([
+      { name: "local_id", data_type: "INTEGER", nullable: false },
+    ])
+  })
+
+  duckdbTest("reads complete table and view metadata with a constant number of queries", async () => {
+    const driver = await open()
+    await driver.execute("CREATE SCHEMA empty_schema")
+    await driver.execute("CREATE TABLE events (id INTEGER NOT NULL, label VARCHAR)")
+    await driver.execute("CREATE VIEW event_ids AS SELECT id FROM events")
+    await driver.execute("ATTACH ':memory:' AS other")
+    await driver.execute("CREATE TABLE other.main.events (foreign_id BIGINT)")
+    for (let i = 0; i < 24; i++) {
+      await driver.execute(`CREATE TABLE item_${i} (id INTEGER)`)
+    }
+    allSpy = spyOn(duckdb.Connection.prototype, "all")
+
+    const metadata = await driver.introspectSchema!()
+
+    expect(allSpy).toHaveBeenCalledTimes(2)
+    expect(metadata.schemas).toEqual(["empty_schema", "main"])
+    expect(metadata.tables).toHaveLength(26)
+    expect(metadata.tables.find((table) => table.name === "events")).toEqual({
+      schema: "main",
+      name: "events",
+      type: "table",
+      columns: [
+        { name: "id", data_type: "INTEGER", nullable: false },
+        { name: "label", data_type: "VARCHAR", nullable: true },
+      ],
+    })
+    expect(metadata.tables.find((table) => table.name === "event_ids")).toMatchObject({
+      schema: "main",
+      type: "view",
+      columns: [{ name: "id", data_type: "INTEGER", nullable: true }],
+    })
+  })
+
+  duckdbTest("does not truncate a large metadata inventory", async () => {
+    const driver = await open()
+    await driver.execute(
+      `CREATE TABLE wide_table (${Array.from({ length: 1100 }, (_, i) => `field_${i} INTEGER`).join(", ")})`,
+    )
+
+    const metadata = await driver.introspectSchema!()
+
+    expect(metadata.tables).toHaveLength(1)
+    expect(metadata.tables[0].columns).toHaveLength(1100)
+    expect(metadata.tables[0].columns.at(-1)?.name).toBe("field_1099")
+  })
+
+  duckdbTest("rejects an already aborted introspection without querying", async () => {
+    const driver = await open()
+    allSpy = spyOn(duckdb.Connection.prototype, "all")
+    const controller = new AbortController()
+    controller.abort(new Error("Index cancelled"))
+
+    await expect(driver.introspectSchema!(controller.signal)).rejects.toThrow("Index cancelled")
+    expect(allSpy).not.toHaveBeenCalled()
+  })
+
+  duckdbTest("stops waiting on abort and starts no further metadata queries", async () => {
+    const driver = await open()
+    let complete: ((error: Error | null, rows: unknown[]) => void) | undefined
+    allSpy = spyOn(duckdb.Connection.prototype, "all").mockImplementation(function (_sql, ...args) {
+      complete = args.at(-1) as typeof complete
+      return this
+    })
+    const controller = new AbortController()
+    const pending = driver.introspectSchema!(controller.signal)
+    controller.abort(new Error("Index cancelled"))
+
+    await expect(pending).rejects.toThrow("Index cancelled")
+    complete!(null, [{ schema_name: "main" }])
+    await Promise.resolve()
+    expect(allSpy).toHaveBeenCalledTimes(1)
+  })
+})