Distributed datalog engine speeds large data joins with scalable design
Distributed Datalog at Scale with Worst-Case Optimal Joins
Databases
Summary
Many computer tasks need to combine information from large tables, but doing this can slow down if the method creates huge in-between steps. The authors show a way to do these joins more efficiently by handling one piece at a time and sharing work smartly across many computers. They built a system that applies this method repeatedly for logic programs, making it faster on big supercomputers. Their tests show it can handle complex tasks better than current tools while balancing speed and memory use.
What this means in practice
- •For database engineers: Run complex recursive queries efficiently on distributed clusters to analyze large relational datasets without intermediate data overload.
- •For supercomputing programmers: Implement large-scale logic-based data analysis on HPC systems using a distributed join approach that scales with compute nodes.
Authors
Kunting Qi, Yihao Sun, Kristopher Micinski, Thomas Gilray, Sidharth Kumar
Abstract
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.