Distributed graph databases that facilitate streaming data insertion and low latency graph queries

A novel distributed graph database is provided that is designed for efficient graph data storage and processing on modern computing architectures. In particular a single node graph database and a runtime & communication layer allows for composing a distributed graph database from multiple single node instances.

BACKGROUND

The subject disclosure relates to database systems, and more specifically, to distributed graph databases that facilitate streaming data insertion and queries.

SUMMARY

According to an embodiment, a computer-implemented method to reduce number of messages required to add a new edge by employing asynchronous communication, comprises: using a processor, operatively coupled to at least one memory, to execute the following acts: receiving a request at a first machine to add a first target; adding the first target at the first machine, generating a unique VIDT, and forwarding the VIDT to a second machine wherein the second machine adds a vertex, and generates a corresponding VIDS, comprising the acts of: Prepare EID as {ShardID, MAXEID}, incrementing MAXEID; adding an outgoing edge {VIDS,VIDT, LID, EID}; forwarding {VIDS,VIDT,LID,EID} to the first machine; and adding at the first machine the incoming edge.

In another embodiment, a computer-implemented method for efficient throughput edge addition, comprises: using a processor, operatively coupled to at least one memory, to execute the following acts: determine vertex placement, based on a hash or an arbitrary placement function; place outgoing edge requests into appropriate queues of a firehose; and place incoming edge requests into appropriate queues of the firehose, wherein for each queue, in parallel: send requests to add vertices for all sources in an outgoing edges set, and all targets in an incoming edges set, and wait for vertex ids of all added vertices and MAXEID from each machine, respectively. Build the final edge tuples for each queue in the form of {VIDS, VIDT, LID, EID} based on the vids returned and insert the outgoing and incoming edge tuples and their corresponding shards.

In yet another embodiment, a method to provide low latency graph queries, comprises: using a processor, operatively coupled to at least one memory, to execute the following acts: employing a query manager to perform graph queries; and employing the query manager to manage multiple threads of execution to handle multiple concurrent queries from one or more clients; wherein for a complete traversal, a thread running on the query manager performs multiple requests to various shards during multiple waves corresponding to traversal levels, and wherein the thread will maintain all partial results until the traversal finishes (max depth, max nodes, max time allowed) and then return results to clients.

In some embodiments, elements described in connection with the computer-implemented method(s) can be embodied in different forms such as a system, a computer program product, or another form.

DETAILED DESCRIPTION

Distributed graphs present unique challenges with respect to different disparate machines and maintaining sharing of unique edge identifiers (IDs). Sequential updating is the most simple manner for maintaining distributed graphs. Reduction of number of steps in connection with such sharing can improve data throughput and integrity. In this disclosure,FIG. 1illustrates a novel architecture for a system100designed to ingest a large number of vertices and edges while simultaneously or concurrently allowing a high query throughput with low latency for each individual query. The system can include the following modules: a firehose104; a graph database server (or alternatively graph database component)106; a query manager108; and client(s)110. A graph database can maintain a graph dataset in the memory of a computing device, on a hard drive of a computing device or on both memory and hard drive in a hybrid solution. A distributed graph database can store a graph dataset on a collection of separate physical computing devices, such as multiple different computers, each with their own separate processor and memory.

As a prelude to the detailed discussion regarding the novel architecture and its aforementioned modules, a foundation for better understanding the architecture is provided through three broad categories: graph data structure libraries, graph processing frameworks, where the emphasis is on the programming models, and graph databases, where the focus is on storage.

Graph libraries: Graph libraries can provide in-memory-only processing. For example, BOOST Graph library (BGL) provides a generic graph library where users can customize multiple aspects of a data structure including directness, in memory storage, and vertex and edge properties. This flexibility facilitates users customizing the data structure for particular needs. Parallel BOOST graph library, Standard Adaptive Parallel Library (STAPL) and Galois, provide in memory parallel graph data structures. These projects provide generic algorithms to access all vertices and edges, possibly in parallel, without knowledge of underlying in-memory storage implementation. Our graph database employs a similar design philosophy with these libraries but extends these works with support for persistent storage and a flexible runtime for better work scheduling.

Graph processing frameworks: Pregel and Giraph employ a parallel programming model called Bulk Synchronous Parallel (BSP) where the computation consists of a sequence of iterations. In each iteration, the framework invokes a user-defined function for each vertex in parallel. This function usually reads messages sent to this vertex from a last iteration, sends messages to other vertices that will be processed at a next iteration, and modifies the state of this vertex and its outgoing edges. GraphLab is a parallel programming and computation framework targeted for sparse data and iterative graph algorithms Pregel, Giraph and GraphLab are good at processing sparse data with local dependencies using iterative algorithms. However they are not designed to answer ad hoc queries and process graphs with rich properties.

TinkerPop is an open-source graph ecosystem consisting of key interfaces and tools needed in the graph processing space including the property graph model (Blueprints), data flow (Pipes), graph traversal and manipulation (Gremlin), graph-object mapping (Frames), graph algorithms (Furnace) and graph server (Rexster). Interfaces can be defined by TinkerPop. As an example, Titan adheres to many APIs defined by TinkerPop and uses data stores such as HBase and Cassandra as the scale-out persistent layer. TinkerPop focuses on defining data exchange formats, protocols and APIs, rather than offering a software with good performance.

Graph stores: Neo4J provides a disk-based, pointer-chasing graph storage model that stores graph vertices and edges in a de-normalized, fixed-length structure and uses pointer-chasing instead of index-based method to visit them. By this means, Neo4J avoids index access and provides better graph traversal performance than disk-based relational database management system (RDBMS) implementations.

Distributed Graph

The distributed graph database is a composition of a fixed set of single node graph databases called shards. The distributed graph is in charge of managing a list of computation nodes and mapping of shards to nodes and implements an API such that users see only one database instance and not a collection of distributed services. Thus, upon instantiating a distributed graph, a naive user can have access to the same interface as that with the sequential vertices and edges handled internally by the distributed graph API.

The graph can distribute its vertices by default based on a hash function applied to the external vertex identifier. An edge can be located with its source vertex by default. Thus a typical distributed graph method can perform as its first step the computation to decide the shard where a particular vertex or edge is located or the shard where it will be allocated. Subsequently the method invocation can be forwarded to the shard in charge to finish the method execution.

InFIG. 2, a distributed system running two distributed graph instances is shown. In general, a machine can host an arbitrary number of shards with each shard running as a distinct process. One or more shards, or in some embodiments each shard, can be uniquely mapped to a host and port as specified by a host file created during the cluster configuration. The cluster depicted has three physical machines hosting the distributed graph system: Machine 0, 1, and 2. Within one machine, there are a number of shards that can belong to multiple graph databases. One or more shards, or in some embodiments all the shards, of a database can be managed by a distributed graph object responsible for address resolution and method forwarding. Data in a distributed graph database can be accessed by any of the nodes of the database. Additionally, in the model a QueryManager (QM) is introduced as a separate process running on its own machine with its own hardware resources. The QM can instantiate a same distributed graph object as all the other shards except a different flag can be passed to a constructor to signal that this instance is not a shard but a proxy. In proxy mode, a distributed graph object will not own any data and any method invocation is performed at a remote location. In the following section, main communication primitives that can be used to exchange data between nodes of the database are described.

Single Node Graph Database

The single node graph database implements a property graph model. Each graph is identified by a user-specified graph name and is comprised of vertices, edges, and properties (e.g., attributes) associated with each vertex and edge. Each vertex is identified by a unique external vertex ID specified by a user and an automatically generated unique internal vertex ID. Each edge is identified by vertex IDs of its source and target vertices and an automatically generated unique edge ID. Multiple edges between a same pair of vertices are allowed.

In some embodiments, one or more vertices or edges (or, in one embodiment, each vertex or edge) is associated with a string label that can be used to categorize vertices and/or edges and facilitate efficient traversal (e.g., only traverse edges of a specific label). The property set of a vertex and/or edge can be or include a list of key-value pairs where each key is a property name and the value associated with the key is the value of the corresponding property for this vertex and/or edge. Property values can be strings, numbers (integer, float, double), vector of numbers, or composite values consisted of strings and numbers. In some embodiments, multiple values for a single property, and properties (e.g., meta data) of properties can be those that are supported to be compliant with Apache TinkerPop 3.

Internally vertex-centric representations can be used to store vertices and edges, along with the maps for vertex and edge properties. An underlying high-performance key-value store can be used to store the above representations in memory and/or on disk.

A rich set of graph APIs can be provided to support most, if not all, fundamental graph operations, including graph creation and/or deletion, data ingestion (e.g., add vertex can be edge one at a time or via batch loading of files in comma-separated value (csv) format), graph update (e.g., delete vertex and/or edge, set and/or update or delete vertex and/or edge properties), graph traversal (iterate through vertices and edges of each vertex), data retrieval (e.g., get vertex and/or edge properties), graph search (e.g., find vertex and/or edge by ID, build or search property index).

Messaging Layer

Applications written within the subject framework can be executed in a Single Program Multiple Data (SPMD) fashion similar to Message Passing Interface (MPI). The binary corresponding to an application can be executed on multiple machines and each instance can have its own identity and know how many nodes make up the computation. After an application starts it can access local memory and local storage. When remote data needs to be processed, communication can be employed. In some embodiments, the distributed graph system can use Remote Procedure Call (RPC) as its core communication abstraction. The RPC can be abstracted on top of a native communication library such as sockets, Message Passing Interface (MPI), Parallel Active Message Interface (PAMI) or Global-Address Space Networking (GASnet) inheriting advantages and disadvantages of the underlying layers. The RPC abstractions provides to the distributed system developers a high level abstraction that helps with productivity and portability of the system.

The RPC API exposed to the user can be exemplified inFIG. 3. At the very high level, RPC can allow users to invoke a function with optional arguments on a remote machine. In embodiments described herein, RPC can employ a user specification of the destination machine, the function to be invoked on the remote host and the arguments to be passed to the function. Such specification can be as shown inFIG. 4, line33. The arguments that will be passed from source machine to the target machine can be encapsulated into a single structure together with a special method that is used to serialize and de-serialize custom data members to a stream of bytes. In one embodiment, the Cereal library is used for this purpose and the user can enumerate all data members of the structure as shown inFIG. 4, lines11-17. The library can serialize by default all basic data types and most of the standard template library (STL) containers including vectors and strings. The library call also recursively serializes data members that are struct or class, provided the data members have their own serialization method already defined.

The function pointer corresponding to the function to be executed remotely can be converted to a unique integer number before being sent on the network. On the receiver side the unique integer can be converted back to a function pointer local to the destination machine. In general one can not assume that a function pointer has the same value on the different nodes where the RPC will be executed. The mapping from a global to unique integer and the inverse operation can be achieved using the register_rpc utility as shown inFIG. 4, line29. In turn, this call can store into a first table the mapping from function pointer to a unique identifier and in a second table the inverse mapping from a unique identifier to the function pointer. The registration of the RPC functions can be performed as one of the first steps of a program using the infrastructure we present here.

The RPC functions can be registered by all processes of a computation before being invoked. In some embodiments, this can be accomplished using a barrier like concept. The computation to be invoked remotely can be implemented in the subject framework with two functions. A first function that can be invoked remotely will receive as arguments an identity of a sender, a byte buffer corresponding to serialized arguments, and/or size of the buffer. This function can de-serialize the byte buffer into the argument that the user passed when invoking the RPC and subsequently invoke the user function with the argument passed in by the sender. The reason for this double invocation is the fact that, in this embodiment, the subject RPC is a pure library approach and the re is not a separate tool to hide some of the implementation details from the user. For example,FIG. 4, lines22-26shows the function in charge of receiving the byte array including the de-serialization step and the invocation of the user requested function. The user specific computation can be performed as part of lines20-21. In the embodiment shown, a simple RPC invocation call chain is over once the framework invokes the remote user function. In some embodiments, a return value may be required. In this case the invoking process can pass in the list of arguments the local memory address of a variable and the identity (FIG. 4, lines4-7). This can be used by a second RPC invoked by the destination node to write back the result value. The process waiting for a return value can often need to spin while waiting for the results to come back and while spinning it can often perform a polling call to execute other possible incoming RPC. Both the arguments and the return values can be arbitrarily large data structures in some embodiments.

Runtime

In general, each individual process (or, in some embodiments, one or more processes) can receive RPC requests from multiple sources. In order to provide a high throughput of executed RPCs per second, a multithreaded task based runtime was employed. Within the system, each RPC invocation (or, in some embodiments, one or more RPC invocations) when received (or, in some embodiments, after receipt) from the network is encapsulated within a task and placed into a runtime scheduler for execution. The runtime scheduler can maintain a pool of threads and dispatch individual tasks to individual threads. The scheduler can also allow for work stealing to keep the load balanced. The same runtime can be also used within the framework to execute parallel computations within one SMP node.

Runtime Scheduler and RPC Interaction:

After an RPC request is received on one of the incoming communication channels, in some embodiments, the messaging layer will only extract the argument and prepare a task that will be placed for execution similar to the example shown inFIG. 3, lines1-10and16-17. As such, the polling thread can extract RPC requests and post them for execution very fast. In some embodiments, the number of concurrent RPC requests that will be executed concurrently will be proportional with the number of threads used by the scheduler. Individual RPC requests can invoke additional RPCs as part of their body, as shown in the example inFIG. 4when a return value was requested. Accordingly, it is possible in general that multiple tasks executing simultaneously or concurrently can post RPC requests concurrently thus calling for some amount of serialization when accessing the communication channels (sockets). Because individual channels can be maintained between pairs of processes, only synchronization of access to these channels for requests that have the same compute process destination is called for.

Query Manager

The foregoing introduced a distributed graph database design. The data has been shown as being distributed in shards across different computation nodes, with possible multiple shards per physical machine. The disclosed embodiments also provide for detail regarding how a graph database can be accessed by clients. Referring back toFIG. 1within the subject system100, database clients are distinguished into two categories. A first category called query manager(s)108and a second category called regular client(s)110. The query manager(s)108can be a database client that will run on a powerful node in terms of memory and computing resources like cores and hardware threads. The query manager(s)108can open a communication channel with each individual process (or, in some embodiments, one or more processes) of the distributed database and afterwards is capable of posting RPCs to any of the nodes of the database for performing various graph operations. Regular client(s)110will not connect to the distributed database directly but rather they will connect to a query manager108and this will execute the query for the client110. The extra level of indirection has the following advantages. First it allows for a level of control for database load. The query manager108may decide to delay the clients answers rather than loading a database with requests that can't be handled. Secondly it is often the case that a query on a graph database is a complex computation like a breadth first search (BFS) or finding a path between a source and a target. For these queries the query manager108may end up accessing all shards of the database multiple times. Thus the query manager108can maintain all this partial state while performing the query and it returns to the client110when the final answer is available. It is also possible that the query manager108can perform additional number of optimizations like some information caching to further optimize response times for advanced read queries.

Query Manager Implementation:

The query manager108when creating an access point to a graph database will instantiate the same distributed graph class as all the other processes of the database. The only difference will be a flag passed to the constructor that will inform the address resolution module that none of the graph data is local and everything needs to be accessed using RPC. As shown inFIG. 5the graph instantiated by the query manager108acts as a proxy to the database data.

Graph Queries and Analytics

Continuing to refer toFIG. 5, the query manager108will handle most if not all basic queries like add/delete/get vertex, edge, property. Additionally the query manager108can implement graph specific queries like various traversals. In this section, what is disclosed is how queries for the distributed graph can be implemented. As a simple example, consider a simplified breadth first search where we go only a given number of levels deep. The analytic will be provided with the distributed graph instance, the starting vertex and the number of levels. First, a partitioned frontier is instantiated which will maintain a list of vertex identifiers grouped by the shard to which they belong. We start by adding the initial starting vertex to it (lines2-3). Next, an iterative process is started where the next BFS frontier is computed based on a current frontier. For this, the system100extracts from the frontier vertices that all live in a certain shard and we post an asynchronous request to that shard to collect and retrieve all neighbors of this set of source vertices (lines7-10). This is performed using the graph method get_all_edges_async( ) which internally will use the RPC mechanism previously described. It should be noted that none of the results are available as soon as method is invoked. The system100only started the computation on remote shards when the method returns this. After the requests to all relevant shards are posted next the system100starts waiting for result(s) to arrive back. It is possible that by the time the system100finishes the last invocation some of the results may be already back, thus this flexible RPC mechanisms allows the system100to overlap communication with data retrieval.

In a second phase, the system100waits for data from a particular shard to arrive and next it processes received data preparing the next wave of the BFS. Additional analytics can be implemented in a similar execution model with the algorithm described in this section. It is contemplated to exploit distributed asynchronous algorithm(s) to perform various analytics. The RPC mechanism the system100employs can allow, for example, for a traversal to start from a query manager node, but next the traversal can be forwarded by the individual shards of the database which asynchronously may send result data back to the query manager108.

Clients

Regular clients110will connect to a query manager108. There can be more than one query manager108per system but still a small number in the order of tens. Regular clients110can be in the order of hundreds and they will communicate with a query manager108over a network protocol. Currently a query manager108can start an HTTP server and accept REST queries from clients110that are subsequently mapped into graph operations.

A client110can request multiple graph traversals to be performed for particular vertex ids. Traversals can happened concurrently from possible multiple threads. Requests are posted to the query manager108which will perform the data aggregation for the whole traversal. The rest API can be issued from a browser, JavaScript, Java or Python program.

Another novel concept for a distributed database that is introduced in the subject novel framework is the Firehose104, an extension for optimizing the ingestion of data. The single node graph database that is extended to provide the distributed version is optimized for a single writer, multiple readers scenario. The single node database supports multiple concurrent readers alongside a writer. However if multiple threads are trying to access the database for write operations they will be simply serialized. For this reason in the subject design each process running a shard of the database creates an additional thread that is in charge of only write operations. The main thread reads from file/socket a line (source, target, timestamp, . . . ). A decision is made regarding the destination shard (ShID) based on source vertex. The firehose104can use different placement functions (e.g., Hashing, Explicit placement and an additional key value store for placement tracking). The firehose104places the data in the queue of the thread in charge of shard ShID. Each thread is in charge of one shard. It reads from local queue and pushes on a socket connection of the data. Data pushes are buffered and no explicit return values are expected for maximum throughput.

In a lot of large scale practical applications there is often a continuous stream of vertices and edges being created and lots of read queries executed simultaneous with the stream of inserts. For these applications, the Firehose104will be in charge to add the edges, possibly in a batched mode, while query manager as introduced in this section will be mainly in charge of read only queries. The Firehose104will run as a separate process, opening communication channels to all the shards of the database. At the same time the Firehose104will connect into the client existing infrastructure accepting requests for adding vertices, edges or updating properties for existing vertices and edges.

With respect to the graph database server (or alternatively graph database component)106, a highly scalable solution is provided using multiple shards per node (OS instance, machine) and multiple nodes (cluster). Each shard server will have a connection to all other shards for fast data exchange. These will be used for asynchronous queries and load balancing long adjacency lists. Each shard will provide a connector for the firehose104for fast data insertion. The firehose104can connect/disconnect at its own pace. Each vertex and edge has an unique identifier that has shard id embedded in it. Assuming one machine has access to internal vertex identifier it has access to the shard/machine where the vertex is located. The system100provides a runtime that supports a highly concurrent execution of requests. One thread inserts/updates (ReadWrite Transactions). Multiple threads perform read operations/transactions.

Vertex and Edge Management

In the subject distributed graph database each vertex and edge is uniquely identified by an internal vertex and edge identifier respectively. In this section we discuss how identifiers are generated and managed while adding items to the database. Edges (outgoing and incoming) are stored as tuples of such identifiers to save storage and improve the data lookup performance.

Additionally vertex and edge properties are stored as key, value pairs using the vertex or edge ids as keys. Internal Vertex identifiers Each vertex has a unique numeric internal identifier. This is allocated when the vertex is created and it won't be reused for any other vertex in the database. In a single node graph database producing a unique id is done by incrementing a variable each time a vertex is added. We will refer to this variable as MAX VID and an unsigned 64 bit number can be used to represent it. When the database is first created this is initialized to zero. To reduce storage requirements a numeric label identifier can be embedded within the binary representation of the vertex identifier, for example in the most significant bits. The vertex identifier is returned to the caller when the vertex is created or by the find_vertex method with an external identifier. For a distributed graph database the system100ensures a unique vertex identifier by using the following protocol. First a vertex is uniquely associated with a shard by using either a default hash function or an arbitrary placement function provided by the user. The distributed graph maintains a mapping from the shard identifier to the physical machine where the shard is stored. The vertex will be added to the shard and the machine as previously identified. When adding a vertex to a shard a vertex identifier is generated by incrementing the shard-local MAX VID variable and the overall global identifier of the vertex becomes the following triplet {LabelId,ShardId,LocalV ertexIdentifier}. Edge Identifiers: Each edge has associated an unique edge identifier. A major challenge for our design comes from the fact that we allow multiple edges between the same two vertices and because we store both incoming and outgoing edges. Assuming we have two vertices A and B and we add an edge {A,B} followed by another edge from {A,B}. Using an unique edge identifier allows to distinguish between the two edges: {A,B,eid1} and {A,B,eid2}. For undirected graphs or graphs where we track incoming edges we store two edge tuples in the database. For example, for the edge {A,B} we store one outgoing edge {A,B,eid1} and one incoming edge {B,A,eid1}. Both edges will know they are part of the same edge because they share the same unique identifier. For a single node graph database an edge identifier can be easily generated by incrementing a MAX EID, unsigned 64 bit integer. For the distributed database the number of actions to be performed when adding an edge increases due to the fact that the source and the target may live in two different shards on two different machines. Assuming a vertex A is mapped to shard1 and a vertex B is mapped to shard 2, at a minimum, for the edge {A,B} we store one outgoing edge {A,B,eid1} on shard1 and one incoming edge {B,A,eid1} on shard2. The edge identifier will be generated in shard1 and communicated to shard2 together with the rest of the arguments when adding the incoming edge. It is also valid to generate the id in shard2 and communicated to shard1 provided the shard identifier is also embedded in the most significant bits of the edge identifier.

Efficient Edge Addition for a Distributed Database

A very common operation for graph databases is adding an edge between a source and a target vertex without adding vertices a priori. For example add_edge (A, Knows, B). This turns out to be a complex operation as shown inFIG. 6. First the operation will add two vertices A, and B if they don't exist already (FIG. 6, lines2,3), create a label Knows if it doesn't exist already (FIG. 6, line4), add the outgoing edge for A (FIG. 6, line5) and add the incoming edge to B (FIG. 6, line6). Note that each of these invocations produce vertex and edge ids that are subsequently used creating data dependencies between the five steps of the method.

Basic algorithm: A straightforward approach to implement the steps depicted inFIG. 6is to execute the code above on a client or query manager node and execute all 5 steps synchronously one after another. Thus for each of the steps 2 to 6 we will have two messages exchanged over the network: one to invoke remotely the operation and one to return results used in the subsequent steps. Thus there will be a total of at least seven messages exchanged (the last step doesn't have to return anything). If confirmation of the final step is required then this approach will take a total of eight steps. 2) Asynchronous algorithm: A first improvement we propose in this patent is to use asynchronous RPC mechanism that our runtime natively supports. For this approach we first forward the add edge method to the node where the destination is allocated (DEST SHARD). On this shard the destination vertex is found or created and its id(VIDT) is forwarded with the rest of the arguments to the machine where the source is located (SOURCE SHARD). Here the source vertex will be located or created (VIDS), the outgoing edge will be created {VIDS,VIDT,LID,EID} and finally forward the invocation back to DEST SHARD to add the incoming edge using the edge id previously generated.

Thus, the number of communication steps are reduced from seven down to three. A fourth step can be optionally employed if a confirmation of the method termination is required on the client initiating the operation.

Batched edge and vertex addition: It is often the case that edges and vertices are added to the database at very high rates and it is acceptable by the user's application that the vertices or edges are added in a batched fashion. In this section we describe a novel mechanism for adding items using batches. We previously introduced the notion of Firehose for optimizing fast insert rate operations and the batching mechanism presented in this section is implemented as part of the Firehose. Let's assume a set of edges are to be added to the database using the semantic described inFIG. 6. The Firehose will collect a batch of them of size N and perform the following processing on them: 1) Create 2*P queues where P is the number of shards. For each shard there will be one outgoing and one incoming edges queue. 2) For each add edge request place one entry in the outgoing edges queue corresponding to the shard where the source vertex of the edge is allocated. Similarly we place an entry in the incoming edges queue corresponding to the shard where the target vertex is allocated. 3) For each pair of queues for each shard we collect the set of vertices to be added to the shard and we send one bulk request to the shard to add the vertices. This step can be done in parallel for all pairs of queues and their corresponding shard. The request will return the vertex identifiers for all newly added vertices. It will also reserve an edge id range on the shard and the edge id range is also returned to the Firehose. 4) Based on all vertex identifiers returned and edge ranges reserved the Firehose will prepare the final tuples corresponding to the edges to be added. The edge tuples containing only internal ids will be sent to the database shards to be inserted. This insertion also happens in parallel for all shards. 5) Optionally, the mapping from external vertex id to internal vertex id can be cached on the Firehose such that to minimize the number of vertices information sent to the shards in step 3.

This novel approach for performing batched edge addition provides the highest amount of parallelism and the lowest number of messages exchanged compared to the other two methods previously introduced. For a given batch of N method invocations, the basic algorithm will perform 7*N communication messages, synchronizing for each step. The asynchronous algorithm performs 3*N messages if the invoking thread doesn't require confirmation termination or 4*N messages if confirmation is required. The batched approached will exchange four larger granularity messages per shard for the whole set of N invocations for a total of P*4 messages. Usually P will be much smaller than N. While the batched method send much fewer messages, there is more data per message sent. However most networks perform better when data is aggregated in bigger chunks.

FIG. 8illustrates an example property graph model in connection with the subject novel architecture. Collection of vertices, edges and their associate properties; Vertex: External Id: Unique Identity, URL, UUID, etc: String Internal Unique Numeric Id: UINT64; and Edge: Label: String, Internal Unique Numeric ID: UINT64.

FIG. 9illustrates a dynamic storage schema in connection with the disclosed architecture. Add edge in a single node database. For each edge we add one entry in the out going list of the source and one entry in the incoming list of the target. The database maintains a variable: MAXEID for the last edge identifier generated; initially this is initialized with zero. Every time a new edge is added:

We add a new mapping <VIDT, {VIDS, MAXEID, LABELID}>

We increment MAXEID by one

It is mandatory that the entry for outgoing list of VIDS, and the entry for the incoming list of VIDT share the same value for MAXEID

FIG. 10illustrates a distributed graph example. Vertices are distributed on multiple machines. Outgoing edges of a vertex are stored where the source vertex is allocated. Incoming edges of a vertex are stored on the machine where the target vertex is allocated. There are three methodologies: (1) Method to reduce the number of exchanged messages for edge addition using asynchronous communication; (2): Method to reduce the number of exchanged messages for edge addition using firehose and batching; and (3) Method for low latency high throughput graph query execution.

With respect to vertex placement, a vertex is allocated to a machine using some form of hashing or an arbitrary placement function. The vertex is added to the designated machine and the machine will generate an unique vertex identifier according to the single node algorithm.

For a simple add edge in a distributed database, the basic algorithm is as shown inFIG. 11.

FIGS. 12 and 13illustrate methods to reduce the number of messages required to add a new edge by using asynchronous communication. At1302, a request is received at a first machine to add a first target. At1304, the methodology performs the act of adding the first target at the first machine, generating a unique VIDT, and forwarding the VIDT to a second machine wherein the second machine adds a vertex, and generates a corresponding VIDS, comprising the acts of: Prepare EID as {ShardID, MAXEID}; incrementing MAXEID; adding an outgoing edge {VIDS,VIDT, LID, EID}; forwarding {VIDS,VIDT,LID,EID} to the first machine. At1306, the methodology adds at the first machine the incoming edge.

In accordance with an optional embodiment, a maximum of three communication steps are performed per add edge request. In yet another embodiment, a maximum of four communication steps are performed per add edge request, wherein a fourth step is employed to confirm termination to a client initiating an operation.

FIGS. 14-18illustrate a second methodology in accordance with the claimed subject matter for a high throughput edge addition with batched solution using the firehose. At1802, the method determines vertex placement, based on a hash or an arbitrary placement function. At1804, place outgoing edge requests into appropriate queues of a firehose. At1806, place incoming edge requests into appropriate queues of the firehose, wherein for each queue, in parallel: send requests to add vertices for all sources in an outgoing edges set, and all targets in an incoming edges set, and wait for vertex ids of all added vertices and MAXEID from each machine, respectively. In accordance with an embodiment, an ingest process is divided into batches, and wherein the acts are executed for respective batches. In accordance with another embodiment, for all vertices added insert into a map (hash table) the pairing from external vertex identifier to internal vertex identifier <A,VIDA>, <B,VIDB>, <C,VIDC> and <D,VIDD>. In yet another embodiment, for all outgoing and incoming edges map from external ids to internal ids and edge identifiers. Upon receiving the ids for all vertices added the firehose will build for each queue, for all the tuples in each queue the following info: {VIDS, VIDT, LID, EID}. The method further comprises sending edge quads to respective machines for insertion. Furthermore, the mapping from external to internal can be cached for use in a next iteration.

FIGS. 19-20illustrate example methodologies in connection with the query manager. At2002, a query manager is employed to perform graph queries. At2004, the method employs the query manager to manage multiple threads of execution to handle multiple concurrent queries from one or more clients; wherein for a complete traversal, a thread running on the query manager performs multiple requests to various shards during multiple waves corresponding to traversal levels, and wherein the thread will maintain all partial results until the traversal finishes (max depth, max nodes, max time allowed) and then return results to clients. In an embodiment, additional query managers can be instantiated to accommodate increasing load.

FIG. 21illustrates example performance results in connection with the claimed and disclosed novel subject matter.

In order to provide a context for the various aspects of the disclosed subject matter,FIG. 22as well as the following discussion are intended to provide a general description of a suitable environment in which the various aspects of the disclosed subject matter can be implemented.FIG. 22illustrates a block diagram of an example, non-limiting operating environment in which one or more embodiments described herein can be facilitated. Repetitive description of like elements employed in other embodiments described herein is omitted for sake of brevity.

With reference toFIG. 22, a suitable operating environment2200for implementing various aspects of this disclosure can also include a computer2212. The computer2212can also include a processing unit2214, a system memory2216, and a system bus2218. The system bus2218couples system components including, but not limited to, the system memory2216to the processing unit2214. The processing unit2214can be any of various available processors. Dual microprocessors and other multiprocessor architectures also can be employed as the processing unit2214. The system bus2218can be any of several types of bus structure(s) including the memory bus or memory controller, a peripheral bus or external bus, and/or a local bus using any variety of available bus architectures including, but not limited to, Industrial Standard Architecture (ISA), Micro-Channel Architecture (MSA), Extended ISA (EISA), Intelligent Drive Electronics (IDE), VESA Local Bus (VLB), Peripheral Component Interconnect (PCI), Card Bus, Universal Serial Bus (USB), Advanced Graphics Port (AGP), Firewire (IEEE 1394), and Small Computer Systems Interface (SCSI).

The system memory2216can also include volatile memory2220and nonvolatile memory2222. The basic input/output system (BIOS), containing the basic routines to transfer information between elements within the computer2212, such as during start-up, is stored in nonvolatile memory2222. Computer2212can also include removable/non-removable, volatile/non-volatile computer storage media.FIG. 22illustrates, for example, a disk storage2224. Disk storage2224can also include, but is not limited to, devices like a magnetic disk drive, floppy disk drive, tape drive, Jaz drive, Zip drive, LS-100 drive, flash memory card, or memory stick. The disk storage2224also can include storage media separately or in combination with other storage media. To facilitate connection of the disk storage2224to the system bus2218, a removable or non-removable interface is typically used, such as interface2226.FIG. 22also depicts software that acts as an intermediary between users and the basic computer resources described in the suitable operating environment2200. Such software can also include, for example, an operating system2228. Operating system2228, which can be stored on disk storage2224, acts to control and allocate resources of the computer2212.

System applications2230take advantage of the management of resources by operating system2228through program modules2232and program data2234, e.g., stored either in system memory2216or on disk storage2224. It is to be appreciated that this disclosure can be implemented with various operating systems or combinations of operating systems. A user enters commands or information into the computer2212through input device(s)2236. Input devices2236include, but are not limited to, a pointing device such as a mouse, trackball, stylus, touch pad, keyboard, microphone, joystick, game pad, satellite dish, scanner, TV tuner card, digital camera, digital video camera, web camera, and the like. These and other input devices connect to the processing unit2214through the system bus2218via interface port(s)2238. Interface port(s)2238include, for example, a serial port, a parallel port, a game port, and a universal serial bus (USB). Output device(s)2240use some of the same type of ports as input device(s)2236. Thus, for example, a USB port can be used to provide input to computer2212, and to output information from computer2212to an output device2240. Output adapter2242is provided to illustrate that there are some output devices2240like monitors, speakers, and printers, among other output devices2240, which require special adapters. The output adapters2242include, by way of illustration and not limitation, video and sound cards that provide a means of connection between the output device2240and the system bus2218. It should be noted that other devices and/or systems of devices provide both input and output capabilities such as remote computer(s)2244.

Computer2212can operate in a networked environment using logical connections to one or more remote computers, such as remote computer(s)2244. The remote computer(s)2244can be a computer, a server, a router, a network PC, a workstation, a microprocessor based appliance, a peer device or other common network node and the like, and typically can also include many or all of the elements described relative to computer2212. For purposes of brevity, only a memory storage device2246is illustrated with remote computer(s)2244. Remote computer(s)2244is logically connected to computer2212through a network interface2248and then physically connected via communication connection2250. Network interface2248encompasses wire and/or wireless communication networks such as local-area networks (LAN), wide-area networks (WAN), cellular networks, etc. LAN technologies include Fiber Distributed Data Interface (FDDI), Copper Distributed Data Interface (CDDI), Ethernet, Token Ring and the like. WAN technologies include, but are not limited to, point-to-point links, circuit switching networks like Integrated Services Digital Networks (ISDN) and variations thereon, packet switching networks, and Digital Subscriber Lines (DSL). Communication connection(s)2250refers to the hardware/software employed to connect the network interface2248to the system bus2218. While communication connection2250is shown for illustrative clarity inside computer2212, it can also be external to computer2212. The hardware/software for connection to the network interface2248can also include, for exemplary purposes only, internal and external technologies such as, modems including regular telephone grade modems, cable modems and DSL modems, ISDN adapters, and Ethernet cards.