fix: collision-safe, resolve-once managed-database addressing (#39)

CHANGELOG.md

@@ -7,14 +7,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
-### Fixed
-
-- The API key now resolves from `HOTDATA_API_KEY` following the README quickstart: setting the env var (with no `credentials=` argument) populates the destination, instead of leaving `credentials.api_key` unset and failing with an opaque `NoneType` error deep in `hotdata-framework`. Missing `api_key`/`workspace_id` now raises a clear `ConfigurationValueError` at setup naming the missing field.
-
 ### Changed
 
 - **Breaking:** `workspace_id` moved out of `HotdataCredentials` (authentication) to a top-level `hotdata(workspace_id=...)` param / `HotdataClientConfiguration` field (configuration), matching the SDK's `Configuration(api_key=, workspace_id=)` shape. It is a **param with no environment-variable fallback** — the `HOTDATA_WORKSPACE` env var is no longer read on this path (the API key remains env-backed, as a secret). Passing `workspace_id` inside a `credentials={...}` dict still works but is deprecated (hoisted with a `DeprecationWarning`); constructing `HotdataCredentials(workspace_id=...)` now raises `TypeError` — pass `workspace_id=` to `hotdata(...)` instead.
 
+### Fixed
+
+- The API key now resolves from `HOTDATA_API_KEY` following the README quickstart: setting the env var (with no `credentials=` argument) populates the destination, instead of leaving `credentials.api_key` unset and failing with an opaque `NoneType` error deep in `hotdata-framework`. Missing `api_key`/`workspace_id` now raises a clear `ConfigurationValueError` at setup naming the missing field.
+- Managed-database name resolution is now collision-safe and resolved once per run. Hotdata database names are not unique, and the destination previously took the first `list_databases` match on every operation — so a name collision could silently read from, write to, or **drop** the wrong database. It now raises a clear error when a name matches more than one database, resolves the name to its record a single time per run (cached on the shared config), and addresses every subsequent operation (load/add/list/query/drop) by id.
+
 
 ## [0.10.0] - 2026-07-20
 

src/hotdata_dlt_destination/hotdata_client.py

@@ -5,20 +5,82 @@
 from hotdata_framework.databases import ManagedDatabase
 from hotdata_framework.managed_client import ManagedDatabaseClient
 
+from hotdata_dlt_destination.errors import HotdataTerminalError
+
 
 class HotdataClient(ManagedDatabaseClient):
     """Managed-database client used by the dlt destination.
 
-    Adds cross-run schema evolution on top of the shared ``hotdata_framework``
-    client. The base client only creates a managed database with its initial
-    tables; this override additionally reconciles tables on an already-existing
-    database. When a later run requires a table that the database is missing,
-    the table is declared in place via ``add_managed_table`` (the table is added
-    empty and populated by the subsequent load) — no data is moved and existing
-    tables, including dlt's ``_dlt_version`` / ``_dlt_loads`` /
-    ``_dlt_pipeline_state`` bookkeeping, are left untouched.
+    Adds two things on top of the shared ``hotdata_framework`` client:
+
+    * **Cross-run schema evolution** — when a later run requires a table the
+      database is missing, the table is declared in place via
+      ``add_managed_table`` (added empty, populated by the subsequent load); no
+      data is moved and existing tables, including dlt's ``_dlt_version`` /
+      ``_dlt_loads`` / ``_dlt_pipeline_state`` bookkeeping, are left untouched.
+    * **Collision-safe, resolve-once addressing** — a database name is resolved
+      to its record once per run (cached via :meth:`bind_run_cache`) and every
+      subsequent operation addresses the database by id. Resolution raises on an
+      ambiguous name instead of silently taking the first match.
     """
 
+    # Run-scoped store bound via bind_run_cache(); resolution is cached on it so
+    # the whole run reuses one resolved/created record.
+    _run_cache: object | None = None
+
+    def bind_run_cache(self, cache: object) -> None:
+        """Bind a run-scoped store so a database resolves to its record once.
+
+        ``cache`` is any object that tolerates a ``_hotdata_resolved_db``
+        attribute — in practice the shared ``HotdataClientConfiguration``
+        instance, which every client built for a run points at.
+        """
+        self._run_cache = cache
+
+    # --- resolution -------------------------------------------------------
+
+    def _collision_safe_resolve(self, name_or_id: str) -> ManagedDatabase:
+        """Resolve a name/id to its record, raising on an ambiguous name.
+
+        Hotdata database names are not unique. Taking the first match can read,
+        write, or drop the wrong database, so a name that matches more than one
+        database raises instead. An id (matched exactly) is unambiguous.
+        """
+        databases = self._request_with_retry(self._runtime.list_managed_databases)
+        by_name = [db for db in databases if db.description == name_or_id]
+        if len(by_name) > 1:
+            raise HotdataTerminalError(
+                f"Managed database name {name_or_id!r} is ambiguous: "
+                f"{len(by_name)} databases share it (ids: {sorted(db.id for db in by_name)}). "
+                "Address it by id to disambiguate."
+            )
+        if by_name:
+            return by_name[0]
+        by_id = [db for db in databases if db.id == name_or_id]
+        if by_id:
+            return by_id[0]
+        raise KeyError(name_or_id)
+
+    def _resolve(self, name_or_id: str) -> ManagedDatabase:
+        """Resolve once per run, then serve the cached (id-addressable) record."""
+        cache = self._run_cache
+        if cache is not None:
+            cached = getattr(cache, "_hotdata_resolved_db", None)
+            if cached is not None and name_or_id in (
+                cached.id,
+                getattr(cached, "description", None),
+            ):
+                return cached
+        db = self._collision_safe_resolve(name_or_id)
+        self._cache_db(db)
+        return db
+
+    def _cache_db(self, db: ManagedDatabase | None) -> None:
+        if self._run_cache is not None:
+            self._run_cache._hotdata_resolved_db = db
+
+    # --- lifecycle --------------------------------------------------------
+
     def ensure_managed_database(
         self,
         name: str,
@@ -30,72 +92,72 @@ def ensure_managed_database(
     ) -> ManagedDatabase:
         # keys: table name -> key columns (enables delete/update/upsert on it)
         keys = keys or {}
-        runtime = self._runtime
 
-        # Resolve is called directly (not via _request_with_retry) so its KeyError
-        # "not found" signal is preserved rather than mapped to a terminal error.
         try:
-            db = runtime.resolve_managed_database(name)
+            db = self._resolve(name)
         except KeyError:
             if not create_if_missing:
                 raise
-            return self._request_with_retry(
-                lambda: runtime.create_managed_database(
+            db = self._request_with_retry(
+                lambda: self._runtime.create_managed_database(
                     description=name, schema=schema, tables=sorted(set(tables)), keys=keys
                 )
             )
+            self._cache_db(db)
+            return db
 
         existing = {
             managed_table.table
             for managed_table in self._request_with_retry(
-                lambda: runtime.list_managed_tables(name, schema=schema)
+                lambda: self._runtime.list_managed_tables(db.id, schema=schema)
             )
         }
         # Declare any newly-required tables additively, in place, carrying their
         # key. dlt calls ``initialize_storage`` with the full table set before any
         # load job runs, so by load time this is normally a no-op.
         for table in sorted(set(tables) - existing):
-            self._add_managed_table(name, table, schema=schema, key=keys.get(table))
+            self._add_managed_table(db.id, table, schema=schema, key=keys.get(table))
         return db
 
     def _add_managed_table(
-        self, name: str, table: str, *, schema: str, key: list[str] | None = None
+        self, database: str, table: str, *, schema: str, key: list[str] | None = None
     ) -> None:
-        runtime = self._runtime
         self._request_with_retry(
-            lambda: runtime.add_managed_table(name, table, schema=schema, key=key)
+            lambda: self._runtime.add_managed_table(database, table, schema=schema, key=key)
         )
 
     def drop_managed_database(self, name: str) -> None:
         """Delete the managed database if it exists (used for dlt dev_mode / refresh)."""
-        runtime = self._runtime
         try:
-            db = runtime.resolve_managed_database(name)
+            db = self._resolve(name)
         except KeyError:
             return
-        self._request_with_retry(lambda: runtime.delete_managed_database(db.id))
+        self._request_with_retry(lambda: self._runtime.delete_managed_database(db.id))
+        self._cache_db(None)
 
     def resolve_managed_database(self, name: str) -> ManagedDatabase:
-        """Resolve a managed database by display name to its record (carrying ``.id``).
+        """Resolve a managed database by display name (or id) to its record.
 
-        Delegates to the runtime client, preserving its ``KeyError`` "not found" signal.
+        Raises ``KeyError`` when nothing matches and ``HotdataTerminalError`` when
+        the name is shared by more than one database.
         """
-        return self._runtime.resolve_managed_database(name)
+        return self._resolve(name)
+
+    def load_managed_table(self, database: str, table: str, **kwargs):
+        """Load parquet into a managed table, addressing the database by id."""
+        db = self._resolve(database)
+        return super().load_managed_table(db.id, table, **kwargs)
 
     def execute_sql(self, sql: str, *, database: str) -> pa.Table:
         """Run a SQL query scoped to ``database`` and return the result as Arrow.
 
-        The read/dataset interface goes through here. The base client has no
-        general query entrypoint of its own — only the private database-scoped
-        submit + Arrow fetch that :meth:`fetch_table` uses — so this mirrors that
-        dance for arbitrary SQL: resolve the managed database name to its id,
-        submit the query, poll until the result is ready, and fetch it as a
-        ``pyarrow.Table``. An empty table is returned when the query produces no
-        out-of-band result (e.g. a statement with no result set).
+        Resolves the managed database to its id (once per run), submits the query,
+        polls until the result is ready, and fetches it as a ``pyarrow.Table``. An
+        empty table is returned when the query produces no out-of-band result.
         """
 
         def operation() -> pa.Table:
-            db = self._runtime.resolve_managed_database(database)
+            db = self._resolve(database)
             result_id = self._query_database_scoped(sql, database_id=db.id)
             if result_id is None:
                 return pa.table({})
@@ -109,10 +171,38 @@ def operation() -> pa.Table:
 
     def list_managed_tables(self, database: str, *, schema: str) -> list:
         """List the managed tables in ``database``/``schema`` (used by ``has_dataset``)."""
-        runtime = self._runtime
+        db = self._resolve(database)
         return self._request_with_retry(
-            lambda: runtime.list_managed_tables(database, schema=schema)
+            lambda: self._runtime.list_managed_tables(db.id, schema=schema)
         )
 
+    def table_is_synced(self, database: str, table: str, *, schema: str) -> bool:
+        db = self._resolve(database)
+        for managed_table in self._request_with_retry(
+            lambda: self._runtime.list_managed_tables(db.id, schema=schema)
+        ):
+            if managed_table.table == table:
+                return managed_table.synced
+        return False
+
+    def fetch_table(self, *, database: str, schema: str, table: str) -> pa.Table | None:
+        def operation() -> pa.Table | None:
+            db = self._resolve(database)
+            if not self._table_is_synced_for(db, table, schema=schema):
+                return None
+            sql = f'SELECT * FROM "default"."{schema}"."{table}"'
+            result_id = self._query_database_scoped(sql, database_id=db.id)
+            if result_id is None:
+                return None
+            return self._fetch_result_arrow(result_id, database_id=db.id)
+
+        return self._request_with_retry(operation)
+
+    def _table_is_synced_for(self, db: ManagedDatabase, table: str, *, schema: str) -> bool:
+        for managed_table in self._runtime.list_managed_tables(db.id, schema=schema):
+            if managed_table.table == table:
+                return managed_table.synced
+        return False
+
 
 __all__ = ["HotdataClient"]

src/hotdata_dlt_destination/job_client.py

@@ -107,6 +107,8 @@ def _hotdata_api(config: HotdataClientConfiguration) -> Iterator[HotdataClient]:
         max_retries=config.max_retries,
         retry_backoff_seconds=config.retry_backoff_seconds,
     )
+    # Share the run's resolved-database cache across every short-lived client.
+    api.bind_run_cache(config)
     try:
         yield api
     finally:

src/hotdata_dlt_destination/sql_client.py

@@ -177,6 +177,7 @@ def open_connection(self) -> HotdataClient:
             max_retries=self._config.max_retries,
             retry_backoff_seconds=self._config.retry_backoff_seconds,
         )
+        self._client.bind_run_cache(self._config)
         return self._client
 
     def close_connection(self) -> None: