Multi-master data replication in a distributed multi-tenant system
View Patent ↗A multi-master replication system is disclosed. The multi-master replication system allows a large set of peer instances to collaboratively replicate data to each other.
1. A method comprising:
monitoring for changes to a replicated database by an instance of the replicated database, wherein the instance is one of a plurality instances of the replicated database, and each instance is separately responsible for multiple database tables local to the respective instances, and wherein the plurality of instances of the replicated database comprise two or more data centers each data center having a leader instance and one or more non-leader instances, the one or more non-leader instances to communicate with other non-leader instances of the same data center and the leader instance of the same data center and not to instances of other data centers, the leader instance to communicate with leader instances of other data centers and not with non-leader instances of other data centers, the monitoring including
detecting by the instance a change to at least one record of the local database tables;
in response to the detecting, adding information to a chunk, wherein a chunk stores metadata about record changes and corresponds to a discrete unit of progress and is specific to a single instance and provide a single consistent unit for both change detection and transport; and
serving the chunk to another instance of the replicated database via one or more leader instances to replicate the detected change to the other instance; and
receiving by the instance a chunk from a remote instance of the replicated database, which has metadata identifying a set of changed records for database tables of the remote instance;
in response to the receiving, requesting underlying data for changed records indicated in the received chunk via the one or more leader instances; and
upserting the changed records into the local database tables.
2. The method of claim 1 , wherein detecting the change further comprises:
identifying modifications to rows of the local database tables;
for each identified modification, writing a metadata entry to an indexing queue table; and
defining chunks by grouping metadata entries of the indexing queue table into multiple segments, each chunk identified by a first and a last record of a respective segment.
3. The method of claim 2 , wherein each chunk includes a source server identifier, an entity type, and a database partition number.
4. The method of claim 2 , wherein identifying the modifications further comprises:
creating a pool of change detection threads to monitor writes to the local database tables.
5. The method of claim 2 , further comprising:
keeping a store of multiple chunks to serve to remote instances.
6. The method of claim 5 , further comprising:
merging data from multiple chunks into a single chunk.
7. The method of claim 1 , wherein serving the chunk further comprises:
processing an HTTP LIST function to indicating available chunks to the other instance; and
processing an HTTP GET function to return underlying data for a specific chunk.
8. The method of claim 1 , wherein the instance includes a modification timestamp indicating a most recent update of the instance, and wherein requesting the underlying data further comprises:
updating the modification timestamp with a modification timestamp of the received chunk.
9. The method of claim 1 , wherein receiving the chunk further comprises:
identifying a time-based trigger to update the instance; and
requesting chunks from one or more remote instances,
wherein the instance requests underlying data and upserts changed records for each received chunk.
10. The method of claim 1 , wherein upserting the changed records further comprises:
inserting new records into the local database tables for records that do not exist in the local database tables; and
merging changed data into the local database tables for records that do exist in the local database tables.
11. The method of claim 10 , wherein upserting the changed records further comprises:
deleting rows from the local database tables marked as delete in the received chunk; and
persisting the deleted rows into a delete table.
12. The method of claim 1 , wherein upserting the changed records further comprises:
determining if a modification timestamp of a changed row of data is more recent than a corresponding row in the local database tables; and
merging changed data into the local database tables when the modification timestamp is more recent; otherwise,
ignoring the changed data.
13. A non-transitory machine readable storage medium storing instructions, which when executed by a machine cause the machine to perform operations including:
monitoring for changes to a replicated database by an instance of the replicated database, wherein the instance is one of a plurality instances of the replicated database, and each instance is separately responsible for multiple database tables local to the respective instances, and wherein the plurality of instances of the replicated database comprise two or more data centers each data center having a leader instance and one or more non-leader instances, the one or more non-leader instances to communicate with other non-leader instances of the same data center and the leader instance of the same data center and not to instances of other data centers, the leader instance to communicate with leader instances of other data centers and not with non-leader instances of other data centers, the monitoring including
detecting by the instance a change to at least one record of the local database tables;
in response to the detecting, adding information to a chunk, wherein a chunk stores metadata about record changes and corresponds to a discrete unit of progress and is specific to a single instance and provide a single consistent unit for both change detection and transport; and
serving the chunk to another instance of the replicated database via one or more leader instances to replicate the detected change to the other instance; and
receiving by the instance a chunk from a remote instance of the replicated database, which has metadata identifying a set of changed records for database tables of the remote instance;
in response to the receiving, requesting underlying data for changed records indicated in the received chunk via the one or more leader instances; and
upserting the changed records into the local database tables.
14. The machine-readable storage medium of claim 13 , wherein the instructions for detecting the change further comprise instructions for
identifying modifications to rows of the local database tables;
for each identified modification, writing a metadata entry to an indexing queue table; and
defining chunks by grouping metadata entries of the indexing queue table into multiple segments, each chunk identified by a first and a last record of a respective segment.
15. The machine-readable storage medium of claim 14 , wherein each chunk includes a source server identifier, an entity type, and a database partition number.
16. The machine-readable storage medium of claim 14 , wherein the instructions for identifying the modifications further comprise instructions for
creating a pool of change detection threads to monitor writes to the local database tables.
17. The machine-readable storage medium of claim 14 , further comprising instructions for
keeping a store of multiple chunks to serve to remote instances.
18. The machine-readable storage medium of claim 17 , further comprising instructions for
merging data from multiple chunks into a single chunk.
19. The machine-readable storage medium of claim 13 , wherein the instructions for serving the chunk further comprise instructions for
processing an HTTP LIST function to indicating available chunks to the other instance; and
processing an HTTP GET function to return underlying data for a specific chunk.
20. The machine-readable storage medium of claim 13 , wherein the instance includes a modification timestamp indicating a most recent update of the instance, and wherein the instructions for requesting the underlying data further comprise instructions for
updating the modification timestamp with a modification timestamp of the received chunk.
21. The machine-readable storage medium of claim 13 , wherein the instructions for receiving the chunk further comprise instructions for
identifying a time-based trigger to update the instance; and
requesting chunks from one or more remote instances,
wherein the instance requests underlying data and upserts changed records for each received chunk.
22. The machine-readable storage medium of claim 13 , wherein the instructions for upserting the changed records further comprise instructions for
inserting new records into the local database tables for records that do not exist in the local database tables; and
merging changed data into the local database tables for records that do exist in the local database tables.
23. The machine-readable storage medium of claim 22 , wherein the instructions for upserting the changed records further comprise instructions for
deleting rows from the local database tables marked as delete in the received chunk; and
persisting the deleted rows into a delete table.
24. The machine-readable storage medium of claim 13 , wherein the instructions for upserting the changed records further comprise instructions for
determining if a modification timestamp of a changed row of data is more recent than a corresponding row in the local database tables; and
merging changed data into the local database tables when the modification timestamp is more recent; otherwise,
ignoring the changed data.
25. An apparatus for managing a replicated database in an on-demand database system, comprising:
a processor; and
a memory device having a sequence of instructions stored thereon, which when executed by the processor, cause the processor to carry out operations including
monitoring for changes to a replicated database by an instance of the replicated database, wherein the instance is one of a plurality instances of the replicated database, and each instance is separately responsible for multiple database tables local to the respective instances, and wherein the plurality of instances of the replicated database comprise two or more data centers each data center having a leader instance and one or more non-leader instances, the one or more non-leader instances to communicate with other non-leader instances of the same data center and the leader instance of the same data center and not to instances of other data centers, the leader instance to communicate with leader instances of other data centers and not with non-leader instances of other data centers, the monitoring including
detecting by the instance a change to at least one record of the local database tables;
in response to the detecting, adding information to a chunk, wherein a chunk stores metadata about record changes and corresponds to a discrete unit of progress and is specific to a single instance and provide a single consistent unit for both change detection and transport; and
serving the chunk to another instance of the replicated database via one or more leader instances to replicate the detected change to the other instance; and
receiving by the instance a chunk from a remote instance of the replicated database, which has metadata identifying a set of changed records for database tables of the remote instance;
in response to the receiving, requesting underlying data for changed records indicated in the received chunk via the one or more leader instances; and
upserting the changed records into the local database tables.
26. The apparatus of claim 25 , wherein the instructions for detecting the change further comprise instructions for
identifying modifications to rows of the local database tables;
for each identified modification, writing a metadata entry to an indexing queue table; and
defining chunks by grouping metadata entries of the indexing queue table into multiple segments, each chunk identified by a first and a last record of a respective segment.
27. The apparatus of claim 26 , wherein each chunk includes a source server identifier, an entity type, and a database partition number.
28. The apparatus of claim 26 , wherein the instructions for identifying the modifications further comprise instructions for
creating a pool of change detection threads to monitor writes to the local database tables.
29. The apparatus of claim 26 , further comprising instructions for
keeping a store of multiple chunks to serve to remote instances.
30. The apparatus of claim 29 , further comprising instructions for
merging data from multiple chunks into a single chunk.
31. The apparatus of claim 25 , wherein the instructions for serving the chunk further comprise instructions for
processing an HTTP LIST function to indicating available chunks to the other instance; and
processing an HTTP GET function to return underlying data for a specific chunk.
32. The apparatus of claim 25 , wherein the instance includes a modification timestamp indicating a most recent update of the instance, and wherein the instructions for requesting the underlying data further comprise instructions for
updating the modification timestamp with a modification timestamp of the received chunk.
33. The apparatus of claim 25 , wherein the instructions for receiving the chunk further comprise instructions for
identifying a time-based trigger to update the instance; and
requesting chunks from one or more remote instances,
wherein the instance requests underlying data and upserts changed records for each received chunk.
34. The apparatus of claim 25 , wherein the instructions for upserting the changed records further comprise instructions for
inserting new records into the local database tables for records that do not exist in the local database tables; and
merging changed data into the local database tables for records that do exist in the local database tables.
35. The apparatus of claim 34 , wherein the instructions for upserting the changed records further comprise instructions for
deleting rows from the local database tables marked as delete in the received chunk; and
persisting the deleted rows into a delete table.
36. The apparatus of claim 25 , wherein the instructions for upserting the changed records further comprise instructions for
determining if a modification timestamp of a changed row of data is more recent than a corresponding row in the local database tables; and
merging changed data into the local database tables when the modification timestamp is more recent; otherwise,
ignoring the changed data.