DtPipe.Core 1.2.3

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
└── PipelineEngine.cs   # Headless pipeline engine for library use

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

var engine = new PipelineEngine(logger);
long rowsWritten = await engine.RunAsync(
    reader,
    writer,
    pipeline: transformers,   // optional
    batchSize: 50_000,
    ct: cancellationToken);

License

MIT