Prakrishta.Data.Bulk 1.0.1

Prakrishta.Data.Bulk

High performance, extensible bulk operations for .NET. Prakrishta.Data.Bulk is a provider agnostic, pipeline based bulk engine designed for speed, flexibility, and testability. It complements Prakrishta.Data by enabling large scale insert, update, and delete operations with minimal overhead.

Features

• Fastest‑in‑class bulk insert performance
• TVP‑based stored procedure strategy (fastest for pure inserts)
• Staging table strategy with MERGE (best for upsert/update/delete)
• Zero reflection, zero EF Core overhead
• Linear scaling from 1k → 50k+ rows
• Async/await support
• BenchmarkDotNet‑verified performance
• Works with SQL Server, Azure SQL, LocalDB
• Pure ADO.NET — no EF Core required
• Clean, extensible architecture

Feature Comparison Table

Getting Started (Quickstart Guide)

1. Install the NuGet package

dotnet add package Prakrishta.Data.Bulk

2. Define your entity

public sealed class SalesRecord
{
    public int Id { get; set; }
    public DateTime SaleDate { get; set; }
    public decimal Amount { get; set; }
}

3. Create a bulk engine full example

var builder = WebApplication.CreateBuilder(args);

// Register Bulk Engine
builder.Services.AddBulkEngine(opts =>
{
    opts.DefaultStrategy = BulkStrategyKind.StoredProcedureTvp;
});

var app = builder.Build();

// Resolve engine
var bulk = app.Services.GetRequiredService<BulkEngine>();

// Sample data
var items = new List<SalesRecord>
{
    new() { Id = 1, SaleDate = DateTime.UtcNow, Amount = 100 },
    new() { Id = 2, SaleDate = DateTime.UtcNow, Amount = 200 }
};

// Insert
await bulk.InsertAsync(
    items,
    "dbo.Sales",
    "dbo.SalesType",
    "dbo.Sales_Insert");

// Partition Switch
await bulk.ReplacePartitionAsync(
    items,
    "dbo.FactSales",
    opts => opts
        .UseStagingTable("dbo.FactSales_Staging_7")
        .ForPartition(7));

app.Run();

4. Service Collection Extensions

public static class ServiceCollectionExtensions
{
    public static IServiceCollection AddBulkEngine(
        this IServiceCollection services,
        string connectionString,
        Action<BulkOptions>? configure = null)
    {
        var options = new BulkOptions();
        configure?.Invoke(options);

        services.AddSingleton(options);

        // Factories
        services.AddSingleton<IBulkCopyFactory, SqlBulkCopyFactory>();
        services.AddSingleton<IDbConnectionFactory, SqlConnectionFactory>();

        // Register schema resolver
        services.AddSingleton<ISchemaResolver>(sp =>
        {
            var factory = sp.GetRequiredService<IDbConnectionFactory>();
            return new SchemaResolver(factory, connectionString);
        });

        // Strategy selector
        services.AddSingleton<BulkStrategySelector>();

        // Strategies
        services.AddSingleton<IBulkStrategy>(sp =>
            new StoredProcedureTvpStrategy(
                sp.GetRequiredService<IDbConnectionFactory>()));

        services.AddSingleton<IBulkStrategy>(sp =>
            new StagingTableStrategy(
                sp.GetRequiredService<BulkOptions>(),
                sp.GetRequiredService<IBulkCopyFactory>(),
                sp.GetRequiredService<IDbConnectionFactory>()));

        services.AddSingleton<IBulkStrategy>(sp =>
            new TruncateAndReloadStrategy(
                sp.GetRequiredService<BulkOptions>(),
                sp.GetRequiredService<IBulkCopyFactory>(),
                sp.GetRequiredService<IDbConnectionFactory>()));

        services.AddSingleton<IBulkStrategy>(sp =>
            new PartitionSwitchStrategy(
                sp.GetRequiredService<IBulkCopyFactory>(),
                sp.GetRequiredService<IDbConnectionFactory>()));

        // Strategy dictionary
        services.AddSingleton<IDictionary<BulkStrategyKind, IBulkStrategy>>(sp =>
        {
            var strategies = sp.GetServices<IBulkStrategy>();
            return strategies.ToDictionary(s => s.Kind, s => s);
        });

        // Pipeline
        services.AddSingleton<IBulkPipeline, BulkPipelineEngine>();

        // Register BulkEngine
        services.AddSingleton<BulkEngine>(sp =>
        {
            var factory = sp.GetRequiredService<IDbConnectionFactory>();
            var pipeline = sp.GetRequiredService<IBulkPipeline>();
            return new BulkEngine(connectionString, factory, pipeline);
        });


        return services;
    }
}

5. TVP Type for SalesRecord

CREATE TYPE dbo.SalesType AS TABLE
(
    Id          INT            NOT NULL,
    SaleDate    DATETIME2(7)   NOT NULL,
    Amount      DECIMAL(18,2)  NOT NULL
);

✔ Must match your C# entity ✔ Must match your staging table ✔ Must NOT include identity or constraints

6. Auto‑Drop + Recreate Script

IF OBJECT_ID('dbo.SalesRecord_Staging', 'U') IS NOT NULL
    DROP TABLE dbo.SalesRecord_Staging;

CREATE TABLE dbo.SalesRecord_Staging
(
    Id          INT            NOT NULL,
    SaleDate    DATETIME2(7)   NOT NULL,
    Amount      DECIMAL(18,2)  NOT NULL
);

CREATE CLUSTERED INDEX IX_SalesRecord_Staging_Id
    ON dbo.SalesRecord_Staging (Id);

Performance Benchmarks

Key Findings

• EFCore.BulkExtensions performs well for small batches due to low setup overhead.
• Prakrishta’s stored-proc strategy is the fastest overall, especially at large batch sizes.
• Prakrishta’s staging-table strategy is also extremely fast and scales linearly.
• At 10k–50k rows, Prakrishta is 2× faster than EFCore.BulkExtensions.
• Staging-table strategy even outperforms raw SqlBulkCopy at large sizes.
• Performance is linear, predictable, and optimized for high-volume ingestion.

Performance Chart (Markdown)

This chart visualizes the current benchmark results for 50,000 rows, the most meaningful scale for real‑world ETL and ingestion workloads.

Milliseconds (lower is better)

Bulk Insert Performance (50,000 rows)

Prakrishta (Stored Proc)   | ████████████████████████████ 188 ms
Prakrishta (Staging)       | ██████████████████████████████ 195 ms
Raw SqlBulkCopy            | ███████████████████████████████ 203 ms
EFCore.BulkExtensions      | █████████████████████████████████████████████ 395 ms

Bulk Insert Performance (10,000 rows)

Prakrishta (Stored Proc)   | ████████████████ 37.3 ms
Prakrishta (Staging)       | ████████████████████ 48.2 ms
Raw SqlBulkCopy            | ████████████████████ 49.2 ms
EFCore.BulkExtensions      | █████████████████████████████████ 87.4 ms

Bulk Insert Performance (1,000 rows)

Prakrishta (Stored Proc)   | ████████ 10.0 ms
EFCore.BulkExtensions      | ████████ 11.4 ms
Prakrishta (Staging)       | █████████ 12.8 ms
Raw SqlBulkCopy            | ██████████ 14.6 ms

Why Prakrishta Is Faster

Prakrishta.Data.Bulk achieves industry‑grade performance because it:

• Uses pure ADO.NET
• Avoids EF Core overhead
• Eliminates reflection
• Uses optimized TVP ingestion
• Uses linear‑scaling staging tables
• Minimizes SQL Server I/O and logging
• Reduces round trips
• Produces predictable, stable performance curves

This is why Prakrishta Data Bulk engine is:

• Faster than EFCore.BulkExtensions
• Faster than Raw SqlBulkCopy at scale
• The fastest overall at 50K rows (stored‑proc strategy)

Choosing the Right Strategy

Different workloads benefit from different bulk‑loading strategies. Prakrishta.Data.Bulk gives you three optimized paths — each designed for a specific class of problems.

1. Stored Procedure Strategy (TVP‑based) — Best Overall for Inserts

Use when you want:

• Maximum raw insert speed
• Minimal SQL Server overhead
• A single round‑trip to the database
• No MERGE logic
• No staging table

Ideal for:

• High‑volume inserts
• ETL ingestion
• Logging pipelines
• Append‑only tables
• Scenarios where the target table has no complex constraints

Why choose it:

Fastest strategy at 1k, 10k, and 50k rows. Outperforms EFCore.BulkExtensions and even Raw SqlBulkCopy.

2. Staging Table Strategy — Best for Upserts, Updates & Deletes

Use when you need:

• MERGE semantics
• Update‑or‑insert behavior
• Delete‑or‑insert behavior
• Full control over matching keys
• Idempotent ingestion

Ideal for:

• Slowly changing dimensions (SCD)
• Data warehouse loads
• Sync jobs
• Reconciliation pipelines
• Any scenario requiring deterministic upsert logic

Why choose it:

Linear scaling, extremely stable, and 2× faster than EFCore.BulkExtensions at medium and large batch sizes.

3. Raw SqlBulkCopy Strategy — Baseline / Custom Scenarios

Use when you want:

• Absolute minimal overhead
• Full control over the SqlBulkCopy pipeline
• Custom batching or streaming logic
• No MERGE or stored proc logic

Ideal for:

• Internal pipelines
• Custom ETL frameworks
• Scenarios where you want to build your own logic on top of SqlBulkCopy

Why choose it:

Great baseline — and your strategies outperform it at scale.

4. When to Choose Which Strategy

Performance Badges

Attribute‑Based Configuration

The Bulk Engine supports strongly‑typed attributes that allow you to configure schema, table names, TVP names, stored procedures, and column mappings directly on your entity classes. This provides a clean, declarative alternative to fluent configuration and integrates seamlessly with automatic schema discovery. Attributes are optional — the engine continues to work with conventions and fluent overrides.

Why Use Attributes?

Attributes allow you to:

• Keep configuration close to your entity model
• Avoid repeating table/TVP/procedure names in multiple places
• Override conventions without using fluent API
• Disable automatic schema discovery when schema is explicitly defined
• Customize column names or ignore properties
• Mark key columns explicitly

They also follow a clear precedence model:

Precedence Order (Highest → Lowest)

• Fluent API overrides
• Attributes
• Automatic schema discovery
• Conventions (dbo.TableName, property name)

Property‑Level Attributes

Column Rename

[BulkColumn("CustomerName")]
public string Name { get; set; }

Maps the property to a different column name.

Ignore Property

[BulkIgnore]
public string TempValue { get; set; }

Ignored during:

• Insert
• Update
• Delete
• TVP generation
• Partition switch staging

Explicit Key

[BulkKey]
public Guid CustomerId { get; set; }

Overrides the default "Id" convention.

How Attributes Interact with Fluent API

Attributes provide defaults, but fluent API always wins:

[BulkSchema("sales")]
[BulkTable("Customer")]
public class Customer { ... }

await bulk
    .For<Customer>()
    .ToTable("custom.Customers")   // overrides attribute
    .InsertAsync(items);

Final table name: custom.Customers

How Attributes Interact with Schema Discovery

If schema is not provided via:

• .InSchema("...")
• [BulkSchema("...")]
• .ToTable("schema.Table")

Then the engine automatically discovers the schema from the database:

SELECT TABLE_SCHEMA
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_NAME = 'Customer'

If multiple schemas contain the same table, the engine throws a clear error and instructs the user to specify a schema explicitly.

Example Entity Using All Attributes

[BulkSchema("sales")]
[BulkTable("Customer")]
[BulkTvp("CustomerType")]
[BulkInsertProcedure("Customer_Insert")]
[BulkUpdateProcedure("Customer_Update")]
[BulkDeleteProcedure("Customer_Delete")]
public class Customer
{
    [BulkKey]
    public int CustomerId { get; set; }

    [BulkColumn("CustomerName")]
    public string Name { get; set; }

    [BulkIgnore]
    public string TempValue { get; set; }
}

Usage:

await bulk.For<Customer>().InsertAsync(customers);

Everything resolves automatically:

• Table → sales.Customer
• TVP → sales.CustomerType
• Insert SP → sales.Customer_Insert
• Column map → { CustomerId, CustomerName }

License

MIT License — free for commercial and open‑source use.