A Survey and Experimental Analysis of Distributed Subgraph Matching
Longbin Lai Zhu Qing Zhengyi Yang Xin Jin Zhengmin Lai Ran Wang Kongzhang Hao Xuemin Lin Lu Qin Wenjie Zhang Ying Zhang Zhengping Qian Jingren Zhou

Improved Distributed Approximations for Maximum-Weight Independent Set
    Ken-ichi KawarabayashiNII, Japan, 
		Seri KhouryUniversity of California, Berkeley,  seri_khoury@berkeley.eduAaron SchildUniversity of California, Berkeley, Gregory SchwartzmanNII, Japan, 
    
================================================================================================================================================================================================




Recently there emerge many distributed algorithms that aim at solving subgraph matching at scale. Existing algorithm-level comparisons failed to provide a systematic view to the pros and cons of each algorithm mainly due to the intertwining of strategy and optimization. In this paper, we identify four strategies and three general-purpose optimizations from representative state-of-the-art works. We implement the four strategies with the optimizations based on the common dataflow system for systematic strategy-level comparison. Our implementation covers all representation algorithms. We conduct extensive experiments for both unlabelled matching and labelled matching to analyze the performance of distributed subgraph matching under various settings, which is finally summarized as a practical guide. 








§ INTRODUCTION


Given a query graph Q and a data graph G, subgraph matching is defined as finding all subgraph instances of G that are isomorphic to Q. In this paper, we assume that the query graph and data graph are undirected [Our implementation can seamlessly handle directed case.] simple graphs, and may be unlabelled or labelled. In this work, we mainly focus on unlabelled case given that most distributed algorithms are developed under this setting. We also demonstrate some results of labelled case due to its practical usefulness. Subgraph matching is one of the most fundamental operations in graph analysis, and has been used in a wide spectrum of applications <cit.>. 

As subgraph matching problem is in general computationally intractable <cit.>, and data graph nowadays is growing beyond the capacity of one single machine, people are seeking efficient and scalable algorithms in the distributed context. Unless otherwise specified, in this paper we consider a simple hash partition of the graph data, that is the graph is randomly partitioned by vertices, and each vertex's neighbors will be placed in the same partition. 

By treating query vertices as attributes and the matched results as relational tables, we can express subgraph matching via natural joins. The problem is accordingly transformed into seeking optimal join plan, where the optimization goal is typically to minimize the communication cost. In this paper, we focus on an in-depth survey and comparison of representative distributed subgraph matching algorithms that follow such join scheme.



 §.§ State-of-the-arts.
 
In order to solve subgraph matching using join, existing works studied various join strategies, which can be categorized into three classes, namely “Binary-join-based subgraph-growing algorithms" (), “Worst-case optimal vertex-growing algorithms" () and “Shares of Hypercube" (). We also include category for algorithms that do not clearly belong to the above categories. 

. The strategy computes subgraph matching via a series of binary joins. It first decomposes the original query graph into a set of join units whose matches can serve the base relation of the join. Then the base relations are joined based on a predefined join order. The algorithms differ in the selections of join unit and join order. Typical choices of join unit are star (a tree of depth 1) in <cit.>, (an edge or intersection of two edges) in <cit.>, and clique (a graph whose vertices are mutually connected) in <cit.>. Most existing algorithms adopt the easier-solving left-deep join order <cit.> except , which explores the optimality-guaranteed bushy join <cit.>.  
 
. Given {v_0, v_1, ⋯, v_n} as the query vertices, strategy first computes all matches of {v_0} that can present in the results, then matches of {v_0, v_1}, and so forth until constructing the results. Ngo et al. proposed the worst-case optimal join algorithm <cit.>, based on which Ammar et al. implemented in dataflow system <cit.> and showed its worst-case optimality <cit.>. In this paper, we also find out that the algorithm (with “overlapped decomposition” [Decompose the query graph into join units that are allowed to overlap edges]) is also a variant of GenericJoin, and is hence worst-case optimal. 


. strategy treats the computation of the query with n vertices as an n-dimensional hypercube. It partitions the hypercube across w workers in the cluster, and then each worker can compute its own share locally with no need of exchanging data. As a result, it typically renders much less communication cost than that of and algorithms. adopts the idea of for subgraph matching. In order to properly partition the computation without missing results, needs to duplicate each edge in multiple workers. As a result, can almost carry the whole graph in each worker for certain queries <cit.> and thus scale out poorly. 

. Shao et al. proposed <cit.> that processes subgraph matching via breadth-first-style traversal. Staring from an initial query vertex, iteratively expands the partial results by merging the matches of certain vertex's unmatched neighbors. It has been pointed out in <cit.> that is actually a variant of . Very recently, Qiao et al. proposed <cit.> that aims at resolving the “output crisis” by compressing the (intermediate) results. The idea is to first compute the matches of the vertex cover of the query graph, then the remaining vertices' matches can be compressed as intersection of the vertex cover's neighbors to avoid costly cartesian product.


We summarize the algorithms, the categories and their relationships in algorithms.




Optimizations. Apart from join strategies, existing algorithms also explored a variety of optimizations, some of which are query- or algorithm-specific, while we spotlight three general-purpose optimizations, , and . aims to divide the whole computation into sub-tasks that can be evaluated independently in order to save resource (memory) allocation. precomputes and indices the triangles (3-cycles) of the graph to facilitate pruning. attempts to maintain the (intermediate) results in a compressed form to reduce resource allocation and communication cost.


Batching. Subgraph matching, especially unlabelled, can produce massive (intermediate) results, and cause huge memory burden. is a way to relieve the burden via divide-and-conquer. The actual implementations may vary, but the simple idea is: one picks up the initial query vertex and divides its candidate matched data vertices (in unlabelled matching, it is the whole data vertex set) into b shares; then the computation is divided into b independent parts, each of which matches the initial query vertex to one share of data vertices. By expectation, we can reduce the resource allocation to 1/b while applying .

TrIndexing. This optimization precomputes and indices the triangles (3-cycle) of the graph to facilitate candidate pruning. In addition to the default hash partition, Lai et al. proposed “triangle partition” to also incorporate the edges among the neighbors (it forms triangles with the anchor vertex) in the partition. “Triangle partition” allows to use as the join unit, which greatly reduces the intermediate results of certain queries and improves the performance. As will be shown very soon, “triangle partition” in de facto is a storage-efficient variant of . In the experiment of <cit.>, the authors showed that can also utilize to improve the performance, while the authors only demonstrated this optimization from the “4-clique” query.

Compression. Subgraph matching is a typical combinatorial problem, and can easily produce results of exponential size. aims to maintain the (intermediate) results in a compressed form to reduce resource allocation and communication cost. Lai et al. proposed “clique compression” for , observing that a large clique in the data graph naturally contains many small cliques that can match join unit. In <cit.>, Qiao et al. proposed to study in subgraph matching in general. The idea is to first compute the matches of the vertex cover of the query graph, then the remaining vertices' matches can be compressed as intersection of the vertex cover's neighbors to avoid costly cartesian product. 




 §.§ Motivations.
 In this paper, we survey seven representative algorithms to solve distributed subgraph matching: <cit.>, <cit.>, <cit.>, <cit.>, <cit.>, <cit.> and <cit.>. While all these algorithms embody some good merits in theory, existing algorithm-level comparisons failed to provide a systematic view to the pros and cons of each algorithm due to several reasons. Firstly, the prior experiments did not take into consideration the differences of languages and the cost of the systems on which each implementation is based (algorithms). Secondly, some implementations hardcode query-specific optimizations for each query, which makes it hard to judge whether the observed performance is from the algorithmic advancement or hardcoded optimization. Thirdly, all and algorithms (more precisely, their implementations) intertwined join strategy with some optimizations of , and . We show in algorithms how each optimization has been applied in current implementation. For example, only adopted and some query-specific , while considered in general, but only for one specific query (was only discussed in paper, but not implemented). People naturally wonder that “maybe it is better to adopt A strategy with B optimization”, but unfortunately none of existing implementation covers that combination. Last but not least, there misses an important benchmarking of the strategy, that is to maintain the whole graph in each partition and parallelize embarrassingly <cit.>. strategy requires no communication, and it should be the most efficient strategy when each machine can hold the whole graph (the case for most experimental settings nowadays).










algorithms summarizes the surveyed algorithms via the category of strategy, the optimality guarantee, and the status of current implementations including the based platform and how the three optimizations are adopted. 



 §.§ Our Contributions

To address the above issues, we aims at a systematic, strategy-level benchmarking of distributed subgraph matching in this paper. To achieve that goal, we implement all strategies, together with the three general-purpose optimizations for subgraph matching based on the dataflow system <cit.>. Note that our implementation covers all seven representative algorithms. Here, we use as the base system as it incurs less cost <cit.> than other popular systems like Giraph <cit.>, Spark <cit.> and GraphLab <cit.>, so that the system's impact can be reduced to the minimum.

We implement the benchmarking platform using our best effort based on the papers of each algorithm and email communications with the authors. Our implementation is  (1) generic to handle arbitrary query, and does not include any hardcoded optimizations; (2) flexible that can configure , and optimizations in any combination for and algorithms; and (3) efficient that are comparable to and sometimes even faster than the original hardcoded implementation. Note that the three general-purpose optimizations are mainly used to reduce communication cost, and is not useful to the and strategies, while we still devote a lot of efforts into their implementations. Aware that their performance heavily depends on the local algorithm, we implement and compare the state-of-the-art local subgraph matching algorithms proposed in <cit.>, <cit.> (for unlabelled matching), and <cit.> (for labelled matching), and adopt the best-possible implementation. For , we refer to <cit.> to implement “Hypercube Optimization” for better hypercube sharing. 






We notice that all existing algorithms are on the track of optimizing communication cost, either by developing algorithm of certain optimality, or by compressing the intermediate results, or even by duplicating the data to avoid communication. Obviously the explosive nature of the (intermediate) results in distributed subgraph matching can incur huge communication cost, but subgraph matching itself is traditionally a computation-intensive task <cit.>. One is immediately intrigued by the following question: 



0.87
Is distributed subgraph matching more a communication-intensive or computation-intensive task nowadays?




It is important to know the answer because it helps direct future researches into the more critical part. In this paper, we will find it out via empirical studies.



We make the following contributions in the paper.

(1) A benchmarking platform based on dataflow system for distributed subgraph matching. We implement four distributed subgraph matching strategies (and the general optimizations) that covers seven state-of-the-art algorithms: <cit.>,  <cit.>, <cit.>, <cit.>, <cit.>, <cit.> and <cit.>. Our implementation is generic to handle arbitrary query, including the labelled and directed query, and thus can guide practical use.

(2) Three general-purpose optimizations - , and . We investigate the literature on the optimization strategies, and spotlight the three general-purpose optimizations. We propose heuristics to incorporate the three optimizations into and strategies, with no need of query-specific adjustments from human experts. The three optimizations can be flexibly configured in any combination.

(3) In-depth experimental studies. In order to extensively evaluate the performance of each strategy and the effectiveness of the optimizations, we use data graphs of different sizes and densities, including sparse road network, dense ego network, and web-scale graph that is larger than each machine's configured memory. We select query graphs of various characteristics that are either from existing works or suitable for benchmarking purpose. In addition to running time, we measure the communication cost, memory usage and other metrics to help reason the performance. 

(4) A practical guide of distributed subgraph matching. Through empirical analysis covering the variances of join strategies, optimizations, join plans, we propose a practical guide for distributed subgraph matching. We also inspire interesting future work based on the experimental findings. 




 §.§ Organizations
. The rest of the paper is organized as follows. backgrounds defines the problem of subgraph matching and introduces preliminary knowledge. algorithms surveys the representative algorithms, and our implementation details following the categories of , , and . opt investigates the three general-purpose optimizations and devises heuristics of applying them to and algorithms. experiments demonstrates the experimental results and our in-depth analysis. related_works discusses the related works, and conclusion concludes the whole paper.



§ PRELIMINARIES





 §.§ Problem Definition



Graph Notations. A graph g is defined as a 3-tuple, g = (V_g, E_g, L_g), where V_g is the vertex set and E_g ⊆ V_g × V_g is the edge set of g, and L_g is a label function that maps each vertex μ∈ V_g and/or each edge e ∈ E_g to a label. Note that for unlabelled graph, L_g simply maps all vertices and edges to ∅. For a vertex μ∈ V_g, denote 𝒩_g(μ) as the set of neighbors, d_g(μ) = |𝒩_g(μ)| as the degree of μ, d_g = 2|E_g|/|V_g| and D_g = max_μ∈ V(g) d_g(μ) as the average and maximum degree, respectively. A subgraph g' of g, denoted g' ⊆ g, is a graph that satisfies V_g'⊆ V_g and E_g'⊆ E_g. 

Given V' ⊆ V_g, we define induced subgraph g(V') as the subgraph induced by V', that is g(V') = (V', E(V'), L_g), where E(V') = {e = (μ, μ') |  e ∈ E_g , μ∈ V' μ' ∈ V'}. We say V' ⊆ V_g is a vertex cover of g, if ∀ e = (μ, μ') ∈ E_g, μ∈ V' or μ' ∈ V'. A minimum vertex cover V^c_g is a vertex cover of g that contains minimum number of vertices. A connected vertex cover is a vertex cover whose induced subgraph is connected, among which a minimum connected vertex cover, denoted V^cc_g, is the one with the minimum number of vertices.

Data and Query Graph. We denote the data graph as G, and let N = |V_G|, M = |E_G|. Denote a data vertex of id i as u_i where 1 <= i <= N. Note that the data vertex has been reordered such that if d_G(u) < d_G(u'), then id(u) < id(u'). We denote the query graph as Q, and let n = |V_Q|, m = |E_Q|, and V_Q = {v_1, v_2, ⋯, v_n}.


In this work, we partition the data graph G across w workers in two ways:



  * Hash partition, denoted Φ^γ(G). Given a random function γ: V_G → [1, w], we place (u, 𝒩_G(u)) in the worker γ(u). We use this strategy by default without otherwise specified.

  * Triangle partition, denoted Φ^τ(G). On top of Φ^γ(G), ∀ u ∈ V_G, we further include the edge (u', u”) in the worker γ(u), where u', u”∈𝒩_G(u), (u', u”) ∈ E_G, and id(u) < id(u')  id(u) < id(u”). Obviously, (u, u', u”) form a triangle in G with u as the smallest vertex.



Subgraph Matching. Given a data graph G and a query graph Q, we define subgraph isomorphism:


	 (Subgraph Isomorphism.) Subgraph isomorphism is defined as a bijective mapping f: V(Q) → V(G) such that, (1) ∀ v ∈ V(Q), L_Q(v) = L_G(f(v)); (2) ∀ (v, v') ∈ E(Q), (f(v), f(v')) ∈ E(G), and L_Q((v, v')) = L_G((f(v), f(v'))). A subgraph isomorphism is called a Match in this paper. With the query vertices listed as {v_1, v_2, ⋯, v_n}, we can simply represent a match f as {u_k_1, u_k_2, ⋯, u_k_n}, where f(v_i) = u_k_i for 1 <= i <= n.
	
	
	The Subgraph Matching problem aims at finding all matches of Q in G. Denote R_G(Q) (or R(Q) when the context is clear) as the result set of Q in G. As prior works <cit.>, we apply symmetry breaking for unlabelled matching to avoid duplicate enumeration caused by automorphism.  Specifically, we first assign partial order O_Q to the query graph according to <cit.>. Here, O_Q ⊆ V_Q × V_Q, and (v_i, v_j) ∈ O_Q means v_i < v_j. In unlabelled matching, a match f must satisfy the order constraint: ∀ (v, v') ∈ O_Q, it holds f(v) < f(v'). Note that we do not consider order constraint in labelled matching. 


	
In subgraph_isomorphism, we present a query graph Q and a data graph G. For unlabelled matching, we give the partial order O_Q under the query graph. There are three matches: {u_1, u_2, u_6, u_5}, {u_2, u_5, u_3, u_6} and {u_4, u_3, u_6, u_5}. It is easy to check that these matches satisfy the order constraint. Without the order constraint, there are actually four automorphic[Automorphism is an isomorphism from one graph to itself.] matches corresponding to each above match <cit.>. For labelled matching, we use different fillings to represent the labels. There are two matches accordingly - {u_1, u_2, u_6, u_5} and {u_4, u_3, u_6, u_5}.  




By treating the query vertices as attributes and data edges as relational table, we can write subgraph matching query as a multiway-way join of the edge relations. For example, regardless of label and order constraints, the query of subgraph_isomorphism can be written as the following join


    R(Q)    = E(v_1, v_2)  E(v_2, v_3) 
        E(v_3, v_4)  E(v_1, v_4)  E(v_2, v_4).

 
This motivates researchers to leverage join operation for large-scale subgraph matching, given that join can be easily distributed, and it is natively supported in many distributed data engines like Spark <cit.> and Flink <cit.>.




 §.§ Massively Parallel Computation Model


Massively Parallel Computation Model () <cit.> defines a multi-round computation among w workers in a cluster. Initially, the input data is distributed arbitrarily across the cluster. Within one round of computation, each worker can receive messages from all workers, use them for local computing, and produce the output data that will be routed to corresponding workers in the next round. model emphasizes communication cost due to its expensiveness in the distributed setting. As previous works <cit.>, we measure the cost by the number of data tuples the workers receive. Under the model, an algorithm is evaluated by three parameters: (1) r, the number of rounds; (2) _max, the maximum cost of any worker during the computation; (3) _tot, the total cost of all workers throughout the computation. 





 §.§ Generic Join: Worst-case Optimality

Ngo et al. gave a class of the worst-case optimal join algorithm called GenericJoin <cit.>. Let the join be R(V) = _F ⊆Ψ R(F), where Ψ = {U | U ⊆ V} and V = ⋃_U∈Ψ U. Given a vertex subset U ⊆ V, let Ψ_U = {V' | V' ∈Ψ V' ∩ U ≠∅}, and for a tuple t ∈ R(V), denote t_U as t's projection on U. We then show the GenericJoin in generic_join.



In generic_join, the original join is recursively decomposed into two parts R(I) and R(J) regarding the disjoint sets I and J.  From line 5, it is clear that R(I) will record R(V)'s projection on I, thus we have |R(I)| ≤ |R(V)|, where R(V) is the maximum possible results of the query. Meanwhile, in line 7, the semi-join R(U) ⋉ t_I = {r |  r ∈ R(U)  r_(U ∪ I) = t_(U ∪ I)} only retains those R(J) w.r.t. t_I that can end up in the join result, which infers that the  R(J) must also be bounded by the final results. This intuitively explains the worst-case optimality of GenericJoin, while we refer interested readers to <cit.> for a complete proof. 

 


 §.§ Timely Dataflow System


is a distributed data-parallel dataflow system <cit.>. The minimum processing unit of is a worker, which can be simply seen as a process that occupies a CPU core. Typically, one physical multi-core machine can run several workers. follows the shared-nothing dataflow computation model <cit.> that abstracts the computation as a dataflow graph. In the dataflow graph, the vertex (a.k.a. operator) defines the computing logics and the edges in between the operators represent the data streams. One operator can accept multiple input streams, feed them to the computing, and produce (typically) one output stream.  
After the dataflow graph for certain computing task is defined, it is distributed to each worker in the cluster, and further translated into a physical execution plan. Based on the physical plan, each worker can accordingly process the task in parallel while accepting the corresponding input portion. 


Traditional centralised distributed architecture relies on a master to coordinate the computation, while follows a decentralised design that drives the computation via the availability of data rather than centralised control. To achieve that purpose, let each data bear a logical timestamp, which can be viewed as an array of integers, and is partially ordered. In , there are two system calls:

	
  * (t): To notify the operator about an event at time t.
	
  * ((O_up, O), d, t): The upstream operator O_up sends the data d to this operator at time t.

Correspondingly, there are two user-defined callback functions:

	
  * (t): The function will be queued on the operator for further call in response to (t).
	
  * ((O_up, O), d, t): The function will be queued on the operator while receiving data d from upstream operator at t.

The queued  and  functions are invoked in such an order that (t) is invoked only after no further calls of ((O_up, O), d, t') with t' ≤ t. This constraint guarantees the correctness of a dataflow, in the sense that once (t) is invoked, all data with smaller timestamps must be available to process. 

To explain the process, we implement a operator on that joins two data streams regarding equal timestamp. Let the two input sources be _0 and _1, and each input data d can be split into key-value pair as (d.k, d.v).



We use two hash tables in the join operator to cache the data for further process. Once a data of both inputs carrying a new (and always larger) timestamp is arriving, we notify the operators in line <ref> and line <ref> to queue the  call at the timestamp. Upon invoking (t), we know that every data with t' ≤ t is ready to go due to the guarantee of dataflow system. Thus we iterate these timestamps in both hash tables, join the corresponding data and output the results in line <ref>.



Discussion. Suppose there are two machines A and B in the cluster, among which A is a quicker executor. The inputs of the join are coming from external data source at a constant pace. While implementing the above join scenario in a centralised architecture, a master machine will coordinate the computation by placing a barrier for synchronisation between two consecutive join steps (timestamps). Thus, in addition to a synchronisation cost, A must always wait for the slower B to move to the next step. In contrast, is more flexible in controlling the computation, as revealed in join_timely. When A completes its share of computation, it can continue to invoke  for the next timestamp, with no need (and indeed impossible) to wait for B. If a barrier is required, one can manually implement a barrier operator[<https://github.com/frankmcsherry/timely-dataflow/blob/master/examples/barrier.rs>] that caches the join's output data until all data of the same timestamp is produced. 




§ ALGORITHM SURVEY


We survey the distributed subgraph matching algorithms following the categories of , , , and . We also show that is a variant of <cit.>, and is thus worst-case optimal.



 §.§ BinJoin



The simplest algorithm uses data edges as the base relation, which starts from one edge, and expands by one edge in each join. For example, to solve the join of subgraph_isomorphism, a simple plan is shown in simple_join_a. The join plan is straightforward, but the intermediate results, especially R_2 (a 3-path), can be huge. 


To improve the performance of , people devoted their efforts into: (1) using more complex base relations other than edge; (2) devising better join plan P. The base relations B_[q] represent the matches of a set of sub-structures [q] of the query graph Q. Each p ∈ [q] is called a join unit, and it must satisfy V_Q = ⋃_p∈[q] V_p and E_Q = ⋃_p∈[q] E_p. With the data graph partitioned across the cluster, <cit.> constrains the join unit to be the structure whose results can be independently computed within each partition (i.e. embarrassingly parallel <cit.>). It is not hard to see that when each vertex has full access to the neighbors in the partition, we can compute the matches of a k-star (a star of k leaves) rooted on the vertex u by enumerating all k-combinations within 𝒩_G(u). Therefore, star is a qualified and indeed widely used join unit.

Given the base relations, the join plan P determines an order of processing binary joins. A join plan is left-deep[More precisely it is deep, and can further be left-deep and right-deep. In this paper, we assume that it is left-deep following the prior work <cit.>.] if there is at least a base relation involved in each join, otherwise it is bushy. For example, the join plan in simple_join_a is left-deep, and a bushy join plan is shown in simple_join_b. Note that the bushy plan avoids the expensive R_2 in the left-deep plan, and is generally better. 

StarJoin. As the name suggests, uses star as the join unit, and it follows the left-deep join order. 
To decompose the query graph, it first locates the vertex cover of the query graph, and each vertex in the cover and its unused neighbors naturally form a star <cit.>. A plan for subgraph_isomorphism is

    (J_1)  R(Q) = (v_2; {v_1, v_3, v_4}) (v_4; {v_2, v_3}),

where (r; L) denotes a relation (the matches of the star) with r as the root, and L as the set of leaves.

TwinTwigJoin. Enumerating a k-star on a vertex of degree d will render O(d^k) cost. We refer star explosion to the case while enumerating stars on a large-degree vertex. Lai et al. proposed <cit.> to address the issue of by forcing the join plan to use (a star of at most two edges) instead of a general star as the join unit. Intuitively, this would help ameliorate the star explosion by constraining the cost of each join unit from d^k of arbitrary k to at most d^2. follows to use left-deep join order. 
The authors proved that is instance optimal to , that is given any general plan in the left-deep join order, we can rewrite it as an alternative plan that draws no more cost (in the big O sense) than the original , where the cost is evaluated based on Erdös-Rényi random graph () model <cit.>.





A plan for subgraph_isomorphism is

    (J_1)      R_1(v_1, v_2, v_3, v_4) = 
       (v_1; {v_2, v_4}) (v_2; {v_3, v_4}); 
    
    (J_2)      R(Q) = R_1(v_1, v_2, v_3, v_4) (v3; {v4}),

where (r; L) denotes a relation with r as the root, and L as the leaves.

CliqueJoin. hampers star explosion to some extent, but still suffers from the problems of long execution (Ω(m/2) rounds) and suboptimal left-deep join plan. resolves the issues by extending in two aspects. Firstly, applies the “triangle partition” strategy (trindex), which enables to use clique, in addition to star, as the join unit. The use of clique can greatly shorten the execution especially when the query is dense, although it still degenerates to when the query contains no clique subgraph. Secondly, exploits the bushy join plan to approach optimality. 
A plan for subgraph_isomorphism is:

    (J_1)  R(Q) = ({v_1, v_2, v_4}) ({v_2, v_3, v_4}),

where (V) denotes a relation of the involving vertices V. 

Implementation Details. We implement the strategy based on the join framework proposed in <cit.> to cover , and .

We use power-law random graph () model <cit.> to estimate the cost as <cit.>, and implement the dynamic programming algorithm <cit.> to compute the cost-optimal join plan. 
Once the join plan is computed, we translate the plan into dataflow that processes each binary join using a Join operator. We implement the Join operator following 's official “pipeline” HashJoin example[<https://github.com/TimelyDataflow/timely-dataflow/blob/master/examples/hashjoin.rs>]. We modify it into “batching-style” - the mappers (senders) shuffle the data based on the join key, while the reducers (receivers) maintain the received key-value pairs in a hash table (until mapper completes) for join processing. The reasons that we implement the join as “batching-style” are, (1) its performance is similar to “pipeline” join as a whole; (2) it replays the original implementation in Hadoop; and (3) it favors the optimization (batching).


Directly implementing the join will cause huge memory burden. We follow the idea of external MergeSort to address the issue. Specifically, we inject a Buffer-and-Batch operator for the two data streams before they arrive at the Join operator. Buffer-and-Batch functions in two parts:

    
	
  * Buffer: While the operator receives data from the upstream, it buffers the data until reaching a given threshold. Then the buffer is sorted according to the join key's hash value and spilled to the disk. The buffer is reused for the next batch of data.
	
  * Batch:  After the data to join is fully received, we read back the data from the disk in a batching manner, where each batch must include all join keys whose hash values are within a certain range.

While one batch of data is delivered to the Join operator, we can leverage Timely to supervise the progress and hold the next batch until the current batch completes. This way, the internal memory requirement is one batch of the data. We notice that and were originally implemented in Hadoop, and the external hash join is already implied in Hadoop's “Shuffle” stage.





Referring clique_join, let

    ({(v_1, v_4), v_2}): {((1, 2), 3), ((2, 3), 4), ((3, 4), 5)}, 
    ({(v_2, v_4), v_1}): {((1, 2), 4), ((2, 3), 5), ((3, 4), 6)}.

Note that we isolate the join keys with extra bracket in the relations for better explanation. Suppose the batch size b = 2, the number of vertices n = 6, and the join key's hash value is simply the first vertex's id. The join will be split into ⌜n/b⌝ = 3 batches: in the first batch, we join the tuples whose join key's hash value is within [1, 2], i.e. {((1, 2), 3), ((2, 3), 4)} and {((1, 2), 4), ((2, 3), 5)}; in the second batch, we join those within [3, 4], i.e. {((3, 4), 5)} and {((3, 4), 6)}; in the third batch, we join those within [5, 6], and there is no tuple in that range. 




In the following, we tend to use algorithms instead of explicitly specifying , and . We further notice that when bushy join plan is adopted, the only boundary between them is that is the algorithm with , while and are the ones without (trindex). 





 §.§ WOptJoin


strategy processes subgraph matching by matching vertices in a predefined order. Given the query graph Q and V_Q = {v_1, v_2, ⋯, v_n} as the matching order, the algorithm starts from an empty set, and computes the matches of the subset {v_1, ⋯, v_i} in the i^th rounds. Denote the partial results after the i^th (i < n) round as R_i, and p = {u_k_1, u_k_2, ⋯, u_k_i}∈ R_i is one of the tuples. In the i + 1^th round, the algorithm expands the results by matching v_i + 1 with u_k_i + 1 for p iff. ∀_1 ≤ j ≤ i (v_j, v_i + 1) ∈ E_Q, (u_k_j, u_k_i + 1) ∈ E_G. It is immediate that the candidate matches of v_i+1, denoted C(v_i+1), can be obtained by intersecting the relevant neighbors of the matched vertices as

    C(v_i + 1) = ⋂_∀_1 ≤ j ≤ i (v_j, v_i + 1) ∈ E_Q𝒩_G(u_k_j).

 
BiGJoin. adopts the strategy in dataflow system. The main challenge is to implement the intersection efficiently using dataflow. For that purpose, the authors designed the following three operators: 

    
	
  * Count: Checking the number of neighbors of each u_k_j in intersection and recording the location (worker) of the one with the smallest neighbor set.
	
  * Propose: Attaching the smallest neighbor set to p as (p; C(v_i + 1)).
	
  * Intersect: Sending (p; C(v_i+1)) to the worker that maintains each u_k_j and update C(v_i+1) = C(v_i+1) ∩𝒩_G(u_k_j).

After intersection, we will expand p by pushing into p every vertex of C(v_i+1). 


Note that the partial results R_i can be huge already, yet in the next round, still needs to Propose on each p ∈ R_i to get the candidate sets for the next query vertex. The memory burden piles up as the progress of the algorithm. To resolve this issue, applies the following batching strategy: it first divides a certain R_i evenly into b parts, and then evaluates each part using a sub-dataflow. These sub-dataflows are physically independent, and can be scheduled to run in a batching manner. Technically speaking, we can apply such batching to any R_i, and even fork sub-batches. For simplicity, the authors implemented that only batches upon R_2, i.e. the edges of the graph. 



Implementation Details. We directly use the authors' implementation <cit.>, but slightly modify the codes to use the common graph data structure. We do not consider the dynamic version of in this paper, as the other strategies currently only support static context. The matching order is determined using a greedy heuristic that starts with the vertex of the largest degree, and consequently selects the next vertex with the most connections (id as tie breaker) with already-selected vertices. 



 §.§ ShrCube


strategy treats the join processing of the query Q as a hypercube of n = |V_Q| dimension. It attempts to divide the hypercube evenly across the workers in the cluster, so that each worker can complete its own share without data communication. However, it is normally required that each data tuple is duplicated into multiple workers. This renders a space requirement of M/w^1-ρ for each worker, where M is size of the input data, w is the number of workers and 0 < ρ≤ 1 is a query-dependent parameter. When ρ is close to 1, the algorithm ends up with maintaining the whole input data in each worker. 

MultiwayJoin. applies the strategy to solve subgraph matching in one single round. Consider w workers in the cluster, a query graph Q with V_Q = {v_1, v_2, …, v_n} vertices and E_Q = {e_1, e_2, …, e_m}, where e_i = (v_i_1, v_i_2). Regarding each query vertex v_i, assign a positive integer as bucket number b_i that satisfies ∏_i=1^n b_i = w. The algorithm then divides the candidate data vertices for v_i evenly into b_i parts via a hash function h: u ↦ z_i, where u∈ V_G, 1≤ z_i≤ b_i. This accordingly divides the whole computation into w shares, each of which can be indiced via an n-ary tuple (z_1, z_2, ⋯, z_n), and is assigned to one worker. Afterwards, regarding each query edge e_i = (v_i_1, v_i_2), maps a data edge (u, u') as (z_1, ⋯, z_i_1 = h(u), ⋯, z_i_2 = h(u'), …, z_n), where other than z_i_1 and z_i_2, each above z_i iterates through {1,2,⋯,b_i}, and the edge will be routed to the workers accordingly. Taking triangle query with E_Q = {(v_1, v_2), (v_1, v_3), (v_2, v_3)} as an example. According to <cit.>, b_1 = b_2 = b_3 = b = √(w) is an optimal bucket number assignment. Each edge (u, u') is then routed to the workers as: (1) (h(u), h(u'), z) regarding (v_1, v_2); (2) (h(u), z, h(u')) regarding (v_1, v_3); (3) (z, h(u), h(u')) regarding (v_2, v_3), where the above z iterates through {1,2,⋯,b}. Consequently, each data edge is duplicated by roughly 3√(w) times, and by expectation each worker will receive 3M/w^1 - 1/3 edges. For unlabelled matching, utilizes the partial order of the query graph (problem_definition) to reduce edge duplication, and details can be found in <cit.>. 

Implementation Details. There are two main impact factors of the performance of . Firstly, the hypercube sharing by assigning proper b_i for v_i. Beame et al. <cit.> generalized the problem of computing optimal hypercube sharing for arbitrary query as linear programming. However, the optimal solution may assign fractional bucket number that is unwanted in practice. An easy refinement is to round down to an integer, but it will apparently result in idle workers. Chu et al. <cit.> addressed this issue via “Hypercube Optimization”, that is to enumerate all possible bucket sequences around the optimal solutions, and choose the one that produces shares (product of bucket numbers) closest to the number of workers. We adopt this strategy in our implementation.

Secondly, the local algorithm. When the edges arrive at the worker, we collect them into a local graph (duplicate edges are removed), and use local algorithm to compute the matches. For unlabelled matching. we study the state-of-the-art local algorithms from “EmptyHeaded” <cit.> and “DualSim”<cit.>. “EmptyHeaded” is inspired by Ngo's worst-case optimal algorithm <cit.> that decomposes the query graph via “Hyper-Tree Decomposition”, computes each decomposed part using worst-case optimal join and finally glues all parts together using hash join. “DualSim” was proposed by <cit.> for subgraph matching in the external-memory setting. The idea is to first compute the matches of V_Q^cc, then the remaining vertices V_Q ∖ V_Q^cc can be efficiently matched by enumerating the intersection of V_Q^cc's neighbors. We find out that “DualSim” actually produces the same query plans as “EmptyHeaded” for all our benchmarking queries (queries) except q_9. We implement both algorithms for q_9 and “DualSim” performs better than “EmptyHeaded” on the GO, US, GP and LJ datasets (datasets). 
As a result, we adopt “DualSim” as the local algorithm for . For labelled matching, we implement “CFLMatch” proposed in <cit.> that has been shown so far to have the best performance.  

Now we let each worker independently compute matches in its local graph. Simply doing so will result in duplicates, so we process deduplication as follows: given a match f that is computed in the worker identified by t_w, we can recover the tuple t^f_e of the matched edge (f(v), f(v')) regarding the query edge e = (v, v'), then the match f is retained if and only if t_w = t^f_e for every e ∈ E_Q. To explain this, let's consider b = 2, and a match {u_0, u_1, u_2} for a triangle query (v_0, v_1, v_2), where h(u_0) = h(u_1) = h(u_2) = 0. It is easy to see that the match will be computed in workers of (0, 0, 0) and (0, 0, 1), while the match in worker (0, 0, 1) will be eliminated as (u_0, u_2) that matches the query edge (v_0, v_2) can not be hashed to (0, 0, 1) regarding (v_0, v_2). We can also avoid deduplication by separately maintaing each edge regarding different query edges it stands for, and use the local algorithm proposed in <cit.>, but it results in too many edge duplicates that drain our memory even when processing a medium-size graph.

 

 §.§ Others

 
 PSgL and its implementation. iteratively processes subgraph matching via breadth-first traversal. All query vertices are configured three status, “white” (initialized), “gray” (candidate) and “black” (matched). Denote v_i as the vertex to match in the i^th round. The algorithm starts from matching initial query vertex v_1, and coloring the neighbors as “gray”. In the i^th round, the algorithm applies the workload-aware expanding strategy at runtime, that is to select the v_i to expand among all current “gray” vertices based on a greedy heuristic to minimize the communication cost <cit.>; the partial results from previous round R_i-1 (specially, R_0 = ∅) will be distributed among the workers based on the candidate data vertices that can match v_i; in the certain worker, the algorithm computes R_i by merging R_i-1 with the matches of the formed by v_i and its “white” neighbors 𝒩^w_Q(v_i), namely (v_i; 𝒩^w_Q(v_i)); after v_i is matched, v_i is colored as “black” and its “white” neighbors will be colored as “gray”; essentially, this process is analogous to by processing R_i = R_i-1(v_i; 𝒩^w_Q(v_i)). Thus, can be seen as an alternative implementation of on Pregel <cit.>. In this work, we also implement using a Pregel on . Note that we introduce Pregel api to as much as possible replay the implementation of . In fact, it is simply wrapping 's primitive operators such as binary_notify and loop [<https://github.com/frankmcsherry/blog/blob/master/posts/2015-09-21.md>], and barely introduces extra cost to the implementation. Our experimental results demonstrate similar findings as prior work <cit.> that 's performance is dominated by <cit.>. Thus, we will not further discuss this algorithm in this paper.



CrystalJoin and its implementation. aims at resolving the “output crisis” by compressing the results of subgraph matching <cit.>. The authors defined a structure called crystal, denoted 𝒬(x, y). A crystal is a subgraph of Q that contains two sets of vertices V_x and V_y (|V_x| = x and |V_y| = y), where the induced subgraph Q(V_x) is a x-clique, and every vertex in V_y connects to all vertices of V_x. We call V_x clique vertices, and V_y the bud vertices. The algorithm first obtains the minimum vertex cover V_Q^c, and then applies the Core-Crystal Decomposition to decompose the query graph into the core Q(V^c_Q) and a set of crystals {𝒬_1(x_1, y_1), …, 𝒬_t(x_t, y_t)}. The crystals must satisfy that ∀ 1 ≤ i ≤ t, Q(V_x_i) ⊆ Q(V_Q^c), namely, the clique part of each crystal is a subgraph of the core. As an example, we plot a query graph and the corresponding core-crystal decomposition  in core_crystals. Note that in the example, both crystals have an edge (i.e. 2-clique) as the clique part. 


With core-crystal decomposition, the computation has accordingly split into three stages:

    
	
  * Core computation. Given that Q(V_Q^c) itself is a query graph, the algorithm can be recursively applied to compute Q(V_Q^c) according to <cit.>.
	
  * Crystal computation. A special case of crystal is 𝒬(x, 1), which is indeed a (x+1)-clique. Suppose an instance of the Q(V_x) is f_x = {u_1, u_2, …, u_x}, we can represent the matches w.r.t. f_x as (f_x, I_y), where I_y = ⋂_i=1^x 𝒩_G(u_i) denotes the set of vertices that can match V_y. This can naturally be extended to the case with y > 1, where any y-combinations of the vertices of I_y together with f_x represent a match. This way, the matches of crystals can be largely compressed. 
	
  * One-time assembly. This stage assembles the core instances and the compressed crystal matches to produce the final results. More precisely, this stage is to join the core instance with the crystal matches.
 


For , we notice some technical obstacles to implement the algorithm according to the paper. Firstly, the core Q(V_Q^c) may be disconnected, a case that can produce exponential number of results. In the original implementation, the authors applied a query-specific optimization, which is against our motivation of solving general queries. Thus, we adapt by replacing the core with the induced subgraph of the minimum connected vertex cover V_Q^cc. 

Secondly, the authors proposed to precompute the cliques up to certain k, while it is often cost-prohibitive to do so in practice. Take UK (datasets) dataset as an example, the triangles, 4-cliques and 5-cliques are respectively about 20, 600 and 40000 times more than the edges of the graph. 
To address this issue, we try to adapt to existing frameworks to avoid precomputing the cliques. We first consider adapting to the join framework of algorithms by treating the crystals as join units, as they are essentially cliques. However, such adaptation is non-trivial as the core part may not be join unit, and needs to be recursively computed. Consequently, we exploit the 's scheme by treating as an order of expanding vertices. Specifically, we first compute the core part using , then for the crystals 𝒬(x, y), we can compute and maintain the V_y matches in the form of arrays without unwrapping them. This will compress the intermediate results to avoid a potential blowout.



	We show the process of the adapted using the query example in core_crystals. It is trivial to compute the core part. Consider an instance of the core, saying {u_1, u_2, u_3}. Let 𝒩_G(u_1) = {u_2, u_3, u_100, …, u_200}, 𝒩_G(u_2) = {u_1, u_100, …, u_200} and 𝒩_G(u_3) = {u_1, u_100, …, u_200}. We then expand the prefix to include the compressed matches of 𝒬_1 as {u_1, u_2, u_3; [u_100, …, u_200]}, in the next step, we can treat this as the “compressed” prefix, and directly expand to include the compressed matches of 𝒬_2 as {u_1, u_2, u_3; [u_100, …, u_200], [u_100, …, u_200]}. When we complete the expanding, we then decompress each record for the actual results.






We notice two technical obstacles to implement according to the paper. Firstly, it is worth noting that the core Q(V_Q^c) may be disconnected, a case that can produce exponential number of results. The authors applied a query-specific optimization in the original implementation to resolve this issue. Secondly, the authors proposed to precompute the cliques up to certain k, while it is often cost-prohibitive to do so in practice. Take UK (datasets) dataset as an example, the triangles, 4-cliques and 5-cliques are respectively about 20, 600 and 40000 times larger than the graph itself. It is worth noting that the main purpose of this paper is not to study how well each algorithm performs for a specific query, which has its theoretical value, but can barely guide practice. After communicating with the authors, we adapt in the following. Firstly, we replace the core Q(V_Q^c) with the induced subgraph of the minimum connected vertex cover Q(V_Q^cc). Secondly, instead of implementing as a strategy, we use it as an alternative join plan (matching order) for . According to , we first match V_Q^cc, while the matching order inside and outside V_Q^cc still follows 's greedy heuristic (vertex-at-a-time). It is worth noting that this adaptation achieves high performance comparable to the original implementation. In fact, we also apply plan to , while it does not perform as well as the version, thus we do not discuss this implementation.

FullRep and its implementation. simply maintains a full replica of the graph in each physical machine. Each worker picks one independent share of computation and solves it using existing local algorithm.

The implementation is straightforward. We let each worker pick its share of computation via a Round-Robin strategy, that is we settle an initial query vertex v_1, and let first worker match v_1 with u_1 to continue the remaining process, and second worker match v_1 with u_2, and so on. This simple strategy already works very well on balancing the load of our benchmarking queries (queries). We use “DualSim” for unlabelled matching and “CFLMatch” for labelled matching as .



 §.§ Worst-case Optimality.





Given a query Q and the data graph G, we denote the maximum possible result set as R_G(Q). Simply speaking, an algorithm is worst-case optimal if the aggregation of the total intermediate results is bounded by Θ(|R_G(Q)|). Todd <cit.> and Ngo et al. <cit.> almost meanwhile proposed the worst-case join algorithm in relational database. Ammar et al. proposed for subgraph matching based on Ngo's algorithm <cit.>. Following , was developed and shown to be worst-case optimal <cit.>.

As for , the optimality has not been claimed in the paper <cit.>. In this work, we further contribute to the finding of 's worst-case optimality. We refer interested readers to the full paper <cit.> for a complete proof.

Given a query Q and the data graph G, we denote the maximum possible result set as R_G(Q). Simply speaking, an algorithm is worst-case optimal if the aggregation of the total intermediate results is bounded by Θ(|R_G(Q)|). Ngo et al. proposed a class of worst-case optimal join algorithm called <cit.>, and we first overview this algorithm. 

GenericJoin. Let the join be R(V) = _F ⊆Ψ R(F), where Ψ = {U | U ⊆ V} and V = ⋃_U∈Ψ U. Given a vertex subset U ⊆ V, let Ψ_U = {V' | V' ∈Ψ V' ∩ U ≠∅}, and for a tuple t ∈ R(V), denote t_U as t's projection on U. We then show the GenericJoin in generic_join. 



In generic_join, the original join is recursively decomposed into two parts R(I) and R(J) regarding the disjoint sets I and J.  From line 5, it is clear that R(I) will record R(V)'s projection on I, thus we have |R(I)| ≤ |R(V)|, where R(V) is the maximum possible results of the query. Meanwhile, in line 7, the semi-join R(U) ⋉ t_I = {r |  r ∈ R(U)  r_(U ∪ I) = t_(U ∪ I)} only retains those R(J) w.r.t. t_I that can end up in the join result, which infers that the  R(J) must also be bounded by the final results. This intuitively explains the worst-case optimality of GenericJoin, while we refer interested readers to <cit.> for a complete proof. 

It is easy to see that is worst-case optimal. In generic_join, we select I in line 4 by popping the edge relation E(v_s, v_i) (s < i) in the i^th step. In line 7, the recursive call to solve the semi-join R(U) ⋉ t_I actually corresponds to the intersection process. 

Worst-case Optimality of CliqueJoin. Note that the two clique relations in clique_join interleave one common edge (v_2, v_4) in the query graph. This optimization, called “overlapping decomposition” <cit.>, eventually contributes to 's worst-cast optimality. Note that it is not possible to apply this optimization to and . We have the following theorem.



is worst-case optimal while applying “overlapped decomposition”.



 
 We implement using generic_join in the following. Note that Q(V) denotes a subgraph of Q induced by V. In line 2, we change the stopping condition to “Q(I) is either a clique or a star”. In line 4, the I is selected such that Q(I) is either a clique or a star. Note that by applying the “overlapping decomposition” in , the sub-query of the J part must be the J-induced graph Q(J), and it will also include the edges of E_Q(I)∩ E_Q(J), which infers that R(Q(J)) = R(Q(J)) ⋉ R(Q(I)) , and just reflects the semi-join in line 7. Therefore, belongs to GenericJoin, and is thus worst-case optimal.





§ OPTIMIZATIONS


We introduce the three general-purpose optimizations, , and in this section, and how we orthogonally apply them to and algorithms. In the rest of the paper, we will use the strategy , , instead of their corresponding algorithms, as we focus on strategy-level comparison. 



 §.§ Batching


Let R(V_i) be the partial results that match the given vertices V_i = {v_s_i, v_s_2, …, v_s_i} (R_i for short if V_i follows a given order), and R(V_j) denote the more complete results with V_i ⊂ V_j. Denote R_j|R_i as the tuples in R_j whose projection on V_i equates R_i. Let's partition R_i into b disjoint parts {R^1_i, R^2_i, …, R^b_i}. We define on R_j|R_i as the technique to independently process the following sub-tasks that compute {R_j|R_i^1, R_j|R_i^2, …, R_j|R_i^b}. Obviously, R_j|R_i = ⋃_k = 1^b R_j | R_i^k. 

WOptJoin. Recall from vertex-at-a-time that progresses according to a predefined matching order {v_1, v_2, …, v_n}. In the i^th round, will Propose on each p ∈ R_i-1 to compute R_i. It is not hard to see that we can easily apply to the computation of R_i|R_i-1 by randomly partitioning R_i-1. For simplicity, the authors implemented on R(Q) | R_1(v_1). Note that R_1(v_1) = V_G in unlabelled matching, which means that we can achieve simply by partitioning the data vertices[Practically, it is more efficient to start from matching the edges instead of the vertices, and we can batch on R(Q)|R_2, where R_2 = E_G.]. For short, we also say the strategy batches on v_1, and call v_1 the batching vertex. We follow the same idea to apply to algorithms.



BinJoin. While it is natural for to batch on v_1, it is non-trivial to pick such a vertex for . Given a decomposition of the query graph {p_1, p_2, …, p_s}, where each p_i is a join unit, we have R(Q) = R(p_1)  R(p_2) ⋯ R(p_s). If we partition R_1(v) so as to batch on v ∈ V_Q, we correspondingly split the join task, and one of the sub-task is R(Q) | R_1^k(v) = R(p_1) | R_1^k(v) ⋯ R(p_s) | R_1^k(v) (R_1^k(v) is one partition of R_1(v)). Observe that if there exists a join unit p where v ∉V_p, we must have R(p) = R(p) | R_1^k(v), which means R(p) have to be fully computed in each sub-task. Let's consider the example query in ttjoin.

    R(Q) = T_1(v_1, v_2, v_4)  T_2(v_2, v_3, v_4)  T_3(v_3, v_4).

Suppose we batch on v_1, the above join can be divided into the following independent sub-tasks:

    R(Q)|R_1^1(v_1)    = (T_1(v_1, v_2, v_4)|R_1^1(v_1))  T_2(v_2, v_3, v_4)  T_3(v_3, v_4), 
    
        R(Q)|R_1^2(v_1)    = (T_1(v_1, v_2, v_4)|R_1^2(v_1))  T_2(v_2, v_3, v_4)  T_3(v_3, v_4), 
       ⋯
    
        R(Q)|R_1^b(v_1)    = (T_1(v_1, v_2, v_4)|R_1^b(v_1))  T_2(v_2, v_3, v_4)  T_3(v_3, v_4).

It is not hard to see that we will have to re-compute T_2(v_2, v_3, v_4) and T_3(v_3, v_4) in all the above sub-tasks. Alternatively, if we batch on v_4, we can avoid such re-computation as T_1, T_2 and T_3 can all be partitioned in each sub-task. Inspired by this, for , we come up with the heuristic to apply on the vertex that presents in as many join units as possible. Note that such vertex can only be in the join key, as otherwise it must at least not present in one side of the join. For complex query, we can still have join unit that does not contain any vertex for after applying the above heuristic. In this case, we either re-compute the join unit, or cache it on disk. Another problem caused by this is potential memory burden of the join. Thus, we devise the join-level following the idea of external MergeSort. Specifically, we inject a Buffer-and-Batch operator for the two data streams before they arrive at the Join operator. Buffer-and-Batch functions in two parts:

    
	
  * Buffer: While the operator receives data from the upstream, it buffers the data until reaching a given threshold. Then the buffer is sorted according to the join key's hash value and spilled to the disk. The buffer is reused for the next batch of data.
	
  * Batch:  After the data to join is fully received, we read back the data from the disk in a batching manner, where each batch must include all join keys whose hash values are within a certain range.

While one batch of data is delivered to the Join operator, Timely allows us to supervise the progress and hold the next batch until the current batch completes. This way, the internal memory requirement is one batch of the data. Note that such join-level is natively implemented in Hadoop's “Shuffle” stage, and we replay this process in to improve the scalability of the algorithm.



 §.§ Triangle Indexing


As the name suggests, precomputes the triangles of the data graph and indices them along with the graph data to prune infeasible results. The authors of <cit.> optimized the 4-clique query by using the triangles as base relations to join, which reduces the rounds of join and network communication. In <cit.>, the authors proposed to not only maintain triangles, but all k-cliques up to a given k. As we mentioned earlier, it incurs huge extra cost of maintaining triangles already, let alone larger cliques. 

In addition to the default hash partition, Lai et al. proposed “triangle partition” <cit.> by also incorporating the edges among the neighbors (it forms triangles with the anchor vertex) in the partition. “Triangle partition” allows to use clique as the join unit <cit.>, which greatly reduces the intermediate results of certain queries and improves the performance. “Triangle partition” is in de facto a variant of , which instead of explicitly materializing the triangles, maintains them in the local graph structure (e.g. adjacency list). As we will show in the experiment (experiments), this will save a lot of space compared to explicit triangle materialization. Therefore, we adopt the “triangle partition” for optimization in this work.

BinJoin. Obviously, becomes with , and (or ) otherwise. With worst-case optimality guarantee (worst_case_opt), should perform much better with , which is also observed in “Exp-1” of experiments.



WOptJoin. In order to match v_i in the i^th round, utilizes Count, Propose and Intersect to process the intersection of intersection. For ease of presentation, suppose v_i+1 connects to the first s query vertices {v_1, v_2, …, v_s}, and given a partial match, {f(v_1), …, f(v_s)}, we have C(v_i + 1)=⋂_j = 1^s 𝒩_G(f(v_j)). In the original implementation, it is required to send (p; C(v_i + 1)) via network to all machines that contain each f(v_j) (1 ≤ j ≤ s) to process the intersection, which can render massive communication cost. In order to reduce the communication cost, we implement for in the following. We first group {v_1, …, v_s} such that for each group U(v_x), we have 

    U(v_x) = {v_x}∪{v_y  |  (v_x, v_y) ∈ E_Q}.

Because of , we have 𝒩_G(f(v_y)) (∀ v_y ∈ U(v_x)) maintain in f(v_x)'s partition. Thus, we only need to send the prefix to f(v_x)'s machine, and the intersection within U(v_x) can be done locally. We process the grouping using a greedy strategy that always constructs the largest group from the remaining vertices.



The “triangle partition” may result in maintaining a large portion of the data graph in certain partition. Lai et al. pointed out this issue, and proposed a space-efficient alternative by leveraging the vertex orderings <cit.>. That is, given the partitioned vertex as u, and two neighbors u' and u” that close a triangle, we place the edge (u', u”) in the partition only when u < u' < u”. Although this alteration reduces storage, it may affect the effectiveness of for and the implementations of and for algorithms. Take as an example, after using the space-efficient “triangle partition”, we should modify the above grouping as:

    U(v_x) = {v_x}∪{v_y  |  (v_x, v_y) ∈ E_Q  (v_x, v_y) ∈ O_Q }.

Note that the order between query vertices are for symmetry breaking (problem_definition), and it may not present in certain query, which makes completely useless for .




 §.§ Compression


Subgraph matching is a typical combinatorial problem, and can easily produce results of exponential size. aims to maintain the (intermediate) results in a compressed form to reduce resource allocation and communication cost. In the following, when we say “compress a query vertex”, we mean maintaining its matched data vertices in the form of an array, instead of unfolding them in line with the one-one mapping of a match (isomorphism). 
Qiao et al. proposed to study in general for subgraph matching. As we introduced in others, first extracts the minimum vertex cover as uncompressed part, and then it can compress the remaining query vertices as the intersection of certain uncompressed matches' neighbors. Such leverages the fact that all dependencies (edges) of the compressed part that requires further computation are already covered by the uncompressed part, thus it can stay compressed until the actual matches are requested. inspires a heuristic for doing , that is to compress the vertices whose matches will not be used in any future computation. In the following, we will apply the same heuristic to the other algorithms. 

BinJoin. Obviously we can not compress any vertex that presents in the join key. What we need to do is to simply locate the vertices to compress in the join unit, namely star and clique. For star, the root vertex must remain uncompressed, as the leaves' computation depends on it. For clique, we can only compress one vertex, as otherwise the mutual connection between the compressed vertices will be lost. In a word, we compress two types of vertices for , (1) non-key and non-root vertices of a star join unit, (2) one non-key vertex of a clique join unit.

WOptJoin. Based on a predefined join order {v_1, v_2, …, v_n}, we can compress v_i (1 ≤ i ≤ n), if there does not exist v_j (i < j) such that (v_i, v_j) ∈ E_Q. In other words, v_i's matches will never be involved in any future intersection (computation). Note that v_n's can be trivially compressed. With , when v_i is compressed, we will maintain its matches as an array instead of unfolding it into the prefix like a normal vertex.





§ EXPERIMENTS




 §.§ Experimental settings


Environments. We deploy two clusters for the experiments: (1) a local cluster of 10 machines connected via one 10GBps switch and one 1GBps switch. Each machine has 64GB memory, 1 TB disk and 1 Intel Xeon CPU E3-1220 V6 3.00GHz with 4 physical cores; (2) an AWS cluster of 40 “r5-2xlarge” instances connected via a 10GBps switch, each with 64GB memory, 8 vCpus and 500GB Amazon EBS storage. By default we use the local cluster of 10 machines with 10GBps switch. We run 3 workers in each machine in the local cluster, and 6 workers in the AWS cluster for . The codes are implemented based on the open-sourced dataflow system <cit.> using Rust 1.32. We are still working towards open-sourcing the codes, and the bins together with their usages are temporarily provided[<https://goo.gl/Xp5BrW>] to verify the results. 

Metrics.  In the experiments, we measure query time T as the slowest worker's wall clock time from an average of three runs. We allow 3 hours as the maximum running time for each test. We use and to indicate a test case runs out of the time limit and out of memory, respectively. By default we will not show the results for clear presentation. We divide T into two parts, the computation time T_comp and the communication time T_comm. We measure T_comp as the time the slowest worker spends on actual computation by timing every computing function. We are aware that the actual communication time is hard to measure as overlaps computation and communication to improve throughput. We consider T - T_comp, which mainly records the time the worker waits data from the network channel (a.k.a. communication time). While the other part of communication that overlaps computation is of less interest as it does not affect the query progress. As a result, we simply let T_comm = T - T_comp in the experiments. We measure the maximum peak memory using Linux's “time -v” in each machine. We define the communication cost as the number of integers a worker receives during the process, and measure the maximum communication cost among the workers accordingly. 

Dataset Formats. We preprocess each dataset as follows: we treat it as a simple undirected graph by removing self-loop and duplicate edges, and format it using “Compressed Sparse Row” (CSR) <cit.>. We relabel the vertex id according to the degree and break the ties arbitrarily. 

Compared Strategies. In the experiments, we implement and with all , and optimizations (opt). is implemented with “Hypercube Optimization” <cit.>, and “DualSim” (unlabelled) <cit.> and “CFLMatch” (labelled) <cit.> as local algorithms. is implemented with the same local algorithms as .

Auxiliary Experiments. We have also conducted several auxiliary experiments in the appendix to study the strategies of , , and .



 §.§ Unlabelled Experiments

Datasets. The datasets used in this experiment are shown in datasets. All datasets except SY are downloaded from public source, which are indicated by the letter in the bracket (S <cit.>, W <cit.>, D <cit.>). All statistics are measured as G is an undirected graph. Among the datasets, GO is a small dataset to study cases of extremely large (intermediate) result set; LJ, UK and FS are three popular datasets used in prior works, featuring statistics of real social network and web graph; GP is the google plus ego network, which is exceptionally dense; US and EU, on the other end, are sparse road networks. These datasets vary in number of vertices and edges, densities and maximum degree, as shown in datasets. We synthesize the SY data according to <cit.> that generates data with real-graph characteristics. Note that the data occupies roughly 80GB space, and is larger than the configured memory of our machine. We synthesize the data because we do not find public accessible data of this size. Larger dataset like Clueweb <cit.> is available, but it is beyond the processing power of our current cluster. 

Each data is hash partitioned (“hash”) across the cluster. We also implement the “triangle partition” (“tri.”) for optimization (trindex). To do so, we use to compute the triangles and send the triangle edges to corresponding partition. We record the time T_* and average number of edges |E_*| of the two partition strategies. The partition statistics are recorded using the local cluster, except for SY that is processed in the AWS cluster. From datasets, 

we can see that |E_tri.| is noticeably larger, around 1-10 times larger than |E_hash|. Note that in GP and UK, which either is dense, or must contain a large dense community, the “triangle partition” can maintain a large portion of data in each partition. While compared to complete triangle materialization, “triangle partition” turns out to be much cheaper. For example, the UK dataset contains around 27B triangles, which means each partition in our local cluster should by average take 0.9B triangles (three integers); in comparison, UK's “triangle partition” only maintains an average of 0.16B edges (two integers) according to datasets. 

We use US, GO and LJ as default datasets in the experiments “Exp-1”, “Exp-2” and “Exp-3” in order to collect useful feedbacks from successful queries, while we may not present certain cases when they do not give new findings. 







Queries. The queries are presented in queries. We also give the partial order under each query for symmetry breaking. The queries except q_7 and q_8 are selected based on all prior works <cit.>, while varying in number of vertices, densities, and the vertex cover ratio |V^cc_Q|/|V_Q|, in order to better evaluate the strategies from different perspectives. The three queries q_7, q_8 and q_9 are relatively challenging given their result scale. For example, the smallest dataset GO contains 2,168B(illion) q_7, 330B q_8 and 1,883B q_9, respectively. For short of space, we record the number of results of each successful query on each dataset in the appendix. Note that q_7 and q_8 are absent from existing works, while we benchmark q_7 considering the importance of path query in practice, and q_8 considering the varieties of the join plans. 



Exp-1: Optimizations. We study the effectiveness of , and for both and strategies, by comparing and with their respective variants with one optimization off, namely “without Batching”, “without Trindexing” and “without Compression”. In the following, we use the suffix of “(w.o.b.)”, “(w.o.t.)” and “(w.o.c.)” to represent the three variants. We use the queries q_2 and q_5, and the results of US and LJ are shown in vary_opt. By default, we use the batch size of 1,000,000 for both and (according to <cit.>) in this experiment, and we reduce the batch size when it runs out of memory, as will be specified.



While comparing with , we observe that barely affects the performance of q_2, but severely for q_5 on LJ (1800s vs 4000s (w.o.b.) ). The reason is that we still apply join-level for that dumps the intermediate data to the disk (batching). While q_5's intermediate data includes the massive results of sub-query Q({v_2, v_3, v_4, v_5}), which incurs huge amount of disk I/O (US does not have this problem as it produces very few results). We also run q_5 without the join-level on LJ, but it fails with . For , is a critical optimization, with the observed performance of better than that of , especially so on LJ. This is expected as actually degenerates to . , on the one hand, allows to run much faster than for both queries on LJ, on the other hand, makes it slower on US. The reason is that US is a sparse dataset with few room for , while itself incurs extra cost. We also compare with on the other sparse graph EU, and the results are the same.  

For strategy, has little impact to the performance. Surprisingly, after using to , the improvement by average is only around 18%. We do another experiment in the same cluster but using 1GBps switch, which shows is over 6 times faster than for both queries on LJ. Note that uses separate threads to buffer received data from the network. Given the same computing speed, a faster network allows the data to be more fully buffered and hence less wait for the following computation. Similar to , greatly improves the performance while querying on LJ, but the opposite on US.


Exp-2 Challenging Queries. We study the challenging queries q_7, q_8 and q_9 in this experiment. We run this experiment using , , and , and show the results of US and GO (LJ failed all cases) in challenging_query. Recall that we split the time into computation time and communication time (experimental_settings), here we plot the communication time as gray filling in each bar of challenging_query.



beats all the other strategies, while fails q_8 and q_9 on GO because of . Although uses the same local algorithm as , it spends a lot of time on deduplication (multiway_join).

We focus on comparing and on GO dataset. On the one hand, outperforms for q_7 and q_8. Their join plans of q_7 are nearly the same except that relies on a global shuffling on v_3 to processing join, while sends the partial results to the machine that maintains the vertex to grow. It is hence reasonable to observe 's poorer performance for q_7 as shuffling is typically a more costly operation. The case of q_8 is similar, so we do not further discuss. On the other hand, even living with costly shuffling, still performs better for q_9. Due to the vertex-growing nature, 's “optimal plan” will have to process the costly sub-query Q({v_1, v_2, v_3, v_4, v_5}). On US dataset, consistently outperforms for these queries. This is because that US does not produce massive intermediate results as LJ, thus 's shuffling cost consistently dominates.

While processing complex queries like q_8 and q_9, we can study varieties of join plans for and . First of all, we want the readers to note that 's join plan for q_8 is different from the optimal plan originally given <cit.>. The original “optimal” plan computes q_8 by joining two tailed triangles (triangle tailed with an edge), while this alternative plan works better by joining the uppers “house-shape” sub-query with the bottom triangle. In theory, the tailed triangle has worse-case bound (AGM bound <cit.>) of O(M^2), smaller than the house's O(M^2.5), and 's actually favors this plan based on cost estimation. However, we find out that the number of tailed triangles is very close to that of the houses on GO, which renders costly process for the original plan to join two tailed triangles. This indicates insufficiency of both cost estimation proposed in <cit.> and worst-case optimal bound <cit.> while computing the join plan, which will be further discussed in discussions.

 Secondly, it is worth noting that we actually report the result of for q_9 while using the plan, as it works better than 's original “optimal” plan. For q_9, will first compute Q(V_Q^cc), namely the 2-path {v_1, v_3, v_5}, thereafter it can compress all remaining vertices v_2, v_4 and v_6. In comparison, the “optimal” plan can only compress v_2 and v_6. In this case, performs better because it configures larger compression. In <cit.>, the authors proved that it renders maximum compression to use the vertex cover as the uncompressed core. However, this may not necessarily result in the best performance, considering that it can be costly to compute the core part. In our experiments, the unlabelled q_4, q_8 and labelled q_8 are cases that plan performs worse than the original plan (with optimization), where plan does not render strictly larger compression while having to process the costly core part. As a result, we only recommend plan when it leads to strictly larger compression.

The final observation is that the computation time dominates most of the evaluated cases, except 's q_8, and 's q_9 on US. We will further discuss this in Exp-3.

Exp-3 All-Around Comparisons. In this experiment, we run q_1 - q_6 using , , and across the datasets GP, LJ, UK, EU and FS. We also run with plan in q_4 as it is the only query that renders different plan from plan, and the results show that the performance with plan is consistently better. We report the results in all_cases, where the communication time is plotted as gray filling. As a whole, among all 35 test cases, achieves the best 85% completion rate, followed by and which complete 71.4% and 68.6% respectively, and performs the worst with just 8.6% completion rate.



typically outperforms the other strategies. Observe that 's performance is often very close to . The reason is that the 's computing plans for these evaluated queries are similar to “DualSim” adopted by . The extra communication cost of has been reduced to very low while adopting optimization. While comparing with , is better for q_3, a clique query (join unit) that requires no join (a case of embarrassingly parallel). performs worse than in most other queries, which, as we mentioned before, is due to the costly shuffling. There is an exception - querying q_1 on GP - where performs better than both and . 

We explain this using our best speculation. GP is a very dense graph, where we observe nearly 100 vertices with degree around 10,000. To process q_1, after computing the sub-query Q({v_1, v_2, v_4}), (and “DualSim”) processes the intersection of v_1 and v_4 (their matches) for v_3. Those larger-degree vertices are now frequently pairing, leading to expensive intersection. In comparison, computes q_1 by joining the sub-query Q({v_1, v_2, v_3}) with Q({v_1, v_3, v_4}). Because both strategies compute Q({v_1, v_2, v_3}), we consider how computes Q({v_1, v_3, v_4}). first locate the matched vertex of v_3, then matches v_1 and v_4 among its neighbors, which is generally cheaper than intersecting the neighbors of v_1 and v_4 to compute v_3. 
Due to the existence of these high-degree pairs, the cost 's intersection can exceed 's shuffling.



We observe that the computation time T_comp dominates in most cases as we mentioned in Exp-2. This is trivially true for and , but it may not be clearly so for and given that they all need to transfer a massive amount of intermediate data. We investigate this and find out two potential reasons. The first one attributes to 's highly optimized communication component, which allows the computation to overlap communication by using extra threads to receive and buffer the data from the network so that it can be mostly ready for the following computation. The second one is the fast network. We re-run these queries using the 1GBps switch, while the results show the opposite trend that the communication time T_comm in turn takes over.

Exp-4 Web-Scale. We run the SY datasets in the AWS cluster of 40 instances. Note that can not be used as SY is larger than the machine's memory. We use the queries q_2 and q_3, and present the results of and (fails all cases due to ) in web_scale_exp. The results are consistent with the prior experiments, but observe that the gap between and while querying q_1 is larger. This is because that we now deploy 40 AWS instances, and 's shuffling cost increases.





 §.§ Labelled Experiments


We use the LDBC social network benchmarking (SNB) <cit.> for labelled matching experiment due to the lack of labelled big graphs in the public. SNB provides a data generator that generates a synthetic social network of required statistics, and a document <cit.> that describes the benchmarking tasks, in which the complex tasks are actually subgraph matching. The join plans of and for labelled experiments are generated as unlabelled case, but we use the label frequencies to break tie.

Datasets. We list the datasets and their statistics in labelled_datasets. These datasets are generated using the "Facebook" mode with a duration of 3 years. The dataset's name, denoted as DGx, represents a scale factor of x. The labels are preprocessed into integers. 





Queries. The queries, shown in labelled_queries, are selected from the SNB's complex tasks with some adaptions: (1) removing the direction of the edges; (2) removing the edge labels; (3) using one-hop edge for multi-hop edges; (4) removing the “no edge” condition; (5) removing all properties except the node types as labels. For (1) and (2), note that our current implementation can support both cases, and we do the adaptations for consistency and simplicity. For (3) and (4), we adapt them because currently they do not conform with the subgraph matching problem studied in this paper. For (5), it is due to our current limitation in supporting property graph. We leave (3), (4) and (5) as interesting future work.

Exp-5 All-Around Comparisons.  We now conduct the experiment using all queries on DG10 and DG60, and present the results in all_labelled_cases. Here we compute the join plans for and by using the unlabelled method, but further using the label frequencies to break tie. The gray filling again represents communication time. outperforms the other strategies in many cases, except that it performs slightly slower than for q_3 and q_5. This is because that q_3 and q_5 are join units, and processes them locally in each machine as , and it does not build indices as “CFLMatch” used in . When comparing to ,   Among all these queries, we only have q_8 that configures different plan (w.r.t. plan) for . The results show that the performance of drops about 10 times while using plan. Note that the core part of q_8 is a 5-path of “Psn-City-Cty-City-Psn” with enormous intermediate results. As we mentioned in unlabelled experiments, it may not always be wise to first compute the vertex-cover-induced core.

We now focus on comparing and . There are three cases that intrigue us. Firstly, observe that performs much better than while querying q_4. The reason is high intersection cost as we discovered on GP dataset in unlabelled matching. Secondly, performs worse than in q_7, which again is because of 's costly shuffling. The third case is q_9, the most complex query in the experiment. performs much better while querying q_9. The bad performance of comes from the long execution plan together with costly intermediate results. The two algorithms all expand the three “Psn”s, and then grow via one of the “City”s to “Cty”, but approaches this using one join (a triangle  a ), while will first expand to “City” then further “Cty”, and the “City” expansion is the culprit of the slower run.





§ DISCUSSIONS AND FUTURE WORK.


We discuss our findings and potential future work based on the experiments in experiments. Eventually, we summarize the findings into a practical guide.

Strategy Selection. is obviously the preferred choice when the machine can hold the graph data, while both and are good alternatives when the graph is larger than the capacity of the machine. For and , on one side, may perform worse than (e.g. unlabelled q_2, q_4, q_5) due to the expensive shuffling operation, on the other side, can also outperform (e.g. unlabelled and labelled q_9) while avoiding costly sub-queries due to query decomposition. One way to choose between and is to compare the cost of their respective join plans, and select the one with less cost. For now, we can either use cost estimation proposed in <cit.>, or summing the worst-case bound, but none of them consistently gives the best solution, as will be discussed in “Optimal Join Plan”. Alternatively, we refer to “EmptyHeaded” <cit.> to study a potential hybrid strategy of and . Note that “EmptyHeaded” is developed in single-machine setting, and it does not take into consideration the impact of , we hence leave such hybrid strategy in the distributed context as an interesting future work. 

Optimizations. Our experimental results suggest always using , using when each machine has sufficient memory to hold “triangle partition”, and using when the data graph is not very sparse (e.g. d_G ≥ 5). often does not impact performance, so we recommend always using due to the unpredictability of the size of (intermediate) results. is critical for , and it can greatly improve by reducing communication cost, while it requires extra storage to maintain “triangle partition”. Amongst the evaluated datasets, each “triangle partition” maintains an average of 30% data in our 10-machine cluster. Thus, we suggest a memory threshold of 60% |E_G| (half for graph and half for running algorithm) for in a cluster of the same or larger scale. Note that the threshold does not apply to extremely dense graph. Among the three optimizations, is the primary performance booster that improves the performance of and by 5 times on average in all but the cases on the very sparse road networks. For such very sparse data graphs, can render more cost than benefits.   



Optimal Join Plan. It is challenging to systematically determine the optimal join plans for both and . From the experiments, we identify three impact factors: (1) the worst-case bound; (2) cost estimation based on data statistics; (3) favoring the optimizations, especially . All existing works only partially consider these factors, and we have observed sub-optimal join plans in the experiments. For example, bases the “optimal” join plan on minimizing the cost estimation, but the join plan does not render the best performance for unlabelled q_8; follows the worst-case optimality, while it may encounter costly sub-queries for labelled and unlabelled q_9; focuses on maximizing the compression, while ignoring the facts that the vertex-cover-induced core part itself can be costly to compute. Additionally, there are other impact factors such as the partial orders of query vertices and the label frequencies, which have not been studied in this work due to short of space. It is another very interesting future work to thoroughly study the optimal join plan while considering all above impact factors.

Computation vs. Communication. We argue that distributed subgraph matching nowadays is a computation-intensive task. This claim holds when the cluster configures high-speed network (e.g. ≥ 10GBps), and the data processor can efficiently overlap computation with communication. Note that computation cost (either 's join or 's intersection) is lower-bounded by the output size that is equal to the communication cost. Therefore, computation becomes the bottleneck if the network condition is good to guarantee the data to be delivered in time. Nowadays, the bandwidth of local cluster commonly exceeds 10GBps, and the overlapping of computation and communication is widely used in distributed systems (e.g. Spark <cit.>, Flink <cit.>). 
As a result, we tend to see distributed subgraph matching as a computation-intensive task, and we advocate future research to devote more efforts into optimizing the computation while considering the following perspectives: (1) the new advancements of hardware, for example the co-processing on GPU in the coupled CPU-GPU architectures <cit.> and the SIMD programming model on modern CPU <cit.>; (2) general computing optimizations such as load balancing strategy and cache-aware graph data accessing <cit.>. 

A Practical Guide. Based on the experimental findings, we propose a practical guide for distributed subgraph matching in program_guide. Note that this program guide is based on current progress of the literature, and future work is needed, for examples to study the hybrid strategy and the impact factors of the optimal join plan, before we can arrive at a solid decision-making to choose between and .


  



§ RELATED WORK


Isomorphism-based Subgraph Matching. In the labelled case, Shang et al. <cit.> used the spanning tree of the query graph to filter infeasible results. Han et al. <cit.> observed the importance of matching order. In <cit.>, the authors proposed to utilize the symmetry properties in the data graph to compress the results. Bi et al. <cit.> proposed based on the “core-forest-leaves” matching order, and obtained performance gain by postponing the notorious cartesian product. 

The unlabelled case is also known as subgraph listing/enumeration, and due to the gigantic (intermediate) results, people have been either seeking scalable algorithms in parallel, or devising techniques to compress the results. Other than the algorithms studied in this paper (algorithms), Kim et al. proposed the external-memory-based parallel algorithm DualSim <cit.>, which maintains the data graph  in blocks on the disk, and matches the query graph by swapping in/out blocks of data to improve I/O efficiency. 

Incremental Subgraph Matching. Computing subgraph matching in a continuous context has recently drawn a lot of attentions. Fan et al. <cit.> proposed incremental algorithm that identifies a portion of the data graph affected by the update regarding the query. The authors in <cit.> used the join scheme as algorithms (edge-at-a-time). The algorithm maintained a left-deep join tree for the query, with each vertex maintaining a partial query and the corresponding partial results. Then one can compute the incremental answers of each partial query in response to the update, and utilizes the join tree to re-construct the results. Graphflow <cit.> solved incremental subgraph matching using join, in the sense that the incremental query can be transformed into m independent joins, where m is the number of query edges. Then they used the worst-case-optimal join algorithm to solve these joins in parallel. Most recently, Kim et al. proposed TurboFlux that maintains data-centric index for incremental queries, which achieves good tradeoff between performance and storage.

Query Languages and Systems. As the increasing demand of subgraph matching in graph analysis, people start to investigate easy-use and highly expressive subgraph matching language. Neo4j introduced Cypher <cit.>, and now people are working on standardizing the semantics of subgraph matching based on Cypher <cit.>. Gradoop <cit.> is a system based on Apache Hadoop that translates a Cypher query into a MapReduce job. Aberger et al. proposed EmptyHeaded based on relational semantics for graph processing, in which they leveraged worst-case optimal join algorithm to solve subgraph matching. Arabesque <cit.> was designed to solve graph mining (continuously computing frequent subgraphs) at scale, while it can be configured for single subgraph query.     




§ CONCLUSIONS


In this paper, we implement four strategies and three general-purpose optimizations for distributed subgraph matching based on dataflow system, aiming for a systematic, strategy-level comparisons among the state-of-the-art algorithms. Based on thorough empirical analysis, we summarize a practical guide, and we also motivate interesting future work for distributed subgraph matching. 













abbrv
  




§ AUXILIARY EXPERIMENTS

Exp-6 Scalability of Unlabelled Matching. We vary the number of machines as 1, 2, 4, 6, 8, 10, and run the unlabelled queries q_1 and q_2 to see how each strategy (, , and ) scales out. We further evaluate “Single Thread”, a serial algorithm that is specially implemented for these two queries. According to <cit.>, we define COST of a strategy as the number of workers it needs to outperform the “Single Thread”, which is a comprehensive measurement of both efficiency and scalability. In this experiment, we query q_1 and q_2 on the popular dataset LJ, and show the results in scalability. Note that we only plot the communication and memory consumption for q_1, as q_2 follows similar trend. We also test on the other datasets, such as the dense dataset GP, the results are also similar.

 

All strategies demonstrate reasonable scaling regarding both queries. In terms of COST, note that is slightly larger than 1, because “DualSim” is implemented in general for arbitrary query, while “SingleThread” uses a hand-tuned implementation. We first analyze the results of q_1. The COST ranking is (1.6), (2.0), (3.1) and (3.7). As expected, scales worse than , while scales worse than because shuffling cost is increasing with the number of machines. In terms of memory consumption, it is trivial that constantly consumes memory of graph size. Due to the use of , both and consume very low memory for both queries. Observe that consumes much more memory than and , even more than the graph data itself. This is because that certain worker may receive more edges (with duplicates) than the graph itself, which increases the peak memory consumption. For communication cost, both and demonstrate reasonable drops as the increment of machines. renders much less communication as expected, but it shows increasing trend. This is actually a reasonable behavior of , as more machines also means more data duplicates. For q_2, the COST ranking is (2.4), (2.75), (3.82) and (71.2). Here, is dramatically larger, with most time spending on deduplication (multiway_join). The trend of memory consumption and communication cost of q_2 is similar with that of q_1, thus is not further discussed.

Exp-7 Vary Desities for Labelled Matching. Based on DG10, We generate the datasets with densities 10, 20, 40, 80 and 160 by randomly adding edges into DG10. Note that the density-10 dataset is the original DG10 in labelled_datasets. We use the labelled queries q_4 and q_7 in this experiment, and show the results in vary_density. 



Exp-8 Vary Labels for Labelled Matching. We generate the datasets with number of labels 0, 5, 10, 15 and 20 based on DG10. Note that there are 5 labels in labelled queries q_4 and q_7, which are called the target labels. The 10-label dataset is the original DG10. For the one with 5 labels, we will replace each label not in the target labels as one random target label. For the ones with more than 10 labels, we randomly choose some nodes to change their labels into some other pre-defined labels until they contain the required number of labels. For the one with zero label, it degenerates into unlabelled matching, and we use unlabelled version of q_4 and q_7 instead. The experiment demonstrates the transition from unlabelled matching to labelled matching, where the biggest drop happens for all algorithms. The drops continue with the increment of the number of labels, but less sharply when there are sufficient number of labels (≥ 10). Observe that when there are very few labels, for example, the 5-label case of q_7, actually performs worse than and . The “CFLMatch” algorithm <cit.> used by relies heavily on label-based pruning. Fewer labels render larger candidate set and more recursive calls, resulting in performance drop of . While fewer labels may enlarge the intermediate results of and , but they are relatively small in the labelled case, and does not create much burden for the 10GBps network.





§ AUXILIARY MATERIALS

All Query Results. In the_results, We show the number of results of every successful query on each dataset evaluated in this work. Note that DG10 and DG60 record the labelled queries of q_1-q_9.

  


Join Plans. We show the detailed join plans of the queries we used in Exp-3 (experiments). We do not present P_2 as it is randomly generated, and gives poor performance in the experiment.





