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

• US 8,380,659 B2
• Filed: 02/09/2010
• Issued: 02/19/2013
• 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:
  US12/703,167
• Publication Number:
  US 20110196827A1
• Time in Patent Office: 1,106 Days
• Field of Search: None
• 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 identifies at least two instances of the database at distinct geographic locations. The method tracks changes to the database by storing deltas. Each delta has a row identifier that identifies the piece of data modified, a sequence identifier that specifies the order in which the deltas are applied to the data, and an instance identifier that specifies where the delta was created. The method determines which deltas to send using an egress map that specifies which combinations of row identifier and sequence identifier have been acknowledged as received at other instances. The method builds a transmission matrix that identifies deltas that have not yet been acknowledged as received. The method then transmits deltas identified in the transmission matrix. After receiving acknowledgement that transmitted deltas have been incorporated into databases at other instances, the method updates the egress map.

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:

identifying a first instance of the distributed database at a first geographic location;

identifying a second instance of the distributed database at a second geographic location;

tracking changes to the distributed database at the first instance by storing deltas, each delta having a row identifier that identifies a piece of data modified, a sequence identifier that specifies an order in which the delta is applied to the second instance, and an instance identifier that specifies an instance where the delta was created;

determining which deltas are to be sent to the second instance using 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;

building a second transmission matrix for the second instance that identifies deltas that have not yet been acknowledged as received at the second instance;

transmitting deltas identified in the second transmission matrix to the second instance;

receiving acknowledgement that transmitted deltas have been incorporated in the second instance; and

updating the second egress map to indicate acknowledged deltas.

2. The method of claim 1, further comprising:

identifying a third instance of the distributed database at a third geographic location distinct from the first and second geographic locations;

determining which deltas are to be sent to the third instance using a third egress map at the first instance, wherein the third egress map specifies which combinations of row identifier and sequence identifier have been acknowledged as received at the third instance;

building a third transmission matrix for the third instance that identifies deltas that have not yet been acknowledged as received at the third instance;

modifying the transmission matrices for the second and third instances to form one or more revised transmission matrices, wherein deltas identified in each revised transmission matrix are transmitted to a respective location to update the instance at the respective location, and deltas identified in at least one of the revised transmission matrices are transmitted to the second location for subsequent transmission from the second location to the third location;

receiving acknowledgement that deltas transmitted to the third instance, either directly or indirectly via the second instance, have been incorporated in the third instance; and

updating the third egress map to indicate acknowledged deltas.

3. The method of claim 2, including assigning a cost for transmissions between pairs of geographic locations, and wherein modifying the transmission matrices includes an analysis of the total cost for transmitting the deltas to the second and third geographic locations.

4. The method of claim 2, wherein modifying the transmission matrices includes determining bandwidth availability between geographic locations of the instances.

5. The method of claim 2, wherein the transmission matrices for the second and third instances are the same, there is only one revised transmission matrix, the one revised transmission matrix is the same as the transmission matrices, and deltas identified in the revised transmission matrix are transmitted to the second geographic location for subsequent transmission to the third geographic location.

6. 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 each instance of the distributed database.

7. The method of claim 1, wherein the second geographic location is distinct from the first geographic location.

8. A server system, comprising a plurality of servers, each server having:

one or more processors;

memory; and

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

identifying a first instance of the distributed database at a first geographic location;

identifying a second instance of the distributed database at a second geographic location;

tracking changes to the distributed database at the first instance by storing deltas, each delta having a row identifier that identifies a piece of data modified, a sequence identifier that specifies an order in which the delta is applied to the second instance, and an instance identifier that specifies an instance where the delta was created;

determining which deltas are to be sent to the second instance using 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;

building a second transmission matrix for the second instance that identifies deltas that have not yet been acknowledged as received at the second instance;

transmitting deltas identified in the second transmission matrix to the second instance;

receiving acknowledgement that transmitted deltas have been incorporated in the second instance; and

updating the second egress map to indicate acknowledged deltas.

9. The server system of claim 8, further comprising instructions for:

identifying a third instance of the distributed database at a third geographic location distinct from the first and second geographic locations;

determining which deltas are to be sent to the third instance using a third egress map at the first instance, wherein the third egress map specifies which combinations of row identifier and sequence identifier have been acknowledged as received at the third instance;

building a third transmission matrix for the third instance that identifies deltas that have not yet been acknowledged as received at the third instance;

modifying the transmission matrices for the second and third instances to form one or more revised transmission matrices, wherein deltas identified in each revised transmission matrix are transmitted to a respective location to update the instance at the respective location, and deltas identified in at least one of the revised transmission matrices are transmitted to the second location for subsequent transmission from the second location to the third location;

receiving acknowledgement that deltas transmitted to the third instance, either directly or indirectly via the second instance, have been incorporated in the third instance; and

updating the third egress map to indicate acknowledged deltas.

10. The server system of claim 9, wherein the transmission matrices for the second and third instances are the same, there is only one revised transmission matrix, the one revised transmission matrix is the same as the transmission matrices, and deltas identified in the revised transmission matrix are transmitted to the second geographic location for subsequent transmission to the third geographic location.

11. A computer readable storage medium storing one or more programs configured for execution by a server computer system having one or more processors and memory storing one or more programs for execution by the one or more processors, the one or more programs comprising instructions to:

identify a first instance of the distributed database at a first geographic location;

identify a second instance of the distributed database at a second geographic location;

track changes to the distributed database at the first instance by storing deltas, each delta having a row identifier that identifies a piece of data modified, a sequence identifier that specifies an order in which the delta is applied to the second instance, and an instance identifier that specifies an instance where the delta was created;

determine which deltas are to be sent to the second instance using 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;

build a second transmission matrix for the second instance that identifies deltas that have not yet been acknowledged as received at the second instance;

transmit deltas identified in the second transmission matrix to the second instance;

receive acknowledgement that transmitted deltas have been incorporated in the second instance; and

update the second egress map to indicate acknowledged deltas.

12. The computer readable storage medium of claim 11, further comprising instructions to:

identify a third instance of the distributed database at a third geographic location distinct from the first and second geographic locations;

determine which deltas are to be sent to the third instance using a third egress map at the first instance, wherein the third egress map specifies which combinations of row identifier and sequence identifier have been acknowledged as received at the third instance;

build a third transmission matrix for the third instance that identifies deltas that have not yet been acknowledged as received at the third instance;

modify the transmission matrices for the second and third instances to form one or more revised transmission matrices, wherein deltas identified in each revised transmission matrix are transmitted to a respective location to update the instance at the respective location, and deltas identified in at least one of the revised transmission matrices are transmitted to the second location for subsequent transmission from the second location to the third location;

receive acknowledgement that deltas transmitted to the third instance, either directly or indirectly via the second instance, have been incorporated in the third instance; and

update the third egress map to indicate acknowledged deltas.

13. The computer readable storage medium of claim 12, wherein the transmission matrices for the second and third instances are the same, there is only one revised transmission matrix, the one revised transmission matrix is the same as the transmission matrices, and deltas identified in the revised transmission matrix are transmitted to the second geographic location for subsequent transmission to the third geographic location.