Dynamic asynchronous traversals for distributed graph queries
Techniques are described for enabling in-memory execution of any-sized graph data query by utilizing both depth first search (DFS) principles and breadth first search (BFS) principles to control the amount of memory used during query execution. Specifically, threads implementing a graph DBMS switch between a BFS mode of data traversal and a DFS mode of data traversal. For example, when a thread detects that there are less than a configurable threshold number of intermediate results in memory, the thread enters BFS-based traversal techniques to increase the number of intermediate results in memory. When the thread detects that there are at least the configurable threshold number of intermediate results in memory, the thread enters DFS mode to produce final results, which generally works to move the intermediate results that are currently available in memory to final query results, thereby reducing the number of intermediate results in memory.
1. A computer-executed method comprising:
executing, by a database server instance of a database management system, a query over graph data maintained in a graph database by the database management system;
wherein the graph data comprises a plurality of vertices and a plurality of edges that represent relationships between the plurality of vertices;
wherein the database server instance is implemented by a plurality of processing threads running on a computing device of the database management system;
wherein said executing the query over the graph data comprises:
a particular thread, of the plurality of processing threads, determining that there are less than a threshold number of intermediate results in one or more intermediate result queues in memory of the computing device;
in response to determining that there are less than the threshold number of intermediate results in the one or more intermediate result queues, the particular thread operating in an intermediate results production mode comprising:
the particular thread using breadth-first traversal techniques to produce one or more intermediate results for the query, and
the particular thread inserting the one or more intermediate results into at least one intermediate result queue of the one or more intermediate result queues; and
in response to determining that there are at least the threshold number of intermediate results in the one or more intermediate result queues and after inserting the one or more intermediate results into the at least one intermediate result queue, modifying executing of said query over the graph data from using the breadth-first traversal techniques in the intermediate results production mode to using depth-first traversal techniques to produce one or more results for the query in a final results production mode.
2. The computer-executed method of claim 1 , wherein:
the computing device of the database management system is a particular computing device;
the one or more intermediate result queues comprise one or both of:
a local intermediate result queue configured to store intermediate results produced by the plurality of processing threads, and
a remote intermediate result queue configured to store intermediate results produced by one or more computing devices, other than the particular computing device, of the database management system.
3. The computer-executed method of claim 1 , wherein:
the one or more intermediate result queues store message buffers;
the particular thread inserting the one or more intermediate results into the at least one intermediate result queue comprises:
the particular thread populating a message buffer with at least a portion of the one or more intermediate results, and
causing a particular intermediate result queue, of the one or more intermediate result queues, to include the message buffer.
4. The computer-executed method of claim 1 , wherein said executing the query over the graph data further comprises:
identifying a remote edge that is owned by a second computing device of the database management system;
in response to identifying the remote edge, including a particular intermediate result that is based, at least in part, on the remote edge in a particular message buffer;
after including the particular intermediate result in the particular message buffer, determining that the particular message buffer is full; and
in response to determining that the particular message buffer is full, sending the particular message buffer to the second computing device.
5. The computer-executed method of claim 4 , wherein:
the computing device of the database management system is a particular computing device;
the particular message buffer is registered with a network interface card (NIC) of the particular computing device;
sending the particular message buffer to the second computing device comprises:
producing communication structures directly from the particular message buffer, and
sending the communication structures, to the second computing device, over a network that communicatively couples the particular computing device and the second computing device.
6. The computer-executed method of claim 4 , wherein:
the computing device of the database management system is a particular computing device;
the method further comprises:
selecting the particular message buffer to store the particular intermediate result based, at least in part, on both (a) an owner of a destination vertex of the remote edge being the second computing device, and (b) a stage of query execution at which the remote edge was discovered;
wherein sending the particular message buffer to the second computing device is further based on determining that a number of unacknowledged message buffers, sent to the second computing device for the stage of query execution, does not exceed a limit on the number of unacknowledged message buffers.
7. The computer-executed method of claim 1 , further comprising:
identifying a plurality of query execution stages for the query;
wherein the plurality of query execution stages is ordered from earliest query execution stage to latest query execution stage;
wherein said executing the query over the graph data is performed based on the plurality of query execution stages;
wherein the particular thread using depth-first traversal techniques to produce the one or more results for the query comprises:
scanning intermediate results, in the one or more intermediate result queues, to identify one or more late intermediate results for one or more of the latest query execution stages, of the plurality of query execution stages;
producing the one or more results for the query based, at least in part, on using depth-first traversal techniques on the one or more late intermediate results.
8. The computer-executed method of claim 1 , further comprising:
identifying a plurality of query execution stages for the query;
wherein the plurality of query execution stages is ordered from earliest query execution stage to latest query execution stage;
wherein said executing the query over the graph data is performed based on the plurality of query execution stages;
wherein the particular thread using the breadth-first traversal techniques to produce the one or more intermediate results for the query comprises:
scanning intermediate results, in the one or more intermediate result queues, to identify one or more early intermediate results for one or more of earlier query execution stages, of the plurality of query execution stages;
producing the one or more intermediate results based, at least in part, on using the breadth-first traversal techniques on the one or more early intermediate results.
9. One or more non-transitory computer-readable media storing one or more sequences of instructions that, when executed by one or more processors, cause:
executing, by a database server instance of a database management system, a query over graph data maintained in a graph database by the database management system;
wherein the graph data comprises a plurality of vertices and a plurality of edges that represent relationships between the plurality of vertices;
wherein the database server instance is implemented by a plurality of processing threads running on a computing device of the database management system;
wherein said executing the query over the graph data comprises:
a particular thread, of the plurality of processing threads, determining that there are less than a threshold number of intermediate results in one or more intermediate result queues in memory of the computing device;
in response to determining that there are less than the threshold number of intermediate results in the one or more intermediate result queues, the particular thread operating in an intermediate results production mode comprising:
the particular thread using breadth-first traversal techniques to produce one or more intermediate results for the query, and
the particular thread inserting the one or more intermediate results into at least one intermediate result queue of the one or more intermediate result queues; and
in response to determining that there are at least the threshold number of intermediate results in the one or more intermediate result queues and after inserting the one or more intermediate results into the at least one intermediate result queue, modifying executing of said query over the graph data from using the breadth-first traversal techniques in the intermediate results production mode to using depth-first traversal techniques to produce one or more results for the query in a final results production mode.
10. The one or more non-transitory computer-readable media of claim 9 , wherein:
the computing device of the database management system is a particular computing device;
the one or more intermediate result queues comprise one or both of:
a local intermediate result queue configured to store intermediate results produced by the plurality of processing threads, and
a remote intermediate result queue configured to store intermediate results produced by one or more computing devices, other than the particular computing device, of the database management system.
11. The one or more non-transitory computer-readable media of claim 9 , wherein:
the one or more intermediate result queues store message buffers;
the particular thread inserting the one or more intermediate results into the at least one intermediate result queue comprises:
the particular thread populating a message buffer with at least a portion of the one or more intermediate results, and
causing a particular intermediate result queue, of the one or more intermediate result queues, to include the message buffer.
12. The one or more non-transitory computer-readable media of claim 9 , wherein said executing the query over the graph data further comprises:
identifying a remote edge that is owned by a second computing device of the database management system;
in response to identifying the remote edge, including a particular intermediate result that is based, at least in part, on the remote edge in a particular message buffer;
after including the particular intermediate result in the particular message buffer, determining that the particular message buffer is full; and
in response to determining that the particular message buffer is full, sending the particular message buffer to the second computing device.
13. The one or more non-transitory computer-readable media of claim 12 , wherein:
the computing device of the database management system is a particular computing device;
the particular message buffer is registered with a network interface card (NIC) of the particular computing device;
sending the particular message buffer to the second computing device comprises:
producing communication structures directly from the particular message buffer, and
sending the communication structures, to the second computing device, over a network that communicatively couples the particular computing device and the second computing device.
14. The one or more non-transitory computer-readable media of claim 12 , wherein:
the computing device of the database management system is a particular computing device;
the one or more sequences of instructions comprise instructions that, when executed by one or more processors, cause:
selecting the particular message buffer to store the particular intermediate result based, at least in part, on both (a) an owner of a destination vertex of the remote edge being the second computing device, and (b) a stage of query execution at which the remote edge was discovered;
wherein sending the particular message buffer to the second computing device is further based on determining that a number of unacknowledged message buffers, sent to the second computing device for the stage of query execution, does not exceed a limit on the number of unacknowledged message buffers.
15. The one or more non-transitory computer-readable media of claim 9 , wherein the one or more sequences of instructions comprise instructions that, when executed by one or more processors, cause:
identifying a plurality of query execution stages for the query;
wherein the plurality of query execution stages is ordered from earliest query execution stage to latest query execution stage;
wherein said executing the query over the graph data is performed based on the plurality of query execution stages;
wherein the particular thread using depth-first traversal techniques to produce the one or more results for the query comprises:
scanning intermediate results, in the one or more intermediate result queues, to identify one or more late intermediate results for one or more of the latest query execution stages, of the plurality of query execution stages;
producing the one or more results for the query based, at least in part, on using depth-first traversal techniques on the one or more late intermediate results.
16. The one or more non-transitory computer-readable media of claim 9 , wherein the one or more sequences of instructions comprise instructions that, when executed by one or more processors, cause:
identifying a plurality of query execution stages for the query;
wherein the plurality of query execution stages is ordered from earliest query execution stage to latest query execution stage;
wherein said executing the query over the graph data is performed based on the plurality of query execution stages;
wherein the particular thread using the breadth-first traversal techniques to produce the one or more intermediate results for the query comprises:
scanning intermediate results, in the one or more intermediate result queues, to identify one or more early intermediate results for one or more of earlier query execution stages, of the plurality of query execution stages;
producing the one or more intermediate results based, at least in part, on using the breadth-first traversal techniques on the one or more early intermediate results.
17. The computer-executed method of claim 1 , wherein using depth-first traversal techniques to produce one or more results for the query is based, at least in part, on particular one or more intermediate results from at least one intermediate result queue of the one or more intermediate result queues.
18. The one or more non-transitory computer-readable media of claim 9 , wherein using depth-first traversal techniques to produce one or more results for the query is based, at least in part, on particular one or more intermediate results from at least one intermediate result queue of the one or more intermediate result queues.