Sharding with Raven

Sharding strategy credits

Full example

Setting up the client for sharding

var shards = new Shards
{
    new DocumentStore
    {
        Identifier = "Users",
        Url = "http://localhost:8081",
        Conventions =
            {
                GenerateDocumentKey = user => "users/" + ((User) user).Name
            }
    },
    new DocumentStore {Identifier = "Blogs", Url = "http://localhost:8082"},
    new DocumentStore {Identifier = "Posts #1", Url = "http://localhost:8083"},
    new DocumentStore {Identifier = "Posts #2", Url = "http://localhost:8084"},
    new DocumentStore {Identifier = "Posts #3", Url = "http://localhost:8085"}
};

var shardStrategy = new ShardStrategy
{
    ShardAccessStrategy = new ParallelShardAccessStrategy(),
    ShardSelectionStrategy = new BlogShardSelectionStrategy(3),
    ShardResolutionStrategy = new BlogShardResolutionStrategy(3),
};

var documentStore = new ShardedDocumentStore(shardStrategy, shards);
documentStore.Initialize();

Using sharding

using (var session = documentStore.OpenSession())
{
    var user = new User { Name = "ayende" };
    var blog = new Blog { Name = "Ayende @ Rahien" };


    session.Store(user);
    session.Store(blog);


    // we have to save to Raven to get the generated id for the blog instance
    session.SaveChanges();
    var posts = new List<Post>();
    for (var i = 0; i < 6; i++)
    {
        var post = new Post
        {
            BlogId = blog.Id,
            UserId = user.Id,
            Content = "Just a post",
            Title = "Post #" + (i + 1)
        };
        posts.Add(post);
        session.Store(post);
    }


    session.SaveChanges();
}

using (var session = documentStore.OpenSession())
{
    session.Query<User>().ToArray();
    session.Query<Blog>().ToArray();
    session.Query<Post>().ToArray();
}

using (var session = documentStore.OpenSession())
{
    session.Load<User>("users/ayende");
    session.Load<Blog>("blogs/1");
    session.Load<Post>("posts/1/2");
    session.Load<Post>("posts/2/2");
}

The sharding strategy

var shardStrategy = new ShardStrategy
{
    ShardAccessStrategy = new ParallelShardAccessStrategy(),
    ShardSelectionStrategy = new BlogShardSelectionStrategy(3),
    ShardResolutionStrategy = new BlogShardResolutionStrategy(3),
};

Shard Selection Strategy

public class BlogShardSelectionStrategy : IShardSelectionStrategy
{
    private readonly int numberOfShardsForPosts;
    private int currentNewShardId;


    public BlogShardSelectionStrategy(int numberOfShardsForPosts)
    {
        this.numberOfShardsForPosts = numberOfShardsForPosts;
    }


    public string ShardIdForNewObject(object obj)
    {
        var shardId = GetShardIdFromObjectType(obj);
        if (obj is Post)
        {
            var nextPostShardId = Interlocked.Increment(ref currentNewShardId) % numberOfShardsForPosts;
            nextPostShardId += 1;// to ensure base 1
            ((Post)obj).Id = "posts/" + nextPostShardId + "/"; // encode the shard id in the in the prefix.
            shardId += nextPostShardId;
        }
        return shardId;
    }


    public string ShardIdForExistingObject(object obj)
    {
        var shardId = GetShardIdFromObjectType(obj);
        if (obj is Post)
        {
            // the format of a post id is: 'posts' / 'shard id' / 'post id'
            var id = ((Post)obj).Id.Split(new[] { '/' }, StringSplitOptions.RemoveEmptyEntries);
            shardId += id[1];// add shard id
        }
        return shardId;
    }


    private static string GetShardIdFromObjectType(object instance)
    {
        if (instance is User)
            return "Users";
        if (instance is Blog)
            return "Blogs";
        if (instance is Post)
            return "Posts #";
        throw new ArgumentException("Cannot get shard id for '" + instance + "' because it is not a User, Blog or Post");
    }
}

((Post) obj).Id = "posts/" + nextPostShardId + "/";

For Users and Blogs, the decision is easy, based on their type. For posts, it is slightly more complicated, but still fairly easy, since we can extract the shard id from the id of the document.

ShardSelectionStrategy helps when we want to save new or existing objects, but ShardResolutionStrategy works for querying and loading.

Shard Resolution Strategy

public class BlogShardResolutionStrategy : IShardResolutionStrategy
{
    private readonly int numberOfShardsForPosts;


    public BlogShardResolutionStrategy(int numberOfShardsForPosts)
    {
        this.numberOfShardsForPosts = numberOfShardsForPosts;
    }


    public IList<string> SelectShardIds(ShardResolutionStrategyData srsd)
    {
        if (srsd.EntityType == typeof(User))
            return new[] { "Users" };
        if (srsd.EntityType == typeof(Blog))
            return new[] { "Blogs" };
        if (srsd.EntityType == typeof(Post))
        {
            if (srsd.Key == null) // general query
                return Enumerable.Range(0, numberOfShardsForPosts).Select(i => "Posts #" + (i + 1)).ToArray();
            // we can optimize better, since the key has the shard id
            // key structure is 'posts' / 'shard id' / 'post id'
            var parts = srsd.Key.Split(new[] { '/' }, StringSplitOptions.RemoveEmptyEntries);
            return new[] { "Posts #" + parts[1] };
        }


        throw new ArgumentException("Cannot get shard id for '" + srsd.EntityType +
        "' because it is not a User, Blog or Post");
    }
}

Cross Shard Transactions

Raven is a transactional system, but when you are using shards, you are talking to several instances at the same time. Each instance is independent and share no data with the other instances. That may cause result in surprising behavior in situations such as this one:

using (var session = documentStore.OpenSession())
{
    var user = new User { Name = "ayende" };
    var blog = new Blog { Name = "Ayende @ Rahien" };

    session.Store(user);
    session.Store(blog);

    session.SaveChanges();
}

When using a single Raven instance, failing to save the Blog will result in the User instance save rollback. But when using sharding, it is possible that the User will be saved (and committed) on one shard, while the Blog failed to save on another shard. The failure to save the Blog on the second server will not result in rollback on the first server.

If you do want such behavior, you need to utilize System.Transactions' Distributed Transaction support, by wrapping the calls inside a TransactionScope, like thus:

using (var tx = new TransactionScope())
using (var session = documentStore.OpenSession())
{
    var user = new User { Name = "ayende" };
    var blog = new Blog { Name = "Ayende @ Rahien" };

    session.Store(user);
    session.Store(blog);

    session.SaveChanges();
    tx.Complete();
}

Now we can relax, knowing that either both changes will complete successfully or both will be rolled back.

Sharding Approaches

How to shard your data

Stable sharding

Data locality