FastIngest.Core 0.5.0

FastIngest

FastIngest is a high-throughput, constant-memory bulk ingestion pipeline for .NET (CSV, XLSX, and NDJSON/JSONL to PostgreSQL, SQL Server, MySQL, SQLite, MongoDB, Cosmos DB, and Elasticsearch). Designed for enterprise workloads processing millions of rows without memory spikes, FastIngest leverages zero-allocation streaming readers, concurrent producer-consumer bounded channels, fluent validation, and native database bulk protocols (such as PostgreSQL binary COPY, SQL Server SqlBulkCopy, and MongoDB unordered BulkWriteAsync).

Architecture Overview

FastIngest processes incoming tabular and line-delimited data using a decoupled producer-consumer pipeline that keeps memory usage constant ($O(1)$) regardless of file size:

Stream (CSV / XLSX / JSONL / NDJSON)
        │
        ▼
┌─────────────────────────────────┐
│ Producer: Streaming Reader      │  <-- Sylvan CSV or PipeReader JSON Lines (CPU)
└──────────────┬──────────────────┘
               │
               ▼
┌─────────────────────────────────┐
│ Producer: Expression / Utf8 JSON│  <-- Zero-reflection binders or Utf8JsonReader
└──────────────┬──────────────────┘
               │
               ▼
┌─────────────────────────────────┐
│ Producer: FluentValidation      │  <-- FailFast or CollectAndContinue
└──────────────┬──────────────────┘
               │
               ▼
┌─────────────────────────────────┐
│ System.Threading.Channels       │  <-- Bounded channel (Capacity: 2 batches)
│ (BoundedChannelFullMode.Wait)   │  <-- Backpressure: strict O(1) memory
└──────────────┬──────────────────┘
               │
               ▼
┌─────────────────────────────────┐
│ Consumer: Native Database Sink  │  <-- High-throughput batch streaming (I/O)
└─────────────────────────────────┘

Key Features

• Concurrent Producer-Consumer Pipelining: Decouples CPU parsing/validation from database I/O socket operations using bounded System.Threading.Channels with backpressure.
• Constant-Memory Streaming: Stream arbitrarily large files (gigabytes to tens of gigabytes) with strict $O(1)$ memory guarantees.
• Line-Delimited JSON (NDJSON / JSONL): High-speed streaming parser over System.IO.Pipelines.PipeReader and System.Text.Json.Utf8JsonReader with automatic format heuristics (.jsonl, .ndjson, or { content peeking).
• 7 Native Database Sinks: Direct bulk protocol integrations for PostgreSQL (COPY), SQL Server (SqlBulkCopy), MySQL (MySqlBulkCopy), SQLite (WAL), MongoDB (BulkWriteAsync), Azure Cosmos DB, and Elasticsearch.
• Validation Strategies:
  • FailFast: Immediately halts ingestion on the first invalid record.
  • CollectAndContinue: Collects invalid row details and exports an error report CSV while allowing valid records to proceed.
• Fluent Pipeline API: Composable, chainable pipeline configuration with channel capacity tuning (WithChannelCapacity), custom JSON serializer options (WithJsonOptions), and progress tracking.
• SemVer 2.0 Driven by Git Tags: Automated versioning via MinVer and seamless CI/CD publishing.

Quickstart

1. Installation

dotnet add package FastIngest.Core
dotnet add package FastIngest.PostgreSql

2. Define Record & Validator

using FluentValidation;

public record CustomerRecord(int Id, string Email, string FullName, decimal Balance);

public class CustomerValidator : AbstractValidator<CustomerRecord>
{
    public CustomerValidator()
    {
        RuleFor(x => x.Email).NotEmpty().EmailAddress();
        RuleFor(x => x.FullName).NotEmpty().MaximumLength(100);
        RuleFor(x => x.Balance).GreaterThanOrEqualTo(0);
    }
}

3. Run Pipeline with PostgreSQL COPY

using FastIngest.Core.Common;
using FastIngest.Core.Pipeline;
using FastIngest.PostgreSql.Extensions;
using Npgsql;

await using var stream = File.OpenRead("large_customers.csv");
await using var connection = new NpgsqlConnection("Host=localhost;Database=mydb;Username=postgres;Password=secret");
await connection.OpenAsync();

var result = await FastIngestPipeline<CustomerRecord>.Create()
    .FromStream(stream, FileType.Csv)
    .WithMapping(mapping =>
    {
        mapping.Map(x => x.Id, "customer_id");
        mapping.Map(x => x.Email, "email");
        mapping.Map(x => x.FullName, "full_name");
        mapping.Map(x => x.Balance, "balance");
    })
    .ValidateWith<CustomerValidator>(options =>
    {
        options.ErrorStrategy = ErrorStrategy.CollectAndContinue;
    })
    .WithBatchSize(5000)
    .WithChannelCapacity(2)
    .OnProgress(progress =>
    {
        Console.WriteLine($"Processed {progress.RowsProcessed} rows ({progress.PercentComplete:F1}%)...");
    })
    .WriteToPostgresAsync(connection, "customers", CancellationToken.None);

Console.WriteLine($"Ingestion Complete! Succeeded: {result.TotalSucceeded}, Failed: {result.TotalFailed}");

if (!result.IsSuccess)
{
    var errorCsv = result.ExportErrorsToCsv();
    await File.WriteAllBytesAsync("ingest_errors.csv", errorCsv);
}

4. Stream Line-Delimited JSON (NDJSON / JSONL)

FastIngest natively streams .jsonl / .ndjson files without loading the entire document into RAM:

await using var jsonlStream = File.OpenRead("customers.jsonl");

var jsonlResult = await FastIngestPipeline<CustomerRecord>.Create()
    .FromStream(jsonlStream, FileType.JsonLines) // Or FileType.Ndjson
    .WithJsonOptions(opt => opt.PropertyNameCaseInsensitive = true)
    .ValidateWith<CustomerValidator>(opt => opt.ErrorStrategy = ErrorStrategy.CollectAndContinue)
    .WithBatchSize(5000)
    .WriteToPostgresAsync(connection, "customers");

5. ASP.NET Core & SignalR Minimal API

Add the FastIngest.AspNetCore package to effortlessly mount 202 Accepted bulk upload endpoints that execute via a background queue and stream progress to connecting clients via SignalR:

using FastIngest.AspNetCore.Extensions;

var builder = WebApplication.CreateBuilder(args);

// Registers SignalR, Background queues, and FastIngest
builder.Services.AddFastIngestAspNetCore(ingest => {
    ingest.AddPostgreSqlSink("Host=localhost;Database=mydb;Username=postgres;Password=secret");
    ingest.RegisterProfilesFromAssemblyContaining<CustomerImportProfile>();
});

var app = builder.Build();

app.MapFastIngestHub("/hubs/fastingest"); // SignalR telemetry hub
app.MapFastIngestUpload<CustomerRecord>("/api/customers/import"); // 202 Accepted Upload

app.Run();

(Clients can pass ?jobId=123 to the upload endpoint and subscribe to ReceiveProgress on the Hub to render progress bars.)

Performance Benchmarks

Official BenchmarkDotNet suite results comparing FastIngest streaming binary COPY against Entity Framework Core 9 (net9.0, Apple M4, PostgreSQL 16 Alpine via Testcontainers):

Run the benchmarks yourself with ./benchmarks/run-benchmarks.sh. See full analysis in docs/benchmarks/performance.md.

LLM & AI Discoverability

FastIngest provides machine-readable documentation endpoints adhering to the llms.txt standard for AI coding assistants (such as Cursor, Windsurf, Claude Code, GitHub Copilot, and ChatGPT):

• Documentation Index: https://fastingest.aebibtech.com/llms.txt
• Full Consolidated Context: https://fastingest.aebibtech.com/llms-full.txt

Repository Structure

├── .github/workflows/ci.yml       # GitHub Actions CI/CD Pipeline
├── Directory.Build.props          # Centralized MSBuild & MinVer configuration
├── FastIngest.sln
├── benchmarks/
│   └── FastIngest.Benchmarks/     # BenchmarkDotNet performance suite
├── docs/                          # VitePress documentation website (https://fastingest.aebibtech.com)
├── src/
│   ├── FastIngest.Core/           # Core pipeline, channels, binders, CSV & NDJSON/JSONL streaming readers
│   ├── FastIngest.PostgreSql/     # PostgreSQL native binary COPY sink
│   ├── FastIngest.SqlServer/      # Microsoft SQL Server SqlBulkCopy sink
│   ├── FastIngest.MySql/          # MySQL MySqlBulkCopy sink
│   ├── FastIngest.Sqlite/         # SQLite WAL batch sink
│   ├── FastIngest.MongoDb/        # MongoDB unordered BulkWrite sink
│   ├── FastIngest.CosmosDb/       # Azure Cosmos DB bulk executor sink
│   ├── FastIngest.Elasticsearch/  # Elasticsearch BulkAsync sink
│   ├── FastIngest.AspNetCore/     # ASP.NET Core endpoints, background workers, and SignalR telemetry
│   └── FastIngest.Extensions.DependencyInjection/ # Engine, DI, and profile registry
├── samples/
│   └── FastIngest.SampleApi/      # Minimal Web API demonstrating ingestion
└── tests/
    └── FastIngest.Tests/          # Unit and integration test suites

License

This project is licensed under the MIT License - see the LICENSE file for details.

Developed & Maintained by Paul Abib Camano (Aebibtech).