feat: FlowSource for streaming-native architecture#125
Merged
Conversation
The streaming sibling of FlowIO<A>: a lazy, resource-safe, error-as-value stream whose sole consumption path is .Compile(), which lands enumeration back inside the FlowIO envelope. Ships the minimum shape per ADR-0023: - FlowSource<A> + FlowSourceCompiler<A> (Drain/Fold/ToList terminals) - lazy Map/Where combinators - deferred acquisition via a FlowResource-derived pull bracket; a source built but never run acquires nothing - terminal-RuntimeError default error channel, with Attempt/SkipErrors/ Rethrow bridging to a per-item FlowSource<EffResult<A>> for dead-lettering Also adds FlowIO.Uncancellable(): bracket release must run even when cancellation is what aborted the stream, but Lift/LiftAsync short-circuit on an already-cancelled token — the release-on-cancellation test caught this. Promotes "Streams" to the fifth Prelude primitive (Prelude README + core-architecture 1.2) per ADR-0023. 14 FlowSource tests green; 69/69 Prelude tests pass. Claude-Session: https://claude.ai/code/session_01XR7Kr3Eupi8bhu4pXCWxn8
Adds the sink side of the compile surface. IFlowSink<T> (in the Prelude, so Into can reference it) is a batch lifecycle — OpenAsync, WriteBatchAsync per BatchSize-sized chunk, CompleteAsync — with DisposeAsync on every exit path so a sink can roll back when completion is not reached. FlowSourceCompiler.Into(sink) drives it inside the effect envelope: it re-chunks the row-at-a-time stream into the sink's BatchSize batches (writer controls batch size, decoupling read batch from write batch), completes on success, and disposes the sink (then releases the byte source) on every path. Pull-based, so a slow sink paces a fast source in O(batch) memory. 3 Into tests (batched write + complete + dispose ordering; dispose-without- complete on failure; sink lifecycle nested inside the resource bracket); 17 FlowSource tests green. Claude-Session: https://claude.ai/code/session_01XR7Kr3Eupi8bhu4pXCWxn8
IFormatRowReader.DeserializeRows gains a [EnumeratorCancellation] CancellationToken (default) so a streaming read is cancellable mid-enumeration. Threaded through every serializer: JSON (DeserializeAsyncEnumerable), CSV (GetRecordsAsync), Excel + Parquet (CopyToAsync for the seek buffer + a ThrowIfCancellationRequested per row/row-group; Parquet also to ParquetReader.CreateAsync). ComposedStorageAdapter's Load and InspectInternal now pass their token through. The default value keeps non-token call sites compiling; test stubs updated to the new signature. 197 core Storage + 31 Csv + 13 Excel + 15 Parquet tests green. Claude-Session: https://claude.ai/code/session_01XR7Kr3Eupi8bhu4pXCWxn8
Derives a read-only streaming view of a collection item: IItem<IEnumerable<TRow>> -> IReadOnlyItem<FlowSource<TRow>>, whose Load() yields a deferred FlowSource (O(batch) peak read memory). - ISupportsStreamingView<TRow>: the capability seam on ComposedStorageAdapter, exposing OpenStreamingSource() which brackets medium.ReadStream() (acquired on first pull, released on every exit path) over reader.DeserializeRows, and reuses TranslateSchemaMismatch so a mid-stream SchemaMismatchException lands as the typed RuntimeError.SchemaMismatch (translation stays in the storage layer; the Prelude never sees the exception). - Streaming Load() is a NEW deferred path (returns a FlowSource description), not an eager IContainerAdapter. - .AsStream() is gated to streaming-capable composed formats (IFormatStreamReader); a direct adapter or non-streaming format throws at wire-up. A .Constrain() wrapper is unwrapped to reach the composed format. - Read-only: the streaming item's Save fails; node concerns delegate to the eager origin. 8 AsStream tests (load+compile, deferred acquisition, fan-out re-acquire, disposal, typed schema-mismatch, wire-up gate, read-only, constrain-unwrap). Claude-Session: https://claude.ai/code/session_01XR7Kr3Eupi8bhu4pXCWxn8
…nsaction) (#122) Add EFCoreBulkSink<T, TContext>, an IFlowSink<T> that consumes a compiled FlowSource one BatchSize-sized batch at a time. It opens a DbContext and a transaction on OpenAsync, issues a BulkInsertAsync per WriteBatchAsync enlisted in that transaction (O(batch) memory instead of the eager path's O(dataset) re-materialisation), commits on CompleteAsync, and rolls back on DisposeAsync-without-Complete so a partial-stream failure leaves no corrupt-but-present rows — keeping IsTransactional honest. BulkSink.Insert<T, TContext>(factory, options) surfaces the sink (mirroring BulkSave.Insert), with IDbContextFactory and Func<TContext> overloads; BatchSize flows from BulkSaveOptions (default 2000). The eager BulkSave API is unchanged. Tests drive FlowSource -> Into(sink) over a shared SQLite in-memory connection: all rows on success, incremental per-batch writes verified by batch count, empty stream commits nothing, and a mid-stream failure rolls the whole write back (asserting a batch was inserted first, then the table is empty). Claude-Session: https://claude.ai/code/session_01XR7Kr3Eupi8bhu4pXCWxn8
Makes the core JSON serializer honest about streaming: it already reads incrementally via JsonSerializer.DeserializeAsyncEnumerable (top-level array elements, no whole-array buffering), so it now implements IFormatStreamReader and declares StorageTraits.CanStream = true. That lets ItemFactory.Enumerable.Json<T>(...).AsStream() produce a working IReadOnlyItem<FlowSource<T>> — the smallest proof that streaming lives in core, not just an extension. The trait/marker drift law in IFormatSerializerLaws now validates the honest claim (CanStream == IFormatStreamReader) for the JSON binding. Corrected the IFormatStreamReader doc which used JSON-as-buffering as its counter-example. 2 JSON streaming tests + 100 JSON/laws tests green. Claude-Session: https://claude.ai/code/session_01XR7Kr3Eupi8bhu4pXCWxn8
Adds SeekableSpill (core, Data.Storage): makes a forward-only byte source seekable by spilling to a bounded temp file (FileOptions.DeleteOnClose) instead of buffering the whole object into a MemoryStream. Peak RAM drops from O(object) to O(copy buffer); an already-seekable source passes through un-owned. Parquet and Excel both drop their per-serializer MemoryStream buffers for the shared primitive (an `await using SeekableSpill`), so a non-seekable (S3/HTTP) Parquet read now holds one row group in RAM regardless of object size — temp-file lifetime is owned by the FlowSource bracket. Parquet already implements IFormatStreamReader, so .AsStream() over Parquet works via #116. 2 SeekableSpill tests + 38 Parquet + 46 Excel tests green. Claude-Session: https://claude.ai/code/session_01XR7Kr3Eupi8bhu4pXCWxn8
Coverage run (FlowthruCoverage / dotnet --collect) flagged StreamingItem at ~46% line coverage — the Exists/LoadUntyped/SaveUntyped/DataType delegation members were unexercised. Adds focused tests; FlowSource itself is at 100%.
… migrate Python to FlowSource (#117, #119) Adds StepContainerKind.Source and teaches Item.Introspection (ContainerKindOf / RowTypeOf) to resolve a Flowthru.Prelude.FlowSource<T> payload to Source with row type T — fixing the misclassification where a FlowSource fell through to Singleton with row type FlowSource<T>. Removes the superseded StepContainerKind.AsyncStream kind and the IAsyncStreamMarshaller marker (unused bare-IAsyncEnumerable scaffolding FlowSource replaces per ADR-0023), updating the FT1301/FT1303 analyzers, their diagnostics, and all affected tests. Migrates the Python extension off the removed markers so the build stays green: the descriptor's doc + tests no longer reference the deleted kind, and its negative-scope assertion now targets Source. The Python step still consumes the eager Enumerable view and materialises via Arrow (ArrowMarshaller.ToList); chunk-wise Arrow streaming is deferred (a v1 step cannot receive a FlowSource, and the wire protocol change was out of scope for this atomic change). Claude-Session: https://claude.ai/code/session_01XR7Kr3Eupi8bhu4pXCWxn8
Add a FUnit streaming affordance for FlowSource<T>: Samples.Source / IEnumerable.AsStream to lift samples into a stream, InvokeStream/RunStream/DrainInto to compile-drain a source (or a step returning one) and unwind the EffResult, TrackingSource to observe bracket release + mid-stream cancellation, and RecordingSink to observe sink lifecycle. Assertions cover both error modes (terminal RuntimeError and per-item dead-letter). Add FT1201 (Flowthru.Storage) trait-honesty analyzer: a format serializer declaring StorageTraits.CanStream = true that either omits the IFormatStreamReader<TRow> marker or whose DeserializeRows body materialises the whole input (whole-document JsonSerializer. Deserialize/DeserializeAsync, or ToList/ToArray over the input stream). The body check is the compile-time signal the runtime drift law cannot give; keyed to the stream parameter to avoid flagging bounded metadata materialisation (e.g. Parquet's row-group reader). Claude-Session: https://claude.ai/code/session_01XR7Kr3Eupi8bhu4pXCWxn8
The intent-level capstone: AddBulkLoad(source.AsStream(), sink) wires a streaming source item to an IFlowSink as an on-DAG identity step, so the common bulk-load gets scheduling / caching / pre-flight / the read cap while the Flow dev writes intent, not Compile/Into. - FlowSinkItem<T> : IItem<FlowSource<T>> — a write-only DAG output whose Save(FlowSource) compiles the source Into the sink; Load fails (a sink is not a readable source). - AddBulkLoad wraps the generic AddStep<FlowSource<T>, FlowSource<T>> with an identity transform, inputs: the streaming source, outputs: a FlowSinkItem. End-to-end test runs a real flow (JSON .AsStream -> AddBulkLoad -> recording sink) through the scheduler and asserts all rows written + IsSuccess. The eager/streaming mismatch analyzer was scoped out: the mismatch is already a compile error (invariant IItem<T>), and #118's FT1201 covers trait honesty. 3 tests green (FlowSinkItem Save/Load + end-to-end flow run). Claude-Session: https://claude.ai/code/session_01XR7Kr3Eupi8bhu4pXCWxn8
Add StreamingParquetOverS3Tests: a Docker-gated tier that seeds a multi-row-group Parquet object into a Testcontainers MinIO and reads it back through the .AsStream() catalog view (ADR-0023) — the O(row group) path. Asserts every streamed read returns all seeded rows in order (count + spot-check + Id checksum), including a wide fan-out of concurrent streaming reads. Gated on TestCapabilities.Docker: Inconclusive (never a failure) on a runtime-less host, mirroring MinioContainerBackend's gating and ParquetOverS3ConcurrencyTests' seeding/concurrency knobs. Per the tier convention the memory ceiling is imposed externally: the test asserts correctness only, and the constrained-container invocation (RSS stays flat as FLOWTHRU_STREAM_ROWS grows) is documented in tests/extensions/CONTRIBUTING.md. No core files touched. Claude-Session: https://claude.ai/code/session_01XR7Kr3Eupi8bhu4pXCWxn8
…SQLite, self-measured memory (#112) The runnable, teachable form of the #111 downstream case: a multi-row-group Parquet dataset bulk-loaded into SQLite two ways over the same schema — an eager Flow (materialise the whole file, O(file)) and a streaming Flow (.AsStream().Map().Where() into an AddBulkLoad + BulkSink.Insert sink, O(batch)). Program.cs instruments each variant with a background RSS sampler, writes the peaks to a Raw memory_samples.csv, and a pure-Flowthru Reporting Flow renders memory_report.md from a checked-in template — the example proving its own thesis. At the 200k-row default, streaming holds peak managed memory to ~15% of eager while both load the same 196,000 rows. Claude-Session: https://claude.ai/code/session_01XR7Kr3Eupi8bhu4pXCWxn8
Spelkington
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Implements the streaming-native reads milestone (ADR-0023): bound a catalog read's peak memory to O(batch) instead of O(file), so Flowthru can process datasets larger than RAM on memory-constrained hosts (AWS Lambda, lightweight ECS/Fargate) — the principled root fix behind the #111 Fargate OOM.
What's in it
FlowSource<T>— a lazy, resource-safe, error-as-value streaming primitive (the streaming sibling ofFlowIO), vendored minimally and consumed only by compiling back intoFlowIO. (feat(core): FlowSource<T> — minimal streaming effect primitive (compile-to-FlowIO) #113)DeserializeRows. (feat(core): thread cancellation into DeserializeRows ([EnumeratorCancellation]) #114)FlowSink<T>+.Compile().Into(sink)batch-sink terminal. (feat(core): FlowSink<T> + FlowSource.Into terminal #115).AsStream()— a read-only streaming catalog view with deferred, bracketedLoad(). (feat(core): .AsStream() catalog view + deferred streaming Load() #116)StepContainerKind.Source+ introspection; the oldAsyncStream/IAsyncStreamMarshallerscaffolding removed. (refactor(core): StepContainerKind.Source; remove AsyncStream/IAsyncStreamMarshaller #117)IFlowSink<T>EFCore.Bulk streaming sink (per-batch, one transaction). (feat(efcore-bulk): IFlowSink<T> streaming sink (per-batch, single transaction) #122)AddBulkLoad— the on-DAG, intent-level bulk-load helper. (feat(core): AddBulkLoad helper + eager/streaming mismatch analyzer #123)StreamingBulkLoadadvanced example — large Parquet → SQLite, self-measuring: eager peak 166 MB vs streaming 25 MB (15% of eager), rendered by a Flowthru Reporting flow.Proof, not just claims
FlowSource.csat 100% line coverage; the whole stack 84–100%..AsStream()→AddBulkLoad→ sink,IsSuccess); test(s3): constrained-memory streaming regression tier (MinIO + cgroup) #124 streams a multi-row-group Parquet over real MinIO (8/8 concurrent reads consistent, driving theSeekableSpill).Issues
Closes #112. Closes #113. Closes #114. Closes #115. Closes #116. Closes #117. Closes #118. Closes #120. Closes #121. Closes #122. Closes #123. Closes #124.
#119 stays open by design (do not auto-close). Python's chunk-wise Arrow marshalling is deferred: a v1 step consumes the eager view (forced-upstream
FlowSourceinputs are forbidden), and true incremental marshalling needs an Arrow-IPC wire-protocol change on both the C# and Python sides. The marker migration landed here (build green); the O(batch) guarantee holds for the C# path but not across a Python step. Tracked as a scoped follow-up in #119.