{'fraction_non_alphanumeric': 0.04945710258515203, 'fraction_numerical': 0.010207400752776444, 'mean_word_length': 3.819498357718722, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
DDSL: Efficient Subgraph Listing on Distributed and Dynamic Graphs

Subgraph listing is a fundamental problem in graph theory and has wide applications in areas like sociology, chemistry, and social networks. Modern graphs can usually be large-scale as well as highly dynamic, which challenges the efficiency of existing subgraph listing algorithms. Recent works have shown the benefits of partitioning and processing big graphs in a distributed system, however, there is only few work targets subgraph listing on dynamic graphs in a distributed environment. In this paper, we propose an efficient approach, called Distributed and Dynamic Subgraph Listing (DDSL), which can incrementally update the results instead of running from scratch. DDSL follows a general distributed join framework. In this framework, we use a Neighbor-Preserved storage for data graphs, which takes bounded extra space and supports dynamic updating. After that, we propose a comprehensive cost model to estimate the I/O cost of listing subgraphs. Then based on this cost model, we develop an algorithm to find the optimal join tree for a given pattern. To handle dynamic graphs, we propose an efficient left-deep join algorithm to incrementally update the join results. Extensive experiments are conducted on real-world datasets. The results show that DDSL outperforms existing methods in dealing with both static dynamic graphs in terms of the responding time.


I. INTRODUCTION
In real-world applications, the underlying data can often be modeled as graphs. For example, the World Wide Web can be treated as a graph, where each single vertex represents a page, and each edge represents a hyper-link between two pages. Give another example, we can model a social network as a graph by treating users as vertices and friend relationships as edges.
In this paper, we study subgraph listing, one of the fundamental problems in graph theory. Given two undirected and unlabeled graphs d and p, it requires to list all subgraphs of the data graph d, which are isomorphic to the pattern graph p. Such a subgraph is also called a match. Subgraph listing has wide applications in different areas. For example, in sociology, chemistry, telecommunication and bioinformatics, subgraph listing is used in comparing large graphs [1], [2]. In activity networks, it is used to monitor potential terrorists by searching threat patterns [3]. It can also be adapted to track social network evolution [4] and to identify properties that are useful in recommendation networks [5].

A. Motivation
Traditional centralized solutions [6], [7], [8] for subgraph listing try to enumerate subgraphs on a single machine. However, the result size can be exponential to the size of the data graph, which causes heavy I/O cost when we want to materialize the result on the disks. In this case, disk I/Os become the bottleneck of the whole algorithm, and centralized algorithms are thus not preferred in dealing with large graphs. To address this issue, other works [9], [10], [11], [12], [13] focus on solving this problem in distributed environments. By partitioning and processing the data graph and the intermediate results (i.e., the partial matches) on the cloud, they reduce the I/O cost on each single machine and the total running time as well.
Another issue is the highly dynamic nature of modern graphs, where vertices and edges are added and deleted from time to time. For example, the number of monthly active users in Facebook increased to more than 1.94 billion in 2017 1 since it was founded in 2004. On average, in every second there are 4.7 new users appear in this social network. In such a situation, when we want to monitor the matches of a certain pattern, it is not efficient to treat the updated graph as a new one and run the algorithm from scratch. Instead, only tracking the change of the result can be more efficient and thus is important.
Although existing distributed algorithms [14], [9], [10], [11], [12], [13] achieve good performance on static graphs, few of them targets exact subgraph listing on distributed and dynamic graphs. In practice, there are two major difficulties to handle dynamic graphs. Firstly, it requires efficient updating of the underlying data storage of d. In works [11], [12], one of the preprocessing steps is to list all k−cliques for k ≤ k 0 in d, which takes significant amount of time. When d changes, whether the storage of cliques can be updated efficiently is unknown, so this could prevent the whole algorithm from efficiently handling dynamic graphs. Secondly, it also requires updating the result efficiently without doing too much redundant calculation, otherwise there would be no difference compared to running the original algorithm from scratch.

B. Overview and Contributions
In this paper, we propose Dynamic and Distributed Subgraph Listing (DDSL), which attempts to solve the exact subgraph listing problem on distributed and dynamic graphs. DDSL can be fitted into general distributed data processing engines like MapReduce [15], Spark [16] and Dryad [17], and we describe it in MapReduce for the ease of presentation. In those systems, the I/O cost and the communication cost are often considered as the bottleneck of the whole algorithm [10], therefore, we conduct our analysis according to the amount of data involved in disk I/Os and communications at each step in our approach.
The whole approach of DDSL can be divided into two stages: 1) Initial Calculation. In this stage, DDSL first constructs the distributed storage of data graph d, and then lists all matches of pattern graph p in data graph d from scratch. 2) Incremental Updating. In this stage, according to the change of d, DDSL first updates the distributed storage of d, and then updates the matches of p. The initial calculation follows a general distributed join framework [11], which is shown in Fig. 1. It has two main building blocks: a distributed storage mechanism, and a set of join units. Firstly, according to the distributed storage mechanism, the data graph d is decomposed into several parts (d 1 , d 2 ). Then the pattern graph p is decomposed into several join units (q 1 , q 2 ), and the matches of each unit can be directly listed from each part of d without join. By gradually joining the matches of all units together, we can obtain the matches of the pattern graph.
Note that, all partial matches generated in this framework are first saved to disks and then loaded for join. Thus, those matches are treated as intermediate results, and are counted into the total cost. In Fig. 1, matches of q 1 and q 2 (M (q 1 , d) and M (q 2 , d), resp.) are intermediate results. Actually, if we view the join order as a tree structure (called the join tree), all nodes except for the root correspond to a match set which belongs to the intermediate results.
Different storage mechanisms may support the direct listing of matches of different join units. For example, two different storage mechanisms have been proposed in [10], [11] to support stars and/or cliques as join units. Intuitively, as Example 1.1 shows, supporting more general units offer more flexibility in finding lower join cost solution. In DDSL we use a distributed storage mechanism called Neighbor-Preserved (NP) storage, which supports all graphs with radius=1 as join Example 1.1: Fig. 2 shows three different join trees of a typical pattern. In Fig. 2a, each join unit is a triangle, so the pattern graph is decomposed into 4 triangles. In Fig. 2b and Fig. 2c, the join units are cliques and R1 units, respectively. Note that a triangle is a special clique of size 3, and a clique is a special R1 unit. In this example, using more general join units can reduce the number of joins. Specifically, using cliques can reduce the number of joins by 2 compared with using triangles, and using R1 units requires no join at all. To further reduce the I/O and communication cost of the join operations, we incorporate the idea of match compression into our method. Specifically, we use the vertex-cover-based compression (VCBC) [12] to compress all the matches of join units. A nice property of the VCBC is that, we can perform join operations directly on the compressed data without decompression, which saves the computation cost. Besides the compression, we derive a comprehensive cost model to estimate the join cost of any given join tree. Based on this cost model, we use a dynamic programming to find the optimal join tree which has the minimum join cost.
In the incremental updating stage, we design an algorithm which can update the NP storage according to the graph changes with a low cost. To update the matches, our idea is to extract a patch set that contains only the newly-appeared matches. Then we can simply merge the old matches with the patch set, and filter out the matches that no longer exist. However, computing the patch set using the regular join is no easier than computing all matches from scratch, because we may not effectively filter out the unnecessary matches on either join side. In DDSL, we design a novel Navigated Join (Nav-join) to extract the patch set with a lower cost. We first decompose the pattern graph into a left-deep tree, and then compute a partial patch set using the Nav-join along this tree. For each involved join, we partition the matches on the left side to the corresponding machines and perform a local exploration, which saves the I/O cost for listing and loading the matches on the right side. The final patch set is then obtained by merging several partial patch sets computed through a set of carefullypicked trees.
In summary, we make the following contributions.
• To the best of our knowledge, DDSL is the first approach that supports unlabeled and undirected pattern matching on dynamic graphs in a distributed environment. • We derive a comprehensive cost model for estimating the cost of a given join tree, and then design a dynamic programming algorithm to find the optimal join tree which has the lowest estimating cost. • To handle dynamic graphs, we design an efficient algorithm update the NP storage within a single MapReduce round. We then propose the novel Nav-join to efficiently compute the patch set. By merging the patch set with the old match set, and filtering out invalid matches, we can obtain the updated result. • We conduct extensive experiments on real-world graphs to demonstrate the efficiency of DDSL on both static and dynamic graphs.

C. Paper Organization
The rest of this paper is organized as follows. In Section II, we introduce the basic concepts in this paper. In Section III, we describe the underlying distributed storage and how we do the initial calculation in DDSL. Then in Section IV and Section V, we propose several optimizations to reduce the cost of initial calculation. To handle dynamic graphs, we propose the incremental updating algorithm in Section VI. Section VIII surveys the related works and compares them to our work. In Section VII, we conduct extensive experiments on real-world networks, to show the efficiency of our approach on both static and dynamic graphs. Finally, we conclude in Section IX.

II. PRELIMINARIES
In this section we introduce the basic concepts and definitions. We also list important notations used in this paper as well as their meanings in Table I.

A. Graph and Subgraph
Given a graph g, we use V (g) and E(g) to denote the vertex set and edge set in g, respectively. Each vertex v ∈ V (g) is associated with a unique id i, and thus is denoted by v i . The data graph d and pattern graph p. v i , u i A vertex which has id i.
An edge that connects v i and v j .
The set of vertices and edges in g.

Ng(v)
The neighbor set of v in g.
The subgraph of g induced by V .
The match set of p on d.
The anchor-center-constrained match set of q.
The optimal join tree of p.

Vc(p)
A vertex cover of p. s A skeleton(a partial match from Vc(p) to V (d)). f |s The compressed form of all matches having s. loc(u) The local graph of vertex u.
The Neighbor-Preserved storage of d.
The set of edges to be added/deleted.

An edge connecting
In this paper we focus on undirected and unlabeled graphs, which means Given two graphs g 1 and g 2 , g 2 is a subgraph of g 1 if V (g 2 ) ⊆ V (g 1 ) and E(g 2 ) ⊆ E(g 1 ). Specifically, g 2 is the subgraph of g 1 induced by vertex set V if (1) V (g 2 ) = V ⊆ V (g 1 ), and (2) E(g 2 ) = E(g 1 )∩(V ×V ). In this case we also denote g 2 by g 1 [V ]. Apparently any graph g is the subgraph of itself induced by V (g).

B. Subgraph Listing
By introducing the concept of graph isomorphism, we can then formally define the subgraph listing problem.
Definition 2.1 (Graph Isomorphism [18]): Given two graphs g 1 and g 2 , an isomorphism from g 1 to g 2 is a bijection f : If there is an isomorphism from g 1 to g 2 , then we say g 1 is isomorphic to g 2 .
Definition 2.2 (Subgraph Listing): Given two connected graphs g 1 and g 2 , subgraph listing requires outputing all subgraphs of g 1 , which are isomorphic to g 2 . Here g 1 is also called the data graph, and g 2 is also called the pattern graph.
For the ease of description, in the rest of this paper, we use d to denote the data graph, and p to denote the pattern graph. Each isomorphism f from p to a valid subgraph of d is called a match. Next we introduce the automorphism of a graph. As it introduces duplicate matches and harms the efficiency of the algorithm, we also describe the technique to break it.
The automorphism of a graph g is the isomorphism f : V (g) → V (g). Any graph has an automorphism f (v i ) = v i , while some graphs have more. In subgraph listing, if the pattern graph p has k automorphisms, one will find k matches p to each valid subgraph of d. Extra effort has to be paid to remove such duplicates, and thus will harm the efficiency. A common way to avoid duplicate results is Symmetric Breaking (SimB) [19]. It assigns a partial order ord = {v i ≺ v j } for some node pairs v i , v j ∈ V (p) × V (p), and a total order Then an isomorphism f is valid if and only if f (v i ) < f (v j ) for any v i ≺ v j . By carefully assigning the partial order, SimB guarantees that for any graph g which is isomorphic to p, there is only one valid isomorphism from p to g. In the following of this paper, we assume that SimB has been applied, and for each valid subgraph of d, we can find only one match. In this case, listing all valid subgraphs is equal to listing all valid matches, which is denoted as M (p, d).

C. Graph Update
For a general undirected and unlabeled graph, there are 4 types of updates: • Adding a vertex; • Deleting an existing vertex, as well as all edges incident to it; • Adding an edge between two existing vertices; • Deleting an existing edge. Since we require d and p to be connected, we only consider edge insertion/deletion in this paper. Adding a vertex u can be automatically done when we add an edge (u, v), and deleting u can be simulated by first deleting all edges incident to u, then deleting u automatically.
One way to handle graph changes is to consider the changes as a stream, and to only deal with one single insertion/deletion at a time. Another way is to consider the changes as several batches, where each batch contains several edges to be inserted and deleted simultaneously. In this paper we use batch updates for two reasons. First, by considering a stream as batches with only one edge, the solution on batch updates can also apply on stream. Second, each time we update the graph and the result, some basic cost like reading the whole graph is unavoidable, and by dealing with multiple edges simultaneously we may save such cost. Under this setting, a graph update U contains two edge sets E d (U ) and E a (U ), where E d (U ) contains edges to be deleted and E a (U ) contains edges to be added.

D. Dynamic Graph and Problem Statement
A dynamic graph is defined as an initial graph d, followed by several updates U 1 , U 2 , . . .. Assuming that after applying each of those updates, we get the updated graphs d , d , . . ., our goal is to output the match set M (p, d), M (p, d ), M (p, d ), . . . efficiently. Here we take a simplified version: only one update exists. The reason is that, if we can compute M (p, d ) efficiently, we can handle the next one by treating d as the new initial graph. We then describe the target problem as follows.
Problem Statement. Given a distributed file system, which can store and load data as key-value pairs, a pattern graph p, a data graph d, and a graph update U , the problem has two goals: where d is the new graph after applying U on d.

E. Power-Law Random Graph Model
In this paper, we use the Power-Law Random (PR) graph model [20] to analyze the data graph d. Based on it, we build a comprehensive estimator to estimate the number of matches given any pattern graph p and data graph d.
It is known that in most real-life networks, the degree of vertices follows the power-law distribution. Thus, modeling the underlying graph as a PR graph can usually give a realistic estimation of its properties. In a PR graph d, if two assumptions hold. One is that self-loop is allowed, and another is that for any Although not all graphs satisfy these two conditions, this model still serves as a good guidance in practice [10].

III. INITIAL CALCULATION
In the initial calculation, the task is to perform regular subgraph listing on the data graph d. As discussed, in this stage we follow a distributed join framework, which has two basic building blocks: 1) The set of supported join units.
2) The distributed storage mechanism of d to support the direct listing of matches of each join unit.
In this section we describe the details of these two building blocks that are used in DDSL.

A. R1 Units
As Example 6.1 shows, using more general join units may reduce the intermediate result size. Thus, instead of using only one or two kinds of join units, in DDSL we use all graphs with radius=1 as our join units, which are called R1 units. They have the following advantages: • R1 units contain a variety of shapes, and thus has the potential to reduce the join cost. • Using R1 units allows us to compress the join results (see Section IV), and most importantly, to incrementally update the join results (see Section VI).
Anchor Vertex. Given an R1 unit q, we can find at least one vertex, who is the common neighbor of all other vertices. We randomly pick such a vertex as the anchor vertex of q. Later we will utilize the anchor vertex to facilitate the listing and join process.

B. Neighbor-Preserved Storage
In the join-based framework, matches of each join unit are listed from each part of the graph storage completely and independently. Suppose d is stored in m parts d 1 , d 2 , . . . , d m , and q is a join unit, then the completeness requires that is, every match exists in at least one partition and no match will be lost. The independence requires that is, no duplicate matches are found in different partitions. If these two properties are satisfied, we can obtain M (q, d) by simply adding all matching set M (q, d i ) together.
To support directly listing the matches of any R1 unit, we use the Neighbor-Preserved (NP) storage mechanism. An NP storage Φ(d) is built upon an arbitrary partition function h, and the local graph loc(u) of each vertex u ∈ V (d).
Definition 3.2 (Partition Function): A partition function h maps vertex ids to partition ids. Given any vertex id, h outputs a partition id j ∈ [1, m], where m is the number of partitions.
Perceptually u i lies on the "center" of loc(u i ), so we call u i a center vertex of d j if h(i) = j. For non-center vertices of d j , we call them border vertices.
One can easily verify that M (q, d i ) does not satisfy the independence requirement, however, we can use an alternative version M ac (q, d i ) which satisfies the anchor-center constraint. We first describe how M ac (q, d i ) is computed, and then proof the independence and completeness of it in Lemma 3.1. Given q and d i , M ac (q, d i ) is computed by listing all matches from q to d i , under the constraint that q's anchor vertex must be matched to one of d i 's center vertices. This can be done using existing in-memory subgraph listing algorithms with minor modifications. For example, in a DFS algorithm, we can put q's anchor vertex in the first level, and only try to match it to d i 's center vertices.
Lemma 3.1: With M ac (q, d j ), any R1 unit can be the join unit w.r.t. an NP storage.
Proof: We proof this by showing the independence and completeness of M ac (q, d i ).
Independence. We proof the independence by showing that any two different sets have no common matches. Let q's anchor vertex be v 0 , and f , f be two arbitrary matches in M ac (q, d i ) and M ac (q, d j ), respectively. Suppose f (v 0 ) = u x and f (v 0 ) = u y , which means h(x) = i and h(y) = j. Since h is a partition function, i = j =⇒ x = y, and thus f = f .
Completeness. We further prove the completeness by showing that any valid match in M (q, d) must exist in one M ac (q, d i ). Let q's anchor vertex be v 0 , and f (v 0 ) = u x . By the definition of an anchor vertex, we have We proved that f ∈ M (q, d h(x) ), and we also know u x is a center vertex in d h(x) . Combining these two we have To build the NP storage Φ(d), we store some duplicate edges between border vertices in each part. Then the important question is: whether the space cost of Φ(d) can be bounded? We analyze and give a confirmative answer to this question w.r.t. the extra edges needed to build Φ(d).
We start with an arbitrary edge (u x , u y ).
where such edges stored in each loc(u z ) are the extra edges introduced to close the triangle (u x , u y , u z ). In other words, for each triangle (u x , u y , u z ) ∈ d, three extra edges (u x , u y ), (u x , u z ) and (u y , u z ) are stored in loc(u z ), loc(u y ) and loc(u x ), respectively. When we take the union of local graphs that belong to the same d i , some edges might be merged. For example, if loc(u x ) and loc(u y ) are both stored in d j , then (u x , u z ) and (u x , u y ) are each stored only once, saving two edges. In summary, the total extra edges introduced The 3 · ∆(d) bound is suitable for a sparse d or a large m. When d is sparse, ∆(d) is small. When m is large, for each triangle in d, there are few chances that two or three of its vertices are center vertices of the same d j . In both cases there are only a few edges that are merged, so 3 · ∆(d) is close to the true extra space cost of Φ(d). However, when d is dense, or m is very small, there are many merged edges, and the gap between 3 · ∆(d) and the true extra cost is large. In the most extreme case, when d itself is a clique, and m = 1, the true extra cost is 0, while ∆(d) ≈ |V (d)| 3 . We thus bound the extra cost in another way. Since d i cannot be larger than d, we simply bound the extra cost as (m − 1) · E(d). Noticed that both bounds are hold for all situations, thus in summary,

C. Pattern Decomposition and Join
With the two building blocks decided, we can perform the subgraph listing within the join framework. Given a pattern graph p, we decompose p into a set of join units Q = {q 1 , q 2 , . . . , q k }, such that: This decomposition can be done by simply generating all possible R1 units inside p, and then exhaustively search a subset that satisfies the second condition.
After listing the matches of all the join units in Q, we then need to join the matches together. Assuming that we have This guarantees that there exists no conflict when merging the mappings in f 1 and f 2 , and the vertex set {f i (v)|∀v ∈ V (p 1 )∩ V (p 2 )} is called the join keys. In Fig. 1, we illustrate a simple join between M (q 1 , d) and M (q 2 , d), where the columns to generate join keys are marked in gray. After the join, we need to check each result, and drop every match that (1) maps two or more vertices in p to the same vertex in d, or (2) violates the partial order ord.
Next we will discuss several optimizations to reduce the cost in this framework.

IV. MATCH COMPRESSION
In general, given any order . However, such a plain storage may occupy a lot of space. For example, the authors in [12] find that for a 6-vertex pattern and a data graph with 0.1 million vertices, the match set can take 10 4 petabytes of storage. They also show that applying match compression can significantly reduce the storage size as well as the I/O cost. Thus, it is worth incorporating the match compression technique into our method.
The idea of match compression is to use a compressed form to store multiple matches together, in order to save the cost of disk I/O and network communication cost. There are different compression strategies in the literature, and in DDSL we use the vertex-cover-based compression (VCBC) [12], because it can be incorporated into the join process without introducing extra I/O cost of decompression. To make the paper selfcontained, we first explain the compression and decompression of VCBC in Section IV-A and Section IV-B, then discuss its compression ratio in Section IV-C. In our analysis, we assume that each vertex can be stored in an integer, and use the number of integers to measure the storage size. For example, if we store all matches in the plain form, then for k matches we need to store k · r integers in total.

A. Compression
Let V c (p) be a vertex cover of p, and s :  Proof: The uncompressed form stores c · (a + b) integers. For each match, f |s needs to store at most b integers, one for each vertex in V (p)\V c (p). Thus, in total f |s needs to store less than a + b · c integers, which saves at least a · (c − 1) integers compared to the uncompressed form.

B. Decompression
The decompression of a compressed match f |s can be done in two steps: i. For all v ∈ V (p)\V c (p), take the Cartesian product of the vertex sets they are mapped to. Each tuple plus the skeleton s becomes a candidate match; ii. For each candidate match, drop it if: (1) it is not an injection, i.e., maps two vertices in p to the same one in d; or (2) it violates the partial order ord. After these two steps, the remaining matches are the recovered matches.
Definition 4.1 (Correct Compressed Match): Given a compressed match f |s of p, f |s is correct if the following condition holds 1) ∀f ∈ M (p, d), which has skeleton s, f can be recovered from f |s; It is proved that the vertex-cover-based compression in Section IV-A is correct [12].

C. The Optimal Compression
A pattern graph can have numerous vertex covers, where each corresponds to a unique compressed match set. Lemma 4.1 gives the lower bound of a specific match set M |s, however, usually we care more about the total compression ratio R given the pattern p and the vertex cover V c (p). Computing the precise ratio requires to compute all matches, which are unavailable unless we finish the subgraph listing task. Therefore, we try to compute a lower bound of R with less information. Theorem 4.1 gives a lower bound of R, given the size of M (p, d) and M (p[V c (p)], d).
Theorem 4.1: Given p, V c (p), and d, the lower bound of the compression ratio is Proof: The uncompressed form stores each match separately, so the storage needed is We break the compressed storage into two parts: the skeleton part and the compressed part.
Each skeleton is a match of the induced graph p[V c (p)] in d, so the number of distinct skeletons is no greater than |M (p[V c (p)], d)|. As skeletons are stored separately in each f |s, so the maximum storage required is: Following Lemma 4.1, the compressed form stores at most |V (p)| − |V c (p)| integers for each match, so the maximum number of integers stored by the compressed part is Combining Equation 4, Equation 5 and Equation 6, we have the lower bound of the compression ratio: Currently the actual values of |M (p, d)| and |M (p[V c (p)], d)| are unknown, and computing these two values is NP-hard. In this case we introduce an estimator of these two values in Section IV-D. For any pattern p, there exists a vertex cover which gives the highest R lower , Such a compression provides the best guaranteed compression ratio, and thus is called the optimal compression.

D. Match Size Estimation
In order to estimate R lower w.r.t. a given vertex cover V c (p) of p, we need to estimate the total number of matches of p in d. To do so, we model the data graph d as a PR graph, and compute the expected match size, which is denoted by E (|M (p, d)

|).
Our computation is based on a random-assignment process. In this process, we first randomly assign a match f from V (p) to V (d). Then we test whether f is a valid match. If f is a valid match, then it satisfy two conditions: . We first consider a simplified version of MC 1 , which is denoted as MC 0 . Given V (p) = {v 1 , v 2 , . . . , v k }, and a degree sequence W = w 1 , w 2 , . . . , w k , where ∀w i : , the degree of vertex f (v i ) in d must be w i . Let X 0 , X 1 and X 2 be the event that MC 0 , MC 1 and MC 2 are satisfied, respectively, then we have wiwjρ.
Note that in the formula of Pr(X 2 |X 0 ), term w i appears once for each adjacent edge of v i , so in total w i appears deg(v i ) times. Therefore, we can write Thus, given W , the probability that f is valid is Recall that MC 0 is a special case of MC 1 , so f satisfies MC 1 if there exists any one W s.t. f satisfies MC 0 . Therefore, the probability that f satisfies MC 1 and MC 2 is Assuming that the pattern graph p has O(1) vertices, the term Then E(|M (p, d)|) is calculated as times the number of all possible assignments, then divided by the number of automorphisms of p satisfying ord, which is where the first term refers to the number of all possible assignments from V (p) to V (d), and Auto(p, ord) refers to the automorphisms of p that satisfy ord. Please note that for the original pattern graph p, there is always only one automorphism satisfying ord, because we apply SimB to break all other automorphisms of p using ord. However, the decomposed subgraphs of p may have several automorphisms w.r.t.. ord, so the last term is not a constant. Again, when p has O(1) vertices, the evaluation of Equation 9 can be evaluated in O(1) time.
Comparison. Authors in [11] also propose a function to estimate the match size. That function relaxes the degree constraints, that is, it allows a vertex in p to be mapped to a vertex with smaller degree in d with a small probability. However, in practice this probability should be 0. Meanwhile, it doesn't consider the partial order ord, which may filter out part of the matches. Thus, the result of that function includes some invalid matches. Our calculation considers there two factors and thus provides a smaller and more precise result.

E. Revised Pattern Decomposition and CC-Join
Because we use match compression in DDSL, the pattern decomposition and join process need to be adjusted to support compressed matches. In this section we describe how DDSL maintains the compressed form f |s from the beginning, and keeps this form along the whole join process.
Given a pattern p and a vertex cover V c (p), DDSL decomposes p into a set of join units Q = {q 1 , q 2 , . . . , q k }, which satisfies the following conditions: Condition (1) and (2) are common requirements in a joinbased framework, to ensure that M (q i , d) can be directly listed and used to recover M (p, d). We add condition (3) to make sure that the compressed form of matches of each q i w.r.t. V c (p) ∩ V (q i ) can be listed directly from Φ(d).
Proposition 4.1 implies that V c (p) ∩ V (q i ) is a vertex cover of q i , which means we can compress the matches of q i w.r.t.
Proposition 4.1: Given any graph p, a vertex cover V c (p), and its subgraph p , V c (p) ∩ V (p ) is a vertex cover of p .
Proof: For any edge (v i , v j ) ∈ E(p ), without loss of generality, assume that Then we prove in Theorem 4.2 that the compressed matches of each q i w.r.t. V c (p)∩V (p ) can be listed directly from Φ(d).
We also summarize the process in Algorithm 1.  After listing the compressed matches of each q i from Φ(d), the next step is to join the matches together. When matches are not compressed, the join of two matches f 1 : , and the join process is simply adding all mappings in f 1 and f 2 together, then check the validity of the result. In DDSL, all the matches are compressed, so the join condition and join process are modified. Before we describe the details, it is important to show the consistency of the compressed forms.
. Proposition 4.2 implies that a vertex v will remain compressed (or not compressed) in every f |s, if we compress the matches of each subgraph p w.r.t. V c (p) ∩ V (p ). We call all such compressed matches the Consistently Compressed (CC) matches under V c (p). Apparently in DDSL, the matches of join units are CC matches under V c (p). To join two CC match sets, we perform the following CC-join.
Join Condition. We reduce the condition on only the skeletons of two matches. Given p and V c (p), two compressed matches f 1 |s 1 of p 1 and f 2 |s 2 of p 2 can be joined if ∀v ∈ Join Process. During the join process, compressed and uncompressed vertices are handled separately. For uncompressed vertices, we simply union the mappings together. For a compressed vertex v, if v appears in both two matches, say f 1 |s 1 and f 2 |s 2 , then we take the intersection of two mapped sets, i.e., v → f 1 |s 1 (v) ∩ f 2 |s 2 (v), otherwise we just keep the only mapping of v. Suppose the join result is f 3 |s 3 , we then check the validity of each compressed vertex v, and remove u from f 3 |s 3 (v) if v → u does not appear in any valid decompressed match. Finally we return f 3 |s 3 (v) as the result.
The join process is summarized in Algorithm 2. Theorem 4.3 shows that after performing CC-join on two correct CC matches, the result is also a correct CC match. Therefore, we can gradually join the CC matches of join units (which are correct) together in a specific order, to obtain all the correct Algorithm 2: CC-Join Input : V c (p), f 1 |s 1 of p 1 , f 2 |s 2 of p 2 , and ord.
if u is not valid then 10 Remove u from f 3 |s 3 (v); 11 return f 3 |s 3 ; CC matches of the pattern graph p. Theorem 4.3: Supposing that by joining two CC matches f 1 |s 1 of p 1 and f 2 |s 2 of p 2 , we get the match f 3 |s 3 of p 3 , then f 3 |s 3 is a correct CC match of p 3 under V c (p), which has skeleton s 3 .
Proof: We first prove the consistency, i.e., f 3 |s 3 is a CC match under V c (p), which has skeleton s 3 .
During the join process, we compute s 3 as . Then we prove the correctness. The skeleton s 3 is obtained by a regular join of two uncompressed matches, so its correctness is guaranteed, and we only focus on compressed vertices.
1) For any f ∈ M (p 3 , d), which has skeleton s 3 , we examine an arbitrary vertex v ∈ V (p 3 )\V c (p). Since v is either in one of V (p 1 )\V c (p) and V (p 2 )\V c (p), or in both, we discuss these two situations separately: • v is in either V (p 1 )\V c (p) or V (p 2 )\V c (p). Without loss of generality, assume that v ∈ V (p 1 )\V c (p).
Thus, for each v ∈ V (p 3 )\V c (p), and for each f ∈ M (p 3 , d), we have f (v) ∈ f 3 |s 3 (v). When decompressing f 3 |s 3 , any valid f ∈ M (p 3 , d) will appear in the Cartesian product, and be in the output.
2) The analysis in (1) also holds in the opposite direction, that is, given an arbitrary vertex v ∈ V (p 3 )\V c (p) and any u ∈ f 3 |s 3 (v), v → u exists either in valid matches in M (p 1 , d) (or M (p 2 , d)), or in both. In each case we can derive that ∀(v, v ) ∈ E(p 3 ) : (u, f 3 |s 3 (v )) ∈ E(d).
Thus, each decompressed match preserves all edges in p 3 , and is a valid match if it passes the validation check in Section IV-B.

F. Revised Optimal Compression
In general, a join process is parallelized in the following way (supposing we are joining M 1 |s 1 of p 1 and M 2 |s 2 of p 2 ). 1) We know the join condition is built on V key = V (p 1 ) ∩ V (p 2 ) ∩ V c (p), and thus we can generate a join key {s(v)|∀v ∈ V key } for each f |s involved in this join. It is obvious that any two matches satisfying the join condition must have the same join key; 2) The distributed system can aggregate matches with the same key together in a worker, and then different workers can perform the join on different join keys simultaneously. If we use the optimal compression which maximizes R lower , it may not be suitable for parallelizing the join process in a distributed system. For example, in Fig. 3, the optimal compression is the one in (c), however, using this compression we cannot generate the join key. Thus, the join degenerates to the Cartesian product between all matches and cannot be parallelized efficiently. Specifically, Lemma 4.2 shows that to well parallelize the join process, the induced graph p[V c (p)] should be connected.
Lemma 4.2: Given p and V c (p), if p[V c (p)] has at least two connected components, then during the CC-join process, there is at least one CC-join involving two patterns p 1 and p 2 , such that V (p 1 ) ∩ V (p 2 ) ∩ V c (p) = ∅.
Proof: Treat the whole process as a join tree. The root corresponds to p it self, where p[V (p) ∩ V c (p)] has more than one connected component. Each leaf corresponds to a join unit q, where q[V (q) ∩ V c (p)] has only one connected component, because q's anchor vertex is in V c (p), and it is connected to all other vertices in V (q). Therefore, we can find at least one node in the join tree, such that 1) it corresponds to a pattern p , where p [V (p )∩V c (p)] has two connected components, and 2) its children corresponds to pattern p 1 and p 2 , where To facilitate the join process, we require the induced graph p[V c (p)] to be connected. Under this condition, we pick the compression that maximizes R lower . Such a compression is called an optimal connected compression.
V. THE OPTIMAL JOIN TREE Given a set of join units Q and a specific join order, the whole join process forms a join tree. A typical join tree T (p) contains a set Q(T (p)) = {q 1 , . . . , q k }, and a set P (T (p)) = {p 1 , p 2 , . . . , p l }, where each q i is a join unit on a leaf node, and each p i is a subgraph of p on an internal node. Assuming that we already have the optimal connected compression of p, choosing different join trees may give us different performance, depending on how much intermediate result is produced. Thus, in DDSL we try to find the optimal join tree, aiming at minimizing the I/O and communication cost produced during the whole join process.
Definition 5.1 (Optimal Join Tree): Given p, V c (p), and a cost function Cost, a join tree T (p) is optimal if 1) The root of T (p) corresponds to p.
2) Each leaf in T (p) corresponds to a join unit.
3) It minimizes Cost(T (p)) among all possible trees that satisfy condition (1) and (2). In MapReduce, the I/O cost and communication cost come from the map data, the shuffle data, and the reduce data. Following the join process, we compute the total cost of any specific join tree T (p) by adding up the cost from each part.
Cost of Processing Leaves. To list all matches of a join unit q, the mappers take Φ(d) as the input, and output all compressed matches of q. The output is directly writing into disk, so no shuffle or reduce is needed. Suppose storing Φ(d) and all matches of q requires S(Φ(d)) and S(q) storage, respectively, then the cost is S(Φ(d)) + S(q). In DDSL, we list the matches of all join units in a single MapReduce round, and for each unit we store its matches in a unique file. In this way, the total cost for processing all leaf nodes in T (p) is

S(q).
Cost of Processing Internal Nodes. To list all matches of a non-unit pattern p i = p l i ∪ p r i , the mappers take the matches of p l i and p r i , and output each match associated with the join key. All key-match pairs are then read and shuffled to their corresponding reducers. Finally each reducer takes matches with the same join key, and output the join result. The total cost incurred in map, shuffle and reduce is 2·S(p l i )+2·S(p r i ), 2 · S(p l i ) + 2 · S(p r i ) and S(p l i ) + S(p r i ) + S(p i ), respectively. Cost of Decompression. After we computing the compressed matches of p, we may need to decompress the results to get the plain matches. This can be done in a single map function. The mappers take S(p) as the input, and output the decompressed matches. Therefore the cost is simply S(p) + |V (p)| · |M (p, d)|. Note that, this decompression is optional. We can decompress the result when we need the plain form, or keep it the compressed form otherwise.
By summarizing the cost of processing all nodes in T (p) and rearranging the terms associated with each pattern, we write the total cost of T (p) as The cost S Φ(d) and S(p) + |V (p)| · |M (p, d)| introduced in processing the leaf nodes and the final decompression, are constants if p and d do not change, so we ignore these terms when possible, and write Equation 10 in a recursive form: if pi is a join unit It suggests that we can use dynamic programming to compute the optimal join tree, which is summarized in Algorithm 3. In line 1 − 3 we initialize the tree set T with all valid join units. From line 4 to line 10 we gradually union the small patters to bigger ones, and keep record of the minimum cost. Note that line 6 ensures that the join key can be generated for every possible join. In line 11 we recursively construct the whole join tree by following its children all the way to the join units.

Algorithm 3: Find Optimal Join Tree
Input : V c (p), p.
Output: The optimal join tree T opt (p).
compute Cost(p ); 9 if p / ∈ T or T (p ).cost > Cost(p ) then 10 T (p ) ← (Cost(p ), p i , p j ); 11 return T opt (p) constructed from T (p); Lemma 5.1 shows that Algorithm 3 can find the join tree with minimum estimation cost. Since there can be at most 2 |E(p)| different subgraphs of p, and each of them might be joined with any other patterns, the time complexity of this algorithm is O(4 |E(p)| ). Considering that we filtered out many invalid joins, the actual running time is much smaller in practice.
Lemma 5.1: The join tree found by Algorithm 3 is optimal (has the lowest estimation cost).
Proof: We use the strong induction to prove that every node in the optimal join tree is stored in T .
1) For each leave node q i , its processing cost is directly computed and stored in T (lines 2-3); 2) For each non-leave node p = p i ∪ p j , if p i and p j are in T , they will be enumerated in lines 5-6, so node p will be explored and stored in T (lines 8-10). 3) For each non-leave node p = p i ∪ p j , assuming that |V (p )| > |V (p i )| ≥ |V (p j )|, line 4 and 5 guarantee that nodes p i and p j are explored and stored in previous steps.
Therefore, the root node T (p) of the optimal tree is in T , and is returned at line 11.

VI. INCREMENTAL UPDATING
In this section we address the problem of handling dynamic graphs. Assuming that a batch U of edge updates is applied on the original data graph d, resulting in a new data graph d , we update Φ(d) and M (p, d) in three steps:

A. Update the NP Storage
To update the NP storage Φ(d), essentially we need to update each local graph loc(u i ) correctly. We first discuss how an edge insertion/deletion influences the edges in loc(u i ), and then extend the discussion to d j , where multiple local graph exists.
Edge Insertion. There are two possible situations that inserting an edge can lead to the change of loc(u i ).
1) Inserting edge (u i , u j ), which means u j becomes a new neighbor of u i . In this case, we need to check all edges adjacent to u j , and add edge (u j , u k ) into loc(u i ) if u k is u i 's neighbor. 2) Inserting edge (u j , u k ), where u j and u k are both u i 's neighbors. In this case, we only need to add (u j , u k ) into loc(u i ). Edge Deletion. Edge deletion is the reverse of edge insertion, so it can be handled by the above two situations with all edge insertions replaced with edge deletions. The only difference is that, in situation (1), we can just delete all edges in loc(u i ) which are adjacent to u j .
Extending our discussion to the update of d k , there are in total three possible cases. C 1 Inserting/deleting edge (u i , u j ), where u i and u j are both center vertices of d k . All edges adjacent to u i or u j must be preserved to store loc(u i ) and loc(u j ), respectively, so we only need to insert/delete edge (u i , u j ). C 2 Inserting/deleting edge (u i , u j ), where u i and u j are both border vertices of d k . For all local graphs in d k , it belongs to situation (2) of insertion/deletion, so we only need to insert/delete edge (u i , u j ). C 3 Inserting/deleting edge (u i , u j ), where u i is a center vertex, and u j is a border vertex. For insertion, we deal as situation (1). For deletion, we first delete (u i , u j ), then for each edge adjacent to u j , we delete it only if no other local graphs contain it. In DDSL we deal with all three cases in a single MapReduce round. Basically, when inserting an edge (u i , u j ) s.t. h(i) = h(j), we need to deal with case C 3 in both d h(i) and d h(j) , which requires to know N d (u j ) and N d (u j ), respectively. Thus, we use the map function to prepare such data, and use the reduce function to update Φ(d). Note that the reduce function takes N d k as one of the inputs, which contains all neighbor sets necessary for adding edges into d k . When dealing with case C 3 , the reduce function can get N d (u i ) (or N d (u j )) from N d k . The whole algorithm is summarized in Algorithm 4.
Each mapper takes d k and U as the input, which causes S(Φ(d))+m·|E(U )| I/O cost. All mappers output the neighbor sets of vertices in U , which are shuffled to the reducer. This incurs at most 3 · ui∈U |N d (u i )| communication cost. The reducers read d k , U and N d k , and then output Φ(d ), so the I/O cost of reduce is S(Φ(d)) + m · |E(U )| + ui∈U |N d (u i )| + S(Φ(d )). In summary, the total cost of updating Φ(d) is

B. Update the Match Set
The update of the match set can be divided into two categories: the removed matches and the newly-appeared matches. In this section we discuss how to identify or extract them separately.
The removed matches exist in M (p, d), but are no longer valid because some edges are deleted. Lemma 6.1 shows that removed matches can be easily detected by checking each match with the deleted edge set E d (U ). Thus, we can remove those matches from M (p, d) during the decompression step, and only output matches that should not be removed. The extra I/O cost comes from reading the deleted edge set E d (U ) by each mapper, which is in total m · |E(U )|. Proof: (If). It is trivial according to the definition of a match.
If these two assumptions hold, then instead of joining two match sets, we may use a partition-and-expand strategy to reduce the join cost. Basically, for a join p i = p j ∪ q k , where q k is a join unit, we can send each f ∈ M new (p j , d ) to several partitions. Then inside each partition d x , we expand f to get the matches of p i . If we choose the partitions carefully, we can guarantee the correctness of the result. Suppose that M new (p j , d ) takes S new (p j ) storage, the total cost is now at most S Φ(d ) + (4m + 1) · S new (p j ) + S new (p i ). According to our assumptions, this cost should be lower than the original bushy-join cost. Based on this idea, we design the Nav-join to compute M new (p, d , q i ) as follows: 1) Find a left-deep tree w.r.t. the join unit set Q, where q i is the lowest leaf. 2) Extract M new (q i , d , q i ) using Algorithm 1 with the constraint that every match must map at least one edge in q i to an inserted edge. 3) For a join in the tree, we compute the result using Navjoin by partitioning the matches on the left side, and expanding them in each part of the NP storage. Note that all matches are still compressed w.r.t. a vertex cover of p. Example 6.2 demonstrates the process of a Nav-join with 3 units and 2 partitions. We now discuss the details and optimizations below. Example 6.2: Consider a join p = (q 1 ∪ q 2 ) ∪ q 3 , and an NP storage of 2 partitions. Fig. 4 shows the 3 steps to compute M new (p, d, q 1 ) using Nav-join. In the first step, we extract the new match sets of q 1 from two partitions separately, and then merge them together. In the second step, we first partition the previous result into 2 parts, s.t. a match f is in partition i only if f can be joined with some matches in d i . Then inside each partition, we expand each match of q 1 to get the new matches of p 1 , where p 1 = q 1 ∪ q 2 . Similarly, in the third step, we partition and expand the matches of p 1 to get the match set M new (p, d, q 1 ).
Optimal Left-Deep Tree. In a join p i = p l i ∪ p r i , the mappers take M new (p l i , d , q i ) as input, and output the matches sent to each d k , which are shuffled to the reducers. The reducers take Φ(d ) and received matches as input, and output M new (p i , d , q i ). According to assumption (1), the main cost would be reading Φ(d ), i.e., S(Φ(d )). Assuming a left-deep tree involves j join units, then the total cost is j · S(Φ(d )), which comes from 1 unit match listing and j − 1 Navjoins. Thus, the optimal left-deep tree is the one involves the minimum number of join units.
Match Navigation. In step (3), supposing the join is p i = p l i ∪ p r i , we decide whether f should be sent to d k as follows: 1) If p r i 's anchor vertex v is used to generate the join-key, and f (v) = u j , then f can only be joined with matches in M (p r i , loc (u j )), where loc (u j ) is the local graph of u j in d . In this case, we only send f to d h(j) . 2) Otherwise, we generate the join-key of f . If a match f can be joined with f , f must have the same join-key, and Step 1: Mnew(q1, d2, q1) Mnew(q1, d, q1) Step 2: Step 3:  thus f (v) must be the common neighbors of all vertices in the join-key. For each vertex u j in the join-key, we send f to d h(j) .
Computing the common neighbors requires to know all neighbors of those vertices in the join-key, however, storing the neighbors of all vertices within d k is costly. For example, in one of the datasets used in our experiments, we find a vertex with more than 10 5 neighbors. If m ≈ 10 2 , this vertex appears in almost every part of Φ(d), and it takes more than 10 7 integers to store its neighbors in every partition. To reduce this storage, we use a bit map of length m for each vertex u, to store the partition ids h(l), ∀u l ∈ N d (u), and check the common partition ids instead of common neighbors. If we use 32-bit integers to simulate bit maps, it takes |V (d )| · m 32 integers in total. When performing the joins of f in d k , we also use the join-key to filter out impossible local graphs in d k . Since the common neighbors now are available in d k , this filtering requires no other storages. Parallelize All Trees. For each q i ∈ Q we need to compute M new (p, d , q i ) according to its left-deep tree. If we compute all sets in serialization, the total cost is S(Φ(d )) · |Q| 2 . We noticed that in each join and unit match listing, the input Φ(d ) can be shared, so we compute all match sets in parallelism. In each MapReduce round, we process the same level of all left-deep trees simultaneously. Since all left-deep tress has the same height |Q| − 1, the calculation can be done in exact |Q| rounds, and the total cost is thus S(Φ(d )) · |Q|.
Match Deduplication. For M new (p, d , q i ), there might be a match f , which maps an edge in q i to an inserted edge, and another edge in q j to an inserted edge. This f will also appear in M new (p, d , q j ) according to our algorithm. To avoid such duplications, we assign a total order {q i < q j if i < j} on Q, then ∀f ∈ M new (p, d , q i ), we keep it only if for any q j < q i , f does not map an edge in q j to an inserted edge. In other words, f can map an edge in q j to an inserted edge if and only if q j > q i . Theorem 6.1: By assigning the total order in the Nav-join, there will be no duplicate matches or lost matches.
Proof: We first show that no math is lost. In the total order, we can find a smallest q i , s.t. q i contains an edge (v i , v j ) which is mapped to an inserted edge by f . It is easy to verify that f ∈ M new (p, d , q i ).
We then show that there is no duplicate matches, i.e., if

VII. EXPERIMENTS
In this section, we conduct two parts of experiments to evaluate the cost of DDSL on static and dynamic graphs separately. Specifically, our experimental study has three main goals: • We illustrate the cheap construction cost and space cost of the NP storage mechanism. • For static graphs, we compare the overall cost of DDSL to the state-of-the-art distributed methods. • For dynamic graphs, we show the efficiency of DDSL in updating the NP storage as well as the matching result.
A. Experiment Setup 1) Datasets and Queries.: In our experiments, we use 4 real-word graphs WebGoogle(WG), WikiTalk(WT), LiveJournal(LJ), and UK-2002(UK) as the data graph, which that are commonly used in recent works [11], [8], [12]. Datasets WG, WT and LJ can be downloaded from SNAP 2 , and dataset UK can be downloaded from WEB 3 . The size of each dataset is listed in Table II. For the pattern graphs, we pick 5 commonly used ones from recent works [11], [8], [12], which are shown in Fig. 5. Below each pattern, we show the corresponding partial order ord used by SimB.
2) Compared Methods.: In the experiments on static graphs, we compare DDSL with two state-of-the-art distributed approaches SEED [11] and Crystal [12]. These two methods are originally designed for the subgraph enumeration task, which does not write the result into disks. In our experiments we make them write all matches into disks to solve the subgraph listing problem. In the experiments on dynamic graphs, since no other works target on the exact problem discussed in this paper, we use Delta-BigJoin [13], a recent approach that handles directed and dynamic graphs, as the competitor. It should be noted that, the running time of DDSL and Delta-BigJoin cannot be directly compared, because these two methods are designed for different tasks. Even if we simulate undirected graphs by duplicating edges, it is still unfair, since some of the optimizations of Delta-BigJoin will not work. Thus, a more proper way is to compare the time increase of each method as the batch size grows.
3) Running Environment.: All methods in our experiments are running on a cluster of 11 machines, one master and 10 slaves. The master has 47GB RAM, two Intel Xeon X5650 CPUs, and one 900GB HDD. Each slave has 125GB RAM, two Intel Xeon E5-2630 CPUs, and one 900GB HDD. We deploy an instance of Apache Hadoop 4 on the cluster. By default we set the number of mappers and reducers to be 200, each with 4GB memory space. To run Delta-BigJoin, we also deploy an instance of Timely Dataflow 5 on the cluster, and set the number of processes to be 200. 4) Parameter Settings.: In the preprocessing step, both SEED and Crystal list all k-cliques for k ≤ k 0 in the data graph, and the user can specify the maximum size k 0 of cliques to be listed. In our experiments, we set k 0 to be 3, which is enough for all the 5 patterns. For DDSL, m is set to be equal to the number of mappers. 5) Evaluation Metrics.: To compare the efficiency of each method, we use the wall-clock time to evaluate the cost of each method. Specifically, we count the elapsed time from submitting a job until it finishes. To show the space cost of NP storage, we use the file size in megabytes as the metric.

B. Experiments on Static Graphs
We divide the experiments on static graphs into two parts. The first part is to evaluate the computation and space cost of preprocessing and storing the data graph. The second part is to evaluate the cost of listing all matches of a given pattern.
1) Preprocessing Costs.: We first study the cost of constructing the NP storage. In Fig. 6a, we compare the preprocessing time of DDSL with SEED and Crystal. In this step, SEED and Crystal list all 3-cliques in the data graph, while DDSL only partition edges to each reducer, and write each part of Φ(d) as a whole in to disks. Compared with WG WT LJ UK Datasets  3) Overall Performance on Static Graphs.: In this part we study the cost of subgraph listing for each method on static graphs. It is notable that the preprocessing is not independent from the listing process. For example, In SEED and Crystal, one can set k 0 to be 4 in stead of 3, so that all 4-cliques are listed. With this additional data, the subgraph listing process of some patterns might be faster, however, listing 4-cliques incurs more I/O cost, and thus the preprocessing takes longer time. Therefore, we use the overall elapsed time as the measurement to better illustrate the overall performance of each method. In Fig. 7 we show the running time of each method on all pattern graphs. The missing bars mean that the running time is larger than 10 4 seconds. In general, DDSL outperforms other two methods in most of the situations. It is slightly slower than Crystal only for processing q 1 and q 3 on LJ. For these two patterns, all three methods performs similar joins while listing the matches, and thus DDSL does not have notable advantages over other two methods. For other three patterns, DDSL has the best performance on all datasets. Specifically, the gap between the running time of DDSL and other two methods are larger when processing more complex patterns. The reason is that, for complex patterns, both SEED and Crystal need to perform several join operations during the process, while DDSL can directly list the matches without join. This result confirms our motivation and the power of supporting R1 units.

C. Experiments on Dynamic Graphs
Experiments in this part also contain two parts. The first part is to evaluate the cost of updating the NP storage, and the second part is to evaluate the cost of updating the matches of a pattern. For both two parts, we generate the update batch by randomly picking edges. Supposing the batch size is b, we randomly picking b/2 edges in d to delete, and then randomly generate b/2 edges that do not exist in d to insert.
1) Cost of Updating the NP Storage: In this part, we enumerate each batch size in {10 2 , 10 3 , 10 4 , 10 5 }, and count the time of updating the NP storage on each dataset. As Fig.  8a shows, DDSL can update the NP storage very efficiently. In comparison to Fig. 6, the updating time is less than the construction time even for the largest batch size. We also noticed that for each dataset, the updating time remains nearly unchanged as the batch size grows, which means the message size during the updating also grows slowly.
2) Cost of Updating the Result: To evaluate the performance of updating the match set, we vary the batch size in {10 2 , 10 3 , 10 4 , 10 5 } on LJ, and count the overall elapsed time of DDSL and Delta-BigJoin. For DDSL, we also include the time for updating of the NP storage, to illustrate its overall performance. We use inf to indicate either the running time is too long, or the memory usage exceeds our capacity.
As Fig. 8b to Fig. 8e show, except for q 2 , the increasing of running time of Delta-BigJoin is much faster than that of DDSL. A possible reason is that, Delta-BigJoin maintains several delta queries at the same time, and as the batch size increases, maintaining those queries causes high pressure to the main memory as well as the communication. Actually, for the smallest pattern q 2 , Delta-BigJoin performs similar to DDSL, which confirms this explanation. On the other hand, DDSL performs stabler. The updating time of all 4 patterns remains to be low. Actually, comparing the updating time with the running time on LJ in Fig. 7, the updating time for every pattern is much smaller than that on static graphs even for the 10 5 batch size.
One may also notice that the updating time of q 1 and q 3 grows faster than that of q 2 and q 4 . The major reason is that, to list the matches of pattern q 1 and q 3 , we need to perform a join operation, while for q 2 and q 4 we directly list the matches. This, again, shows that reducing the join operations can often reduce the total cost.

VIII. RELATED WORK
Centralized Subgraph Listing. Subgraph listing on a single machine has been extensively studied in the literature. Most of the solutions like VF2 [21], QuickSI [22], and GADDI [23] use backtracking and recursing to find the matches. In GraphQL [24] and SPath [25], neighbors' labels of a vertex are used to filtering invalid matches at early steps. Han et al. [26] finds that the matching order can significantly influence the efficiency of the algorithm. More recently, Kim et al. [8] proposes an I/O-efficient algorithm through a dual approach.
Distributed Subgraph Listing. Many approaches are proposed recently to solve the subgraph listing problem in a distributed environment. Sun et al. [14] use the Trinity memory cloud to parallelize a join-based algorithm, which employs STwigs as the join unit. Shao et al. [9] parallelize the traditional DFS algorithm using Pregel [27]. They use several pruning rules as well as the workload-balancing strategy to improve the efficiency. Lai et al. [10], [11] investigate the joinbased algorithms based on MapReduce. They try to reduce the overall I/O cost by introducing different join units and join trees. Gao et al. [28] achieve approximate subgraph listing through message passing. They convert the query graph into a DAG, and use Giraph 6 to pass messages between vertices. Qiao et al. [12] propose a framework to compress the matches with the vertex-cover-based-compression, which can further reduce the I/O cost. More recently, Khaled et al. [13] propose a join algorithm for directed graphs based on a the Timely Dataflow [29] system, which achieves worst-case optimality in terms of computation and communication cost. However, it brings storage pressure to the system by requiring to store the complete data graph on each machine. Also, how to support undirected patterns efficiently in this approach is still an open problem.
Subgraph Listing on Dynamic Graphs. There are only a few works targeting on dynamic graphs. In [28], matches are identified by passing messages between vertices. When edges are inserted/deleted, the +/-messages are passed through the graph to update vertex status as well as the matching results. This method is efficient since many messages are merged within a vertex, however, this also leads to an inexact result. Fan et al. [30] investigate several problems on dynamic graphs, and propose an exact centralized algorithm for dynamic subgraph enumeration. To handle graph changes, it extracts the  union of d p -neighbors for all inserted edges, and applies a centralized static algorithm on this extracted graph. However, as a centralized algorithm, it suffers from the huge I/O cost for the subgraph listing problem on big graphs. For directed graphs, Khaled et al. [13] propose an algorithm to handle graph changes as streaming data.

IX. CONCLUSION
In this paper, we study the problem of subgraph listing on distributed and dynamic graphs. We propose an efficient method, called DDSL, to handle dynamic graphs through two stages: initial calculation and incremental updating. The initial calculation follows a general distributed join framework. In order to reduce the join operations, we use the bounded neighbor-preserved storage mechanism for the data graph, which supports listing the matches of any R1 unit directly. To further reduce the intermediate result size, we incorporate the existing vertex-cover-based compression into this framework. To choose a better join order, we derive a comprehensive cost model, and use a dynamic programming to find the optimal join tree. In the incremental updating stage, we first design an algorithm to update the NP storage, then we propose the novel Nav-join to compute newly-appeared matches. Extensive experiments show that DDSL can handle static subgraph listing with a competitive performance compared with the state-of-the-art distributed methods. Moreover, DDSL can efficiently handle dynamic subgraph listing without computing from scratch. To the best of our knowledge, DDSL is the first approach in the literature that supports unlabeled and undirected pattern matching on dynamic graphs in a distributed environment.