Local format source boundary
============================

This page describes the development source tree. The published ``0.1.0.dev0``
package predates the local source boundary and CSV reader APIs described here.

``pipewhy.formats.verified_local_file(path, sha256_hex=digest, byte_limit=limit)``
opens a regular local file under an explicit lowercase SHA-256 content identity.
The positive byte limit applies to the source and to reads through the returned
binary handle. The context manager owns and closes the handle, checks the content
and file identity before use, and verifies them again on successful exit.

An exception in the caller propagates and closes the handle. A mismatched digest,
changed file, replacement path, unsupported source or exceeded limit fails rather
than reporting a successful read. The check detects accidental changes; it does
not authenticate a producer or guarantee an atomic snapshot against an adversary.
The expected digest must come from the caller's source-version evidence.
``LocalSourceError.reason`` distinguishes invalid configuration, unsupported
sources, byte limits, version mismatches, changed sources and prematurely closed
handles. Filesystem I/O errors remain explicit exceptions.

This boundary does not parse a format or publish a snapshot. A capture coordinator
must wait for complete parsing and successful context exit before reporting COMPLETE
or publishing a snapshot. Remote object sources require separate adapters and their
own version and consistency guarantees.

The low-level CSV and JSON Lines readers also accept a ``SourceOpener``:
a caller-supplied context manager factory receiving the expected digest and byte
limit. The opener owns the binary stream and must enforce its size limit, verify
the complete source against that digest, and check the source again before a
successful exit. Reader batches remain provisional through that exit. A custom
opener is a trusted boundary: the reader cannot prove that an implementation
performed those checks. Snapshot adapters currently accept only the verified
local-file path, and accepting an opener does not itself add a remote connector
or establish S3-compatible storage support.

CSV record batches
------------------

``pipewhy.formats.iter_csv_batches`` reads an explicitly versioned local file
using a ``CsvSchema`` and ``CsvLimits``. Schemas contain ordered ``CsvField``
definitions with stable field IDs, declared types and explicit null markers.
The header must match the schema. Text, booleans, bounded signed and unsigned
integers, finite decimals, finite binary64 values and supported timestamps are
available. Strict normalization is the default; logical normalization must be
selected explicitly.

``CsvSchema.descriptor`` returns the versioned schema description used to derive
its SHA-256 identity. ``CsvSchema.from_descriptor`` validates and reconstructs
that description, rejecting unknown fields, unsupported versions and invalid
field definitions. Each returned description is independent of the schema object.

The caller supplies limits for source bytes, record bytes, field bytes, field
count, records and words per batch, and total records. An oversized or malformed
record raises ``CsvReadError`` without including its value in the error message.
Source failures retain their specific category in the CSV error reason.
The interpreter's current CSV field limit can be inspected with
``csv_field_character_limit()``. The reader does not change that process-wide
setting.

Batches are provisional until the iterator is fully exhausted and source
verification succeeds. Closing it early is not a complete capture. Remote
storage remains separate work.

Configured local CSV source
---------------------------

``pipewhy.adapters.csv.CsvSource`` binds a caller-selected logical source ID,
local path, schema and limits. ``CsvAdapter`` discovers the configured root and
records dataset without opening the file. Complete discovery therefore means the
configured scope is enumerated; it does not assert that the file is readable.

Reads require the exact lowercase SHA-256 of the requested file bytes. A requested
batch size is an upper bound and is capped by the configured batch limit. The
adapter does not retain historical versions; a caller may supply an older version
only while its exact bytes are still available at the configured path. Source
lineage remains UNKNOWN with an explicit reason. Emitted batches remain provisional
until the iterator completes and verifies its source.

CSV snapshots
-------------

The Python ``scan_source`` API accepts a ``CsvSource`` and an exact source
digest. Once a complete read and verification succeed, the capture records the
number of rows actually observed as both its expected and returned counts; this
is not an independent count supplied by the upstream producer. The capture
stores the schema descriptor and can be reopened without the source file.
Failures discard provisional records and publish an unavailable observation
with an explicit reason. Source lineage remains unknown.

Command-line configuration
--------------------------

Select CSV explicitly and supply the schema and resource limits in the TOML
configuration. Loading the configuration does not open the source file::

   schema_version = 1

   [store]
   path = "snapshots"

   [sources.orders]
   kind = "csv"
   path = "orders.csv"
   source_id = "orders"
   schema = { format = "pipewhy.csv-schema", version = 1, encoding = "utf-8", fields = [{ id = 0, name = "amount", type = "int64", nullable = false, null_markers = [] }] }
   limits = { source_bytes = 4096, record_bytes = 1024, field_bytes = 256, field_count = 16, batch_records = 2, batch_words = 1024, max_records = 100 }

The source version for CSV is the full lowercase SHA-256 digest of the exact
file bytes. For example, obtain it with
``python -c "import hashlib; print(hashlib.sha256(open('orders.csv','rb').read()).hexdigest())"``
and pass the result to ``pipewhy scan orders --config pipewhy.toml
--source-version DIGEST``. The example digest command reads the whole file;
for large files, calculate the same digest with a streaming checksum tool.
PipeWhy verifies the digest while reading and never assumes a filename implies
the format. CSV support uses the Python standard library.

JSON Lines record batches
-------------------------

The development-source API ``iter_jsonl_batches`` reads explicitly configured
flat JSON objects, one per UTF-8 line, into provisional normalized batches. A
``JsonlSchema`` declares stable field IDs, scalar types, required fields and
nullability; absent optional fields and present nulls remain distinct. The reader
rejects duplicate or unknown fields, nested data, malformed input and values
outside its explicit limits. Its source digest and batches have the same local
verification and provisional-read rules described above.
The line limit counts raw bytes including any line terminator; the field-byte
limit counts UTF-8 bytes in decoded field names and string values. Numeric token
length and decimal exponent have separate explicit limits.

``JsonlSchema.descriptor`` returns a fresh versioned schema description, and
``JsonlSchema.from_descriptor`` validates and reconstructs it. The schema
identity includes field IDs, names, types, presence and null policies.

``JsonlSource`` and ``JsonlAdapter`` bind a caller-selected logical source ID,
local path, schema and limits. Discovery enumerates only the configured scope and
does not establish that the file is readable. Reads use the exact SHA-256 digest,
and upstream lineage is unknown. This adapter does not retain historical files
or publish snapshots by itself. The Python ``scan_source`` API accepts a
``JsonlSource``, stores the schema descriptor and reopens the capture without the
original file. Complete counts are observed after verification, not an
independent upstream count. Failed reads discard provisional records and retain
an explicit reason.

For command-line scans, select ``kind = "jsonl"`` in a source table. For example::

   [sources.events]
   kind = "jsonl"
   path = "events.jsonl"
   source_id = "events"
   schema = { format = "pipewhy.jsonl-schema", version = 1, fields = [{ id = 0, name = "id", type = "int64", required = true, nullable = false }, { id = 1, name = "note", type = "text", required = false, nullable = true }] }
   limits = { source_bytes = 4096, line_bytes = 1024, field_bytes = 256, field_count = 8, batch_records = 2, batch_words = 1024, max_records = 100, number_bytes = 32, decimal_exponent = 100 }

Include the same top-level ``schema_version`` and store configuration shown for
CSV, then run ``pipewhy scan events --config pipewhy.toml --source-version DIGEST``
with the full lowercase SHA-256 of the exact JSON Lines file bytes. The format is
selected by configuration, never inferred from its file extension.

JSON document record batches
----------------------------

``iter_json_document_batches`` accepts a ``JsonDocumentSchema`` with explicit
``object`` or ``array`` framing and a ``JsonDocumentLimits``. An object is one
record; an array contains flat objects in source order. The initial subset uses
the same declared scalar, null and missing-field policies as JSON Lines. Nested
values, duplicate or unknown keys, malformed documents and mismatched framing
fail explicitly.

The complete document is read into memory under an explicit document-byte bound,
separate from the source-byte bound. Other limits cover fields, numeric tokens,
decimal exponent, record count and batch size. A custom source opener must provide
a blocking binary stream and fulfill the verification obligations above. All
batches remain provisional until source verification completes.

``JsonDocumentSchema.descriptor`` includes the framing and field policies used
to calculate its identity. ``JsonDocumentSchema.from_descriptor`` strictly
validates and reconstructs this description for later capture metadata.

``JsonDocumentSource`` and ``JsonDocumentAdapter`` configure a local path,
logical source ID, schema and limits. Discovery covers the configured scope
without claiming the file is readable; source lineage remains unknown. Reads
require an exact content digest and do not retain old file versions. The Python
``scan_source`` API accepts ``JsonDocumentSource`` and stores the framing and
schema descriptor in a snapshot that can be reopened without the original file.
Complete counts describe records observed after verification, not independent
upstream counts. Failed reads discard provisional data and retain a reason.

For command-line scans, select ``kind = "json-document"`` and declare the
document framing explicitly::

   [sources.orders]
   kind = "json-document"
   path = "orders.json"
   source_id = "orders"
   schema = { format = "pipewhy.json-document-schema", version = 1, framing = "array", fields = [{ id = 0, name = "amount", type = "int64", required = true, nullable = false }] }
   limits = { source_bytes = 4096, document_bytes = 4096, field_bytes = 256, field_count = 8, batch_records = 2, batch_words = 1024, max_records = 100, number_bytes = 32, decimal_exponent = 100 }

With the top-level store configuration shown above, pass the full SHA-256 digest
to ``pipewhy scan orders --config pipewhy.toml --source-version DIGEST``.

Parquet input boundaries
------------------------

``read_parquet_batches`` also accepts a trusted verified source opener. The
reader stages its binary stream locally and independently checks its size and
SHA-256 digest before starting the bounded worker. A failure in the opener's
final verification prevents any decoded records from being returned. This
interface allows storage adapters to supply verified streams; it does not itself
implement a remote storage connector. Snapshot capture currently accepts local
paths.
