Method and system for parallel processing of database queries

• US 20040098359A1
• Filed: 11/14/2002
• Published: 05/20/2004
• Est. Priority Date: 11/14/2002
• Status: Active Grant

• Current Assignee: LexisNexis Risk Data Management, Inc. (RELX PLC)
• Original Assignee: Seisint, Inc.
• Inventors: Bayliss, David, Smith, Jake, Hicks, Nigel, Halliday, Gavin, Poulsen, Ole, Chapman, Richard
• Granted Patent: US 6,968,335 B2
• Time in Patent Office: Days
• US Class Current: 1/1
• CPC Class Codes: G06F 16/24532 of parallel queries
• CPC Class Codes: G06F 16/2471 Distributed queries
• CPC Class Codes: Y10S 707/99933 Query processing, i.e. searching
• CPC Class Codes: Y10S 707/99942 Manipulating data structure, e.g. compression, compaction, compilation
• CPC Class Codes: Y10S 707/99952 Coherency, e.g. same view to multiple users

Abstract

A system and methods for parallel processing of queries to one or more databases are described herein. One or more databases may be distributed among a subset of slave nodes of a global-results processing matrix. A query to the database may be generated using a query-based high-level programming language. The query-based source code then may be converted to intermediary source code in a common programming language and then compiled into a dynamic link library (DLL) or other type of executable. The DLL is then distributed among the slave nodes of the processing matrix, whereupon the slave nodes execute related portions of the DLL substantially in parallel to generate initial query results. The initial query results may then be provided to master node of the global-results processing matrix for additional processing, whereby the master node is adapted to execute one or more associated portions of the DLL on the initial query results.

Claims

1. A system for processing at least one query to a database, the system comprising:

a first processing matrix comprising;

a master node of a first type having a processor and receiving at least one executable compiled from a query; and

a first plurality of slave nodes of a first type each comprising a processor and means for storing data, each first type slave node storing a portion of the database, each first type slave node processor adapted to execute a first portion of the at least one executable on its stored database portion to generate query results, the plurality of first type slave nodes executing the first executable portion substantially in parallel to collectively generate a set of initial query results; and

a second processing matrix receiving the initial query results generated by the first processing matrix and comprising;

a second plurality of slave nodes of a second type each comprising a processor and means for storing a portion of the initial query results, each second type slave node processor adapted to execute a second portion of the at least one executable on its stored initial query results portion to generate a portion of intermediary query results, the plurality of second type slave nodes executing the second executable portion substantially in parallel to collectively generate a set of intermediary query results.

2. The system of claim 1, wherein the second processing matrix further comprises a master node of a second type comprising a processor and means for storing data, the second type master node receiving and storing the intermediary query results generated by the second plurality of slave nodes, the second type master node processor adapted to execute a third portion of the at least one executable on the intermediate query results to generate a resultant set of query results.

3. The system of claim 2, wherein the second plurality of slave nodes receives at least the second portion of the at least one executable from the second master node.

4. The system of claim 1, wherein the second processing matrix further comprises a master node of a second type comprising a processor and means for storing data, the second master node being adapted to receive at least one other executable compiled from an other query, wherein at least a portion of the at least one other executable is executed on the intermediary query results to generate a resultant set of query results.

5. The system of claim 4, wherein the at least one other executable portion is executed by the second master node.

6. The system of claim 4, wherein the at least one other executable portion is executed by the second plurality of slave nodes.

7. The system of claim 4, wherein the other query is based on a user query submission and the system generates and compiles the other query into the at least one other executable, executes the at least one other executable, generates the resultant set of query results, and presents the resultant set of query results in an essentially real-time fashion.

8. The system of claim 7, wherein the resultant set of results is presented to the user having made the query submission.

9. The system of claim 1, wherein each second type slave node stores a substantially different portion of the initial query results and the second plurality of slave nodes collectively stores substantially all of the initial query results.

10. The system of claim 1, further comprising a query server operably connected to at least one of the first processing matrix and the second processing matrix, the query server being adapted to:

generate intermediary source code from query source code, the query source code representing at least one database operation on the database and wherein the query source code is formatted based in part on a query-based programming language; and

compile the intermediary source code to generate the at least one executable.

11. The system of claim 10, wherein the intermediary source code is one of a group consisting of:

C source code, C++ source code, Fortran source code; and

Pascal source code.

12. The system of claim 10, further comprising a query agent operably connected to the query server and at least one of the first processing matrix and the second processing matrix, the query agent being adapted to:

submit the at least one executable to at least one of the first processing matrix and the second processing matrix for execution; and

monitor execution of the at least one executable.

13. The system of claim 12, further comprising a scheduling service module operably connected to the query agent and the query server and being adapted to:

schedule a time of execution of the at least one executable based on at least one characteristic of the at least one executable; and

direct the query agent to submit the executable to at least one of the first processing matrix and the second processing matrix at the scheduled time.

14. The system of claim 13, wherein the at least one characteristic of the at least one executable includes one of a group consisting of:

an expected execution time of the at least one executable;

a priority of the at least one executable;

a type of the at least one database operation represented by the at least one executable; and

a number of slave nodes expected to execute the at least one executable.

15. The system of claim 12, further comprising a work-unit reporting module operably connected to the query agent and being adapted to store at least a portion of the resultant query results.

16. The system of claim 10, further comprising a query builder module operably connected to the query server and being adapted to facilitate a client in generating the query source code using the query-based language.

17. The system of claim 16, wherein the query builder module includes a graphical user interface.

18. The system of claim 16, wherein the query builder module is further adapted to display at least a portion of the resultant query results.

19. The system of claim 16, wherein the query builder module is further adapted to submit a representation of the query source code to the query server over a network.

20. The system of claim 1, further comprising a staging zone adapted to distribute the database among the first plurality of slave nodes, each first type slave node receiving a substantially different database portion.

21. The system of claim 20, wherein data of the database is distributed randomly among the first type slave nodes.

22. The system of claim 20, wherein the staging zone is further adapted to distribute the database using a hash function on a record field of each record of the database.

23. The system of claim 1, wherein the at least one executable includes at least one dynamic link library (DLL).

24. The system of claim 23, wherein each slave node of the first processing matrix further includes a software agent being adapted to execute a portion of the at least one DLL, wherein a software agent at the first master node is configured to execute a second portion of the DLL

25. The system of claim 23, wherein each slave node of the second processing matrix further includes a software agent being adapted to execute at least one portion of the at least one DLL.

26. The system of claim 1, wherein the first slave nodes are substantially homogeneous.

27. The system of claim 1, wherein the second slave nodes are substantially homogeneous.

28. The system of claim 1, wherein the first type slave nodes and the second type slave nodes are substantially homogeneous.

29. The system of claim 1, wherein the first type slave nodes, the first master node, the second type slave nodes, and the second master node are substantially homogeneous.

30. The system of claim 1, wherein the processors of one or more of the second master node, second type slave nodes, first master node, and first type slave nodes are of the type found in general purpose, single-user microcomputers.

31. The system of claim 1, wherein the at least one database operation represented by the first portion of the at least one executable includes one of a group consisting of:

linking, matching, filtering, scoring, sorting, counting, and joining.

32. The system of claim 1, wherein the at least one database operation represented by the second portion of the at least one executable includes one of a group consisting of:

sorting, linking, collating, filtering, counting, deduping, joining, appending, merging, purging, non-hierarchical linking, neural-net scoring, and formatting.

33. The system of claim 1, wherein the plurality of first type slave nodes includes between ten slave nodes and one thousand slave nodes.

34. The system of claim 1, wherein the plurality of first type slave nodes includes between fifty slave nodes and 500 slave nodes.

35. The system of claim 1, wherein the plurality of first type slave nodes includes between 100 slave nodes and 300 slave nodes.

36. The system of claim 1, wherein the plurality of first type slave nodes includes at least one thousand slave nodes.

37. The system of claim 1, wherein each first type slave node stores a substantially different portion of the database and the first plurality of slave nodes collectively stores substantially all of the database.

38. The system of claim 1, wherein the at least one executable represents at least one database operation.

39. The system of claim 38, wherein the first and second portions of the at least one executable both represent at least one database operation.

40. The system of claim 1, wherein the first type slave node means for storing includes memory and at least one disk storage.

41. The system of claim 40, wherein the first type slave node stores the portion of the database at least in memory.

42. The system of claim 1, wherein the second type slave node means for storing includes memory and at least one disk storage.

43. The system of claim 42, wherein the second type slave node stores the portion of the initial query results at least in disk storage.

44. A system for processing at least one query to a database, the system comprising:

a general-purpose (GP) query processing matrix comprising;

a GP master node having a processor and receiving at least one executable compiled from a query; and

a plurality of GP slave nodes each comprising a processor and means for storing data, each GP slave node storing a portion of the database, each GP slave node processor adapted to execute a first portion of the at least one executable on its stored database portion to generate query results, the plurality of GP slave nodes executing the first executable portion substantially in parallel to collectively generate a set of initial query results; and

a global-results (GR) processing matrix receiving the initial query results generated by the general-purpose processing matrix and comprising;

a plurality of GR slave nodes each comprising a processor and means for storing a portion of the initial query results, each GR slave node processor adapted to execute a second portion of the at least one executable on its stored initial query results portion to generate a portion of intermediary query results, the plurality of GR slave nodes executing the second executable portion substantially in parallel to collectively generate a set of intermediary query results; and

a GR master node comprising a processor and means for storing data.

45. The system of claim 44, wherein the GR master node receives and stores the intermediary query results generated by the plurality of GR slave nodes, the second type master node processor adapted to execute a third portion of the at least one executable on the intermediate query results to generate a resultant set of query results.

46. The system of claim 44, wherein the second plurality of slave nodes receives at least the second portion of the at least one executable from the second master node.

47. The system of claim 44, wherein the GR master node is adapted to receive at least one other executable compiled from an other query, wherein at least a portion of the at least one other executable is executed on the intermediary query results to generate a resultant set of query results.

48. The system of claim 47, wherein the at least one other executable portion is executed by the GR master node.

49. The system of claim 47, wherein the at least one other executable portion is executed by the plurality of GR slave nodes.

50. The system of claim 47, wherein the other query is based on a user query submission and the system generates and compiles the other query into the at least one other executable, executes the at least one other executable, generates the resultant set of query results, and presents the resultant set of query results in an essentially real-time fashion.

51. The system of claim 50, wherein the resultant set of results are presented to the user having made the user query submission.

52. The system of claim 44, wherein each second type slave node stores a substantially different portion of the initial query results and the second plurality of slave nodes collectively stores substantially all of the initial query results.

53. The system of claim 44, further comprising a query server operably connected to at least one of the GR processing matrix and the GP processing matrix, the query server being adapted to:

generate intermediary source code from query source code, the query source code representing at least one database operation on the database and wherein the query source code is formatted based in part on a query-based programming language; and

compile the intermediary source code to generate the at least one executable.

54. The system of claim 53, wherein the intermediary source code is one of a group consisting of:

C source code, C++ source code, Fortran source code; and

Pascal source code.

55. The system of claim 53, further comprising a query agent operably connected to the query server and at least one of the GR processing matrix and the GP processing matrix, the query agent being adapted to:

submit the at least one executable to at least one of the GR processing matrix and the GP processing matrix for execution; and

monitor execution of the at least one executable.

56. The system of claim 55, further comprising a scheduling service module operably connected to the query agent and the query server and being adapted to:

schedule a time of execution of the at least one executable based on at least one characteristic of the at least one executable; and

direct the query agent to submit the executable to at least one of the GR processing matrix and the GP processing matrix at the scheduled time.

57. The system of claim 56, wherein the at least one characteristic of the at least one executable includes one of a group consisting of:

an expected execution time of the at least one executable;

a priority of the at least one executable;

a type of the at least one database operation represented by the at least one executable; and

a number of slave nodes expected to execute the at least one executable.

58. The system of claim 55, further comprising a work-unit reporting module operably connected to the query agent and being adapted to store at least a portion of the resultant query results.

59. The system of claim 53, further comprising a query builder module operably connected to the query server and being adapted to facilitate a client in generating the query source code using the query-based language.

60. The system of claim 59, wherein the query builder module includes a graphical user interface.

61. The system of claim 59, wherein the query builder module is further adapted to display at least a portion of the resultant query results.

62. The system of claim 59, wherein the query builder module is further adapted to submit a representation of the query source code to the query server over a network.

63. The system of claim 44, further comprising a staging zone adapted to distribute the database among the GP slave nodes, each GP slave node receiving a substantially different database portion.

64. The system of claim 63, wherein data of the database is distributed randomly among the GP slave nodes.

65. The system of claim 63, wherein the staging zone is further adapted to distribute the database using a hash function on a record field of each record of the database.

66. The system of claim 44, wherein the at least one executable includes at least one dynamic link library (DLL).

67. The system of claim 66, wherein each slave node of the GP processing matrix further includes a software agent being adapted to execute a portion of the at least one DLL, wherein a software agent at the GP master node is configured to execute a second portion of the DLL

68. The system of claim 66, wherein each slave node of the GR processing matrix further includes a software agent being adapted to execute at least one portion of the at least one DLL.

69. The system of claim 44, wherein the GP slave nodes are substantially homogeneous.

70. The system of claim 44, wherein the GR slave nodes are substantially homogeneous.

71. The system of claim 44, wherein the GP slave nodes and the GR slave nodes are substantially homogeneous.

72. The system of claim 44, wherein the GP slave nodes, the GP master node, the GR slave nodes, and the GR master node are substantially homogeneous.

73. The system of claim 44, wherein the processors of one or more of the GR master node, GR slave nodes, GP master node, and GP slave nodes are of the type found in general purpose, single-user microcomputers.

74. The system of claim 44, wherein the at least one database operation represented by the first portion of the at least one executable includes one of a group consisting of:

linking, matching, filtering, scoring, sorting, counting, and joining.

75. The system of claim 44, wherein the at least one database operation represented by the second portion of the at least one executable includes one of a group consisting of:

sorting, linking, collating, filtering, counting, deduping, joining, appending, merging, purging, non-hierarchical linking, neural-net scoring, and formatting.

76. The system of claim 44, wherein the plurality of GP slave nodes includes between ten slave nodes and one thousand slave nodes.

77. The system of claim 44, wherein the plurality of GP slave nodes includes between fifty slave nodes and 500 slave nodes.

78. The system of claim 44, wherein the plurality of GP slave nodes includes between 100 slave nodes and 300 slave nodes.

79. The system of claim 44, wherein the plurality of GP slave nodes includes at least one thousand slave nodes.

80. The system of claim 44, wherein each first type slave node stores a substantially different portion of the database and the first plurality of slave nodes collectively stores substantially all of the database.

81. The system of claim 44, wherein the at least one executable represents at least one database operation.

82. The system of claim 81, wherein the first and second portions of the at least one executable both represent at least one database operation.

83. The system of claim 44, wherein the first type slave node means for storing includes memory and at least one disk storage.

84. The system of claim 83, wherein the first type slave node stores the portion of the database at least in memory.

85. The system of claim 44, wherein the second type slave node means for storing includes memory and at least one disk storage.

86. The system of claim 85, wherein the second type slave node stores the portion of the initial query results at least in disk storage.

87. A system for processing at least one query to a database, the system comprising:

a general-purpose (GP) query processing matrix comprising;

a GP master node comprising a processor, memory, disk storage, and a network interface, the GP master node receiving at least one executable compiled from a query; and

a plurality of GP slave nodes each comprising a processor, memory, disk storage, and a network interface, each GP slave node storing a portion of the database, each GP slave node processor adapted to execute a first portion of the at least one executable using the stored database portion to generate query results, the first portion of the at least one executable representing at least one database operation, the GP slave node executing the first executable portion substantially in parallel with other GP slave nodes to collectively generate a set of initial query results; and

a global-results (GR) processing matrix receiving the initial query results generated by the general-purpose processing matrix and comprising;

a plurality of GR slave nodes each comprising a processor, memory, disk storage, and a network interface, each GR slave node storing a portion of the initial query results, each GR slave node processor adapted to execute a second portion of the at least one executable on the stored initial query results portion to generate a portion of intermediary query results, the second portion of the at least one executable representing at least one database operation, the GR slave node executing the second executable portion substantially in parallel with other GR slave nodes to collectively generate a set of intermediary query results; and

a GR master node comprising a processor, memory, disk storage, and a network interface, the GR master node receiving and storing the intermediary query results generated by the GR slave nodes, the GR master node processor adapted to execute a third portion of the at least one executable on the intermediate query results to generate a refined set of query results, the third portion of the at least one executable representing at least one database operation.

88. The system of claim 87, wherein the GP slave node stores the portion of the database at least in memory.

89. The system of claim 87, wherein the GR slave node stores the portion of the initial query results at least in disk storage.

90. The system of claim 87, wherein the GR master node stores the intermediary query results at least in disk storage.

91. A system for processing at least one query to a database, the system comprising:

a query server for receiving and processing a query;

a general-purpose (GP) query processing matrix having a plurality of processing nodes comprising;

a GP master node having a processor and at least one data storage device, the GP master node being adapted to receive at least one executable compiled from a query;

a plurality of GP slave nodes operably connected to the GP master node, wherein each GP slave node is adapted to;

store a substantially different portion of the database in memory at the slave node; and

execute a first portion of at least one executable using the stored database portion to generate query results, the first portion of the at least one executable representing at least one database operation to be performed on the stored database portion; and

wherein the plurality of GP slave nodes collectively generates a set of initial query results; and

a global-results (GR) processing matrix operably connected to the query server and including;

a plurality of GR slave nodes, each GR slave node being adapted to;

store a substantially different portion of the initial query results on disk storage of the slave node; and

execute a second portion of the at least one executable using the stored initial query results portion to generate a portion of intermediary query results, wherein the plurality of GR slave nodes execute the second portion substantially in parallel to generate a set of intermediary results; and

a master node having a processor and at least one storage device and being operably connected to the plurality of slave nodes.

92. The system of claim 91, wherein the GR master node receives and stores the intermediary query results generated by the plurality of GR slave nodes, the GR master node processor being adapted to execute a third portion of the at least one executable on the intermediate query results to generate a resultant set of query results.

93. The system of claim 92, wherein the plurality of GR slave nodes receives at least the second portion of the at least one executable from the GR master node.

94. The system of claim 91, wherein the GR master node is adapted to receive at least one other executable compiled from an other query, wherein at least a portion of the at least one other executable is executed on the intermediary query results to generate a resultant set of query results.

95. The system of claim 94, wherein the at least one other executable portion is executed by the GR master node.

96. The system of claim 94, wherein the at least one other executable portion is executed by the plurality of GR slave nodes.

97. The system of claim 94, wherein the other query is based on a user query submission and the system generates and compiles the other query into the at least one other executable, executes the at least one other executable, generates the resultant set of query results, and presents the resultant set of query results in an essentially real-time fashion.

98. The system of claim 97, wherein the resultant set of results is presented to the user having made the query submission.

99. A system for processing at least one query to at least one database, the system comprising:

a query server being adapted to;

generate intermediary source code from query source code, the query source code representing least one database operation using the database and wherein the query source code is formatted based in part on a query-based programming language; and

compile the intermediary source code to generate at least one executable;

a general-purpose query processing matrix operably connected to the query server and including;

a plurality of slave nodes, each slave node storing in memory a different portion of the database and being adapted to execute a first portion of the at least one executable using the stored database portion to generate a portion of initial query results, the first portion of the at least one executable representing at least one database operation on the stored database portion, the first portion being executed by the slave node substantially in parallel with an execution of the first portion by other slave nodes;

at least one level of collator nodes, each collator node of each level storing in memory query results from a lower level;

a global-results processing matrix including;

a plurality of slave nodes, each slave node storing on disk storage a different portion of the database and being adapted to execute a first portion of the at least one executable using the stored database portion to generate a portion of initial query results, the first portion of the at least one executable representing at least one database operation on the stored database portion, the first portion being executed by the slave node substantially in parallel with an execution of the first portion by other slave nodes; and

a master node operably connected to the plurality of slave nodes and being adapted to;

store the initial query results on the disk storage of the master node; and

execute a second portion of the at least one executable on the stored initial query results to generate final query results, the second portion of the at least one executable representing at least one database operation on the initial query results.

100. A method for processing at least one query to a database distributed among a plurality of slave nodes, each slave node storing a substantially distinct database portion on disk storage, the method comprising the steps of:

receiving a query in a query-based language source code and compiling at least one executable from the query source code;

executing, at each slave node, a first portion of the at least one executable using the stored database portion to generate a portion of initial query results, the first portion of the at least one executable representing at least one database operation on the stored database portion, the first portion being executed by the slave node substantially in parallel with an execution of the first portion by other slave nodes;

storing the initial query results to disk storage of a master node; and

executing, at the master node, a second portion of the at least one executable using the stored initial query results to generate resultant query results, the second portion of the at least one executable representing at least one database operation on the initial query results.

101. The method of claim 100, further comprising the steps of:

generating intermediary source code from query source code, the query source code representing least one database operation using the database and wherein the query source code is formatted based in part on a query-based programming language; and

compiling the intermediary source code to generate the at least one executable.

102. In a global-results processing matrix including a master node and a plurality of slave nodes, each of the master node and slave nodes including a processor and disk storage, a computer readable medium, the computer readable medium comprising:

a first set of executable instructions being adapted to manipulate the processor of each slave node to execute a first portion of the at least one executable using the a portion of a database stored at the disk storage of the slave node to generate a portion of initial query results, the first portion of the at least one executable representing at least one database operation on the stored database portion, the first portion being executed by the slave node substantially in parallel with an execution of the first portion by other slave nodes; and

a second set of executable instructions being adapted to manipulate the processor of the master node to execute a second portion of the at least one executable using the initial query results to generate final query results, the second portion of the at least one executable representing at least one database operation using the initial query results.

103. The computer readable medium of claim 102, further comprising a third set of executable instructions being adapted to manipulate a processor to:

generate intermediary source code from query source code, the query source code representing least one database operation using the database and wherein the query source code is formatted based in part on a query-based programming language;

compile the intermediary source code to generate the at least one executable; and

provide the at least one executable to the master node.

104. The computer readable medium of claim 102, wherein the at least one executable includes at least one dynamic link library (DLL).