Streaming: fetch only the parts of a dataset an analysis actually touches (split from #151)

[jeremymanning]

Summary

Split out of #151 at the repository owner's request. #151 ("support data sharing") builds the
data-package object: declared data is staged to a private HuggingFace repo (or carried inline when
small) and dereferenced on the worker on demand. That gets the whole declared dataset to the
worker. This issue is about the harder half: getting only the parts of a dataset that a given
analysis actually touches, so an analysis can run on a machine that cannot hold the whole thing.

The owner's framing, quoted in full

From @jeremymanning on #151:

re: streaming: i'm not 100% sure how to leverage this properly. if there is a way we can
systematically figure out which part(s) of the dataset are needed at a given time, then this would
be incredibly powerful because we could (for example) run an analysis on a machine that couldn't
actually fit the full dataset-- along the lines of how HF streaming enables inference on models
when the machine executing the inference can't fit the full dataset. but generalizing this may be
more complicated. perhaps it's possible by doing something like running a mock simulation that
wraps every function recursively to either execute the function (if it takes less than some very
small threshold amount of time) OR simply track which parts of the dataset are accessed inside
that piece. i'm imagining this would use threading and timers in some way, and then interrupt
function calls if they took too long. and then we'd have to do some sort of fancy bookkeeping to
tag different parts of the execution and figure out which parts of the dataset go with which tags,
and then organize the to-be-streamed dataset so that parts could be copied over on demand. unless
i'm missing an obvious solution, this is likely more complex than should be attempted in this
initial "support data sharing" implementation. if so, we should open a new issue to address the
streaming functionality separately, and defer this part of this issue accordingly.

And, on why the deferral:

this is likely more complex than should be attempted in this initial "support data sharing"
implementation.

What this issue must decide before any code

The sketch above has two separable mechanisms, and they have very different risk profiles.

1. Access tracking — discover which parts of a dataset a function touches.
2. Speculative partial execution — run the function under a time budget, interrupting calls that
   run long, and attribute the accesses observed so far to execution "tags".

(2) is the expensive and dangerous part. Interrupting arbitrary user code part-way through, by
threads and timers, is not generally safe: a partially executed function may have already written a
file, posted to an API, or mutated shared state, and there is no way to know from the outside. Any
plan here has to say what class of function it is willing to speculate on, and how a user opts in.

(1) is tractable on its own and may be most of the value. Concretely cheaper alternatives worth
pricing before building the simulator:

• Lazy chunked handles. The data package hands the function an array-like/table-like object whose
  __getitem__ fetches the covering chunk on demand and caches it. No tracing, no interruption; the
  access pattern is the fetch pattern. Costs a wrapper type per supported format.
• Ride existing streaming. datasets.load_dataset(..., streaming=True) and zarr/h5py over a
  remote store already solve this for their own formats. Clustrix may only need to hand the worker
  credentials plus a URI, not invent a mechanism.
• Declared partitioning. The caller says how the dataset splits and which partition a call needs.
  Explicit, unglamorous, and consistent with Make Clustrix a data mover: stage declared inputs and outputs (reverses the "not a data mover" limitation) #151's "declaration, never inference" rule.
• Recorded first run. Run once with full data and record accesses; use the recording to prefetch
  on subsequent runs. Sidesteps interruption entirely, at the cost of needing one full-size run.

Acceptance criteria (to be refined once an approach is chosen)

• A written comparison of the four alternatives above against the simulator sketch, with the
  interruption-safety question answered explicitly.
• Whatever is built must obey Make Clustrix a data mover: stage declared inputs and outputs (reverses the "not a data mover" limitation) #151's rules: declaration over inference, digest verification of every
  fetched part, and no silent fallback to fetching the whole dataset when streaming fails.
• Verified against a real remote store with a dataset larger than the worker's memory. Per this
  repository's standard, a mocked demonstration proves nothing here.

Explicitly out of scope

Same list as #151: no sync/mirror/watch, no DAG or data-derived ordering, no cluster-to-cluster
transfer, no provenance database.

Related

• Make Clustrix a data mover: stage declared inputs and outputs (reverses the "not a data mover" limitation) #151 (the data-package implementation this was split out of)

[jeremymanning]

Tracked as a sub-issue of #160 (deferred). Consolidated after a three-agent audit that verified every open issue against the code; see #160 for the plan and the ordering.