fix(ingestr): pin the DuckDB ADBC driver to the installed duckdb version (#1203)

DuckDB 2.0.0 was released and ingestr installs the newest DuckDB ADBC driver when none is installed, so files it wrote moved to a storage format the locked duckdb package cannot read. SQLBuild now installs the driver matching its duckdb package before running ingestr, unless ADBC_DRIVER_PATH is set, and the ingestr extra includes dbc.

Author: kvlonge

crates/sqlbuild-core/src/lifecycle_events/constants.rs

@@ -183,7 +183,8 @@ pub const OPERATION_KINDS: [&str; 10] = [
     "subprocess",
     "warehouse",
 ];
-pub const OPERATION_NAMES: [&str; 55] = [
+pub const OPERATION_NAMES: [&str; 56] = [
+    "adbc_driver_install",
     "audit_evaluation",
     "clone_execution",
     "clone_finalization",

crates/sqlbuild-core/src/progress/constants.rs

@@ -1,7 +1,7 @@
 //! Lifecycle event types, operation labels and layout of CLI progress lines.
 
 /// Operations with visible progress lines, and the label each renders with.
-pub(crate) const VISIBLE_OPERATION_LABELS: [(&str, &str); 31] = [
+pub(crate) const VISIBLE_OPERATION_LABELS: [(&str, &str); 32] = [
     ("clone_finalization", "Clone finalization"),
     ("discovery_declaration_parse", "Declaration parsing"),
     ("discovery_filesystem_walk", "Filesystem discovery"),
@@ -14,6 +14,7 @@ pub(crate) const VISIBLE_OPERATION_LABELS: [(&str, &str); 31] = [
     ("dbt_command", "dbt command"),
     ("external_source_load", "loader callable"),
     ("ingestr_command", "ingestr subprocess"),
+    ("adbc_driver_install", "DuckDB driver install"),
     ("managed_source_load", "loader callable"),
     ("python_asset", "Python asset attempt"),
     ("python_check", "Python check attempt"),
@@ -38,7 +39,8 @@ pub(crate) const VISIBLE_OPERATION_LABELS: [(&str, &str); 31] = [
     ("relation_promotion", "Promote relation"),
 ];
 /// Operations a live transient owner shows on a terminal when no resource attempt owns them.
-pub(crate) const TTY_TRANSIENT_OWNER_OPERATIONS: [&str; 18] = [
+pub(crate) const TTY_TRANSIENT_OWNER_OPERATIONS: [&str; 19] = [
+    "adbc_driver_install",
     "dbt_command",
     "discovery_declaration_parse",
     "discovery_filesystem_walk",

pyproject.toml

@@ -49,7 +49,8 @@ clickhouse = [
   "clickhouse-connect>=0.8",
 ]
 ingestr = [
-  "ingestr>=1.0.0",
+  "dbc>=0.3.1",
+  "ingestr>=1.1.73",
 ]
 dlt = [
   "dlt[duckdb,filesystem,sql_database]>=1.28.0,<2",
@@ -72,7 +73,8 @@ dev = [
   "dbt-snowflake==1.12.0b1",
   "dbt-sqlserver>=1.9.0,<1.11",
   "dlt[duckdb,filesystem,sql_database]>=1.28.0,<2",
-  "ingestr>=1.0.0",
+  "dbc>=0.3.1",
+  "ingestr>=1.1.73",
   "jsonschema>=4.23.0",
   "maturin>=1.9,<2",
   "pre-commit>=4.3.0",

src/sqlbuild/.agents/skills/sqlbuild/references/docs/integrations/ingestr.md

@@ -31,6 +31,11 @@ uv pip install 'sqlbuild[ingestr]'
 
 This installs `ingestr` alongside SQLBuild. The `ingestr` CLI must be available on `PATH`.
 
+ingestr reads and writes DuckDB through an ADBC driver. Before the first ingestr run, SQLBuild
+installs the DuckDB driver that matches its own `duckdb` package (with `dbc`, included in the extra)
+into `<environment>/etc/adbc/drivers`, so the files ingestr writes stay readable by SQLBuild. Set
+`ADBC_DRIVER_PATH` to manage the driver yourself.
+
 ## How it works
 
 1. Declare an `ingestr` block on a source in `sources/*.yml`

src/sqlbuild/integrations/ingestr/_helpers/runner.py

@@ -5,6 +5,8 @@
 import shutil
 import subprocess
 import sys
+import tomllib
+from importlib import metadata
 from os import environ
 from pathlib import Path
 
@@ -14,6 +16,7 @@
 from sqlbuild.runtime.observability.classes.operation_lifecycle import OperationLifecycle
 
 _DUCKDB_DRIVER_BOOTSTRAP_ERROR: str = "ADBC driver still not available after installation"
+_DUCKDB_MANIFEST: str = "duckdb.toml"
 
 
 def run_ingestr_command(
@@ -26,6 +29,7 @@ def run_ingestr_command(
             "This source uses ingestr, but the ingestr CLI is not available. "
             "Install it with: pip install 'sqlbuild[ingestr]'"
         )
+    ensure_duckdb_driver()
     command_display: str = format_ingestr_command(command)
     try:
         completed: subprocess.CompletedProcess[str] = _run_observed_ingestr_subprocess(
@@ -81,7 +85,74 @@ def _ingestr_subprocess_env() -> dict[str, str]:
     env: dict[str, str] = dict(environ)
     if env.get("ADBC_DRIVER_PATH"):
         return env
-    driver_path: Path = Path(sys.prefix) / "etc" / "adbc" / "drivers"
+    driver_path: Path = _managed_driver_path()
     if driver_path.exists():
         env["ADBC_DRIVER_PATH"] = str(driver_path)
     return env
+
+
+def _managed_driver_path() -> Path:
+    return Path(sys.prefix) / "etc" / "adbc" / "drivers"
+
+
+def ensure_duckdb_driver() -> None:
+    """Install the DuckDB ADBC driver matching the duckdb package unless the user set one."""
+
+    if environ.get("ADBC_DRIVER_PATH"):
+        return
+    try:
+        duckdb_version: str = metadata.version("duckdb")
+    except metadata.PackageNotFoundError:
+        return
+    driver_path: Path = _managed_driver_path()
+    if _installed_driver_version(driver_path / _DUCKDB_MANIFEST) == duckdb_version:
+        return
+    dbc: str | None = _dbc_executable()
+    if dbc is None:
+        raise IngestrIntegrationError(
+            "ingestr needs the dbc driver installer to pin its DuckDB driver to duckdb "
+            f"{duckdb_version}. Install it with: pip install 'sqlbuild[ingestr]'"
+        )
+    driver_path.mkdir(parents=True, exist_ok=True)
+    with OperationLifecycle(
+        operation_kind="subprocess",
+        operation_name="adbc_driver_install",
+        auto_fail_base_exceptions=False,
+    ) as lifecycle:
+        completed: subprocess.CompletedProcess[str] = subprocess.run(
+            (dbc, "install", "--quiet", f"duckdb={duckdb_version}"),
+            capture_output=True,
+            check=False,
+            env={**environ, "ADBC_DRIVER_PATH": str(driver_path)},
+            text=True,
+        )
+        if completed.returncode == 0:
+            lifecycle.completed(exit_code=0)
+        else:
+            lifecycle.failed(
+                error_code=f"exit_{completed.returncode}",
+                exit_code=completed.returncode,
+                signal_number=-completed.returncode if completed.returncode < 0 else None,
+            )
+    if completed.returncode != 0:
+        raise IngestrIntegrationError(
+            f"could not install the DuckDB ADBC driver for duckdb {duckdb_version}: "
+            f"{(completed.stderr or completed.stdout).strip()}"
+        )
+
+
+def _installed_driver_version(manifest: Path) -> str | None:
+    try:
+        with manifest.open("rb") as handle:
+            version: object = tomllib.load(handle).get("version")
+    except (OSError, tomllib.TOMLDecodeError):
+        return None
+    return version if isinstance(version, str) else None
+
+
+def _dbc_executable() -> str | None:
+    for name in ("dbc", "dbc.exe"):
+        beside_python: Path = Path(sys.executable).parent / name
+        if beside_python.exists():
+            return str(beside_python)
+    return shutil.which("dbc")

tests/unit/src/sqlbuild/integrations/ingestr/_test_types.py

@@ -38,3 +38,18 @@ class IngestrRunnerTestCase:
     returncode: int
     expected_signal_number: int
     expected_event_types: tuple[str, ...] = ()
+
+
+@dataclass(frozen=True)
+class IngestrDriverTestCase:
+    description: str
+    driver_files: tuple[tuple[str, str], ...]
+    environment: dict[str, str]
+    expected_installs: tuple[tuple[str, ...], ...]
+
+
+@dataclass(frozen=True)
+class IngestrDriverFailureTestCase:
+    description: str
+    stderr: str
+    expected_message: str

tests/unit/src/sqlbuild/integrations/ingestr/test_runner.py

@@ -3,18 +3,26 @@
 from __future__ import annotations
 
 import subprocess
+from pathlib import Path
 
 import pytest
 
-from sqlbuild.integrations.ingestr._helpers.runner import run_ingestr_command
+from sqlbuild.integrations.ingestr._helpers.runner import ensure_duckdb_driver, run_ingestr_command
 from sqlbuild.integrations.ingestr.exceptions import IngestrIntegrationError
 from sqlbuild.observability import (
     EventDispatcher,
     LifecycleEvent,
     dispatcher_scope,
     invocation_scope,
 )
-from tests.unit.src.sqlbuild.integrations.ingestr._test_types import IngestrRunnerTestCase
+from tests.unit.src.sqlbuild.integrations.ingestr._test_types import (
+    IngestrDriverFailureTestCase,
+    IngestrDriverTestCase,
+    IngestrRunnerTestCase,
+)
+
+_RUNNER: str = "sqlbuild.integrations.ingestr._helpers.runner"
+_INSTALL: tuple[str, ...] = ("dbc", "install", "--quiet", "duckdb=1.5.2")
 
 
 @pytest.mark.parametrize(
@@ -39,6 +47,9 @@ def test_given_signalled_ingestr_when_run_then_safe_terminal_excludes_child_stre
         "sqlbuild.integrations.ingestr._helpers.runner.shutil.which",
         lambda executable: executable,
     )
+    monkeypatch.setattr(
+        "sqlbuild.integrations.ingestr._helpers.runner.ensure_duckdb_driver", lambda: None
+    )
     monkeypatch.setattr(
         "sqlbuild.integrations.ingestr._helpers.runner._run_ingestr_subprocess",
         lambda **kwargs: subprocess.CompletedProcess(
@@ -92,6 +103,9 @@ def test_given_unobserved_ingestr_interruption_when_run_then_only_start_is_publi
         "sqlbuild.integrations.ingestr._helpers.runner.shutil.which",
         lambda executable: executable,
     )
+    monkeypatch.setattr(
+        "sqlbuild.integrations.ingestr._helpers.runner.ensure_duckdb_driver", lambda: None
+    )
     monkeypatch.setattr(
         "sqlbuild.integrations.ingestr._helpers.runner._run_ingestr_subprocess",
         lambda **kwargs: (_ for _ in ()).throw(KeyboardInterrupt),
@@ -105,3 +119,88 @@ def test_given_unobserved_ingestr_interruption_when_run_then_only_start_is_publi
         _ = run_ingestr_command(("ingestr", "run"))
 
     assert tuple(event.event_type for event in events) == test_case.expected_event_types
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    (
+        IngestrDriverTestCase(
+            description="missing driver installs the duckdb package version",
+            driver_files=(),
+            environment={},
+            expected_installs=(_INSTALL,),
+        ),
+        IngestrDriverTestCase(
+            description="driver of another version is replaced",
+            driver_files=(("duckdb.toml", "version = '2.0.0'\n"),),
+            environment={},
+            expected_installs=(_INSTALL,),
+        ),
+        IngestrDriverTestCase(
+            description="matching driver is kept",
+            driver_files=(("duckdb.toml", "version = '1.5.2'\n"),),
+            environment={},
+            expected_installs=(),
+        ),
+        IngestrDriverTestCase(
+            description="user driver path is left alone",
+            driver_files=(),
+            environment={"ADBC_DRIVER_PATH": "/custom/adbc"},
+            expected_installs=(),
+        ),
+    ),
+    ids=lambda case: case.description,
+)
+def test_given_duckdb_driver_state_when_ensuring_then_installs_only_a_mismatch(
+    test_case: IngestrDriverTestCase,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    driver_path: Path = tmp_path / "etc" / "adbc" / "drivers"
+    driver_path.mkdir(parents=True)
+    _ = [(driver_path / name).write_text(text) for name, text in test_case.driver_files]
+    calls: list[tuple[tuple[str, ...], dict[str, str]]] = []
+    monkeypatch.setattr(f"{_RUNNER}.sys.prefix", str(tmp_path))
+    monkeypatch.setattr(f"{_RUNNER}.environ", dict(test_case.environment))
+    monkeypatch.setattr(f"{_RUNNER}.metadata.version", lambda name: "1.5.2")
+    monkeypatch.setattr(f"{_RUNNER}._dbc_executable", lambda: "dbc")
+    monkeypatch.setattr(
+        f"{_RUNNER}.subprocess.run",
+        lambda args, **kwargs: (
+            calls.append((args, kwargs["env"])) or subprocess.CompletedProcess(args, 0, "", "")
+        ),
+    )
+
+    ensure_duckdb_driver()
+
+    assert tuple(args for args, _ in calls) == test_case.expected_installs
+    assert all(env["ADBC_DRIVER_PATH"] == str(driver_path) for _, env in calls)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    (
+        IngestrDriverFailureTestCase(
+            description="failed install names the duckdb version",
+            stderr="registry unreachable",
+            expected_message="duckdb 1.5.2: registry unreachable",
+        ),
+    ),
+    ids=lambda case: case.description,
+)
+def test_given_failing_driver_install_when_ensuring_then_names_the_duckdb_version(
+    test_case: IngestrDriverFailureTestCase,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    monkeypatch.setattr(f"{_RUNNER}.sys.prefix", str(tmp_path))
+    monkeypatch.setattr(f"{_RUNNER}.environ", {})
+    monkeypatch.setattr(f"{_RUNNER}.metadata.version", lambda name: "1.5.2")
+    monkeypatch.setattr(f"{_RUNNER}._dbc_executable", lambda: "dbc")
+    monkeypatch.setattr(
+        f"{_RUNNER}.subprocess.run",
+        lambda args, **kwargs: subprocess.CompletedProcess(args, 1, "", test_case.stderr),
+    )
+
+    with pytest.raises(IngestrIntegrationError, match=test_case.expected_message):
+        ensure_duckdb_driver()