FLOW FOR MULTI-MASTER REPLICATION IN DISTRIBUTED STORAGE

• US 20160179915A1
• Filed: 03/04/2015
• Published: 06/23/2016
• Est. Priority Date: 12/19/2014
• Status: Active Grant

• Current Assignee: Microsoft Corporation, Microsoft Technology Licensing LLC (Microsoft Corporation)
• Original Assignee: Microsoft Technology Licensing LLC (Microsoft Corporation)
• Inventors: GREBNOV, ILYA, BANINA, SAMUEL, LAMANNA, CHARLES, LAM, KEVIN
• Granted Patent: US 9,864,791 B2
• Time in Patent Office: Days
• US Class Current: 1/1
• CPC Class Codes: G06F 16/2282 Tablespace storage structures; Management thereof
• CPC Class Codes: G06F 16/2365 Ensuring data consistency and integrity
• CPC Class Codes: G06F 16/27 Replication, distribution or synchronisation of data between databases or within a distributed database system; Distributed database system architectures therefor
• CPC Class Codes: G06F 16/278 Data partitioning, e.g. horizontal or vertical partitioning
• CPC Class Codes: G06F 16/9535 Search customisation based on user profiles and personalisation

Abstract

Embodiments are directed to replicating data in distributed storage. A replication message may be retrieved from a message queue associated with a source table. The replication message may include a row identifier. One or more target storages within a same replication group as the source table may be identified. A row from each of the one or more target storages may be obtained corresponding to the row identifier. A winning row may be determined from the obtained rows based on a latest timestamp of the row. A replication operation may be created based on the winning row. The replication operation may be performed on the obtained rows from each of the target storages.

Claims

1. One or more computer-storage media storing computer-executable instructions that, when executed by a computing device having a processor, cause the computing device to perform a method of replicating data in distributed storage, the method comprising:

retrieving a replication message from a message queue associated with a source table, the source table being one of a plurality of source tables in a replication group, the replication message comprising a row identifier;

identifying one or more target storages within the replication group, the target storages including tables within the replication group;

obtaining a table row corresponding to the row identifier and a first entity tag (eTag) for the table row from each of the one or more target storages, an eTag comprises an identifier for a specific version of the table row;

determining a winning row from the obtained rows based on a latest client timestamp of the obtained rows, the winning row being a table row with a latest version of the table row;

creating replication operations based on the winning row, wherein a replication operation comprises instructions on inserting data from the winning row in to one or more target storages; and

performing batch execution of the replication operations to the one or more target storages.

2. The media of claim 1, wherein the obtaining the table row corresponding to the row identifier comprises:

determining if a matching row corresponding to the row identifier exists for each of the one or more target storages; and

when a matching row exists for one of the one or more target storages the table, returning the matching row and the first eTag corresponding to the matching row.

3. The media of claim 1, wherein the obtaining the table row corresponding to the row identifier comprises:

determining if a matching row corresponding to the row identifier exists for each of the one or more target storages; and

when a matching row does not exist, returning a placeholder empty row instance and a first eTag corresponding to the placeholder empty row instance.

4. The media of claim 1, wherein the determining a winning row from the obtained rows based on a latest client timestamp of the obtained row comprises:

obtaining a client timestamp indicating a last operation date of an obtained row from the obtained rows of the one or more target storages; and

returning the obtained row with a latest client timestamp.

5. The media of claim 1, the method further comprising:

determining that at least one obtained row contains a placeholder empty row instance, wherein when the at least one obtained row contains a placeholder empty row instance, the creating replication operations comprises creating a replication operation to insert the winning row to the target storage with the placeholder empty row instance.

6. The media of claim 1, wherein the creating replication operations comprises creating a replication operation to replace an existing row instance of the one or more table stores with the winning row.

7. The media of claim 1, the method further comprising:

determining that the winning row is marked for deletion, wherein when the winning row is marked for deletion, the creating replication operations comprises creating at least one replication operation to delete a row from at least one of the one or more target storages.

8. The media of claim 2, wherein the performing batch execution of the replication operations comprises:

obtaining a second eTag for each of the obtained rows of the one or more target storages, the second eTag comprising the identifier for the specific version of each of the obtained rows;

comparing the first eTag and second eTag of each obtained row; and

when the first eTag and second eTag of an obtained row are different;

halting performance of the replication operations on the obtained rows of the one or more target storages; and

requeueing the replication message to the message queue.

9. The media of claim 1, wherein the replication message comprises instructions to replicate data to one or more table rows of the source tables in the replication group, wherein the instructions include unique row identifiers for each table row.

10. The media of claim 1, the method further comprising when the replication operations are successfully performed at the target storages within the replication group, dequeuing the replication message from the message queue.

11. The media of claim 1, the method further comprising when at least one operation of the replication operations is unsuccessful, requeuing the replication message to the message queue.

12. A computer-implemented method of processing replication messages in distributed storage, the method comprising:

receiving a replication message from a message queue associated with a first table among a plurality of tables, at least two tables of the plurality of tables being located at different locations;

obtaining a row identifier identifying a row to be updated from the replication message;

fetching a row from each of the plurality of tables corresponding to the row identifier;

determining a winning row based on a latest timestamp of the fetched row from each of the plurality of tables, the winning row being the obtained row with the latest timestamp; and

replicating the winning row among the plurality of tables.

13. The method of claim 12, wherein the fetching a row from each of the plurality of tables corresponding to the row identifiers comprises:

determining if a matching row corresponding to the row identifiers exists in the table; and

when a matching row exists in the table, returning the matching row and a first entity tag (eTag) corresponding to the matching row, an eTag comprises an identifier for a specific version of the table row.

14. The method of claim 12, wherein when a fetched row from a table of the plurality of tables is a placeholder row, the replicating the winning row comprises creating an insert operation for the table corresponding to the winning row.

15. The method of claim 12, further comprising:

determining that the winning row is marked for deletion, wherein when the winning row is marked for deletion, the replicating the winning row comprises creating a delete operation for the fetched row from each of the plurality of tables.

16. The method of claim 12, wherein the replicating the winning row comprises creating a replace operation for the fetched row from each of the plurality of tables with the winning row, wherein a first eTag is retained.

17. The method of claim 12, wherein the replicating the winning row comprises:

obtaining a second eTag for the each of the fetched rows of the plurality of tables, the second eTag representing a state of the fetched row at the time the second eTag was generated;

comparing the first eTag and second eTag;

when the first eTag and second eTag are different;

halting replicating the winning row among the plurality of tables; and

requeueing the replication message to the message queue.

18. A system for guaranteeing replication and eventual consistency in multi-master configuration, comprising:

a message queue;

a plurality of tables; and

a replication worker component configured to;

receive a replication message from a message queue associated with a first table among the plurality of tables, at least two tables of the plurality of tables being located at different locations;

obtain row identifiers from the replication message;

determine if a matching row corresponding to the row identifiers exists in each of the plurality of tables;

when a matching row exists in a table of the plurality of tables, return the matching row and client timestamp and first entity tag (eTag) corresponding to the matching row;

when a matching row does not exist in a table of the plurality of tables, return a placeholder row instance and first eTag corresponding to the placeholder row instance;

select a winning row from the returned rows with a latest client timestamp, the winning row being the returned row with the latest client timestamp indicating a latest version of the returned row;

create replication operations based on the winning row;

perform batch execution of the replication operations to the target storages; and

remove the replication message from the message queue.

19. The system of claim 18, wherein the replication worker is further configured to:

receive a second replication message from a message queue associated with a second table among the plurality of tables;

determine that a new winning row exists if a matching row corresponding to the second row identifiers exists in each of the plurality of tables;

when a matching row exists in a table of the plurality of tables;

returning the matching row;

when a matching row does not exist in a table of the plurality of tables, returning a placeholder row;

obtaining a timestamp for each of the returned rows from each of the plurality of tables;

selecting a winning row from the returned rows with a latest timestamp;

replicate the winning row among the plurality of tables; and

removing the message from the message queue.

20. The system of claim 18, wherein each of the plurality of tables is configured to:

when a matching row exists in a table of the plurality of tables;

return the matching row;

when a matching row does not exist in a table of the plurality of tables, return a placeholder row;

return a timestamp and eTag for each of the returned rows; and

replicate the winning row in the table.