feat: generic EtlPipeline core (#147) - #276
Merged
Merged
Conversation
Add a generic, format-agnostic ETL pipeline to Wolfgang.Etl.Abstractions that format packages (Csv/Json/Xml/SqlBulkCopy) will extend with class-named source factories and sink terminators. - EtlPipeline: static entry with the Source sentinel (for format-package source extensions) and From(IAsyncEnumerable) / From(ExtractorBase) built-in factories. Named EtlPipeline (not Pipeline) to avoid clashing with the existing fluent Pipeline builder and System.IO.Pipelines. - IEtlPipeline<T>: lazy, strongly-typed operator surface — Where/Select/SelectMany (sync + async), Distinct (custom comparer), Take/Skip, Tap (sync + async), Buffer, To(LoaderBase) terminator, and AsAsyncEnumerable escape hatch. - IEtlPipelineSink.RunAsync(progress, ct) with end-to-end cancellation. - EtlPipelineProgress record (extracted/loaded/filtered/errored/elapsed). Nothing runs until the terminal sink enumerates the stream. Cancellation is observed per record. 30 unit tests cover every operator (sync + async), Buffer, Distinct-with-comparer, an operator exception, mid-stream cancellation, progress counters, AsAsyncEnumerable, and all argument guards. All 287 tests pass; src builds clean across all TFMs (warnings-as-errors). Closes #147 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
ETL-Transformers already implements Where/Select/SelectMany/Distinct/Take/ Skip/Buffer/etc. as ITransformAsync transformers. Reimplementing them inside Abstractions duplicated that logic (a third copy after System.Linq.Async and ETL-Transformers), and Abstractions can't depend on Transformers to reuse it (Transformers -> Abstractions). Strip the operators from the core. IEtlPipeline<T> now exposes only the plumbing: Through(ITransformAsync) / Through(ITransformWithCancellationAsync) to append transformer stages, To(LoaderBase) to terminate, and AsAsyncEnumerable() to escape. The LINQ-flavored operators will ship as extension methods in Wolfgang.Etl.Transformers (layered on Through), reusing the transformers that already exist there — gated on the Abstractions 0.16.0 release, same as the format-package factories. EtlPipelineProgress drops RecordsFiltered/RecordsErrored (the core no longer sees per-record operator decisions) and reports RecordsExtracted / RecordsLoaded / Elapsed. 272 tests pass; src builds clean across all TFMs. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Chris-Wolfgang
changed the title
Runnable Net8.0 + Net4.8 example that builds a generic EtlPipeline from an IAsyncEnumerable source, chains three Through stages (string -> int -> int -> string, showing the element type flowing across the chain and same-type stages being fine), terminates with a LoaderBase sink, and reports EtlPipelineProgress (RecordsExtracted at the source, RecordsLoaded at the sink). Wired into the solution under the existing Net80/Net48 folders. Both build clean; the Net8.0 build runs end-to-end. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Document the format-agnostic EtlPipeline core (From/Through/To/AsAsyncEnumerable) alongside the fluent Pipeline: chained Through with type flow, EtlPipelineProgress, cancellation, the AsAsyncEnumerable escape hatch, the operator (Wolfgang.Etl.Transformers) and format-package extension layers, a Pipeline-vs-EtlPipeline comparison table, and a link to Example8. Add a matching key-feature bullet in the introduction. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Chris-Wolfgang
commented
Jul 20, 2026
Chris-Wolfgang
commented
Jul 20, 2026
Chris-Wolfgang
commented
Jul 20, 2026
The DoubleTransformer read as 'produces a double' but was typed int -> int. Make it genuinely int -> double (item * 2.0) and FormatTransformer double -> string, so the chain now flows string -> int -> double -> string — three distinct types, and the transformer's name matches its output type. Format with :F1 so the double is visible in the output. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Add two Through overloads that take a stream-to-stream delegate instead of an ITransformAsync class: Through<TOut>(Func<IAsyncEnumerable<T>, IAsyncEnumerable<TOut>>) Through<TOut>(Func<IAsyncEnumerable<T>, CancellationToken, IAsyncEnumerable<TOut>>) These are the delegate form of the transformer primitive (same contract as ITransformAsync.TransformAsync), so they stay at the core's 'append a stage' level and let callers supply a one-off stage inline without declaring a class. Distinct from a per-element Func<T,TOut> (that's Select, an operator that lives in Wolfgang.Etl.Transformers). Unlike AsAsyncEnumerable()+From(), Through keeps the pipeline's extracted/loaded counting intact. 4 new tests; 276 pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Add a "Supplying a stage" subsection to the generic EtlPipeline docs: Through accepts either an ITransformAsync class or a stream-to-stream delegate (Func<IAsyncEnumerable<T>, IAsyncEnumerable<TOut>>, plus a cancellation-aware overload) for one-off inline stages, distinct from a per-element Select operator. Also note the minimal From(...).To(...) form when the source output already matches the loader input. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This was referenced Jul 20, 2026
This was referenced Jul 22, 2026
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
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 Abstractions-side scope of #147 — the core of a generic, format-agnostic ETL pipeline. Operators live downstream (see the placement note).
Placement decision (option B)
ETL-Transformers already ships
Where/Select/SelectMany/Distinct/Take/Skip/Buffer/… asITransformAsynctransformers. Reimplementing them here duplicated that logic, and Abstractions can't depend on Transformers to reuse it (the dependency runs Transformers → Abstractions). So the core stays minimal and the operators become extension methods inWolfgang.Etl.Transformers(layered onThrough), reusing the transformers that already exist there.What's here (core only)
EtlPipeline— static entry: theSourcesentinel (extension-dispatch target for format-package source factories) + built-inFrom(IAsyncEnumerable<T>)/From(ExtractorBase<T,TProgress>)factories. NamedEtlPipeline, notPipeline, sincePipelineis the existing fluent builder.IEtlPipeline<T>— the plumbing:Through(ITransformAsync<T,TOut>)/Through(ITransformWithCancellationAsync<T,TOut>)to append transformer stages,To<TProgress>(LoaderBase<T,TProgress>)to terminate,AsAsyncEnumerable()to escape to rawIAsyncEnumerable.IEtlPipelineSink.RunAsync(progress, ct)— end-to-end cancellation.EtlPipelineProgress—RecordsExtracted/RecordsLoaded/Elapsed(the two ends the core observes).Verification
Fromboth sources,Throughsingle/chained/cancellation-aware,AsAsyncEnumerable, transformer exception, mid-stream cancellation, progress counters, all argument guards).srcbuilds clean across all TFMs (net462 → net10.0), warnings-as-errors.Follow-ups (gated on Abstractions 0.16.0 shipping)
Where/Select/Distinct/… extension methods onIEtlPipeline<T>that wrap the existing transformers viaThrough.EtlPipeline.Source.CsvExtractor<T>(...)source factories + sink terminators.Notes
vNext;Closes #147fires when vNext merges tomain.PublicAPI.Unshipped.txt(RS0017 enforced; RS0016 off — reconcile at release).