Add bounded batch data plane

CHANGELOG.md

@@ -12,13 +12,44 @@ 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
 
 - The CLI, scheduler, and legacy Python runner now share the same application boundary while core
   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
 

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.

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

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),
     }