DtPipe.Core 1.8.2

DtPipe.Core

The core pipeline engine for DtPipe. Contains all abstractions, models, and pipeline logic with no external dependencies beyond Microsoft.Extensions.Logging.Abstractions.

Suitable for use as a standalone NuGet package in custom ETL pipelines.

Package

<PackageReference Include="DtPipe.Core" Version="1.0.0" />

What's Inside

DtPipe.Core/
├── Abstractions/       # Core interfaces (row and columnar)
├── Attributes/         # [ComponentOption] attribute
├── Dialects/           # ISqlDialect, SQL generation helpers
├── Helpers/            # Shared utility helpers
├── Infrastructure/     # Arrow type mapping, schema factory, row↔columnar bridges
│   └── Arrow/
├── Models/             # Shared data models (PipeColumnInfo, etc.)
├── Options/            # IOptionSet, IQueryAwareOptions, IKeyAwareOptions, OptionsRegistry
├── Pipelines/          # PipelineExecutionPlan, PipelineSegment, DAG orchestration
│   └── Dag/
├── Security/           # SQL query validator
└── Validation/         # Schema and constraint validators

Key Abstractions

Key Options Interfaces

Implementing a Custom Reader

// 1. Define options
public record MyReaderOptions : IProviderOptions, IQueryAwareOptions
{
    public static string Prefix => "mydb";
    public static string DisplayName => "MyDB Reader";
    public string? Query { get; set; }
}

// 2. Implement IStreamReader
public class MyStreamReader : IStreamReader
{
    public IReadOnlyList<PipeColumnInfo>? Columns { get; private set; }
    public Task OpenAsync(CancellationToken ct) { ... }
    public IAsyncEnumerable<ReadOnlyMemory<object?[]>> ReadBatchesAsync(int batchSize, CancellationToken ct) { ... }
    public ValueTask DisposeAsync() { ... }
}

// 3. Implement IProviderDescriptor<IStreamReader>
public class MyReaderDescriptor : IProviderDescriptor<IStreamReader>
{
    public string ProviderName => "mydb";
    public Type OptionsType => typeof(MyReaderOptions);
    public bool RequiresQuery => true;
    public bool CanHandle(string cs) => cs.StartsWith("mydb:");
    public IStreamReader Create(string cs, object options, IServiceProvider sp)
        => new MyStreamReader(cs, ((MyReaderOptions)options).Query!);
}

Running the Pipeline Directly

Headless execution goes through DagOrchestrator — even a single-branch (linear) run goes through it once, giving uniform cancellation and channel wiring:

var orchestrator = new DagOrchestrator(logger, channelRegistry, readerFactories);
var job = new JobDefinition { Input = "csv:in.csv", Output = "parquet:out.parquet" };
var dag = new JobDagDefinition
{
    // BranchDefinition.FromJob is the one projection from a job to a branch; building the
    // record by hand is how four call sites drifted apart.
    Branches = new[] { BranchDefinition.FromJob("main", job) }
};

int exitCode = await orchestrator.ExecuteAsync(dag, (branch, ctx, ct) =>
{
    // Open the branch's reader, run transformers, write to its writer.
    return RunBranchAsync(branch, ctx, ct);
}, cancellationToken);

License

MIT