Add DuckDB data diff e2e coverage

Author: anandgupta42

docs/internal/2026-06-08-dbt-pr-review-self-improvement-loop.md

@@ -285,6 +285,43 @@ Implemented result:
 - Real-world corpus floor preserved: 15/15 bad cases caught, 0/5 false
   positives.
 
+## Current Acceptance Checkpoint
+
+Demo matrix with local core, base artifacts in `target-base`, head artifacts in
+`target`, and AI disabled:
+
+| Branch | Expected | Actual | Deterministic evidence |
+|---|---:|---:|---|
+| `demo/safe-refactor` | `APPROVE` | `APPROVE` | no findings |
+| `demo/join-key-breakage` | `REQUEST_CHANGES` | `REQUEST_CHANGES` | `altimate_core.structural_diff:SC010`, core equivalence |
+| `demo/test-removal` | `COMMENT` | `COMMENT` | `dbt-patterns:removed_tests` |
+| `demo/new-pii-exposure` | `REQUEST_CHANGES` | `REQUEST_CHANGES` | `altimate_core.classify_pii`, core equivalence |
+| `demo/mart-select-star` | `COMMENT` | `COMMENT` | core equivalence warning |
+| `demo/incremental-without-guard` | `COMMENT` | `COMMENT` | core dbt config lint + deterministic catalog cost notes |
+
+DuckDB e2e proof:
+
+- Branch: `demo/join-key-breakage`.
+- `dbt build --profiles-dir . --target dev`: passed `PASS=14 WARN=0 ERROR=0`.
+- Reviewer: `REQUEST_CHANGES` with critical `join_risk` from
+  `altimate_core.structural_diff` rule `join_key_regression` (`SC010`).
+
+Warehouse data-diff e2e proof:
+
+- Test: `packages/opencode/test/altimate/data-diff-duckdb-e2e.test.ts`.
+- Connection: local DuckDB via `@altimateai/drivers/duckdb`.
+- Base/head relations: `base_orders` and `head_orders`.
+- Key columns: `order_id`; compared value column: `amount`.
+- Observed warehouse delta: one row only in head and one updated value.
+- Validation:
+  `ALTIMATE_RUN_WAREHOUSE_E2E=1 bun test --timeout 30000 test/altimate/data-diff-duckdb-e2e.test.ts`.
+- Default fast-loop behavior:
+  `bun test --timeout 30000 test/altimate/data-diff-duckdb-e2e.test.ts`
+  skips unless `ALTIMATE_RUN_WAREHOUSE_E2E=1` is set.
+- Implementation note: fixed the DuckDB connector to use the two-argument
+  constructor when no open options are required; Bun's native binding can miss
+  the callback when `undefined` is passed as the second argument.
+
 ## Warehouse E2E Policy
 
 Use local DuckDB first for fast end-to-end checks, then add BigQuery runs when a

packages/drivers/src/duckdb.ts

@@ -56,26 +56,29 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
         new Promise<any>((resolve, reject) => {
           let resolved = false
           let timeout: ReturnType<typeof setTimeout> | undefined
+          let instance: any
           const opts = accessMode ? { access_mode: accessMode } : undefined
-          const instance = new duckdb.Database(
-            dbPath,
-            opts,
-            (err: Error | null) => {
-              if (resolved) { if (instance && typeof instance.close === "function") instance.close(); return }
-              resolved = true
-              if (timeout) clearTimeout(timeout)
-              if (err) {
-                const msg = err.message || String(err)
-                if (msg.toLowerCase().includes("locked") || msg.includes("SQLITE_BUSY") || msg.includes("DUCKDB_LOCKED")) {
-                  reject(new Error("DUCKDB_LOCKED"))
-                } else {
-                  reject(err)
-                }
+          const onOpen = (err: Error | null) => {
+            if (resolved) {
+              if (instance && typeof instance.close === "function") instance.close()
+              return
+            }
+            resolved = true
+            if (timeout) clearTimeout(timeout)
+            if (err) {
+              const msg = err.message || String(err)
+              if (msg.toLowerCase().includes("locked") || msg.includes("SQLITE_BUSY") || msg.includes("DUCKDB_LOCKED")) {
+                reject(new Error("DUCKDB_LOCKED"))
               } else {
-                resolve(instance)
+                reject(err)
               }
-            },
-          )
+            } else {
+              resolve(instance)
+            }
+          }
+          instance = opts
+            ? new duckdb.Database(dbPath, opts, onOpen)
+            : new duckdb.Database(dbPath, onOpen)
           // Bun: native callback may not fire; fall back after 2s
           timeout = setTimeout(() => {
             if (!resolved) {

packages/opencode/test/altimate/data-diff-duckdb-e2e.test.ts

@@ -0,0 +1,44 @@
+import { afterAll, beforeAll, describe, expect, test } from "bun:test"
+
+import { runDataDiff } from "../../src/altimate/native/connections/data-diff"
+import * as Registry from "../../src/altimate/native/connections/registry"
+
+const RUN = process.env.ALTIMATE_RUN_WAREHOUSE_E2E === "1"
+const e2eTest = RUN ? test : test.skip
+
+describe("data.diff DuckDB e2e", () => {
+  beforeAll(() => {
+    process.env.ALTIMATE_TELEMETRY_DISABLED = "true"
+  })
+
+  afterAll(() => {
+    delete process.env.ALTIMATE_TELEMETRY_DISABLED
+    Registry.reset()
+  })
+
+  e2eTest("detects value and row deltas through a real DuckDB warehouse", async () => {
+    Registry.reset()
+    Registry.setConfigs({ duck_e2e: { type: "duckdb", path: ":memory:" } })
+    const conn = await Registry.get("duck_e2e")
+
+    await conn.execute("CREATE TABLE base_orders (order_id INTEGER, amount INTEGER)")
+    await conn.execute("CREATE TABLE head_orders (order_id INTEGER, amount INTEGER)")
+    await conn.execute("INSERT INTO base_orders VALUES (1, 100), (2, 200)")
+    await conn.execute("INSERT INTO head_orders VALUES (1, 100), (2, 250), (3, 300)")
+
+    const result = await runDataDiff({
+      source: "base_orders",
+      target: "head_orders",
+      key_columns: ["order_id"],
+      extra_columns: ["amount"],
+      source_warehouse: "duck_e2e",
+      target_warehouse: "duck_e2e",
+      algorithm: "joindiff",
+    })
+
+    expect(result.success).toBe(true)
+    expect((result.outcome as any)?.mode).toBe("diff")
+    expect((result.outcome as any)?.stats?.exclusive_table2).toBe(1)
+    expect((result.outcome as any)?.stats?.updated).toBe(1)
+  })
+})