Scaling Graph Query Processing: Focused Exploration on Partitions Using Load Balancing

arXiv:2610.32015v1 [cs.DB] 23 Oct 2026

Scaling Graph Query Processing: Focused Exploration on Partitions Using Load Balancing Harshit Modi

Abhishek Santra

Sharma Chakravarthy

IT Lab & CSE Department University of Texas at Arlington Arlington, Texas, USA harshitashokkum.modi@mavs.uta.edu

IT Lab & CSE Department University of Texas at Arlington Arlington, Texas, USA abhishek.santra@mavs.uta.edu

IT Lab & CSE Department University of Texas at Arlington Arlington, Texas, USA sharmac@cse.uta.edu

Abstract—The focus of this paper is on query processing and not search (approximate or exact). The goal is to find all or k answers and the expressiveness of a query (as opposed to a search expression which is typically simpler) is considered important. Although SQL query processing has matured over the last several decades, query processing over graph data has not received the same attention. Only recently, systems, such as Neo4J, and others have started using graphs for data representation on a large scale. Their approaches and algorithms remain proprietary. In this paper, we address the problem of answering graph queries and their efficient processing on arbitrary-sized graphs. We propose the partitioning of graphs and the use of the distributed paradigm to achieve efficiency and scalability. Our approach takes a subgraph as query input including wild cards and comparison operations. Our partitioned approach leverages focused exploration and distributed processing to achieve the desired parallelism and improve the response time. We also present heuristics to reduce the number of partitions loaded for query processing. We present our partitioning algorithm, distributed processing architecture used, and load balancing for computing query answers. We establish the correctness of the approach in the presence of partitions, compare heuristics for load balancing, and extensive experimental results over large synthetic and real-world graph data sets. Index Terms—Graph Querying, Expressive Queries, Scalable Processing, Map/Reduce

I. M OTIVATION As the amount of structured data and application characteristics are increasing, graphs used for modeling them are becoming larger and they cannot be stored in main memory or use a single processor for bounding the response time. For example, Facebook has around 2.3 billion active users [1], computer science bibliography website, DBLP, has over 5 million records of conferences/workshop papers and journal articles [2], and Internet Movie Database (IMDb) contains over 360 million data items with information of 123 million people and 6.5 million titles [3]. This brings significant challenges in querying large graphs. We propose a partitioned approach along with leveraging a distributed processing framework for processing queries on large graphs. When query evaluation is done on partitioned graphs, answers may span multiple partitions which brings new challenges to make sure: i) all answers (or k answers) are computed (completeness) and ii) all computed answers are correct (soundness). Our approach allows us to match partition

sizes to the processor capability as well as control the amount of parallel processing. For this approach, we also propose a number of heuristics for choosing which and how many partitions to evaluate after each iteration to minimize the total number of partitions loaded and to reduce total response time (Load balancing). We also formulate a load factor to assess the efficacy of a heuristic used. Based on the availability/affordability of resources, our approach has the ability to adjust to specified/affordable amount of parallel processing. In the ideal case, we can match the number of partitions with the corresponding number of resources in each iteration for maximum parallelism and the best response time. Problem Statement: The problem is to develop a distributed processing framework to process graph queries on very large graphs with reasonable or controlled response time. Specifically, given a large graph G and available resources (in terms of total number of processors/cores), develop a graph partition strategy P , a set of partition loading heuristics H, and a distributed algorithm that can efficiently and correctly (satisfy soundness and completeness) process any expressive graph query Q. The sub-problems of this are: i) expressive query specification, ii) developing partitioning strategies, iii) distributed processing algorithms to process partitions using proposed heuristics and obtaining all (or k) answers, and iv) establishing soundness and completeness of results. A. Why Distributed Processing? Having experimented with parallel processing approaches [4], we believe an established distributed processing paradigm is better suited for this work. Then there is the question of which one to use and why among many choices, such as Spark, Hadoop, Pregel, etc. We want to assert that our work is not for measuring absolute response time. Instead, our goal is to develop a scalable query processing approach using a paradigm and validate it analytically and experimentally. Any choice of framework is neutral to our focus and the use of Hadoop and Map/Reduce is based on availability and convenience. It is true that some algorithms can benefit from, say, Spark. But our algorithms do not have such in-memory storage requirements across iterations to take advantage of, say Spark. Finally, our approach is architecture agnostic and

the algorithms do not change much. Only the absolute values will change and trends are likely to remain the same. We have chosen the publicly available Hadoop Map/Reduce framework as a distributed paradigm that has been used to scale computations horizontally to accommodate very large data sizes. This paradigm has been successfully used for a number of graph problems, such as search [5] and approximate matches [6], [7], counting triangles [8], etc. Apache Hadoop, which is an open-source implementation of Map/Reduce, has been used extensively. The choice of Map/Reduce instead of traditional parallel processing is obvious. Partitioning still remains the same. However, using multiple processors, scheduling them, passing data among processors, and loading needs to be done by the query processing system. In addition, aggregating partial results and bookkeeping for that is quite complicated. On top of this, failure recovery of processors needs to be handled by the query processing system. This was experienced in our previous work [9], [10] which led to the choice of the Map/Reduce paradigm. Hadoop framework offers a number of advantages and also some challenges. The main challenge is to cast the algorithm into Map/Reduce framework and figure out how to exploit the shuffle and merging. Correctness of the algorithm, in the presence of partitioning, is also to be addressed. However, the Hadoop framework handles heterogeneous processors, different size partitions to match processing resources, choice of number of processors for parallelism, and robust execution (fault tolerance) in the presence of failures. If we implement our algorithm using traditional multiprocessor implementation, extensive coding is needed for inter-process communication and bookkeeping. In our implementation, the map, shuffle, and reduce paradigm are used to reduce bookkeeping and inter-process communication. Moreover, Map/Reduce has also been used for partitioning the graph based on the number of resources available. B. Choice of Data Sets We have chosen IMDb and DBLP data sets for experimental purposes for a reason. Although the data sets ate not extremely large, they allow us to verify query results comparing them with their database versions and writing corresponding SQL queries. However, the approach is general and scales to any number of Mappers/Reducers depending on the data size. This also helps us demonstrate the expressiveness of our query input with SQL in contrast to other work. The contributions of this paper are: • Scalable approach to graph querying on arbitrary sized graphs – directed, undirected, labeled. • Support expressive graph queries - handle relational comparison operators (>, >=, <, <=, =, <>), range operators (Between, IN), wild cards (multiple character match (%), single character match ( ), and Boolean operators AND and OR. • A variable-size range partitioning Map/Reduce algorithm that takes a single pass to create graph partitions of different sizes and their adjacency lists.

Several heuristics to optimize the loading of partitions for efficient query evaluation and their analysis. • Map/Reduce algorithm for leveraging distributed processing for querying. • Extensive experimental analysis on large real-world and synthetic data sets using a broad range of queries to validate the proposed heuristics and the approach. •

II. R ELATED W ORK Although there has been a lot of effort on graph search [6], [11], [12], [13], [7] and this paper is about graph querying, we limit ourselves to discussing earlier querying languages [14] and approaches for graphs. SPARQL query language[15] is a W3C recommendation for querying RDF (Resource Description Framework) graph data. RDF is a directed, labeled graph data format for representing information on the Web. It describes a graph by a set of triples, each of which describes a pair (attribute and value) or an interconnection between two nodes. The SPARQL query language works primarily through a primitive triple pattern matching technique with simple constraints on query nodes and edges. Cypher[16] is Neo4j’s [17] graph query language, that allows for expressive and efficient data querying in a property graph. Patterns are expressed syntactically following a “pictorial” intuition to encode nodes and edges with arrows between them. However, above mentioned applications require the query to be represented in a specific language model. Hence writing structured queries requires extensive experience in query language and the data model, as well as a good understanding of the particular data set. Our approach takes a query as a graph and finds all instances of it in a very large graph. It is to be noted that there is some work addressing subgraph enumeration ([18], [19]) where triangles, cliques, diamonds, etc. are identified, which is not the same problem as query evaluation, which is the focus of the paper. Moreover, such work will not be able to process the large class of expressive queries that have been taken care of in the proposed approach. In a nutshell, as compared to earlier work on graph query processing, this work differs in a number of ways. Scalability is being addressed in a general-purpose manner using a different paradigm. This work extends upon our earlier work on partitioned non-Map/Reduce approach for query processing [4], [20], [9]. This paper handles more expressive query representations. Also, the heuristics proposed in this paper are different and relevant to the framework used. Finally, there are no restrictions on the number of entities or labels that can be specified as part of a query. We can find all (or k) exact matches in very large graphs. Using the approach proposed in this paper, “query relaxation” [21] can be easily incorporated as the expression computed is similar to an SQL query, and edge labels are similar to attribute values. In this approach, answers can also be ranked to output results based on a given threshold.

III. L ARGE G RAPH Q UERY P ROCESSING C HALLENGES Three major challenges for querying large graphs are: i) how to partition the graph to minimize computation overhead for answers spanning multiple partitions, ii) how to preserve the correctness of the results in the presence of partitions, and iii) what heuristics are appropriate for improving computation efficiency. These issues arise due to graph partitioning, the need to load and process partitions independently (and may be more than once), and the bookkeeping required to compute correct and complete results. Need for Variable-size Partitioning: There are several ways to partition a graph and this has been researched extensively (e.g., KaHIP [22], [23] and Metis [24].) Existing approaches, typically, make multiple passes over the input or produce equal-sized partitions, thus affecting the overall performance. Moreover, they are unable to match partition sizes based on the processing resource capabilities. Variability of processing resources is common in Map/Reduce framework.

Fig. 1: Range Partitioning Adjacency Lists Vid 1 2 3

Adj List (z,2,B,1,D) (z,2,B,1,D), (x,3,A,2,B) (x,3,A,2,B), (i,3,A,4,C), (x,3,A.5,B)

Vid 4 5 6 7 8

Adj List (i,3,A,4,C), (y,5,B,4,C) (x,3,A,5,B), (y,5,B,4,C), (x,6,A,5,B) (x,6,A,5,B), (p,7,E,6,A), (q,6,A,8,D) (p,7,E,6,A) (q,6,A,8,D)

TABLE I: Partition 1 TABLE II: Partition 2 Instead of arbitrary/random partitioning or partitioning using a power of 2 (as is done in Metis), we use range partitioning for its flexibility, superior performance during preprocessing, and effectiveness in substructure discovery [25]. We use the contiguous range of vertex ids (a range need not have all numbers in the range) to partition the input graph and create an adjacency list for each partition. In the example given in Figure 1, to partition the graph into two using given range information, we assign the first 3 nodes (vertex ids 1 to 3) to partition 1 and the other 5 nodes (vertex ids 4 to 8) to partition 2. With this, adjacency lists for those two partitions are created as shown in Tables I and II. Range partitioning takes one pass on the input graph and is done using Map/Reduce only once and the cost is amortized over processing as many queries on the graph as

needed. We also generate meta information during partitioning for computing heuristics to be used during query evaluation. In each partition, the information collected include query plan start nodes1 , intermediate query plan nodes, and connected components. Partitions generated after range partitioning contain partitioned graphs (could be a forest) and separately its adjacency list. As query expansion on these partitioned graphs needs corresponding adjacency list partition, we need to direct this partition to the processor holding the appropriate adjacency list. Without range information among vertex ids across the adjacency partitions, this operation requires the complete information of all vertices and their partition adjacency lists. Range information allows us to direct a partial answer to the correct partition in which the adjacency list of node is stored. Also, as we use the mapping between partition id and vertex id range stored as range information, this gives us finer control over the size of the partitions for heterogeneous processing resources. Correctness: As answers to a query can be part of multiple partitions and even come back to an already processed partition, several cases (shown in Figure 2) need to be systematically dealt with for answer correctness.

Fig. 2: Challenges for Querying Partitioned Graphs 1) All answers are within a single partition (Figure 2 (a)) 2) Answers span more than one partition (Figure 2 (b)) 3) Answers span one or more partitions that are already processed (Figure 2 (c)) In addition to the above, there are additional challenges to make sure answers are correct in the presence of non-unique node and edge labels (node ids are unique.) They are: Overlapping Answers: Figure 3 shows an example where the answer instances overlap. That is, the vertex or edge is part of more than one answer (due to their label.) Here, our query processing should identify all the instances as separate answers for correct results. Note that node B in partition P3 needs to be expanded twice once for each answer to get correct answers.

Fig. 3: Overlapping answers 1 Query sizes are significantly smaller compared to the graph/partition and

this information can be collected for multiple queries as well.

Multi-Path Query Plans: Based on query characteristics, a query plan can have more than one path from top to bottom. If the plan contains multiple traversal paths, then the complete answers should match the query edges present in all paths. Figure 4 shows a multiple path query plan and the issues with processing it. The partial answer found in partition P1 continues for node (14, B) in partition P2 and becomes start node in that partition. Then, we load and process all start nodes in partition P2. However, we need to keep track of which query plan node needs to be expanded (as there is another B node in the query plan). If we expand the node (14, B) in P2, thinking it is node (2, B) of query plan, the result will be incorrect (using the edge B → C from the query plan). Thus, without proper bookkeeping, we may find incorrect results. This situation warrants not only keeping track of which node to expand, but how it needs to be expanded with respect to query plan.

Fig. 4: Issue with multiple path query plan Duplicates in Query Expansion: Independent and one-edgeat-a-time expansion of the graph from query start node and continuing may lead to the generation of duplicates. Hence, duplicate elimination [26] is needed to find correct answers and reduce the cost of query expansion. Figure 5 shows an example of how duplicates are formed during query expansion. The circled instances highlighted in same color have the same set of vertex ids and edges between them. In the presence of duplicates, query processing not only uses resources, but also generates incorrect answers. We use canonical representation to identify duplicates. IV. Q UERY P ROCESSING A PPROACH AND I TS C ORRECTNESS Our approach uses the Map/Reduce framework to support querying graphs of arbitrary sizes. We use a parameter k to indicate the number of available resources (in our case Mappers). For a given k, we choose the best k partitions to be processed in parallel in each iteration using heuristics and evaluate a given query plan.

Fig. 5: Duplicates in Query Expansion Constrained Query Plan Expansion: In our approach, a query plan2 is used to determine the order in which nodes should be expanded to obtain the answer instances that match the given query. Constrained query expansion in a partition starts from initial instances of the start nodes and then we expand all instances one edge at a time, independently according to query plan. As we expand, if the query edge goes out of current partition, we update the corresponding partition with the partial answer instance. We do so until there are no more edges to visit in the query plan (or plans) or no more answers to process in that partition. Removing Duplicates: Duplicates have the same vertex ids and the same connectivity including node labels and edge direction and labels, if present. To remove duplicates we convert the expanded instance to its canonical form. We employ a lexicographic ordering on edge label for canonical form. If there are multiple edges with the same edge label in a substructure, they are ordered on the source vertex label. If the source vertex label is also the same, they are further ordered on the destination vertex label. If edge label, and vertex labels are also identical, then source and destination vertex ids are used for ordering. Canonical forms of duplicates are shown in Figure 5. Routing Partial Answers: When all query results are completely inside a single partition, complete answers are computed and stored while processing that partition. When query results span multiple partitions, we compute partial answers in one partition and carry that instance to the appropriate partition which has further query edges. We keep constructing the answers by routing it to appropriate partition and loading additional partitions until the query plan is completely matched. The complete answers are collected when the answer instances have all the edges according to the query plan. When an answer instance comes back to a partition that has already been processed, it is a special case of the above. As we carry instances and process additional partitions, we ensure correctness by loading those partitions as many times as needed. Handling Overlapping Answers: Our system makes sure that 2 Given a graph query, several plans are possible. Generating a low-cost query plan is discussed in [9].

our final result set contains all overlapping answers as separate instances. We compute query on each initial substructure systematically and if any answer goes out of partition, we carry them as separate instances and systematically keep finding complete answers. For the example discussed in Figure 3, the initial instances (5, A) and (6, A) will be carried separately to partition P3 and when we process partition P3, node (7, B) will be used for expanding both instances. Hence, all the overlapping answers come out correctly.

Fig. 6: Marking node for correctness Addressing Multi-Path Query Plans: Query plan nodes are numbered for this purpose. The issue with the multi-path query plan is handled by marking the node that continues to a different partition with the node id of the query plan node (which is unique). Figure 6 shows the relative marking of node 14 for our previous example (Figure 4). Once this partial answer continues to partition P2, we would know which query edge to expand further from the relative node id of the node and find correct results. In our case node (14:4, B) in the partial answer will be further expanded for query edge (14, B) → (15, D). A. Input Query and Plan Representations Apart from the input graph, the query is given to our system as a graph. Of course, the query is a much smaller graph as computed to the input graph and has query-related decorations, such as wild cards, Boolean conditions, BETWEEN operator, and Boolean AND and OR specifications. Nodes and edges are labeled (including wild cards). This query is converted into a query plan with a root node and other nodes and edges taken from the query. Figures 7 and 8 show two queries expressed as graphs along with their English description and one of the plans generated for each query. We have developed a query optimizer in [9] which uses a catalog generated for the input graph to perform a cost-based estimation of the plan. For this, we are using a plan that does not use a wildcard as the root node or top-level nodes as they match a large number of alternatives. We support the OR operator by combining the results of two plans as it corresponds to a UNION query. An example query is shown for each data set along with the plan. These query results are further elaborated in Section VII. The plan is input as a text file (with extension .g) indicating node ids and their labels, followed by edges along with edge labels and nodes connected to them. Direction of edges can also be specified although we are using undirected plans here. Answer computation starts from the root nodes and generates all matches for the rest of the query plan. The search is constrained and does not generate node and edge labels not present in the plan.

B. Correctness of the Proposed Approach For correctness, we need to find all exact matches of the query plan in the whole graph. As we have partitions of the graph, the result set should have answers from within a partition or across partitions. In our approach, we have addressed each case of answers spanning partitions. Also depending on query characteristics, we have addressed both single path and multiple path query plans. Overall the correctness of the query processing is operationally ensured by i) Using Meta Information File (MIF) to correctly identify partitions that have start nodes (whether initial or continuation), ii) Updating Meta Information File to include partitions that have continuation or to remove partitions which have been processed, iii) Carrying intermediate results (instances with marked nodes) to the correct graph partition which will then load respective adjacency list for further expansion, and iv) Constructing partial answers and continuing expansion systematically from the last visited node. We have tested the correctness empirically (apart from the above operational correctness) as well by embedding graphs with known frequency and different structures in generated synthetic graphs using Subgen [27] and evaluating complete or partial queries of embedded graphs. It is not included due to space constraints. V. M AP /R EDUCE A RCHITECTURE FOR PARTITIONING AND Q UERY P ROCESSING This section discusses the Map/Reduce algorithm for query processing. Input query, which is a graph (see Figures 7 and 8), is converted into a representation of edges and labels (including direction and wild cards) for processing. An input query is very small in terms of number of nodes and edges. Figure 9 shows the overall query processing architecture using Map/Reduce. The input graph is range partitioned based on the resources available using the first pair of Map/Reduce processing. Meta Information File (MIF) is generated during partitioning, collecting query start and potential continuation nodes in each partition. Based on k, partitions are chosen using heuristics (discussed in Section VI.) Choosing partitions and processing a query continues until there are no more partitions to be processed. Map/Reduce query processor takes a query plan, adjacency list partition files and generates partial/complete answers as well as updates MIF for the next iteration. When finished all complete answers are written out.

Fig. 9: MR-GQP System Architecture

Fig. 7: Query Graph and Plan for DBLP Q3: Find all conference papers along with the conference names that are published in 2005 or 2008 or 2009 by the author whose first name start with ‘John’

Fig. 8: Query Graph and Plan for IMDB Q3: Find movies along with their actors and producer information by company American Broadcasting Company (ABC) after year 1970 where the genre should not be comedy or horror.

Input Graph Representation: Our input graph is represented as a sequence of unordered edges (or 1-edge substructure). This makes it easier to distribute the graph partitions among many machines. Both undirected and directed graphs can be used for our approach but we have considered only directed graphs in our system as it has relations explicitly represented using directions. Table III shows the edge input representation of the graph in figure 10. Each edge is completely represented by a 5-element tuple ⟨edge label, source vertex id, source vertex label, destination vertex id, destination vertex label ⟩. Both partitioning and query processing are done using the Map/Reduce framework. The driver program that runs Map/ Reduce jobs is implemented in Java. The configuration file with parameters is passed to the driver program. Range partitioning is done in a single iteration of Map/Reduce, which generates graph partition files and adjacency list partition files. In addition, if query plans are available, partitioning generates an initial Meta Information File (MIF). MIF contains eligible partitions for query evaluation. After MIF is generated, our driver program applies the specified heuristic using the contents of MIF and chooses top-k partitions. When the chosen

Edges (z,2,B,1,D) (x,3,A,2,B) (i,3,A,4,C) (x,3,A,5,B) (y,5,B,4,C) (x,6,A,5,B) (p,7,E,6,A) (q,8,D,6,A)

TABLE III: Graph Edge List Representation Fig. 10: Example Input Graph

partitions set becomes empty, then query evaluation has been completed and we terminate the computation. Else we would have k or less partitions to process at a time, using which a Map/Reduce based query processing does query expansion on those partitions in parallel and outputs the complete answers

to Final All Answers (FAA) file and writes the intermediate answers to the relevant graph partition files for further expansion. During the processing of partitions, MIF is updated and will be used by the driver program to select the next set of partitions to be loaded in the next iteration. This is continued until eligible partitions set in MIF are empty. A. Range Partitioning using Map/Reduce Algorithm 1 gives the process of range partitioning using Map/ Reduce. Algorithm 1 Range Partitioning INPUT: Graph edge list, Range info, Query Plan OUTPUT: Graph partitions, Adjacency list partitions, MIF Class Mapper 1: function MAP(key = line no, value = (one) edge instance) 2: get the source vertex id(sV id) and destination vertex id(dV id) from edge 3: emit(newKey = sV id, newValue = edge instance) 4: emit(newKey = dV id, newValue = edge instance) 5: end function Class Partitioner 6: function GET PARTITION(key = vertex id, value = one edge instance) 7: reducerId = get partition id of vertex from Range Info 8: return reducerId 9: end function Class Reducer 10: function SETUP 11: pId = get id of current reducer 12: partitionRange = get range of partition pId from Range Info 13: edgeList = set to hold unique edges in current partition 14: create new files for graph partition(pId) and MIF 15: end function 16: function REDUCE(key = vertex id, valueList = list of adjacent edges) 17: get vertex id(vId) from key and vertex label(vLabel) from valueList 18: if vLabel is Query Plan start node then 19: update MIF with ⟨ partition id = pId, label = vLabel, count = 1, 20: is start node = True ⟩ 21: else if vLabel is Query Plan intermediate node then 22: update MIF with ⟨ partition id = pId, label = vLabel, count = 1, 23: is start node = False ⟩ 24: end if 25: create initial substructure ⟨vId, vLabel⟩ and write out to graph partition(pId) 26: append every edge from valueList to edgeList 27: create adjacency list (AL) for vertex (vId) 28: emit(newKey = vId, newValue = AL) 29: end function 30: function CLEANUP 31: S = make sets for all vertices in current partition(pId) 32: for each edge < sV id, dV id > do in edgeList 33: if sVid and dVid not in same set then 34: union both sets of sVid and dVid 35: end if 36: end for 37: count of connected components(CC) = total unique sets in S 38: update MIF with ⟨ partition id = pId, connected components = CC ⟩ 39: end function

Split and Shuffle Edges: Based on the number of mappers available, the Map/Reduce paradigm divides the input graph edge list into input splits and given as the input to the mapper. The map function then processes a single edge at a time and for both the vertex ids in the edge, it outputs the vertex id as the key and the edge instance as the value (line 1 to 5).

The shuffle phase sorts the mapper output based on vertex ids and groups all the records that have the same vertex id. As we want to divide output records based on range information, we use the custom partitioner that partitions keys with a custom hash function based on range information (lines 6 to 9). This will route all the output records with vertex id in the same range to the same reducer. Creating Graph Partition and Adjacency List Partition: Once the reducer gets the desired set of keys, we create graph partition files that contain initial substructures and adjacency list partitions that contain the adjacency list of all the vertices in the same partition. First, inside the setup of reducer, we create new graph partition file and local MIF (line 14). After that, the reduce function gets key as vertex id and value list as all adjacent edges from that vertex (line 16). For each vertex, we create initial substructure and output it to the corresponding graph partition file (line 25). Also, for each vertex, a single adjacency list is created. And finally, the reducer emits key as vertex id and value as adjacency list (line 27 and 28). The emitted keys that belong to the same partition are written to a single file creating the adjacency list partition. Collecting Meta Information and creating MIF: As we create initial substructures inside the reducer, we compare them to the query plan start node or intermediate nodes and if there is a match, we update the occurrences of those instances in the current partition in MIF (lines 18 to 24). Also, as the reducer gets all the edges within single partition, we collect all the edges (line 26) and in reducer cleanup function, using every edge in a partition, we find connected components using the find and union algorithm and count of connected components in that partition is written to MIF (lines 31 to 38). As each reducer processes separately, local MIF for each reducer will be created and then the driver program collects them to create global MIF. B. Query Processing using Map/Reduce

Fig. 11: Query Processing using Map/Reduce Algorithm 2 presents the algorithms for the mapper and reducer classes for query processing. The Map/Reduce block diagram has been illustrated in Fig. 11. The top k or less graph partition files chosen from eligible partitions after applying heuristics are taken as input in each query processing iteration. Query expansion inside mapper: For query processing, input to each mapper is a graph partition that contains initial substructures or partial answer instances. In the setup function, the mapper loads the respective adjacency list partition and query plan (lines 3 and 4). We then create a set to hold

unique completed answers from the current partition (line 5). The map function takes the key as the line number and value as the initial substructure or partial answer instance (line 7). Then we start query expansion for those instances that have query plan start node (line 8) and do a systematic expansion for every query edge present in the query plan and store the intermediate substructures in a set (lines 9 to 32). In the case of a partial answer instance, we will continue expansion only if the last visited node has been marked with the relative id of the current query node (line 16). To expand a node, we use the adjacency list of that node and according to the query plan, find all edges that match with the current query edge and expand the substructure with that edge (lines 18 and 19). If the newly expanded node belongs to a different partition, then we mark that node relative to the last visited query node in the query plan and the mapper emits a new partition id as key and marks partial answer instance as the value (lines 25 and 26). Otherwise, if the expanded node is in the same partition, we create a canonical instance of the answer and store them in the answers set to remove duplicates (lines 20 to 22). We then continue with the next query edge from the query plan. When there is no more edge to expand from the query plan and no more partitions to be loaded, our answer set contains all complete answers. We collect them in the global answer set for that mapper (line 30). In the cleanup function, the complete answers are written in the Final All Answers (FAA) file (line 34). Aggregating partial answers and updating graph partitions in reducer: After mappers have been processed, the shuffle phase groups the output of mappers on the partition id and sends them to the corresponding reducer. The reducer processes key-value lists generated for each partition. The reduce function gets the key as partition id and value list containing partial answers in that partition (line 36). For each partial answer substructure instance present in the list, it updates the MIF to include continuing nodes in appropriate partitions (lines 37 to 40). According to partition id, it then loads the corresponding graph partition file from HDFS and write out partial answers in that file (line 41). VI. H EURISTICS FOR S CHEDULING PARTITIONS Both meta-information (collected at the time of partitioning and in each iteration) as well as heuristics based on them are leveraged to find all answers efficiently. Heuristics are used for: i) identifying which k partitions need to be loaded in each iteration and ii) loading the minimum number of required partitions (totally) to answer a query. This is validated experimentally. A. Heuristics Derived From Meta Information: In this paper, we have proposed three heuristics to determine an efficient order of loading the k partitions for query processing. Random loading of partitions for evaluating a query (RANDOM heuristic) is used as a baseline heuristic against which others are compared. Initially, we tried using individual metrics, such as MIN SN (MAX SN) or minimum

Algorithm 2 Map/Reduce-Based Query Processing INPUT: Top k graph partitions and adjacency list partitions, Query Plan OUTPUT: Updated graph partitions, Final answers file(FAA), Updated MIF Class Mapper 1: function SETUP 2: currentP artitionId = partition Id from currently loaded graph partition 3: adjMap = load corresponding adjacency partition 4: load Query Plan file 5: finalAnswers = Set to hold completed answers 6: end function 7: function MAP(key = line no, value = initial or partial answer instance) 8: if value contains Query Plan start node then 9: answerSet = Set to hold unique substructures after each query edge 10: expansion 11: add initial value substructure to answerSet 12: while all the nodes and edges in the plan get visited do 13: for ans substructure in answerSet do 14: get last visited node (v) from ans 15: if node (v) has been marked and doesn’t match with Query node 16: then continue for next query edge 17: aL = get adjacency list of v from adjMap 18: expand the current node to the desired node by adding edge e 19: from aL based on the plan 20: if pId of expanded node matches currentPartitionId then 21: create canonical instance newAns 22: remove ans from answerSet and add newAns to answerSet 23: else 24: mark expanded node in ans substructure with relative id of last visited query node 25: emit(newKey = pId of expanded node, newValue = 26: expanded ans substructure) 27: end if 28: end for 29: end while 30: finalAnswers ∪ answerSet 31: end if 32: end function 33: function CLEANUP 34: write out finalAnswers to FAA file in HDFS 35: end function Class Reducer 36: function REDUCE(key = partition Id, valueList = intermediate results) 37: for value in valueList do 38: get the last expanded vertex (v) from the value 39: update MIF file with ⟨ partition id = key, label = null, count = 1, 40: is start node = True ⟩ 41: write out value to graph partition (pId) in HDFS 42: end for 43: end function

(maximum) number of start nodes in a partition and MIN CC or minimum number of connected components. As each of them individually does not capture the partition characteristic, we have combined them using a weighted formula and normalization to get better heuristics. In the experimental section, we compare both individual and composed heuristics to highlight the need for combining multiple individual heuristics to get better ones3 . 1) SN CC: Here the idea is to give higher loading/scheduling priority to the partitions that contain more 3 For the composition of heuristics, based on the intuition, many compositions were tried before arriving at ones which better captured the essence of the partition characteristic.

Data Set DBLP IMDB Synthetic DBLP small IMDB small

number of start nodes and less number of connected components as they mostly will contain the maximum number of query answer instances. Based on this intuition, the SN CC heuristic is calculated as follows: HSN CC = (e−0.04∗CC ) ∗ S

(1)

where in a given partition, S is the number of start nodes and CC is the number of connected components. CC is used to dampen the effect of start nodes a bit. With this formula, the partition containing more start nodes and less connected components is given a higher heuristic value. But for partitions that have more connected components, the heuristic value decreases which is what we want. Partitions with a higher value for this heuristic will be given higher priority for loading. 2) SN CC2k : From the set of eligible partitions, first, top 2k partitions are chosen with maximum number of start nodes. From this set, k partitions are selected that have the least number of connected components. Thus, this heuristic utilizes both start nodes (MAX SN) and connected components (MIN CC) in each partition to choose the top k partitions. 3) SN IN: In addition to start nodes in each partition, in this heuristic we consider the query plan intermediate nodes to choose top-k partitions. The partitions having large number of intermediate nodes compared to start nodes will have potentially more answers (though partial) and will be needed multiple times during processing. The formula used for this heuristic is: S HSN IN = ln(S + 1) ∗ S + I2

Our experimental analysis compares all of the above heuristics to establish an efficiency pattern with respect to the choice of heuristic. VII. E XPERIMENTAL A NALYSIS Processing Environment: All the experiments have been performed using Java with Hadoop on Comet cluster at SDSC (San Diego Supercomputer Center (https://www.sdsc. edu/services/hpc/hpc systems.html). The Comet Cluster has 1944 nodes and each node has 24 cores (built on two 12-core Intel Xeon E5 2.5 GHz processors) with 128 GB memory, and 320GB SSD for local scratch space.

#Vertices 3.3M 0.7M 400K 12K 11K

#Edges 13.5M 3.5M 1.2M 30K 33K

TABLE IV: Data Set Description Query Name

English Query

Key Operator

#Ans

DBLP Queries DBLP Q1

Find papers along with their authors, conferences, and the publication year, which are published between 1990 and 1995 and the paper name includes the term “Transaction Management in Multidatabase Systems”

Range (BETWEEN)

18

DBLP Q2

Find papers along with their authors and publication year, which are published after 1995 in the conference VLDB or SIGMOD and the paper name includes the term “Transaction Management in Multidatabase Systems”

Comparison (>)

45

DBLP Q3

Find all conference papers along with the conference names that are published in 2005 or 2008 or 2009 by the author whose first name starts with ‘John’. (Query Graph and Plan in Figure 7)

Special Character (%)

7685

IMDB Queries IMDB Q1

Find movies produced by “Adam Sandler” where the genres should be comedy and Sci-Fi but the year should not equal to 2000

AND

24

IMDB Q2

Find all actors (both Male and Female) who have worked in a movie owned by the company “Paramount Pictures” where the genre of the movie is either Drama or Action.

Union (OR)

6824

IMDB Q3

Find movies along with their actors and producer information by company American Broadcasting Company (ABC) after year 1970 where the genre should not be comedy or horror. (Query Graph and Plan in Figure 8)

Combination (>, NOT)

7252

(2)

where for a given partition, S is the number of start nodes and I is the number of intermediate nodes. Thus, the higher the value of SN IN for a partition, the higher will be its loading priority. It gives preference to the higher number of start nodes, but also takes into account the number of intermediate nodes. If the number of intermediate nodes is 0 or small, it does not have much effect. If the number of intermediate nodes increases, the heuristic value decreases.

#Partitions 120 60 10 3 3

TABLE V: DBLP and IMDB Queries Data Set Description: For this work labeled graphs are required on which query processing can be performed. As shown in Table IV, we have used real-world (DBLP and IMDB) and synthetic graphs to show the effectiveness and scalability of our approach with the largest DBLP graph having 3+ million nodes and 13+ million edges. The DBLP data set contains information about authors, papers, conferences, and years as vertex labels and edges showing relationships among these vertex labels. Similarly, the IMDB data set contains information about movies, year, genre, person, gender, and

movie company as vertex labels and edges represent the relationship among these vertex labels. The synthetic graph contains 2000 unique vertex labels and 4000 unique edge labels. Correctness of Results: For establishing the correctness of our algorithms, we have used an existing system, QP-Subdue, which is a non-partitioned, main-memory query processor for graph databases [9]. However, QP-Subdue cannot handle larger graph sizes. Hence we have used synthetic data sets with embedded subgraphs as well as smaller sizes of both DBLP and IMDB graphs to verify the correctness, empirically, as shown in table IV for DBLP small and IMDB small. For the DBLP and IMDb graph data sets, we took queries having different characteristics (shown in Table V). We have used queries with comparison operator, wild cards (%), range queries (BETWEEN), AND query, OR query and combination of logical and comparison operators (>, NOT). Figure 7 shows a sample query graph for the DBLP wild card query (DBLP Q3). A. Results of Partitioned Graph Querying Speedup with number of partitions processed in parallel in any iteration (k): The speedup achieved for different DBLP and IMDb queries by increasing the value of k with respect to response time is shown in Figure 12a. The purpose of this experiment is to show the change in response time with respect to resources used. As can be seen, for DBLP Q1 (BETWEEN operator query) there is significant improvement in response time (starting with 52%) until 64 partitions (k) and tapers off after that. For the wild card query (DBLP Q3), shown in Figure 12b, even at 120 partitions there is significant improvement as the number of answers is very large due to wildcard (see Table V last column)! Similar speedups are also observed for the IMDb queries, shown in Figure 12c (IMDb Q2 - OR Query) and Figure 12d (IMDb Q3 - Combination Query). Effect of number of partitions for given mappers: There are three parameters that affect the response time: i) the number of partitions (based on processor memory availability) or map tasks, ii) the number of physical mappers available for processing, and iii) the number of partitions loaded in each iteration. The total number of partitions used in Figure 13 is 120. Ideally number of partitions that can be used for each iteration needs to match the number of available mappers. When we cannot match # of partitions in an iteration to the # of mappers, multiple partitions have to be processed by each mapper in each iteration. In this experiment, we are interested in the effect on response time as we increase the k value (# of partitions processed in each iteration) for a given number of mappers (fixed resources). We have used the Wild Card (%) query here as it generates the maximum number of answers (DBLP Q3). Figure 13 shows two things: i) as the number of mappers (or resources) increases, response time goes down even if multiple partitions are processed due to an increase in k by each mapper (groups with different mapper values) and ii) when the number

of mappers is held constant and k is increased making more partitions to be processed in each iteration, response time only increases slightly as multiple partitions are processed by the mappers (2nd and 3rd bars compared to the 1st bar in each set). Also, as the number of mappers increases, shuffle time increases which comes out as a bit higher value of response time as mapper size reaches 32. The takeaway is to match the number of partitions loaded in each iteration to the number of mappers available, if possible, for the best response time. Effect of Heuristics: Instead of scheduling partitions for execution randomly, it would be better if we can reduce the total number of partitions loaded using heuristics. We have done these experiments to see the effect of each heuristic (discussed in Section VI) on query processing for all queries of DBLP and IMDb graphs. We only show the wild card (DBLP Q3) and combination (IMDB Q3) queries due to space constraints. For the initial analysis, we chose k = 1 and compared the response time for each heuristic. To see the effect of k, we increased k to 2, 4, 8, and 16 given that we have that many mappers available. Number of reducers for this set of experiments are fixed to 2. For the wild card (%) query (Figure 14a) and the combination query (Figure 14b), it is observed in terms of execution time that the worst heuristics are RANDOM, MIN SN, MIN CC and MAX SN, while the best heuristics are SN CC2k , SN CC and SN IN. Thus, it has been showcased that the proposed composed heuristics perform much better than the individual heuristics. B. Load Factor Analysis Based on the chosen k value, we need to identify which k partitions need to be loaded in each iteration. In the ideal case, we want to load the minimum number of required partitions to answer a query. A required partition is one in which one or more query plan nodes exist. In processing a query over a partitioned graph, minimizing the total number of partitions loaded over all the iterations can be used as an efficiency measure. The optimal case is loading the number of required partitions only once. That is, the optimal load factor must be 1. Load Factor: Apart from the number of partitions loaded, the query processing can take more iterations or fewer iterations depending upon the chosen value of k (and the partitions used in each iteration.) The ideal number of required iterations is when we load only the required partitions. We define ‘Load Factor’ which measures the efficiency of query processing by comparing the number of iterations needed for the ideal case. The formula for load factor is defined as follows (illustrated in Figure 15): Load Factor =

Total iterations taken for query processing Required number of iterations

(3)

where, Required number of iterations = ⌈

unique partitions that has answers ⌉ k (4)

(a) DBLP (BETWEEN query)

(b) DBLP (Wild Card (%) query)

(c) IMDB (OR query)

(d) IMDB (Combination query)

Fig. 12: Response time Comparison of queries as k increases

Fig. 13: Effect of changing k for given mappers

The closer the value of load factor is to 1 the better the scheduling based on the chosen heuristic. We propose a suite of heuristics to identify which is better according to graph or query characteristics. If there are p′ required number of partitions and for a given k value, H(a) loads p′ +a partitions and H(b) loads p′ +b partitions, then we can say H(a) is better than H(b) if a < b. In other words, the less the number of repeated partitions (or load factor), the better the heuristic. We further extend our heuristics comparison by taking the load factor for each heuristic into consideration. Figures 16a and 16b show the load factor values for each

heuristic for our given DBLP and IMDb queries, respectively. For instance in figure 16a, for k = 1, in case of BETWEEN query, SN CC has the lowest load factor value i.e. it requires less number of iterations and hence, less partitions are loaded for this heuristic. SN CC2k is the second best heuristic and SN IN is the third best heuristic in terms of Load Factor. Here, the load factor for the RANDOM heuristic is more than any other. Hence more partitions are loaded for the RANDOM heuristic, making it inefficient. The second worst is MIN SN. For all other DBLP and IMDb queries, the top 3 heuristics that have the lowest load factor values are SN CC, SN IN, and SN CC2k . The presented analysis for load factor for all queries also matches with our previous heuristics comparison. We also see a similar pattern in load factor values for every value of k. In every scenario, the top 3 heuristics are SN CC, SN IN and SN CC2k . In summary, we have conducted extensive experiments on radically different data sets to present the results of our findings. Unfortunately, we are able to present few results due to space constraints. The queries that we have validated are quite expressive and are comparable to relational queries. We have also validated our approach for scalability of computation and verified the efficacy of proposed heuristics. We have shown speedup with respect to the number of mappers used (or k), compared all heuristics for all query types, and compared load factor for all heuristics. We have identified three heuristics

(a) DBLP (Wild Card (%) Query) (a) DBLP Queries

(b) IMDb (Combination Query)

Fig. 14: Heuristics Comparison (b) IMDb Queries

Fig. 16: Load Factor Analysis

Fig. 15: Load Factor Analysis: Optimal vs Non-optimal partitions loading

(SN CC, SN CC2k and SN IN) that do better for all query types. VIII. C ONCLUSIONS In this paper, we have proposed and demonstrated a scalable approach to efficient graph query processing using the Hadoop Map/Reduce framework that correctly computes the answers for different types of graph queries (including wild cards) efficiently. We have established the operational correctness of the algorithm as well in addition to experimental validation. Experimental results on large real-world and synthetic data sets show the effectiveness and efficiency of this approach.

Future work includes extending this framework for additional cost-based optimizations as is commonly done in the relational model. Also, techniques for partitioning the graph for a workload (of multiple queries) and processing multiple queries in parallel (rather than sequentially and in addition to processing a single query in parallel) need to be developed. Whether partitioning changes make any difference for workloads with multiple queries needs to be investigated as well. ACKNOWLEDGMENTS For this work, Dr. Sharma Chakravarthy was partly supported by NSF Grants CCF-1955798 and CNS-2120393.

R EFERENCES [1] Internet Live Stats, “Facebook Active Users Stats,” https://www. internetlivestats.com/watch/facebook-users/, 2020, online; Accessed: 24May-2020. [2] DBLP, “DBLP Data Stats,” https://dblp.uni-trier.de/statistics/ recordsindblp, 2020, online; Accessed: 24-May-2020. [3] IMDb, “IMDb Data Stats,” https://www.imdb.com/pressroom/stats/, 2020, online; Accessed: 24-May-2020. [4] J. Bodra, “Processing queries over partitioned graph databases: An approach and it’s evaluation,” Master’s thesis, The University of Texas at Arlington, May 2016. [Online]. Available: http: //itlab.uta.edu/students/alumni/MS/Jay D Bodra/JBod MS2016.pdf [5] S. Sakr and G. Al-Naymat, “Graph indexing and querying: A review,” IJWIS, vol. 6, pp. 101–120, 06 2010. [6] R. Giugno and D. Shasha, “Graphgrep: A fast and universal method for querying graphs,” in Object recognition supported by user interaction for service robots, vol. 2. IEEE, 2002, pp. 112–115. [7] H. Tong, C. Faloutsos, B. Gallagher, and T. Eliassi-Rad, “Fast best-effort pattern matching in large attributed graphs,” in Proceedings of the 13th ACM SIGKDD international conference on Knowledge discovery and data mining, 2007, pp. 737–746. [8] S. Arifuzzaman, M. Khan, and M. V. Marathe, “Patric: a parallel algorithm for counting triangles in massive networks,” in CIKM, 2013, pp. 529–538. [9] S. Das, A. Goyal, and S. Chakravarthy, “Plan before you execute: A costbased query optimizer for attributed graph databases,” in International Conference on Big Data Analytics and Knowledge Discovery. Springer, 2016, pp. 314–328. [10] J. Bodra, S. Das, A. Santra, and S. Chakravarthy, “Query processing on large graphs: scalability through partitioning,” in International Conference on Big Data Analytics and Knowledge Discovery. Springer, 2018, pp. 271–288. [11] X. Yan, P. S. Yu, and J. Han, “Graph indexing: a frequent structure-based approach,” in Proceedings of the 2004 ACM SIGMOD international conference on Management of data, 2004, pp. 335–346. [12] J. Cheng, Y. Ke, W. Ng, and A. Lu, “Fg-index: towards verification-free query processing on graph databases,” in Proceedings of the 2007 ACM SIGMOD international conference on Management of data, 2007, pp. 857–872. [13] J. Cheng, Y. Ke, and W. Ng, “Efficient query processing on graph databases,” ACM Transactions on Database Systems (TODS), vol. 34, no. 1, pp. 1–48, 2009. [14] R. Angles, M. Arenas, P. Barceló, A. Hogan, J. Reutter, and D. Vrgoč, “Foundations of modern query languages for graph databases,” ACM Computing Surveys (CSUR), vol. 50, no. 5, pp. 1–40, 2017. [15] E. Prud’hommeaux, “Sparql query language for rdf, w3c recommendation,” http://www. w3. org/TR/rdf-sparql-query/, 2008. [16] Neo4j, “Cypher Query Language,” https://neo4j.com/developer/cypher/, 2011. [17] Neo4j-Inc, “Neo4j,” https://neo4j.com, 2007. [18] Z. Yang, L. Lai, X. Lin, K. Hao, and W. Zhang, “HUGE: an efficient and scalable subgraph enumeration system,” CoRR, vol. abs/2103.14294, 2021. [Online]. Available: https://arxiv.org/abs/2103.14294 [19] K. Ammar, F. McSherry, S. Salihoglu, and M. Joglekar, “Distributed evaluation of subgraph queries using worst-case optimal and lowmemory dataflows,” Proc. VLDB Endow., vol. 11, no. 6, pp. 691–704, 2018. [Online]. Available: http://www.vldb.org/pvldb/vol11/ p691-ammar.pdf [20] A. Goyal, “Qp-subdue: Processing queries over graph databases,” 2015. [21] T. Gaasterland, “Cooperative answering through controlled query relaxation.” IEEE Expert Systems, vol. 12, no. 5, pp. 48–59, 1997. [22] P. Sanders and C. Schulz, “Engineering multilevel graph partitioning algorithms,” in Algorithms - ESA 2011 - 19th Annual European Symposium, Saarbrücken, Germany, September 5-9, 2011. Proceedings, 2011, pp. 469–480. [23] C. Schulz. (2013) High quality graph partitioning.phd thesis. [Online]. Available: http://algo2.iti.kit.edu/schulz/dissertation christian schulz.pdf [24] G. Karypis and V. Kumar, “A fast and high quality multilevel scheme for partitioning irregular graphs,” SIAM J. Sci. Comput., vol. 20, no. 1, pp. 359–392, Dec. 1998.

[25] S. Das, “Divide and Conquer Approach to Scalable Substructure Discovery: Partitioning Schemes, Algorithms, Optimization and Performance Analysis Using Map/Reduce Paradigm,” Ph.D. dissertation, The University of Texas at Arlington, May 2017. [Online]. Available: http://itlab.uta.edu/students/alumni/PhD/Soumyava Das/SDas PhD2017.pdf [26] S. Das and S. Chakravarthy, “Duplicate reduction in graph mining: Approaches, analysis, and evaluation,” vol. 30, no. 8, 2018, pp. 1454– 1466. [27] D. Cook and L. Holder, “Subgen: Synthetic Graph Generator,” http: //ailab.wsu.edu/subdue, 2007.

Related documents

• Scaling Graph Query Processing: Focused Exploration on Partitions Using Load Balancing#1447382
• A Monte Carlo tree search algorithm for optimization of load scalability in database systems#956229
• DBRepro: Automated Database Synthesis via a Hybrid Constraint-Solving Approach for Reproducing Slow Queries#618639
• Enabling Data Dependency-based Query Optimization#158629
• Efficient load-balancing through distributed token dropping#384952