diff --git a/README.md b/README.md index 0f88719..addb1fd 100644 --- a/README.md +++ b/README.md @@ -238,7 +238,7 @@ To pin to one specific release instead, use the exact tag the live badges under [Releasing a new version](#releasing-a-new-version)): ```python -%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.6" +%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.10" # staging's latest release (early access) %pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.10-staging" diff --git a/docs/dev-notes.md b/docs/dev-notes.md new file mode 100644 index 0000000..04f01d5 --- /dev/null +++ b/docs/dev-notes.md @@ -0,0 +1,923 @@ +# Dev notes — findings & ideas + +Scratchpad for observations and ideas about `eea_datalakehouse`, collected while +the development team is away. **Nothing here has been applied to the code** — +every entry is a proposal to discuss when the team is back. + +Started: 2026-08-18 · Branch at time of writing: `chore/tidy-dds-ingestion-copyover` + +## How to use this file + +- One entry per finding, newest at the bottom of its section. +- Keep entries small and self-contained: what was observed, why it matters, what + we'd suggest. No entry implies a decision has been made. +- Status: `open` (needs discussion) · `agreed` (team said yes, not yet done) · + `rejected` (discussed, deliberately not doing) · `done` (implemented elsewhere). + +### Entry template + +```markdown +### + +- **Status:** open +- **Where:** `path/to/file.py:42` +- **Observation:** what is actually there today. +- **Why it matters:** the concrete consequence. +- **Suggestion:** the smallest change that would address it. +- **Open question:** anything we need the team to decide. +``` + +## Findings + +### Dangling link to `docs/python-client-guide.md` + +- **Status:** open +- **Where:** `src/eea_datalakehouse/dds_ingestion/README.md` (last line of + "Managing the transfer") +- **Observation:** the README points at + [`docs/python-client-guide.md`](../../docs/python-client-guide.md) as the + "Full class reference". That file does not exist in the repository, and + `git log` shows it was never committed and later deleted — it has simply never + been there. +- **Why it matters:** the link is the only pointer to a full API reference for + `FolderIngest`, so a notebook author following the README hits a 404 on + GitHub. It also reads as if documentation exists that nobody can find. +- **Suggestion:** either write the guide, or drop the link and inline the short + method list that is already in that README. +- **Open question:** was the guide drafted somewhere outside this repo (wiki, + Confluence, the DDS repo) and just needs copying over? + +### The API is shaped for a developer, not for a custodian writing a one-off script + +- **Status:** open +- **Where:** `src/eea_datalakehouse/catalog/` (on `development`/`main` — **not** on + `chore/tidy-dds-ingestion-copyover`, so all line numbers below are as of + `development`), plus `pyproject.toml` +- **Premise:** the people writing this code are data custodians, working in a + command-line / batch-scripting style — short scripts, run once, per dataflow. + The library currently asks them to work like application developers instead. + +The concrete things that force the developer style, numbered below. Each is +small on its own; together they are why a five-line job takes a page of Python. +Sub-sections `1a`, `2a` and `7` are proposals rather than observations. + +#### 1. Credentials are threaded through every call site — and don't need to be + +`Catalog.__init__` (`catalog/client.py:125-136`) takes `base_url` and `token` as +required positional arguments, plus an optional `username`. Nothing resolves +them: every caller supplies all three, on every construction. + +The library's own debug harness shows the cost — `debugger/debug_run.py` repeats + +```python +catalog = Catalog(DREMIO_BASE_URL, DREMIO_TOKEN, username=DREMIO_USERNAME) +``` + +at **20+ separate call sites** (lines 143, 159, 186, 210, 220, 235, 245, 252, +259, 280, 290, 311, 330, 341, 353, 368, 387, 401, 414, …). If the authors' own +script looks like that, a custodian's will too. + +**The machinery to avoid this already exists in this same library, twice:** + +- `dds_ingestion/folder.py:136-143` — `FolderIngest` resolves its own + credentials from the environment when the caller doesn't inject a client, via + `load_creds()` / `load_base_url()` (`dds_ingestion/credentials.py:48-76`). + So one half of the library auto-resolves and the other half refuses to. +- `dds_ingestion/common/dremio_identity.py` — a full identity resolver + (`resolve()`, `endpoint()`, `dds_credentials()`) that already knows the + precedence a custodian needs: the JupyterLab "Dremio Catalog" settings panel + first, then whatever `%init` bound into the kernel, then the process + environment including the local `_DREMIO_USER` / `_DREMIO_PWD` / + `DREMIO_BASE_URL` aliases (`dremio_identity.py:177-205`, `242-260`, + `115-119`). It resolves exactly the four values `Catalog` demands: + `DREMIO_USERNAME`, `DREMIO_TOKEN`, `DREMIO_URL`, `DDS_BASE_URL`. + +So: **no, credentials do not need to be passed.** `Catalog` simply never calls +the resolver that the package next door already ships. + +- **Suggestion:** make every argument to `Catalog()` optional and fall back to + `dremio_identity.resolve()`, so `Catalog()` with no arguments works in a Hub + kernel, in the local stack, and in a cron job — with explicit arguments still + winning when someone needs to override. Same treatment for the `DREMIO_URL` + vs `DDS_BASE_URL` split, which a custodian should never have to think about. +- **Open question:** is `dremio_identity` deliberately kept inside + `dds_ingestion.common` (i.e. ingest-only), or should it move up to + `eea_datalakehouse.common` and become the one identity path for ingest, + catalog and anything later? + +#### 1a. Answering it directly: yes, `%init` is enough — with two gaps + +A real `%init` in this environment exposes: + +``` +DREMIO_USERNAME adm_bliki +DREMIO_TOKEN set (hidden, 64 chars) +DREMIO_URL https://dremio.eea.europa.eu:9047 +DREMIO_PASSWORD (not defined) +DDS_BASE_URL http://localhost:8000/ +SCHEDULER_URL http://gpu02.pdmz.eea:8080/scheduler +``` + +Lined up against what `Catalog(base_url, token, username=...)` demands: +`DREMIO_URL` → `base_url`, `DREMIO_TOKEN` → `token`, `DREMIO_USERNAME` → +`username`. **All three are already there.** Nothing needs to be typed, and +`dremio_identity.resolve()` already returns exactly this dict +(`BOUND_VARS = SECRET_VARS + SERVICE_VARS`, `dremio_identity.py:95-109`). + +`DREMIO_PASSWORD (not defined)` is not a blocker: the Flight handshake uses +`authenticate_basic_token(username, token)` — the PAT as the password, not +`DREMIO_PASSWORD` (`catalog/sql.py:255-262`, `319-321`). And the trailing slash +on `DDS_BASE_URL` is harmless — `RestSqlExecutor`, `CatalogRestClient` and +`IngestClient` all `rstrip("/")` their base URL. + +Two things do need deciding before an argument-less `Catalog()` would work +everywhere: + +**Gap 1 — `resolve()` does not read the kernel globals `%init` binds.** +`endpoint()` looks in three places: the settings panel, then `ip.user_ns` (what +`%init` bound), then the environment (`dremio_identity.py:187-201`). +`resolve()` looks in only two: the panel and the environment +(`dremio_identity.py:242-260`) — no `ip.user_ns`. It works today because the Hub +injects those variables into the *environment* and `%init` merely re-exports +them as globals, so the environment lookup happens to find the same values. But +anything that exists only as a kernel global — a value the extension binds +without exporting, or one a user reassigns in a cell — is invisible to +`resolve()` while being visible to `endpoint()`. Two resolvers, two different +precedence chains. + +**Gap 2 — `%init` exposes no Flight location, and `datacopy` is Flight-only.** +`datacopy` / `datamove` always run over Arrow Flight (`catalog/client.py:49-51`, +`205-245`), which needs a `grpc://host:port`, not the REST URL. None of the six +`%init` variables carries one. With nothing given, `_default_flight_location` +derives it from `DREMIO_URL`: `https://dremio.eea.europa.eu:9047` → +`grpc+tls://dremio.eea.europa.eu:32010` (`catalog/sql.py:69-87`) — and that +code's own comment says port 32010 is Dremio's documented default, +"unverified against this project's actual deployment". So making credentials +automatic would make `datacopy` silently depend on a guess; if the guess is +wrong it fails as `Socket closed`, which reads like a network fault rather than +a misconfiguration. + +- **Suggestion:** one resolver, used by everything. Fold `endpoint()`'s + three-source precedence into `resolve()` so both agree, then let `Catalog()` + and `FolderIngest()` default every connection argument from it — explicit + arguments still win. Add `DREMIO_FLIGHT_LOCATION` + (`catalog/sql.py:43`) to whatever `%init` exports, or verify 32010 for this + deployment and drop the "unverified" caveat. +- **Open question:** is `%init` ours to extend? If the variable list is owned by + the `jupyter_dremio` Hub extension, adding the Flight location is a + cross-repo change, and until then the library needs a documented default. + +#### 2. `idempotency_key` is mandatory everywhere and reaches nothing + +Every one of the 19 `Catalog` methods declares `*, idempotency_key: str` with +**no default** (`catalog/client.py:174-332`) — including the pure reads: +`gettablesfrom`, `gettableitemsfrom`, `getwikifrom`, `gettagsfrom`, +`getmetafromwiki`. + +Two observations about what that key actually does: + +- **It is never sent to Dremio.** The REST executor posts `{"sql": sql}` and + nothing else (`catalog/sql.py:165`); no header carries it. Its only job is to + be a lookup key in a local JSON file at + `~/.cache/eea_datalakehouse/catalog_retry_state.json` + (`catalog/retry_state.py:23-26`), written only when a call raises + `EngineStartingError`. +- **The executor layer already treats it as optional** — + `SqlExecutor.execute(sql, *, idempotency_key: str | None = None)` + (`catalog/sql.py:62`). The layer users touch is stricter than the layer that + consumes it. + +The predictable result is in the example you sent: `idempotency_key="343242dsrew"`. +A keyboard mash is the rational response to a required argument whose purpose +isn't visible at the call site — but it is also the handle you would need to +type back into `catalog.retry_pending(...)` later, so the mash quietly +forfeits the one feature the argument exists for. + +The name also over-promises. For `datacopy`, re-running with the same key does +**not** make the operation a no-op: with the default `overwrite=False` the second +run raises `CatalogOperationError: target ... already exists — pass +overwrite=True to replace it` (`catalog/operations.py:402-408`). "Idempotency +key" reads as "safe to re-run"; it isn't. + +- **Suggestion:** make it optional and fill it in behind the scenes — see 2a. +- **Open question:** was a server-side idempotency contract ever intended (a + header Dremio or DDS would honour), or is local retry bookkeeping the whole + design? + +#### 2a. Proposed: make `idempotency_key` optional and prefill it + +**Position to put to the team** (proposed here, not yet discussed with them): +the idempotency key is a sound concept for larger, multi-step transactions — +that is where knowing "this is the same unit of work as before" earns its +keep — but it is overkill for one-time execution code. It should be an +**optional** parameter, filled in behind the scenes when the caller doesn't +supply one. + +The signature change is backwards compatible and stops at the operations layer: +`idempotency_key: str | None = None` on the 19 `Catalog` methods and their +`operations.*` counterparts. Nothing below needs touching — `SqlExecutor` +already declares it optional (`catalog/sql.py:62`), and every existing caller +that passes a key keeps working unchanged. + +**How to prefill — the one real design choice.** + +*Option A: a fresh random key per call (uuid4).* Simplest to implement, but the +key is the handle you need for `catalog.retry_pending(key)`, and with a random +one the custodian never sees it. It would have to be surfaced somehow — returned +on the result, exposed as `catalog.last_idempotency_key`, or `retry_pending()` +with no argument meaning "the most recent pending one". Every stalled run also +leaves a *new* orphan entry in the retry-state file, which nothing prunes. + +*Option B (recommended): derive it deterministically* from the operation name +plus the arguments that identify the work — for `datacopy`, the resolved source +and target paths. Re-running the same script regenerates the same key, which +matters because **re-running the script is what a custodian actually does** after +a cold-engine stall. They never have to learn `retry_pending` exists: + +- first run stalls → `retry_state.record(...)` writes the derived key, attempts 1 + (`catalog/retry_state.py:82-91`); +- the custodian re-runs the same script → same derived key → attempts 2 under the + same entry, not a second orphan; +- it succeeds → `retry_state.clear(key)` removes it + (`catalog/operations.py`, end of `_run_steps` / `_run_actions`). + +The retry-state file stays bounded by the number of distinct operations, not by +the number of attempts. + +One caution worth stating so nobody trips on it later: with a derived key, two +genuinely separate runs with identical arguments (copy, delete the target, copy +again) share a key. Since the key is only local bookkeeping and is cleared on +success, that is harmless — but it is a real consequence of B and should be a +deliberate choice, not a surprise. + +**Where the parameter should stay explicit** + +- The read-only operations should not take one at all — `gettablesfrom`, + `gettableitemsfrom`, `getwikifrom`, `gettagsfrom`, `getmetafromwiki`. There is + nothing to resume; today `_fetch_step` records them anyway. +- Genuinely multi-step operations keep the explicit parameter available, because + that is the case the concept was designed for — notably `datamove`, whose own + module docstring flags the CREATE-succeeded-then-DROP-stalled gap that a blind + retry cannot resolve (`catalog/operations.py:30-37`). +- Anything a scheduler drives, where the caller wants a stable handle it chose + itself rather than one the library derived. + +- **Open question:** should `retry_state` gain an age-out or a cap regardless? + `list_pending()` exists (`catalog/retry_state.py:114`) but nothing ever prunes + the file, and with auto-filled keys entries will be created more often than + they are today. + +#### 3. There is no command line + +No `[project.scripts]` in `pyproject.toml` (neither branch), no `__main__.py`, +and no `argparse` / `click` / `typer` anywhere under `src/`. "Command-line / +batch scripting style" therefore means: write a Python file, import the right +class, construct it with credentials, call a method, print the result. + +The good news is the groundwork is already right: `catalog/operations.py` +exposes every operation as a plain module-level function taking an executor +(`operations.table2view(executor, ...)`, `datacopy(executor, ...)`, …), and +`Catalog`'s methods are thin delegations to them (`client.py:174-332`). A CLI +would be an adapter over that layer, not a rewrite. + +- **Suggestion:** one console entry point, one subcommand per verb, arguments + in the same order as the Python functions: + + ```bash + eea-catalog datacopy SRC DST --create-target-folder + eea-catalog gettablesfrom bwd --json + eea-ingest ./my_data biodiversity.uploads --format parquet --parallelism 4 + ``` + + with credentials resolved per #1 (nothing on the command line), a non-zero + exit code on failure, and `--json` for anything a shell script needs to parse. +- **Open question:** do custodians run these on their own machine, in the + project container, or as a scheduled Hub notebook? That decides whether the + CLI or a still-simpler Python one-liner is the primary surface. + +#### 4. The verbs that match the custodians' vocabulary are the unimplemented ones + +`draft2version` and `publishversion` both raise `NotImplementedError` +(`catalog/operations.py:326` and `:340`). Those are precisely the two +domain-level actions in the draft → version → publish workflow. + +Your example is that exact workflow — promoting +`…bwd.draft.bw_assessment.assessments` to +`…bwd.versions.2025_6.bw_assessment.assessments` — done by hand with the +generic `datacopy` and two fully spelled-out paths. The custodian ends up +performing the plumbing that the domain verb was meant to hide. + +- **Suggestion:** treat `draft2version` / `publishversion` as the priority, and + let them own the path arithmetic (see #5): `draft2version("bw_assessment", + version="2025_6")` rather than two 7-segment strings. +- **Open question:** what is the intended semantics — is a version a physical + copy (today's `datacopy` behaviour), or a view repoint? `publishversion`'s + docstring says the consumer view repoint, but the draft→version step is + undecided in code. + +#### 5. Catalog paths are opaque strings, though the taxonomy is known + +`datacopy` takes two free-form dotted strings. In your example they are seven +segments long and differ only in the middle: + +``` +catalog.water_management_resources.bathing_water.bwd.draft.bw_assessment.assessments +catalog.water_management_resources.bathing_water.bwd.versions.2025_6.bw_assessment.assessments +``` + +Everything before and after `draft` / `versions.2025_6` is repeated by hand, +and nothing validates the shape — `_quote_path` just splits on `.` and quotes +each segment (`catalog/operations.py:90-96`). A typo in segment 2 surfaces as +a Dremio error, not a local one. + +Meanwhile the library already encodes this taxonomy elsewhere: +`dds_ingestion/common/catalog.py` documents the path as +`{source}/{domain}/{subdomain}/{dataflow}/...` and hard-codes +`DATAFLOW_DEPTH = 3`, refusing to create anything above that level. + +- **Suggestion:** a small path type or builder shared by both packages, so the + custodian names the dataflow and the stage and the library assembles the + string — and so a wrong number of segments fails locally, before any call. +- **Open question:** is `lakehouse_structure.md` (referenced from + `dds_ingestion/common/catalog.py`) the authoritative taxonomy? If so it could + drive validation directly. + +#### 6. Batch-unfriendly output + +- `logger = logging.getLogger("eea_datalakehouse.dds_ingestion")` + (`dds_ingestion/folder.py:39`) — no handler is configured anywhere in the + package, so in a plain script every `logger.info` about resuming or + re-uploading goes nowhere unless the custodian configures logging first. +- Progress is a tqdm bar (`dds_ingestion/progress.py:37-43`), sensible in a + notebook; in a cron log it writes carriage returns, and the no-tqdm fallback + prints one line per file. +- Results come back as Python objects (`IngestOutcome`, `SqlResult`, + `TableInfo`), so reporting what happened requires more Python. + +- **Suggestion:** as part of the CLI in #3 — a default log line per step to + stderr, results to stdout as JSON on request, and progress that detects a + non-TTY and degrades to periodic lines. + +#### 7. Proposed: a session context, so paths stop being absolute + +**Position to put to the team** (proposed here, not yet discussed with them): +let the user set the session "context" once — the SQL `USE ` idea — so +the library has a default schema to resolve short paths against. This is the +direct fix for #5, and it is what makes the command-line mental model work: +`cd` once, then use relative names. + +On your own example it collapses two 7-segment strings into two short ones: + +```python +catalog.use("catalog.water_management_resources.bathing_water.bwd") +catalog.datacopy("draft.bw_assessment.assessments", + "versions.2025_6.bw_assessment.assessments") +``` + +Everything the two paths had in common moves into the `use()` line, said once. +What is left is exactly what differs — which is the part the custodian is +actually thinking about. + +**The important finding: this cannot be Dremio's own context.** + +Dremio's `POST /api/v3/sql` does accept a `context` array alongside `sql`, and +the client currently sends only `sql` (`catalog/sql.py:165`) — so passing it +through looks like the obvious one-line implementation. It isn't, because only +one of the three ways this library names things would be affected by it: + +1. **SQL DDL** — `CREATE TABLE {target} AS SELECT * FROM {source}` + (`catalog/operations.py:419-424`). A server-side context *would* apply here. +2. **INFORMATION_SCHEMA lookups** — `_entry_exists` and `_entry_kind` + (`catalog/operations.py:190-232`) and `gettablesfrom` + (`catalog/operations.py:613-650`) compare `"TABLE_SCHEMA"` against the path + as a **string literal**. `TABLE_SCHEMA` always holds the full dotted path, so + a session context does not rewrite it — a relative path simply fails to + match, `_entry_exists` returns `False`, and `datacopy` reports + `source ... does not exist` for a source that is right there. Silent and + confusing. +3. **Dremio's catalog REST API** — `_lookup_by_path` does + `"/".join(path.split("."))` against `/api/v3/catalog/by-path/…` + (`catalog/rest.py:55-60`) and `create_folder` posts `path.split(".")` + (`catalog/rest.py:261`). That API has no context concept at all; it needs the + absolute path. Folders, wiki and tags all go through it. + +So the context has to be **resolved client-side** — joined onto the front of a +relative path before anything else happens — which is also the better answer: +it keeps one code path for all three mechanisms, it works identically over REST +and Flight, and it lets a wrong path fail locally instead of as a Dremio error. +There is already precedent for the library resolving paths on the caller's +behalf: `_resolve_target_path` implements `cp source dest/` semantics +(`catalog/operations.py:234-249`). + +**Where the context should come from** — the same three-source pattern as the +credentials in #1, so a script can set it without touching code: + +- `catalog.use("…")` — mirrors the SQL the custodians already know, and can be + called more than once in a script that touches two dataflows; +- `Catalog(context="…")` — for the one-shot case; +- an environment variable (say `EEA_CATALOG_CONTEXT`), so a batch job sets it + alongside the credentials and the script body carries no paths at all. Worth + asking whether `%init` should export it too. + +**Where to put the boundary.** `dds_ingestion/common/catalog.py` already draws +one: `DATAFLOW_DEPTH = 3` — segments 0-2 (`source` / `domain` / `subdomain`) +must already exist and are never created, and from index 3 down is the +dataflow's own space. That gives a principled default context of the first +three segments (`catalog.water_management_resources.bathing_water`), leaving +`bwd.draft.…` relative. Your example suggests the more useful day-to-day +boundary is one deeper — the dataflow itself (`….bathing_water.bwd`) — since +that is what the two paths actually share. Both are defensible; the team should +pick one rather than leaving it to each script. + +**The one thing that must not be fudged: relative vs absolute.** Guessing — +"try it relative, fall back to absolute" — is the wrong answer here, because +the failure mode is silent misresolution into a real but wrong location. Most +operations would merely error, but `deletefolder(cascade=True)` +(`catalog/rest.py:340`) deletes a subtree depth-first, and `createfolder` +would quietly build a new one in the wrong place. Suggested rule: with a +context set, **every path is relative to it**, and an absolute path is written +with an explicit marker (a leading `.`, or an `absolute("…")` helper) — the +same unambiguous split a shell makes between `foo` and `/foo`. Whatever the +rule, destructive operations should echo the fully resolved path before acting. + +- **Open question:** should the context also apply to + `FolderIngest(target_catalog_path=…)`, so ingest and catalog take paths the + same way? They are the two halves of the same job and currently share no path + handling at all. +- **Open question:** one context, or a source/target pair? A draft→version + promotion has a common prefix; a copy between two dataflows does not, and + would need one of the two paths spelled out in full. + +#### 8. Proposed: `datacopy` should accept a folder and copy everything in it + +**Position to put to the team** (proposed here, not yet discussed with them): +a custodian should be able to point `datacopy` at a dataset folder and have +every table inside it copied, rather than making one call per table. + +**Today it doesn't just refuse — it refuses with the wrong reason.** +`datacopy`'s first step is `check_source_exists`, which calls `_entry_exists` +(`catalog/operations.py:190-206`): an `INFORMATION_SCHEMA."TABLES"` lookup for +`TABLE_SCHEMA = ` and `TABLE_NAME = `. A folder has no row there, +so the call fails with + +``` +CatalogOperationError: source '…bwd.draft.bw_assessment' does not exist +``` + +for a folder that plainly does exist. The library can already tell the +difference — `CatalogRestClient.is_folder()` (`catalog/rest.py:76-97`) — and it +already uses it, but only on the *other* side: `_resolve_target_path` asks +`is_folder` about the **target** to implement `cp source dest/` semantics +(`catalog/operations.py:234-249`). Folder-aware on the target, blind on the +source. Even without the feature, the message should say "is a folder" rather +than "does not exist". + +**Every building block for the real thing already exists:** + +| Need | Already there | +|---|---| +| detect a folder source | `CatalogRestClient.is_folder` (`rest.py:76`) | +| enumerate its contents | `gettablesfrom` — every table/view under a schema, **at any depth, in one query** (`operations.py:613-650`) | +| create target subfolders | `ensure_folder_path` — creates missing levels, tolerates already-there (`rest.py:275-312`) | +| tell a table from a view | `_entry_kind` (`operations.py:209-232`) | +| land inside a folder | `_resolve_target_path` (`operations.py:234-249`) | + +So this is wiring existing parts together, not new machinery. + +**And it is what `draft2version` needs anyway.** Per the taxonomy in +`dds_ingestion/common/catalog.py`, the dataset folder is +`…bwd.draft.bw_assessment` and `assessments` is a table inside it. "Copy a +dataset folder" and "promote a draft to a version" are the same operation at +the same level — so #8 is the mechanism #4 is missing, not a separate feature. + +**What the team has to decide** — none of these have an obvious default: + +- **Views.** `gettablesfrom` returns tables *and* views, but deliberately does + not select `TABLE_TYPE` (its own docstring says so). Two problems follow. + Copying a view with `CREATE TABLE … AS SELECT *` silently **materialises it + into a table** — the copy is a different kind of thing from the original. And + copying it *as* a view leaves its definition pointing at the original source + tables, so the copied folder isn't self-contained. `datamove` already treats + table/view confusion as a bug class worth auto-detecting + (`operations.py:447-457`). Cheap first step: add `TABLE_TYPE` to + `gettablesfrom`'s SELECT, so a folder copy knows per entry without N extra + round-trips. +- **Depth.** `gettablesfrom` is already recursive — its `LIKE 'schema.%'` + matches across dots, so it returns the whole subtree. That may be exactly + right for a dataset folder, but it should be a stated choice (`cp -r`), not + inherited by accident from the helper. +- **Partial failure.** One CTAS becomes N. `_run_actions` records + `step k/N` and `retry_pending` re-dispatches the operation **from the + beginning** (`operations.py:1167-1195`) — so with the default + `overwrite=False`, a resumed folder copy fails on every table that already + made it. It needs skip-what-exists, or a per-table record of what succeeded. + There is a precedent in this same library: `FolderIngest._already_done()` + skips files the server reports as already uploaded + (`dds_ingestion/folder.py:322-341`), and `retry_state` already persists a + free-form `params` dict that could carry a done-list + (`catalog/retry_state.py:40`). +- **`overwrite=True` at folder scale.** Does it mean "replace each colliding + table" or "replace the target folder"? Very different blast radius, and the + current wording doesn't extend cleanly. +- **Parallelism.** `FolderIngest` copies 4-way with a `ThreadPoolExecutor` + (`dds_ingestion/folder.py:363`). Tempting here, but `datacopy` is Flight-only + and `FlightSqlExecutor` holds one `FlightClient` with a lazily-cached + `_auth_header` — that lazy initialisation is unguarded, and this repo has not + verified concurrent statement execution on one Flight client. N concurrent + CTAS may also just queue on the Dremio engine. Worth measuring before + assuming it helps. +- **Return type.** `datacopy` returns one `SqlResult`. A folder copy needs a + per-table summary — the ingest side already has the shape for this in + `IngestOutcome` (`copied` / `skipped` / per-item results, + `dds_ingestion/folder.py:57-71`). + +- **Caveat for whoever implements it:** `is_folder` is best-effort by design — + it returns `False` on *any* lookup failure, because some Dremio source types + (the internal Arctic/Nessie-backed ones) answer a nested by-path lookup with + a 400 rather than a 404 (`catalog/rest.py:84-96`). On such a source, folder + detection silently degrades back to today's "does not exist". That needs a + decision too: fail loudly, or fall back to treating the path as a table. +- **Open question:** should `datamove` get the same treatment? It carries the + identical single-entry assumption, and moving a dataset folder is an equally + natural request — but copy-then-drop across N entries has a much worse + partial-failure story, which the module docstring already flags for the + single-entry case (`operations.py:30-37`). + +#### 8a. Proposed: split it into two verbs rather than overloading `datacopy` + +**Position to put to the team:** instead of `datacopy` detecting what it was +pointed at, have two explicit verbs — one for a single table, one for a whole +folder. + +This is the better shape, and not only on taste. Four concrete reasons: + +- **Detection is best-effort, so overloading is a silent-behaviour-change + risk.** `is_folder` returns `False` on any lookup failure, including the 400 + that Arctic/Nessie-backed sources answer with (`catalog/rest.py:84-96`). An + overloaded `datacopy` would quietly run the *single-table* path on a + false negative and fail with the misleading "does not exist". With the intent + declared in the verb, the same false negative becomes a clear error — "this + is not a folder" — because the library knows what the caller meant. +- **The return types genuinely differ.** A single copy returns `SqlResult`; a + folder copy needs a per-entry summary (see #8). Overloading forces a union + return that every caller has to narrow. +- **The parameters don't overlap.** Depth/recursion, per-entry parallelism and + a done-list only make sense for the folder verb; `overwrite` means different + things at the two scales. Overloading means parameters that are silently + ignored half the time — the same trap `Catalog(username=…)` already sets, + where the argument matters only if you happen to call a Flight operation. +- **It matches the naming already in use.** The existing verbs are lowercase + concatenations — `gettablesfrom`, `gettableitemsfrom`, `draft2version`, + `createfolder`. `tablecopy` / `datasetcopy` sit in that style without + introducing a new convention. + +**The one thing to settle first: "dataset" is an overloaded word here.** + +- In this project's taxonomy it means the folder — `common/catalog.py` calls + `bw_assessment` "the dataset folder", one level below `draft`. +- In **Dremio's** own vocabulary a *dataset* is a table or a view — that is what + PDS (physical dataset) and VDS (virtual dataset) stand for. + +So to a Dremio-fluent reader `datasetcopy()` would suggest copying exactly one +table, which is the opposite of the intent. Three ways out, in the order I'd +rank them: + +1. `tablecopy()` / `foldercopy()` — "folder" is already this library's own word + for the thing (`createfolder`, `deletefolder`, `ensure_folder_path`, + `is_folder`), so it stays internally consistent and collides with nothing. +2. `tablecopy()` / `datasetcopy()` as suggested, with the EEA meaning stated + loudly in the docstring — fine if the custodians' vocabulary is what matters + and Dremio's is not. +3. Keep one `datacopy` with an explicit `recursive=True` — the `cp -r` model, + which fits the command-line framing of #7, but keeps the union return type. + +- **Open question:** if this splits, what happens to `datacopy`/`datamove`? + Both are exported (`catalog/__init__.py:29-30`), used throughout + `debugger/debug_run.py`, and documented in the README — so it's an alias or a + deprecation, not a rename. And does `datamove` split the same way + (`tablemove` / `foldermove`), or stay single-entry given its worse + partial-failure story? + +#### 9. Proposed: settle the naming convention — `tableToView`, `datasetToViews` + +**Position to put to the team:** replace the `2`-as-"to" names with spelled-out +word separation — `table2view` → `tableToView` — and add the folder-level +counterpart `datasetToViews()`, matching the `tablecopy`/`datasetcopy` split in +#8a. + +**The underlying problem is bigger than the `2`: this module currently uses +three naming styles at once.** + +| Style | Names | +|---|---| +| lowercase run-together | `datacopy`, `datamove`, `deleteview`, `createfolder`, `deletefolder`, `deletetags`, `deletewiki`, `gettablesfrom`, `gettableitemsfrom`, `gettagsfrom`, `getwikifrom`, `getmetafromwiki`, `settagsto`, `setwikito`, `publishversion` | +| digit-for-word | `table2view`, `draft2version`, `setmeta2wiki` | +| snake_case | `retry_pending` | + +All 19 are public, all in the same `__all__` (`catalog/__init__.py:56-88`) and +the same class — including `retry_pending`, which is snake_case sitting beside +`gettablesfrom` in `Catalog` (`catalog/client.py:250`, `:327`). + +And the layer immediately below is *entirely* snake_case. `Catalog.createfolder` +→ `operations.createfolder` → `catalog_rest.create_folder` and +`ensure_folder_path` (`catalog/operations.py:1104-1109`). The same concept is +spelled two different ways one line apart. Same for `deletefolder` → +`delete_folder`, and for `is_folder` / `is_table_or_view` / `get_wiki` / +`set_wiki` / `get_tags` / `set_tags`, plus `load_creds`, `load_base_url`, +`scan_folder`, `ingest_folder`, `resolve_executor`, `list_pending` elsewhere in +the library. + +So this is one decision covering all 19 names, not a fix for three of them. + +**Which convention.** The instinct is right either way — `gettableitemsfrom` is +hard to read and `2` is not a word. The two candidates: + +- **camelCase** (`tableToView`, `datasetToViews`, `getTablesFrom`) — as + proposed. Reads well, and it matches the JSON world these operations talk to: + Dremio's own fields are camelCase (`entityType`, and the settings keys + `ddsServerUrl` / `accessToken` in `dremio_identity.py:85-90`). +- **snake_case** (`table_to_view`, `dataset_to_views`, `get_tables_from`) — + PEP 8, and it matches the ~20 functions in this library that already use it, + including the exact methods these verbs delegate to. It is also the only + option that can be *enforced*: ruff's `N` (pep8-naming) rules would keep it + from drifting again, and `select` currently omits `N` + (`pyproject.toml:53`) so nothing catches the drift today. Choosing camelCase + means `N` can never be turned on without a wall of `noqa`. + +My recommendation is snake_case, for the enforceability and for consistency +with the layer underneath — but the important thing is that one of them is +chosen for all 19 rather than the styles continuing to mix. Either beats what +is there now. + +**The verb matrix, if #8a and this both go ahead** (camelCase spelling shown +as proposed; substitute snake_case throughout if that wins): + +| single entry | whole folder | +|---|---| +| `tableToView` | `datasetToViews` | +| `tableCopy` | `datasetCopy` | +| `tableMove` | `datasetMove` | + +- Note the plural in `datasetToViews` — deliberate, since the operation yields + many views, but it makes the set irregular next to `datasetCopy`. Worth + deciding whether plurals track the output or the names stay uniform. +- `draft2version` and `setmeta2wiki` are the other two `2` names and should move + in the same sweep (`draftToVersion`, `setMetaToWiki`). +- The "dataset" vs Dremio's PDS/VDS meaning of the word, raised in #8a, applies + here too: `datasetToViews` reads to a Dremio-fluent user as "turn one dataset + into several views". + +**Migration.** These names are already released — tags up to `v0.1.10`, and the +README tells notebooks to install from `@main` / `@staging`, so custodians' +existing scripts import them. They are also used throughout +`debugger/debug_run.py` and documented in both READMEs. So: rename in one sweep, +keep the old names as thin aliases emitting a `DeprecationWarning` for one +release, then drop them. Worth doing now precisely because adoption is still +small — the cost only grows. + +- **Open question:** does the rename extend to the `Catalog` class's own + parameters (`target_catalog_path`, `conflict_mode`) and to `dds_ingestion`'s + public surface, or is it scoped to the catalog verbs? + +#### 10. Proposed: `createfolder` / `deletefolder` should take more than one folder + +**Position to put to the team:** both should accept a set of folders, not a +single path — `mkdir -p a b c` and `rm -r a b c` take multiple operands, and +that is the shell model these verbs are already imitating (their own docstrings +say "`rmdir` vs `rm -r`", `catalog/operations.py:1131`). + +Today both are strictly single-path: `createfolder(catalog_rest, path, *, +create_parents=False, …)` (`catalog/operations.py:1070`) and +`deletefolder(catalog_rest, path, *, cascade=False, …)` +(`catalog/operations.py:1119`). Setting up a dataflow means one call per folder. + +**This is the cheapest of the batch features to add** (cf. #8), because both +underlying operations are already idempotent: + +- `create_folder` tolerates a 409 and reports whether it newly created the + folder (`catalog/rest.py:255-273`); +- `delete_folder` returns early when the folder is already gone + (`catalog/rest.py:352-354`). + +So a partially-completed batch can simply be re-run — which matters, because a +batch shares one idempotency key and `retry_pending` re-dispatches the operation +**from the start** (`catalog/operations.py:1167-1195`). For #8's copy that is a +real problem; here it is harmless. + +**What the team has to decide:** + +- **Ordering is not the order the user typed, and it differs by verb.** + - Deleting without `cascade` must go **deepest-first**, or removing a parent + before its child fails with "folder is not empty" + (`catalog/rest.py:359-364`). + - Creating without `create_parents` must go **shallowest-first**, or creating + `a.b.c` before `a.b` fails with "parent does not exist" + (`catalog/operations.py:1098-1102`). + + Opposite sorts, and a naive `for p in paths:` loop gets both wrong. Sorting by + depth inside the batch verb makes a list of related folders "just work" — + which is most of the value of the feature. +- **Report what happened.** Both currently return `None`, and + `createfolder(create_parents=True)` **throws away information it already + has**: `ensure_folder_path` returns the list of levels it actually created + (`catalog/rest.py:275-312`) and `create_folder` returns whether the folder was + new — then `createfolder` discards both. A batch needs a per-path result + (created / already existed / failed) and the pieces are already there. +- **Fail-fast or continue-on-error?** For creates, continuing and reporting is + probably right. For deletes it is not obviously right, and the default should + be a deliberate choice rather than whatever falls out of the loop. +- **Echo the resolved paths before deleting.** Especially once #7's context + makes the written paths relative — a batch `deletefolder(cascade=True)` is the + single most destructive call in this library. + +**On API shape — this one can safely accept both, unlike #8a.** The argument +against overloading `datacopy` was that folder-vs-table detection is +best-effort and can silently pick the wrong path. Here there is no detection: +`str` vs `Sequence[str]` is unambiguous at runtime, so `path: str | +Sequence[str]` keeps every existing call working and needs no new verb. If the +team prefers explicit plural verbs anyway (`createfolders` / `deletefolders`), +that is a naming call under #9, not a correctness one. + +- **Note for whoever implements it:** `deletefolder` is the only operation that + bypasses the shared `_run_steps` / `_run_actions` helpers and hand-rolls its + own record-and-clear (`catalog/operations.py:1135-1142`). A batch version + should go through the shared helper so step numbering and retry recording + behave like everything else. +- **Note:** the `_OPERATIONS` registry that `retry_pending` dispatches through + (`catalog/operations.py:1145-1163`) is keyed by operation name — any new or + renamed verb from #9 or #8a has to be added there too, or its retries break + with a `KeyError`. + +#### What "simple" could look like + +Sketch to argue about, not a proposal to implement — the same job as your +example, with #1, #2, #4, #5 and #7 addressed. Each line that disappears is one +of the findings above: + +```python +from eea_datalakehouse.catalog import Catalog + +# no URL, no token, no username — resolved from the panel / %init / env (#1) +with Catalog() as catalog: + catalog.use("catalog.water_management_resources.bathing_water.bwd") # (#7) + catalog.datacopy("draft.bw_assessment.assessments", # (#2: no key) + "versions.2025_6.bw_assessment.assessments") +``` + +and once the domain verb exists (#4), with the context supplying everything +above the dataflow: + +```python +with Catalog() as catalog: + catalog.use("catalog.water_management_resources.bathing_water.bwd") + catalog.draft2version("bw_assessment", version="2025_6") +``` + +or, with #3, nothing in Python at all: + +```bash +export EEA_CATALOG_CONTEXT=catalog.water_management_resources.bathing_water.bwd +eea-catalog draft2version bw_assessment --version 2025_6 +``` + +Compare against what the same job takes today — full credentials, a hand-made +idempotency key, and the taxonomy typed twice: + +```python +with Catalog(DREMIO_URL, DREMIO_TOKEN, username=DREMIO_USERNAME) as catalog: + catalog.datacopy( + "catalog.water_management_resources.bathing_water.bwd.draft.bw_assessment.assessments", + "catalog.water_management_resources.bathing_water.bwd.versions.2025_6.bw_assessment.assessments", + create_target_folder=True, + idempotency_key="343242dsrew", + ) +``` + +### Half the public surface has no docstring — and a notebook is where that shows + +- **Status:** open +- **Where:** the whole package; worst in `dds_ingestion/models.py`, + `dds_ingestion/progress.py`, `dds_ingestion/client.py` (this branch) and + `catalog/client.py` (on `development`) +- **Observation:** counting public classes/functions/methods (`__init__` + included, `_private` names excluded): + + | tree | public defs | no docstring | + | --- | --- | --- | + | this branch (`data_preparation` + `dds_ingestion`) | 82 | 29 (35%) | + | `development` (adds `catalog/`) | 152 | 75 (49%) | + + Note what is *not* missing: type information. `mypy` runs `strict = true` + (`pyproject.toml:56`) and a scan of both trees finds no unannotated parameter + and one unannotated return in total. Only the prose is absent, and unevenly: + + - `catalog/client.py` — 20 of 22 public members, including every verb + (`table2view`, `draft2version`, `publishversion`, `datacopy`, `datamove`, + …). Its own class docstring is thorough; the methods under it have none. + See the next entry — those docstrings do exist, one layer down. + - `dds_ingestion/client.py` — `IngestClient.begin:107`, `.commit:136`, + `.close:90`, and both `__init__`s. + - `dds_ingestion/models.py` — 15 of 27, i.e. every `from_json`/`as_payload` + and the `Progress:134` class itself. + - `dds_ingestion/progress.py` — the `ProgressBar:13` protocol's `update`/ + `close`, though the module docstring does spell the contract out. + - `data_preparation/` — classes and methods are documented well; but none of + the four modules, nor the package `__init__.py`, has a module docstring. +- **Why it matters:** this library is driven from a notebook by design — + `dds_ingestion/__init__.py` opens with "A generated Jupyter notebook imports + :class:`FolderIngest`…". In JupyterLab, `Shift+Tab` inside a call and `obj?` + in a cell both render the same two things and nothing else: the signature and + the docstring. Where the docstring is empty a custodian gets a bare signature + and has to leave the notebook and open the source to learn what `commit()` + returns or what `Progress` carries. We have documentation — in READMEs — but + it is not reachable from where people are typing, which is the one place the + question actually gets asked. +- **Suggestion:** a one-line summary on everything named in an `__all__`, in the + NumPy style `data_preparation/transformation.py:19` already uses, plus module + docstrings for the four `data_preparation` modules. To stop it drifting back, + turn on ruff's pydocstyle rules for `src/`: `pyproject.toml:50` currently + selects `E,F,I,UP,B,C4,SIM`, so nothing has ever checked for a docstring — + adding `"D"` with `[tool.ruff.lint.pydocstyle] convention = "numpy"` would. +- **Open question:** do we want this on the `from_json`/`as_payload` wire + helpers too, or only on the surface a notebook touches? A `D` rule set covers + both unless we add a per-file-ignore for `models.py`. + +### `Catalog`'s methods drop the docstrings that `operations` already wrote + +- **Status:** open +- **Where:** `src/eea_datalakehouse/catalog/client.py:174-249` vs + `catalog/operations.py` (on `development`) +- **Observation:** all 19 public verbs in `operations.py` carry a real + docstring — `datacopy:343` "Copy data from `source_path` into `target_path`.", + `draft2version:315` "Promote the draft table at `draft_path` into a permanent + `version_path`.", and so on down the list. `Catalog`'s methods are thin + delegates onto them (`client.py:174` `table2view` → `operations.table2view`, + `:205` `datacopy` → `operations.datacopy`, same for the rest) and **not one + delegate has a docstring**, so none of that text reaches the object a notebook + actually holds. +- **Why it matters:** the writing is already done and paid for; it is just one + layer below where anyone looks. `Shift+Tab` on `catalog.datacopy(` — the + natural way to check argument order mid-cell — shows the signature and + nothing else, while `operations.datacopy?` shows the paragraph that would have + answered the question. `help(Catalog)` has the same hole. +- **Suggestion:** the cheapest fix that cannot drift is to copy the text at + class-definition time instead of retyping it — `datacopy.__doc__ = + operations.datacopy.__doc__` after the class body, or a tiny + `_delegates_to(operations.datacopy)` decorator that assigns `__doc__` **only**. + Deliberately not `functools.wraps`: that also rebinds `__name__`, + `__wrapped__` and `__signature__`, so the inspector would start showing the + module function's signature — including the leading `executor` argument that + `Catalog` supplies itself, which is exactly the confusion we'd be trying to + remove. If we'd rather stay explicit and boring, a one-line summary plus "See + `operations.datacopy`." is 19 lines of typing, once. +- **Open question:** is `Catalog` the documented surface and `operations` the + internals, or are both public? `catalog/__init__.py` exports both and its + module docstring points at `operations` for callers who want to manage the + executor themselves — which argues both, and so argues for docstrings that + read correctly in either place. + +### Nothing pins the notebook environment the library is written for + +- **Status:** open +- **Where:** `pyproject.toml:13-32` (`dependencies`) and `:34-41` + (`[project.optional-dependencies]`) +- **Observation:** the package is notebook-facing by design, but nothing in the + repo installs, pins, or exercises a notebook: no `jupyterlab` or `ipykernel` + in the runtime dependencies or in the `dev` extra, no `.ipynb` anywhere in the + tree, and `tqdm` — the bar `dds_ingestion/progress.py` drives — comes in as + the plain terminal build rather than its notebook widget variant. +- **Why it matters:** if we want to tell custodians "the API reference is + `Shift+Tab`", we need to be able to say *in which Jupyter*. Completion + behaviour is version-dependent (JupyterLab 4 is where the inline completer and + the LSP integration became first-class), and "it completes for me, not for + you" is unanswerable without a reproducible environment. It also means nobody + has ever run the notebook path in CI. +- **Suggestion:** add a `notebook` extra — `jupyterlab>=4,<5` and `ipykernel`, + optionally `jupyterlab-lsp` + `python-lsp-server` if we want static completion + as well as the runtime kind — so `pip install -e ".[notebook]"` reproduces + what we document against. Two related loose ends worth folding in: `py.typed` + exists only under `dds_ingestion/`, not in `catalog/`, `data_preparation/` or + the package root, so a type checker or an LSP treats the rest of the package + as untyped despite `mypy strict`; and `pyproject.toml` declares no package + data at all, so it is worth confirming the marker actually lands in a built + wheel (`python -m build && unzip -l dist/*.whl | grep py.typed`). +- **Open question:** do custodians run JupyterLab themselves, or in a hosted + environment where we control neither the Lab version nor the installed + extensions? That decides whether static/LSP completion is worth chasing at + all — if it isn't, everything rests on runtime introspection, which is to say + on the docstrings in the two entries above. + +## Ideas / parking lot + +_Larger or fuzzier things that aren't defects — worth a conversation, not a ticket yet._ + +## Questions for the team + +_Things we can't resolve from the code alone._ + +- Is `docs/` intended to become a real documentation folder in this repo, or is + documentation hosted elsewhere? +- Which Jupyter do custodians actually run — JupyterLab 4 locally, an older + Notebook, or a hosted environment we don't control? If we intend tab-completion + and `Shift+Tab` to be the day-to-day API reference, that version is a + supported-platform decision, not a preference. diff --git a/docs/read-only-ingest-client-plan.md b/docs/read-only-ingest-client-plan.md new file mode 100644 index 0000000..6befe8f --- /dev/null +++ b/docs/read-only-ingest-client-plan.md @@ -0,0 +1,113 @@ +# Read-only ingest — the client half (DI-11.9) — **DONE** + +> Landed: `sub_path` (DI-11.12), the retry guard and `placement` (DI-11.7), +> `CommitResult.storage_path` and the docs below (DI-11.9). mypy clean, 52 tests +> pass. What is still open is recorded at the end of this file. + +Companion note for **this** repository. The full design lives with the server +work it depends on: + +``` +EEALakeHouse/Components/DremioDocumentService/devplans/2026-08-21/ + Development Plan- Read-Only Ingest to Permanent S3 (DI-11).md +``` + +Read that first — this file only records what is built **here**, and when. + +## The short version + +`FolderIngest` already has the attribute: `intent="read_only" | "editable"` +(`src/eea_datalakehouse/dds_ingestion/folder.py:103`, `models.py:13`), sent to +`POST /ingest/begin` (`client.py:121`). On EEA production's managed catalog the +server currently ignores it for placement — every ingest stages to a temp prefix, +is copied into the catalog's Iceberg storage, and the upload is deleted. + +DI-11 makes `read_only` mean *store the files permanently* under +`local_s3/dh-prod-data/read///` and register them. +**Where the bytes land is chosen by the server**, in `begin`'s presigned targets, +so this package needs no new parameter and no new call. + +## What changed here + +0. ✅ **`sub_path` — the one new parameter** (DI-11.12). `FolderIngest(..., + sub_path="2026")`, passed straight through to `begin`; the server validates, + normalises and scopes on it. It is what lets a read-only table accumulate: + + ```python + FolderIngest(folder="./bw_2026", target_catalog_path="…/bwd/reference", + data_format="parquet", intent="read_only", + table_name="water_temperature", sub_path="2026").run() + # → read/…/water_temperature/2026/*.parquet, alongside 2024/ and 2025/ + ``` + + A custodian whose local folder *already* has the structure needs nothing — + `scan_folder` (`folder.py:80-92`) preserves sub-folders and the server keeps + the original `rel_path`. `sub_path` is for filing a **flat** folder under a + name they choose. Worth a worked example in `README.md`, and note that + `replace` with a `sub_path` clears only that year. +1. ✅ **Docs and docstrings** — `folder.py` class docstring + the `intent` + parameter, `dds_ingestion/README.md:38`, `__init__.py:16`: say what each + intent now does (permanent raw folder + catalog view vs. Iceberg copy). +2. ✅ **`CommitResult.storage_path`** (`models.py`) — optional, additive; the + physical Dremio path the server reports for a permanent ingest, so a notebook + can print where the files actually are. Parse permissively as everywhere else + (absent ⇒ `None`), so it works against an older server. +3. ✅ **Resume/retry note** (`folder.py:305-322`, `retry()`): `_already_done`'s + docstring says a re-run is "an S3 overwrite of the identical key, which is + harmless". Against a permanent folder in `append` mode that is only true if + the server reuses the keys instead of numbering them. Align the wording with + whatever DI-11.7 settles, and steer users to `attach()` + `retry()` rather + than re-running `run()`. +4. ✅ **Tests** — `tests/dds_ingestion/`: `storage_path` parsing, and a + respx-mocked `begin` whose `key_prefix` is a permanent `read/...` prefix, + asserting the client uploads to the issued targets unchanged. + +## What does not change + +- No new constructor argument, no new method, no new endpoint. +- `editable` behaviour, the upload/parallelism/progress machinery, credentials. +- Nothing has to change for the client to keep working while the server runs + with `DDS_INGEST_READ_MODE=staged` (its default). + +## When + +DI-11 **step 4** ("Prove") in the rollout table — after the server feature exists +behind its flag, and it is the first client pointed at a `permanent` server. The +JupyterLab extension is deliberately left alone until step 5. + +## Noted for later: deleting a read-only table + +Because read-only files are permanent, dropping the view or dataset from the +catalog leaves the objects in S3. The agreed home for that cleanup is a +**delete-table operation in this package** — `eea_datalakehouse.catalog` has +`deleteview` and `deletefolder` today, but nothing that removes a table together +with its backing data. **Not implemented as part of DI-11**, recorded here (and +in §8.5 of the DI-11 plan) so it is a known gap rather than a surprise. + +When it is picked up: this package holds no S3 credentials by design, so the +delete needs a DDS-side endpoint for the read folder to call — not boto3 in a +notebook. + +## Related: `common/catalog.py` moves server-side + +`dds_ingestion/common/catalog.py` (on `development`) walks a catalog path and +creates the missing dataflow levels, refusing to invent a domain or subdomain. +That walk is exactly what a read-only ingest needs before its view can be +created, so **DI-11.2 ports it into DDS**, where the extension and any REST +client get it too and the taxonomy rule is enforced in one place. Once that +lands, the client-side copy is either a thin convenience wrapper over +`POST /api/v1/catalog/{path}/folder` or redundant — decide when DI-11.2 is done, +not before. + +This is a *catalog* concern only: the `read/` prefixes in S3 need no creating at +all — see §3.6 of the DI-11 plan. + +## One thing to settle first + +There are two copies of this client: this repo +(`src/eea_datalakehouse/dds_ingestion/`) and a snapshot in the DDS repo +(`clients/dds_ingest/dds_ingest/`). They have already drifted — `folder.py`, +`client.py` and `models.py` differ; `credentials.py` and `progress.py` are +identical. **This repo is where the client is developed.** Whether the DDS-side +copy is re-synced, thinned to a test fixture, or dropped is a separate decision, +not part of DI-11 — but decide it before editing both by hand again. diff --git a/src/eea_datalakehouse/dds_ingestion/README.md b/src/eea_datalakehouse/dds_ingestion/README.md new file mode 100644 index 0000000..cc8417e --- /dev/null +++ b/src/eea_datalakehouse/dds_ingestion/README.md @@ -0,0 +1,116 @@ +# eea_datalakehouse.dds_ingestion + +Notebook-side client for the **Dremio Document Service (DDS) Ingest API** (DI-8.4 / DI-8.5). + +A generated Jupyter notebook imports `FolderIngest` to transfer a folder of data +files to S3 and register it as a Dremio table — coordinated entirely through the +DDS REST API. The **only** object-storage access is the presigned URLs issued by +DDS; this package never uses an S3 SDK or S3 credentials. + +## Install + +This is a subpackage of the **EEADataLakehouse** library, not a distribution of +its own — install the library from the repository root: + +```bash +pip install -e ".[dev]" # dev extra adds ruff / mypy / pytest / respx +``` + +In a notebook kernel, pinned to a ref that carries this subpackage: + +```python +%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@development" +``` + +## Usage (inside a notebook kernel) + +Dremio credentials are read from the injected kernel env vars `_DREMIO_USER` / +`_DREMIO_PWD`, and the service URL from `DDS_BASE_URL`. None of these are ever +logged or printed. + +```python +from eea_datalakehouse.dds_ingestion import FolderIngest + +outcome = FolderIngest( + folder="./my_data", + target_catalog_path="biodiversity.uploads", + data_format="parquet", # one of parquet | csv | json + intent="read_only", # or "editable" — see below + conflict_mode="fail", + parallelism=4, # concurrent uploads (default 4) +).run() + +print(outcome.commit.table_path, outcome.commit.record_count) +print(outcome.commit.storage_path) # where the files are, when they are kept +``` + +`intent` decides what the transfer leaves behind: + +| | `read_only` | `editable` | +|---|---|---| +| your uploaded files | kept — they **are** the table | copied in, then deleted | +| the table | the folder, registered, with a view at the catalog path | an Iceberg table | +| good for | published data, data that accumulates | a table you will write to | + +(Storing read-only files permanently is a server setting. Where it is not +enabled, both intents stage and load as before and only the table's shape +differs — `storage_path` is then `None`.) + +`run()` performs `begin → upload(all files) → commit`, and leaves the session +handle available afterwards. + +## Read-only data that grows + +`intent="read_only"` on a server configured for permanent storage keeps your +files as uploaded and registers them as the table — nothing is copied into the +catalog. `sub_path` files each upload under a named sub-folder, so one table can +accumulate: + +```python +FolderIngest( + folder="./bw_2026", # a flat folder of parquet files + target_catalog_path="water_management_resources/bathing_water/bwd/reference", + data_format="parquet", + intent="read_only", + table_name="water_temperature", + sub_path="2026", # → .../water_temperature/2026/ + conflict_mode="fail", # refuses if 2026 is already there +).run() +``` + +Next year, the same call with `sub_path="2027"` adds to the same table; +`conflict_mode="replace"` with a `sub_path` re-does **that year only** and leaves +the others standing. If your local folder already has the structure +(`2026/*.parquet`), you do not need `sub_path` — the layout is preserved as-is. +The name is normalised server-side to the EEA convention (lowercase, +underscores), so `"Q1 2026"` becomes `q1_2026`. + +## Managing the transfer + +A transfer is a server-side session, so it can be inspected and resumed: + +```python +with FolderIngest(folder="./my_data", target_catalog_path="catalog/theme/sub", + data_format="parquet") as job: + try: + job.run() + except Exception: + if job.status().is_resumable: + job.retry() # re-runs only the step that failed — no re-upload +``` + +`job.session_id` · `status()` · `estimate()` · `retry()` · `cancel()` · +`close()` · `FolderIngest.attach(session_id, ...)` to pick a transfer up in a +later kernel. + +**Full class reference:** [`docs/python-client-guide.md`](../../docs/python-client-guide.md). + +## Layout + +| Path | Purpose | +|---|---| +| `dds_ingestion/credentials.py` | env-var creds + redacted `DremioCreds` | +| `dds_ingestion/models.py` | typed request/response models for the contract | +| `dds_ingestion/client.py` | thin, unit-testable HTTP client (`IngestClient`) | +| `dds_ingestion/progress.py` | tqdm progress bar with graceful fallback | +| `dds_ingestion/folder.py` | `FolderIngest` orchestration (scan/parallel/session) | diff --git a/src/eea_datalakehouse/dds_ingestion/__init__.py b/src/eea_datalakehouse/dds_ingestion/__init__.py index 6a57109..54d714c 100644 --- a/src/eea_datalakehouse/dds_ingestion/__init__.py +++ b/src/eea_datalakehouse/dds_ingestion/__init__.py @@ -1,4 +1,5 @@ -"""Ingestion into DDS (Dremio Document Service), notebook-side (DI-8.4/8.5). +"""eea_datalakehouse.dds_ingestion — notebook-side client for the Dremio +Document Service Ingest API. A generated Jupyter notebook imports :class:`FolderIngest` to transfer a folder of data files to S3 (via DDS-issued presigned URLs only) and register it as a @@ -12,15 +13,21 @@ folder="./my_data", target_catalog_path="biodiversity.uploads", data_format="parquet", - intent="read_only", + intent="read_only", # keep the files as the table | "editable" → Iceberg parallelism=4, ).run() print(outcome.commit.table_path, outcome.commit.record_count) + print(outcome.commit.storage_path) # where the files are, if kept + +``intent`` decides what the transfer leaves behind. On a server configured for +permanent read-only storage, ``"read_only"`` keeps the uploaded files where the +table lives and registers them; ``"editable"`` loads them into an Iceberg table +and deletes the upload. See :class:`FolderIngest` for the full rule. """ from __future__ import annotations -from .client import IngestApiError, IngestClient +from .client import IngestApiError, IngestClient, S3UploadError from .credentials import ( DremioCreds, MissingCredentialsError, @@ -31,6 +38,7 @@ DEFAULT_PARALLELISM, FolderIngest, IngestOutcome, + IngestStateError, ingest_folder, scan_folder, ) @@ -38,10 +46,12 @@ BeginResult, CommitResult, DataFormat, + EstimateResult, FileSpec, Intent, Progress, S3Plan, + StageResult, StatusResult, UploadPart, UploadTarget, @@ -53,15 +63,19 @@ "CommitResult", "DataFormat", "DremioCreds", + "EstimateResult", "FileSpec", "FolderIngest", "IngestApiError", "IngestClient", "IngestOutcome", + "IngestStateError", "Intent", "MissingCredentialsError", "Progress", "S3Plan", + "S3UploadError", + "StageResult", "StatusResult", "UploadPart", "UploadTarget", diff --git a/src/eea_datalakehouse/dds_ingestion/client.py b/src/eea_datalakehouse/dds_ingestion/client.py index 1e42d79..cc6b8b7 100644 --- a/src/eea_datalakehouse/dds_ingestion/client.py +++ b/src/eea_datalakehouse/dds_ingestion/client.py @@ -1,9 +1,25 @@ """Thin, unit-testable HTTP client for the DDS Ingest API (v0.1). This layer knows nothing about local folders, parallelism or progress bars: it -only translates the four ingest endpoints to/from the typed models in -:mod:`eea_datalakehouse.dds_ingestion.models`. The orchestration logic lives in -:mod:`eea_datalakehouse.dds_ingestion.folder`. +only translates the ingest endpoints to/from the typed models in +:mod:`eea_datalakehouse.dds_ingestion.models`. The orchestration logic +lives in :mod:`eea_datalakehouse.dds_ingestion.folder`. + +One method per endpoint: + +=========================== ========================================== +:meth:`IngestClient.begin` ``POST /api/v1/ingest/begin`` +:meth:`~IngestClient.commit` ``POST /api/v1/ingest/commit`` +:meth:`~IngestClient.get_status` ``GET /api/v1/ingest/{id}`` +:meth:`~IngestClient.list_sessions` ``GET /api/v1/ingest`` +:meth:`~IngestClient.retry` ``POST /api/v1/ingest/{id}/retry`` +:meth:`~IngestClient.cancel` ``DELETE /api/v1/ingest/{id}`` +:meth:`~IngestClient.estimate` ``POST /api/v1/ingest/estimate`` +:meth:`~IngestClient.stage` ``POST /api/v1/ingest/stage`` +=========================== ========================================== + +plus :meth:`~IngestClient.upload_file` / :meth:`~IngestClient.upload_file_multipart`, +which talk to S3 through the presigned targets ``begin`` issued. Authentication uses a Bearer token (the Dremio PAT, which is the kernel's ``_DREMIO_PWD``) sent **only on the DDS API calls** — never on the presigned S3 @@ -23,22 +39,62 @@ BeginResult, CommitResult, DataFormat, + EstimateResult, FileSpec, Intent, + StageResult, StatusResult, UploadTarget, ) DEFAULT_TIMEOUT = 60.0 +# Error bodies are quoted back to the caller; cap them so an HTML error page +# does not bury the status line it came with. +_MAX_ERROR_CHARS = 500 class IngestApiError(RuntimeError): - """A DDS ingest API call returned a non-success status.""" + """A DDS ingest API call returned a non-success status. - def __init__(self, status_code: int, message: str) -> None: - super().__init__(f"DDS ingest API error {status_code}: {message}") + ``where`` names the call that failed (e.g. ``POST /api/v1/ingest/begin``). + It is part of the message because a transfer makes several different calls + to two different systems, and "API error 500" alone does not say which one + broke — see :class:`S3UploadError` for the other system. + """ + + def __init__(self, status_code: int, message: str, *, where: str | None = None) -> None: + super().__init__( + f"DDS ingest API error {status_code}" + f"{f' on {where}' if where else ''}: {message}" + ) self.status_code = status_code self.message = message + self.where = where + + +class S3UploadError(IngestApiError): + """A pre-signed upload to object storage failed — NOT a DDS API error. + + The bytes go straight from here to S3 using the target ``begin`` issued, so + this failure is the object storage's (or whatever proxy sits in front of + it): a rejected policy, an unreachable endpoint, a gateway error. Reported + as its own class so a caller can tell "your upload never landed" from "the + Document Service refused the request", and so the message names the file and + the URL that actually answered. + """ + + def __init__( + self, status_code: int, message: str, *, rel_path: str, url: str + ) -> None: + RuntimeError.__init__( + self, + f"S3 upload of {rel_path!r} failed: HTTP {status_code} from {url}: {message}", + ) + self.status_code = status_code + self.message = message + self.where = url + self.rel_path = rel_path + self.url = url class IngestClient: @@ -95,6 +151,7 @@ def begin( conflict_mode: str, files: list[FileSpec], table_name: str | None = None, + sub_path: str | None = None, idempotency_key: str | None = None, multipart: bool | None = None, ) -> BeginResult: @@ -108,6 +165,8 @@ def begin( } if table_name is not None: body["table_name"] = table_name + if sub_path: + body["sub_path"] = sub_path if idempotency_key is not None: body["idempotency_key"] = idempotency_key if multipart is not None: @@ -132,12 +191,83 @@ def commit( return CommitResult.from_json(data) def get_status(self, session_id: str) -> StatusResult: + """Current state of one session (``GET /api/v1/ingest/{id}``).""" resp = self._http.get( f"{self._base_url}/api/v1/ingest/{session_id}", headers=self._auth_headers ) - self._raise_for_status(resp) + self._raise_for_status(resp, f"GET /api/v1/ingest/{session_id}") return StatusResult.from_json(resp.json()) + def list_sessions(self, state: str | None = None) -> list[StatusResult]: + """Your ingest sessions (``GET /api/v1/ingest``). + + ``state`` filters server-side: ``"active"`` (still running), + ``"failed"``, or ``"all"``. Omit for the server's default view. + """ + params = {"state": state} if state else None + resp = self._http.get( + f"{self._base_url}/api/v1/ingest", params=params, headers=self._auth_headers + ) + self._raise_for_status(resp, "GET /api/v1/ingest") + payload = resp.json() + rows = payload if isinstance(payload, list) else payload.get("sessions", []) + return [StatusResult.from_json(row) for row in rows] + + def retry(self, session_id: str) -> StatusResult: + """Re-run a failed session's last stage (``POST /api/v1/ingest/{id}/retry``). + + Only a ``failed`` session can be retried, and only the Dremio load stage + is resumable — the server kept the staged files for exactly that case + (see :attr:`StatusResult.is_resumable`). A session whose upload never + completed, or whose failure a re-run cannot fix, raises + :class:`IngestApiError` telling you to upload again. + """ + resp = self._http.post( + f"{self._base_url}/api/v1/ingest/{session_id}/retry", + headers=self._auth_headers, + ) + self._raise_for_status(resp, f"POST /api/v1/ingest/{session_id}/retry") + return StatusResult.from_json(resp.json()) + + def cancel(self, session_id: str) -> None: + """Discard a session and its staged data (``DELETE /api/v1/ingest/{id}``). + + Use it to abandon a transfer: the staged objects are deleted and the + session is dropped. It does NOT remove a table that a previous commit + already created. + """ + resp = self._http.delete( + f"{self._base_url}/api/v1/ingest/{session_id}", headers=self._auth_headers + ) + self._raise_for_status(resp, f"DELETE /api/v1/ingest/{session_id}") + + def estimate(self, session_id: str) -> EstimateResult: + """Row-count + size class for the staged data (``POST .../estimate``). + + A pre-flight check between upload and commit. On a managed catalog the + count comes from the staged Parquet footers (CSV/JSON have none and + report 0). + """ + data = self._post("/api/v1/ingest/estimate", {"session_id": session_id}) + return EstimateResult.from_json(data) + + def stage(self, session_id: str, rel_path: str, data: bytes) -> StageResult: + """Upload one file THROUGH the server (``POST /api/v1/ingest/stage``). + + The fallback when the S3 endpoint is not reachable from this kernel: DDS + writes the bytes with its own catalog credentials instead of handing back + a presigned URL. Slower — the bytes transit the server — so prefer + :meth:`upload_file`, and use this when that fails to connect. + """ + resp = self._http.post( + f"{self._base_url}/api/v1/ingest/stage", + data={"session_id": session_id, "rel_path": rel_path}, + files={"file": (rel_path.rsplit("/", 1)[-1], data)}, + headers=self._auth_headers, + ) + self._raise_for_status(resp, "POST /api/v1/ingest/stage") + return StageResult.from_json(resp.json()) + def upload_file(self, target: UploadTarget, data: bytes) -> str | None: """Upload one file's bytes to its presigned target. @@ -171,7 +301,7 @@ def upload_file(self, target: UploadTarget, data: bytes) -> str | None: resp = self._http.request( target.method, target.url, content=data, headers=target.headers ) - self._raise_for_status(resp) + self._raise_for_upload(resp, rel_path=target.rel_path, url=target.url) return resp.headers.get("ETag") def upload_file_multipart( @@ -209,7 +339,7 @@ def upload_file_multipart( start = index * chunk body = data[start : start + chunk] if chunk else b"" resp = self._http.put(part.url, content=body) - self._raise_for_status(resp) + self._raise_for_upload(resp, rel_path=target.rel_path, url=part.url) etag = resp.headers.get("ETag") if etag is None: raise IngestApiError( @@ -226,14 +356,19 @@ def _post(self, path: str, body: dict[str, Any]) -> dict[str, Any]: resp = self._http.post( f"{self._base_url}{path}", json=body, headers=self._auth_headers ) - self._raise_for_status(resp) + self._raise_for_status(resp, f"POST {path}") result: dict[str, Any] = resp.json() return result @staticmethod - def _raise_for_status(resp: httpx.Response) -> None: - if resp.is_success: - return + def _error_message(resp: httpx.Response) -> str: + """The most useful text an error response carries. + + A DDS error is ``{error, message, path}`` JSON, so quote its ``message``. + Anything else — an S3 XML fault, a proxy's HTML error page, Starlette's + bare ``Internal Server Error`` — is quoted verbatim but capped: those + pages run to kilobytes and only the first lines identify the sender. + """ message = resp.text try: payload = resp.json() @@ -241,4 +376,25 @@ def _raise_for_status(resp: httpx.Response) -> None: message = str(payload.get("message") or payload.get("error") or message) except ValueError: pass - raise IngestApiError(resp.status_code, message) + message = message.strip() + if len(message) > _MAX_ERROR_CHARS: + message = message[:_MAX_ERROR_CHARS] + "… (truncated)" + return message or f"(empty {resp.status_code} response body)" + + @classmethod + def _raise_for_status(cls, resp: httpx.Response, where: str) -> None: + """Raise :class:`IngestApiError` naming ``where`` unless the call succeeded.""" + if resp.is_success: + return + raise IngestApiError(resp.status_code, cls._error_message(resp), where=where) + + @classmethod + def _raise_for_upload( + cls, resp: httpx.Response, *, rel_path: str, url: str + ) -> None: + """Raise :class:`S3UploadError` unless the pre-signed upload succeeded.""" + if resp.is_success: + return + raise S3UploadError( + resp.status_code, cls._error_message(resp), rel_path=rel_path, url=url + ) diff --git a/src/eea_datalakehouse/dds_ingestion/folder.py b/src/eea_datalakehouse/dds_ingestion/folder.py index 2f71dad..fb852b7 100644 --- a/src/eea_datalakehouse/dds_ingestion/folder.py +++ b/src/eea_datalakehouse/dds_ingestion/folder.py @@ -9,8 +9,9 @@ * show a progress bar (tqdm, degrading gracefully if absent); * resume — re-running skips files already uploaded for the session. -The class depends only on :class:`~eea_datalakehouse.dds_ingestion.client.IngestClient`, -so the HTTP layer can be mocked or swapped in tests. +The class depends only on +:class:`~eea_datalakehouse.dds_ingestion.client.IngestClient`, so the HTTP +layer can be mocked or swapped in tests. """ from __future__ import annotations @@ -19,6 +20,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from pathlib import Path +from typing import Any from .client import IngestClient from .credentials import DremioCreds, load_base_url, load_creds @@ -26,8 +28,10 @@ BeginResult, CommitResult, DataFormat, + EstimateResult, FileSpec, Intent, + StatusResult, UploadTarget, ) from .progress import make_progress_bar @@ -42,15 +46,29 @@ } +class IngestStateError(RuntimeError): + """An operation was asked for in a state that cannot support it. + + Raised locally, before any HTTP call — e.g. committing a transfer that never + began, or retrying one that is still running. + """ + + @dataclass(slots=True) class IngestOutcome: - """Summary returned by :meth:`FolderIngest.run`.""" + """Summary returned by :meth:`FolderIngest.run` and :meth:`FolderIngest.retry`. - begin: BeginResult + ``begin`` is ``None`` when the transfer was resumed rather than started here + (:meth:`FolderIngest.attach` had no ``begin`` of its own to report), and + ``resumed`` says which happened. + """ + + begin: BeginResult | None commit: CommitResult files_uploaded: int files_skipped: int etags: dict[str, str] = field(default_factory=dict) + resumed: bool = False def scan_folder(folder: Path, data_format: DataFormat) -> list[FileSpec]: @@ -74,7 +92,37 @@ def scan_folder(folder: Path, data_format: DataFormat) -> list[FileSpec]: class FolderIngest: - """Orchestrate ingest of a local folder into a Dremio table via DDS.""" + """Orchestrate ingest of a local folder into a Dremio table via DDS. + + ``intent`` decides what the transfer leaves behind, and on a server + configured for permanent read-only storage that includes **where the files + end up**: + + * ``"read_only"`` — the upload is stored where the table lives and kept. The + folder is registered as the dataset (which is also what builds Dremio's + metadata: schema, file listing, Parquet statistics) and a view at the + catalog path points at it. Nothing is copied, and the files keep the shape + they were exported in. Use it for data that is published, not edited. + * ``"editable"`` — the upload is staged, loaded into an Iceberg table in the + catalog, and then deleted. Use it for a table that will be written to. + + Against a server that has not enabled permanent storage, both intents stage + and load as before; the difference is then only the table's shape. Either way + the destination is the server's to choose — this class uploads to the targets + ``begin`` issues. + + ``sub_path`` files this upload under a named sub-folder of the table — the + accumulating-dataset shape, one year at a time:: + + FolderIngest(folder="./bw_2026", target_catalog_path=..., data_format="parquet", + intent="read_only", table_name="water_temperature", + sub_path="2026").run() + + It applies to a **read-only** ingest whose files are stored permanently; the + server refuses it otherwise rather than filing the data somewhere else. A + folder that already has the structure locally needs nothing: ``scan_folder`` + keeps sub-folders and the server preserves them. + """ def __init__( self, @@ -85,6 +133,7 @@ def __init__( intent: Intent = "read_only", conflict_mode: str = "fail", table_name: str | None = None, + sub_path: str | None = None, parallelism: int = DEFAULT_PARALLELISM, idempotency_key: str | None = None, multipart: bool | None = None, @@ -103,11 +152,17 @@ def __init__( self.intent = intent self.conflict_mode = conflict_mode self.table_name = table_name + self.sub_path = sub_path self.parallelism = parallelism self.idempotency_key = idempotency_key self.multipart = multipart self.show_progress = show_progress + # Set once ``begin`` runs (or by ``attach``); the handle every session + # command below works from. + self._session_id: str | None = None + self._begin: BeginResult | None = None + # The client owns the credentials; if the caller did not inject one we # build it from the kernel environment. Creds never leave the client. if client is not None: @@ -119,51 +174,217 @@ def __init__( self._client = IngestClient(resolved_url, resolved_creds) self._owns_client = True + # -- session handle --------------------------------------------------- + + @property + def session_id(self) -> str | None: + """Id of the transfer this object is driving, once ``begin`` has run.""" + return self._session_id + + @classmethod + def attach( + cls, + session_id: str, + folder: str | Path, + target_catalog_path: str, + *, + data_format: DataFormat, + **kwargs: Any, + ) -> FolderIngest: + """Bind to an EXISTING session instead of starting a new one. + + For picking a transfer back up in a later kernel — inspect it with + :meth:`status`, resume it with :meth:`retry`, or abandon it with + :meth:`cancel`. ``folder`` still has to point at the same local data, so + a re-upload is possible if the staged copy is gone. + """ + job = cls(folder, target_catalog_path, data_format=data_format, **kwargs) + job._session_id = session_id + return job + + def close(self) -> None: + """Release the HTTP client, if this object created it.""" + if self._owns_client: + self._client.close() + + def __enter__(self) -> FolderIngest: + return self + + def __exit__(self, exc_type: object, exc: object, tb: object) -> None: + self.close() + # -- orchestration ---------------------------------------------------- def run(self) -> IngestOutcome: - """Execute the full begin → upload → commit flow.""" + """Execute the full begin → upload → commit flow. - try: - files = scan_folder(self.folder, self.data_format) - if not files: - raise FileNotFoundError( - f"no {self.data_format} files found under {self.folder}" - ) - begin = self._client.begin( - target_catalog_path=self.target_catalog_path, - intent=self.intent, - data_format=self.data_format, - conflict_mode=self.conflict_mode, - files=files, - table_name=self.table_name, - idempotency_key=self.idempotency_key, - multipart=self.multipart, + The client is left open afterwards so the same object can still + :meth:`status`, :meth:`retry` or :meth:`cancel` the transfer. Use it as a + context manager (or call :meth:`close`) to release it. + """ + begin = self.begin() + etags, multipart_etags, uploaded, skipped = self._upload_all(begin) + commit = self.commit(multipart_etags=multipart_etags or None) + return IngestOutcome( + begin=begin, + commit=commit, + files_uploaded=uploaded, + files_skipped=skipped, + etags=etags, + ) + + def begin(self) -> BeginResult: + """Open the transfer and get the presigned upload targets (step 1 of 3). + + The server validates the target here — an unauthorised or non-existent + catalog folder fails now, before a single byte is uploaded. + """ + files = scan_folder(self.folder, self.data_format) + if not files: + raise FileNotFoundError( + f"no {self.data_format} files found under {self.folder}" ) - etags, multipart_etags, uploaded, skipped = self._upload_all(begin) - commit = self._client.commit( - session_id=begin.session_id, - multipart_etags=multipart_etags or None, + begin = self._client.begin( + target_catalog_path=self.target_catalog_path, + intent=self.intent, + data_format=self.data_format, + conflict_mode=self.conflict_mode, + files=files, + table_name=self.table_name, + sub_path=self.sub_path, + idempotency_key=self.idempotency_key, + multipart=self.multipart, + ) + self._session_id = begin.session_id + self._begin = begin + return begin + + def commit( + self, *, multipart_etags: list[dict[str, object]] | None = None + ) -> CommitResult: + """Finalise the transfer: load the staged files into the table (step 3). + + This is where the Dremio work happens (``CREATE TABLE`` + ``COPY INTO`` + on a managed catalog), so it is the step most likely to fail — see + :meth:`retry`. + """ + if self._session_id is None: + raise IngestStateError("no session to commit — call begin() or run() first") + return self._client.commit( + session_id=self._session_id, multipart_etags=multipart_etags + ) + + # -- session management ----------------------------------------------- + + def status(self) -> StatusResult: + """Current server-side state of this transfer.""" + if self._session_id is None: + raise IngestStateError("no session yet — call begin() or run() first") + return self._client.get_status(self._session_id) + + def estimate(self) -> EstimateResult: + """Row count + size class of the staged data, between upload and commit.""" + if self._session_id is None: + raise IngestStateError("no session yet — call begin() or run() first") + return self._client.estimate(self._session_id) + + def cancel(self) -> None: + """Abandon this transfer and delete its staged data. + + Does not drop a table an earlier successful commit already created. + """ + if self._session_id is None: + raise IngestStateError("no session to cancel") + self._client.cancel(self._session_id) + self._session_id = None + + def retry(self) -> IngestOutcome: + """Re-run this transfer from the step that failed. + + Two cases, decided by the server's own report of where it broke: + + * **The load failed** (``CREATE TABLE`` / ``COPY INTO``) and the staged + files were kept — the common case, e.g. Dremio was briefly unavailable + or the target folder was fixed afterwards. Only that step re-runs; the + upload is NOT repeated. + * **Anything else** — the upload never finished, or the failure was one a + re-run cannot fix (an unresolved collision, an unreadable file), so the + staged copy is gone. A fresh session is opened and the folder is + uploaded again. Any ``idempotency_key`` is dropped for that attempt, or + the server would just replay the failed session. + + **A transfer whose files are stored permanently is never re-uploaded + blindly.** There the upload landed in the table's own folder and stayed, + so a fresh session would add a *second* copy — the server numbers an + incoming name that already exists, precisely so an append can never + overwrite live data, and that protection turns a silent re-run into + duplicated rows. Such a transfer is resumable server-side by design, so + the first branch handles it; if the server says it is not, this raises + rather than guessing. + + Raises :class:`IngestStateError` if there is nothing to retry — no + session, one that is still running, or one that already succeeded. + """ + if self._session_id is None: + raise IngestStateError("no session to retry — call run() first") + status = self._client.get_status(self._session_id) + if status.status == "done": + raise IngestStateError( + f"transfer {self._session_id} already succeeded; nothing to retry" + ) + if status.status in ("pending", "uploading", "committing"): + raise IngestStateError( + f"transfer {self._session_id} is {status.status}; wait for it to " + "finish before retrying" ) + if status.is_resumable: + logger.info("resuming the load step of %s", self._session_id) + resumed = self._client.retry(self._session_id) return IngestOutcome( - begin=begin, - commit=commit, - files_uploaded=uploaded, - files_skipped=skipped, - etags=etags, + begin=self._begin, + commit=CommitResult.from_json(resumed.raw), + files_uploaded=0, + files_skipped=len(self._begin.s3.uploads) if self._begin else 0, + resumed=True, ) - finally: - if self._owns_client: - self._client.close() + if status.stores_permanently: + raise IngestStateError( + f"transfer {self._session_id} stored its files permanently and " + "cannot be resumed, so re-running it would upload a second copy " + "alongside the first. Inspect the table, then re-ingest " + "deliberately — with conflict_mode='replace' to redo it, or a " + "sub_path for data that belongs beside what is already there." + ) + # Nothing to resume server-side: start over with a new session, which + # means dropping the idempotency key that would replay the failed one. + logger.info( + "transfer %s cannot be resumed (%s); re-uploading under a new session", + self._session_id, + status.raw.get("failed_stage") or status.status, + ) + self._session_id = None + self._begin = None + self.idempotency_key = None + return self.run() # -- upload phase ----------------------------------------------------- def _already_done(self, session_id: str) -> set[str]: """Return rel_paths already uploaded for this session (resume support). - Queries the session status; the server reports ``files_done`` and may - list completed rel_paths under ``raw["uploaded"]``. We only skip files - the server explicitly names, so resume never wrongly drops a file. + Queries the session status and skips only the files the server + explicitly names under ``raw["uploaded"]``, so resume never wrongly drops + a file. NOTE: the server populates that list for **server-run** transfers + (``POST /ingest/folder``); a client-uploaded session reports nothing + there, so re-running one uploads every file again. + + Re-running the SAME session is harmless — the keys are identical, so the + second upload overwrites the first. Starting a NEW session is a different + matter where the files are stored permanently: the server numbers an + incoming name that already exists, precisely so an append cannot + overwrite live data, and that turns a re-run into a second copy rather + than an overwrite. :meth:`retry` refuses exactly that case; prefer + :meth:`FolderIngest.attach` + :meth:`retry` over re-running :meth:`run`. """ try: @@ -242,11 +463,15 @@ def ingest_folder( data_format: DataFormat, **kwargs: object, ) -> IngestOutcome: - """Convenience wrapper: build a :class:`FolderIngest` and run it.""" + """Convenience wrapper: build a :class:`FolderIngest`, run it, close it. - return FolderIngest( + One-shot. If the transfer might need a :meth:`FolderIngest.retry`, drive the + class directly (ideally as a context manager) so the session handle survives. + """ + with FolderIngest( folder, target_catalog_path, data_format=data_format, **kwargs, # type: ignore[arg-type] - ).run() + ) as job: + return job.run() diff --git a/src/eea_datalakehouse/dds_ingestion/models.py b/src/eea_datalakehouse/dds_ingestion/models.py index 1ed1da9..04ea0ab 100644 --- a/src/eea_datalakehouse/dds_ingestion/models.py +++ b/src/eea_datalakehouse/dds_ingestion/models.py @@ -113,12 +113,20 @@ def from_json(cls, data: dict[str, Any]) -> BeginResult: @dataclass(frozen=True, slots=True) class CommitResult: - """Response from ``POST /api/v1/ingest/commit``.""" + """Response from ``POST /api/v1/ingest/commit``. + + ``table_path`` is where the table is **queried** — the catalog path. For a + read-only ingest whose files are stored permanently, ``storage_path`` is + where those files physically **are** (the Dremio path of the promoted + folder); it is ``None`` for a staged ingest, whose upload was copied into the + catalog's own storage and deleted, and against any server predating it. + """ session_id: str status: str table_path: str | None = None record_count: int | None = None + storage_path: str | None = None @classmethod def from_json(cls, data: dict[str, Any]) -> CommitResult: @@ -127,6 +135,7 @@ def from_json(cls, data: dict[str, Any]) -> CommitResult: status=data["status"], table_path=data.get("table_path"), record_count=data.get("record_count"), + storage_path=data.get("storage_path"), ) @@ -147,7 +156,12 @@ def from_json(cls, data: dict[str, Any]) -> Progress: @dataclass(frozen=True, slots=True) class StatusResult: - """Response from ``GET /api/v1/ingest/{session_id}``.""" + """Response from ``GET /api/v1/ingest/{session_id}`` (and the list/retry calls). + + ``status`` is the lifecycle state — ``pending`` → ``uploading`` → + ``committing`` → ``done``, or ``failed`` / ``cancelled``. ``raw`` keeps the + whole payload so fields the client does not model yet stay reachable. + """ status: str progress: Progress @@ -160,3 +174,98 @@ def from_json(cls, data: dict[str, Any]) -> StatusResult: progress=Progress.from_json(data.get("progress") or {}), raw=data, ) + + @property + def session_id(self) -> str | None: + value = self.raw.get("session_id") + return str(value) if value is not None else None + + @property + def table_path(self) -> str | None: + value = self.raw.get("table_path") + return str(value) if value is not None else None + + @property + def record_count(self) -> int | None: + value = self.raw.get("record_count") + return int(value) if value is not None else None + + @property + def placement(self) -> str: + """Where this transfer's files live: ``staged`` or ``read_permanent``. + + ``staged`` (the default, and what a server without DI-11 reports by + omitting the field) means the upload was copied into the catalog and + deleted. ``read_permanent`` means the files were stored where the table + lives and kept — so re-uploading them is not a safe way to recover. + """ + value = self.raw.get("placement") + return str(value) if value else "staged" + + @property + def stores_permanently(self) -> bool: + """Whether this transfer's uploaded files ARE the table (DI-11).""" + return self.placement == "read_permanent" + + @property + def error(self) -> str | None: + """Why the transfer failed, including the stage — e.g. ``"Dremio load + failed: ..."`` or ``"S3 upload failed: ..."``. ``None`` unless failed.""" + value = self.raw.get("error") + return str(value) if value is not None else None + + @property + def is_terminal(self) -> bool: + """Whether the session has finished, one way or another.""" + return self.status in ("done", "failed", "cancelled") + + @property + def is_resumable(self) -> bool: + """Whether :meth:`FolderIngest.retry` can re-run this transfer. + + A failed transfer is resumable when the server kept its staged files — + which it does when the upload landed and only the Dremio load failed on + a managed catalog. Anything else has to be uploaded again. + + This reads the server's own ``resumable`` verdict rather than re-deriving + it from ``failed_stage``: a load-stage failure the server could not hold + the bytes for reports ``failed_stage="load"`` too, and inferring from that + sent :meth:`~eea_datalakehouse.dds_ingestion.folder.FolderIngest.retry` + down the resume path to be told the staged data was gone — instead of + simply re-uploading. A server that does not send the field reads as not + resumable, which is the safe direction: the transfer is re-uploaded + under a new session. + """ + return self.status == "failed" and bool(self.raw.get("resumable")) + + +@dataclass(frozen=True, slots=True) +class EstimateResult: + """Response from ``POST /api/v1/ingest/estimate`` (pre-flight sizing).""" + + record_count: int + size_class: str + + @classmethod + def from_json(cls, data: dict[str, Any]) -> EstimateResult: + return cls( + record_count=int(data.get("record_count", 0)), + size_class=str(data.get("size_class", "")), + ) + + +@dataclass(frozen=True, slots=True) +class StageResult: + """Response from ``POST /api/v1/ingest/stage`` (server-proxied upload).""" + + rel_path: str + key: str + bytes_written: int | None = None + + @classmethod + def from_json(cls, data: dict[str, Any]) -> StageResult: + return cls( + rel_path=str(data["rel_path"]), + key=str(data["key"]), + bytes_written=data.get("bytes"), + ) diff --git a/tests/dds_ingestion/test_client.py b/tests/dds_ingestion/test_client.py index b260448..723aaaa 100644 --- a/tests/dds_ingestion/test_client.py +++ b/tests/dds_ingestion/test_client.py @@ -3,7 +3,11 @@ import httpx import pytest import respx -from eea_datalakehouse.dds_ingestion.client import IngestApiError, IngestClient +from eea_datalakehouse.dds_ingestion.client import ( + IngestApiError, + IngestClient, + S3UploadError, +) from eea_datalakehouse.dds_ingestion.credentials import DremioCreds from eea_datalakehouse.dds_ingestion.models import FileSpec, UploadPart, UploadTarget @@ -175,3 +179,127 @@ def test_error_response_raises_with_message(creds: DremioCreds) -> None: ) assert exc.value.status_code == 409 assert "exists" in str(exc.value) + + +@respx.mock +def test_dds_error_names_the_call_that_failed(creds: DremioCreds) -> None: + """A 500 with Starlette's bare body must still say WHICH call broke. + + The message an ingest reported was "DDS ingest API error 500: Internal + Server Error" — true, and useless: a transfer calls begin, upload and commit + against two different systems. + """ + respx.post(f"{BASE_URL}/api/v1/ingest/begin").mock( + return_value=httpx.Response(500, text="Internal Server Error") + ) + with IngestClient(BASE_URL, creds) as client, pytest.raises(IngestApiError) as exc: + client.begin( + target_catalog_path="x", + intent="read_only", + data_format="parquet", + conflict_mode="fail", + files=[FileSpec("a.parquet", 6)], + ) + + assert exc.value.where == "POST /api/v1/ingest/begin" + assert "POST /api/v1/ingest/begin" in str(exc.value) + assert "Internal Server Error" in str(exc.value) + + +@respx.mock +def test_failed_presigned_upload_is_not_reported_as_a_dds_error( + creds: DremioCreds, +) -> None: + """The upload goes straight to object storage; its failures are its own.""" + respx.post("https://s3.example/bucket").mock( + return_value=httpx.Response(500, text="gateway said no") + ) + target = UploadTarget( + rel_path="a.parquet", url="https://s3.example/bucket", method="POST" + ) + with IngestClient(BASE_URL, creds) as client, pytest.raises(S3UploadError) as exc: + client.upload_file(target, b"PAR1") + + assert exc.value.rel_path == "a.parquet" + assert exc.value.url == "https://s3.example/bucket" + message = str(exc.value) + assert message.startswith("S3 upload of 'a.parquet' failed: HTTP 500") + assert "https://s3.example/bucket" in message + assert "DDS ingest API error" not in message + # …and it is still an IngestApiError, so existing handling keeps working. + assert isinstance(exc.value, IngestApiError) + + +@respx.mock +def test_long_error_bodies_are_truncated(creds: DremioCreds) -> None: + respx.post(f"{BASE_URL}/api/v1/ingest/commit").mock( + return_value=httpx.Response(502, text="x" * 5000) + ) + with IngestClient(BASE_URL, creds) as client, pytest.raises(IngestApiError) as exc: + client.commit(session_id="sess-1") + + assert exc.value.message.endswith("… (truncated)") + assert len(exc.value.message) < 600 + + +@respx.mock +def test_begin_sends_sub_path_only_when_given(creds: DremioCreds) -> None: + """DI-11.12: the named sub-folder rides on ``begin`` and nowhere else. + + Omitted when unset, so a client that never uses it sends the body it always + sent — an older server sees no new field. + """ + import json + + route = respx.post(f"{BASE_URL}/api/v1/ingest/begin").mock( + return_value=httpx.Response( + 200, + json={ + "session_id": "sess-1", + "status": "open", + "s3": {"bucket": "b", "key_prefix": "p/", "uploads": []}, + "collision": None, + }, + ) + ) + kwargs: dict[str, object] = { + "target_catalog_path": "bio.uploads", + "intent": "read_only", + "data_format": "parquet", + "conflict_mode": "append", + "files": [FileSpec("a.parquet", 6)], + } + with IngestClient(BASE_URL, creds) as client: + client.begin(**kwargs) # type: ignore[arg-type] + assert "sub_path" not in json.loads(route.calls.last.request.read()) + + client.begin(sub_path="2026", **kwargs) # type: ignore[arg-type] + assert json.loads(route.calls.last.request.read())["sub_path"] == "2026" + + +def test_commit_result_carries_the_physical_location() -> None: + """DI-11.9: ``storage_path`` says where permanently-stored files actually are. + + ``table_path`` is where the table is queried; the two differ for a read-only + ingest that keeps its files. Absent (a staged ingest, or a server predating + the field) parses as ``None`` rather than failing — the models are + deliberately permissive about keys they do not know. + """ + from eea_datalakehouse.dds_ingestion.models import CommitResult + + kept = CommitResult.from_json( + { + "session_id": "s1", + "status": "done", + "table_path": "catalog/water/bwd/assessments", + "record_count": 12, + "storage_path": "local_s3/dh-prod-data/read/water/bwd/assessments", + } + ) + assert kept.table_path == "catalog/water/bwd/assessments" + assert kept.storage_path == "local_s3/dh-prod-data/read/water/bwd/assessments" + + staged = CommitResult.from_json( + {"session_id": "s2", "status": "done", "table_path": "x/y", "record_count": 1} + ) + assert staged.storage_path is None diff --git a/tests/dds_ingestion/test_folder.py b/tests/dds_ingestion/test_folder.py index 99fcacd..c2b6025 100644 --- a/tests/dds_ingestion/test_folder.py +++ b/tests/dds_ingestion/test_folder.py @@ -6,7 +6,11 @@ from pathlib import Path import pytest -from eea_datalakehouse.dds_ingestion.folder import FolderIngest, scan_folder +from eea_datalakehouse.dds_ingestion.folder import ( + FolderIngest, + IngestStateError, + scan_folder, +) from eea_datalakehouse.dds_ingestion.models import ( BeginResult, CommitResult, @@ -28,6 +32,9 @@ def __init__(self, *, already_uploaded: list[str] | None = None) -> None: self.commit_calls: list[dict[str, object]] = [] self.closed = False self._already_uploaded = already_uploaded or [] + # Set to drive ``get_status`` from a test (e.g. a failed, unresumable + # session); ``None`` keeps the default "open" shape. + self.status_payload: dict[str, object] | None = None # concurrency instrumentation self._lock = threading.Lock() self._active = 0 @@ -46,6 +53,8 @@ def begin(self, *, files: list[FileSpec], **kwargs: object) -> BeginResult: ) def get_status(self, session_id: str) -> StatusResult: + if self.status_payload is not None: + return StatusResult.from_json(self.status_payload) return StatusResult.from_json( { "status": "open", @@ -204,3 +213,86 @@ def test_multipart_etags_threaded_into_commit(data_folder: Path) -> None: entry = by_rel["a.parquet"] assert entry["upload_id"] == "up-a.parquet" assert entry["parts"] == [{"part_number": 1, "etag": '"etag-a.parquet"'}] + + +def test_sub_path_is_passed_through_to_begin(tmp_path: Path) -> None: + """DI-11.12: the class carries the custodian's chosen sub-folder, nothing more. + + Normalising and scoping it is the server's job — the client would otherwise + be a second opinion about where the data lives. + """ + (tmp_path / "a.parquet").write_bytes(b"PAR1") + client = FakeClient() + job = FolderIngest( + tmp_path, + "catalog/water/bwd", + data_format="parquet", + table_name="water_temperature", + sub_path="2026", + client=client, + show_progress=False, + ) + job.run() + assert client.begin_calls[0]["sub_path"] == "2026" + + +def test_sub_path_defaults_to_none(tmp_path: Path) -> None: + (tmp_path / "a.parquet").write_bytes(b"PAR1") + client = FakeClient() + FolderIngest( + tmp_path, "catalog/water/bwd", data_format="parquet", + client=client, show_progress=False, + ).run() + assert client.begin_calls[0]["sub_path"] is None + + +def test_retry_refuses_to_re_upload_a_permanently_stored_transfer( + tmp_path: Path, +) -> None: + """DI-11.7: a blind re-run there adds a second copy, so it must not happen. + + The server numbers an incoming name that already exists — precisely so an + append can never overwrite live data — which turns a silent "start over" + into duplicated rows rather than a clean retry. + """ + (tmp_path / "a.parquet").write_bytes(b"PAR1") + client = FakeClient() + client.status_payload = { + "status": "failed", + "progress": {}, + "failed_stage": "load", + "resumable": False, + "placement": "read_permanent", + } + job = FolderIngest( + tmp_path, "catalog/water/bwd", data_format="parquet", + intent="read_only", client=client, show_progress=False, + ) + job._session_id = "sess-1" + + with pytest.raises(IngestStateError) as excinfo: + job.retry() + assert "second copy" in str(excinfo.value) + # And nothing was re-uploaded behind the user's back. + assert client.begin_calls == [] + assert client.uploaded == [] + + +def test_retry_still_starts_over_for_a_staged_transfer(tmp_path: Path) -> None: + # The staged path is unchanged: its files were discarded, so re-uploading is + # the only way forward and is safe. + (tmp_path / "a.parquet").write_bytes(b"PAR1") + client = FakeClient() + client.status_payload = { + "status": "failed", + "progress": {}, + "failed_stage": "upload", + "resumable": False, + } + job = FolderIngest( + tmp_path, "catalog/water/bwd", data_format="parquet", + client=client, show_progress=False, + ) + job._session_id = "sess-1" + job.retry() + assert len(client.begin_calls) == 1 diff --git a/tests/dds_ingestion/test_session.py b/tests/dds_ingestion/test_session.py new file mode 100644 index 0000000..5acb601 --- /dev/null +++ b/tests/dds_ingestion/test_session.py @@ -0,0 +1,277 @@ +"""Session management on :class:`FolderIngest` — status, retry, cancel, attach. + +The important behaviour is :meth:`FolderIngest.retry`, which resumes at the step +that failed. The server tells it which step that was; these tests pin both +branches: + +* the load failed and the staged files were kept → re-run ONLY the load, no + re-upload; +* anything else → the staged copy is gone, so upload again under a NEW session + (and drop the idempotency key, or the server would replay the failed one). +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest + +from eea_datalakehouse.dds_ingestion.folder import FolderIngest, IngestStateError +from eea_datalakehouse.dds_ingestion.models import ( + BeginResult, + CommitResult, + S3Plan, + StatusResult, + UploadTarget, +) + + +class SessionClient: + """Fake IngestClient that serves a scripted sequence of session states.""" + + def __init__(self, *statuses: dict[str, Any]) -> None: + self._statuses = list(statuses) + self.begin_calls: list[dict[str, Any]] = [] + self.commit_calls: list[str] = [] + self.retry_calls: list[str] = [] + self.cancel_calls: list[str] = [] + self.uploaded: list[str] = [] + self.closed = False + self._next_session = 0 + + # -- endpoints used by FolderIngest -- + def begin(self, *, files: list[Any], **kwargs: Any) -> BeginResult: + self.begin_calls.append({"files": files, **kwargs}) + self._next_session += 1 + return BeginResult( + session_id=f"sess-{self._next_session}", + status="uploading", + s3=S3Plan( + bucket="b", + key_prefix="p/", + uploads=tuple( + UploadTarget(rel_path=f.rel_path, url=f"https://s3.test/{f.rel_path}") + for f in files + ), + ), + ) + + def get_status(self, session_id: str) -> StatusResult: + payload = self._statuses.pop(0) if self._statuses else {"status": "done"} + return StatusResult.from_json({"session_id": session_id, **payload}) + + def upload_file(self, target: UploadTarget, data: bytes) -> str | None: + self.uploaded.append(target.rel_path) + return f'"etag-{target.rel_path}"' + + def commit(self, *, session_id: str, **kwargs: Any) -> CommitResult: + self.commit_calls.append(session_id) + return CommitResult( + session_id=session_id, status="done", table_path="t", record_count=2 + ) + + def retry(self, session_id: str) -> StatusResult: + self.retry_calls.append(session_id) + return StatusResult.from_json( + { + "session_id": session_id, + "status": "done", + "table_path": "bio.uploads.t", + "record_count": 7, + } + ) + + def cancel(self, session_id: str) -> None: + self.cancel_calls.append(session_id) + + def close(self) -> None: + self.closed = True + + +def _job(folder: Path, client: SessionClient, **kwargs: Any) -> FolderIngest: + return FolderIngest( + folder, + "bio.uploads", + data_format="parquet", + show_progress=False, + client=client, # type: ignore[arg-type] + **kwargs, + ) + + +# --- the session handle ------------------------------------------------------ + + +def test_session_id_is_exposed_after_begin(data_folder: Path) -> None: + client = SessionClient() + job = _job(data_folder, client) + assert job.session_id is None + job.begin() + assert job.session_id == "sess-1" + + +def test_attach_binds_to_an_existing_session(data_folder: Path) -> None: + client = SessionClient({"status": "failed", "failed_stage": "load"}) + job = FolderIngest.attach( + "sess-earlier", + data_folder, + "bio.uploads", + data_format="parquet", + show_progress=False, + client=client, + ) + assert job.session_id == "sess-earlier" + assert job.status().status == "failed" + + +def test_commit_before_begin_is_a_state_error(data_folder: Path) -> None: + job = _job(data_folder, SessionClient()) + with pytest.raises(IngestStateError): + job.commit() + + +def test_cancel_discards_the_session(data_folder: Path) -> None: + client = SessionClient() + job = _job(data_folder, client) + job.begin() + job.cancel() + assert client.cancel_calls == ["sess-1"] + assert job.session_id is None + + +# --- retry ------------------------------------------------------------------- + + +def test_retry_resumes_the_load_without_re_uploading(data_folder: Path) -> None: + # The stranded-transfer case: the upload landed, Dremio failed, the server + # kept the staged files. Only the load re-runs. + client = SessionClient( + { + "status": "failed", + "failed_stage": "load", + "resumable": True, + "error": "Dremio load failed: boom", + } + ) + job = _job(data_folder, client) + job.begin() + client.uploaded.clear() + + outcome = job.retry() + + assert client.retry_calls == ["sess-1"] + assert client.uploaded == [] # nothing re-uploaded + assert client.begin_calls == [client.begin_calls[0]] # no new session + assert outcome.resumed is True + assert outcome.commit.record_count == 7 + + +def test_retry_re_uploads_a_load_failure_that_kept_nothing(data_folder: Path) -> None: + # A load-stage failure the server could NOT hold the bytes for still reports + # failed_stage="load". Inferring resumability from that sent retry down the + # resume path to be told the staged data was gone; the server's own verdict + # sends it to re-upload instead. + client = SessionClient( + { + "status": "failed", + "failed_stage": "load", + "resumable": False, + "error": "Dremio load failed: 'x.parquet' could not be read as Parquet", + } + ) + job = _job(data_folder, client) + job.begin() + client.uploaded.clear() + + outcome = job.retry() + + assert client.retry_calls == [] # server retry not attempted + assert len(client.begin_calls) == 2 # a second session + assert sorted(client.uploaded) == ["a.parquet", "sub/b.parquet"] + assert outcome.resumed is False + + +def test_retry_after_an_upload_failure_starts_a_fresh_session(data_folder: Path) -> None: + # The staged files were cleaned up, so there is nothing to resume: upload + # again under a new session. + client = SessionClient( + {"status": "failed", "failed_stage": "upload", "error": "S3 upload failed"} + ) + job = _job(data_folder, client) + job.begin() + client.uploaded.clear() + + outcome = job.retry() + + assert client.retry_calls == [] # server retry not attempted + assert len(client.begin_calls) == 2 # a second session + assert sorted(client.uploaded) == ["a.parquet", "sub/b.parquet"] + assert outcome.resumed is False + assert job.session_id == "sess-2" + + +def test_retry_drops_the_idempotency_key_when_re_uploading(data_folder: Path) -> None: + # Reusing the key would make the server replay the SAME failed session, which + # is exactly the loop that made a retry look like it did nothing. + client = SessionClient({"status": "failed", "failed_stage": "upload"}) + job = _job(data_folder, client, idempotency_key="key-1") + job.begin() + assert client.begin_calls[0]["idempotency_key"] == "key-1" + + job.retry() + + assert client.begin_calls[1]["idempotency_key"] is None + + +@pytest.mark.parametrize("state", ["pending", "uploading", "committing"]) +def test_retry_refuses_while_still_running(data_folder: Path, state: str) -> None: + client = SessionClient({"status": state}) + job = _job(data_folder, client) + job.begin() + with pytest.raises(IngestStateError, match=state): + job.retry() + + +def test_retry_refuses_a_finished_transfer(data_folder: Path) -> None: + client = SessionClient({"status": "done"}) + job = _job(data_folder, client) + job.begin() + with pytest.raises(IngestStateError, match="already succeeded"): + job.retry() + + +def test_retry_without_a_session_is_a_state_error(data_folder: Path) -> None: + job = _job(data_folder, SessionClient()) + with pytest.raises(IngestStateError): + job.retry() + + +# --- status flags ------------------------------------------------------------ + + +def test_is_resumable_follows_the_servers_verdict() -> None: + kept = StatusResult.from_json( + {"status": "failed", "failed_stage": "load", "resumable": True} + ) + # Same stage, but the server cleaned up: NOT resumable. Deriving this from + # failed_stage alone is what made retry ask to resume data that was gone. + cleaned = StatusResult.from_json( + {"status": "failed", "failed_stage": "load", "resumable": False} + ) + swept = StatusResult.from_json({"status": "failed", "failed_stage": "upload"}) + running = StatusResult.from_json({"status": "committing", "resumable": True}) + # A server too old to send the field reads as not resumable — the safe way to + # be wrong, since the transfer is simply re-uploaded. + silent = StatusResult.from_json({"status": "failed", "failed_stage": "load"}) + assert kept.is_resumable is True + assert cleaned.is_resumable is False + assert swept.is_resumable is False + assert running.is_resumable is False + assert silent.is_resumable is False + + +def test_is_terminal_covers_every_end_state() -> None: + for state in ("done", "failed", "cancelled"): + assert StatusResult.from_json({"status": state}).is_terminal is True + assert StatusResult.from_json({"status": "uploading"}).is_terminal is False