FractalDataWorks.Data.DataContainers.Abstractions 0.4.0-preview.6

FractalDataWorks.Data.DataContainers.Abstractions

Data container abstractions for reading, writing, and manipulating structured data across different storage formats. Defines contracts for format-agnostic data access with schema support and streaming capabilities.

Overview

Data container abstractions that extend Data.Abstractions with contracts for:

• Format-Agnostic Containers: Read/write data regardless of underlying format (CSV, JSON, SQL, Parquet)
• Schema-Aware Operations: IDataSchema integration for type validation and field lookups
• Streaming Data Access: IDataReader and IDataWriter for efficient record-by-record processing
• In-Memory Datasets: IRuntimeDataSet for LINQ-like query operations
• Write Mode Control: TypeCollection-based write modes (Append, Overwrite, CreateNew, Upsert)

Target Frameworks: .NET Standard 2.0, .NET 10.0

Dependencies: FractalDataWorks.Data.Abstractions, FractalDataWorks.Results, FractalDataWorks.Collections

Key Interfaces

IDataContainer

From IDataContainer.cs:24-143:

public interface IDataContainer
{
    string Id { get; }
    string Name { get; }
    string ContainerType { get; }
    IDataSchema Schema { get; }
    IReadOnlyDictionary<string, object> Metadata { get; }
    DataLocation Location { get; }
    IContainerConfiguration Configuration { get; }

    Task<IGenericResult> ValidateReadAccess(DataLocation location);
    Task<IGenericResult> ValidateWriteAccess(DataLocation location);
    Task<IGenericResult<ContainerMetrics>> GetReadMetrics(DataLocation location);
    Task<IGenericResult<IDataReader>> CreateReader(DataLocation location);
    Task<IGenericResult<IDataWriter>> CreateWriter(DataLocation location, IContainerWriteMode? writeMode = null);
    Task<IGenericResult<IDataSchema>> DiscoverSchema(DataLocation location, int sampleSize = 1000);
}

IDataReader

From IDataReader.cs:20-167:

public interface IDataReader : IDisposable
{
    IDataSchema Schema { get; }
    long CurrentPosition { get; }
    IReadOnlyDictionary<string, object> SourceMetadata { get; }
    bool HasMoreRecords { get; }
    bool SupportsSeek { get; }

    Task<IGenericResult<IReadOnlyDictionary<string, object>?>> ReadRecord(CancellationToken cancellationToken = default);
    Task<IGenericResult<T?>> ReadRecord<T>(CancellationToken cancellationToken = default) where T : class;
    Task<IGenericResult<IEnumerable<IReadOnlyDictionary<string, object>>>> ReadRecords(int maxRecords, CancellationToken cancellationToken = default);
    Task<IGenericResult<IEnumerable<T>>> ReadRecords<T>(int maxRecords, CancellationToken cancellationToken = default) where T : class;
    Task<IGenericResult<IEnumerable<T>>> ReadAllRecords<T>(CancellationToken cancellationToken = default) where T : class;
    Task<IGenericResult> Seek(long position, CancellationToken cancellationToken = default);
    Task<IGenericResult> Reset(CancellationToken cancellationToken = default);
    IReaderStatistics GetStatistics();
}

IDataWriter

From IDataWriter.cs:21-164:

public interface IDataWriter : IDisposable
{
    IDataSchema Schema { get; }
    long CurrentPosition { get; }
    IReadOnlyDictionary<string, object> DestinationMetadata { get; }
    IContainerWriteMode WriteMode { get; }
    bool SupportsBatch { get; }
    bool SupportsTransactions { get; }

    Task<IGenericResult> WriteRecord(IReadOnlyDictionary<string, object> record, CancellationToken cancellationToken = default);
    Task<IGenericResult> WriteRecord<T>(T record, CancellationToken cancellationToken = default) where T : class;
    Task<IGenericResult<long>> WriteRecords(IEnumerable<IReadOnlyDictionary<string, object>> records, CancellationToken cancellationToken = default);
    Task<IGenericResult<long>> WriteRecords<T>(IEnumerable<T> records, CancellationToken cancellationToken = default) where T : class;
    Task<IGenericResult> Flush(CancellationToken cancellationToken = default);
    Task<IGenericResult<IWriterStatistics>> Finalize(CancellationToken cancellationToken = default);
    Task<IGenericResult<IWriteTransaction>> BeginTransaction(CancellationToken cancellationToken = default);
    IWriterStatistics GetStatistics();
}

IDataRow

From IDataRow.cs:11-72:

public interface IDataRow
{
    T GetValue<T>(string fieldName);
    T GetValue<T>(int ordinal);
    bool TryGetValue<T>(string fieldName, out T value);
    bool TryGetValue<T>(int ordinal, out T value);
    object? GetValue(string fieldName);
    object? GetValue(int ordinal);
    bool HasField(string fieldName);
    IReadOnlyList<string> FieldNames { get; }
    int FieldCount { get; }
    IReadOnlyDictionary<string, object?> AsDictionary();
}

IRuntimeDataSet

From IRuntimeDataSet.cs:11-127:

public interface IRuntimeDataSet
{
    string Name { get; }
    IDataSchema Schema { get; }
    bool IsSuccess { get; }
    string? ErrorMessage { get; }
    int RowCount { get; }
    IEnumerable<IDataRow> Rows { get; }

    IDataRow GetRow(int index);
    IRuntimeDataSet Where(Func<IDataRow, bool> predicate);
    IRuntimeDataSet Select(Func<IDataRow, IDataRow> selector);
    decimal Sum(string fieldName);
    decimal Average(string fieldName);
    decimal Min(string fieldName);
    decimal Max(string fieldName);
    int Count();
    int Count(Func<IDataRow, bool> predicate);
    IEnumerable<IGrouping<object?, IDataRow>> GroupBy(string fieldName);
    IRuntimeDataSet OrderBy(string fieldName, bool descending = false);
    IRuntimeDataSet Take(int count);
    IRuntimeDataSet Skip(int count);
    IDataRow? FirstOrDefault();
    IDataRow? FirstOrDefault(Func<IDataRow, bool> predicate);
    IReadOnlyList<IDataRow> ToList();
    IDataRow[] ToArray();
}

Container Write Modes

From ContainerWriteModes/IContainerWriteMode.cs:9-25:

public interface IContainerWriteMode : ITypeOption<int, ContainerWriteModeBase>
{
    bool PreservesExistingData { get; }
    bool RequiresExistingContainer { get; }
    bool FailsIfExists { get; }
}

Available write modes via TypeCollection lookup:

• ContainerWriteModes.Append - Add to existing data
• ContainerWriteModes.Overwrite - Replace existing data
• ContainerWriteModes.CreateNew - Fail if container exists
• ContainerWriteModes.Upsert - Insert or update based on key

Usage Patterns

Creating DataRows

From IDataRow.cs:77-302:

// Create row from schema and values
var schema = DataSchema.FromFields([
    new SchemaField("CustomerId", typeof(int), 0),
    new SchemaField("Name", typeof(string), 1)
]);
var row = new DataRow(schema, [1, "Acme Corp"]);

// Access values by name (dictionary lookup)
var name = row.GetValue<string>("Name");

// Access values by ordinal (direct array access)
var id = row.GetValue<int>(0);

// Safe access with TryGetValue
if (row.TryGetValue<string>("Email", out var email))
{
    Console.WriteLine(email);
}

// Create single-field row
var singleRow = DataRow.SingleField("Count", 42);

// Create from dictionary
var dict = new Dictionary<string, object?> { ["CustomerId"] = 1, ["Name"] = "Acme" };
var rowFromDict = DataRow.FromDictionary(schema, dict);

Working with RuntimeDataSet

From RuntimeDataSet.cs:13-217:

// Create successful dataset
var schema = DataSchema.FromFields([
    new SchemaField("Id", typeof(int), 0),
    new SchemaField("Amount", typeof(decimal), 1)
]);
var rows = new[] { new DataRow(schema, [1, 100.50m]), new DataRow(schema, [2, 250.75m]) };
var dataSet = RuntimeDataSet.FromRows("Orders", schema, rows);

// Query operations (LINQ-like)
var filtered = dataSet.Where(row => row.GetValue<decimal>("Amount") > 100m);
var sorted = dataSet.OrderBy("Amount", descending: true);
var top5 = dataSet.Take(5);

// Aggregations
var total = dataSet.Sum("Amount");
var average = dataSet.Average("Amount");
var maxAmount = dataSet.Max("Amount");

// Create failure dataset
var failed = RuntimeDataSet.Failure("Orders", "Connection timeout");
if (!failed.IsSuccess)
{
    Console.WriteLine(failed.ErrorMessage);
}

Schema and Field Operations

From DataSchema.cs:12-151 and DataSchemaExtensions.cs:11-60:

// Create schema
var schema = new DataSchema(
    id: "users-v1",
    name: "Users",
    version: "1.0",
    fields: new ISchemaField[]
    {
        new SchemaField("Id", typeof(int), 0) { IsRequired = true, IsPrimaryKey = true },
        new SchemaField("Email", typeof(string), 1) { IsRequired = true, MaxLength = 255 },
        new SchemaField("CreatedAt", typeof(DateTime), 2) { IsRequired = true }
    }
);

// Field operations
var fieldNames = schema.FieldNames();
var emailField = schema.GetField("Email");
var ordinal = schema.GetOrdinal("Email");
var hasField = schema.HasField("Phone");

// Schema operations
var projected = schema.ProjectTo(new[] { "Id", "Email" });
var extended = schema.ExtendWith(new[] { new SchemaField("Phone", typeof(string), 3) });

Best Practices

1. Use IDataReader/IDataWriter for Streaming: For large datasets, use streaming interfaces rather than loading all data into memory
2. Check IsSuccess on RuntimeDataSet: Always verify dataset validity before accessing rows
3. Prefer Ordinal Access: Use GetValue<T>(int ordinal) for performance-critical loops
4. Dispose Readers/Writers: Both implement IDisposable; use using statements
5. Validate Schemas: Use ValidateReadAccess/ValidateWriteAccess before operations
6. Use TypeCollection for Write Modes: Access write modes via ContainerWriteModes.ByName() or static properties

Dependencies

• FractalDataWorks.Data.Abstractions: IDataSchema, DataLocation, IContainerConfiguration
• FractalDataWorks.Results: IGenericResult for operation outcomes
• FractalDataWorks.Collections: TypeCollection infrastructure for write modes

Related Packages

• FractalDataWorks.Data.Files: File-based container implementations (CSV, JSON)
• FractalDataWorks.Data.MsSql: SQL Server container implementations
• FractalDataWorks.Data.Http: HTTP/REST container implementations