Implementing database synchronization with entity framework core

Implementing database synchronization with entity framework core

Ok, so far, our synchronization framework is only implemented for an in-memory database that we use for testing purposes.

Now let’s implement a different use case, lets add synchronization functionality to an entity framework core DbContext.

As I explained before, the key part of synchronizing data using delta encoding is to be able to track the differences that happen to a data object, in this case, a relational database.

these are the task that we need to do to accomplish our goal

  1. Find out how entity framework converts the changes that happen to the objects to SQL commands
  2. Decide what information we need to track and save as a delta
  3. Create the infrastructure to save deltas (IDeltaStore)
  4. Create the infrastructure to process deltas (IDeltaProcessor)
  5. Implement the synchronization node functionality in an Entity Framework DbContext(ISyncClientNode)
  6. Create a test scenario

 

1 Find out how entity framework converts the changes that happen to the objects to SQL commands

In our companies (BitFrameworks & Xari) we have been working in data synchronization for a while, but all this work has been done in the XPO realm.

We know that in most ORMs frameworks there is a layer of the ORM that is in charge of translating the changes made to objects into SQL commands, the trick is to locate this layer. So while I was trapped in Mexico waiting for a flight back to Phoenix, I decided to dig into entity framework’s core GitHub report, this is what I found https://github.com/dotnet/efcore/blob/b18a7efa7c418e43184db08c6d1488d6600054cb/src/EFCore.Relational/Update/Internal/BatchExecutor.cs#L161

public virtual async Task<int> ExecuteAsync(
           IEnumerable<ModificationCommandBatch> commandBatches,
           IRelationalConnection connection,
           CancellationToken cancellationToken = default)

As you can see one of the parameters is an IEnumerable of ModificationCommandBatch https://github.com/dotnet/efcore/blob/main/src/EFCore.Relational/Update/ModificationCommandBatch.cs this command batch exposes a read-only list of modification commands (ModificationCommand)

https://github.com/dotnet/efcore/blob/cc53b3e80755e5d882bb21ef10e0e0e33194d9bd/src/EFCore.Relational/Update/ModificationCommandBatch.cs#L30

public abstract class ModificationCommandBatch
{
    /// <summary>
    ///     The list of conceptual insert/update/delete <see cref="ModificationCommands" />s in the batch.
    /// </summary>
    public abstract IReadOnlyList<IReadOnlyModificationCommand> ModificationCommands { get; }

now let’s take look into the ModificationCommand https://github.com/dotnet/efcore/blob/main/src/EFCore.Relational/Update/ModificationCommand.cs this class provides all the information about the changes that will be converted into SQL commands, which means that if we serialize this object and save it as a delta we can then send it to another node and replicate the changes…VOILA!!!

Now here is a stone in our path, the class https://github.com/dotnet/efcore/blob/main/src/EFCore.Relational/Update/ModificationCommand.cs is not serializable or to say it in a better way NOT easily serializable, so let’s stop here for a moment and move to a different task

So now we know where the changes that we need to keep track of are, now let’s try to understand how those changes are converted into SQL commands and then executed into the database.

2 Decide what information we need to track and save as a delta

Entity framework core uses dependency injection to be able to handle different database engines so the idea here is that there are a lot of small services that can be replaced in other to create a different implementation, for example, SQLite, SqlServer, Postgres, etc …

After a lot of digging, I found that the service that is in charge of generating the update commands (insert, update and delete) UpdateSqlGenerator

https://github.com/dotnet/efcore/blob/main/src/EFCore.Relational/Update/UpdateSqlGenerator.cs

this class implements IUpdateSqlGenerator https://github.com/dotnet/efcore/blob/main/src/EFCore.Relational/Update/IUpdateSqlGenerator.cs and as you can see all methods receive a string builder and a ModificationCommand so this is the service  in charge of translating the ModificationCommand into SQL commands and SQL commands are easy to serialize because they are just text, so this is what we are going to serialize and save as a delta

    public interface IUpdateSqlGenerator
    {
        /// <summary>
        ///     Generates SQL that will obtain the next value in the given sequence.
        /// </summary>
        /// <param name="name">The name of the sequence.</param>
        /// <param name="schema">The schema that contains the sequence, or <see langword="null" /> to use the default schema.</param>
        /// <returns>The SQL.</returns>
        string GenerateNextSequenceValueOperation(string name, string? schema);

        /// <summary>
        ///     Generates a SQL fragment that will get the next value from the given sequence and appends it to
        ///     the full command being built by the given <see cref="StringBuilder" />.
        /// </summary>
        /// <param name="commandStringBuilder">The builder to which the SQL fragment should be appended.</param>
        /// <param name="name">The name of the sequence.</param>
        /// <param name="schema">The schema that contains the sequence, or <see langword="null" /> to use the default schema.</param>
        void AppendNextSequenceValueOperation(
            StringBuilder commandStringBuilder,
            string name,
            string? schema);

        /// <summary>
        ///     Appends a SQL fragment for the start of a batch to
        ///     the full command being built by the given <see cref="StringBuilder" />.
        /// </summary>
        /// <param name="commandStringBuilder">The builder to which the SQL fragment should be appended.</param>
        void AppendBatchHeader(StringBuilder commandStringBuilder);

        /// <summary>
        ///     Appends a SQL command for deleting a row to the commands being built.
        /// </summary>
        /// <param name="commandStringBuilder">The builder to which the SQL should be appended.</param>
        /// <param name="command">The command that represents the delete operation.</param>
        /// <param name="commandPosition">The ordinal of this command in the batch.</param>
        /// <returns>The <see cref="ResultSetMapping" /> for the command.</returns>
        ResultSetMapping AppendDeleteOperation(
            StringBuilder commandStringBuilder,
            IReadOnlyModificationCommand command,
            int commandPosition);

        /// <summary>
        ///     Appends a SQL command for inserting a row to the commands being built.
        /// </summary>
        /// <param name="commandStringBuilder">The builder to which the SQL should be appended.</param>
        /// <param name="command">The command that represents the delete operation.</param>
        /// <param name="commandPosition">The ordinal of this command in the batch.</param>
        /// <returns>The <see cref="ResultSetMapping" /> for the command.</returns>
        ResultSetMapping AppendInsertOperation(
            StringBuilder commandStringBuilder,
            IReadOnlyModificationCommand command,
            int commandPosition);

        /// <summary>
        ///     Appends a SQL command for updating a row to the commands being built.
        /// </summary>
        /// <param name="commandStringBuilder">The builder to which the SQL should be appended.</param>
        /// <param name="command">The command that represents the delete operation.</param>
        /// <param name="commandPosition">The ordinal of this command in the batch.</param>
        /// <returns>The <see cref="ResultSetMapping" /> for the command.</returns>
        ResultSetMapping AppendUpdateOperation(
            StringBuilder commandStringBuilder,
            IReadOnlyModificationCommand command,
            int commandPosition);
    }

3 Create the infrastructure to save deltas (Implementing IDeltaStore)

Now is time to create a delta store, this is an easy one since we only need to inherit from our delta store base and save the information in an entity framework DbContext, so here is the implementation

https://github.com/egarim/SyncFramework/blob/main/src/EntityFrameworkCore/BIT.Data.Sync.EfCore/EFDeltaStore.cs

if you want to compare it with other delta store implementations you can take a look at the in-memory version here

https://github.com/egarim/SyncFramework/blob/main/src/BIT.Data.Sync/Imp/MemoryDeltaStore.cs

4 Create the infrastructure to process deltas (implementing IDeltaProcessor)

So far, we know what we need to store in the deltas which basically is SQL commands and their parameters so it means to process those SQL Commands our delta processor needs to create a database connection and execute SQL commands

https://github.com/egarim/SyncFramework/blob/main/src/EntityFrameworkCore/BIT.Data.Sync.EfCore/EFDeltaProcessor.cs

public EFDeltaProcessor(DbContext dBContext) 
{
    _dBContext = dBContext;
  
}
public EFDeltaProcessor(string connectionstring, string DbEngineAlias, string ProviderInvariantName)
{

    this.CurrentDbEngine = DbEngineAlias;
    this.connectionString = connectionstring;

    try
    {
        factory = DbProviderFactories.GetFactory(ProviderInvariantName);
    }
    catch (Exception ex)
    {
        Debug.WriteLine(ex.Message);
        throw new Exception("There was a problem creating the database connection using DbProviderFactories.GetFactory. Please your make sure the DbProviderFactory for your database is registered https://docs.microsoft.com/en-us/dotnet/api/system.data.common.dbproviderfactories.registerfactory?view=net-5.0", ex);
    }
    //TODO check provider registration later

    //DbProviderFactories.RegisterFactory("Microsoft.Data.SqlClient", SqlClientFactory.Instance);
}

there are a few things to notice in that class, first, it has 2 constructors because we need 2 different ways to create the connection to the database, one using the entity framework DbContext and one using ADO.NET DbProviderFactory

All the magic happens in the ProcessDeltas method, this method is in charge of, extract the content of the deltas and transforming them into SQL commands and parameters, and then executing the command.

please notice that the content of each delta is an instance of ModificationCommandData

https://github.com/egarim/SyncFramework/blob/main/src/EntityFrameworkCore/BIT.Data.Sync.EfCore/Data/ModificationCommandData.cs

which is a class that allows us to store multiple SQL commands (for different database engines) and their parameters

5 Implement the synchronization node functionality in an Entity Framework DbContext(ISyncClientNode)

At the moment we are able to produce and process deltas for entity framework relational, so the next step is to implement the functionality of synchronization client node by implementing the following interface

https://github.com/egarim/SyncFramework/blob/main/src/BIT.Data.Sync/Client/ISyncClientNode.cs

namespace BIT.Data.Sync.Client
{
    public interface ISyncClientNode
    {
        IDeltaProcessor DeltaProcessor { get; }
        IDeltaStore DeltaStore { get; }
        ISyncFrameworkClient SyncFrameworkClient { get; }
        string Identity { get;  }

    }
}

https://github.com/egarim/SyncFramework/blob/main/src/EntityFrameworkCore/BIT.Data.Sync.EfCore/SyncFrameworkDbContext.cs

The server-side

I’m not going to show the implementation of the server since that implementation is generic and uses the same delta store and delta processor that we created at the beginning of this article. for more information check the following links

Adding network support

https://www.jocheojeda.com/2021/10/17/syncframework-adding-network-support/

Testing network support

https://www.youtube.com/watch?v=mSl0n0O5QIg&t=4s

 

The next post its going to be a video testing a simple synchronization scenario, see you in the next post!!!

 

 

Parts of a Synchronization Framework

Parts of a Synchronization Framework

In the last post, we talked about what are deltas and how by using them we can synchronize data structures.

So, in this post, I will describe the necessary parts needed to implement Delta-based synchronization, let’s start

  • Data Object: any database, object, graph, or file system that we are tracking for synchronization
  • Delta: a delta is a data difference between the original and the current state of a data object
  • Node: is a point of the synchronization network, there are 2 types of nodes
    1. Client node: a node that is able to produce and process deltas
    2. Server node: a node that is used only to exchange deltas, it can optionally process deltas too.
  • Delta Store: storage where you can save deltas so you can later exchange them with other nodes
  • Delta Processor: a utility that helps you includes the deltas created in other nodes in your current copy of the data object

Now let’s implement this concept to synchronize a database

Ok so each database that we want to synchronize will be a client node and a client node should be able to produce and store deltas and to process deltas produced by other nodes, so our database node should look like the following diagram

The server node can be as simple as just delta storage exposed in an HTTP API as you can see in the basic server node diagram or use several delta storages and delta processors as show on the complex server node diagram

                                         

 

And with those diagrams, I finish this post, in the next post we will implement those concepts in C# code