Dask DataFrames
===============

Install ``pipewhy[dask]`` to capture a trusted local Dask DataFrame. The first
adapter accepts explicitly declared pandas ``int64``, nullable ``Int64`` and
Python-backed string columns. The index is excluded. Other DataFrame types
and remote cluster execution are not supported by this adapter.

.. code-block:: python

   from pathlib import Path
   import dask
   import dask.dataframe as dd
   import pandas as pd
   from pipewhy.adapters.dask import DaskSource
   from pipewhy.cli.capture import scan_source
   from pipewhy.formats import DaskLimits, DaskSchema, DaskWorkerLimits
   from pipewhy.formats import PandasField, PandasLimits, PandasSchema, dask_source_version
   from pipewhy.snapshots import LocalSnapshotStore

   data = pd.DataFrame({"id": pd.array([1, 2, 3], dtype="Int64")})
   with dask.config.set({"dataframe.convert-string": False}):
       frame = dd.from_pandas(data, npartitions=2, sort=False)
   schema = DaskSchema(PandasSchema((PandasField(0, "id", "Int64"),)))
   limits = DaskLimits(PandasLimits(1000, 10, 1024, 1000000, 100, 10000), 1000000, 10)
   workers = DaskWorkerLimits(2147483648, 90000, 100000, 1000000)
   source = DaskSource("people", frame, schema, limits, workers)
   version = dask_source_version(frame, schema=schema, limits=limits, worker_limits=workers)
   snapshot = scan_source(source, LocalSnapshotStore(Path("snapshots")), version)

Each partition executes in a fresh supervised local process with an explicit
memory limit and deadline. The worker request is limited to 65,536 bytes. Graph bytes, partition count, records, fields,
encoded staging, output frames and output bytes are bounded. The graph is
prepared and serialized in the caller process before worker execution; its
CPU time and transient serialization memory are not covered by the worker
limit. The graph-byte limit bounds serialized output, not graph construction.
During verification, retained records plus one additional partition can be in
memory; the staging limit applies to each observation. Total runtime can
include one deadline per partition per observation.

Only use trusted graphs: Dask tasks execute Python code, and the adapter stages
serialized tasks temporarily for its workers. The temporary files are removed
after each worker completes. Source versions include ordered partition results;
capture repeats the observation to detect accidental changes before publishing.
This does not guarantee an atomic read of mutable or nondeterministic sources.
Strict normalization is the default; logical normalization is opt-in.
Snapshots reopen without Dask or the original source.

Dask Arrays
-----------

One- and two-dimensional Dask Arrays use an explicit ``DaskArraySchema``
wrapping a ``NumpySchema``. The first version supports signed and unsigned
64-bit integers, 64-bit floats and booleans. Chunk sizes must be known; a
two-dimensional array must have a single chunk across its columns. Row chunks
are executed in order, and each computed block must match its declared size.
Structured arrays and masked arrays are rejected.

.. code-block:: python

   import dask.array as da
   import numpy as np
   from pipewhy.adapters.dask import DaskArraySource
   from pipewhy.formats import DaskArrayLimits, DaskArraySchema, NumpyLimits, NumpySchema, dask_array_source_version

   values = da.from_array(np.array([[1, 2], [3, 4]], dtype=">i8"), chunks=(1, 2))
   array_schema = DaskArraySchema(NumpySchema((0, 1), "big", 2))
   array_limits = DaskArrayLimits(NumpyLimits(1000, 10, 1000000, 100, 10000), 1000000, 10)
   source = DaskArraySource("measurements", values, array_schema, array_limits, workers)
   version = dask_array_source_version(values, schema=array_schema, limits=array_limits, worker_limits=workers)
   snapshot = scan_source(source, LocalSnapshotStore(Path("snapshots")), version)

Array capture uses the same trusted-graph, worker, staging and repeated-read
boundaries described above. Its schema identity is distinct from a DataFrame
or a local NumPy array.