TAPER: query-aware, partition-enhancement for large, heterogenous graphs

In this paper we present TAPER, a graph re-partitioning system that is sensitive to evolving query workloads. Let \(\mathcal{Q} = \{(Q_1, n_1) \dots (Q_h,n_h)\}\) denote a query workload, where \(n_i\) is the relative frequency of \(Q_i\) in \(\mathcal{Q}\), and let \(P_k(G)\) be an existing k-way partitioning of G. This could be for instance a simple hash-based partitioning, or one based on an established method such as Metis [12] multilevel partitioning [12] or spectral recursive octasection [8].

The goal of TAPER is to enhance \(P_k(G)\), by computing a new partitioning \(P_k'(G,\mathcal{Q})\) from \(P_k(G)\) that takes \(\mathcal{Q}\) into account. The new partitioning is obtained by swapping vertices across the partitions of \(P_k(G)\), using heuristics that attempt to minimise the total probability of \( ipt \), denoted total \( extroversion \), that occur during execution of any of the queries in \(\mathcal{Q}\). As this method only involves moving relatively few vertices from one partition to another, it is much less expensive than a complete re-partitioning, even after many iterations.

Furthermore, by virtue of its incremental nature, TAPER is able to track changes \(\mathcal{Q} \rightarrow \mathcal{Q'}\) in the workload, by re-partitioning its own partitioning, i.e.,In general, given an initial, possibly workload-agnostic, and non-optimal initial partitioning \(P^0_k(G)\), TAPER can be used to compute a progression of partitionings:These partitionings have the property that each \(P^i_k(G,\mathcal{Q}_{i}) \) exhibits lower extroversion than \(P^0_k(G)\) given \(\mathcal{Q}_i\): it is approximately optimised for that workload.

TAPER makes use of space-efficient main-memory data structures to encode \(\mathcal{Q}\) and to associate estimates of traversal probability with the edges in G. These are then used to calculate the extroversion of each vertex in its partition. A TAPER re-partitioning step, as in Definition 1, is actually several internal iterations of a vertex-swapping procedure aimed at reducing extroversion one vertex at a time. Each time a new TAPER invocation is required (1) we update our data structures for \(\mathcal{Q'}\) and \(G'\), recompute vertex extroversion and begin a new round of iterations.

Our specific contributions are as follows: Firstly, from the notion of stability of a partition [4] we derive an operational metric of partitioning quality, expressed in terms of extroversion for each vertex; Secondly, we describe an encoding of traversal probabilities for each edge in G, given \(\mathcal{Q}\), which is space-efficient and show how they can be updated following the evolution of \(\mathcal{Q}\); Thirdly, we show how TAPER makes use of these structures to iteratively achieve a re-partitioning step (Definition 1).

We present an extensive evaluation of the TAPER system using both real and synthetic graphs of varying sizes, and compare its performance and scalability against one-off workload-agnostic partitionings obtained using the popular Metis approach,¹ without edge weights. In our experiments we use both a simple hash-based partitioning as well as a Metis partitioning as a starting point \(P^0_k(G)\) for one invocation of TAPER. Our results show that such an invocation converges to a stable quality within 6–7 internal iterations, and that the resulting new partitioning \(P^{i+1}_k(G,\mathcal{Q})\) exhibits \(70\%\) quality improvement when a hash-based \(P^0_k(G)\) starting point is used, and about \(30\%\) improvement when using a Metis initial partitioning.

Finally, we show experimentally how the quality of a partitioning degrades following successive simulated changes in \(\mathcal{Q}\), and how it is successfully restored by repeatedly invoking TAPER on the current partitioning and the new workload.

Two main strands of prior work are relevant to our study: (1) workload aware replication and data placement in distributed databases; and (2) graph partitioning.

In the context of online applications, distributed database queries that traverse multiple partitions are expensive [20], incurring high communication cost and, in some implementations, resource contention. In order to achieve good latencies, distributed databases must find a data placement strategy which minimises these transactions, and the overhead they cause, whilst maintaining a balanced load across all machines involved. This is also known as reducing average query span, i.e. the number of partitions involved in answering a query.

For graph data, balanced graph partitioning has been exhaustively studied in literature since the 1970s [8, 12, 13, 16, 23, 25, 26, 28, 30], and a number of practical implementations are available [12, 23]. We do not seek a new graph partitioning algorithm; rather, to propose a workload-driven method for improving partitions that already exist.

This goal is orthogonal to that of the Loom [7] system, also proposed by the present authors. Loom uses a variant of the traversal probability encoding presented here (Sect. 4) to produce a workload-aware partitioning of a graph stream. This may seem to support our goal with TAPER, however, like other streaming graph partitioners [26, 28], once Loom has assigned a vertex to a partition it will never reassign it. This has several implications: firstly, Loom assumes that a workload \(\mathcal{Q}\) is known a priori with respect to vertex assignment; secondly, Loom’s partitionings are vulnerable to aforementioned degrading quality given changes to \(\mathcal{Q}\) over time; and thirdly, Loom may only partition a graph fully, rather than take an existing partitiong as input. This is in contrast with TAPER, which tracks an ongoing workload \(\mathcal{Q} \rightarrow \mathcal{Q'}\) and may continuously adapt an existing graph partitioning for good performance given a random \(Q_i \in \mathcal{Q'}\).

Note that TAPER’s goal of refining an existing partitioning is supported by ParMetis [13], a parallel implementation of the established Metis partitioner [12]. However, this process is communication and computation intensive and as such is often only used as a “one-off” step, rather than for repeated repartitioning of a graph [29]. Additionally, like other existing partitioners, ParMetis is agnostic to a changing query workload. Although ParMetis may be provided with weights which correspond to the traversals of individual edges by a query workload, such weights are much too expensive to maintain over time [32].

Aside from Loom and ParMetis, Curino et al. [3, 20] have proposed systems to tackle a problem related to our own: that of workload-aware data placement in distributed RDBMS. In particular, Schism [3] captures a query workload over a period of time, modelling it as a graph, each edge of which represents tuples involved in the same transaction. This graph is then partitioned using existing “one-off” techniques to achieve a min edge-cut. Mapped back to the original database, this partitioning represents an arrangement of records which causes a minimal number of transactions in the captured workload to be distributed. In SWORD, Quamar et al. [22] build upon the ideas presented in Schism [3]. They use a compressed representation of the workload graph and perform incremental re-partitioning to improve the partitioning’s scalability and sensitivity to workload changes.

Although the goal of these works and our own is similar, there exist major differences in approach. For instance, in Schism, edges directly represent the elements accessed by a query, rather than their labels as we do. However this fine grained approach produces very large graphs, which are expensive to both partition and store, and may impact scalability [22]. Furthermore, these works are focused on a relational data model, where typical workloads overwhelmingly consist of short, 1–2 “hop” queries. This justifies Quamar et al’s simplifying decision, when repartitioning a graph, to consider only queries which span a single partition. However, this assumption does not hold for general graph path and pattern matching queries. It is unclear how SWORD’s approach would perform given a workload containing many successive join operations, equivalent to the traversals required for graph pattern matching.

Other systems, such as LogGP [30] and CatchW [25], are explicitly workload-aware graph partitioners, but focus on improving analytical workloads designed for the bulk synchronous parallel (BSP) model of computation. In the BSP model a graph processing job is performed in a number of supersteps, synchronised between partitions. Both systems keep a historical log of edges traversed in previous supersteps and use this to predict future traversals and move involved vertices so that their edges are internal to partitions, reducing \( ipt \) in subsequent supersteps.

Yet more partitioning systems [29, 31] consider other aspects of a graph’s workload, besides edge traversals. Xu et al. [31] note that, for many BSP workloads, reducing the total communication time of a workload by simply minimising a graph’s inter-partition edges may not be an appropriate optimisation. This is due to the fact that a cluster of machines may be heterogeneous, i.e. machines may have different levels of computing power, or be virtualised across various physical hosts, affording different levels of inter-machine communication. Xu et al. therefore propose modelling the topology of a cluster upon which a graph partitioning will reside, as a graph itself. Subsequently, given some basic information about a workload’s communication and computation characteristics, vertices and edges of the data graph are assigned to appropriate machines, minimising workload execution time.

The work of Vaquero et al. [29] is perhaps the most relevant to our own with TAPER. They propose a system of iterative vertex swapping to adapt to graph changes over time (e.g. vertex additions and removals). This is highly effective at maintaining a good partitioning over time, w.r.t min edge-cut. However, because the system optimises for min edge-cut, it does not consider the heterogeneity of a graph, or that of its query workload. Thus the work we present here could complement that of Vaquero et al: they consider changes to the graph whilst we consider changes to its workload.

Not only are all these systems [3, 22, 25, 29‐31] focused on different types of workload than we are with TAPER, but in an online setting, workload summaries which directly measure individual edge traversals [3, 25, 30] are vulnerable to overfitting. Path queries in particular are often only executed over a subset of the graph, i.e. find Post connected to Person a. If this subset changes (e.g. to Person j) then such a granular workload summary would be inaccurate. By considering instead the vertex labels in a workload’s queries and the graph structure local to a given edge e, TAPER is able to predict e’s traversal probability even when it has not yet been traversed.

Further prior work has focused on exploiting statistical properties of a query workload, to efficiently manage graph data through replication [18, 21, 32]. Whilst [18, 21] focus on 1-hop queries over social networks, Yang et al. [32] propose algorithms to efficiently analyse general online query workloads and to dynamically replicate “hotspots” (cross partition clusters of vertices that are being frequently traversed), thereby temporarily dissipating network load. Whilst highly effective at dealing with unbalanced query workloads, Yang et al. focus solely upon the replication of vertices and edges using temporary secondary partitions. They do not improve upon the initial partitioning, nor do they consider workload characteristics when producing it. This can result in replication mechanisms doing far more work than is necessary over time, adversely affecting the performance of a system. As a result, the enhancement techniques we present here would also complement many workload aware replication approaches, such as that proposed by Yang et al.

The broad goal of TAPER is to increase the quality of a k-way partitioning (Sect. 1.1). Here we define the measure of partition quality which we aim to increase. For this, we extend the notion of partition stability, first introduced by Delvenne et al. [4] in the context of multi-resolution community detection in graphs; stability is described in terms of network flow. The main intuition is that, when a partition is stable, a flow that originates from a point within a partition and moves randomly along paths should be trapped within the same partition for a long time t. Network flow in graphs is modelled as a random walk, where discrete time t is measured as the number of steps. More precisely, the stability of a partition \(V_i\) is defined as the probability that it contains the same random walker both at time \(t_0\) and at time \(t_0 + t\), less the probability for an independent walker to be in \(V_i\) : \(p(V_i, t_0, t_0+t) - p(V_i, t_0, \infty )\). Note that this definition allows for the possibility of a walker crossing multiple partition boundaries before returning to its initial partition at any time during the \([t_0, t_0 + t]\) interval. The overall stability of a partitioning \(P_k(G)\) is the sum of the stability of all partitions \(V_i\) where \(1\le i\le k\). In other words, the greater the stability of a partitioning, the higher the probability that a random walker, having traversed t steps, will be in the same partition where it started.

We extend stability by creating a new measure of partition quality which we will refer to as workload-aware stability. Our extensions are driven by two main requirements. Firstly, TAPER aims to improve the quality of a graph partitioning by minimising the probability of expensive inter-partition traversals, when executing a given query workload \(\mathcal{Q}\) (Definition 1 in Sect. 1.1). Using stability, which models network flow as random walkers that traverse paths in a graph, gives us more flexibility than other measures of partition quality, such as edge-cut, when we try to incorporate information on a query workload. Stability’s ‘walkers’, represented by the probabilities in a transition matrix, may be modified to account for the specific graph patterns associated with the queries in \(\mathcal{Q}\), along with their relative frequency. This will reveal different dominant traversal patterns and produce a measure of quality more closely correlated with the cost of executing \(\mathcal{Q}\) over a particular graph partitioning. Secondly, the current definition of stability as given above is limited, as it does not account for the probability that a walker crosses partition boundaries multiple times within t steps. In contrast, we need to be able to estimate the probability that the walker does not leave the partition within the interval.

We address both requirements by extending the well-known notion of a biased random walk over a graph. Rather than uniform transition probabilities, such a random walk assumes the more general Markov property; that is, the probability of a transition from vertex \(v_k\) to \(v_j\) only depends on the prior probability of being in \(v_k\):In this case, the probabilities \(Pr(v_k \rightarrow v_j| v_k)\) are captured by a transition matrix M:and the probability of a t-steps walk from \(v_k\) to \(v_j\) is computed as \(M^t[k,j]\). However, taking into account the query matching patterns, as per our requirements above, invalidates the Markov property, because the probability of a transition \(v_k \rightarrow v_j\) now depends on the specific path through which we arrive at \(v_k\):in general, for any two paths \(p \ne p'\) leading to \(v_k\).

In other words, in order to account for query matching patterns of length up to t, where t is defined by the query expressions in \(\mathcal{Q}\), we use a multi-step (non-random) walk model over the graph, which has memory of the last t steps. Each transition probability \(v_k \rightarrow v_j\) is now explicitly conditioned on the paths, of length up to t, which lead to \(v_k\).

To represent these probabilities, we extend M to a set of matrices:where t denotes the longest query matching pattern in \(\mathcal{Q}\), and \( VM ^{(k)}\) has dimension \(1 \le k \le t\). We use the term Visitor Matrix (\( VM \)) to refer to (4).

The definition of \( VM \) is by induction, where the base cases are the prior probabilities \(Pr(v_{i})\) to be in \(v_i\), for \( VM ^{(1)}\), and the normal transition matrix M, for \( VM ^{(2)}\). Formally:for \(2 < k \le t\). Figure 2 shows a representation of a Visitor Matrix with \(t=3\), using a 2-dimensional matrix layout where \( VM ^{(3)}\) is “appended” to \( VM ^{(2)}\). The cells in the matrix store probabilities for paths in the example graph to the right (originally Fig. 1), relative to query expression \(Q_1\). For example, path \(1 \rightarrow 2 \rightarrow 3\) is an instance of query pattern abc, and its probability is stored in \( VM ^{(3)}[1,2,3]\) (similarly for the other highlighted elements in the matrix). A \( VM \), like any finite transition matrix, is right-stochastic, i.e., each row sums to 1, and the cells represent all paths up to length t. We show how compute the elements of \( VM (t)\) for a given query workload \(\mathcal{Q}\) in Sect. 4.2.

In practice, we assume \( VM (t)\) is partitioned into n sub-matrics \( VM _i(t)\), one for each of n partitions, because we can always find a permutation of the rows and columns of \( VM \) such that \( VM _i(t)\) is a contiguous sub-matrix of \( VM \). Thus, in the following we use \( VM _i(t)\) to refer the \( VM \) for partition \(V_i\).

Note that the visitor matrix is impractically large to compute, with a space complexity of \(O(|V|^t)\). In Sect. 5.2 we present heuristics that are designed to reduce both space complexity, as well as to avoid computing some of the cells in the \( VM \).

Informally, we define the extroversion of a vertex v to be the likelihood that it is the source of an inter-partition traversal, given any of the query patterns in \(\mathcal{Q}\). TAPER seeks to enhance a partitioning by determining a series of vertex swaps between graph partitions such that their total extroversion is minimised. This is an extension of the general graph partitioning problem, a classic approach to which is the algorithm KL/FM, proposed by Kernighan and Lin [14] and later improved upon by Fiduccia and Mattheyes [5]. They present techniques that attempt to find sets of vertices and edges which, when moved between two halves of a graph bisection, produce an arrangement that is globally optimal for some criteria (usually min edge-cut). Karypis and Kumar [12] subsequently generalise this technique to address the problem of k-way partitioning, in an algorithm that they call Greedy Refinement. Greedy Refinement selects a random boundary vertex³ and orders the partitions to which it is adjacent by the potential gain (reduction in edge-cut) of moving the vertex there, subject to some partition balance constraints. If a move does not satisfy chosen balance constraints, progressively less beneficial destination partitions are considered. Finally, the move will be performed. Greedy Refinement has been shown to converge within 4–8 iterations. It is this algorithm which we use as the basis for our TAPER’s vertex swapping procedure. However, rather than reduction in edge-cut, we use the reduction in extroversion as our measure of gain for evaluating vertex swaps.

There are some other key differences between our own approach, and that of Greedy Refinement. Firstly, Greedy Refinement considers vertices at random from the boundary set, whilst we consider only the set of most extroverted vertices, in descending order of extroversion. This reduces the number of swaps performed and so should improve performance. Secondly, Greedy Refinement is designed to operate on a graph compressed using a matching algorithm, so every vertex move corresponds to the movement of a cluster of vertices in the original graph. Without this trait, Greedy Refinement would be more susceptible to being trapped in local optimisation minima: as vertex clusters are iteratively moved across partition boundaries edge-cut may temporarily increase. We do not operate on a compressed graph; instead we opt for a simple flood fill approach, detailed in Sect. 5.5. Using traversal probabilities, precomputed in the visitor matrix, we identify a vertex v’s family: those vertices likely to be the source of traversals to v. This is the clique of vertices which should accompany a swapping candidate to a new partition.

We now formally define a vertex’s extroversion (symmetrically, along with its introversion), in terms of the \( VM \). Given \(v \in V_i\), we have seen that a VM cell \( VM _{i}^{(k+2)}[v_{i_1}, \ldots , v_{i_{k}}, v, v']\) denotes the probability of a transition from v to \(v'\), given a path \(p = v_{i_1} \rightarrow \ldots \rightarrow v_{i_k} \rightarrow v\) that matches a query pattern. Let \( paths (v, V_i)\) denote the set of all such paths in \(V_i\), i.e. those that match a query pattern in \(\mathcal{Q}\) and end in v. We define \( introversion (v)\) of v as the total probability of such transition occurring, summed over every path \(p \in paths (v, V_i)\) and every destination vertex \(v' \in V_i\). Formally:where for path \(p= v_{i_1} \rightarrow \cdots \rightarrow v_{i_k} \rightarrow v\) of length \(k+1\), we have:and the total intra-partition traversal probability is divided by the total probability of all traversal paths to v:to account for the percentage of the traversals from v that are internal.

Symmetrically, we define the extroversion of vertex v as the total likelihood of inter-partition traversal \(v \rightarrow v'\), where \(v \in V_i\) and \(v' \in V_j\), \(j\ne i\). As the VM is right stochastic and we may assume that a partition’s VM forms a sub-matrix of the global VM, inter-partition probabilities are the complement to 1 of the intra-partition probabilities:

Given a trie, such as in Fig. 3b, we associate a probability to each node in the trie, reflecting the relative likelihood that a sequence of vertices with those labels will be traversed in the graph. These probabilities are periodically (re)calculated by considering both the individual contribution of each query Q to the trie structure, as well as the frequency with which Q appears in the workload during some preceding time t.

To understand these calculations, consider again Fig. 3, where we assume that \(Q_1\), \(Q_2\) each occur once in \(\mathcal{Q}\) over time t, i.e., they have the same relative frequency. Starting from root \(\mathcal {E}\), consider transition \(\mathcal {E} \rightarrow a\). Its probability can be expressed as:where the conditional probabilities are computed using the labels on the nodes and the \(Pr(Q_i)\) are the relative frequencies of the \(Q_i\). In the example we have \(Pr(Q_1) = Pr(Q_2) = .5\), \(Pr(\mathcal {E} \rightarrow a |Q_1) = 1\) because a is the only possible first match in \(Q_1\)’s pattern, and \(Pr(\mathcal {E} \rightarrow a |Q_2) = .5\) because initially \(Q_2\) can match both a and c, with equal probability. Thus, \(Pr(\mathcal {E} \rightarrow a) = 1 \cdot .5 + .5 \cdot .5 = .75\).

We can now use \(Pr(\mathcal {E} \rightarrow a)\) to compute \(Pr(\mathcal {E} \rightarrow a \rightarrow b)\) and \(Pr(\mathcal {E} \rightarrow a \rightarrow c)\):whereand \(Pr(\mathcal {E} \rightarrow a \rightarrow b | Q_2) = 0\) because pattern \(\mathcal {E} \rightarrow a \rightarrow b\) is not feasible for \(Q_2\). Thus, \(Pr(\mathcal {E} \rightarrow a \rightarrow b) = .5 \cdot .5 = .25\).

Formally, we identify each node n in the trie by the sequence of k steps \((n_1,n_2,\ldots ,n_k)\) required to reach it from the root node, \(\mathcal {E}\). A probability label p(n) is then associated with each node, its value computed as follows:The individual terms of the sum are conditional probabilities over the path in the trie to node N. As we have seen in the example, these conditional probabilities over the paths are computed recursively on the length k:

The TPSTry encodes the current likelihood of traversing from a vertex with some label, to any connected vertex with some other label (Sect. 4.1). This is an abstraction over the values we actually need for the visitor matrix, which are vertex-to-vertex transition probabilities. We may derive the desired vertex transition probabilities, from a path of previously traversed vertices \(p = {p_1, p_2, \ldots , p_k}\), as follows. First we look up the the path’s corresponding sequence of vertex labels in the pattern summary trie. This returns a set of child trie nodes \(n \in N\) which represent legal labels for the next vertex to be traversed, along with each label’s associated probability p(n). Subsequently, the traversal probabilities for each label are uniformly distributed amongst those neighbours of \(p_k\) which share that label. This produces a vector of traversal probabilities, one for each neighbour of the \(p_k\). This vector corresponds to a row in the visitor matrix.

For each path of traversals with length \(<\!\!\!t\), the VM assumes that a subsequent traversal is guaranteed, i.e. the total traversal probability in each row is 1, and the VM is stochastic. In reality some paths of traversals must have a total length \(<\!\!t\), either because a query expression defines a path of a shorter length, or because a vertex does not have a neighbour with the label required by a query expression. A query execution engine would stop traversing in such a scenario. We represent this non-zero probability of no subsequent traversal from a vertex as probability to traverse to the same vertex,⁶ as this is equivalent to intra-partition traversal probability.

In Sect. 2.3 we described a VM cell as containing the probability of traversing to a vertex v given some preceding sequence of traversals \(p_1 \rightarrow p_2 \rightarrow \cdots \rightarrow p_{t-1}\). Formally, we compute the value of a cell \( VM ^{(t)}[p_1, \ldots , p_{t-1}, v]\) aswhere \( l : V \rightarrow L_V\) is the labelling function for a graph G, and \(N_G: V \rightarrow V\) corresponds to the set of neighbours of v such that \((v, n) = e\in E\) for all \(n\in N_G(v)\). The latter term of this definition uniformly distributes the traversal probability to a vertex with label l across all of of \(p_{t-1}\)’s l labelled neighbours.

The relative frequency of d from \(a\rightarrow b\) is also 0.5 . Vertex 2 has the neighbours 1,3,4 and 5 with the labels a, c, d and c respectively. As an example, the probability of traversing to vertex 3 is the probability of traversing to a c labelled vertex, divided by the number of c labelled neighbours of 2.Therefore, as shown in Fig. 4 (left), we have \( VM ^{(3)}[1,2,*] = (0,0,0.25,0.5,0.25,0)\).

In the previous section (Sect. 4.1) we mention that TPSTry probabilities are periodically updated to reflect query frequencies changing over time. We do not recompute VM cells for each change to the TPSTry, instead they are lazily re-evaluated each time a vertex swapping iteration (Sect. 3.1) is triggered. Additionally, we store a snapshot of the TPSTry at the point of the pervious vertex swapping iteration; if a trie node’s probability remains the same between two iterations, we are able to safely avoid recomputing its associated VM cells.

We have implemented a system prototype on top of the Tinkerpop graph processing framework,⁷ which allows us to use any of several popular GDBMS to store G. Though our prototype, built using the Akka framework,⁸ is designed to be distributed across multiple hosts, in the current implementation input graph partitionings reside on a single host. Partitions are defined in terms of vertex-cut, as opposed to edge-cut: inter-partition connections are represented by flagging cut vertices and annotating them with the partitions they belong to. We have extended Tinkerpop so that multiple edge-disjoint subgraphs are treated as a single, global, graph and queried using the Gremlin query language.⁹ An inter-partition traversal is detected when a Gremlin query retrieves the external neighbours of a cut vertex. Our test architecture is shown in Fig. 5. It simulates a distributed deployment, where each partition is logically isolated, managed by a separate instance of the TAPER algorithm implementation. Each instance is responsible for updating the Visitor Matrix for its partition, and also determines the rank of extroverted vertices to evict at each iteration.

As noted in Sect. 2.3, the space complexity of the VM for each partition \(V_i\) grows with the number of vertices in the partition, and exponentially with the length of the query patterns: \(O(|V_i|^t)\). Here we discuss two heuristics, aimed at reducing the portion of the VMs that need to be explicitly represented or computed for each partition, reducing both the time and space complexity of the TAPER algorithm.

TPSTry (Sect. 4) is implemented as two separate data structures: (i) a trie multimap, where each trie node maps to the set of queries which could be responsible for a traversal path with the associated sequence of vertex labels; and (ii) a sorted table mapping queries to their respective frequencies. These frequencies are approximated using a sketch datastructure which samples the occurrences of each query within a sliding window of time t.

TAPER relies upon an ordering of the vertices in a partition by their likelihood to be the source of inter-partition traversals. In order to produces this order, we group the rows of a partition’s visitor matrix by the final vertex of the paths they represent and then derive their extroversion (Sect. 3.2).

As a result of the heuristics defined above, not all vertices are represented in the visitor matrix. Therefore we refer to the sorted set of vertices produced as a partial extroversion ordering. Rather than grouping the rows of a pre-existing matrix, we define a corecursive algorithm to efficiently produce such rows consecutively during initial VM construction. This greatly simplifies the process of maintaining a running total of intra-partition transition probabilities for each vertex, as required for the heuristics presented in Sect. 5.2. A simplified version of the procedure is expressed in Algorithm 1. Consider again our earlier example of vertex 3 in partition B (Fig. 1), along with the pattern summary trie in Fig. 4 (right). Vertex 3 has the label c, which does exist as a prefix in the trie. It has local neighbours 5 and 6, along with external neighbours 2 and 4, labelled c,a,b and d respectively. The external transition probability from 3 given a path of (3) is \(1 - \Sigma (0, 1, 0)\) multiplied by the probability to have made the sequence of traversals which that path represents (\(\frac{0.25}{|c|} = 0.125\)). In this case: 0.

The prefixes cc and ac also exist in the trie, therefore (5, 3) and (6, 3) are further potential paths through 3. The external transition probabilities from 3 given paths of (5, 3) and (6, 3) are \((1-\Sigma (0, 0, 1))\cdot 0.125\) and \((1-\Sigma (0, 0.25, 0.5))\cdot 0.25\) respectively. Note that the total probability for the path (6, 3, j), \(j\in V_B\) is \(<1\) because acd is also a prefix in the pattern summary trie, and vertex 3 is adjacent to the “external” vertex 4. The final external transition probability from 3 is 0.06; its extroversion \(0.06 \cdot \frac{1}{0.5} = 0.12\).

To achieve its aims, TAPER improves distributed query performance by reducing the probability of inter-partition traversals when answering queries. For each partition, given a sorted collection of the vertices with the highest extroversion, TAPER must reduce this probability without mutating the underlying graph structure.

To this end, we propose a simple variation on the k-way Kernighan-Lin algorithm proposed by Karypis and Kumar [12] (Sect. 3.1). This is a two-step, symmetric process, shown in Fig. 6: firstly, given a priority queue of candidate vertices with high extroversion, compute the preferred destination partition for each vertex, along with the clique of neighbours which should accompany it (its family); secondly, when offered a new group of vertices, a partition should compute potential gains in introversion and decide whether or not to accept the offer.

We determine a swapping candidate’s family set with a simple recursive procedure: Given each family member (initially just the candidate), we examine its local neighbours; if a traversal from a neighbour to the member is more likely than not, then it is added to the family. Once the family-set has been determined, we evaluate the total loss in introversion the sending partition would suffer from their loss. This process is highly efficient as all the relevant values are preserved in the visitor matrix, either from calculating the introversion of vertices, or from constructing a candidates set in the previous step.

When on the receiving end of a swap, a partition should calculate the total local introversion of a family, and compare it to the potential loss to the offering partition. Partitions are “cooperative” rather than greedy, so if the introversion gain for a receiving partition partition is not greater than the loss for a sending one, the swap is rejected. If a swap is rejected the offering partition will try less “preferable” destinations until all partitions adjacent to the candidate vertex are exhausted. In this event the candidate vertex and its family remain in their original partition.

This process runs independently for each partition, swapping extroverted vertices to their preferred neighbouring partitions. A vertex may only be swapped once per iteration of the algorithm. Note that, because partitions calculate potential vertex swaps independently of one another, it is possible that highly extroverted connected vertices would “flip-flop” between two partitions without ever improving workload-aware stability. In order to avoid this we adopt another technique from (Par)Metis [13]. First, every partition is assigned an ordered identifier. Subsequently, we split vertex swapping into two phases: Initially, given its queue of extroverted vertices, a partition will only attempt swaps whose destination partition has an id lower than itself. In the second phase, each partition will do the reverse, swapping vertices with partitions of a higher id. For each iteration, the ordering of these phases flips (higher\(\rightarrow \)lower, lower\(\rightarrow \)higher, ...). This process prevents vertices “chasing” one another continuously by ensuring that when the introversion loss is calculated by the second partition, both offending vertices will be present locally.

When swaps have been attempted for every vertex in each partition’s extroverted queue, the iteration is considered finished. Where necessary, we update each partition’s VM by adding and removing rows for swapped vertices, the resulting subgraph acts as input to a subsequent iteration of vertex-swapping. Repeated iterations of this process will produce the desired result: an enhanced partitioning with a lower overall probability for inter-partition traversals, better workload-aware stability given a query workload \(\mathcal{Q}\).

For all experiments we initially partition the test graphs using either hash or Metis. Except where otherwise stated, these are 8-way partitions. As mentioned, we consider the quality of a partitioning as its workload-aware stability (Sect. 2.2), corresponding to the probability of inter-partition traversals when executing a workload \(\mathcal{Q}\). We measure this experimentally by executing snapshots of query workloads over partitioned graphs and counting (Sect. 5) the number of inter-partition traversals \( ipt \). All algorithms, data structures and dataset pre-processing steps, including calcVMRows and the TPSTry, are publicly available.¹⁰ All our experiments are performed on a machine with a 3.1Ghz Intel Core i7 CPU and 16GB of RAM.