MULTI-MASTER DATA REPLICATION IN A DISTRIBUTED MULTI-TENANT SYSTEM

• US 20120259894A1
• Filed: 10/04/2011
• Published: 10/11/2012
• Est. Priority Date: 04/11/2011
• Status: Active Grant

• Current Assignee: Salesforce.com, Inc.
• Original Assignee: Salesforce.com, Inc.
• Inventors: Hansma, Scott, Burstein, Paul, VARLEY, Ian
• Granted Patent: US 9,396,242 B2
• Time in Patent Office: Days
• US Class Current: 707/795
• CPC Class Codes: G06F 16/2322 using timestamps
• CPC Class Codes: G06F 16/2379 Updates performed during online database operations; commit processing
• 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/273 Asynchronous replication or reconciliation
• CPC Class Codes: G06F 16/275 Synchronous replication
• CPC Class Codes: G06F 16/93 Document management systems
• CPC Class Codes: G06F 16/9535 Search customisation based on user profiles and personalisation
• CPC Class Codes: G06F 16/954 Navigation, e.g. using categorised browsing

Abstract

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.

Claims

1. A method, comprising:

a plurality of instances, each being separately responsible for a plurality of database tables, regularly replicating updates to the database tables using a plurality of chunks;

thereby consistently

detecting, serving, requesting, and upsetting changes to the database tables.

2. The method of claim 1, wherein the detecting comprises:

creating a pool of change detection threads;

when a row in a replicated database table is modified, writing a trigger into an indexing queue table;

watching the indexing queue table and pulling out aggregates of changed rows and breaking them up into the chunks;

creating chunk definitions by looking at the first and last record in a selected chunk; and

persisting the chunk definitions to the database.

3. The method of claim 1, wherein the serving comprises:

utilizing an HTTP LIST function for returning a set of the available chunks; and

utilizing an HTTP GET function for returning the underlying data for exactly one of the chunks obtained by the HTTP LIST function.

4. The method of claim 1, wherein the requesting comprises:

wherein each instance has its own up to date marker showing how recently that instance has been updated, and further wherein each chunk has metadata comprising a modstamp;

a first instance requesting data from database tables belonging to a second instance by;

checking an instance configuration table to determine whether the second instance is a leader or not, and whether the second instance is reachable or not;

requesting chunks belonging to the second instance either directly from the second instance or from a leader responsible for servicing requests made of the second instance;

upserting the requested chunks; and

determining whether all requested chunks were successfully upserted, and if so, advancing an up to date marker of the first instance to the modstamp of the most recent chunk.

5. The method of claim 1, wherein the upserting comprises:

passing a blob (binary large object) of records into a PL/SQL procedure;

checking that a modstamp of a new row is strictly later than the existing modstamp;

otherwise, ignoring the new row;

running a DELETE statement for delete rows; and

persisting delete rows into a delete table.

6. The method of claim 5, wherein the upserting comprises:

using two core PL/SQL statements, comprising;

MERGE for the inserts and updates, and

DELETE for the deletes.

7. The method of claim 1, wherein the metadata for each of the chunks further comprises:

a source server ID, an entity type, and a database partition number.

8. The method of claim 1, further comprising:

regularly reviewing all chunks for any outstanding non-upserted chunks; and

after a predetermined time period, compacting the non-upserted chunks into catchup chunks.

9. The method of claim 8, wherein catchup chunks further comprise:

being marked with a special bit showing a status of being catchup chunks;

being not cached in binary form, either on a client or a server; and

can only come directly from a source instance and not from a leader.

10. A machine-readable medium carrying one or more sequences of instructions for implementing a method for providing an interface for object relationships, comprising:

a plurality of instances, each being separately responsible for a plurality of database tables, regularly replicating updates to the database tables using a plurality of chunks;

thereby consistently

detecting, serving, requesting, and upserting changes to the database tables.

11. A method for performing replication in a multi-tenant database system, the method comprising:

a plurality of instances, each being separately responsible for a plurality of database tables, regularly replicating updates to the database tables using a plurality of chunks;

thereby consistently

detecting, serving, requesting, and upserting changes to the database tables.

12. A method for idempotent application of modifications to a multi-tenant database system, comprising:

a plurality of instances, each being separately responsible for a plurality of database tables, regularly replicating updates to the database tables using a plurality of chunks;

regularly checking timestamps of database records on a record-by-record basis, thereby making the application of modified data to be idempotent and tolerant of reordering; and

using markers for deleted data so that the deleted data can be applied to the database records in a same stream as insert data or update data.

13. The method of claim 12, further comprising:

maintaining a map of instances and leaders within each of a plurality of data centers via XML files that are created and maintained as part of a source code configuration tree; and

using the map in deciding which instance to ask for specific data.

14. The method of claim 12, wherein each instance contains pointers to all other instances, including a unique one-character instance ID, a URL, and other information.

15. The method of claim 13, wherein the step of regularly replicating includes synchronizing the XML configuration information into a single instance configuration table, thereby

maintaining a consistent view of instances and leaders information across the entire multi-tenant database system.

16. The method of claim 15, further comprising:

routing all database updates into the instance configuration table, so that every actor, whether client or server, shares a same view of a current state of the instances and leaders map.

17. The method of claim 13, further comprising:

dividing the instances into leaders and non-leaders, wherein at any given time there is exactly one leader within any one of the plurality of data centers.

18. The method of claim 17, further comprising:

a leader instance running a DELETE statement for deleting rows and also persisting these statements into its own delete table; and

a request to perform a delete firing a trigger that writes the deleted key to a separate delete record/table, so that the delete markers are then treated the same as inserts and updates.

19. The method of claim 15, further comprising:

wherein the source of data for the instance configuration table is the information in the XML files.

20. The method of claim 12, further comprising:

utilizing two HTTP operations for serving changed data;

LIST, which returns a set of the available chunks, by entity, date, or other criteria; and

GET, which returns the underlying data for exactly one chunk, by a unique chunk ID; and

initiating the LIST operation on multiple separate chunks each having a variety of records, resulting in a variety of chunks each identified by the unique chunk ID.

21. The method of claim 12, wherein the step of regularly replicating comprises detecting, serving, requesting, and upserting changes to the database tables.

22. The method of claim 21, wherein the requesting comprises:

wherein each instance has its own up to date marker showing how recently that instance has been updated, and further wherein each chunk has metadata comprising a modstamp;

a first instance requesting data from database tables belonging to a second instance by;

checking the instance configuration table to determine whether the second instance is a leader or not, and whether the second instance is reachable or not;

requesting chunks belonging to the second instance either directly from the second instance or from a leader responsible for servicing requests made of the second instance;

upserting the requested chunks; and

determining whether all requested chunks were successfully upserted, and if so, advancing an up to date marker of the first instance to the modstamp of the most recent chunk.

23. A method of achieving dynamic hierarchical arrangement of nodes for optimal geographic data distribution, comprising:

dividing instances into leaders and non-leaders within a particular data center;

dynamically re-adjusting designation of leaders and non-leaders based at least upon load-size and reachability including but not limited to system failures or planned outages; and

steering instance data to traverse multiple hops via leaders, so as to ensure that each database update only traverses a data center link a minimum number of times.

24. The method of claim 23, further comprising:

within any specific data center, one instance acting as a leader and all other non-leader instances in the data center communicating with each other and with the leader but not to any other instances outside the data center; and

leaders talking to the other instances within their specific data center and also talking to remote leaders who provide data for all the instances within that leader'"'"'s own particular data center.

25. The method of claim 24, further comprising:

selecting leaders by looking at a leader ID which is a configured value for each known instance and is stored within instance XML configuration files, and

picking the lowest-numbered reachable instance in each data center to be the leader.

26. The method of claim 25, further comprising:

in the event of a tie where two instances have the same number, breaking the tie by referring to the instance name in ascending order, thereby

selecting leaders in order of desirability based on their load.

27. The method of claim 24, further comprising:

one or more instances acting as a proxy for changes made elsewhere than that instance, particularly but not exclusively if that instance us acting as a leader.

28. A method of achieving efficiency within a database environment, comprising:

a plurality of instances, each being separately responsible for a plurality of database tables, regularly replicating updates to the database tables using a plurality of chunks;

regularly reviewing all chunks for any outstanding non-upserted chunks;

after a predetermined time period, compacting the non-upserted chunks into catchup chunks;

marking the catchup chunks with a special bit to signify a status of being a catchup chunk; and

during a process of caching catchup chunks, performing no binary caching, either on a client or a server.

29. The method of claim 28, wherein a catchup chunk originates only from a source instance and not from a leader instance.

30. The method of claim 28, wherein a chunk is either cached or recalculated, according to which is more efficient.

31. The method of claim 28, wherein transport logic is ignorant of what data is contained within a particular chunk.

32. The method of claim 28, further comprising:

arranging that chunks are specific to a single instance so that no overlap exists between chunks from difference instances.

33. The method of claim 28, further comprising:

providing a ShouldLookForLostChunks flag, wherein if set to “

true”

, each iteration of the client attempting to fetch and upsert the most recent chunks returned by a LIST command, but also any prior active chunks that have yet to be upserted.

34. The method of claim 28, wherein said regularly replicating comprises:

detecting, serving, requesting, and upserting changes to the database tables.

35. The method of claim 28, wherein the requesting comprises:

wherein each instance has its own up to date marker showing how recently that instance has been updated, and further wherein each chunk has metadata comprising a modstamp;

a first instance requesting data from database tables belonging to a second instance by;

checking an instance configuration table to determine whether the second instance is a leader or not, and whether the second instance is reachable or not;

requesting chunks belonging to the second instance either directly from the second instance or from a leader responsible for servicing requests made of the second instance;

upserting the requested chunks; and

determining whether all requested chunks were successfully upserted, and if so, advancing an up to date marker of the first instance to the modstamp of the most recent chunk.

36. A method of controlling a batch size within a relational database using a combination of date and unique ID endpoints, comprising:

packaging a plurality of updates to database records into chunks using timestamps;

checking all database records having the same timestamp to determine whether the total amount of records exceeds a predetermined maximum batch size;

re-checking the records both for having the same timestamp and for an entity ID; and

allowing only those records having the same timestamp and the same entity ID to be packaged into a particular chunk.

37. A machine-readable medium carrying one or more sequences of instructions for idempotent application of modifications to a multi-tenant database system in a multi-tenant database system, which instructions, when executed by one or more processors, cause the one or more processors to carry out the steps of:

a plurality of instances, each being separately responsible for a plurality of database tables, regularly replicating updates to the database tables using a plurality of chunks;

regularly checking timestamps of database records on a record-by-record basis, thereby making the application of modified data to be idempotent and tolerant of reordering; and

using markers for deleted data so that the deleted data can be applied to the database records in a same stream as insert data or update data.

38. The machine-readable medium of claim 37, further comprising:

maintaining a map of instances and leaders within each of a plurality of data centers via XML files that are created and maintained as part of a source code configuration tree; and

using the map in deciding which instance to ask for specific data.

39. The machine-readable medium of claim 37, wherein each instance contains pointers to all other instances, including a unique one-character instance ID, a URL, and other information.

40. The machine-readable medium of claim 37, wherein the step of regularly replicating includes synchronizing the XML configuration information into a single instance configuration table, thereby

maintaining a consistent view of instances and leaders information across the entire multi-tenant database system.

41. The machine-readable medium of claim 40, further comprising:

routing all database updates into the instance configuration table, so that every actor, whether client or server, shares a same view of a current state of the instances and leaders map.

42. The machine-readable medium of claim 38, further comprising:

dividing the instances into leaders and non-leaders, wherein at any given time there is exactly one leader within any one of the plurality of data centers.

43. The machine-readable medium of claim 42, further comprising:

a leader instance running a DELETE statement for deleting rows and also persisting these statements into its own delete table; and

a request to perform a delete firing a trigger that writes the deleted key to a separate delete record/table, so that the delete markers are then treated the same as inserts and updates.

44. The machine-readable medium of claim 40, further comprising:

wherein the source of data for the instance configuration table is the information in the XML files.

45. The machine-readable medium of claim 37, further comprising:

utilizing two HTTP operations for serving changed data;

LIST, which returns a set of the available chunks, by entity, date, or other criteria; and

GET, which returns the underlying data for exactly one chunk, by a unique chunk ID; and

initiating the LIST operation on multiple separate chunks each having a variety of records, resulting in a variety of chunks each identified by the unique chunk ID.

46. The machine-readable medium of claim 37, wherein the step of regularly replicating comprises detecting, serving, requesting, and upserting changes to the database tables.

47. The machine-readable medium of claim 46, wherein the requesting comprises:

wherein each instance has its own up to date marker showing how recently that instance has been updated, and further wherein each chunk has metadata comprising a modstamp;

a first instance requesting data from database tables belonging to a second instance by;

checking the instance configuration table to determine whether the second instance is a leader or not, and whether the second instance is reachable or not;

requesting chunks belonging to the second instance either directly from the second instance or from a leader responsible for servicing requests made of the second instance;

upserting the requested chunks; and

determining whether all requested chunks were successfully upserted, and if so, advancing an up to date marker of the first instance to the modstamp of the most recent chunk.

48. A machine-readable medium carrying one or more sequences of instructions for achieving dynamic hierarchical arrangement of nodes for optimal geographic data distribution in a multi-tenant database system, which instructions, when executed by one or more processors, cause the one or more processors to carry out the steps of:

dividing instances into leaders and non-leaders within a particular data center;

dynamically re-adjusting designation of leaders and non-leaders based at least upon load-size and reachability including but not limited to system failures or planned outages; and

steering instance data to traverse multiple hops via leaders, so as to ensure that each database update only traverses a data center link a minimum number of times.

49. The machine-readable medium of claim 48, further comprising:

within any specific data center, one instance acting as a leader and all other non-leader instances in the data center communicating with each other and with the leader but not to any other instances outside the data center; and

leaders talking to the other instances within their specific data center and also talking to remote leaders who provide data for all the instances within that leader'"'"'s own particular data center.

50. The machine-readable medium of claim 49, further comprising:

selecting leaders by looking at a leader ID which is a configured value for each known instance and is stored within instance XML configuration files, and

picking the lowest-numbered reachable instance in each data center to be the leader.

51. The machine-readable medium of claim 50, further comprising:

in the event of a tie where two instances have the same number, breaking the tie by referring to the instance name in ascending order, thereby

selecting leaders in order of desirability based on their load.

52. The machine-readable medium of claim 49, further comprising:

one or more instances acting as a proxy for changes made elsewhere than that instance, particularly but not exclusively if that instance us acting as a leader.

53. A machine-readable medium carrying one or more sequences of instructions for achieving efficiency within a multi-tenant database system, which instructions, when executed by one or more processors, cause the one or more processors to carry out the steps of:

a plurality of instances, each being separately responsible for a plurality of database tables, regularly replicating updates to the database tables using a plurality of chunks;

regularly reviewing all chunks for any outstanding non-upserted chunks;

after a predetermined time period, compacting the non-upserted chunks into catchup chunks;

marking the catchup chunks with a special bit to signify a status of being a catchup chunk; and

during a process of caching catchup chunks, performing no binary caching, either on a client or a server.

54. The machine-readable medium of claim 53, wherein a catchup chunk originates only from a source instance and not from a leader instance.

55. The machine-readable medium of claim 53, wherein a chunk is either cached or recalculated, according to which is more efficient.

56. The machine-readable medium of claim 53, wherein transport logic is ignorant of what data is contained within a particular chunk.

57. The machine-readable medium of claim 53, further comprising:

arranging that chunks are specific to a single instance so that no overlap exists between chunks from difference instances.

58. The machine-readable medium of claim 53, further comprising:

providing a ShouldLookForLostChunks flag, wherein if set to “

true”

, each iteration of the client attempting to fetch and upsert the most recent chunks returned by a LIST command, but also any prior active chunks that have yet to be upserted.

59. The machine-readable medium of claim 53, wherein said regularly replicating comprises:

detecting, serving, requesting, and upserting changes to the database tables.

60. The machine-readable medium of claim 53, wherein the requesting comprises:

wherein each instance has its own up to date marker showing how recently that instance has been updated, and further wherein each chunk has metadata comprising a modstamp;

a first instance requesting data from database tables belonging to a second instance by;

checking an instance configuration table to determine whether the second instance is a leader or not, and whether the second instance is reachable or not;

requesting chunks belonging to the second instance either directly from the second instance or from a leader responsible for servicing requests made of the second instance;

upserting the requested chunks; and

determining whether all requested chunks were successfully upserted, and if so, advancing an up to date marker of the first instance to the modstamp of the most recent chunk.

61. A machine-readable medium carrying one or more sequences of instructions for controlling a batch size within a relational database using a combination of date and unique ID endpoints in a multi-tenant database system, which instructions, when executed by one or more processors, cause the one or more processors to carry out the steps of:

packaging a plurality of updates to database records into chunks using timestamps;

checking all database records having the same timestamp to determine whether the total amount of records exceeds a predetermined maximum batch size;

re-checking the records both for having the same timestamp and for an entity ID; and

allowing only those records having the same timestamp and the same entity ID to be packaged into a particular chunk.