Flow for multi-master replication in distributed storage

• US 9,864,791 B2
• Filed: 03/04/2015
• Issued: 01/09/2018
• 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
• Primary Examiner(s): Lu, Kuen
• Application Number:
  US14/638,357
• Publication Number:
  US 20160179915A1
• Time in Patent Office: 1,042 Days
• Field of Search: 707624
• 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, wherein an eTag comprises an identifier for a specific version of the table row, and wherein the obtaining comprises for each target storage;

determining if a matching row corresponding to the row identifier exists in the target storage; and

when a matching row exists in the target storage, returning the matching row and the first entity tag (eTag) corresponding to the matching row;

determining whether at least one of the obtained rows has a later client timestamp than the table row corresponding to the row identifier from the source table;

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 comprises:

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

3. 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.

4. 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.

5. 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.

6. 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.

7. The media of claim 1, 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;

replacing the winning row with the row with differing first eTag and second eTag as a new winning row, the new winning row representing a later version of the row than the winning row; and

performing the replication operations on the obtained rows of the one or more target storages using the new winning row.

8. 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.

9. 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.

10. 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.

11. 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, the fetching comprising for each table;

determining if a matching row corresponding to the row identifier 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, wherein an eTag comprises an identifier for a specific version of the row;

determining whether at least one fetched row from the plurality of tables has a later timestamp than the identified row from the first table;

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 fetched row with the latest timestamp; and

replicating the winning row among the plurality of tables.

12. The method of claim 11, 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.

13. The method of claim 11, 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.

14. The method of claim 11, 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 the first eTag is retained.

15. The method of claim 11, 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, replacing the winning row with the fetched row with different first and second eTags;

replicating the fetched row with different first and second eTags among the plurality of tables.

16. A computer-implemented system comprising:

one or more processors; and

one or more computer-storage media storing computer-useable instructions that, when executed by the one or more processors, cause the one or more processors to perform a 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, the fetching comprising for each table;

determining if a matching row corresponding to the row identifier 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, wherein an eTag comprises an identifier for a specific version of the row;

determining whether at least one fetched row from the plurality of tables has a later timestamp than the identified row from the first table;

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 fetched row with the latest timestamp; and

replicating the winning row among the plurality of tables.

17. The system of claim 16, 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.

18. The system of claim 16, wherein the method further comprises:

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.

19. The system of claim 16, 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 the first eTag is retained.

20. The system of claim 16, 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, replacing the winning row with the fetched row with different first and second eTags;

replicating the fetched row with different first and second eTags among the plurality of tables.