PaperlayerView exact revision on arXiv

Source-linked AI summary

Massively Parallel Sort-Merge Joins in Main Memory Multi-Core Database Systems

Martina-Cezara Albutiu, Alfons Kemper, Thomas Neumann

arXiv:1207.0145v1cs.DB

TL;DR

The paper addresses scalable join processing for large main-memory, multi-core, NUMA systems, where sort-merge joins had received less attention than hash joins. It develops MPSM algorithms using local partial sorting and parallel run processing instead of a final global merge. MPSM scales almost linearly with cores and achieves a factor-of-four advantage over Vectorwise in the reported evaluation.

  • Problem

    Main-memory databases need join algorithms suited to growing RAM, massively parallel cores, and NUMA, while sort-merge joins were considered inferior to hash joins.

  • Method

    MPSM uses partial partition-based sorting, processes independently created sorted runs in parallel, and performs sorting on local NUMA memory partitions without a final merge.

  • Results

    MPSM scales almost linearly with employed cores and outperforms competing hash joins and Vectorwise, achieving a factor of four over Vectorwise.

  • Takeaways & Limitations

    Sort-based parallel joins can provide competitive performance for very large main-memory databases on modern multi-core systems.

  • Takeaways & Limitations

    The paper focuses on the response-time-optimal range-partitioned in-core MPSM variant; disk-based D-MPSM is only sketched for future analysis.

Abstract

from arXiv · show

Two emerging hardware trends will dominate the database system technology in the near future: increasing main memory capacities of several TB per server and massively parallel multi-core processing. Many algorithmic and control techniques in current database technology were devised for disk-based systems where I/O dominated the performance. In this work we take a new look at the well-known sort-merge join which, so far, has not been in the focus of research in scalable massively parallel multi-core data processing as it was deemed inferior to hash joins. We devise a suite of new massively parallel sort-merge (MPSM) join algorithms that are based on partial partition-based sorting. Contrary to classical sort-merge joins, our MPSM algorithms do not rely on a hard to parallelize final merge step to create one complete sort order. Rather they work on the independently created runs in parallel. This way our MPSM algorithms are NUMA-affine as all the sorting is carried out on local memory partitions. An extensive experimental evaluation on a modern 32-core machine with one TB of main memory proves the competitive performance of MPSM on large main memory databases with billions of objects. It scales (almost) linearly in the number of employed cores and clearly outperforms competing hash join proposals - in particular it outperforms the "cutting-edge" Vectorwise parallel query engine by a factor of four.

1. INTRODUCTION

Main-memory database systems must exploit growing RAM capacities, multi-core parallelism, and NUMA locality rather than relying on disk-oriented assumptions. MPSM addresses these conditions with synchronization-avoiding, NUMA-friendly processing that scales across cores and outperforms competing joins.

  • Hardware trends: Main-memory database systems increasingly target servers with several TB of RAM and highly parallel multi-core processing.These systems are no longer I/O bound, making intra-query and intra-operator parallelism important.
  • NUMA locality: NUMA-friendly processing places data so threads work mostly on local memory, which is faster than accessing remote memory.The system must account for both multi-core parallelization and RAM and cache hierarchies.
  • NUMA locality: A 3× performance penalty occurs when sorting ignores NUMA boundaries on a 1 TB, 32-core machine processing 1600M tuples.The NUMA-affine benchmark sorts each 50M-tuple chunk in its local NUMA partition, whereas the unfavorable case uses a globally allocated array.
  • NUMA locality: Sequential scans of remote memory are acceptable because hardware prefetching mitigates the cost of sequential access.This observation motivates avoiding random remote writes while permitting sequential remote reads.
  • MPSM approach: MPSM avoids a hard-to-parallelize final merge by processing independently sorted runs in parallel on local memory partitions.Its design follows rules that avoid random remote writes, rely on sequential remote reads, and avoid fine-grained synchronization.
  • Evaluation: MPSM scales almost linearly with core count, outperforms Wisconsin hash join by up to an order of magnitude, and achieves a factor-of-four advantage over Vectorwise.The evaluation covers very large main-memory databases with hundreds of GB of data and up to 32 cores.

2. THE BASIC IDEA OF MPSM

MPSM avoids a globally merged sort order by sorting chunks locally and joining the resulting runs in parallel, targeting NUMA-aware scalability. Its basic form trades extra sequential scanning for locality, minimal synchronization, and skew resistance.

  • MPSM design: MPSM sorts each data chunk locally, then joins independently created runs in parallel instead of performing a hard-to-parallelize final merge.The output is partially sorted rather than globally sorted, which can still support operations such as early aggregation.
  • MPSM design: Unlike Wisconsin hash join and radix join, MPSM avoids shared data structures and cross-NUMA writes during local run generation.Sequential scans during joining allow the hardware prefetcher to hide much of the remote-access overhead.
  • B-MPSM algorithm: B-MPSM divides both inputs into equal chunks, sorts each worker’s chunks into runs, and assigns each worker one private chunk while scanning the complete public input.Run generation is performed independently in local memory without worker synchronization.
  • B-MPSM algorithm: B-MPSM is absolutely skew resistant and needs only one synchronization point, to ensure that public-input runs are ready before joining begins.Private-input sorting may continue while other workers start their join phase.
  • B-MPSM algorithm: Each worker sorts chunks of both relations but processes all sorted public-input runs, so the join phase scans the complete public input regardless of thread count.P-MPSM adds range partitioning of private input to reduce this repeated join-phase work for in-core processing.

2. IntroSort (Introspection Sort) [20]

The sorting routine uses introspective sorting: quicksort is bounded by a recursion limit, heapsort handles deeper cases, and insertion sort finishes tiny partitions. It is reported as faster than STL sort in parallel local-run generation.

  • IntroSort (Introspection Sort): IntroSort limits Quicksort to 2 · log(N) recursion levels, switches to heapsort if necessary, and leaves partitions below 16 elements for insertion sort.This combines the usual fast path of Quicksort with safeguards against excessive recursion and a specialized finish for small partitions.
  • IntroSort (Introspection Sort): 30% faster than STL sort, the routine retained its advantage when up to 32 workers sorted local runs in parallel.Each worker sorts a separate data chunk, avoiding synchronization-heavy parallel sorting.

A DISK-BASED VARIANT

D-MPSM adapts MPSM for RAM-constrained processing by spooling sorted runs to disk and synchronously scanning them through the key domain. P-MPSM instead range-partitions private input to reduce join work and achieve near-linear parallel scalability.

  • D-MPSM: D-MPSM spools sorted runs to disk and processes only active run pages in RAM during the join.Workers progress synchronously through the sorted key domain, while processed pages are released and upcoming pages are prefetched.
  • D-MPSM: A page index ordered by key values coordinates processing and identifies pages that can be released or prefetched.The index stores each page’s first join key value and its run identifier.
  • D-MPSM: D-MPSM performance is bounded by the time required to write and read both inputs, making sufficient I/O bandwidth necessary for many cores.The paper specifically identifies a very large number of disks as necessary to exploit multicore processing in this mode.
  • P-MPSM: P-MPSM range-partitions private input so each worker handles only a fraction of the join-key domain.The approach extends B-MPSM with a prologue that assigns range-partitioned private data to workers and avoids much of the join-phase work.
  • P-MPSM: For parallelism T ≥ 2 and |R| ≤ |S|, range partitioning pays off, and P-MPSM scales almost linearly with T.The private input is preferably the smaller relation, while the public input is the larger relation.
  • P-MPSM: Private-input redistribution uses radix clustering, histograms, prefix sums, and sequential writes to precomputed worker-specific regions.This design is branch-free, comparison-free, and synchronization-free during scattering.

4. SKEW RESILIENCE OF P-MPSM

Skew-resilient P-MPSM dynamically derives partition bounds from the key distributions of both inputs. It combines CDF and fine-grained histogram information to balance worker costs rather than merely partition sizes.

  • 4. SKEW RESILIENCE OF P-MPSM: Skew resilience comes from dynamically computing partition bounds from information about the key distributions in R and S.This replaces statically determined bounds and is presented as a low-overhead refinement of P-MPSM.
  • 4.1 Global S Distribution: Sorted public-input runs provide equi-height histograms that are merged into a global CDF at almost no construction cost.The CDF represents skewed distributions in S and can support configurable worker counts through interpolation.
  • 4.2 Global R Distribution Histogram: Fine-grained radix histograms on private chunks use B leading bits to obtain 2^B bins and more precise R distribution information.The resulting clusters can be merged into T partitions with balanced workloads and little histogramming overhead.
  • 4.2 Global R Distribution Histogram: In the illustrated skewed domain [0,32), seven values fall below 8, three fall in [8,16), three in [16,24), and one is at least 24.The example then combines the first cluster with the remaining three clusters under a correlated R–S distribution assumption.
  • 4.3 Partitioning the Private Input R: Global splitters are chosen by combining R histograms with the S CDF and minimizing the largest estimated worker cost.The cost approximation includes sorting R partitions, processing private runs, and processing relevant S data.
  • 4.3 Partitioning the Private Input R: The resulting R partitions need not have equal sizes because balancing processing cost is more important than balancing tuple counts.S is partitioned implicitly by its sorting order, avoiding separate partitioning overhead for the public input.

5. EXPERIMENTAL EVALUATION

The evaluation tests MPSM on very large in-memory joins across NUMA-aware hardware, varying data sizes, multiplicities, core counts, roles, and skew. MPSM scales nearly linearly and outperforms Vectorwise and Wisconsin hash join in the reported experiments.

  • Experimental setup: The experiments use a one-TB, 32-core server and in-memory datasets with 1.6 billion tuples in R and S sized from 1× to 16× |R|.The evaluated query is an equi-join that feeds payloads through processing while producing one output tuple.
  • Uniform data comparison: MPSM outperforms Vectorwise by a factor of four on uniform data across tested multiplicities.Wisconsin performs poorly at these data volumes and core counts because its global hash table crosses NUMA partitions.
  • Core scalability: MPSM scales almost linearly with worker-thread count through 32 physical cores, while hyperthreading beyond 32 cores provides no further improvement.At parallelism level 64, performance remains stable because all physical cores are already fully utilized.
  • Role reversal: Role reversal improves execution time as S becomes larger than R by assigning the smaller relation to the private role.The improvement affects range partitioning and joining, while the summed sorting costs remain unchanged.
  • Skew resilience: Location skew in S can reduce complexity because each private partition may find its join partners in only one public run.With less pronounced location skew, performance falls between the two extreme cases; other experiments did not exploit location skew.
  • Skew resilience: Dynamic splitter computation balances sort-and-join work under negatively correlated skew by combining each R partition with its corresponding S join range.Equal-cardinality R partitions leave workers handling low join keys with substantially more join work.
  • Evaluation summary: MPSM’s reported advantage is conservative because the experiments did not exploit pre-existing sort order or the quasi-sortedness of its results.Both properties would favor MPSM in complete query execution plans according to the evaluation summary.

6. RELATED WORK

Prior parallel join work largely emphasized hash-based partitioning and hash joins, including NUMA-aware and GPU-oriented designs. The paper positions MPSM as a parallel sort-merge alternative that avoids global merging and synchronization-heavy or NUMA-costly steps.

  • Parallel join processing: Parallel join processing originated with hash-based partitioning for database machines and later extended to distributed and multi-core systems.The shared-memory setting makes data distribution more efficient but still requires attention to memory locality.
  • Hash joins: The Wisconsin hash join builds a global shared hash table across NUMA partitions, requiring latches during construction and remote random reads during probing.These behaviors violate the paper’s stated scalability principles for synchronization and locality.
  • GPU and related techniques: GPU join research combines nested-loop, sort-merge, and hash joins with massive threading, local memory communication, and histogram-based radix partitioning.MPSM adapts the histogram approach for synchronization-free partitioning of its private input and uses a Radix/IntroSort implementation.
  • Output properties: MPSM produces sorted runs within each worker partition rather than completely sorted output, a physical property the authors identify as potentially useful for later operations.The comparative experiments did not exploit this property or possible pre-existing sorting.

7. CONCLUSIONS AND FUTURE WORK

MPSM is a massively parallel equi-join approach designed for large main-memory systems, with scalability grounded in NUMA-aware execution. The paper also identifies extensions to other join variants and disk-based processing as future work.

  • Conclusions: MPSM joins independently sorted runs in parallel, avoiding fine-grained synchronization and random access to remote NUMA partitions.Its scalability is attributed to exploiting NUMA characteristics on modern high-capacity servers.
  • Conclusions: MPSM effectively joins main-memory data with billions of tuples while scaling almost linearly with the number of cores.The reported evaluation used a 32-core, 1TB server, with the conclusion projecting scalability beyond that system.
  • Conclusions: MPSM’s linear core scalability is presented as promising for future servers with several TB of memory and hundreds of cores.These hardware trends are identified as important for main-memory databases and operational or real-time business intelligence.
  • Future Work: Future work will develop MPSM algorithmic details for outer, semi, and nonequi joins.The current paper concentrates on the response-time-optimal range-partitioned in-core variant.
  • Future Work: A follow-up paper will analyze memory-constrained disk-based D-MPSM processing, which is only sketched here.The authors describe this variant as promising for large batch queries running alongside transactions and real-time BI analytics.
Loading 1207.0145v1…