Method and system for efficiently replicating data in non-relational databases

• US 8,938,418 B2
• Filed: 09/03/2013
• Issued: 01/20/2015
• Est. Priority Date: 02/09/2010
• Status: Active Grant

• Current Assignee: Google LLC (Alphabet Inc.)
• Original Assignee: Google Inc. (Alphabet Inc.)
• Inventors: Zunger, Yonatan
• Primary Examiner(s): Pham, Hung Q
• Application Number:
  US14/017,117
• Publication Number:
  US 20140012812A1
• Time in Patent Office: 504 Days
• US Class Current: 707/609
• CPC Class Codes: G06F 16/2329 using versioning
• 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/275 Synchronous replication

Abstract

A method replicates data between instances of a distributed database. The method tracks changes to the distributed database at a first instance by storing deltas. Each delta includes a row identifier that identifies a row having a base value, and a sequence identifier that specifies an order in which the delta is applied to the base value to compute a current value for the row. The method identifies a set of deltas to send to a second instance based in part on an egress map at the first instance, wherein the egress map specifies which combinations of row identifier and sequence identifier have been acknowledged as received at the second instance. The method then transmits the identified set of deltas to the second instance. After receiving acknowledgement that the deltas in the identified set of deltas have been incorporated into the second instance, the egress map is updated accordingly.

Claims

1. A method of replicating data for a distributed database between a plurality of instances, each instance comprising one or more server computers with memory and one or more processors, the method comprising:

tracking changes to the distributed database at a first instance by storing deltas, wherein each delta includes;

a row identifier that identifies a row having a base content value; and

a sequence identifier that specifies an order in which the delta is applied to the base content value to compute a current value for the row;

identifying a set of deltas to send to a second instance based, at least in part, on a second egress map at the first instance, wherein the second egress map specifies which combinations of row identifier and sequence identifier have been acknowledged as received at the second instance;

transmitting the identified set of deltas to the second instance;

receiving acknowledgement that the deltas in the identified set of deltas have been incorporated into the second instance; and

updating the second egress map to indicate the acknowledged deltas.

2. The method of claim 1, wherein each delta further includes an instance identifier that specifies an instance where the delta was created.

3. The method of claim 1, wherein transmitting the identified set of deltas includes building a transmission matrix for the second instance that identifies deltas that have not yet been acknowledged as received at the second instance.

4. The method of claim 1, wherein the first instance and second instance are at distinct geographic locations.

5. The method of claim 1, wherein each sequence identifier comprises a timestamp and a unique tie breaker value that is assigned based on hardware and/or software at the first instance.

6. The method of claim 1, further comprising:

determining that the deltas in the identified set of deltas have not yet been acknowledged as received at a third instance;

transmitting a message to the second instance instructing the second instance to forward the identified set of deltas to the third instance;

receiving acknowledgement that the deltas in the identified set of deltas have been incorporated into the third instance; and

updating a third egress map to indicate the acknowledged deltas.

7. The method of claim 1, wherein transmitting the identified set of deltas to the second instance comprises:

partitioning the identified set of deltas into a first set and a second set;

transmitting the first set directly to the second instance; and

transmitting the second set to a third instance, with instructions to forward the second set to the second instance.

8. The method of claim 7, wherein the partitioning is based, at least in part, on analysis of total cost for transmitting deltas between pairs of instances.

9. The method of claim 7, wherein the partitioning is based, at least in part, on available bandwidth between pairs of instances.

10. The method of claim 7, wherein the partitioning is based, at least in part, on identifying a further set of deltas to send to the third instance using a third egress map at the first instance, and wherein the third egress map specifies which combinations of row identifier and sequence identifier have been acknowledged as received at the third instance.

11. The method of claim 10, wherein the second set equals the identified further set.

12. A computer system for replicating data for a distributed database between a plurality of instances, wherein each instance comprises one or more servers and each server has:

one or more processors;

memory; and

one or more programs stored in the memory, the one or more programs comprising instructions for;

tracking changes to the distributed database at a first instance by storing deltas, wherein each delta includes;

a row identifier that identifies a row having a base content value; and

a sequence identifier that specifies an order in which the delta is applied to the base content value to compute a current value for the row;

identifying a set of deltas to send to a second instance based, at least in part, on a second egress map at the first instance, wherein the second egress map specifies which combinations of row identifier and sequence identifier have been acknowledged as received at the second instance;

transmitting the identified set of deltas to the second instance;

receiving acknowledgement that the deltas in the identified set of deltas have been incorporated into the second instance; and

updating the second egress map to indicate the acknowledged deltas.

13. The computer system of claim 12, further comprising:

determining that the deltas in the identified set of deltas have not yet been acknowledged as received at a third instance;

transmitting a message to the second instance instructing the second instance to forward the identified set of deltas to the third instance;

receiving acknowledgement that the deltas in the identified set of deltas have been incorporated into the third instance; and

updating a third egress map to indicate the acknowledged deltas.

14. The computer system of claim 12, wherein transmitting the identified set of deltas to the second instance comprises:

partitioning the identified set of deltas into a first set and a second set;

transmitting the first set directly to the second instance; and

transmitting the second set to a third instance, with instructions to forward the second set to the second instance.

15. The computer system of claim 14, wherein the partitioning is based, at least in part, on analysis of total cost for transmitting deltas between pairs of instances.

16. The computer system of claim 14, wherein the partitioning is based, at least in part, on identifying a further set of deltas to send to the third instance using a third egress map at the first instance, and wherein the third egress map specifies which combinations of row identifier and sequence identifier have been acknowledged as received at the third instance.

17. A non-transitory computer readable storage medium storing one or more programs configured for execution by one or more processors of a computer system for replicating data for a distributed database between a plurality of instances, the one or more programs comprising instructions for:

tracking changes to the distributed database at a first instance by storing deltas, wherein each delta includes;

a row identifier that identifies a row having a base content value; and

a sequence identifier that specifies an order in which the delta is applied to the base content value to compute a current value for the row;

identifying a set of deltas to send to a second instance based, at least in part, on a second egress map at the first instance, wherein the second egress map specifies which combinations of row identifier and sequence identifier have been acknowledged as received at the second instance;

transmitting the identified set of deltas to the second instance;

receiving acknowledgement that the deltas in the identified set of deltas have been incorporated into the second instance; and

updating the second egress map to indicate the acknowledged deltas.

18. The computer readable storage medium of claim 17, further comprising:

determining that the deltas in the identified set of deltas have not yet been acknowledged as received at a third instance;

transmitting a message to the second instance instructing the second instance to forward the identified set of deltas to the third instance;

receiving acknowledgement that the deltas in the identified set of deltas have been incorporated into the third instance; and

updating a third egress map to indicate the acknowledged deltas.

19. The computer readable storage medium of claim 17, wherein transmitting the identified set of deltas to the second instance comprises:

partitioning the identified set of deltas into a first set and a second set;

transmitting the first set directly to the second instance; and

transmitting the second set to a third instance, with instructions to forward the second set to the second instance.

20. The computer readable storage medium of claim 19, wherein the partitioning is based, at least in part, on identifying a further set of deltas to send to the third instance using a third egress map at the first instance, and wherein the third egress map specifies which combinations of row identifier and sequence identifier have been acknowledged as received at the third instance.