PaperlayerView exact revision on arXiv

Source-linked AI summary

Distributed Evaluation of Subgraph Queries Using Worstcase Optimal LowMemory Dataflows

Khaled Ammar, Frank McSherry, Semih Salihoglu, Manas Joglekar

arXiv:1802.03760v1cs.DCcs.DB

TL;DR

Subgraph queries must be evaluated on large, rapidly changing graphs, motivating efficient distributed processing. The paper recasts worst-case optimal joins as dataflow algorithms for static and dynamic queries; its algorithms provide worst-case resource guarantees, including under skew, and are evaluated on graphs up to 64 billion edges.

  • Problem

    Subgraph queries need to run on large static and dynamic graphs as graph scale and update speeds grow.

    Source: §1 · p. 1

  • Method

    The paper recasts worst-case optimal join algorithms as distributed dataflow computations, developing BiGJoin, Delta-BiGJoin, and skew-resilient BiGJoin-S.

    Source: §1.3 · p. 2

  • Results

    The algorithms achieve worst-case optimal computation and communication with linear total memory, while per-worker costs decrease linearly with added workers even on skewed inputs.

    Source: §1 · p. 1

  • Takeaways & Limitations

    The algorithms and theory apply beyond subgraph queries to computing and maintaining relational equi-joins.

    Source: §1 · p. 1

    In a preliminary implementation, BiGJoin-S overheads exceeded their benefits; workload-balanced techniques remain to be evaluated in an implementation.

    Source: §3.4 · p. 5, §7 · p. 12

Abstract

from arXiv · show

We study the problem of finding and monitoring fixed-size subgraphs in a continually changing large-scale graph. We present the first approach that (i) performs worst-case optimal computation and communication, (ii) maintains a total memory footprint linear in the number of input edges, and (iii) scales down per-worker computation, communication, and memory requirements linearly as the number of workers increases, even on adversarially skewed inputs. Our approach is based on worst-case optimal join algorithms, recast as a data-parallel dataflow computation. We describe the general algorithm and modifications that make it robust to skewed data, prove theoretical bounds on its resource requirements in the massively parallel computing model, and implement and evaluate it on graphs containing as many as 64 billion edges. The underlying algorithm and ideas generalize from finding and monitoring subgraphs to the more general problem of computing and maintaining relational equi-joins over dynamic relations.

1. INTRODUCTION

The paper targets efficient, scalable subgraph evaluation on large static and dynamic graphs, where growing graph scale and update rates motivate distributed processing. It adapts worst-case optimal joins to dataflow and develops algorithms for static and dynamic queries, including a skew-resilient variant.

1.1 Joins and Worst-case Optimality

  • The AGM bound is the maximum possible output size for given input relations, and worst-case optimality keeps computation or communication within O(MaxOutQ).For a triangle query on a graph with IN edges, MaxOutQ is IN^3/2.Source: §1.1 · p. 1

1.2.1 Edge-at-a-time Approaches

  • Edge-at-a-time approaches can require O(IN^2) computation for triangles, exceeding the IN^3/2 AGM bound because their first join can be that large.Changing join order or preprocessing does not correct this asymptotic suboptimality.Source: §1.2.1 · p. 2

1.2.2 The Shares Algorithm

  • Shares can reduce communication and round counts, but its cumulative memory grows as O(w^(1−ε)IN), so adding workers can scale resources inefficiently.For triangles, ε = 1/2; ε is often smaller for other queries.Source: §1.2.2 · p. 2

1.2.3 Vertex-at-a-time Approaches

  • Vertex-at-a-time algorithms extend partial matches one vertex at a time and avoid intermediate data larger than MaxOutQ.NPRR and Leapfrog TrieJoin are instances of Generic Join; TurboISO uses a vertex-at-a-time strategy but is not worst-case optimal.Source: §1.2.3 · p. 2

1.3 Our Approach and Contributions

  • The paper recasts Generic Join as a distributed dataflow primitive, yielding BiGJoin for static queries, Delta-BiGJoin for dynamic queries, and BiGJoin-S for balanced static evaluation.The algorithms and theory also apply to relational equi-joins.Source: §1 · p. 1, §1.3 · p. 2
  • BiGJoin and Delta-BiGJoin monitor complex subgraphs on graphs with up to 64B edges using just over eight bytes per edge.The evaluation used a 16-machine cluster.Source: §1.3 · p. 2
  • BiGJoin-S guarantees worst-case computation and communication costs and workload balance on every query, with per-worker memory as low as O(IN/w).BiGJoin can leave most work to one worker on adversarial inputs; Delta-BiGJoin achieves the same guarantees as BiGJoin for insertion-only workloads.Source: §1.3 · p. 2

2. PRELIMINARIES

The preliminaries define the relational query notation and Generic Join's incremental construction of query results. They then specify the distributed computation model and the Timely Dataflow mechanisms used to coordinate processing and bound memory.

2.1 Notation

  • A query Q joins n relations over m attributes, with input size IN equal to the sum of the relation sizes.The relations may contain different subsets of the attributes.Source: §2.1 · p. 3

2.2 Generic Join

  • Generic Join orders attributes and iteratively extends each prefix tuple by intersecting the next-attribute extensions supplied by the relations.Extension indices provide constant-time size and membership checks and retrieval time linear in the extension-set size; GJ runs in O(mnMaxOutQ) when these properties hold.Source: §2.2 · p. 3

2.3 Massively Parallel Computation Model

  • The MPC model measures rounds, maximum per-worker load per round, and total communication; the paper adds M for memory use.Worker memory excludes output tuples, which are written to storage outside the cluster.Source: §2.3 · p. 3

2.4 Timely Dataflow

  • Timely Dataflow distributes data-parallel operators across workers and uses logical progress to coordinate processing.Operators can delay inputs until others finish, helping synchronize workers and drain downstream queues to bound memory.Source: §2.4 · p. 3

3. ALGORITHMS

The algorithms extend query prefixes through a shared dataflow primitive, adapting it for static joins, incremental maintenance, memory bounds, and skew resilience. BiGJoin-S balances work under arbitrary input skew while retaining the stated cumulative computation, communication, and memory bounds.

3.1 Dataflow Primitive

  • The shared primitive extends Pj tuples to Pj+1; batching Proposal output bounds queued elements by B = wB′, while Count and Intersect can stream outputs.Proposal operators synchronize between batches; the Count and Intersect steps remain unchanged by batching.Source: §3 · p. 3, §3.1.2 · p. 4, §3.1.3 · p. 4

3.2 Joins on Static Relations: BiGJoin

  • BiGJoin assembles batched prefix-extension dataflows for static queries and matches Generic Join’s O(mnMaxOutQ) computation and communication costs.Its cumulative memory is O(mIN + mB), but it cannot guarantee workload balance on adversarial inputs.Source: §3.2 · p. 4

3.3.1 Delta-GJ

  • Delta-GJ computes changes with one independently executed GJ delta query per relation and is worst-case optimal for insertion-only workloads, with total computation O(mn²MaxOutQ).Its performance under workloads mixing insertions and deletions is left for future work.Source: §3.3.1 · p. 5

3.3.2 Delta-BiGJoin

  • Delta-BiGJoin seeds each relation-specific delta dataflow with that relation’s changes and maintains indices incrementally; for insertions, its computation and communication are O(mn²MaxOutQ).Its cumulative memory is O(mnIN(z) + mB).Source: §3.3.2 · p. 5

3.4.1 Skew-resilient Extension Indices

  • BiGJoin-S distributes extension-index contents and divides proposal and intersection ranges among workers to address skew in index sizes and workloads.Its resolver and lookup aggregation also address skewed routing of prefixes and candidate extensions.Source: §3.4 · p. 5, §3.4.1 · p. 5, §3.4.2 · p. 6

3.4.2 Balance and Extension Resolving Operators

  • BiGJoin-S uses O(mnMaxOutQ) cumulative computation and communication and O(mnIN + mB) memory; each worker does O(B′) work per round with probability at least 1−O(1/IN).These guarantees apply for the theorem’s stated sufficiently large, logarithmic-size batches.Source: §3.4.2 · p. 6
  • When B′ < IN/w, BiGJoin-S has optimal MPC load.The load is O(mnIN/w + mB′).Source: §3.4.2 · p. 6

3.4 A Work-balanced Dataflow: BiGJoin-S

  • Delta-BiGJoin-S can be skew-resilient when updates are large enough; a single-tuple update may provide too little work to distribute evenly.Implementing and evaluating the workload-balancing techniques is left for future work.Source: §3.4 · p. 5, §3.4.2 · p. 6

4. IMPLEMENTATION

The implementation uses Timely Dataflow to realize BiGJoin and Delta-BiGJoin, with machinery also suitable for queries beyond graph subqueries. Its prefix-extension stages support static evaluation and timestamped, weighted updates for dynamic evaluation.

4. IMPLEMENTATION

  • The implementation’s underlying machinery also suits more general queries, including a ternary-relation input described elsewhere in the paper.Source: §4 · p. 6

4.1 Prefix Extension in Timely Dataflow

  • Prefix extension starts from prefixes and extends each with relations constraining the next attribute; dynamic prefixes also carry timestamps and signed weights.Operators shuffle prefixes to the worker holding the relevant index; shared partitioning lets relation uses share an index, with forward and reverse edge indexes for graphs.Source: §4.1 · pp. 6–7

4.1.1 Count Minimization

  • Count minimization selects the smallest relevant neighbor-list size and its relation index before forwarding the prefix for candidate proposal.The operator compares outgoing or incoming neighbor counts for the prefix’s current vertex.Source: §4.1.1 · p. 7

4.1.2 Candidate Proposal

  • Candidate proposal sends a prefix to the relation selected by min-i and extends it with that vertex’s outgoing or incoming neighbors.The candidate extensions travel together in one output stream.Source: §4.1.2 · p. 7

4.1.3 Intersection

  • Intersection filters candidate extensions against the appropriate neighbor lists for each involved relation, then forwards valid extensions as prefixes or compact final outputs.Source: §4.1.3 · p. 7

4.2 The BiGJoin Dataflow

  • BiGJoin’s static dataflow seeds length-two prefixes from edges, extends them in global attribute order, and uses pre-built, memory-mapped indices.The evaluated static graphs were small enough to use the whole graph as each worker’s index.Source: §4.2 · p. 7
  • BiGJoin processes prefixes in batches, which controls peak memory but does not guarantee bounded intermediate results during a batch.The implementation did not include downstream-draining back-pressure techniques, which allow proposal stages to wait for downstream dataflow to drain.Source: §4.2 · p. 7

4.3 The Delta-BiGJoin Dataflow

  • Delta-BiGJoin uses separate delta-query dataflows and multi-versioned indices to sequence simultaneous updates correctly while supporting queries against different index versions.Updates carry logical timestamps and signed integer weights to distinguish time and additions from deletions.Source: §4.3 · pp. 7–8

5. EVALUATION

The evaluation compares static-graph baselines, tests update capacity and scaling on large graphs, and examines optimizations and batch-size sensitivity. Common Crawl experiments focus on maintaining query results under updates rather than enumerating all instances.

5.1 Experimental Setup

  • The evaluation spans five subgraph queries and datasets from 68 million-edge LiveJournal to 64 billion-edge Common Crawl; the largest graph is evaluated through incremental maintenance.Because Common Crawl has an estimated 2.36 × 10^16 diamonds, the experiments avoid enumerating all answers.Source: §5.1 · p. 8

5.2 Baseline measurements

  • BiGJoinT reaches COST at 2 cores and Delta-BiGJoinT at 4; EmptyHeaded is faster on one machine, while its index-build time can be non-trivial.COST measures the cores needed to beat an optimized single-threaded implementation.Source: §5.2.1 · p. 8, §5.2.2 · p. 9, Table 2 · p. 9

5.2.3 Arabesque

  • Arabesque considers roughly 30x more prefixes than BiGJoinT, corresponding to 10x–20x higher running times in the reported comparison.Arabesque ran only on LiveJournal among the tested datasets because it ran out of memory on the others.Source: Table 3 · p. 9, §5.2.3 · p. 9

5.3 Capacity and Scaling

  • The computation processes roughly 1M updates per second and reports between 10M and 100M changed triangles per second.For 4-cliques, 200K input changes take 226 seconds at 224 workers, while output changes remain in the tens of millions per second.Source: §5.3 · p. 10
  • As Common Crawl data and worker counts double, maximum index size stays near 8 billion tuples, while more workers generally improve throughput and reduce per-machine memory.The largest out-degree is about 45 million, and the reported balance holds despite such high-degree vertices.Source: Figure 5 · p. 10, §5.3 · p. 10

5.4 Generality and Specializations

  • Symmetry breaking, triangle indexing through a ternary relation, and factorization are integrated; rewriting the 4-clique query with triangles reduces complexity and intermediate prefixes.The paper also demonstrates support for general relations beyond binary edge tables.Source: §5.4 · pp. 10–11
  • With symmetry breaking and triangle indexing, BiGJoin is competitive with or can outperform SEED+O on 4- and 5-clique queries, but timings come from, not identical hardware.The comparison illustrates that BiGJoin can incorporate prior optimizations; it is not presented as a controlled superiority claim.Source: Table 5 · p. 11, §5.4 · p. 11

5.5 Sensitivity to Batch Size

  • As batch size grows, runtime initially improves and then levels off around 10K, while memory usage rises; up to 100K, that increase is small.The experiments use Delta-BiGJoin on the UK graph with 16 workers.Source: §5.5 · p. 11, Figure 6 · p. 11

6. RELATED WORK

Related work covers distributed join strategies, incremental view maintenance, subgraph-query systems, and streaming algorithms. The cited approaches differ in their guarantees, memory requirements, and whether they enumerate or approximate results.

Distributed Multiway Join Algorithms

  • Distributed join alternatives include a two-relation algorithm [26], GYM, and DBP; GYM and DBP can require superlinear cluster memory for cyclic queries.GYM uses O(n) rounds; DBP uses three rounds but can also compute superlinear intermediate relations.Source: §6 · p. 11

Incremental View Maintenance

  • Prior incremental view-maintenance techniques may lack theoretical optimality and materialize large intermediates; LFTJ-Inc has a stronger trace-edit guarantee but sensitivity indices can reach the AGM bound.LFTJ-Inc is based on Leapfrog TrieJoin [59].Source: §6 · p. 12

Systems and Algorithms For One-time Subgraph Queries

  • Many general subgraph systems use edge-at-a-time strategies, while traversal ordering, workload balancing, and symmetry breaking can complement the paper’s algorithms.PSgL is described as edge-at-a-time; RDF engines TrinityRDF and Spartex also use such strategies.Source: §6 · p. 12

Streaming and Semi-streaming Algorithms

  • Streaming and semi-streaming subgraph studies may use slightly superlinear space and often approximate counts, whereas this paper studies enumeration.The cited examples include triangle-finding variants.Source: §6 · p. 12

7. FUTURE WORK

The paper identifies workload balance, query symmetry, and stronger-than-worst-case guarantees as future research directions. It reports that BiGJoin-S overhead exceeded its benefits in a preliminary implementation.

7. FUTURE WORK

  • BiGJoin-S’s preliminary implementation had overheads higher than its benefits, motivating better workload-balanced versions and study of when skew hurts BiGJoin and DeltaBiGJoin.The authors describe understanding skew’s effects and guarding against it effectively as future work.Source: §7 · p. 12
  • Query symmetries can make delta-queries compute identical prefixes, suggesting automatic sharing of those computations across dataflows.For 4-clique, the passage names dQ2 and dQ3 as computing the same P2 and P3 prefixes.Source: §7 · p. 12
  • The authors propose practical algorithms with guarantees stronger than worst-case optimality, such as certificate-complexity optimality for computation.Certificate complexity is described as the smallest proof size needed to verify output correctness and as strictly stronger than worst-case optimality.Source: §7 · p. 12

A. EXAMPLE EXECUTION OF GJ

The example walks through Generic Join (GJ) on a triangle query by extending prefixes one attribute at a time. At each extension, it intersects candidate sets, starting from the smaller set when comparing sizes.

A. EXAMPLE EXECUTION OF GJ

  • GJ begins with the empty prefix and forms P1 by intersecting the candidate extensions for a1.For the example, the intersection produces P1 = {(1), (6), (7)}.Source: A. EXAMPLE EXECUTION OF GJ · p. 14
  • For each P1 prefix, GJ extends it to P2 by intersecting the candidate extensions for a2.For prefix (1), the smaller candidate set is {6}, and the intersection produces (1, 6).Source: A. EXAMPLE EXECUTION OF GJ · p. 14
  • GJ extends each P2 prefix to P3 using candidate extensions for a3, again beginning with the smaller set.The example yields the triangles (1, 6, 7), (6, 7, 1), and (7, 1, 6).Source: A. EXAMPLE EXECUTION OF GJ · p. 14

B. PROOF OF LEMMA 3.1

The proof bounds BiGJoin’s stored prefixes and intermediate data through per-operator invariants, then relates its cumulative computation and communication to Generic Join. It also bounds the rounds needed to extend batches of prefixes.

B. PROOF OF LEMMA 3.1

  • BiGJoin stores all prefix sets using cumulative memory O(mwB′) = O(mB), with at most 2B′ prefixes per Proposal operator for each Pj.The proof also bounds intermediate data from candidate extensions being intersected by O(wB′) per round.Source: B. PROOF OF LEMMA 3.1 · p. 14
  • BiGJoin’s cumulative computation and communication are each O(mnMaxOutQ), matching its stated cumulative computation cost of Generic Join.The analysis attributes this to starting each prefix intersection from the relation with the fewest extensions and sending each candidate extension to another operator.Source: B. PROOF OF LEMMA 3.1 · p. 14
  • Extending a batch of B′ prefixes takes at most 2n + 1 rounds of computation.The proof counts at most n Intersect and n Count operators in each dataflow primitive.Source: B. PROOF OF LEMMA 3.1 · p. 14

C. PROOF OF THEOREM 3.2

The theorem proof bounds the cumulative cost of Delta-GJ’s repeated delta-query evaluations by Generic Join’s cost on the final relations. It accounts for indexing, prefix checks, and subsequent prefix extensions.

C. PROOF OF THEOREM 3.2

  • For each dQi, the cumulative cost across z evaluations is asymptotically at most Generic Join’s cost on the final relations using the same attribute ordering.The proof considers each delta query separately and relies on the final-relation query Q(z).Source: C. PROOF OF THEOREM 3.2 · pp. 14–15
  • Delta-GJ’s cumulative cost for indexing updated tuples matches Generic Join’s indexing cost on the final relations.This comparison applies to indexing each tuple in the update sequence into the Ext indices.Source: C. PROOF OF THEOREM 3.2 · p. 15
  • After computing Pri, Delta-GJ’s cost to extend its tuples through later prefixes is no greater than Generic Join’s cost with the same global attribute ordering.The proof uses that each delta query’s relations are subsets of the corresponding final relations.Source: C. PROOF OF THEOREM 3.2 · p. 15
  • Membership checks add O(mnIN) cumulative cost across relations, which the proof says is subsumed by O(mn2MaxOutQ).The checks determine whether each updated tuple belongs in Pri.Source: C. PROOF OF THEOREM 3.2 · p. 15

D. PROOF OF LEMMA 3.3

Delta-BiGJoin’s computation and communication costs match the stated asymptotic bounds, while its cumulative memory is bounded by the memory of BiGJoin on the full query. Its total round bound includes work per update and the prefix-extension batches.

D. PROOF OF LEMMA 3.3

  • Delta-BiGJoin’s computation and communication cost is O(mn^2 MaxOutQ), and its cumulative memory is O(mnIN(z)+mB).The memory bound follows because its indices and intermediate data are no larger than those BiGJoin uses on Q(z).Source: D. PROOF OF LEMMA 3.3 · p. 15
  • O(mn^2 MaxOutQ/B′ + zmn^2) rounds bound Delta-BiGJoin’s computation across the z updates.The bound combines prefix-extension work with additional O(zmn) rounds per delta query.Source: D. PROOF OF LEMMA 3.3 · p. 15

E. PROOF OF THEOREM 3.4

The proof bounds BiGJoin-S’s per-worker load and total memory, and shows that its cumulative computation and communication costs match BiGJoin asymptotically. It controls skew by aggregating requests and balancing batches of candidate extensions.

E. PROOF OF THEOREM 3.4

  • BiGJoin-S has asymptotically the same cumulative computation and communication costs as BiGJoin.For each prefix, Count does O(n) work and Extension-Resolve and Intersect do O(n × min-c) work, where min-c is the minimum count after count minimization.Source: E. PROOF OF THEOREM 3.4 · p. 15
  • BiGJoin-S maintains at most 4B (p, min-i, start, end) tuples in any Pj.Extending B tuples generates at most B + w^2 ≤ 2B tuples, using B′ ≥ w.Source: E. PROOF OF THEOREM 3.4 · p. 15
  • O(mnIN + mB) bounds BiGJoin-S’s total cumulative memory for indices and prefix tuples.The two terms account for storing the indices and the Pj tuples, respectively.Source: E. PROOF OF THEOREM 3.4 · p. 15
  • Local aggregation and balancing keep each worker’s Extension-Resolve, Intersect, Count, and Balance requests or tuples to O(B′) per batch.The lookup-load argument gives the skewed Extension-Resolve and Intersect bounds with high probability; Count uses the same argument, while Balance uses B′ ≥ w.Source: E. PROOF OF THEOREM 3.4 · p. 15
  • O(IN/w + B′) bounds each worker’s load with probability at least 1−O(1/(IN×MaxOutQ)).The initial input contributes O(IN/w); subsequent batches contribute O(B′) communication and proportional computation.Source: E. PROOF OF THEOREM 3.4 · p. 15

F. COMPARISON WITH SHARES

The comparison finds that BiGJoin-S has lower memory and better load scaling than Hypercube, while Hypercube uses fewer rounds. Hypercube’s worst-case communication optimality applies only to a subclass of queries.

F. COMPARISON WITH SHARES

  • Hypercube is worst-case communication optimal when τ* ≤ ρ*, but not when ρ* < τ* except if one relation contains all query attributes.The triangle query is an example where τ* = ρ* = 3/2.Source: F. COMPARISON WITH SHARES · p. 16
  • At high parallelism on the example query with ρ* = 2 and τ* = 3, Hypercube’s communication is Ω(IN^3), exceeding BiGJoin-S’s O(mnIN^2).The comparison uses load L = IN/w^(1/3) = O(1), which implies w = IN^3.Source: F. COMPARISON WITH SHARES · p. 16
  • Hypercube requires more cluster memory than BiGJoin-S, even on queries where Hypercube is worst-case optimal, when BiGJoin-S’s batch is sufficiently small.The passage compares Hypercube’s Ω(w^(1−1/ρ*)IN) cluster memory with BiGJoin-S’s O(mnIN + mB).Source: F. COMPARISON WITH SHARES · p. 16
  • BiGJoin-S’s load can be O(mnIN/w), whereas Hypercube’s load is O(IN/w^(1/τ*)) or O(IN/w^(1/ρ*)) and does not scale linearly with workers when τ*, ρ* > 1.The BiGJoin-S bound is obtained by choosing a sufficiently small batch size.Source: F. COMPARISON WITH SHARES · p. 16
  • Hypercube runs a constant number of rounds, while BiGJoin-S uses more rounds.The passage states that Hypercube performs better on round count.Source: F. COMPARISON WITH SHARES · p. 16
Loading 1802.03760v1…