Source-linked AI summary
Distributed Datalog at Scale with Worst-Case Optimal Joins
Kunting Qi, Yihao Sun, Kristopher Micinski, Thomas Gilray, Sidharth Kumar
TL;DR
Cyclic multi-way joins create costly intermediates in binary-join systems, while distributed WCOJ had not addressed recursive fixpoint evaluation on shared-nothing HPC systems. This paper presents a distributed-memory Datalog engine that extends HyperCube partitioning to iterative WCOJ execution. Across its evaluated benchmarks, the system demonstrates strong scalability and competitive performance against existing Datalog engines.
Problem
Distributed WCOJ had been studied for one-shot, non-recursive queries, leaving recursive fixpoint evaluation on shared-nothing HPC systems an open challenge.
Source: §I · p. 1, §VI · p. 11Method
The engine extends HyperCube partitioning to recursive WCOJ, reusing the layout and redistributing only newly derived tuples in each iteration.
Source: §VII · p. 11, §VI · p. 11Results
The system demonstrates strong scalability and is faster than established single-node engines on most benchmarks, with reported speedups up to 14.3× from 32 to 2,048 processes.
Source: A. Scalability under Shared-Nothing… · p. 8, §VII · p. 11Takeaways & Limitations
Partition dimensionality trades improved locality against communication complexity and replication, making it a design choice tied to workload and memory feasibility.
Source: C. Impact of Partition Dimensionality · p. 6, §VII · p. 11
Abstract
from arXiv · showhide
Scalable logic-based analytics are frequently dominated by cyclic multi-way joins over large relational datasets. State-of-the-art systems evaluate these joins through chains of binary joins, incurring substantial overhead from large intermediate relations that must be materialized and redistributed across the network. Worst-Case Optimal Joins (WCOJ) offer a provably superior alternative by evaluating joins one variable at a time, eliminating intermediate blowup. However, existing WCOJ implementations target either shared-memory machines or one-shot, non-recursive queries on dataflow and big-data platforms, leaving WCOJ-based recursive evaluation on distributed-memory HPC systems an open problem. We present, to our knowledge, the first distributed-memory Datalog engine that uses WCOJ for recursive fixpoint evaluation. Our design builds on HyperCube partitioning, which hash-partitions relations along variable-aligned grid dimensions and replicates them along missing dimensions, and extends it from a single shuffle to iterative execution: in every semi-naive iteration, only newly derived tuples are repartitioned and re-replicated, using MPI all-to-all collectives and per-dimension all-gathers over subcube communicators, while the same layout is reused across iterations so that all WCOJ intersections remain rank-local. We evaluate our system on the Aurora and Polaris leadership-class supercomputers at Argonne National Laboratory using graph mining and program analysis benchmarks, demonstrating strong scalability, outperforming state-of-the-art Datalog engines on most benchmarks, and characterizing how partition dimensionality trades runtime against memory feasibility.
I. INTRODUCTION
Recursive Datalog workloads require scalable evaluation of cyclic joins, but binary-join intermediates can overwhelm computation and communication. The paper introduces a distributed-memory WCOJ engine that reuses a HyperCube layout for iterative, rank-local evaluation.
I. INTRODUCTION
- The introduction illustrates Datalog analytics with a cyclic triangle query and a recursive Peer Influence query over graph relations.The Peer Influence rule derives influence through a shared neighbor and joins three Edge occurrences with the Influenced relation.Source: §I · p. 1
- Binary joins can create oversized intermediate relations that must be materialized and redistributed, and recursive queries repeat this communication bottleneck.The central challenges are coordinating partitions across all WCOJ intersection variables and redistributing newly derived tuples between fixpoint iterations.Source: §I · pp. 1–2
- WCOJ intersects relations across variables and prunes candidates before materialization, avoiding the large intermediates produced by chained binary joins.The paper describes this approach as keeping intermediates within the worst-case optimal bound established by the AGM inequality.Source: §I · p. 2
- HyperCube partitioning maps selected join variables to process-grid dimensions, partitioning relations on present attributes and replicating them along missing dimensions.Increasing dimensionality can reduce replication for relations carrying the selected variables, while increasing communication; the choice exposes a runtime and memory trade-off.Source: §I · p. 2
- Up to 14.3× speedup on Polaris and 13.4× on Aurora was achieved when scaling from 32 to 2,048 processes.The system was faster than established single-node engines on most recursive and cyclic query benchmarks and outperformed existing distributed Datalog engines on transitive closure.Source: §I · p. 2
- The paper presents, to its authors’ knowledge, the first distributed-memory Datalog engine using WCOJ for recursive semi-naïve fixpoint evaluation.Each iteration repartitions and re-replicates only newly derived tuples using MPI collectives, while reusing the same layout so WCOJ intersections remain rank-local.Source: §I · p. 2
II. BACKGROUND
Datalog expresses relational rules whose application can recursively derive facts until evaluation reaches a fixpoint. Semi-naïve evaluation tracks newly derived tuples to avoid redundant derivation.
II. BACKGROUND
- Datalog rules use a head atom for a derived fact and body atoms for the conditions under which it holds.Shared variables across body atoms impose equality constraints corresponding to relational equi-joins.Source: §II · p. 3
- Evaluation iteratively applies rules until no new intensional facts can be derived, reaching a fixpoint.The input extensional database contains base facts, while the intensional database contains derived facts.Source: §II · p. 3
- Transitive closure is expressed with rules for direct edges and for extending already-known paths by another edge.The paper uses this recursive query as an example of deriving pairs connected by a directed path.Source: §II · p. 3
- Semi-naïve evaluation maintains a delta relation of tuples from the previous iteration and restricts the next iteration to joining that delta against existing facts.The paper identifies this incremental strategy as the basis of its evaluation approach.Source: §II · p. 3
A. Worst-Case Optimal Joins (WCOJ)
Chained binary joins can materialize many paths that do not contribute to a cyclic query’s output. WCOJ instead intersects relations variable by variable, pruning nonviable candidates before materializing them.
A. Worst-Case Optimal Joins (WCOJ)
- Binary plans scan one relation and probe others, whereas WCOJ cross-references relations dynamically as it extends variable bindings.Figure 1 contrasts binary intermediates with WCOJ’s pruning for a triangle query.Source: A. Worst-Case Optimal Joins (WCOJ) · p. 3, Figure 1 · p. 3, Figure 2
- For a triangle query, binary intermediates can reach O(|E|^2), while the output is bounded by O(|E|^3/2).In the dense-star example, 15 of 16 materialized paths are discarded when the closing edge is checked.Source: A. Worst-Case Optimal Joins (WCOJ) · p. 3
- WCOJ evaluates a triangle by intersecting relations one variable at a time, pruning candidates that cannot contribute to an output before materialization.The triangle plan intersects x, then y, then z across the relevant relations.Source: A. Worst-Case Optimal Joins (WCOJ) · pp. 3–4
- Prior distributed WCOJ systems addressed one-shot, non-recursive queries, while recursive WCOJ on shared-nothing HPC clusters remained an open challenge.Naively distributed attribute-at-a-time evaluation can trigger many synchronous remote probes and a network-latency bottleneck.Source: A. Worst-Case Optimal Joins (WCOJ) · p. 4
B. Distributed Datalog Processing
Distributed Datalog systems can process data beyond single-machine memory, but binary-join intermediates remain costly for cyclic workloads. Distributed WCOJ requires a partitioning scheme that supports intersections across variables and can be reapplied efficiently during recursion.
B. Distributed Datalog Processing
- Existing distributed Datalog engines include Spark-based BigDatalog and RaSQL, Distributed SociaLite, and MPI-based Slog, which still relies on materialized binary joins.The cited engines have limitations on scaling efficiency or complex and cyclic queries.Source: B. Distributed Datalog Processing · p. 4
III. WORKLOAD PARTITIONING FOR DISTRIBUTED WORST-CASE OPTIMAL JOINS
- In shared-nothing execution, oversized binary-join intermediates must be redistributed across processes, increasing computation and communication overhead.Distributed WCOJ had been studied for one-shot, non-recursive queries, leaving partitioning for iterative queries as a central challenge.Source: §III · p. 4
- The proposed approach builds on HyperCube partitioning, mapping join variables to process-grid dimensions and replicating relations along dimensions their attributes do not cover.MPI collectives over subcube communicators implement the layout, which is intended to support data locality while minimizing communication.Source: §III · p. 4
A. Hypercube Partitioning and Replication for Distributed WCOJ
WCOJ partitions by shared variables rather than making independent relation-pair decisions, so a hypercube layout aligns process-grid dimensions with selected join variables. Tuples are partitioned on dimensions present in their relation and replicated along missing dimensions, allowing WCOJ intersections to run locally.
A. Hypercube Partitioning and Replication for Distributed WCOJ
- WCOJ uses simultaneous intersections across shared variables, making partitioning by individual relation pairs unsuitable for multi-way joins.A hypercube layout uses selected join variables as grid dimensions, with relations partitioned or replicated according to the variables they contain.Source: A. Hypercube Partitioning and… · p. 4
- In a 2D layout, relations are partitioned on keys they contain and replicated across ranks along missing dimensions.Figure 3 contrasts this with a 1D layout, where relations lacking the single partition key are replicated.Source: Figure 3 · p. 5, A. Hypercube Partitioning and… · p. 5
- Tuples fully covering the selected keys are routed to one rank; tuples missing keys are replicated to ranks consistent with their known coordinates.The implementation uses global MPI_Alltoall and MPI_Alltoallv redistribution, then selective MPI_Allgather and MPI_Allgatherv within subcube communicators for missing dimensions.Source: A. Hypercube Partitioning and… · p. 5
- This placement keeps the relation fragments needed for WCOJ available locally while avoiding fine-grained remote lookups during evaluation.The algorithm materializes local fragments, builds local indexes, and then runs local WCOJ.Source: A. Hypercube Partitioning and… · pp. 5–6
B. Cost Model for Replication under Hypercube Partitioning
The cost model measures how hypercube partitioning replicates each relation according to which selected partition keys its schema covers. It then aggregates those per-relation costs to expose how partition-key choice affects total replicated data.
B. Cost Model for Replication under Hypercube Partitioning
- Relations covering all selected partition keys are fully partitioned without replication; otherwise, they are replicated along each missing dimension.For relation Ri, the missing dimensions are D \ keys(Ri).Source: B. Cost Model for Replication under… · p. 6
- The replication factor γ_i counts the hypercube cells containing each original tuple of relation Ri.If the product over missing dimensions is empty, γ_i = 1.Source: B. Cost Model for Replication under… · p. 6
- The model totals replicated relation sizes, showing that relations covering more partition keys incur less replication and those missing more dimensions incur more.Replication can grow especially when the partition counts for missing dimensions are large; partition-key choice determines the query's total replication cost.Source: B. Cost Model for Replication under… · p. 6
C. Impact of Partition Dimensionality
Partition dimensionality affects both local data placement and the replication needed for complete joinability. Its impact depends on how selected keys align with relation schemas, so no dimensionality is best independently of the query.
C. Impact of Partition Dimensionality
- The effect of partition dimensionality depends on alignment between selected keys and relation schemas, not dimensionality alone.More dimensions can improve locality while increasing communication complexity and replication for dimensions a relation lacks.Source: C. Impact of Partition Dimensionality · p. 6
- With 1D partitioning on x, T(z, y) lacks the selected key and is replicated across all 64 partitions.Influenced(x), R(x, y), and S(x, z) are fully partitioned on x.Source: C. Impact of Partition Dimensionality · p. 7
- With 2D partitioning on x and z, S(x, z) is fully partitioned while Influenced(x), R(x, y), and T(z, y) are replicated along their missing dimension.Compared with 1D, replication is distributed more evenly across partially covered relations.Source: C. Impact of Partition Dimensionality · p. 7
- With 3D partitioning on x, y, and z, each binary edge atom is replicated along one dimension, while unary Influenced(x) is replicated along two.Across the layouts, 1D burdens one uncovered relation, 2D spreads replication more evenly, and 3D shifts more replication onto the unary relation.Source: C. Impact of Partition Dimensionality · p. 7
IV. ITERATIVE WCOJ IN PARALLEL
The system extends distributed WCOJ to recursive queries by evaluating each semi-naïve delta against local relation state and redistributing only newly derived tuples. Reusing the same hypercube layout keeps WCOJ intersections rank-local across iterations until the global delta is empty.
IV. ITERATIVE WCOJ IN PARALLEL
- Recursive Peer Influence requires repeated distributed WCOJ rounds because newly influenced vertices can trigger further derivations.Unlike one-shot joins, each iteration's output becomes the next input delta until the global fixpoint is reached.Source: §IV · p. 7
- New tuples are redistributed with MPI_Alltoall(v), replicated along missing dimensions with MPI_Allgather(v) when needed, then deduplicated to form the next delta.The iterative procedure reapplies the hypercube layout to each iteration's newly derived tuples.Source: Figure 4 · p. 7, §IV · p. 8
- Each iteration evaluates recursive rules with the current delta and local full relation state, then retains genuinely new tuples as the next delta.A global reduction detects an empty delta and terminates evaluation at the fixpoint.Source: §IV · p. 7
- The same partition-and-replication layout is reused so WCOJ's prefix-driven intersection work remains rank-local across iterations.Communication is deferred until after local evaluation, avoiding fine-grained remote lookups during the intersection loop.Source: §IV · p. 7
V. EVALUATION
The evaluation tests the system on two leadership-class HPC machines using graph and program-analysis queries. It compares end-to-end runtimes against single-node and multi-node systems.
V. EVALUATION
- Experiments run on Polaris and Aurora, with 32 MPI ranks per Polaris node and at most 64 per Aurora node.Polaris nodes have 32-core CPUs and 512 GB memory; Aurora nodes have two 52-core CPUs, 1 TB memory, and 128 GB CPU-attached HBM.Source: §V · p. 8
- The benchmark suite covers five queries spanning recursive and non-recursive joins, graph analytics, and program analysis, using four graphs and two synthetic input programs.The queries are PEER INFLUENCE, TRIANGLE, transitive closure, same generation, and Andersen-style context-sensitive points-to analysis; points-to analysis uses MEDIUM and LARGE programs.Source: §V · p. 8
- End-to-end runtime includes input reading, redistribution, replication, index construction, iterative evaluation, and result materialization; distributed results report the slowest rank.The comparisons include Soufflé, FlowLog, and DuckDB for single-node systems, and Slog and Multi-Node Multi-GPU Datalog for multi-node systems.Source: §V · p. 8
A. Scalability under Shared-Nothing Execution
The system achieves strong scaling across workloads on Polaris and Aurora, but scaling benefits vary by workload and partitioning layout. Two-dimensional partitioning is faster when feasible, while three-dimensional partitioning can remain feasible at smaller process counts; communication becomes a larger runtime share at scale.
A. Scalability under Shared-Nothing Execution
- Most benchmarks scale well through 256–512 processes, with diminishing gains afterward; scaling regularity varies by workload on both systems.Andersen scales most smoothly, while PEER INFLUENCE is less regular; TRIANGLE and SG keep improving with diminishing returns, and TC runtime increases beyond 512 processes.Source: A. Scalability under Shared-Nothing… · pp. 8–9, Table II · p. 9, Table III · p. 9
- 21.2% to 52.2% of Andersen (Large) runtime is communication on Polaris as process count rises from 32 to 2,048; Aurora rises from 16.7% to 58.6%.Join and indexing account for a large runtime fraction at small process counts, while communication takes an increasing share at larger counts.Source: A. Scalability under Shared-Nothing… · p. 9, Figure 5 · p. 10
- 2D is faster when both layouts are feasible, while 3D remains feasible for some Polaris cases that exceed 2D memory limits.On Polaris, 2D PEER INFLUENCE (Orkut) falls from 30.765 s at 64 processes to 7.955 s at 2,048, versus 68.945 s to 23.317 s in 3D; 2D runs out of memory at 32 processes, while 3D remains feasible.Source: A. Scalability under Shared-Nothing… · pp. 9–10, Table II · p. 9, Table III · p. 9
- 3D has a larger communication share than 2D for PEER INFLUENCE (Orkut), while 2D has lower runtime when memory permits.On Polaris, communication share rises from 26.5% to 63.9% in 2D, versus 35.7% to 86.8% in 3D; on Aurora, it rises from 24.4% to 78.1% in 2D, versus 43.9% to 88.8% in 3D.Source: A. Scalability under Shared-Nothing… · pp. 9–10, Table IV · p. 9
B. Comparison with Existing Systems
The system is competitive on one node and compares favorably with distributed systems on transitive closure. Prior work covers one-shot WCOJ, GPU Datalog, incremental processing, and robust join techniques, while recursive distributed-memory WCOJ remains uncombined.
B. Comparison with Existing Systems
- On one Polaris node, the system is competitive and often faster on recursive and multi-way workloads, while FlowLog performs particularly well on Andersen and SG (G10).DuckDB is the closest competitor on the two TRIANGLE workloads, but times out on PEER (Orkut) and returns incorrect results on mutually recursive Andersen analyses.Source: B. Comparison with Existing Systems · p. 10, Table V · p. 10
- Across process counts, the system outperforms Slog by 8.9–11.2× and the multi-node multi-GPU engine by 1.5–3.7× on TC (USRoad).Table VI reports the distributed TC (USRoad) runtime comparison on Polaris; the stated speedups are from the prose.Source: B. Comparison with Existing Systems · p. 10, Table VI · p. 11
VI. RELATED WORK
- Prior distributed WCOJ systems target one-shot queries; this work combines HyperCube partitioning with iterative delta redistribution for recursive fixpoint evaluation.Myria supports recursive Datalog with binary joins, while prior WCOJ systems target one-shot conjunctive or subgraph queries.Source: §VI · pp. 10–11
- GPU Datalog work includes WCOJ on a single GPU, while this system uses no accelerators and describes distributed partitioning as orthogonal to local join acceleration.Combining the framework with multi-node GPU Datalog is proposed as future work.Source: §VI · p. 11
- Unlike incremental Datalog engines that track insertions and deletions, this work targets distributed batch processing; streaming support remains future work.The passage identifies batch processing as the dominant execution model for established Datalog applications such as program analysis.Source: §VI · p. 11
- Robust join techniques mainly target acyclic queries, whereas the WCOJ approach also handles cyclic queries prevalent in Datalog workloads.The cited robust techniques use early pruning, backjumping, Bloom filters, or predicate transfer.Source: §VI · p. 11
VII. CONCLUSION AND FUTURE WORK
The conclusion reports strong scaling and favorable benchmark comparisons, with partition dimensionality trading runtime against memory feasibility. At high process counts, communication is a dominant cost and motivates optimizing non-uniform all-to-all exchanges.
VII. CONCLUSION AND FUTURE WORK
- Up to 14.3× speedup from 32 to 2,048 processes; the system is faster than established single-node engines on most benchmarks and outperforms distributed Datalog engines on transitive closure.These results come from graph pattern mining and program analysis benchmarks on Polaris and Aurora.Source: §VII · p. 11
- 2D partitioning minimizes runtime, while 3D partitioning remains feasible where 2D runs out of memory.The conclusion frames partition dimensionality as a runtime–memory trade-off.Source: §VII · p. 11
- At 2,048 processes, communication accounts for 52–59% of Andersen (Large) runtime and 87–89% of PEER INFLUENCE (Orkut) runtime under 3D partitioning.The implementation uses MPI_Alltoallv for non-uniform exchanges, and delta sizes change the communication pattern across iterations.Source: §VII · p. 11
- The authors identify optimized non-uniform all-to-all algorithms, machine-learning-driven tuning, cost-based partition choices, skew-aware replication, and GPU-accelerated local joins as future directions.The best exchange algorithm and parameters depend on message-size distribution, process count, and machine.Source: §VII · p. 11