diff --git a/CHANGELOG.md b/CHANGELOG.md index b11275b..86e1fdc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,14 @@ Notable changes to Loafer are documented here. This project follows requests, execution plans, batch envelopes, events, snapshots, and results. - Runtime ports and local adapters for cancellation, checkpoints, secret resolution, event publication, and interactive transform review. +- An opt-in bounded row-local ETL data plane with per-batch envelopes, validation, schema + versioning, quarantine output, rolling row/byte/checksum reconciliation, cancellation + boundaries, and atomic CSV/JSON publication. +- Explicit `fail`, `evolve`, `quarantine`, and `coerce` schema-drift policies plus required-column + and column-type validation. +- Native PDF text/table provenance, file/page/time limits, and configurable page failure handling. +- Run-scoped PostgreSQL staging with transactional replace, create-once, append, and keyed-upsert + publication plus an explicit delivery guarantee in validated execution plans. ### Changed @@ -19,6 +27,29 @@ Notable changes to Loafer are documented here. This project follows execution orchestration remains independent of client frameworks. - Durable application contracts now exclude credentials, connector instances, iterators, provider clients, row payloads, and other ephemeral runtime objects. +- AI row-local transforms now generate and version one validated artifact per run and reuse it for + every bounded batch. +- SQL transforms are classified as global relational work, and the volume benchmark now exercises + the declared row-local path. + +### Fixed + +- Cancellation, transform failures, and target failures during bounded file runs now discard + run-scoped temporary output instead of publishing a final partial file. +- CSV encoding detection now scans in bounded chunks instead of allocating the entire source file + during connection. +- The Linux process-tree benchmark now tolerates sandbox workers exiting during `/proc` sampling + instead of aborting on the normal `ESRCH` race, and treats absent Git tooling in production + images as optional provenance rather than a benchmark failure. + +### Known limitations + +- MongoDB row-local runs remain rejected until a tested staging/merge protocol replaces direct + partial batch effects. PostgreSQL append is intentionally at-least-once across an ambiguous + target-commit/checkpoint gap; keyed upsert is the replay-safe merge mode. +- Undeclared/materialized transforms and local SQL ETL still retain full-run state. The bounded + path passed the clean production-image 30M-row gate at 118.23 MiB peak process-tree RSS. +- PDF extraction supports native text and tables; OCR remains unimplemented. ## [0.4.0] - 2026-07-29 diff --git a/PRODUCTION_READINESS.md b/PRODUCTION_READINESS.md index fa76634..85d5b6f 100644 --- a/PRODUCTION_READINESS.md +++ b/PRODUCTION_READINESS.md @@ -77,11 +77,13 @@ operational envelope. ## Release blockers for 30–100M+ ETL -### P0 — End-to-end materialization +### P0 — End-to-end materialization outside the declared row-local path `loafer/transform/__init__.py::materialize_input_rows` drains every source chunk into one list. -AI, custom Python, SQL, and multi-step ETL runners use it. `loafer/agents/load.py` then writes from -`transformed_data`, which is also a full-run list. +Undeclared/materialized AI, custom Python, SQL, and multi-step ETL runners still use it. +`loafer/agents/load.py` then writes from `transformed_data`, which is also a full-run list. Declared +`row_local` custom, AI, and custom/AI pipelines bypass this graph and keep one bounded batch in +flight. Impact: @@ -118,11 +120,12 @@ Required correction: - Give workers leases and cooperative cancellation at batch boundaries. - Advance checkpoints only after the corresponding target effect is durable. -### P0 — Database partial outputs can appear final +### P0 — Database partial outputs outside staged targets can appear final CSV and JSON targets now publish atomically and discard unpublished temporary files on failure. -PostgreSQL still commits every small insert batch, and its target adapter commits table/index -creation separately. +Declared row-local PostgreSQL runs now commit batches only to a hidden run-scoped table and publish +with one final transaction. The legacy direct PostgreSQL adapter still commits every small insert +batch, and MongoDB has no equivalent staging protocol. Impact: @@ -178,6 +181,14 @@ uv run python benchmarks/full_pipeline.py \ --rows 10000000 \ --rss-limit-mb 2048 \ --report benchmarks/results/10m.json + +uv run python benchmarks/full_pipeline.py \ + --rows 30000000 \ + --chunk-size 10000 \ + --rss-limit-mb 512 \ + --sandbox-memory-mb 256 \ + --timeout-seconds 3600 \ + --report benchmarks/results/30m-row-local.json ``` The harness generates deterministic input, executes the real CLI and transform subprocess, samples @@ -195,27 +206,32 @@ the current custom identity transform: |---:|---:|---:|---|---| | 1,000,000 | 13.36s | 1,310.1 MiB | exact row count/SHA-256 | [`1m.json`](benchmarks/results/1m.json) | | 10,000,000 | 19.54s before cutoff | 2,056.7 MiB | terminated at 2 GiB; no final/temp output | [`10m.json`](benchmarks/results/10m.json) | +| 30,000,000 row-local | 2,028.26s | 118.23 MiB | exact row count/SHA-256; clean production image | [`30m-row-local.json`](benchmarks/results/30m-row-local.json) | Input generation time is excluded from pipeline wall time. These results demonstrate full-run -materialization rather than a bounded-memory curve. Environment, image, limits, and interpretation -are recorded with the [versioned benchmark artifacts](benchmarks/results/README.md). +materialization in the Phase 0 path and bounded memory in the declared Phase 2 row-local path. +Environment, limits, provenance caveats, and interpretation are recorded with the +[versioned benchmark artifacts](benchmarks/results/README.md). ## High-priority production gaps -### P1 — Data quality is sample-based +### P1 — Data quality is complete only on the declared row-local path -Validation is primarily computed from the schema sample. It cannot prove whole-run null rates, -type consistency, uniqueness, referential integrity, or rejected-row counts. +Declared row-local runs validate every batch and aggregate column/null/rejected metrics with +quarantine output. The legacy materialized graph remains primarily schema-sample based, and neither +path yet provides declared uniqueness or referential-integrity checks. -Build batch-level validation with aggregated run metrics and a quarantine output. +Extend the same contract to global/materialized plans and add uniqueness and referential-integrity +policies where their storage requirements are explicit. -### P1 — Schema drift policy is implicit +### P1 — Schema drift policy remains implicit outside row-local execution -Target schema is inferred from the first row/chunk. Later columns and type changes can fail or be -silently coerced depending on the adapter. +Declared row-local runs expose schema versions and `fail | evolve | quarantine | coerce` policy. +Legacy target behavior is still inferred from the first row/chunk, so later columns and type +changes can fail or be silently coerced depending on the adapter. -Add declared contracts, schema versions, and explicit `fail | evolve | quarantine | coerce` -policies. +Extend the declared schema contract to global/materialized execution and add compatibility tests +for representative wide and nested records. ### P1 — The sandbox is a resource limiter, not a complete isolation boundary @@ -370,6 +386,29 @@ Exit gate: - cancellation and target failure do not publish a false success or final partial output; - native PDF text/table fixtures prove page provenance, limits, and failure reporting. +**Current status:** implementation and clean production-image verification complete. Declared +row-local custom, AI, and custom/AI pipeline transforms now use `BatchEnvelope` units without +populating full-run `raw_data` or `transformed_data`; AI artifacts are generated and versioned once +per run. Every batch receives +schema-drift and validation policy, quarantine metadata, row/byte/checksum reconciliation, +cooperative cancellation, and a final checkpoint only after atomic CSV/JSON publication or a +PostgreSQL staging-table transaction. PostgreSQL replace/create-once/append/upsert modes now hide +all batches until final swap or merge; live-database tests cover atomic visibility, failure cleanup, +append, and idempotent keyed upsert. Target and transform failure tests prove that existing output +is preserved and temporary output is discarded. Native PDF fixtures cover text, ruled tables, +provenance, file/page limits, enforced page timeout, and fail/skip reporting; OCR remains explicitly +unimplemented. + +The production-image exit run processed 30,000,000 deterministic rows in 2,028.26 seconds with +118.23 MiB peak process-tree RSS under a 512 MiB cap, exact input/output row and SHA-256 +reconciliation, atomic publication, and no temporary output. Its +[`30m-row-local.json`](benchmarks/results/30m-row-local.json) report pins source revision +`b2d474b`, image ID `sha256:70d60d4c…`, Python 3.11.15, container limits, and disk-backed storage. +MongoDB remains intentionally rejected by the row-local path until it implements staging/merge +publication rather than partial direct batch effects. PostgreSQL append is at-least-once across an +ambiguous target-commit/checkpoint gap; deterministic keyed upsert is replay-safe. Local/global SQL +remains classified as global relational work and is not presented as bounded row-local execution. + ### Phase 3 — Add durable metadata and single-node recovery **Goal:** make runs observable and resumable before adding distributed transport. @@ -546,15 +585,16 @@ Exit gate: ## What to implement next -With Phase 0 and Phase 1 complete, start Phase 2: +Finish the bounded data-plane gate before starting durable metadata: -1. Add a `transform_batch` execution path for declared row-local transforms. -2. Keep bounded `BatchEnvelope` units flowing through CSV extract → validate → transform → - staged JSON publication without populating full-run `raw_data` or `transformed_data`. -3. Generate and version AI transform artifacts once per run, then execute the validated artifact - per batch. -4. Reconcile batch/input/output/rejected counts and checksums, and test cancellation or target - failure without false success or final partial output. +1. Extend the pinned production-image curve to 1M/10M and representative wide-row/custom-transform + workloads; the 30M narrow identity gate is complete. +2. Keep MongoDB blocked until an equivalent tested staging/merge protocol exists, and extend the + PostgreSQL live failure matrix to connection loss during final publication. +3. Add a spill-capable local global-relational plan with explicit disk/memory/temp limits, while + continuing to prefer ELT pushdown. +4. Expand schema evolution compatibility tests across supported targets and representative wide + or nested records. Do not add Better Auth, PostgreSQL run metadata, NATS, or distributed workers until the bounded single-node data-plane contract is real. diff --git a/README.md b/README.md index aee2556..bd84204 100644 --- a/README.md +++ b/README.md @@ -22,9 +22,13 @@ LLM-generated artifacts. - Local scheduling, daemon management, run summaries, and logs - Optional Gemini, OpenAI, Claude, and Qwen providers - Resource-limited Python transform subprocesses on Linux and macOS +- Declared row-local ETL with bounded batches, per-batch validation, schema policies, + reconciliation checksums, atomic CSV/JSON publication, and transactional PostgreSQL staging -Source and target connectors process chunks, but some ETL transform paths still materialize a full -run. Do not assume bounded memory for 30–100M-row jobs yet. See +Bounded execution is opt-in because applying a global transform independently to chunks changes its +meaning. Undeclared transforms and local SQL ETL still use the materialized compatibility path. Do +not assume bounded memory for 30–100M-row jobs until the workload has passed the reproducible +full-pipeline benchmark for its row width and transform class. See [Production readiness](PRODUCTION_READINESS.md) for the verified limits and release gates. The `v0.4.0` release baseline's four-column custom identity path completed 1M rows at roughly 1.28 @@ -34,6 +38,11 @@ other transforms, and concurrent runs require their own capped benchmark. The ve environment provenance are in [`benchmarks/results/`](benchmarks/results/README.md). +The declared row-local four-column identity workload has passed a clean production-image 30M-row +gate at 118.23 MiB peak process-tree RSS under a 512 MiB cap, with exact row-count/SHA-256 +reconciliation and no temporary output. See +[`30m-row-local.json`](benchmarks/results/30m-row-local.json). + ## Install Python 3.11 or newer is required. @@ -112,6 +121,62 @@ print(result.status, result.snapshot.rows_loaded) not contain source rows, credentials, connectors, iterators, or live LLM provider objects. The legacy `loafer.runner.run_pipeline()` API remains available as a compatibility facade. +## Bounded row-local execution + +Declare `row_local` only when every output row depends on rows in the current batch, such as maps, +filters, normalization, or independent enrichment: + +```yaml +mode: etl +chunk_size: 5000 + +source: + type: csv + path: ./input/orders.csv + +transform: + type: custom + path: ./transforms/normalize_order.py + +target: + type: json + path: ./output/orders.json + write_mode: overwrite + +execution: + transform_class: row_local + schema_drift: fail # fail | evolve | quarantine | coerce + # quarantine_path: ./output/rejected.json + +validation: + required_columns: [id, amount] + column_types: + id: string + max_null_rate: 0.1 + strict: true + on_failure: fail # fail | quarantine +``` + +This path never populates full-run `raw_data` or `transformed_data`. It emits a `BatchEnvelope` for +each batch, validates every row, keeps rolling row/byte/checksum totals, checks cancellation at safe +boundaries, and generates an AI transform artifact once per run before reusing it for every batch. +Rejected rows are written with batch and reason metadata when quarantine is configured. + +Current publication guarantees: + +| Target | Declared row-local behavior | +|---|---| +| JSON / CSV | Run-scoped temporary file; atomically renamed only after every batch succeeds | +| PostgreSQL `replace` | Hidden run-scoped table; final table replacement occurs in one transaction and deterministic replay replaces it again | +| PostgreSQL `error` | Hidden run-scoped table; create-once rename occurs in one transaction; retry after an ambiguous success requires reconciliation | +| PostgreSQL `append` | Hidden run-scoped table; all rows merge in one transaction; retry after a target-commit/checkpoint gap is at-least-once and can duplicate rows | +| PostgreSQL `upsert` | Hidden run-scoped table; keyed merge occurs in one transaction and deterministic replay is idempotent by the declared key | +| MongoDB | Rejected at config validation until a tested staging/merge publication protocol exists | + +SQL is classified as `global_relational`; joins, aggregates, sorts, windows, and large +deduplication must use ELT pushdown or a spill-capable engine rather than per-batch execution. +`loafer validate` exposes the selected delivery guarantee in the execution plan. + ## Transform options ### SQL @@ -159,6 +224,25 @@ transform: Each step receives the previous step's output. See [`examples/pipelines/multi_step_transform.yaml`](examples/pipelines/multi_step_transform.yaml). +## PDF extraction limits + +The native PDF source streams page records with file/page provenance and table provenance: + +```yaml +source: + type: pdf + path: ./documents/report.pdf + extract_tables: true + max_pages: 500 + max_file_size_mb: 100 + page_timeout_seconds: 30 + total_timeout_seconds: 300 + page_failure_policy: fail # fail | skip +``` + +`skip` records a redacted page diagnostic while continuing with later pages. OCR is not +implemented; `ocr_applied` remains `false` in provenance. + ## CLI ```text diff --git a/benchmarks/full_pipeline.py b/benchmarks/full_pipeline.py index 06747b1..4faf7fc 100644 --- a/benchmarks/full_pipeline.py +++ b/benchmarks/full_pipeline.py @@ -89,7 +89,13 @@ def _process_group_rss_bytes(process_group: int) -> int: # (field 5), and fields[21] is RSS pages (field 24). if int(fields[2]) == process_group: total_pages += int(fields[21]) - except (FileNotFoundError, IndexError, PermissionError, ValueError): + except ( + FileNotFoundError, + ProcessLookupError, + IndexError, + PermissionError, + ValueError, + ): continue return total_pages * page_size @@ -188,16 +194,33 @@ def _preflight(work_directory: Path, rows: int) -> None: def _git_revision(repository: Path) -> str | None: - result = subprocess.run( - ["git", "rev-parse", "HEAD"], - cwd=repository, - capture_output=True, - text=True, - check=False, - ) + try: + result = subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=repository, + capture_output=True, + text=True, + check=False, + ) + except OSError: + return None return result.stdout.strip() if result.returncode == 0 else None +def _git_worktree_dirty(repository: Path) -> bool | None: + try: + result = subprocess.run( + ["git", "status", "--porcelain"], + cwd=repository, + capture_output=True, + text=True, + check=False, + ) + except OSError: + return None + return bool(result.stdout) if result.returncode == 0 else None + + def _tail(path: Path, max_characters: int = 4000) -> str: try: text = path.read_text(encoding="utf-8", errors="replace") @@ -251,6 +274,9 @@ def run_benchmark( "transform:", " type: custom", f" path: {json.dumps(str(transform_path))}", + "execution:", + " transform_class: row_local", + " schema_drift: fail", f"chunk_size: {chunk_size}", "streaming_threshold: 1", "sandbox:", @@ -314,7 +340,7 @@ def run_benchmark( "rows_requested": rows, "rows_output": output_rows, "chunk_size": chunk_size, - "transform_class": "custom_identity_current_materializing_path", + "transform_class": "custom_identity_row_local", "generation_seconds": round(generation_seconds, 6), "wall_seconds": round(process.wall_seconds, 6), "throughput_rows_per_second": _verified_throughput( @@ -344,6 +370,7 @@ def run_benchmark( "machine": platform.machine(), "loafer_version": loafer_version, "git_revision": _git_revision(repository), + "git_worktree_dirty": _git_worktree_dirty(repository), }, "process": asdict(process), } diff --git a/benchmarks/results/30m-row-local.json b/benchmarks/results/30m-row-local.json new file mode 100644 index 0000000..394cef8 --- /dev/null +++ b/benchmarks/results/30m-row-local.json @@ -0,0 +1,51 @@ +{ + "chunk_size": 10000, + "correct": true, + "cpu_system_seconds": 86.19, + "cpu_user_seconds": 1921.59, + "environment": { + "container_cpus": 4, + "container_image": "loafer:verify-b2d474b", + "container_image_id": "sha256:70d60d4c7347569849f8b423f02a81353aba7305adbdbbc2afd8797a761c7275", + "container_memory_mb": 2048, + "container_memory_swap_mb": 2048, + "git_revision": null, + "git_worktree_dirty": null, + "loafer_version": "0.4.1.dev0+b2d474b", + "machine": "x86_64", + "platform": "Linux-6.12.96+deb13-amd64-x86_64-with-glibc2.41", + "python": "3.11.15", + "source_revision": "b2d474ba7b90926c19fd542041b0c6a950890472", + "work_directory_storage": "disk-backed bind mount" + }, + "generation_seconds": 61.391424, + "input_bytes": 1140588914, + "input_sha256": "717426ac9c3710761b50156ee8c32d66ec144c11ae4d8e2dd08c4de3289e8184", + "output_bytes": 1140588914, + "output_published": true, + "output_sha256": "717426ac9c3710761b50156ee8c32d66ec144c11ae4d8e2dd08c4de3289e8184", + "peak_process_tree_rss_bytes": 123969536, + "peak_process_tree_rss_mb": 118.227, + "process": { + "cpu_system_seconds": 86.19, + "cpu_user_seconds": 1921.59, + "peak_rss_bytes": 123969536, + "returncode": 0, + "termination_reason": null, + "wall_seconds": 2028.2585178489971 + }, + "returncode": 0, + "rows_output": 30000000, + "rows_requested": 30000000, + "rss_limit_mb": 512, + "sandbox_memory_mb": 256, + "schema_version": 1, + "status": "succeeded", + "stderr_tail": "", + "temporary_outputs": [], + "termination_reason": null, + "throughput_rows_per_second": 14791.014, + "timeout_seconds": 3600, + "transform_class": "custom_identity_row_local", + "wall_seconds": 2028.258518 +} diff --git a/benchmarks/results/README.md b/benchmarks/results/README.md index 6a15c07..8312dbc 100644 --- a/benchmarks/results/README.md +++ b/benchmarks/results/README.md @@ -1,6 +1,6 @@ -# Phase 0 release benchmark baseline +# Full-pipeline benchmark artifacts -These reports close the Phase 0 benchmark gate for Loafer `v0.4.0`. +The `1m.json` and `10m.json` reports record the Phase 0 release baseline for Loafer `v0.4.0`. ## Provenance @@ -34,3 +34,22 @@ Input generation is excluded from pipeline wall time. The 10M report has `correct: false` because the requested pipeline did not complete; for the Phase 0 workload-envelope gate, its safe termination and lack of published output are the expected evidence. + +## Bounded row-local production-image gate + +[`30m-row-local.json`](30m-row-local.json) records the Phase 2 implementation gate run on +2026-07-30. A deterministic CSV → custom identity → CSV pipeline processed 30,000,000 rows with +chunk size 10,000 under a 512 MiB process-tree RSS limit and a 256 MiB sandbox limit. + +The run completed in 2,028.26 seconds at 14,791.01 rows/second with 118.23 MiB peak process-tree +RSS. All 30,000,000 output rows were counted, the 1,140,588,914-byte input and output SHA-256 +digests matched, and no temporary output remained. + +The production image was built from clean source revision +`b2d474ba7b90926c19fd542041b0c6a950890472` and installed Loafer +`0.4.1.dev0+b2d474b` on Python 3.11.15. Its immutable local image ID was +`sha256:70d60d4c7347569849f8b423f02a81353aba7305adbdbbc2afd8797a761c7275`. +Docker provided 4 CPUs, a 2 GiB memory/memory-swap envelope, and disk-backed bind storage; the +harness independently enforced the 512 MiB process-tree RSS limit. Do not place large benchmark +files on `tmpfs` inside a memory-capped container because cgroup accounting includes those +unevictable file pages. diff --git a/loafer/adapters/sources/csv_source.py b/loafer/adapters/sources/csv_source.py index a0c71b6..d62140e 100644 --- a/loafer/adapters/sources/csv_source.py +++ b/loafer/adapters/sources/csv_source.py @@ -18,6 +18,7 @@ from loafer.ports.connector import SourceConnector logger = logging.getLogger(__name__) +_ENCODING_SCAN_CHARS = 1024 * 1024 class CsvSourceConnector(SourceConnector): @@ -51,10 +52,13 @@ def connect(self) -> None: try: self._file = open(self._path, encoding=self._encoding, newline="") - self._file.read() + while self._file.read(_ENCODING_SCAN_CHARS): + pass self._file.seek(0) except UnicodeDecodeError: logger.warning("UTF-8 decode failed for %s, falling back to latin-1", self._path) + if self._file is not None: + self._file.close() self._file = open(self._path, encoding="latin-1", newline="") self._actual_encoding = "latin-1" diff --git a/loafer/adapters/sources/pdf.py b/loafer/adapters/sources/pdf.py index b9f7867..7b8352f 100644 --- a/loafer/adapters/sources/pdf.py +++ b/loafer/adapters/sources/pdf.py @@ -2,8 +2,14 @@ from __future__ import annotations +import contextlib +import signal +import threading +import time +from pathlib import Path from typing import Any +from loafer.exceptions import ConnectorError from loafer.ports.connector import SourceConnector @@ -14,25 +20,51 @@ def __init__( self, path: str, extract_tables: bool = True, + max_pages: int | None = None, + max_file_size_mb: int = 100, + page_timeout_seconds: float = 30.0, + total_timeout_seconds: float = 300.0, + page_failure_policy: str = "fail", ) -> None: self._path = path self._extract_tables = extract_tables + self._max_pages = max_pages + self._max_file_size_mb = max_file_size_mb + self._page_timeout_seconds = page_timeout_seconds + self._total_timeout_seconds = total_timeout_seconds + self._page_failure_policy = page_failure_policy self._doc: Any = None + self._diagnostics: list[str] = [] def connect(self) -> None: + path = Path(self._path) + try: + size_bytes = path.stat().st_size + except OSError as exc: + raise ConnectorError(f"failed to open PDF: {exc}") from exc + limit_bytes = self._max_file_size_mb * 1024 * 1024 + if size_bytes > limit_bytes: + raise ConnectorError( + f"PDF file is {size_bytes} bytes, exceeding the configured " + f"{self._max_file_size_mb}MB limit" + ) + try: import pdfplumber except ImportError: - from loafer.exceptions import ConnectorError - raise ConnectorError("PDF connector requires 'pdfplumber'") try: self._doc = pdfplumber.open(self._path) except Exception as exc: - from loafer.exceptions import ConnectorError - raise ConnectorError(f"failed to open PDF: {exc}") from exc + self._diagnostics = [] + page_count = len(self._doc.pages) + if self._max_pages is not None and page_count > self._max_pages: + self.disconnect() + raise ConnectorError( + f"PDF has {page_count} pages, exceeding the configured {self._max_pages}-page limit" + ) def disconnect(self) -> None: if self._doc: @@ -41,23 +73,31 @@ def disconnect(self) -> None: def stream(self, chunk_size: int) -> Any: if self._doc is None: - from loafer.exceptions import ConnectorError - raise ConnectorError("not connected") chunk: list[dict[str, Any]] = [] + document_started = time.monotonic() for page_num, page in enumerate(self._doc.pages, start=1): - text = page.extract_text() or "" - row: dict[str, Any] = { - "page": page_num, - "text": text, - } - - if self._extract_tables: - tables = page.extract_tables() - row["tables"] = tables if tables else [] - row["table_count"] = len(tables) if tables else 0 - + elapsed = time.monotonic() - document_started + if elapsed >= self._total_timeout_seconds: + raise ConnectorError( + f"PDF extraction exceeded the configured " + f"{self._total_timeout_seconds:g}s document timeout before page {page_num}" + ) + + page_budget = min( + self._page_timeout_seconds, + self._total_timeout_seconds - elapsed, + ) + try: + with _time_limit(page_budget): + row = self._extract_page(page_num, page) + except Exception as exc: + message = f"PDF page {page_num} extraction failed: {exc}" + if self._page_failure_policy == "fail": + raise ConnectorError(message) from exc + self._diagnostics.append(message) + continue chunk.append(row) if len(chunk) >= chunk_size: @@ -71,3 +111,60 @@ def count(self) -> int | None: if self._doc is None: return None return len(self._doc.pages) + + def diagnostics(self) -> list[str]: + return list(self._diagnostics) + + def _extract_page(self, page_num: int, page: Any) -> dict[str, Any]: + text = page.extract_text() or "" + source_path = str(Path(self._path).resolve()) + row: dict[str, Any] = { + "page": page_num, + "page_number": page_num, + "text": text, + "provenance": { + "source_path": source_path, + "page_number": page_num, + "content_type": "native_pdf", + "ocr_applied": False, + }, + } + + if self._extract_tables: + tables = page.extract_tables() or [] + row["tables"] = tables + row["table_count"] = len(tables) + row["table_provenance"] = [ + { + "source_path": source_path, + "page_number": page_num, + "table_index": index, + "row_count": len(table), + "column_count": max((len(table_row) for table_row in table), default=0), + } + for index, table in enumerate(tables) + ] + return row + + +@contextlib.contextmanager +def _time_limit(seconds: float) -> Any: + """Enforce a wall-clock limit where SIGALRM is available.""" + if not hasattr(signal, "SIGALRM") or threading.current_thread() is not threading.main_thread(): + started = time.monotonic() + yield + if time.monotonic() - started > seconds: + raise TimeoutError(f"page exceeded {seconds:g}s timeout") + return + + def _raise_timeout(_signum: int, _frame: Any) -> None: + raise TimeoutError(f"page exceeded {seconds:g}s timeout") + + previous_handler = signal.getsignal(signal.SIGALRM) + signal.signal(signal.SIGALRM, _raise_timeout) + previous_timer = signal.setitimer(signal.ITIMER_REAL, seconds) + try: + yield + finally: + signal.setitimer(signal.ITIMER_REAL, *previous_timer) + signal.signal(signal.SIGALRM, previous_handler) diff --git a/loafer/adapters/targets/postgres_staging.py b/loafer/adapters/targets/postgres_staging.py new file mode 100644 index 0000000..21b5ce2 --- /dev/null +++ b/loafer/adapters/targets/postgres_staging.py @@ -0,0 +1,300 @@ +"""Run-scoped PostgreSQL staging target with atomic final publication.""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any + +from psycopg2 import sql + +from loafer.adapters.postgres_sql import column_list, qualified_identifier +from loafer.core.identifiers import split_qualified_name +from loafer.exceptions import LoadError +from loafer.ports.connector import TargetConnector + + +class PostgresStagingTargetConnector(TargetConnector): + """Write batches to a hidden table, then merge/swap in one transaction.""" + + def __init__( + self, + url: str, + table: str, + write_mode: str, + key: list[str] | None, + run_id: str, + ) -> None: + self._url = url + self._table = table + self._write_mode = write_mode + self._key = key or [] + self._run_id = run_id + schema, _table_name = split_qualified_name(table) + identity = f"{table}:{run_id}".encode() + stage_name = f"_loafer_stage_{hashlib.sha256(identity).hexdigest()[:24]}" + self._staging_table = f"{schema}.{stage_name}" + self._conn: Any = None + self._cursor: Any = None + self._columns: list[str] = [] + self._staging_created = False + self._published = False + self._rows_written = 0 + + def connect(self) -> None: + try: + import psycopg2 + + self._conn = psycopg2.connect(self._url) + self._conn.autocommit = False + self._cursor = self._conn.cursor() + self._drop_staging() + self._conn.commit() + except Exception as exc: + self.disconnect() + raise LoadError(f"failed to connect PostgreSQL staging target: {exc}") from exc + + def disconnect(self) -> None: + if self._conn is not None and not self._published: + try: + self._conn.rollback() + if self._cursor is not None: + self._drop_staging() + self._conn.commit() + except Exception: + try: + self._conn.rollback() + except Exception: + pass + if self._cursor is not None: + self._cursor.close() + self._cursor = None + if self._conn is not None: + self._conn.close() + self._conn = None + + def write_chunk(self, chunk: list[dict[str, Any]]) -> int: + if self._conn is None or self._cursor is None: + raise LoadError("connect() must be called before write_chunk()") + if not chunk: + return 0 + + try: + import psycopg2.extras + + if not self._staging_created: + self._columns = list(chunk[0]) + self._create_staging(chunk[0]) + self._staging_created = True + self._assert_columns(chunk) + + query = sql.SQL("INSERT INTO {} ({}) VALUES %s").format( + qualified_identifier(self._staging_table), + column_list(self._columns), + ) + values = [self._serialize(row) for row in chunk] + psycopg2.extras.execute_values( + self._cursor, + query, + values, + template=None, + page_size=min(len(chunk), 1000), + ) + self._conn.commit() + except Exception as exc: + self._conn.rollback() + raise LoadError(f"staging batch insert failed ({len(chunk)} rows): {exc}") from exc + + self._rows_written += len(chunk) + return len(chunk) + + def finalize(self) -> None: + if self._conn is None or self._cursor is None: + return + if not self._staging_created: + self._publish_empty() + self._published = True + return + + try: + if self._write_mode == "replace": + self._replace() + elif self._write_mode == "error": + self._publish_new(error_if_exists=True) + elif self._write_mode == "append": + self._append() + elif self._write_mode == "upsert": + self._upsert() + else: + raise LoadError(f"unsupported PostgreSQL write mode: {self._write_mode}") + self._conn.commit() + self._published = True + except Exception as exc: + self._conn.rollback() + if isinstance(exc, LoadError): + raise + raise LoadError(f"failed to publish PostgreSQL staging table: {exc}") from exc + + def _create_staging(self, sample: dict[str, Any]) -> None: + definitions = [ + sql.SQL("{} {}").format( + sql.Identifier(column), + sql.SQL(_infer_pg_type(value)), + ) + for column, value in sample.items() + ] + query = sql.SQL("CREATE TABLE {} ({})").format( + qualified_identifier(self._staging_table), + sql.SQL(", ").join(definitions), + ) + self._cursor.execute(query) + + def _replace(self) -> None: + self._cursor.execute( + sql.SQL("DROP TABLE IF EXISTS {}").format(qualified_identifier(self._table)) + ) + self._rename_staging() + + def _publish_new(self, *, error_if_exists: bool) -> None: + if self._table_exists(self._table): + if error_if_exists: + raise LoadError(f"table '{self._table}' already exists and write_mode is 'error'") + return + self._rename_staging() + + def _append(self) -> None: + if not self._table_exists(self._table): + self._rename_staging() + return + self._cursor.execute( + sql.SQL("INSERT INTO {} ({}) SELECT {} FROM {}").format( + qualified_identifier(self._table), + column_list(self._columns), + column_list(self._columns), + qualified_identifier(self._staging_table), + ) + ) + self._drop_staging() + + def _upsert(self) -> None: + if not self._key: + raise LoadError("PostgreSQL staged upsert requires key columns") + if not self._table_exists(self._table): + self._rename_staging() + self._ensure_unique_index() + return + + self._ensure_unique_index() + updates = [column for column in self._columns if column not in self._key] + if updates: + action = sql.SQL("DO UPDATE SET {}").format( + sql.SQL(", ").join( + sql.SQL("{} = EXCLUDED.{}").format( + sql.Identifier(column), + sql.Identifier(column), + ) + for column in updates + ) + ) + else: + action = sql.SQL("DO NOTHING") + query = sql.SQL("INSERT INTO {} ({}) SELECT {} FROM {} ON CONFLICT ({}) {}").format( + qualified_identifier(self._table), + column_list(self._columns), + column_list(self._columns), + qualified_identifier(self._staging_table), + column_list(self._key), + action, + ) + self._cursor.execute(query) + self._drop_staging() + + def _publish_empty(self) -> None: + target_exists = self._table_exists(self._table) + if self._write_mode == "replace": + if target_exists: + self._cursor.execute( + sql.SQL("TRUNCATE TABLE {}").format(qualified_identifier(self._table)) + ) + self._conn.commit() + return + raise LoadError( + "cannot publish an empty PostgreSQL replacement because no output schema " + "or existing target table is available" + ) + if self._write_mode == "error" and target_exists: + raise LoadError(f"table '{self._table}' already exists and write_mode is 'error'") + if not target_exists: + raise LoadError( + "cannot publish an empty PostgreSQL run because no output schema " + "or existing target table is available" + ) + self._conn.commit() + + def _rename_staging(self) -> None: + _schema, final_name = split_qualified_name(self._table) + self._cursor.execute( + sql.SQL("ALTER TABLE {} RENAME TO {}").format( + qualified_identifier(self._staging_table), + sql.Identifier(final_name), + ) + ) + self._staging_created = False + + def _drop_staging(self) -> None: + self._cursor.execute( + sql.SQL("DROP TABLE IF EXISTS {}").format(qualified_identifier(self._staging_table)) + ) + self._staging_created = False + + def _table_exists(self, table: str) -> bool: + schema, table_name = split_qualified_name(table) + self._cursor.execute( + """ + SELECT EXISTS ( + SELECT FROM information_schema.tables + WHERE table_schema = %s AND table_name = %s + ) + """, + (schema, table_name), + ) + return bool(self._cursor.fetchone()[0]) + + def _ensure_unique_index(self) -> None: + identity = f"{self._table}:{','.join(self._key)}".encode() + index_name = f"loafer_uq_{hashlib.sha256(identity).hexdigest()[:16]}" + self._cursor.execute( + sql.SQL("CREATE UNIQUE INDEX IF NOT EXISTS {} ON {} ({})").format( + sql.Identifier(index_name), + qualified_identifier(self._table), + column_list(self._key), + ) + ) + + def _assert_columns(self, chunk: list[dict[str, Any]]) -> None: + expected = set(self._columns) + for row in chunk: + if set(row) != expected: + raise LoadError( + "PostgreSQL staging batch schema changed after the first output batch" + ) + + def _serialize(self, row: dict[str, Any]) -> tuple[Any, ...]: + return tuple( + json.dumps(row[column]) if isinstance(row[column], (dict, list)) else row[column] + for column in self._columns + ) + + +def _infer_pg_type(value: Any) -> str: + if value is None: + return "TEXT" + if isinstance(value, bool): + return "BOOLEAN" + if isinstance(value, int): + return "BIGINT" + if isinstance(value, float): + return "DOUBLE PRECISION" + if isinstance(value, (dict, list)): + return "JSONB" + return "TEXT" diff --git a/loafer/agents/extract.py b/loafer/agents/extract.py index fde2dbd..7a98796 100644 --- a/loafer/agents/extract.py +++ b/loafer/agents/extract.py @@ -116,7 +116,12 @@ def extract_agent(state: PipelineState) -> PipelineState: state.get("cursor_value"), ) peekable = _PeekableStream(raw_iter) - peekable_stream = _counting_stream(peekable, state, cursor_column) + peekable_stream = _counting_stream( + peekable, + state, + cursor_column, + connector, + ) state["stream_iterator"] = peekable_stream state["rows_extracted"] = ( count if count is not None and not client_side_incremental else 0 @@ -176,6 +181,7 @@ def _counting_stream( stream_iter: Iterator[list[dict[str, Any]]], state: PipelineState, cursor_column: str | None = None, + connector: SourceConnector | None = None, ) -> Iterator[list[dict[str, Any]]]: """Wrap a stream iterator to count rows and track the max cursor as consumed.""" total = 0 @@ -192,6 +198,8 @@ def _counting_stream( # placeholder count set at extract time (BUG-5). if total == 0: state.setdefault("warnings", []).append("Source returned 0 rows") + if connector is not None: + state.setdefault("warnings", []).extend(connector.diagnostics()) def _filter_incremental_stream( diff --git a/loafer/application/service.py b/loafer/application/service.py index 5e5efb4..8a22acf 100644 --- a/loafer/application/service.py +++ b/loafer/application/service.py @@ -62,6 +62,22 @@ def _config_digest(config: PipelineConfig) -> str: return hashlib.sha256(rendered.encode("utf-8")).hexdigest() +def _delivery_guarantee(config: PipelineConfig) -> str: + if config.execution.transform_class != "row_local": + return "target_defined" + if config.target.type in {"csv", "json"}: + return "atomic_run_publication" + if config.target.type == "postgres": + if config.target.write_mode == "replace": + return "atomic_transactional_replace" + if config.target.write_mode == "error": + return "atomic_transactional_create_once" + if config.target.write_mode == "upsert": + return "idempotent_keyed_atomic_merge" + return "at_least_once_atomic_merge" + return "unsupported" + + def _build_plan( config: PipelineConfig, request: RunRequest, @@ -91,9 +107,12 @@ def _build_plan( source_type=config.source.type, target_type=config.target.type, transform_type=config.transform.type, + transform_class=config.execution.transform_class, chunk_size=config.chunk_size, streaming_threshold=config.streaming_threshold, validation_strict=config.validation.strict, + schema_drift_policy=config.execution.schema_drift, + delivery_guarantee=_delivery_guarantee(config), llm_provider=config.llm.provider, llm_model=config.llm.model, incremental_column=config.incremental.column if config.incremental else None, @@ -150,8 +169,24 @@ def _snapshot( target_type=plan.target_type, transform_type=plan.transform_type, rows_extracted=max(0, int(state.get("rows_extracted", 0))), - rows_transformed=max(0, len(state.get("transformed_data", []))), + rows_transformed=max( + 0, + int( + state.get("rows_transformed", 0) + if state.get("batches_completed", 0) + else len(state.get("transformed_data", [])) + ), + ), rows_loaded=max(0, int(state.get("rows_loaded", 0))), + rows_rejected=max(0, int(state.get("rows_rejected", 0))), + rows_filtered=max(0, int(state.get("rows_filtered", 0))), + batches_completed=max(0, int(state.get("batches_completed", 0))), + bytes_in=max(0, int(state.get("bytes_in", 0))), + bytes_out=max(0, int(state.get("bytes_out", 0))), + input_checksum=state.get("input_checksum"), + output_checksum=state.get("output_checksum"), + schema_version=state.get("schema_version"), + transform_artifact_version=state.get("transform_artifact_version"), validation_passed=bool(state.get("validation_passed", False)), duration_ms=durations, warnings=warnings, @@ -259,11 +294,15 @@ def _stream_prepared( reviewer=self._reviewer, secret_resolver=self._secrets, provider_factory=self._provider_factory, + cancellation=self._cancellation, + checkpoints=self._checkpoints, ) sequence = 0 while True: - if self._cancellation.is_cancelled(request.run_id): + if plan.transform_class != "row_local" and self._cancellation.is_cancelled( + request.run_id + ): raise PipelineError(f"Pipeline cancelled (run_id={request.run_id})") try: stage, status, state = next(updates) @@ -278,6 +317,7 @@ def _stream_prepared( stage=stage, status=StageStatus(status), snapshot=_snapshot(state, plan, config), + batch=state.get("last_batch_envelope") if stage == "batch" else None, ) self._events.publish(event) yield event, state diff --git a/loafer/cli.py b/loafer/cli.py index 952ef08..308c1c7 100644 --- a/loafer/cli.py +++ b/loafer/cli.py @@ -529,6 +529,10 @@ def _print_summary_table(state: dict[str, Any], mode: str, failed_stage: str | N if token_usage: total_tokens = token_usage.get("total_tokens", 0) extras.append(f"Tokens: {total_tokens:,}") + if state.get("batches_completed"): + extras.append(f"Batches: {state['batches_completed']:,}") + if state.get("rows_rejected"): + extras.append(f"Rejected: {state['rows_rejected']:,}") console.print(f"\n[dim]{' | '.join(extras)}[/dim]") warnings = state.get("warnings", []) @@ -629,6 +633,10 @@ def run( status = event.status.value state = event.snapshot.model_dump(mode="python") final_state = state + if node_name == "batch": + if active_animator is not None: + active_animator.pulse() + continue label = _get_stage_label(node_name, state) row_info = _get_row_info(node_name, state) @@ -720,9 +728,12 @@ def validate( table.add_row("Source", plan.source_type) table.add_row("Target", plan.target_type) table.add_row("Transform", plan.transform_type) + table.add_row("Transform class", plan.transform_class) table.add_row("Chunk size", str(plan.chunk_size)) table.add_row("Streaming threshold", str(plan.streaming_threshold)) table.add_row("Validation strict", str(plan.validation_strict)) + table.add_row("Schema drift", plan.schema_drift_policy) + table.add_row("Delivery guarantee", plan.delivery_guarantee) table.add_row("LLM provider", plan.llm_provider) table.add_row("LLM model", plan.llm_model) diff --git a/loafer/config.py b/loafer/config.py index 9248c73..3af444e 100644 --- a/loafer/config.py +++ b/loafer/config.py @@ -142,6 +142,11 @@ class PdfSourceConfig(BaseModel): type: Literal["pdf"] path: str extract_tables: bool = True + max_pages: int | None = None + max_file_size_mb: int = 100 + page_timeout_seconds: float = 30.0 + total_timeout_seconds: float = 300.0 + page_failure_policy: Literal["fail", "skip"] = "fail" @field_validator("path") @classmethod @@ -150,6 +155,20 @@ def path_must_exist(cls, v: str) -> str: raise ValueError(f"PDF file not found: {v}") return v + @field_validator("max_pages", "max_file_size_mb") + @classmethod + def positive_limits(cls, value: int | None) -> int | None: + if value is not None and value <= 0: + raise ValueError("PDF page and file-size limits must be positive") + return value + + @field_validator("page_timeout_seconds", "total_timeout_seconds") + @classmethod + def positive_time_limits(cls, value: float) -> float: + if value <= 0: + raise ValueError("PDF timeout limits must be positive") + return value + SourceConfig = Annotated[ PostgresSourceConfig @@ -300,9 +319,49 @@ def steps_must_be_non_empty(cls, v: list[Any]) -> list[Any]: # Validation config +SchemaType = Literal[ + "null", + "boolean", + "integer", + "float", + "string", + "datetime", + "object", + "array", +] + + class ValidationConfig(BaseModel): max_null_rate: float = 0.5 strict: bool = False + required_columns: list[str] = Field(default_factory=list) + column_types: dict[str, SchemaType] = Field(default_factory=dict) + on_failure: Literal["fail", "quarantine"] = "fail" + + @field_validator("max_null_rate") + @classmethod + def null_rate_is_fraction(cls, value: float) -> float: + if not 0 <= value <= 1: + raise ValueError("max_null_rate must be between 0 and 1") + return value + + @field_validator("required_columns") + @classmethod + def required_columns_are_unique(cls, value: list[str]) -> list[str]: + normalized = [column.strip() for column in value] + if any(not column for column in normalized): + raise ValueError("required_columns cannot contain empty names") + if len(set(normalized)) != len(normalized): + raise ValueError("required_columns cannot contain duplicates") + return normalized + + +class ExecutionConfig(BaseModel): + """Data-plane semantics that must be declared independently of chunk size.""" + + transform_class: Literal["materialized", "row_local", "global_relational"] = "materialized" + schema_drift: Literal["fail", "evolve", "quarantine", "coerce"] = "fail" + quarantine_path: str | None = None # Sandbox config @@ -502,6 +561,7 @@ class PipelineConfig(BaseModel): streaming_threshold: int = 10_000 destructive_filter_threshold: float = 0.3 validation: ValidationConfig = Field(default_factory=ValidationConfig) + execution: ExecutionConfig = Field(default_factory=ExecutionConfig) llm: LLMConfig = Field(default_factory=LLMConfig) incremental: IncrementalConfig | None = None sandbox: SandboxConfig = Field(default_factory=SandboxConfig) @@ -513,6 +573,55 @@ def chunk_size_must_be_positive(cls, v: int) -> int: raise ValueError("chunk_size must be a positive integer") return v + @model_validator(mode="after") + def validate_execution_contract(self) -> PipelineConfig: + execution = self.execution + transform_types = ( + [step.type for step in self.transform.steps] + if isinstance(self.transform, PipelineTransformConfig) + else [self.transform.type] + ) + if "sql" in transform_types and execution.transform_class == "materialized": + execution.transform_class = "global_relational" + + if execution.transform_class == "row_local": + if self.mode != "etl": + raise ValueError( + "execution.transform_class 'row_local' is only supported in ETL mode; " + "use ELT pushdown for global relational work" + ) + if "sql" in transform_types: + raise ValueError( + "SQL transforms cannot be declared row_local because joins, sorts, windows, " + "aggregates, and deduplication require global semantics; use ELT pushdown or " + "execution.transform_class 'global_relational'" + ) + if self.target.type not in {"csv", "json", "postgres"}: + raise ValueError( + "bounded row_local execution currently requires an atomically published " + "CSV/JSON target or the PostgreSQL staging target" + ) + + needs_quarantine = ( + execution.schema_drift == "quarantine" or self.validation.on_failure == "quarantine" + ) + if needs_quarantine and not execution.quarantine_path: + raise ValueError( + "execution.quarantine_path is required when schema drift or validation failures " + "use the quarantine policy" + ) + if execution.schema_drift == "evolve" and self.target.type in {"csv", "postgres"}: + target_constraint = ( + "the CSV header is fixed after the first batch" + if self.target.type == "csv" + else "the PostgreSQL staging table schema is fixed by the first batch" + ) + raise ValueError( + f"schema_drift 'evolve' is not supported for {self.target.type} targets because " + f"{target_constraint}; use JSON or choose fail/quarantine/coerce" + ) + return self + @model_validator(mode="before") @classmethod def normalise_transform(cls, data: Any) -> Any: diff --git a/loafer/connectors/registry.py b/loafer/connectors/registry.py index 442dc6b..286fd24 100644 --- a/loafer/connectors/registry.py +++ b/loafer/connectors/registry.py @@ -20,6 +20,9 @@ from loafer.adapters.targets.json_target import JsonTargetConnector as _JsonTarget from loafer.adapters.targets.mongo import MongoTargetConnector as _MongoTarget from loafer.adapters.targets.postgres import PostgresTargetConnector as _PgTarget +from loafer.adapters.targets.postgres_staging import ( + PostgresStagingTargetConnector as _PgStagingTarget, +) from loafer.exceptions import ConnectorError from loafer.ports.connector import SourceConnector, TargetConnector @@ -106,6 +109,23 @@ def get_target_connector(config: TargetConfig) -> TargetConnector: return _build_target(connector_cls, config) +def get_staged_target_connector( + config: TargetConfig, + *, + run_id: str, +) -> TargetConnector: + """Instantiate a target that hides every batch until final publication.""" + if config.type == "postgres": + return _PgStagingTarget( + config.url, + config.table, + config.write_mode, + config.key, + run_id, + ) + return get_target_connector(config) + + def _build_source( cls: type[SourceConnector], config: SourceConfig, @@ -152,7 +172,15 @@ def _build_source( incremental_value=cursor_value, ) case "pdf": - return cls(config.path, config.extract_tables) # type: ignore[call-arg] + return cls( # type: ignore[call-arg] + config.path, + config.extract_tables, + config.max_pages, + config.max_file_size_mb, + config.page_timeout_seconds, + config.total_timeout_seconds, + config.page_failure_policy, + ) msg = f"source connector '{config.type}' not implemented" raise RegistryError(msg) @@ -213,4 +241,5 @@ def _resolve_url(url: str) -> dict[str, Any]: CsvTargetConnector = _CsvTarget JsonTargetConnector = _JsonTarget PostgresTargetConnector = _PgTarget +PostgresStagingTargetConnector = _PgStagingTarget MongoTargetConnector = _MongoTarget diff --git a/loafer/contracts.py b/loafer/contracts.py index f71ddb8..ef12a83 100644 --- a/loafer/contracts.py +++ b/loafer/contracts.py @@ -56,9 +56,12 @@ class ExecutionPlan(ContractModel): source_type: str target_type: str transform_type: str + transform_class: str = "materialized" chunk_size: int = Field(gt=0) streaming_threshold: int = Field(gt=0) validation_strict: bool + schema_drift_policy: str = "fail" + delivery_guarantee: str = "target_defined" llm_provider: str llm_model: str incremental_column: str | None = None @@ -91,9 +94,13 @@ class BatchEnvelope(ContractModel): rows_in: int = Field(ge=0) rows_out: int = Field(ge=0) rows_rejected: int = Field(ge=0) + rows_filtered: int = Field(default=0, ge=0) bytes_in: int = Field(ge=0) bytes_out: int = Field(ge=0) checksum: str | None = None + input_checksum: str | None = None + output_checksum: str | None = None + duration_ms: float = Field(default=0.0, ge=0) class Checkpoint(ContractModel): @@ -135,6 +142,15 @@ class RunSnapshot(ContractModel): rows_extracted: int = Field(ge=0) rows_transformed: int = Field(ge=0) rows_loaded: int = Field(ge=0) + rows_rejected: int = Field(default=0, ge=0) + rows_filtered: int = Field(default=0, ge=0) + batches_completed: int = Field(default=0, ge=0) + bytes_in: int = Field(default=0, ge=0) + bytes_out: int = Field(default=0, ge=0) + input_checksum: str | None = None + output_checksum: str | None = None + schema_version: str | None = None + transform_artifact_version: str | None = None validation_passed: bool duration_ms: dict[str, float] = Field(default_factory=dict) warnings: tuple[str, ...] = () @@ -154,6 +170,7 @@ class RunEvent(ContractModel): status: StageStatus occurred_at: datetime = Field(default_factory=_utc_now) snapshot: RunSnapshot + batch: BatchEnvelope | None = None class RunResult(ContractModel): diff --git a/loafer/core/batches.py b/loafer/core/batches.py new file mode 100644 index 0000000..a2be888 --- /dev/null +++ b/loafer/core/batches.py @@ -0,0 +1,331 @@ +"""Pure bounded-batch schema, validation, and reconciliation policies.""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass, field +from typing import Any + +from loafer.config import ValidationConfig +from loafer.exceptions import ValidationError + +_DATETIME_RE = re.compile( + r"^\d{4}-\d{2}-\d{2}(?:[T ]\d{2}:\d{2}(?::\d{2})?(?:\.\d+)?" + r"(?:Z|[+-]\d{2}:?\d{2})?)?$" +) + + +def value_type(value: Any) -> str: + """Return the stable schema type used by batch contracts.""" + if value is None: + return "null" + if isinstance(value, bool): + return "boolean" + if isinstance(value, int): + return "integer" + if isinstance(value, float): + return "float" + if isinstance(value, dict): + return "object" + if isinstance(value, list): + return "array" + if isinstance(value, str) and _DATETIME_RE.match(value): + return "datetime" + return "string" + + +def infer_schema(rows: list[dict[str, Any]]) -> dict[str, str]: + """Infer a deterministic column/type mapping for one bounded batch.""" + observed: dict[str, set[str]] = {} + for row in rows: + for column, value in row.items(): + observed.setdefault(column, set()).add(value_type(value)) + + schema: dict[str, str] = {} + for column in sorted(observed): + types = observed[column] - {"null"} + if not types: + schema[column] = "null" + elif len(types) == 1: + schema[column] = next(iter(types)) + else: + schema[column] = "mixed" + return schema + + +def schema_version(schema: dict[str, str]) -> str: + """Return a content-addressed version for a schema mapping.""" + payload = json.dumps(schema, sort_keys=True, separators=(",", ":")).encode("utf-8") + return f"sha256:{hashlib.sha256(payload).hexdigest()}" + + +def canonical_row_bytes(row: dict[str, Any]) -> bytes: + """Serialize a row deterministically for counts, bytes, and checksums.""" + return ( + json.dumps( + row, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + default=str, + ).encode("utf-8") + + b"\n" + ) + + +@dataclass +class RollingRowsDigest: + """Chunk-boundary-independent checksum and byte accumulator.""" + + _digest: Any = field(default_factory=hashlib.sha256) + rows: int = 0 + bytes: int = 0 + + def update(self, rows: list[dict[str, Any]]) -> None: + for row in rows: + payload = canonical_row_bytes(row) + self._digest.update(payload) + self.rows += 1 + self.bytes += len(payload) + + @property + def checksum(self) -> str: + return f"sha256:{self._digest.hexdigest()}" + + +@dataclass(frozen=True) +class RejectedRow: + """One rejected source row and its machine-readable reason.""" + + row: dict[str, Any] + stage: str + reason: str + + +@dataclass(frozen=True) +class SchemaBatchResult: + rows: list[dict[str, Any]] + rejected: list[RejectedRow] + schema: dict[str, str] + version: str + evolved: bool = False + + +class SchemaTracker: + """Apply one explicit schema-drift policy without retaining prior rows.""" + + def __init__(self) -> None: + self._schema: dict[str, str] | None = None + + @property + def schema(self) -> dict[str, str]: + return dict(self._schema or {}) + + @property + def version(self) -> str | None: + return schema_version(self._schema) if self._schema is not None else None + + def apply( + self, + rows: list[dict[str, Any]], + policy: str, + ) -> SchemaBatchResult: + if self._schema is None: + self._schema = infer_schema(rows) + return SchemaBatchResult( + rows=list(rows), + rejected=[], + schema=self.schema, + version=schema_version(self._schema), + ) + + accepted: list[dict[str, Any]] = [] + rejected: list[RejectedRow] = [] + evolved = False + + if policy == "evolve": + incoming = infer_schema(rows) + merged = _merge_schemas(self._schema, incoming) + evolved = merged != self._schema + self._schema = merged + accepted = list(rows) + else: + for row in rows: + differences = _schema_differences(row, self._schema) + if not differences: + accepted.append(row) + continue + + reason = "; ".join(differences) + if policy == "fail": + raise ValidationError(f"schema drift detected: {reason}") + if policy == "quarantine": + rejected.append(RejectedRow(row=row, stage="schema", reason=reason)) + continue + if policy == "coerce": + try: + accepted.append(_coerce_row(row, self._schema)) + except (TypeError, ValueError) as exc: + rejected.append( + RejectedRow( + row=row, + stage="schema", + reason=f"schema coercion failed: {exc}", + ) + ) + continue + raise ValidationError(f"unknown schema drift policy: {policy}") + + return SchemaBatchResult( + rows=accepted, + rejected=rejected, + schema=self.schema, + version=schema_version(self._schema), + evolved=evolved, + ) + + +@dataclass(frozen=True) +class ValidationBatchResult: + rows: list[dict[str, Any]] + rejected: list[RejectedRow] + warnings: list[str] + column_counts: dict[str, dict[str, int]] + + +def validate_batch( + rows: list[dict[str, Any]], + config: ValidationConfig, +) -> ValidationBatchResult: + """Validate every row in a batch and apply fail/quarantine semantics.""" + rejected_by_index: dict[int, list[str]] = {} + columns = sorted({column for row in rows for column in row}) + column_counts: dict[str, dict[str, int]] = { + column: {"total_count": len(rows), "null_count": 0} for column in columns + } + + for index, row in enumerate(rows): + reasons: list[str] = [] + for column in config.required_columns: + if column not in row or row[column] is None: + reasons.append(f"required column '{column}' is missing or null") + for column, expected in config.column_types.items(): + if column not in row or row[column] is None: + continue + actual = value_type(row[column]) + if not _type_compatible(actual, expected): + reasons.append(f"column '{column}' expected {expected}, got {actual}") + if reasons: + rejected_by_index[index] = reasons + + for column in columns: + if column not in row or row[column] is None: + column_counts[column]["null_count"] += 1 + + warnings: list[str] = [] + if rows: + for column, counts in column_counts.items(): + null_rate = counts["null_count"] / counts["total_count"] + if null_rate <= config.max_null_rate: + continue + message = ( + f"Column '{column}' null rate {null_rate:.2%} exceeds " + f"max_null_rate {config.max_null_rate:.2%}" + ) + if config.strict: + for index, row in enumerate(rows): + if column not in row or row[column] is None: + rejected_by_index.setdefault(index, []).append(message) + else: + warnings.append(message) + + if rejected_by_index and config.on_failure == "fail": + first_index = min(rejected_by_index) + raise ValidationError( + f"batch validation failed at row {first_index}: " + f"{'; '.join(rejected_by_index[first_index])}" + ) + + accepted = [row for index, row in enumerate(rows) if index not in rejected_by_index] + rejected = [ + RejectedRow( + row=rows[index], + stage="validation", + reason="; ".join(reasons), + ) + for index, reasons in sorted(rejected_by_index.items()) + ] + return ValidationBatchResult( + rows=accepted, + rejected=rejected, + warnings=warnings, + column_counts=column_counts, + ) + + +def _type_compatible(actual: str, expected: str) -> bool: + if actual == expected: + return True + return expected == "float" and actual == "integer" + + +def _schema_differences(row: dict[str, Any], schema: dict[str, str]) -> list[str]: + differences: list[str] = [] + missing = sorted(set(schema) - set(row)) + extra = sorted(set(row) - set(schema)) + if missing: + differences.append(f"missing columns: {', '.join(missing)}") + if extra: + differences.append(f"new columns: {', '.join(extra)}") + + for column in sorted(set(row) & set(schema)): + value = row[column] + if value is None or schema[column] in {"mixed", "null"}: + continue + actual = value_type(value) + if not _type_compatible(actual, schema[column]): + differences.append(f"column '{column}' changed from {schema[column]} to {actual}") + return differences + + +def _merge_schemas( + current: dict[str, str], + incoming: dict[str, str], +) -> dict[str, str]: + merged = dict(current) + for column, incoming_type in incoming.items(): + current_type = merged.get(column) + if current_type is None or current_type == "null": + merged[column] = incoming_type + elif incoming_type not in {current_type, "null"}: + merged[column] = "mixed" + return dict(sorted(merged.items())) + + +def _coerce_row(row: dict[str, Any], schema: dict[str, str]) -> dict[str, Any]: + return {column: _coerce_value(row.get(column), expected) for column, expected in schema.items()} + + +def _coerce_value(value: Any, expected: str) -> Any: + if value is None or expected in {"mixed", "null"}: + return value + if _type_compatible(value_type(value), expected): + return float(value) if expected == "float" else value + if expected == "string": + return str(value) + if expected == "integer": + return int(value) + if expected == "float": + return float(value) + if expected == "boolean": + if isinstance(value, str): + normalized = value.strip().lower() + if normalized in {"true", "1", "yes"}: + return True + if normalized in {"false", "0", "no"}: + return False + if isinstance(value, (int, float)): + return bool(value) + raise ValueError(f"cannot coerce {value!r} to {expected}") diff --git a/loafer/data_plane.py b/loafer/data_plane.py new file mode 100644 index 0000000..4385885 --- /dev/null +++ b/loafer/data_plane.py @@ -0,0 +1,427 @@ +"""Bounded single-node ETL data plane for declared row-local transforms.""" + +from __future__ import annotations + +import time +import uuid +from collections.abc import Iterator +from typing import TYPE_CHECKING, Any + +from loafer.config import JsonTargetConfig, PipelineConfig +from loafer.connectors.registry import ( + get_source_connector, + get_staged_target_connector, + get_target_connector, +) +from loafer.contracts import BatchEnvelope, Checkpoint +from loafer.core.batches import ( + RejectedRow, + RollingRowsDigest, + SchemaTracker, + canonical_row_bytes, + validate_batch, +) +from loafer.core.destructive import detect_destructive_operations, raise_if_destructive +from loafer.exceptions import PipelineError, ValidationError +from loafer.llm.schema import build_schema_sample +from loafer.transform.batch_runner import ( + PreparedBatchTransform, + prepare_transform_artifact, + transform_batch, +) + +if TYPE_CHECKING: + from loafer.graph.state import PipelineState + from loafer.ports.connector import SourceConnector, TargetConnector + from loafer.ports.runtime import CancellationPort, CheckpointPort + +_INCREMENTAL_PUSHDOWN_SOURCES = {"postgres", "mysql", "sqlite", "rest_api"} +_PARTITION_ID = "default" + + +def uses_bounded_data_plane(config: PipelineConfig) -> bool: + """Return whether the config explicitly selected bounded row-local execution.""" + return config.mode == "etl" and config.execution.transform_class == "row_local" + + +def stream_bounded_pipeline( + config: PipelineConfig, + state: PipelineState, + *, + dry_run: bool, + cancellation: CancellationPort | None, + checkpoints: CheckpointPort | None, +) -> Iterator[tuple[str, str, PipelineState]]: + """Flow source batches through validation, transform, and staged publication.""" + source: SourceConnector | None = None + target: TargetConnector | None = None + quarantine: TargetConnector | None = None + active_stage = "extract" + started = time.monotonic() + stage_ms = {"extract": 0.0, "validate": 0.0, "transform": 0.0, "load": 0.0} + + input_digest = RollingRowsDigest() + output_digest = RollingRowsDigest() + input_schema = SchemaTracker() + output_schema = SchemaTracker() + artifact: PreparedBatchTransform | None = None + last_envelope: BatchEnvelope | None = None + quality_columns: dict[str, dict[str, int]] = {} + quality_warnings: set[str] = set() + destructive_seen: set[tuple[str, str]] = set() + + state["is_streaming"] = True + state["raw_data"] = [] + state["transformed_data"] = [] + yield ("extract", "running", state) + + try: + source = _source_connector(config, state) + source.connect() + if not dry_run: + target = get_staged_target_connector( + config.target, + run_id=state["run_id"], + ) + target.connect() + if config.execution.quarantine_path: + quarantine = get_target_connector( + JsonTargetConfig( + type="json", + path=config.execution.quarantine_path, + write_mode="overwrite", + ) + ) + quarantine.connect() + + source_iterator = iter(source.stream(config.chunk_size)) + batch_index = 0 + source_offset = 0 + + while True: + _raise_if_cancelled(cancellation, state["run_id"]) + read_started = time.monotonic() + try: + raw_rows = next(source_iterator) + except StopIteration: + stage_ms["extract"] += (time.monotonic() - read_started) * 1000 + break + stage_ms["extract"] += (time.monotonic() - read_started) * 1000 + + raw_rows = _filter_incremental_rows(config, state, raw_rows) + if not raw_rows: + continue + + batch_index += 1 + batch_started = time.monotonic() + batch_input_digest = RollingRowsDigest() + batch_input_digest.update(raw_rows) + input_digest.update(raw_rows) + state["rows_extracted"] = input_digest.rows + _advance_cursor(config, state, raw_rows) + + active_stage = "validate" + validate_started = time.monotonic() + schema_result = input_schema.apply(raw_rows, config.execution.schema_drift) + validation_result = validate_batch(schema_result.rows, config.validation) + rejected = [*schema_result.rejected, *validation_result.rejected] + _merge_quality_counts(quality_columns, validation_result.column_counts) + quality_warnings.update(validation_result.warnings) + stage_ms["validate"] += (time.monotonic() - validate_started) * 1000 + + accepted_rows = validation_result.rows + if state.get("schema_sample") == {} and accepted_rows: + state["schema_sample"] = build_schema_sample( + accepted_rows, + max_sample_rows=5, + ) + state["schema_version"] = schema_result.version + + if rejected: + _write_rejections( + quarantine, + rejected, + run_id=state["run_id"], + batch_id=f"batch-{batch_index:08d}", + ) + state["rows_rejected"] = state.get("rows_rejected", 0) + len(rejected) + + active_stage = "transform" + transform_started = time.monotonic() + if artifact is None and accepted_rows: + artifact = prepare_transform_artifact(state, accepted_rows[:5]) + state["transform_artifact_version"] = artifact.version + state["last_error"] = None + transformed = ( + transform_batch(artifact, accepted_rows, state) + if artifact is not None and accepted_rows + else [] + ) + transformed, output_rejected = _apply_output_schema( + output_schema, + transformed, + config.execution.schema_drift, + ) + if output_rejected: + _write_rejections( + quarantine, + output_rejected, + run_id=state["run_id"], + batch_id=f"batch-{batch_index:08d}", + ) + state["rows_rejected"] = state.get("rows_rejected", 0) + len(output_rejected) + batch_filtered = max( + 0, + len(accepted_rows) - len(transformed) - len(output_rejected), + ) + state["rows_filtered"] = state.get("rows_filtered", 0) + batch_filtered + + warnings = detect_destructive_operations( + {"raw_data": accepted_rows}, + {"transformed_data": transformed}, + state.get("destructive_filter_threshold", 0.3), + ) + raise_if_destructive(warnings, state.get("auto_confirmed", False)) + for warning in warnings: + key = (warning.reason.value, warning.message) + if key not in destructive_seen: + state.setdefault("destructive_warnings", []).append(warning) + destructive_seen.add(key) + stage_ms["transform"] += (time.monotonic() - transform_started) * 1000 + + _raise_if_cancelled(cancellation, state["run_id"]) + active_stage = "load" + load_started = time.monotonic() + written = len(transformed) if dry_run else _write_target(target, transformed) + stage_ms["load"] += (time.monotonic() - load_started) * 1000 + + batch_output_digest = RollingRowsDigest() + batch_output_digest.update(transformed) + output_digest.update(transformed) + state["rows_transformed"] = output_digest.rows + state["rows_loaded"] = state.get("rows_loaded", 0) + written + state["batches_completed"] = batch_index + state["bytes_in"] = input_digest.bytes + state["bytes_out"] = output_digest.bytes + state["input_checksum"] = input_digest.checksum + state["output_checksum"] = output_digest.checksum + state["validation_passed"] = True + + source_start = source_offset + source_offset += len(raw_rows) + last_envelope = BatchEnvelope( + run_id=state["run_id"], + stage_id="load", + partition_id=_PARTITION_ID, + batch_id=f"batch-{batch_index:08d}", + attempt=0, + source_position_start={"offset": source_start}, + source_position_end={"offset": source_offset - 1}, + schema_version=schema_result.version, + transform_artifact_version=artifact.version if artifact is not None else None, + rows_in=len(raw_rows), + rows_out=len(transformed), + rows_rejected=len(rejected) + len(output_rejected), + rows_filtered=batch_filtered, + bytes_in=batch_input_digest.bytes, + bytes_out=batch_output_digest.bytes, + checksum=batch_output_digest.checksum, + input_checksum=batch_input_digest.checksum, + output_checksum=batch_output_digest.checksum, + duration_ms=(time.monotonic() - batch_started) * 1000, + ) + state["last_batch_envelope"] = last_envelope + state["raw_data"] = [] + state["transformed_data"] = [] + yield ("batch", "done", state) + + if input_digest.rows == 0: + raise ValidationError("Source returned 0 rows — nothing to validate") + + state["validation_report"] = { + "rows_checked": input_digest.rows, + "rows_rejected": state.get("rows_rejected", 0), + "columns": quality_columns, + "hard_failures": [], + "soft_warnings": sorted(quality_warnings), + } + for warning in sorted(quality_warnings): + if warning not in state.setdefault("warnings", []): + state["warnings"].append(warning) + + diagnostics = source.diagnostics() + for diagnostic in diagnostics: + if diagnostic not in state.setdefault("warnings", []): + state["warnings"].append(diagnostic) + + _raise_if_cancelled(cancellation, state["run_id"]) + active_stage = "load" + if quarantine is not None: + quarantine.finalize() + if target is not None: + target.finalize() + state["target_published"] = True + + if checkpoints is not None and last_envelope is not None and not dry_run: + checkpoints.save( + Checkpoint( + checkpoint_id=uuid.uuid4().hex, + run_id=state["run_id"], + partition_id=_PARTITION_ID, + batch_id=last_envelope.batch_id, + source_position=last_envelope.source_position_end, + ) + ) + + state["duration_ms"].update(stage_ms) + state["duration_ms"]["total"] = (time.monotonic() - started) * 1000 + yield ("extract", "done", state) + yield ("validate", "done", state) + yield ("transform", "done", state) + yield ("load", "skipped" if dry_run else "done", state) + except Exception as exc: + state["last_error"] = str(exc) + state["duration_ms"].update(stage_ms) + state["duration_ms"]["total"] = (time.monotonic() - started) * 1000 + yield (active_stage, "failed", state) + if isinstance(exc, PipelineError): + raise + raise PipelineError(f"Pipeline failed (run_id={state['run_id']}): {exc}") from exc + finally: + if quarantine is not None: + quarantine.disconnect() + if target is not None: + target.disconnect() + if source is not None: + source.disconnect() + + +def _source_connector(config: PipelineConfig, state: PipelineState) -> SourceConnector: + incremental = config.incremental + if incremental is None or config.source.type not in _INCREMENTAL_PUSHDOWN_SOURCES: + if incremental is not None: + state.setdefault("warnings", []).append( + f"Incremental filtering for source type '{config.source.type}' is applied " + "client-side and scans the source" + ) + return get_source_connector(config.source) + return get_source_connector( + config.source, + incremental_column=incremental.column, + incremental_param=incremental.param or incremental.column, + cursor_value=state.get("cursor_value"), + ) + + +def _filter_incremental_rows( + config: PipelineConfig, + state: PipelineState, + rows: list[dict[str, Any]], +) -> list[dict[str, Any]]: + incremental = config.incremental + if incremental is None or config.source.type in _INCREMENTAL_PUSHDOWN_SOURCES: + return rows + from loafer.core.incremental import filter_rows_after_cursor + + return filter_rows_after_cursor( + rows, + incremental.column, + state.get("cursor_value"), + ) + + +def _advance_cursor( + config: PipelineConfig, + state: PipelineState, + rows: list[dict[str, Any]], +) -> None: + if config.incremental is None: + return + from loafer.core.incremental import max_cursor + + state["new_cursor"] = max_cursor( + rows, + config.incremental.column, + state.get("new_cursor"), + ) + + +def _apply_output_schema( + tracker: SchemaTracker, + rows: list[dict[str, Any]], + policy: str, +) -> tuple[list[dict[str, Any]], list[RejectedRow]]: + if not rows: + return [], [] + result = tracker.apply(rows, policy) + rejected = [ + RejectedRow(row=item.row, stage="transform_output", reason=item.reason) + for item in result.rejected + ] + return result.rows, rejected + + +def _write_target( + target: TargetConnector | None, + rows: list[dict[str, Any]], +) -> int: + if target is None: + raise PipelineError("target connector is unavailable") + return target.write_chunk(rows) + + +def _write_rejections( + quarantine: TargetConnector | None, + rejected: list[RejectedRow], + *, + run_id: str, + batch_id: str, +) -> None: + if not rejected: + return + if quarantine is None: + first = rejected[0] + raise ValidationError( + f"{len(rejected)} rows require quarantine but no quarantine target is configured; " + f"first rejection: {first.reason}" + ) + quarantine.write_chunk( + [ + { + "_loafer": { + "run_id": run_id, + "batch_id": batch_id, + "stage": item.stage, + "reason": item.reason, + }, + "row": item.row, + } + for item in rejected + ] + ) + + +def _merge_quality_counts( + aggregate: dict[str, dict[str, int]], + batch: dict[str, dict[str, int]], +) -> None: + for column, counts in batch.items(): + target = aggregate.setdefault(column, {"total_count": 0, "null_count": 0}) + target["total_count"] += counts["total_count"] + target["null_count"] += counts["null_count"] + + +def _raise_if_cancelled( + cancellation: CancellationPort | None, + run_id: str, +) -> None: + if cancellation is not None and cancellation.is_cancelled(run_id): + raise PipelineError(f"Pipeline cancelled (run_id={run_id})") + + +def rows_size(rows: list[dict[str, Any]]) -> int: + """Return canonical serialized byte size for tests and diagnostics.""" + return sum(len(canonical_row_bytes(row)) for row in rows) + + +__all__ = ["rows_size", "stream_bounded_pipeline", "uses_bounded_data_plane"] diff --git a/loafer/engine.py b/loafer/engine.py index 42cac54..28f241f 100644 --- a/loafer/engine.py +++ b/loafer/engine.py @@ -20,7 +20,12 @@ from collections.abc import Iterator from loafer.llm.base import LLMProvider - from loafer.ports.runtime import ReviewPort, SecretResolver + from loafer.ports.runtime import ( + CancellationPort, + CheckpointPort, + ReviewPort, + SecretResolver, + ) _PROVIDER_ENV_VARS = { "gemini": "GEMINI_API_KEY", @@ -125,6 +130,7 @@ def _build_initial_state( chunk_size=config.chunk_size, streaming_threshold=config.streaming_threshold, destructive_filter_threshold=config.destructive_filter_threshold, + execution_config=config.execution, raw_data=[], transformed_data=[], schema_sample={}, @@ -141,7 +147,19 @@ def _build_initial_state( generated_sql=None, run_id=run_id or uuid.uuid4().hex[:12], rows_extracted=0, + rows_transformed=0, rows_loaded=0, + rows_rejected=0, + rows_filtered=0, + batches_completed=0, + bytes_in=0, + bytes_out=0, + input_checksum=None, + output_checksum=None, + schema_version=None, + transform_artifact_version=None, + last_batch_envelope=None, + target_published=False, duration_ms={}, warnings=[], is_streaming=False, @@ -181,6 +199,8 @@ def execute_pipeline( reviewer: ReviewPort | None = None, secret_resolver: SecretResolver | None = None, provider_factory: ProviderFactory | None = None, + cancellation: CancellationPort | None = None, + checkpoints: CheckpointPort | None = None, ) -> PipelineState: """Execute a validated ETL or ELT pipeline. @@ -215,6 +235,26 @@ def execute_pipeline( factory = provider_factory or _build_llm_provider state["llm_provider"] = factory(config, secret_resolver) + from loafer.data_plane import stream_bounded_pipeline, uses_bounded_data_plane + + if uses_bounded_data_plane(config): + try: + for _stage, _status, _state in stream_bounded_pipeline( + config, + state, + dry_run=dry_run, + cancellation=cancellation, + checkpoints=checkpoints, + ): + pass + except PipelineError: + raise + except Exception as exc: + raise PipelineError(f"Pipeline failed (run_id={state['run_id']}): {exc}") from exc + if not dry_run: + _persist_cursor(state) + return state + mode = config.mode if mode == "etl": @@ -259,6 +299,8 @@ def stream_pipeline( reviewer: ReviewPort | None = None, secret_resolver: SecretResolver | None = None, provider_factory: ProviderFactory | None = None, + cancellation: CancellationPort | None = None, + checkpoints: CheckpointPort | None = None, ) -> Iterator[tuple[str, str, PipelineState]]: """Execute a validated pipeline and yield per-stage runtime updates. @@ -286,6 +328,22 @@ def stream_pipeline( factory = provider_factory or _build_llm_provider state["llm_provider"] = factory(config, secret_resolver) + from loafer.data_plane import stream_bounded_pipeline, uses_bounded_data_plane + + if uses_bounded_data_plane(config): + try: + yield from stream_bounded_pipeline( + config, + state, + dry_run=dry_run, + cancellation=cancellation, + checkpoints=checkpoints, + ) + finally: + if not dry_run and state.get("target_published", False): + _persist_cursor(state) + return + mode = config.mode if mode == "etl": diff --git a/loafer/graph/state.py b/loafer/graph/state.py index cc713a3..a8b4895 100644 --- a/loafer/graph/state.py +++ b/loafer/graph/state.py @@ -43,6 +43,7 @@ class PipelineState(TypedDict, total=False): chunk_size: int streaming_threshold: int destructive_filter_threshold: float + execution_config: Any # Data (mutated per agent) raw_data: list[dict[str, Any]] @@ -74,7 +75,19 @@ class PipelineState(TypedDict, total=False): # Execution metadata run_id: str rows_extracted: int + rows_transformed: int rows_loaded: int + rows_rejected: int + rows_filtered: int + batches_completed: int + bytes_in: int + bytes_out: int + input_checksum: str | None + output_checksum: str | None + schema_version: str | None + transform_artifact_version: str | None + last_batch_envelope: Any | None + target_published: bool duration_ms: dict[str, float] warnings: list[str] is_streaming: bool diff --git a/loafer/ports/connector.py b/loafer/ports/connector.py index edeb5f6..5275b44 100644 --- a/loafer/ports/connector.py +++ b/loafer/ports/connector.py @@ -37,6 +37,10 @@ def read_all(self) -> list[dict[str, Any]]: def count(self) -> int | None: """Row count if cheap to compute, else ``None``.""" + def diagnostics(self) -> list[str]: + """Return redacted warnings collected while streaming.""" + return [] + # -- context manager ----------------------------------------------------- def __enter__(self) -> SourceConnector: diff --git a/loafer/transform/batch_runner.py b/loafer/transform/batch_runner.py new file mode 100644 index 0000000..b730fd4 --- /dev/null +++ b/loafer/transform/batch_runner.py @@ -0,0 +1,211 @@ +"""Prepare row-local transform artifacts once and execute them per batch.""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from loafer.config import ( + AITransformConfig, + CustomTransformConfig, + PipelineTransformConfig, +) +from loafer.core.sandbox import run_sandboxed +from loafer.exceptions import TransformError +from loafer.graph.state import PipelineState +from loafer.llm.schema import build_schema_sample +from loafer.transform.ai_runner import AiTransformRunner +from loafer.transform.code_validator import validate_transform_function + + +@dataclass(frozen=True) +class CodeArtifact: + """One validated custom or AI-generated row-local code artifact.""" + + name: str + kind: str + code: str + + +@dataclass(frozen=True) +class PreparedBatchTransform: + """Immutable transform artifact reused for every batch in one run.""" + + version: str + steps: tuple[CodeArtifact, ...] + stop_on_empty: bool = False + + +def prepare_transform_artifact( + state: PipelineState, + sample_rows: list[dict[str, Any]], +) -> PreparedBatchTransform: + """Generate/load and validate all code once from a bounded sample.""" + config = state.get("transform_config") + if isinstance(config, CustomTransformConfig): + steps = (_custom_artifact(config.path, config.name or "custom"),) + return _prepared(steps) + if isinstance(config, AITransformConfig): + steps = _prepare_ai_artifacts(config, state, sample_rows) + return _prepared(steps) + if isinstance(config, PipelineTransformConfig): + return _prepare_pipeline(config, state, sample_rows) + raise TransformError( + "bounded row_local execution supports custom, AI, or custom/AI pipeline transforms" + ) + + +def transform_batch( + artifact: PreparedBatchTransform, + rows: list[dict[str, Any]], + state: PipelineState, +) -> list[dict[str, Any]]: + """Execute the same prepared artifact against one bounded batch.""" + current = list(rows) + for step in artifact.steps: + current = _execute(step.code, current, state) + if not current and artifact.stop_on_empty: + raise TransformError( + f"row-local transform step '{step.name}' produced 0 rows and " + "stop_on_empty is enabled" + ) + return current + + +def _prepare_pipeline( + config: PipelineTransformConfig, + state: PipelineState, + sample_rows: list[dict[str, Any]], +) -> PreparedBatchTransform: + artifacts: list[CodeArtifact] = [] + current_sample = list(sample_rows) + + for index, step in enumerate(config.steps): + name = step.name or f"step_{index}" + if isinstance(step, CustomTransformConfig): + step_artifacts = (_custom_artifact(step.path, name),) + elif isinstance(step, AITransformConfig): + step_artifacts = _prepare_ai_artifacts(step, state, current_sample, name=name) + else: + raise TransformError( + "SQL pipeline steps require global relational semantics and cannot execute " + "through transform_batch" + ) + + artifacts.extend(step_artifacts) + for artifact in step_artifacts: + current_sample = _execute(artifact.code, current_sample, state) + if not current_sample and config.stop_on_empty: + raise TransformError( + f"row-local transform step '{name}' produced 0 sample rows and " + "stop_on_empty is enabled" + ) + + return _prepared(tuple(artifacts), stop_on_empty=config.stop_on_empty) + + +def _prepare_ai_artifacts( + config: AITransformConfig, + state: PipelineState, + sample_rows: list[dict[str, Any]], + *, + name: str | None = None, +) -> tuple[CodeArtifact, ...]: + custom: CodeArtifact | None = None + if config.custom_path: + custom = _custom_artifact(config.custom_path, f"{name or 'ai'}_custom") + + if config.bypass_ai: + if custom is None: + raise TransformError( + "bypass_ai is set but no custom_path is configured for the row-local transform" + ) + return (custom,) + + prompt_rows = list(sample_rows) + if custom is not None and config.custom_order == "custom_first": + prompt_rows = _execute(custom.code, prompt_rows, state) + + provider = state.get("llm_provider") + if provider is None: + raise TransformError("row-local AI transform requires an LLM provider") + + generator = AiTransformRunner() + generated = generator._generate_ai_code( + provider, + build_schema_sample(prompt_rows, max_sample_rows=5), + config.instruction, + custom.code if custom is not None else None, + state, + ) + if not generated: + raise TransformError("AI provider returned no transform artifact") + + if config.review: + reviewer = state.get("reviewer") + approved = reviewer.approve_transform(generated) if reviewer is not None else False + if not approved: + return (custom,) if custom is not None else () + + ai = CodeArtifact(name=name or config.name or "ai", kind="ai", code=generated) + if custom is None: + return (ai,) + if config.custom_order == "custom_first": + return custom, ai + return ai, custom + + +def _custom_artifact(path: str, name: str) -> CodeArtifact: + source = Path(path) + if not source.exists(): + raise TransformError(f"transform file not found: {source}") + code = source.read_text(encoding="utf-8") + valid, reason = validate_transform_function(code) + if not valid: + raise TransformError(f"Custom transform validation failed: {reason}") + return CodeArtifact(name=name, kind="custom", code=code) + + +def _prepared( + steps: tuple[CodeArtifact, ...], + *, + stop_on_empty: bool = False, +) -> PreparedBatchTransform: + serialized = json.dumps( + [{"name": step.name, "kind": step.kind, "code": step.code} for step in steps], + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + version = f"sha256:{hashlib.sha256(serialized).hexdigest()}" + return PreparedBatchTransform( + version=version, + steps=steps, + stop_on_empty=stop_on_empty, + ) + + +def _execute( + code: str, + rows: list[dict[str, Any]], + state: PipelineState, +) -> list[dict[str, Any]]: + sandbox = state.get("sandbox_config") + timeout = getattr(sandbox, "timeout", 60) + max_memory_mb = getattr(sandbox, "max_memory_mb", 512) + return run_sandboxed( + code, + rows, + timeout=timeout, + max_memory_mb=max_memory_mb, + ) + + +__all__ = [ + "CodeArtifact", + "PreparedBatchTransform", + "prepare_transform_artifact", + "transform_batch", +] diff --git a/skills/loafer-engine/references/execution-contract.md b/skills/loafer-engine/references/execution-contract.md index c526234..c60795e 100644 --- a/skills/loafer-engine/references/execution-contract.md +++ b/skills/loafer-engine/references/execution-contract.md @@ -6,8 +6,8 @@ run_id, stage_id, partition_id, batch_id, attempt source_position_start, source_position_end schema_version, transform_artifact_version -rows_in, rows_out, rows_rejected -bytes_in, bytes_out, checksum +rows_in, rows_out, rows_rejected, rows_filtered +bytes_in, bytes_out, input_checksum, output_checksum ``` ## Row-local flow @@ -24,6 +24,14 @@ read bounded batch Bound the number of batches in flight. Cancellation occurs between safe boundaries. Retry the same artifact and config version. +Loafer's implemented local row-local profile currently uses one batch in flight and atomically +publishes CSV/JSON files after every batch succeeds. PostgreSQL writes to a hidden run-scoped table +and publishes in one final transaction: `replace` is replayable replacement, `error` is atomic +create-once, `append` is at-least-once across a target-commit/checkpoint gap, and deterministic +`upsert` is idempotent by its declared key. MongoDB is rejected until it provides an equivalent +tested staging/merge publication protocol. The validated execution plan exposes the selected +delivery guarantee. + ## Global relational flow Prefer source/target pushdown. Otherwise write partitioned intermediate Arrow/Parquet data and use diff --git a/skills/loafer-engineering/references/architecture.md b/skills/loafer-engineering/references/architecture.md index 2476bfb..ef62a98 100644 --- a/skills/loafer-engineering/references/architecture.md +++ b/skills/loafer-engineering/references/architecture.md @@ -40,11 +40,14 @@ YAML → config.load_config → application.RunPipeline creates a credential-free ExecutionPlan → engine._build_initial_state - → engine selects ETL or ELT LangGraph - → extract agent resolves source adapter and samples schema - → validate agent applies sample-based checks - → ETL: transform runner → load target adapter - → ELT: load_raw target adapter → in-target SQL transform + → declared row_local ETL: bounded data plane + → source batch → schema/validation → prepared transform_batch → staged file target + → BatchEnvelope + rolling reconciliation → atomic publication → final checkpoint + → otherwise engine selects ETL or ELT LangGraph + → extract agent resolves source adapter and samples schema + → validate agent applies sample-based checks + → ETL: materialized transform runner → load target adapter + → ELT: load_raw target adapter → in-target SQL transform → engine persists the local incremental cursor after graph completion → application emits sanitized RunEvent / RunResult contracts ``` @@ -66,6 +69,7 @@ record. Persistence and client surfaces use the credential-free contracts in `lo | `loafer/agents/` | LangGraph stage functions | | `loafer/transform/` | AI, Python, SQL, and multi-step execution | | `loafer/graph/` | Separate ETL and ELT topology | +| `loafer/data_plane.py` | Bounded row-local batch orchestration and atomic publication | | `loafer/engine.py` | In-process graph composition and execution | | `loafer/application/` | Plan, run, validate, and connector-listing use cases | | `loafer/runner.py` | Backward-compatible Python facade over the application service | @@ -84,6 +88,8 @@ Verify before relying on it, but the repository currently contains: - Gemini, Claude, OpenAI, and Qwen provider adapters. - Cursor-based incremental extraction with a local JSON state file. - Postgres and Mongo upsert modes. +- Declared row-local ETL through bounded batches with schema policies, per-batch validation, + quarantine, checksums, cancellation, and atomic CSV/JSON publication. - Python transform subprocess sandboxing on supported operating systems. - CLI validation, connector listing, runs, scheduling, daemon management, logs, and initialization. - Unit, integration, end-to-end, smoke, and opt-in benchmark tests. @@ -99,9 +105,9 @@ Re-check the repository because this reference is a snapshot, not a substitute f ## Architectural risks -- The source stream is drained by `materialize_input_rows()` into a list for AI, custom, SQL, and - multi-step ETL transforms. The ETL load agent then writes from `transformed_data`. -- Sample-based validation does not validate every partition. +- Undeclared/materialized AI, custom, SQL, and multi-step ETL still drain the source stream through + `materialize_input_rows()`; only explicitly declared row-local work uses the bounded data plane. +- The legacy graph path remains sample-validated; the bounded row-local path validates every batch. - Local JSON watermark state is not sufficient for concurrent or distributed workers. - CSV and JSON targets publish atomically, but the local watermark/state file does not yet use the same publication protocol. diff --git a/tests/e2e/test_bounded_data_plane.py b/tests/e2e/test_bounded_data_plane.py new file mode 100644 index 0000000..c782a48 --- /dev/null +++ b/tests/e2e/test_bounded_data_plane.py @@ -0,0 +1,330 @@ +"""End-to-end contracts for declared bounded row-local execution.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from loafer.adapters.runtime import ( + EnvironmentSecretResolver, + InputReviewPort, + NeverCancelled, + NullEventPublisher, +) +from loafer.application import RunRequest, get_local_application +from loafer.application.service import RunPipeline +from loafer.core.batches import RollingRowsDigest +from loafer.exceptions import PipelineError +from loafer.llm.base import TransformPromptResult + + +def _bounded_config( + tmp_path: Path, + *, + transform_code: str | None = None, + rows: int = 5, + chunk_size: int = 2, +) -> tuple[Path, Path, Path]: + source = tmp_path / "input.csv" + source.write_text( + "id,name\n" + "".join(f"{index},name-{index}\n" for index in range(1, rows + 1)), + encoding="utf-8", + ) + transform = tmp_path / "transform.py" + transform.write_text( + transform_code + or ( + "def transform(data):\n" + " return [{**row, 'name': row['name'].upper()} for row in data]\n" + ), + encoding="utf-8", + ) + output = tmp_path / "output.json" + config = tmp_path / "pipeline.yaml" + config.write_text( + "\n".join( + [ + "name: bounded-data-plane", + "mode: etl", + "source:", + " type: csv", + f" path: {source}", + "target:", + " type: json", + f" path: {output}", + "transform:", + " type: custom", + f" path: {transform}", + "execution:", + " transform_class: row_local", + " schema_drift: fail", + f"chunk_size: {chunk_size}", + "", + ] + ), + encoding="utf-8", + ) + return config, source, output + + +def test_csv_custom_json_flows_as_bounded_batches(tmp_path: Path) -> None: + config, _source, output = _bounded_config(tmp_path) + request = RunRequest(config_path=str(config), run_id="bounded-e2e", auto_confirm=True) + + events = list(get_local_application().run_pipeline.stream(request)) + batch_events = [event for event in events if event.stage == "batch"] + final = events[-1].snapshot + + assert len(batch_events) == 3 + assert all(event.batch is not None for event in batch_events) + assert all(event.batch.rows_in <= 2 for event in batch_events if event.batch) + assert ( + len( + { + event.batch.transform_artifact_version + for event in batch_events + if event.batch is not None + } + ) + == 1 + ) + assert final.rows_extracted == 5 + assert final.rows_transformed == 5 + assert final.rows_loaded == 5 + assert final.rows_rejected == 0 + assert final.batches_completed == 3 + assert final.input_checksum + assert final.output_checksum + assert events[-1].stage == "load" + assert json.loads(output.read_text(encoding="utf-8"))[-1]["name"] == "NAME-5" + assert ( + get_local_application().validate(config).plan.delivery_guarantee == "atomic_run_publication" + ) + + +def test_run_checksums_reconcile_with_published_rows(tmp_path: Path) -> None: + config, _source, output = _bounded_config(tmp_path, rows=4, chunk_size=1) + + result = get_local_application().run_pipeline.run( + RunRequest(config_path=str(config), run_id="checksum-e2e", auto_confirm=True) + ) + published = json.loads(output.read_text(encoding="utf-8")) + expected_output = RollingRowsDigest() + expected_output.update(published) + + assert result.snapshot.output_checksum == expected_output.checksum + assert result.snapshot.rows_loaded == expected_output.rows + assert result.snapshot.bytes_out == expected_output.bytes + + +def test_bounded_runtime_state_retains_no_run_sized_row_lists(tmp_path: Path) -> None: + config, _source, _output = _bounded_config(tmp_path, rows=7, chunk_size=2) + + state = get_local_application().run_pipeline.run_state( + RunRequest(config_path=str(config), run_id="bounded-state", auto_confirm=True) + ) + + assert state["raw_data"] == [] + assert state["transformed_data"] == [] + assert state["rows_extracted"] == 7 + assert state["rows_transformed"] == 7 + assert state["batches_completed"] == 4 + + +def test_transform_failure_does_not_replace_existing_output(tmp_path: Path) -> None: + config, _source, output = _bounded_config( + tmp_path, + transform_code=( + "def transform(data):\n" + " for row in data:\n" + " if row['id'] == '3':\n" + " return 1 / 0\n" + " return data\n" + ), + rows=4, + chunk_size=2, + ) + original = [{"existing": True}] + output.write_text(json.dumps(original), encoding="utf-8") + + with pytest.raises(PipelineError, match="division by zero"): + get_local_application().run_pipeline.run( + RunRequest(config_path=str(config), run_id="target-failure", auto_confirm=True) + ) + + assert json.loads(output.read_text(encoding="utf-8")) == original + assert list(tmp_path.glob(".output.json.*.tmp")) == [] + + +def test_target_failure_discards_all_staged_batches( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from loafer.adapters.targets.json_target import JsonTargetConnector + from loafer.connectors import registry + from loafer.exceptions import LoadError + + class _FailingJsonTarget(JsonTargetConnector): + def __init__(self, path: str, write_mode: str = "overwrite") -> None: + super().__init__(path, write_mode) + self._batch_calls = 0 + + def write_chunk(self, chunk: list[dict[str, Any]]) -> int: + self._batch_calls += 1 + written = super().write_chunk(chunk) + if self._batch_calls == 2: + raise LoadError("injected target failure") + return written + + monkeypatch.setitem(registry._TARGET_REGISTRY, "json", _FailingJsonTarget) + config, _source, output = _bounded_config(tmp_path, rows=5, chunk_size=2) + original = [{"existing": True}] + output.write_text(json.dumps(original), encoding="utf-8") + + with pytest.raises(PipelineError, match="injected target failure"): + get_local_application().run_pipeline.run( + RunRequest(config_path=str(config), run_id="load-failure", auto_confirm=True) + ) + + assert json.loads(output.read_text(encoding="utf-8")) == original + assert list(tmp_path.glob(".output.json.*.tmp")) == [] + + +class _CancelAfterFirstBatch: + def __init__(self) -> None: + self.calls = 0 + + def is_cancelled(self, run_id: str) -> bool: + assert run_id == "cancel-bounded" + self.calls += 1 + return self.calls >= 3 + + +class _RecordingCheckpoints: + def __init__(self, output: Path) -> None: + self.output = output + self.saved: list[object] = [] + + def load(self, run_id: str, partition_id: str) -> None: + del run_id, partition_id + return None + + def save(self, checkpoint: object) -> None: + assert self.output.exists() + self.saved.append(checkpoint) + + +def _run_use_case( + cancellation: object, + checkpoints: object, +) -> RunPipeline: + return RunPipeline( + cancellation=cancellation, # type: ignore[arg-type] + checkpoints=checkpoints, # type: ignore[arg-type] + secrets=EnvironmentSecretResolver(), + events=NullEventPublisher(), + reviewer=InputReviewPort(), + ) + + +def test_cancellation_at_batch_boundary_discards_staged_output(tmp_path: Path) -> None: + config, _source, output = _bounded_config(tmp_path, rows=6, chunk_size=2) + use_case = _run_use_case(_CancelAfterFirstBatch(), _RecordingCheckpoints(output)) + + with pytest.raises(PipelineError, match="cancelled"): + use_case.run( + RunRequest(config_path=str(config), run_id="cancel-bounded", auto_confirm=True) + ) + + assert not output.exists() + assert list(tmp_path.glob(".output.json.*.tmp")) == [] + + +def test_checkpoint_is_saved_only_after_atomic_publication(tmp_path: Path) -> None: + config, _source, output = _bounded_config(tmp_path, rows=3, chunk_size=2) + checkpoints = _RecordingCheckpoints(output) + use_case = _run_use_case(NeverCancelled(), checkpoints) + + use_case.run( + RunRequest(config_path=str(config), run_id="checkpoint-bounded", auto_confirm=True) + ) + + assert len(checkpoints.saved) == 1 + + +def test_validation_quarantine_reconciles_rejected_rows(tmp_path: Path) -> None: + config, _source, output = _bounded_config(tmp_path, rows=3, chunk_size=2) + quarantine = tmp_path / "quarantine.json" + text = config.read_text(encoding="utf-8") + text = text.replace( + " schema_drift: fail\n", + ( + " schema_drift: fail\n" + f" quarantine_path: {quarantine}\n" + "validation:\n" + " column_types:\n" + " id: integer\n" + " on_failure: quarantine\n" + ), + ) + config.write_text(text, encoding="utf-8") + + result = get_local_application().run_pipeline.run( + RunRequest(config_path=str(config), run_id="quarantine-bounded", auto_confirm=True) + ) + + assert result.snapshot.rows_extracted == 3 + assert result.snapshot.rows_loaded == 0 + assert result.snapshot.rows_rejected == 3 + assert json.loads(output.read_text(encoding="utf-8")) == [] + rejected = json.loads(quarantine.read_text(encoding="utf-8")) + assert len(rejected) == 3 + assert rejected[0]["_loafer"]["stage"] == "validation" + + +def test_ai_artifact_is_generated_once_and_reused_per_batch(tmp_path: Path) -> None: + source = tmp_path / "input.csv" + source.write_text("id\n1\n2\n3\n", encoding="utf-8") + output = tmp_path / "output.json" + config = tmp_path / "pipeline.yaml" + config.write_text( + "\n".join( + [ + "source:", + " type: csv", + f" path: {source}", + "target:", + " type: json", + f" path: {output}", + "transform:", + " type: ai", + " instruction: copy every row", + "execution:", + " transform_class: row_local", + "chunk_size: 1", + "llm:", + " provider: openai", + " model: test-model", + " api_key: test-key", + "", + ] + ), + encoding="utf-8", + ) + provider = MagicMock() + provider.generate_transform_function.return_value = TransformPromptResult( + code="def transform(data):\n return data\n", + raw_response="identity", + token_usage={"total_tokens": 10}, + ) + + get_local_application(provider_factory=lambda _config, _secrets: provider).run_pipeline.run( + RunRequest(config_path=str(config), run_id="ai-bounded", auto_confirm=True) + ) + + provider.generate_transform_function.assert_called_once() + assert len(json.loads(output.read_text(encoding="utf-8"))) == 3 diff --git a/tests/integration/test_postgres_staging.py b/tests/integration/test_postgres_staging.py new file mode 100644 index 0000000..4c1102a --- /dev/null +++ b/tests/integration/test_postgres_staging.py @@ -0,0 +1,127 @@ +"""Integration coverage for atomic PostgreSQL run publication.""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from loafer.adapters.targets.postgres_staging import PostgresStagingTargetConnector + +pytestmark = pytest.mark.integration + +_TABLE = "public.test_loafer_staged_publication" + + +def _target( + postgres_url: str, + *, + run_id: str, + write_mode: str, + key: list[str] | None = None, +) -> PostgresStagingTargetConnector: + return PostgresStagingTargetConnector( + postgres_url, + _TABLE, + write_mode, + key, + run_id, + ) + + +def _seed_old_target(pg_conn: Any) -> None: + cursor = pg_conn.cursor() + cursor.execute(f"DROP TABLE IF EXISTS {_TABLE}") + cursor.execute(f"CREATE TABLE {_TABLE} (id BIGINT, name TEXT)") + cursor.execute(f"INSERT INTO {_TABLE} (id, name) VALUES (1, 'old')") + + +def _rows(pg_conn: Any) -> list[tuple[int, str]]: + cursor = pg_conn.cursor() + cursor.execute(f"SELECT id, name FROM {_TABLE} ORDER BY id") + return cursor.fetchall() + + +def test_replace_is_invisible_until_atomic_publication( + postgres_url: str, + pg_conn: Any, +) -> None: + _seed_old_target(pg_conn) + target = _target(postgres_url, run_id="replace-success", write_mode="replace") + try: + target.connect() + target.write_chunk([{"id": 2, "name": "new"}]) + target.write_chunk([{"id": 3, "name": "newer"}]) + + assert _rows(pg_conn) == [(1, "old")] + + target.finalize() + assert _rows(pg_conn) == [(2, "new"), (3, "newer")] + finally: + target.disconnect() + pg_conn.cursor().execute(f"DROP TABLE IF EXISTS {_TABLE}") + + +def test_disconnect_before_finalize_preserves_target_and_discards_stage( + postgres_url: str, + pg_conn: Any, +) -> None: + _seed_old_target(pg_conn) + target = _target(postgres_url, run_id="replace-cancelled", write_mode="replace") + stage = target._staging_table + try: + target.connect() + target.write_chunk([{"id": 2, "name": "unpublished"}]) + target.disconnect() + + assert _rows(pg_conn) == [(1, "old")] + cursor = pg_conn.cursor() + cursor.execute("SELECT to_regclass(%s)", (stage,)) + assert cursor.fetchone()[0] is None + finally: + target.disconnect() + pg_conn.cursor().execute(f"DROP TABLE IF EXISTS {_TABLE}") + + +def test_append_merges_all_staged_rows_in_one_transaction( + postgres_url: str, + pg_conn: Any, +) -> None: + _seed_old_target(pg_conn) + target = _target(postgres_url, run_id="append-success", write_mode="append") + try: + target.connect() + target.write_chunk([{"id": 2, "name": "new"}]) + + assert _rows(pg_conn) == [(1, "old")] + + target.finalize() + assert _rows(pg_conn) == [(1, "old"), (2, "new")] + finally: + target.disconnect() + pg_conn.cursor().execute(f"DROP TABLE IF EXISTS {_TABLE}") + + +def test_upsert_is_atomic_and_idempotent_by_key( + postgres_url: str, + pg_conn: Any, +) -> None: + _seed_old_target(pg_conn) + try: + for run_id in ("upsert-one", "upsert-two"): + target = _target( + postgres_url, + run_id=run_id, + write_mode="upsert", + key=["id"], + ) + try: + target.connect() + target.write_chunk([{"id": 1, "name": "updated"}, {"id": 2, "name": "new"}]) + target.finalize() + finally: + target.disconnect() + + assert _rows(pg_conn) == [(1, "updated"), (2, "new")] + finally: + pg_conn.cursor().execute(f"DROP TABLE IF EXISTS {_TABLE}") diff --git a/tests/unit/connectors/test_csv_source.py b/tests/unit/connectors/test_csv_source.py index bd906e3..95c3cf4 100644 --- a/tests/unit/connectors/test_csv_source.py +++ b/tests/unit/connectors/test_csv_source.py @@ -3,6 +3,7 @@ from __future__ import annotations from typing import Any +from unittest.mock import patch import pytest @@ -87,3 +88,38 @@ def test_count_after_stream(self, tmp_path: Any) -> None: rows = conn.read_all() assert len(rows) == 3 assert conn.count() == 3 + + def test_connect_scans_encoding_with_bounded_reads(self, tmp_path: Any) -> None: + f = tmp_path / "data.csv" + f.write_text("id,name\n1,Alice\n2,Bob\n", encoding="utf-8") + real_open = open + read_sizes: list[int] = [] + calls = 0 + + class TrackingFile: + def __init__(self, handle: Any) -> None: + self._handle = handle + + def read(self, size: int = -1) -> str: + read_sizes.append(size) + return self._handle.read(size) + + def __iter__(self) -> Any: + return iter(self._handle) + + def __getattr__(self, name: str) -> Any: + return getattr(self._handle, name) + + def tracked_open(*args: Any, **kwargs: Any) -> Any: + nonlocal calls + calls += 1 + handle = real_open(*args, **kwargs) + return TrackingFile(handle) if calls == 1 else handle + + with patch("builtins.open", side_effect=tracked_open): + connector = CsvSourceConnector(str(f)) + connector.connect() + connector.disconnect() + + assert read_sizes + assert all(size > 0 for size in read_sizes) diff --git a/tests/unit/connectors/test_pdf_source.py b/tests/unit/connectors/test_pdf_source.py index cc1f378..6af7cb0 100644 --- a/tests/unit/connectors/test_pdf_source.py +++ b/tests/unit/connectors/test_pdf_source.py @@ -2,6 +2,7 @@ from __future__ import annotations +import time from pathlib import Path import pytest @@ -72,6 +73,53 @@ def pdf_path(self, tmp_path: Path) -> Path: pdf.write_bytes(pdf_content) return pdf + @pytest.fixture + def table_pdf_path(self, tmp_path: Path) -> Path: + """Create a native-text PDF containing a ruled two-column table.""" + content = b"""0.5 w +72 720 m 300 720 l S +72 690 m 300 690 l S +72 660 m 300 660 l S +72 660 m 72 720 l S +180 660 m 180 720 l S +300 660 m 300 720 l S +BT /F1 12 Tf 82 702 Td (Name) Tj ET +BT /F1 12 Tf 190 702 Td (Value) Tj ET +BT /F1 12 Tf 82 672 Td (Alice) Tj ET +BT /F1 12 Tf 190 672 Td (42) Tj ET +""" + objects = [ + b"<< /Type /Catalog /Pages 2 0 R >>", + b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>", + ( + b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] " + b"/Contents 4 0 R /Resources << /Font << /F1 5 0 R >> >> >>" + ), + b"<< /Length " + str(len(content)).encode() + b" >>\nstream\n" + content + b"endstream", + b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>", + ] + payload = bytearray(b"%PDF-1.4\n") + offsets = [0] + for number, obj in enumerate(objects, start=1): + offsets.append(len(payload)) + payload.extend(f"{number} 0 obj\n".encode()) + payload.extend(obj) + payload.extend(b"\nendobj\n") + xref_offset = len(payload) + payload.extend(f"xref\n0 {len(objects) + 1}\n".encode()) + payload.extend(b"0000000000 65535 f \n") + for offset in offsets[1:]: + payload.extend(f"{offset:010d} 00000 n \n".encode()) + payload.extend( + ( + f"trailer\n<< /Size {len(objects) + 1} /Root 1 0 R >>\n" + f"startxref\n{xref_offset}\n%%EOF\n" + ).encode() + ) + path = tmp_path / "native-table.pdf" + path.write_bytes(payload) + return path + def test_connect_and_disconnect(self, pdf_path: Path) -> None: from loafer.connectors.registry import PdfSourceConnector @@ -116,6 +164,13 @@ def test_stream_includes_tables_when_enabled(self, pdf_path: Path) -> None: all_rows = [row for chunk in chunks for row in chunk] assert "tables" in all_rows[0] assert "table_count" in all_rows[0] + assert "table_provenance" in all_rows[0] + assert all_rows[0]["provenance"] == { + "source_path": str(pdf_path.resolve()), + "page_number": 1, + "content_type": "native_pdf", + "ocr_applied": False, + } def test_stream_excludes_tables_when_disabled(self, pdf_path: Path) -> None: from loafer.connectors.registry import PdfSourceConnector @@ -128,6 +183,26 @@ def test_stream_excludes_tables_when_disabled(self, pdf_path: Path) -> None: all_rows = [row for chunk in chunks for row in chunk] assert "tables" not in all_rows[0] + def test_native_table_fixture_preserves_page_and_table_provenance( + self, + table_pdf_path: Path, + ) -> None: + from loafer.connectors.registry import PdfSourceConnector + + with PdfSourceConnector(str(table_pdf_path), extract_tables=True) as conn: + rows = conn.read_all() + + assert rows[0]["tables"] == [[["Name", "Value"], ["Alice", "42"]]] + assert rows[0]["table_provenance"] == [ + { + "source_path": str(table_pdf_path.resolve()), + "page_number": 1, + "table_index": 0, + "row_count": 2, + "column_count": 2, + } + ] + def test_stream_chunking(self, pdf_path: Path) -> None: from loafer.connectors.registry import PdfSourceConnector @@ -169,3 +244,65 @@ def test_file_not_found_raises(self, tmp_path: Path) -> None: conn = PdfSourceConnector(str(tmp_path / "missing.pdf")) with pytest.raises(Exception, match="failed to open PDF"): conn.connect() + + def test_page_limit_is_enforced_before_extraction(self, pdf_path: Path) -> None: + from loafer.connectors.registry import PdfSourceConnector + + conn = PdfSourceConnector(str(pdf_path), max_pages=2) + + with pytest.raises(Exception, match=r"3 pages.*2-page limit"): + conn.connect() + assert conn._doc is None + + def test_file_size_limit_is_enforced_before_parser_open(self, tmp_path: Path) -> None: + from loafer.connectors.registry import PdfSourceConnector + + large = tmp_path / "large.pdf" + large.write_bytes(b"x" * (1024 * 1024 + 1)) + conn = PdfSourceConnector(str(large), max_file_size_mb=1) + + with pytest.raises(Exception, match=r"exceeding.*1MB limit"): + conn.connect() + + def test_skip_policy_reports_page_failure(self) -> None: + from loafer.connectors.registry import PdfSourceConnector + + class _BrokenPage: + def extract_text(self) -> str: + raise RuntimeError("broken content stream") + + class _Document: + def __init__(self) -> None: + self.pages = [_BrokenPage()] + + conn = PdfSourceConnector("/tmp/fake.pdf", page_failure_policy="skip") + conn._doc = _Document() + + assert list(conn.stream(chunk_size=1)) == [] + assert conn.diagnostics() == ["PDF page 1 extraction failed: broken content stream"] + + def test_page_timeout_is_enforced_and_reported(self, monkeypatch: pytest.MonkeyPatch) -> None: + from loafer.connectors.registry import PdfSourceConnector + + class _Page: + pass + + class _Document: + def __init__(self) -> None: + self.pages = [_Page()] + + conn = PdfSourceConnector( + "/tmp/fake.pdf", + page_timeout_seconds=0.01, + page_failure_policy="skip", + ) + conn._doc = _Document() + + def _slow_page(_page_num: int, _page: object) -> dict[str, object]: + time.sleep(0.1) + return {} + + monkeypatch.setattr(conn, "_extract_page", _slow_page) + + assert list(conn.stream(chunk_size=1)) == [] + assert "page exceeded 0.01s timeout" in conn.diagnostics()[0] diff --git a/tests/unit/connectors/test_postgres_staging_target.py b/tests/unit/connectors/test_postgres_staging_target.py new file mode 100644 index 0000000..d880d66 --- /dev/null +++ b/tests/unit/connectors/test_postgres_staging_target.py @@ -0,0 +1,167 @@ +"""Tests for run-scoped PostgreSQL staging publication.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from loafer.adapters.targets.postgres_staging import PostgresStagingTargetConnector +from loafer.config import PostgresTargetConfig +from loafer.connectors.registry import get_staged_target_connector +from loafer.exceptions import LoadError +from tests.postgres_sql import render_sql + + +def _connector( + *, + table: str = "analytics.events", + write_mode: str = "replace", + key: list[str] | None = None, +) -> PostgresStagingTargetConnector: + connector = PostgresStagingTargetConnector( + "postgresql://user:pass@localhost/db", + table, + write_mode, + key, + "run/with unsafe text", + ) + connector._conn = MagicMock() + connector._cursor = MagicMock() + return connector + + +def test_stage_name_is_safe_and_keeps_target_schema() -> None: + connector = _connector(table='analytics.events"; DROP TABLE users; --') + + assert connector._staging_table.startswith("analytics._loafer_stage_") + assert '"' not in connector._staging_table + assert ";" not in connector._staging_table + + +def test_create_staging_quotes_adversarial_columns() -> None: + connector = _connector() + + connector._create_staging({'name"; DROP TABLE users; --': "safe"}) + + query = render_sql(connector._cursor.execute.call_args.args[0]) + stage_name = connector._staging_table.split(".", 1)[1] + assert query == ( + f'CREATE TABLE "analytics"."{stage_name}" ("name""; DROP TABLE users; --" TEXT)' + ) + + +def test_write_chunk_inserts_only_into_stage_and_commits() -> None: + connector = _connector() + + with patch("psycopg2.extras.execute_values") as execute_values: + assert connector.write_chunk([{"id": 1, "name": "Ada"}]) == 1 + + query = render_sql(execute_values.call_args.args[1]) + assert query.startswith('INSERT INTO "analytics"."_loafer_stage_') + assert '"analytics"."events"' not in query + connector._conn.commit.assert_called_once() + + +def test_write_chunk_rolls_back_and_wraps_stage_creation_failure() -> None: + connector = _connector() + connector._cursor.execute.side_effect = RuntimeError("database unavailable") + + with pytest.raises(LoadError, match="staging batch insert failed"): + connector.write_chunk([{"id": 1}]) + + connector._conn.rollback.assert_called_once() + + +def test_replace_drops_final_and_renames_stage_in_one_finalize_commit() -> None: + connector = _connector() + connector._staging_created = True + connector._columns = ["id"] + + connector.finalize() + + queries = [render_sql(call.args[0]) for call in connector._cursor.execute.call_args_list] + assert queries[0] == 'DROP TABLE IF EXISTS "analytics"."events"' + assert queries[1].startswith('ALTER TABLE "analytics"."_loafer_stage_') + assert queries[1].endswith(' RENAME TO "events"') + connector._conn.commit.assert_called_once() + assert connector._published is True + + +def test_append_merges_stage_then_drops_it() -> None: + connector = _connector(write_mode="append") + connector._staging_created = True + connector._columns = ["id", "name"] + connector._table_exists = MagicMock(return_value=True) + + connector.finalize() + + queries = [render_sql(call.args[0]) for call in connector._cursor.execute.call_args_list] + assert queries[0].startswith('INSERT INTO "analytics"."events" ("id", "name") SELECT ') + assert queries[1].startswith('DROP TABLE IF EXISTS "analytics"."_loafer_stage_') + connector._conn.commit.assert_called_once() + + +def test_upsert_quotes_keys_and_columns() -> None: + connector = _connector( + write_mode="upsert", + key=['id"; DROP TABLE users; --'], + ) + connector._staging_created = True + connector._columns = ['id"; DROP TABLE users; --', "name"] + connector._table_exists = MagicMock(return_value=True) + + connector.finalize() + + queries = [render_sql(call.args[0]) for call in connector._cursor.execute.call_args_list] + merge = next(query for query in queries if query.startswith("INSERT INTO")) + assert 'ON CONFLICT ("id""; DROP TABLE users; --")' in merge + assert '"name" = EXCLUDED."name"' in merge + + +def test_disconnect_discards_unpublished_stage() -> None: + connector = _connector() + connector._staging_created = True + mock_conn = connector._conn + mock_cursor = connector._cursor + + connector.disconnect() + + queries = [render_sql(call.args[0]) for call in mock_cursor.execute.call_args_list] + assert any(query.startswith("DROP TABLE IF EXISTS") for query in queries) + mock_conn.rollback.assert_called_once() + mock_conn.close.assert_called_once() + + +def test_empty_append_requires_an_existing_target_schema() -> None: + connector = _connector(write_mode="append") + connector._table_exists = MagicMock(return_value=False) + + with pytest.raises(LoadError, match="no output schema"): + connector.finalize() + + +def test_empty_replace_truncates_existing_target() -> None: + connector = _connector(write_mode="replace") + connector._table_exists = MagicMock(return_value=True) + + connector.finalize() + + query = render_sql(connector._cursor.execute.call_args.args[0]) + assert query == 'TRUNCATE TABLE "analytics"."events"' + assert connector._published is True + + +def test_registry_selects_staging_adapter_for_bounded_postgres() -> None: + config = PostgresTargetConfig( + type="postgres", + url="postgresql://localhost/db", + table="public.output", + write_mode="upsert", + key="id", + ) + + connector = get_staged_target_connector(config, run_id="run-123") + + assert isinstance(connector, PostgresStagingTargetConnector) + assert connector._key == ["id"] diff --git a/tests/unit/test_batch_policies.py b/tests/unit/test_batch_policies.py new file mode 100644 index 0000000..190e886 --- /dev/null +++ b/tests/unit/test_batch_policies.py @@ -0,0 +1,99 @@ +"""Tests for bounded-batch schema, validation, and checksum policies.""" + +from __future__ import annotations + +import pytest + +from loafer.config import ValidationConfig +from loafer.core.batches import ( + RollingRowsDigest, + SchemaTracker, + validate_batch, +) +from loafer.exceptions import ValidationError + + +def test_rolling_checksum_is_independent_of_batch_boundaries() -> None: + rows = [{"id": 1, "name": "a"}, {"id": 2, "name": "b"}, {"id": 3, "name": "c"}] + whole = RollingRowsDigest() + split = RollingRowsDigest() + + whole.update(rows) + split.update(rows[:1]) + split.update(rows[1:]) + + assert split.rows == whole.rows == 3 + assert split.bytes == whole.bytes + assert split.checksum == whole.checksum + + +def test_schema_fail_policy_rejects_new_columns() -> None: + tracker = SchemaTracker() + tracker.apply([{"id": 1, "name": "a"}], "fail") + + with pytest.raises(ValidationError, match="new columns: extra"): + tracker.apply([{"id": 2, "name": "b", "extra": True}], "fail") + + +def test_schema_quarantine_rejects_only_drifted_rows() -> None: + tracker = SchemaTracker() + tracker.apply([{"id": 1}], "quarantine") + + result = tracker.apply([{"id": 2}, {"id": "wrong"}], "quarantine") + + assert result.rows == [{"id": 2}] + assert len(result.rejected) == 1 + assert "changed from integer to string" in result.rejected[0].reason + + +def test_schema_coerce_normalizes_rows_to_baseline() -> None: + tracker = SchemaTracker() + first = tracker.apply([{"id": 1, "active": True}], "coerce") + second = tracker.apply( + [{"id": "2", "active": "false", "ignored": "drop"}], + "coerce", + ) + + assert first.version == second.version + assert second.rows == [{"active": False, "id": 2}] + assert second.rejected == [] + + +def test_schema_evolve_changes_content_addressed_version() -> None: + tracker = SchemaTracker() + first = tracker.apply([{"id": 1}], "evolve") + second = tracker.apply([{"id": 2, "name": "new"}], "evolve") + + assert second.evolved is True + assert second.schema == {"id": "integer", "name": "string"} + assert second.version != first.version + + +def test_validation_fail_reports_first_invalid_row() -> None: + config = ValidationConfig( + required_columns=["id"], + column_types={"id": "integer"}, + on_failure="fail", + ) + + with pytest.raises(ValidationError, match="row 1"): + validate_batch([{"id": 1}, {"id": "bad"}], config) + + +def test_validation_quarantine_keeps_valid_rows_and_counts_nulls() -> None: + config = ValidationConfig( + max_null_rate=0.25, + strict=True, + required_columns=["id"], + column_types={"id": "integer"}, + on_failure="quarantine", + ) + + result = validate_batch( + [{"id": 1, "name": "a"}, {"id": None, "name": "b"}], + config, + ) + + assert result.rows == [{"id": 1, "name": "a"}] + assert len(result.rejected) == 1 + assert result.column_counts["id"] == {"total_count": 2, "null_count": 1} diff --git a/tests/unit/test_bounded_config.py b/tests/unit/test_bounded_config.py new file mode 100644 index 0000000..78f055e --- /dev/null +++ b/tests/unit/test_bounded_config.py @@ -0,0 +1,174 @@ +"""Configuration contracts for bounded and global execution semantics.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from loafer.application.service import _delivery_guarantee +from loafer.config import PipelineConfig +from loafer.exceptions import ConfigError + + +def _custom_path(tmp_path: Path) -> Path: + path = tmp_path / "transform.py" + path.write_text("def transform(data):\n return data\n", encoding="utf-8") + return path + + +def test_existing_pipelines_keep_materialized_execution_default(tmp_path: Path) -> None: + config = PipelineConfig( + source={"type": "rest_api", "url": "https://example.test/rows"}, + target={"type": "json", "path": str(tmp_path / "out.json")}, + transform={"type": "custom", "path": str(_custom_path(tmp_path))}, + ) + + assert config.execution.transform_class == "materialized" + assert config.execution.schema_drift == "fail" + + +def test_sql_is_classified_as_global_relational_by_default(tmp_path: Path) -> None: + config = PipelineConfig( + source={"type": "rest_api", "url": "https://example.test/rows"}, + target={"type": "json", "path": str(tmp_path / "out.json")}, + transform={"type": "sql", "query": "SELECT * FROM {{source}}"}, + ) + + assert config.execution.transform_class == "global_relational" + + +def test_row_local_sql_is_rejected_with_global_semantics_guidance(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="global semantics"): + PipelineConfig( + source={"type": "rest_api", "url": "https://example.test/rows"}, + target={"type": "json", "path": str(tmp_path / "out.json")}, + transform={"type": "sql", "query": "SELECT * FROM {{source}}"}, + execution={"transform_class": "row_local"}, + ) + + +def test_row_local_mongo_target_is_rejected_until_atomic_protocol_exists() -> None: + with pytest.raises(ValueError, match="PostgreSQL staging target"): + PipelineConfig( + source={"type": "rest_api", "url": "https://example.test/rows"}, + target={ + "type": "mongo", + "url": "mongodb://localhost/db", + "database": "test", + "collection": "output", + }, + transform={"type": "ai", "instruction": "copy rows"}, + execution={"transform_class": "row_local"}, + ) + + +def test_quarantine_policy_requires_a_path(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="quarantine_path"): + PipelineConfig( + source={"type": "rest_api", "url": "https://example.test/rows"}, + target={"type": "json", "path": str(tmp_path / "out.json")}, + transform={"type": "custom", "path": str(_custom_path(tmp_path))}, + execution={ + "transform_class": "row_local", + "schema_drift": "quarantine", + }, + ) + + +def test_csv_cannot_evolve_schema_after_header_publication(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="header is fixed"): + PipelineConfig( + source={"type": "rest_api", "url": "https://example.test/rows"}, + target={"type": "csv", "path": str(tmp_path / "out.csv")}, + transform={"type": "custom", "path": str(_custom_path(tmp_path))}, + execution={ + "transform_class": "row_local", + "schema_drift": "evolve", + }, + ) + + +def test_postgres_staging_cannot_evolve_schema_after_creation(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="staging table schema is fixed"): + PipelineConfig( + source={"type": "rest_api", "url": "https://example.test/rows"}, + target={ + "type": "postgres", + "url": "postgresql://localhost/db", + "table": "public.output", + }, + transform={"type": "custom", "path": str(_custom_path(tmp_path))}, + execution={ + "transform_class": "row_local", + "schema_drift": "evolve", + }, + ) + + +@pytest.mark.parametrize( + ("write_mode", "key", "expected"), + [ + ("replace", None, "atomic_transactional_replace"), + ("error", None, "atomic_transactional_create_once"), + ("append", None, "at_least_once_atomic_merge"), + ("upsert", "id", "idempotent_keyed_atomic_merge"), + ], +) +def test_postgres_delivery_guarantee_is_explicit( + tmp_path: Path, + write_mode: str, + key: str | None, + expected: str, +) -> None: + config = PipelineConfig( + source={"type": "rest_api", "url": "https://example.test/rows"}, + target={ + "type": "postgres", + "url": "postgresql://localhost/db", + "table": "public.output", + "write_mode": write_mode, + "key": key, + }, + transform={"type": "custom", "path": str(_custom_path(tmp_path))}, + execution={"transform_class": "row_local"}, + ) + + assert _delivery_guarantee(config) == expected + + +def test_validation_null_rate_must_be_a_fraction(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="between 0 and 1"): + PipelineConfig( + source={"type": "rest_api", "url": "https://example.test/rows"}, + target={"type": "json", "path": str(tmp_path / "out.json")}, + transform={"type": "custom", "path": str(_custom_path(tmp_path))}, + validation={"max_null_rate": 1.5}, + ) + + +def test_load_config_wraps_bounded_contract_errors(tmp_path: Path) -> None: + from loafer.config import load_config + + config = tmp_path / "pipeline.yaml" + config.write_text( + "\n".join( + [ + "source:", + " type: rest_api", + " url: https://example.test/rows", + "target:", + " type: json", + f" path: {tmp_path / 'out.json'}", + "transform:", + " type: sql", + " query: SELECT * FROM {{source}}", + "execution:", + " transform_class: row_local", + ] + ), + encoding="utf-8", + ) + + with pytest.raises(ConfigError, match="global semantics"): + load_config(config) diff --git a/tests/unit/test_full_pipeline_benchmark.py b/tests/unit/test_full_pipeline_benchmark.py index 56dcd40..0a8a98a 100644 --- a/tests/unit/test_full_pipeline_benchmark.py +++ b/tests/unit/test_full_pipeline_benchmark.py @@ -11,6 +11,8 @@ from benchmarks.full_pipeline import ( _csv_data_row_count, _generate_input, + _git_revision, + _git_worktree_dirty, _process_group_rss_bytes, _sha256, _verified_throughput, @@ -47,6 +49,39 @@ def test_process_group_rss_includes_current_process() -> None: assert _process_group_rss_bytes(os.getpgrp()) > 0 +def test_process_group_rss_ignores_processes_that_exit_during_scan( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class VanishedProcess: + name = "123" + + def __truediv__(self, _name: str) -> VanishedProcess: + return self + + def read_text(self, *, encoding: str) -> str: + raise ProcessLookupError("process exited") + + monkeypatch.setattr(Path, "iterdir", lambda _path: iter([VanishedProcess()])) + + assert _process_group_rss_bytes(123) == 0 + + +def test_git_provenance_is_optional_when_image_has_no_git( + tmp_path: Path, +) -> None: + def missing_git(*_args: object, **_kwargs: object) -> None: + raise FileNotFoundError("git") + + with pytest.MonkeyPatch.context() as monkeypatch: + monkeypatch.setattr( + "benchmarks.full_pipeline.subprocess.run", + missing_git, + ) + + assert _git_revision(tmp_path) is None + assert _git_worktree_dirty(tmp_path) is None + + def test_throughput_is_only_reported_for_verified_output() -> None: assert _verified_throughput(1000, 2.0, correct=True) == 500.0 assert _verified_throughput(10_000_000, 18.0, correct=False) is None