Distributed database system and method having nodes co-ordinated in a decentralized manner

• US 7,480,658 B2
• Filed: 07/14/2004
• Issued: 01/20/2009
• Est. Priority Date: 07/16/2003
• Status: Expired due to Fees

• Current Assignee: Skype Limited (Microsoft Corporation)
• Original Assignee: Joltid Ltd.
• Inventors: Heinla, Ahti, Kasesalu, Priit
• Primary Examiner(s): Breene; John E
• Assistant Examiner(s): LE, THU NGUYET T
• Application Number:
  US10/890,191
• Publication Number:
  US 20050108247A1
• Time in Patent Office: 1,651 Days
• Field of Search: 707/10, 709/243
• US Class Current: 1/1
• 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: H04L 67/1001 for accessing one among a plurality of replicated servers
• CPC Class Codes: H04L 67/1008 based on parameters of servers, e.g. available memory or workload monitoring of computer activity G06F11/30
• CPC Class Codes: H04L 67/101 based on network conditions
• CPC Class Codes: H04L 67/1019 Random or heuristic server selection
• CPC Class Codes: H04L 67/1095 Replication or mirroring of data, e.g. scheduling or transport for data synchronisation between network nodes
• CPC Class Codes: H04L 67/1097 for distributed storage of data in networks, e.g. transport arrangements for network file system [NFS], storage area networks [SAN] or network attached storage [NAS]
• CPC Class Codes: H04L 69/329 in the application layer [OSI layer 7]
• CPC Class Codes: H04L 9/40 Network security protocols

Abstract

There is provided a distributed database system comprising a plurality of participating nodes mutually interconnected via a communication network. The participating nods include a first subset of storage nodes and a second subset of interfacing nodes. The storage nodes are grouped into a plurality of slots, each storage node having stored therein addresses of substantially al other storage nodes in its particular slot. The storage nodes of each slot further have stored therein one or more addresses of storage nodes in all other of the slots of the system. The interfacing nodes each have stored therein addresses of one or more of the storage nodes. The storage nodes are operable to store one or more data records therein and/or retrieve one or more data records therefrom in response to instructions received from the interfacing nodes. The system is coordinated in operation in a decentralised manner.

Claims

1. A distributed database system comprising a, plurality of participating nodes and a communication network for supporting data transfer between the participating nodes, comprising:

the participating nodes comprise a first subset of storage nodes and a second subset of interfacing nodes,

the storage nodes each including a processor and a memory and are operable to store one or more data records therein and/or retrieve one or more data records therefrom in response to instructions received from the interfacing nodes,

the storage nodes are grouped into a plurality of slots, each storage node having stored therein addresses of substantially all other storage nodes in its particular slot,

the storage nodes of each slot further have stored therein one or more addresses of storage nodes in other of the slots of the system,

the interfacing nodes each including a processor and a memory and have stored therein addresses of one or more of the storage nodes,

the participating nodes being operable to be co-ordinated in a decentralized manner at a node level, wherein each of the participating nodes selects, on its own and without any input from other nodes, which of the storage nodes to output data thereto,

the processor of each of the interfacing nodes comprising;

(a) a first analyzing unit configured to analyze data received thereat and for subsequently determining one or more of the slots corresponding to the received data, the system being operable to store the received data as corresponding data records in one or more of the storage nodes of said one or more of the slots corresponding to the received data; and

(b) a second analyzing unit configured to analyze one or more search queries and for determining one or more appropriate corresponding slots, for retrieving data records corresponding to the one or more search queries from one or more of the storage nodes of said one or more corresponding slots and for outputting one or more of the data records in response to the one or more search queries.

2. A system according to claim 1, wherein less than 10% of the participating nodes are arranged to be storage nodes.

3. A system according to claim 1, wherein less than 2% of the participating nodes are arranged to be storage nodes.

4. A system according to claim 1, wherein, when storing one or more data records in one or more of the slots, one or more of the interfacing nodes are operable to store the one or more data records as multiple copies in a plurality of storage nodes in each of said one or more of the slots.

5. A system according to claim 4, wherein the plurality of storage nodes in which the multiple copies of the one or more data records are stored are randomly selected by said one or more of the interfacing nodes.

6. A system according to claim 4, wherein said one or more interfacing nodes are arranged to store said multiple copies of said one or more data records at substantially different time instances to at least partially counteract loss of data records occurring in the system during operation.

7. A system according to claim 6, wherein said one or more interfacing nodes are arranged to store said multiple copies of said one or more data records repetitively at a time interval similar to an average time to failure of the participating nodes, so as to maintain data storage integrity in the system even in the event of failure of a plurality of the participating nodes.

8. A system according to claim 7, wherein said one or more interfacing nodes are arranged to store said multiple copies of said one or more data records in storage nodes which are randomly selected within each of said one or more of the slots each time data record storage is performed.

9. A system according to claim 7, wherein the time interval is in a range of 12 hours to 72 hours.

10. A system according to claim 1, wherein each slot is arranged to each include in a range of 2 to 1000 storage nodes.

11. A system according to claim 1, wherein each slot is arranged to each include in a range of 5 to 20 storage nodes.

12. A system according to claim 1, wherein the first computing means is operable to locate addresses of substantially all storage nodes in a selected slot by:

(a) requesting at least one of the storage nodes at random within the system to send to the first computing means one or more other storage node addresses stored therein corresponding to the selected slot; and

(b) requesting at least one of said one or more other storage nodes to send addresses of substantially all other storage nodes within the selected slot.

13. A system according to claim 1, wherein the second computing means is operable to locate addresses of substantially all storage nodes in a selected slot by:

(a) requesting at least one of the storage nodes at random within the system to send to the second computing means one or more other storage node addresses stored therein corresponding to the selected slot; and

(b) requesting at least one of said one or more other storage nodes to send addresses of substantially all other storage nodes within the selected slot.

14. A system according to claim 1, wherein the first computing means is arranged to iterate over the data received thereat by applying a key=value transform wherein a parameter key corresponds to at least part of a word to be analysed and a parameter value corresponds to an equivalent numerical value for the parameter key.

15. A system according to claim 13, wherein a selection of the one or more appropriate corresponding slots to the parameter key is derived by applying a mathematical modulus function whose maximum value corresponds to the number of slots present within the system.

16. A system according to claim 14, wherein the key=value transform is arranged to employ at least one of case folding and word sub-portion selection.

17. A system according to claim 16, wherein word sub-portion selection relates to selecting a front portion of a word to be analysed, and wherein a slot number for the word to be analyzed is calculated as a hash value derived from the front portion of the word to be analyzed.

18. A system according to claim 16, wherein the length of the sub-portion selected for use is made dependent on an information content associated with the sub-portion.

19. A system according to claim 1, wherein the second computing means is arranged to apply one or more of a key=value transform and a key =<

phrase>

transform for determining one or more appropriate corresponding slot to said one or more search queries.

20. A system according to claim 19, wherein, when said one or more search queries are found to include an information content less than a predefined threshold, the second computing means is operable to send the one or more search query to one or more randomly selected storage nodes.

21. A system according to claim 19, wherein the second computing means is operable to send said one or more search queries to corresponding storage nodes in sequential batches.

22. A system according to claim 21, wherein the second computing means is operable to send said one or more search queries in batches, such that each batch is arranged to be received by in a range of 1 to 50 storage nodes.

23. A system according to claim 22, wherein each batch is received by 3 storage nodes.

24. A method of operating a distributed database system, the method comprising the steps of:

(a) arranging for the system to comprise a plurality of participating nodes and a communication network for supporting data transfer between the participating nodes, the participating nodes comprising a first subset of storage nodes and a second subset of interfacing nodes,

(b) grouping the storage nodes into a plurality of slots, each storage node having stored therein addresses of substantially all other storage nodes in its particular slot,

(c) arranging for the storage nodes of each slot further to have stored therein addresses of one or more storage nodes in other of the slots of the system,

(d) arranging for the interfacing nodes each to have stored therein addresses of one or more of the storage nodes,

(e) arranging for the storage nodes to store one or more data records therein and/or retrieve one or more data records therefrom in response to instructions received from the interfacing nodes,

(f) arranging for the system so that its interfacing nodes each comprise at least one of;

(i) analysing data received thereat and for subsequently determining one or more of the slots corresponding to the received data, the system being operable to store the received data as corresponding data records in one or more of the storage nodes of said one or more of the slots corresponding to the received data; and

(ii) analysing one or more search queries and for determining one or more appropriate corresponding slots, for retrieving data records corresponding to the one or more search queries from one or more of the storage nodes of said one or more corresponding slots and for outputting one or more of the data records in response to the one or more search queries, wherein the participating nodes are operable to be co-ordinated in a decentralized manner at a node level wherein each of the participating nodes selects, on its own and without any input from other nodes, which of the storage nodes to output data thereto.

25. A method according to claim 24, wherein less than 10% of the participating nodes are arranged to be storage nodes.

26. A method according to claim 24, wherein less than 2% of the participating nodes are arranged to be storage nodes.

27. A method according to claim 24, wherein, when storing one or more data records in one or more of the slots, one or more of the interfacing nodes are operable to store the one or more data records as multiple copies in a plurality of storage nodes in each of said one or more of the slots.

28. A method according to claim 27, wherein the plurality of storage nodes in which the multiple copies of the one or more data records are stored are randomly selected by said one or more of the interfacing nodes.

29. A method according to claim 27, wherein said one or more interfacing nodes are arranged to store said multiple copies of said one or more data records at substantially different time instances to at least partially counteract loss of data records occurring in the system during operation, so as to maintain data storage integrity in the system even in the event of failure of plurality of the participating nodes.

30. A method according to claim 29, wherein said one or more interfacing nodes are arranged to store said multiple copies of said one or more data records repetitively at a time interval similar to an average time to failure of the participating nodes.

31. A method according to claim 30, wherein said one or more interfacing nodes are arranged to store said multiple copies of said one or more data records in storage nodes which are randomly selected within each of said one or more of the slots each time data record storage is performed.

32. A method according to claim 30, wherein the time interval is in a range of 12 hours to 72 hours.

33. A method according to claim 24, wherein each slot is arranged to each include in a range of 2 to 1000 storage nodes.

34. A method according to claim 24, wherein each slot is arranged to each include in a range of 5 to 20 storage nodes.

35. A method according to claim 24, wherein the first computing means is operable to locate addresses of substantially all storage nodes in a selected slot by:

(a) requesting at least one of the storage nodes at random within the system to send to the first computing means other storage node addresses stored therein corresponding to the selected slot; and

(b) requesting at least one of said one or more other storage nodes to send addresses of substantially all other storage nodes within the selected slot.

36. A method according to claim 24, wherein the second computing means is operable to locate addresses of substantially all storage nodes in a selected slot by:

(a) requesting at least one of the storage nodes at random within the system to send to the second computing means other storage node addresses stored therein corresponding to the selected slot; and

(b) requesting at least one of said one or more other storage nodes to send addresses of substantially all other storage nodes within the selected slot.

37. A method according to claim 24, wherein the first computing means is arranged to iterate over the data received thereat by applying a key=value transform wherein a parameter key corresponds to at least part of a word to be analysed and a parameter value corresponds to an equivalent numerical value for the parameter key.

38. A method according to claim 37, wherein a selection of the one or more appropriate corresponding slots to the parameter key is derived by applying a mathematical modulus function whose maximum value corresponds to the number of slots present within the system.

39. A method according to claim 37, wherein the key=value transform is arranged to employ at least one of case folding and word sub-portion selection.

40. A method according to claim 39, wherein word sub-portion selection relates to selecting a front portion of a word to be analysed, and wherein a slot number for the word to be analyzed is calculated as a hash value derived from the front portion of the word to be analyzed.

41. A method according to claim 39, wherein the length of the sub-portion selected for use is made dependent on an information content associated with the sub-portion.

42. A method according to claim 24, wherein the second computing means is arranged to apply one or more of a key=value transform and a key =<

phrase>

transform for determining one or more appropriate corresponding slot to said one or more search queries.

43. A method according to claim 41, wherein, when said one or more search queries are found to include an information content less than a predefined threshold, the second computing means is operable to send the one or more search query to one or more randomly selected storage nodes.

44. A method according to claim 42, wherein the second computing means is operable to send said one or more search queries to corresponding storage nodes in sequential batches.

45. A method according to claim 44, wherein the second computing means is operable to send said one or more search queries in batches, such that each batch is arranged to be received by in a range of 1 to 50 storage nodes.

46. A method according to claim 44, wherein each batch is received by 3 storage nodes.

47. A data carrier having stored on it software implementing the method according to claim 24.