Data Import Manager

Overview

DataImportManager (namespace TheTechIdea.Beep.Editor.Importing) is a partial-class orchestrator that moves data between any two IDataSource instances. It supports the full ETL lifecycle: schema compatibility checks, field-level quality rules, multi-phase transformation pipelines, intelligent batch processing, incremental sync, schema drift detection, staging tables, error quarantine, dead-letter replay, and run-history persistence.

DataImportManager (main + partial classes)
├── DataImportManager.Core.cs        ← data-source init, entity discovery
├── DataImportManager.Migration.cs   ← preflight validation, sync-draft
└── DataImportManager.Replay.cs      ← dead-letter replay

Helpers (injected / created internally)
├── IDataImportValidationHelper      ← config and compatibility checks
├── IDataImportTransformationHelper  ← field filter · mapping · defaults · custom fn
├── IDataImportBatchHelper           ← optimal sizing, batch splitting, retry
└── IDataImportProgressHelper        ← logging, progress, metrics

Supporting subsystems
├── Quality/          IDataQualityRule, BuiltInRules (NotNull, Unique, Range, Regex, …)
├── ErrorStore/       IImportErrorStore + ImportErrorRecord
├── History/          IImportRunHistoryStore + ImportRunRecord
├── Staging/          StagingOptions
├── Schema/           SchemaComparator, SchemaSnapshot
└── Profiling/        DataProfiler, DataProfile, FieldProfile

Quick Start

Minimal import (backward-compatible)

using var mgr = new DataImportManager(dmeEditor);

mgr.SourceDataSourceName = "ExternalCRM";
mgr.SourceEntityName     = "Contacts";
mgr.DestDataSourceName   = "MainDB";
mgr.DestEntityName       = "Customers";

var progress = new Progress<IPassedArgs>(a => Console.WriteLine(a.Messege));
using var cts = new CancellationTokenSource();

var result = await mgr.RunImportAsync(progress, cts.Token, null, batchSize: 100);
Console.WriteLine(result.Flag); // Errors.Ok | Errors.Failed

ImportContext entry point (recommended)

var ctx = new ImportContext
{
    Selection = new ImportSelectionContext
    {
        SourceDataSourceName      = "ExternalCRM",
        SourceEntityName          = "Contacts",
        DestinationDataSourceName = "MainDB",
        DestinationEntityName     = "Customers"
    },
    Options = new ImportExecutionOptions
    {
        BatchSize             = 200,
        RunMigrationPreflight = true,
        OnBatchError          = BatchErrorStrategy.Skip
    }
};

using var mgr = new DataImportManager(dmeEditor);
var result = await mgr.RunImportAsync(ctx, progress, CancellationToken.None);

ImportContext

ImportContext is the canonical, serialisable entry-point for all import runs. It replaces the older string-key parameter dictionary.

ImportContext

ImportSelectionContext

ImportExecutionOptions

BatchErrorStrategy enum

DataImportConfiguration

DataImportConfiguration is the expanded configuration object used when callers need fine-grained control. It is accepted by RunImportAsync(config, …) and can be built from an ImportContext via BuildConfigurationFromContext(ctx).

Core identifiers

Data shaping

Defaults & batching

Advanced pipeline phases

Run API

Return value

All async methods return Task<IErrorsInfo>:

Lifecycle Control

Imports run asynchronously. The manager exposes pause / resume / cancel controls that are safe to call from any thread.

ImportState enum

ImportStatus

GetImportStatus() returns a snapshot — mutations do not affect the live state.

ImportPerformanceMetrics

Migration Preflight

Preflight validation checks schema compatibility without writing any records. Enable it via ImportExecutionOptions.RunMigrationPreflight = true or call it directly.

// Direct call (independent of a full RunImportAsync)
var result = await mgr.RunMigrationPreflightAsync(config, log: msg => Console.WriteLine(msg));
if (result.Flag == Errors.Failed)
{
    Console.WriteLine($"Preflight failed: {result.Message}");
    return;
}

// Build sync-draft schema (no records imported)
var draft = await mgr.BuildSyncDraftAsync(config);

Preflight checks performed

1. Source datasource resolves and is connected.
2. Destination datasource resolves and is connected.
3. Source entity structure can be read.
4. Destination entity exists (or will be created if CreateDestinationIfNotExists=true).
5. When AddMissingColumns=false: all source fields must already exist on the destination.

Data Quality Rules

Add IDataQualityRule instances to DataImportConfiguration.QualityRules. Each rule is evaluated per-record before writing. Failures trigger the rule's OnFailure action.

DataQualityAction enum

Built-in rules

config.QualityRules.Add(new NotNullRule("CustomerID", DataQualityAction.Block));
config.QualityRules.Add(new UniqueRule("CustomerID", DataQualityAction.Quarantine));
config.QualityRules.Add(new RangeRule("Age", min: 0, max: 150, DataQualityAction.Block));
config.QualityRules.Add(new RegexRule("Email", @"^[^@]+@[^@]+\.[^@]+$", DataQualityAction.Warn));

// Custom rule
public class PositivePriceRule : IDataQualityRule
{
    public string FieldName  => "Price";
    public string RuleName   => "positive_price";
    public DataQualityAction OnFailure => DataQualityAction.Block;

    public bool Evaluate(object? value, object record) =>
        value != null && decimal.TryParse(value.ToString(), out var d) && d > 0;

    public string FailureMessage(object? value) =>
        $"Price must be positive (got '{value}').";
}

config.QualityRules.Add(new PositivePriceRule());

Schema Drift

The SchemaDriftPolicy controls what happens when the source entity's schema changes between runs. SchemaComparator (static utility) compares two SchemaSnapshot instances and produces a SchemaDriftReport.

SchemaDriftPolicy enum

config.DriftPolicy = SchemaDriftPolicy.AutoAddColumns;

// Manual schema comparison
var baseline = SchemaSnapshot.CaptureFrom(config.SourceData, config.SourceEntityName);
var current  = SchemaSnapshot.CaptureFrom(config.SourceData, config.SourceEntityName);
var report   = SchemaComparator.Compare(baseline, current);

Console.WriteLine($"Added:   {report.AddedFields.Count}");
Console.WriteLine($"Removed: {report.RemovedFields.Count}");
Console.WriteLine($"Altered: {report.AlteredFields.Count}");

Staging

When StagingOptions.Enabled = true, raw records are written to a staging entity (e.g. Customers_raw) first. The pipeline then normalizes and moves them to the final destination. Useful for auditing, validation gates, and multi-step transformations.

config.Staging = new StagingOptions
{
    Enabled                   = true,
    StagingEntitySuffix       = "_staging",
    DropStagingAfterNormalize = false, // keep for audit
    SkipNormalization         = false
};

Error Store & Dead-Letter Replay

Assign an IImportErrorStore to config.ErrorStore to capture quarantined records. Two implementations ship out of the box:

• DataSourceImportErrorStore — writes to an IDataSource entity
• JsonFileImportErrorStore — writes to a JSON file on disk

IImportErrorStore API

ImportErrorRecord

Dead-Letter Replay

// Attach error store to config
config.ErrorStore = new JsonFileImportErrorStore("./errors/");

var result = await mgr.RunImportAsync(config, progress, cts.Token);

// Later — replay records that previously failed
var contextKey = "ExternalCRM/Contacts/MainDB/Customers";
var replayResult = await mgr.ReplayFailedRecordsAsync(contextKey, progress, CancellationToken.None);
Console.WriteLine($"Replay: {replayResult.Flag}");

Data Profiling

DataProfiler (static utility) samples an entity and computes per-field statistics. Results are returned as a DataProfile containing a list of FieldProfile entries. Useful for understanding source data before configuring quality rules.

var profile = await DataProfiler.ProfileAsync(
    editor:         dmeEditor,
    dataSourceName: "ExternalCRM",
    entityName:     "Contacts",
    sampleSize:     2000,
    token:          CancellationToken.None);

foreach (var field in profile.Fields)
{
    Console.WriteLine($"{field.Name}: nulls={field.NullCount}, " +
                      $"distinct={field.DistinctCount}, " +
                      $"min={field.Min}, max={field.Max}");
}

DataProfile properties

Run History

Assign an IImportRunHistoryStore to config.RunHistoryStore to persist run records. Two implementations ship: DataSourceImportRunHistoryStore and JsonFileImportRunHistoryStore.

IImportRunHistoryStore API

Incremental sync with watermarking

// Store and restore last watermark for incremental loads
var historyStore = new JsonFileImportRunHistoryStore("./history/");
config.RunHistoryStore = historyStore;

var last = await historyStore.GetLastSuccessfulRunAsync("CRM/Contacts/DB/Customers");
config.SyncMode          = SyncMode.Incremental;
config.WatermarkColumn   = "ModifiedDate";
config.LastWatermarkValue = last?.WatermarkValue; // null on first run = import all

var result = await mgr.RunImportAsync(config, progress, cts.Token);

Examples

Full pipeline with all features

using var mgr = new DataImportManager(dmeEditor);

var config = mgr.CreateImportConfiguration(
    "ProductFeed", "ExternalCatalog",
    "Products",   "MainDB");

// Field subset
config.SelectedFields = new List<string> { "SKU", "Name", "Price", "Stock" };

// Source filter — only changed in last 7 days
config.SourceFilters.Add(new AppFilter
{
    FieldName   = "ModifiedDate",
    Operator    = ">=",
    FilterValue = DateTime.Today.AddDays(-7).ToString("yyyy-MM-dd")
});

// Quality rules
config.QualityRules.Add(new NotNullRule("SKU",   DataQualityAction.Block));
config.QualityRules.Add(new NotNullRule("Name",  DataQualityAction.Quarantine));
config.QualityRules.Add(new RangeRule("Price", 0.01m, 999999m, DataQualityAction.Block));

// Error store — save quarantined records to JSON
config.ErrorStore = new JsonFileImportErrorStore("./errors/");

// Run history — enable incremental runs
config.RunHistoryStore = new JsonFileImportRunHistoryStore("./history/");
config.SyncMode        = SyncMode.Incremental;
config.WatermarkColumn = "ModifiedDate";

// Staging — write raw records first
config.Staging = new StagingOptions { Enabled = true, StagingEntitySuffix = "_raw" };

// Schema drift
config.DriftPolicy = SchemaDriftPolicy.AutoAddColumns;

// Business logic transformation
config.CustomTransformation = record =>
{
    if (record is Dictionary<string, object> d)
        d["ImportedAt"] = DateTime.UtcNow;
    return record;
};

// Execute
var progress = new Progress<IPassedArgs>(a => Console.WriteLine($"[{a.ParameterInt1}/{a.ParameterInt2}] {a.Messege}"));
var result   = await mgr.RunImportAsync(config, progress, CancellationToken.None);

// Check results
var status = mgr.GetImportStatus();
Console.WriteLine($"Result:      {result.Flag}");
Console.WriteLine($"Processed:   {status.RecordsProcessed}");
Console.WriteLine($"Quarantined: {status.RecordsQuarantined}");
Console.WriteLine($"Throughput:  {status.Metrics?.RecordsPerSecond:F1} rec/s");

Pause / Resume from UI

// Start import on background thread
_ = Task.Run(() => mgr.RunImportAsync(config, progress, cts.Token));

// Pause after 2 seconds (e.g., user clicks Pause button)
await Task.Delay(2000);
mgr.PauseImport();

var snap = mgr.GetImportStatus();
Console.WriteLine($"Paused at record {snap.RecordsProcessed}");

// Resume
await Task.Delay(3000);
mgr.ResumeImport();

Profile source data before import

var profile = await DataProfiler.ProfileAsync(dmeEditor, "ExternalCatalog", "ProductFeed", 5000);

// Log fields with >10% nulls — may need NotNullRule
foreach (var f in profile.Fields.Where(f => f.NullCount * 1.0 / profile.SampleSize > 0.1))
    Console.WriteLine($"High null rate: {f.Name} ({f.NullCount}/{profile.SampleSize})");

Full API Reference

IDataImportManager (complete interface)

Helper Interfaces (all 4)

IDataImportValidationHelper

IDataImportTransformationHelper

IDataImportBatchHelper

IDataImportProgressHelper

SchemaSnapshot & SchemaSnapshotStore

In Schema/: captures entity structure at a point in time for schema drift comparison.

// SchemaSnapshot — static capture method
public class SchemaSnapshot
{
    // Properties
    public string DataSourceName { get; set; }
    public string EntityName { get; set; }
    public DateTime CapturedAt { get; set; }
    public List<EntityField> Fields { get; set; }

    // Static factory
    public static SchemaSnapshot CaptureFrom(IDataSource ds, string entityName);
}

// SchemaComparator — static comparison utility
public static class SchemaComparator
{
    public static SchemaDriftReport Compare(SchemaSnapshot baseline, SchemaSnapshot current);
    // SchemaDriftReport: AddedFields, RemovedFields, AlteredFields lists
}

// SchemaSnapshotStore — persistence for schema snapshots
public class SchemaSnapshotStore
{
    public Task SaveSnapshotAsync(SchemaSnapshot snapshot);
    public Task<SchemaSnapshot?> LoadSnapshotAsync(string contextKey);
    public Task ClearAsync(string contextKey);
}

IWatermarkStore / WatermarkStore (incremental sync)

In Sync/: persists watermark values for incremental data imports.

public interface IWatermarkStore
{
    Task<object?> GetWatermarkAsync(string contextKey);
    Task SaveWatermarkAsync(string contextKey, object value);
    Task ClearAsync(string contextKey);
}

// Implementation
public class WatermarkStore : IWatermarkStore { /* backed by IDataSource or local store */ }

DataQualityEvaluator

In Quality/: orchestrates rule evaluation across a batch.

public class DataQualityEvaluator
{
    public DataQualityEvaluator(List<IDataQualityRule> rules, IImportErrorStore? errorStore = null);

    public QualityEvaluationResult Evaluate(object record);
    // Returns: Passed {bool}, Action {DataQualityAction}, Failures {List<RuleResult>}

    public class QualityEvaluationResult
    {
        public bool Passed { get; set; }
        public DataQualityAction Action { get; set; }
        public List<RuleResult> Failures { get; set; } = new();
    }

    public class RuleResult
    {
        public string RuleName { get; set; }
        public string FieldName { get; set; }
        public string Message { get; set; }
        public DataQualityAction Action { get; set; }
    }
}

LocalStoreFactory

In Factories/: creates local JSON-file backed error store, run history, and watermark stores.

public static class LocalStoreFactory
{
    public static IImportErrorStore CreateErrorStore(string directoryPath);
    public static IImportRunHistoryStore CreateRunHistoryStore(string directoryPath);
    public static IWatermarkStore CreateWatermarkStore(string directoryPath);
}

All Enums

Source File Map