DistributedDataSource distributes entities across multiple physical datasources (shards), each backed by an IProxyCluster. It implements IDataSource and IDistributedDataSource across 12 partial files.
Per-shard failover, circuit breaking, and load balancing are inherited from IProxyCluster.
public interface IPartitionFunction
{
string Name { get; }
int Resolve(PartitionInput input, int bucketCount);
}
public struct PartitionInput
{
public string EntityName { get; set; }
public object[] KeyValues { get; set; }
public string PartitionKeyField { get; set; }
public static PartitionInput ForEntity(string entity, params object[] keys);
}
// 4 built-in partition functions:
public class HashPartitionFunction : IPartitionFunction; // Default
public class RangePartitionFunction : IPartitionFunction; // Range-based
public class ListPartitionFunction : IPartitionFunction; // Explicit key → bucket
public class CompositePartitionFunction : IPartitionFunction; // Chains multiple functions
public static class PartitionFunctionFactory
{
public static IPartitionFunction Create(string type, object config = null);
// type: "hash", "range", "list", "composite"
}
public static class PartitionKeyCoercer
{
public static object Coerce(object value, Type targetType);
// Coerces partition key to correct type for hashing
}
public static class MurmurHash3Helper
{
public static int Hash(byte[] data, uint seed = 0);
public static int HashString(string value, uint seed = 0);
}
Catalog Subsystem
public interface IShardCatalog
{
IReadOnlyDictionary<string, Shard> Shards { get; }
Shard GetShard(string shardId);
bool TryGetShard(string shardId, out Shard shard);
IEnumerable<Shard> GetShardsForPartitionRange(int start, int end);
}
public class Shard
{
public string ShardId { get; set; }
public IProxyCluster Cluster { get; set; }
public ShardState State { get; set; } // Active, Draining, Offline
public int SlotStart { get; set; } // Hash slot range start
public int SlotEnd { get; set; } // Hash slot range end
public DateTime AddedAtUtc { get; set; }
public Dictionary<string, double> Metrics { get; set; }
}
public class ShardCatalog : IShardCatalog
{
public void AddShard(Shard shard);
public bool RemoveShard(string shardId);
public void UpdateShardState(string shardId, ShardState state);
}
public class ShardCatalog.Persistence.cs
{
public Task SaveAsync(string path);
public Task LoadAsync(string path);
}
public enum ShardState { Active, Draining, Offline }
Routing Subsystem
public interface IShardRouter
{
RoutingDecision RouteRead(string entityName, List<AppFilter> filters,
DistributionPlan plan);
RoutingDecision RouteWrite(string entityName, object entity,
DistributionPlan plan);
}
public class RoutingDecision
{
public List<string> TargetShardIds { get; set; }
public PlacementType PlacementType { get; set; }
public bool IsBroadcast { get; set; }
public string PrimaryShardId { get; set; }
public string PartitionKey { get; set; }
}
public class ShardRouter : IShardRouter { /* uses ShardCatalog + PlacementResolver */ }
// Key extraction (.KeyExtraction partial)
public class ShardRouter.KeyExtraction.cs // Extracts partition key from filters/entity
// Dual-write (.DualWrite partial)
public class ShardRouter.DualWrite.cs // Dual-write routing during resharding
public interface IShardRoutingHook
{
RoutingDecision OnRouteRead(RoutingDecision decision, string entityName);
RoutingDecision OnRouteWrite(RoutingDecision decision, string entityName, object entity);
}
public class NullShardRoutingHook : IShardRoutingHook; // Pass-through default
public class ShardRoutingException : Exception
{
public string EntityName { get; }
public PlacementType Placement { get; }
public ShardRoutingException(string message, string entityName, PlacementType placement);
}
Execution Subsystem
public interface IDistributedReadExecutor
{
Task<IEnumerable<object>> ExecuteSingleShardAsync(
string entityName, List<AppFilter> filters, string shardId, CancellationToken ct);
Task<IEnumerable<object>> ExecuteScatterAsync(
string entityName, List<AppFilter> filters, List<string> shardIds, CancellationToken ct);
Task<IEnumerable<object>> ExecuteReplicatedReadAsync(
string entityName, List<AppFilter> filters, List<string> shardIds, CancellationToken ct);
}
public class DistributedReadExecutor : IDistributedReadExecutor { /* 4 partials */ }
// .SingleShard — single-shard with retry/failover
// .Scatter — parallel scatter with degree-of-parallelism cap
// .Replicated — nearest-healthy-replica selection
public class DistributedWriteExecutor
{
public Task<IErrorsInfo> ExecuteRoutedAsync(
string entityName, object entity, string shardId, CancellationToken ct);
public Task<IErrorsInfo> ExecuteBroadcastAsync(
string entityName, object entity, List<string> shardIds, CancellationToken ct);
public Task<IErrorsInfo> ExecuteFanOutAsync(
string entityName, object entity, List<string> shardIds, CancellationToken ct);
}
public class BasicResultMerger : IResultMerger
{
public IEnumerable<object> Merge(IEnumerable<IEnumerable<object>> results);
public IEnumerable<T> MergeTyped<T>(IEnumerable<IEnumerable<T>> results);
}
Query Subsystem
public interface IQueryPlanner
{
QueryPlan Plan(string entityName, List<AppFilter> filters, DistributionPlan plan);
}
public class QueryPlan
{
public QueryIntent Intent { get; set; } // SingleShard, Scatter, BroadcastJoin
public List<string> TargetShardIds { get; set; }
public AggregateSpec Aggregate { get; set; }
public MergeSpec Merge { get; set; }
public bool IsSingleShard { get; }
public bool IsScatter { get; }
}
public enum QueryIntent { SingleShard, Scatter, BroadcastJoin }
public enum AggregateKind { Count, Sum, Avg, Min, Max }
public class AggregateSpec
{
public AggregateKind Kind { get; set; }
public string FieldName { get; set; }
}
public class AggregateAccumulator
{
public void Accumulate(object row);
public void Merge(AggregateAccumulator other);
public object FinalResult { get; }
}
public enum OrderDirection { Asc, Desc }
public class OrderBySpec
{
public string FieldName { get; set; }
public OrderDirection Direction { get; set; }
}
public class MergeSpec
{
public bool Distinct { get; set; }
public List<OrderBySpec> OrderBy { get; set; }
public int? Top { get; set; }
}
public enum MergeOperation { Union, Concat, Intersect }
public interface IQueryAwareResultMerger
{
IEnumerable<object> Merge(IEnumerable<IEnumerable<object>> shardResults,
MergeSpec spec, CancellationToken ct = default);
}
public class QueryAwareResultMerger : IQueryAwareResultMerger
{
// .Sorting — merge-sort from sorted shard results
// .Grouping — group-by aggregation across shards
}
public class QueryPlanner : IQueryPlanner { /* resolves Plan from filters + placement */ }
public class BroadcastJoinRewriter
{
public QueryPlan RewriteForBroadcastJoin(QueryPlan plan, DistributionPlan distPlan);
}
public class RowValueExtractor
{
public static object GetValue(object row, string fieldName);
}
Transactions Subsystem
public interface IDistributedTransactionCoordinator
{
Task<TransactionToken> BeginAsync(string entityName, List<string> shardIds,
TransactionStrategy strategy = TransactionStrategy.SingleShard);
Task PrepareAsync(TransactionToken token);
Task CommitAsync(TransactionToken token);
Task RollbackAsync(TransactionToken token);
}
public enum TransactionStrategy { SingleShard, TwoPhaseCommit, Saga }
public class DistributedTransactionCoordinator : IDistributedTransactionCoordinator
{
// .SingleShard — begin on one shard, no coordination needed
// .TwoPhaseCommit — 2PC: prepare all → commit all (or rollback)
// .Saga — compensating-saga steps on abort
}
public class TransactionToken : IDisposable
{
public string TokenId { get; }
public string EntityName { get; }
public DistributedTransactionStatus Status { get; }
public Task CommitAsync();
public Task RollbackAsync();
public void Dispose();
}
public class DistributedTransactionScope : IDisposable { /* convenience wrapper */ }
public enum DistributedTransactionStatus
{
None, Active, Preparing, Prepared, Committing, Committed,
RollingBack, RolledBack, InDoubt, TimedOut
}
public enum TransactionLogKind { Prepare, Commit, Rollback, Compensate }
public class TransactionLogEntry
{
public string TokenId { get; set; }
public TransactionLogKind Kind { get; set; }
public string ShardId { get; set; }
public DateTime TimestampUtc { get; set; }
public string EntityName { get; set; }
}
public interface IDistributedTransactionLog
{
Task AppendAsync(TransactionLogEntry entry);
Task<IReadOnlyList<TransactionLogEntry>> GetEntriesAsync(string tokenId);
Task ClearAsync(string tokenId);
}
public class InMemoryTransactionLog : IDistributedTransactionLog;
public class FileTransactionLog : IDistributedTransactionLog;
public class TransactionDecisionResolver
{
public DistributedTransactionStatus Resolve(
IReadOnlyList<TransactionLogEntry> log);
}
public class SagaStep
{
public string StepName { get; set; }
public Func<Task> ExecuteAsync { get; set; }
public Func<Task> CompensateAsync { get; set; }
public bool IsCritical { get; set; }
}
Placement Subsystem
public class EntityPlacementResolver
{
public PlacementResolution Resolve(string entityName, List<AppFilter> filters,
DistributionPlan plan);
public PlacementResolution ResolveForWrite(string entityName, object entity,
DistributionPlan plan);
}
public enum PlacementMatchKind { Exact, Wildcard, Default }
public class PlacementResolution
{
public PlacementMatchKind MatchKind { get; set; }
public PlacementType PlacementType { get; set; }
public List<string> ShardIds { get; set; }
public string PartitionKeyField { get; set; }
public bool IsSharded { get; }
public bool IsReplicated { get; }
}
public class EntityPlacementMap
{
public Dictionary<string, List<string>> EntityToShards { get; set; }
public EntityPlacementMap(DistributionPlan plan);
public bool TryGetShards(string entityName, out List<string> shardIds);
}