diff --git a/CHANGELOG.md b/CHANGELOG.md index dd92998..2af3e5d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed + +- Schema evolution now declares missing tables in place via `add_managed_table` instead of recreating the managed database. Adding a table on a later run no longer snapshots, deletes, and reloads existing data — existing tables (including dlt bookkeeping) are left untouched. Requires `hotdata-framework>=0.6.0`. + ## [0.5.0] - 2026-06-29 diff --git a/README.md b/README.md index 4d979bd..79ff3e9 100644 --- a/README.md +++ b/README.md @@ -43,7 +43,7 @@ export HOTDATA_WORKSPACE=your_workspace_id That's it. On first run, the `sales` managed database is created automatically and the `orders` table is loaded. -`hotdata` is a native dlt destination (`JobClientBase` + `WithStateSync`): it supports nested/child tables, preserves dlt's internal columns (`_dlt_id`, `_dlt_load_id`), and persists schema-version, load, and pipeline-state tables in the managed database so **incremental sources resume correctly across runs**. If an existing managed database is missing a declared table on a later run, it is recreated with the union of existing and required tables (managed-database tables can only be declared at creation time); existing data is snapshotted and reloaded so nothing is lost. +`hotdata` is a native dlt destination (`JobClientBase` + `WithStateSync`): it supports nested/child tables, preserves dlt's internal columns (`_dlt_id`, `_dlt_load_id`), and persists schema-version, load, and pipeline-state tables in the managed database so **incremental sources resume correctly across runs**. If an existing managed database is missing a declared table on a later run, the table is added to it in place; existing tables and their data are left untouched. ## Feature support @@ -101,7 +101,7 @@ Where `hotdata` stands against the [dlt destination capability spec](https://dlt | dlt system tables (`_dlt_loads`, `_dlt_version`) | ✅ | Persisted in the managed database | | Pipeline state sync (`WithStateSync`) | ✅ | Incremental sources resume across runs | | New columns | ✅ | Permissive column promotion on append/merge | -| New tables | ⚠️ | Managed-DB tables are declared at creation; adding one triggers a data-preserving recreate (declare all in `declared_tables`) | +| New tables | ✅ | A table missing on a later run is declared in place on the existing database — no recreate, no data movement | | Multiple tables per pipeline | ✅ | Pass every table name via `declared_tables` | ### Staging, transactions & identifiers diff --git a/docs/architecture.md b/docs/architecture.md index 904bdef..b37f4a2 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -25,14 +25,14 @@ Provide a native `dlt` destination for Hotdata managed databases with explicit c - dlt internal columns (`_dlt_id`, `_dlt_load_id`) are preserved; no extra columns are added. - Schema versioning (`_dlt_version`), load tracking (`_dlt_loads`), and pipeline state (`_dlt_pipeline_state`) are persisted as managed tables, so `get_stored_state` lets incremental sources resume across runs. - Write dispositions: `replace`, `append`, `merge`, `upsert`, `insert-only`. Merge identity falls back to `_dlt_id` when no primary key is declared. -- `initialize_storage` declares the full table set (internal + user + schema tables) up front; if a later run needs a table the database lacks, `ensure_managed_database` recreates the database with the union of existing and required tables. Because tables can only be declared at creation time, the recreate snapshots every existing table first and reloads it afterward, so no data (including dlt bookkeeping) is lost. +- `initialize_storage` declares the full table set (internal + user + schema tables) up front; if a later run needs a table the database lacks, `ensure_managed_database` declares it in place via `add_managed_table`. Existing tables (including dlt bookkeeping) are left untouched and no data is moved. ## Ingestion flow 1. `dlt` writes a parquet load file and hands its path plus the `table` schema to `HotdataLoadJob`. 2. Contract mapping converts table metadata into `{database}.{schema}.{table}` naming. 3. Write disposition comes from the dlt table schema, falling back to the destination default. -4. Managed database is resolved or created (`ensure_managed_database`, with union-recreate on missing tables). +4. Managed database is resolved or created (`ensure_managed_database`, which declares any missing tables in place). 5. Load path uses only supported API operations: - `replace`: upload parquet batch and `load_managed_table(replace)` on the target - `append` / `merge` / `upsert` / `insert-only`: fetch existing target rows, combine in Python, then replace the target @@ -49,5 +49,5 @@ Provide a native `dlt` destination for Hotdata managed databases with explicit c ## Known limitations - Managed-table loads only support `mode=replace`; append/merge/upsert/insert-only are emulated via read-modify-write. -- Tables must be declared when the managed database is created; use `declared_tables` for multi-table pipelines. When a declared table is missing on a later run the database is recreated with the table union; existing data is snapshotted and reloaded, so the recreate is transparent (at the cost of re-uploading every table). +- Use `declared_tables` for multi-table pipelines so every table is declared up front. A table that is missing on a later run is added to the existing database in place (via `add_managed_table`) — no recreate and no data movement. - Read-modify-write reads the full target table on every non-replace batch. diff --git a/docs/runbook.md b/docs/runbook.md index e75a9ed..159a020 100644 --- a/docs/runbook.md +++ b/docs/runbook.md @@ -52,5 +52,5 @@ - `401` / `403`: verify `HOTDATA_API_KEY` and `HOTDATA_WORKSPACE`. - `404` on destination paths: verify `HOTDATA_API_BASE_URL` is the API host (for example `https://api.hotdata.dev`). - `429` / `5xx`: increase retry/backoff values. -- `table not declared`: recreate the managed database with all target tables in `declared_tables`, or declare them at create time. +- `table not declared`: list every target table in `declared_tables`. Missing tables are otherwise added to the database in place on the next run. - Append/merge loads re-read the full target table each batch; large tables may be slow until native append/merge API support lands. diff --git a/pyproject.toml b/pyproject.toml index 5ea97b0..85f7ead 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,7 +10,7 @@ requires-python = ">=3.11" dependencies = [ "dlt>=1.28.1", "hotdata>=0.5.0", - "hotdata-framework>=0.5.0", + "hotdata-framework>=0.6.0", "pandas>=2.0", "pyarrow>=14", ] diff --git a/src/hotdata_dlt_destination/hotdata_client.py b/src/hotdata_dlt_destination/hotdata_client.py index 5e0cce0..1f58aaf 100644 --- a/src/hotdata_dlt_destination/hotdata_client.py +++ b/src/hotdata_dlt_destination/hotdata_client.py @@ -1,26 +1,20 @@ from __future__ import annotations -import contextlib -import os -import tempfile - -import pyarrow as pa from hotdata_framework.databases import ManagedDatabase from hotdata_framework.managed_client import ManagedDatabaseClient -from hotdata_dlt_destination.parquet import write_table_parquet - class HotdataClient(ManagedDatabaseClient): """Managed-database client used by the dlt destination. Adds cross-run schema evolution on top of the shared ``hotdata_framework`` - client. Managed-database tables can only be declared at creation time, so - when an existing database is missing a required table the database must be - recreated. To avoid losing data (including dlt's ``_dlt_version`` / - ``_dlt_loads`` / ``_dlt_pipeline_state`` bookkeeping, which powers state - sync), every existing table is snapshotted, the database is recreated with - the union of existing and required tables, and the snapshots are reloaded. + 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. """ def ensure_managed_database( @@ -52,27 +46,16 @@ def ensure_managed_database( lambda: runtime.list_managed_tables(name, schema=schema) ) } - if not set(tables) - existing: - return db + # Declare any newly-required tables additively, in place. 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) + return db - # Snapshot existing data before the destructive recreate so no rows are - # lost when a new table is added on a later run. - all_tables = sorted(existing | set(tables)) - snapshots: dict[str, pa.Table] = {} - for table in sorted(existing): - data = self.fetch_table(database=name, schema=schema, table=table) - if data is not None and data.num_rows: - snapshots[table] = data - - self._request_with_retry(lambda: runtime.delete_managed_database(db.id)) - new_db = self._request_with_retry( - lambda: runtime.create_managed_database( - description=name, schema=schema, tables=all_tables - ) - ) - for table, data in snapshots.items(): - self._reload_table(name, table, schema=schema, data=data) - return new_db + def _add_managed_table(self, name: str, table: str, *, schema: str) -> None: + runtime = self._runtime + self._request_with_retry(lambda: runtime.add_managed_table(name, table, schema=schema)) def drop_managed_database(self, name: str) -> None: """Delete the managed database if it exists (used for dlt dev_mode / refresh).""" @@ -83,16 +66,5 @@ def drop_managed_database(self, name: str) -> None: return self._request_with_retry(lambda: runtime.delete_managed_database(db.id)) - def _reload_table(self, database: str, table: str, *, schema: str, data: pa.Table) -> None: - with tempfile.NamedTemporaryFile(suffix=".parquet", delete=False) as handle: - path = handle.name - try: - write_table_parquet(data, path) - upload_id = self.upload_parquet(path) - self.load_managed_table(database, table, schema=schema, upload_id=upload_id) - finally: - with contextlib.suppress(OSError): - os.unlink(path) - __all__ = ["HotdataClient"] diff --git a/src/hotdata_dlt_destination/job_client.py b/src/hotdata_dlt_destination/job_client.py index d43ffca..7190a1a 100644 --- a/src/hotdata_dlt_destination/job_client.py +++ b/src/hotdata_dlt_destination/job_client.py @@ -234,9 +234,9 @@ def is_storage_initialized(self) -> bool: return False def drop_storage(self) -> None: - # dlt calls this on dev_mode / refresh to wipe the destination. Managed-database - # tables can only be declared at creation time, so drop the whole database; - # initialize_storage recreates it on the next run. + # dlt calls this on dev_mode / refresh to wipe the destination. Dropping the + # whole managed database is the simplest correct reset; initialize_storage + # recreates it on the next run. try: with _hotdata_api(self.config) as api: api.drop_managed_database(self.config.database_name) diff --git a/tests/test_client.py b/tests/test_client.py index f14f7ee..f50860a 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,6 +1,5 @@ from types import SimpleNamespace -import pyarrow as pa import pytest from hotdata_dlt_destination.hotdata_client import HotdataClient @@ -204,6 +203,7 @@ def __init__(self, existing_db, existing_tables) -> None: self._existing_tables = list(existing_tables) self.created: list[tuple] = [] self.deleted: list[str] = [] + self.added: list[tuple] = [] self.uploaded: list[str] = [] self.loaded: list[tuple] = [] @@ -215,6 +215,11 @@ def resolve_managed_database(self, name): def list_managed_tables(self, name, *, schema): return [SimpleNamespace(table=t) for t in self._existing_tables] + def add_managed_table(self, database, table, *, schema): + self.added.append((database, table, schema)) + self._existing_tables.append(table) + return SimpleNamespace(table=table, schema=schema) + def create_managed_database(self, *, description, schema, tables): self.created.append((description, schema, list(tables))) return SimpleNamespace(id="new_db") @@ -276,22 +281,21 @@ def test_ensure_noop_when_all_tables_present() -> None: client.close() -def test_ensure_union_recreate_preserves_data(monkeypatch) -> None: +def test_ensure_adds_missing_table_without_recreate() -> None: rt = _EvoRuntime(existing_db=SimpleNamespace(id="db_1"), existing_tables=["orders"]) client = _client_with(rt) - snapshot = pa.table({"id": [1], "_dlt_id": ["a"]}) - monkeypatch.setattr(client, "fetch_table", lambda *, database, schema, table: snapshot) client.ensure_managed_database( "db", schema="public", tables=["orders", "customers"], create_if_missing=True ) - # Recreated with the union, old database deleted, and the existing table's - # data reloaded into the new database (no data loss). - assert rt.created == [("db", "public", ["customers", "orders"])] - assert rt.deleted == ["db_1"] - assert rt.loaded == [("db", "orders", "public", "up_1")] - assert len(rt.uploaded) == 1 + # The missing table is declared in place; the database is never deleted or + # recreated and no data is moved. + assert rt.added == [("db", "customers", "public")] + assert rt.deleted == [] + assert rt.created == [] + assert rt.uploaded == [] + assert rt.loaded == [] client.close() diff --git a/tests/test_e2e_inmemory.py b/tests/test_e2e_inmemory.py index b3e4b30..b0e700c 100644 --- a/tests/test_e2e_inmemory.py +++ b/tests/test_e2e_inmemory.py @@ -73,6 +73,11 @@ def delete_managed_database(self, name_or_id): for key in [k for k in self.tables if k[0] == name]: del self.tables[key] + def add_managed_table(self, database, table, *, schema): + name = self.id_to_name.get(database, database) + self.declared.setdefault(name, set()).add(table) + return SimpleNamespace(table=table, var_schema=schema, synced=False) + def upload_parquet(self, path): self._n += 1 uid = f"up_{self._n}" @@ -248,8 +253,9 @@ def customers(): pipelines_dir=str(tmp_path / "r1"), ).run(orders()) assert len(backend.rows("e2e_evo", "orders")) == 1 + db_id_before = backend.name_to_id["e2e_evo"] - # Adding a new table triggers the union-recreate; existing data must survive. + # Adding a new table declares it in place (no recreate); existing data survives. dlt.pipeline( pipeline_name="p_evo", destination=_dest("e2e_evo", ["orders", "customers"]), @@ -257,6 +263,8 @@ def customers(): pipelines_dir=str(tmp_path / "r2"), ).run([customers()]) + # The database was never deleted/recreated — same id throughout. + assert backend.name_to_id["e2e_evo"] == db_id_before orders_after = backend.rows("e2e_evo", "orders") assert orders_after is not None and len(orders_after) == 1 and orders_after[0]["id"] == 1 assert backend.rows("e2e_evo", "customers") is not None diff --git a/uv.lock b/uv.lock index 90eb7ee..8f99023 100644 --- a/uv.lock +++ b/uv.lock @@ -301,7 +301,7 @@ dev = [ requires-dist = [ { name = "dlt", specifier = ">=1.28.1" }, { name = "hotdata", specifier = ">=0.5.0" }, - { name = "hotdata-framework", specifier = ">=0.5.0" }, + { name = "hotdata-framework", specifier = ">=0.6.0" }, { name = "pandas", specifier = ">=2.0" }, { name = "pyarrow", specifier = ">=14" }, ] @@ -315,16 +315,16 @@ dev = [ [[package]] name = "hotdata-framework" -version = "0.5.0" +version = "0.6.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "hotdata" }, { name = "pandas" }, { name = "pyarrow" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/72/04/cad9fafac70f0f7b851329b4b3c1824ddbd07c704015cc5d91840e19d73c/hotdata_framework-0.5.0.tar.gz", hash = "sha256:323a69e7beb8fcab35814fad090b0203f3b7b5a857816b93cab09ef081996f78", size = 93962, upload-time = "2026-06-29T07:01:28.25Z" } +sdist = { url = "https://files.pythonhosted.org/packages/3d/97/821fbacc5d22ad5ce3d0d2db995b53188000821efd135342c59c4db7334a/hotdata_framework-0.6.0.tar.gz", hash = "sha256:99303170ee2d0bef2e440680dfa3232ed317f4333cfc7642624e2be04dc76c7f", size = 94468, upload-time = "2026-06-30T17:19:23.711Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/cf/5c/1a7ce4125a1332aaf9a0fd19fd1e659815fa6b83174c040bff8b53acaf3f/hotdata_framework-0.5.0-py3-none-any.whl", hash = "sha256:280e651d261f314e894bcbd7fb65135af100c02537f765d9877e013d2c16c257", size = 14078, upload-time = "2026-06-29T07:01:26.806Z" }, + { url = "https://files.pythonhosted.org/packages/ed/c5/e02b08f8527eb4a020d23bf61bf0f61fc163a604d5da5a01fde3360c3eac/hotdata_framework-0.6.0-py3-none-any.whl", hash = "sha256:a38d28ba8efa8b6869f32604e2440fcff84435415af262306dc0f7da0aee2886", size = 14292, upload-time = "2026-06-30T17:19:22.425Z" }, ] [[package]]