GAMMA - A High Performance Dataflow Database Machine

In one breath — the whole paper, compressed

Gamma is a fully operational relational database machine built from 20 VAX 11/750 processors joined by an 80 megabit/second token ring, with a 160 megabyte disk attached to eight of them. Every relation is horizontally partitioned across all disks using one of four declustering strategies, so scanning is done in parallel by the processor that owns each fragment and the interconnect never has to carry raw disk bandwidth. Query trees are compiled into operator processes that read a stream of tuples and push results through a split table, a small map from hashed or ranged attribute values to destination processes; apart from two control messages to start an operator and one to report its completion, execution is entirely self-scheduling. Joins are hash-partitioned into a building phase and a probing phase, with bit vector filters shipped back to the producers of the inner relation and a subpartitioning scheme for hash table overflow. The measured result is near-linear speedup for both selections and joins from one to eight disks, single-processor times competitive with the commercial IDM500, and one honest negative result about network interface congestion.

Before this paper — the world it landed in

By the mid-1980s the database machine field had produced only a handful of research prototypes and three commercial products, and none had shown that a highly parallel relational machine could actually be built; the most commercially successful of them, the Britton-Lee IDM500, exploited no parallelism at all. The authors' own earlier prototype, DIRECT, had been evaluated and found badly flawed: it treated parallelism as a substitute for indexing, so it burned scarce I/O bandwidth scanning data an index would have skipped. Worse, the number of control messages DIRECT needed for a parallel join grew with the product of the sizes of the two input relations, so message handling swamped the actual processing. Meanwhile the hardware trend was brutal: single-chip CPU performance had improved by at least two orders of magnitude over the decade while commercially available disk bandwidth improved by roughly a factor of three, which made I/O the binding constraint on every existing machine design. The fashionable fix, ganging many small drives into a high-bandwidth mass storage subsystem, only moved the problem, because the data then had to cross an interconnection network of equal bandwidth even though the 90-10 rule says most of it is not needed.

The problem — what was actually breaking

Processor speed and disk bandwidth had diverged sharply over the previous decade, roughly two orders of magnitude of CPU improvement against a factor of three in I/O bandwidth, rendering a number of database machine designs useless.
DIRECT used parallelism as a substitute for indexing, and since an index is precisely a mechanism to avoid searching a large piece of the database, that approach led to disastrous performance once I/O bandwidth became the critical resource.
The control overhead of DIRECT's parallel join algorithms was proportional to the product of the sizes of the two input relations, so even with message passing through shared memory the time spent passing and handling messages dominated processing and I/O time.
Building a mass storage subsystem with, say, 100 megabytes per second of effective bandwidth does not help unless the interconnection network in front of it also carries 100 megabytes per second, and most of the data moved is not needed in the first place.
No prototype or product had demonstrated a highly parallel relational database machine that could be measured: Teradata published no performance data and refused benchmarking requests, DELTA published only sort engine numbers slower than a commercial package on a super-minicomputer, and MBDS reported nothing for complex operations.
Existing partitioned file systems such as VSAM and the Tandem file system required that a file partitioned on a key also be kept sorted on that key at every site, which forces the physical clustering choice to follow the declustering choice.

Core ideas — the contributions, and why they work

One processor per disk drive

Rather than build an exotic parallel read-out storage subsystem, Gamma pairs a conventional disk drive with an ordinary processor and connects the pairs by a network. Fifty drives at two megabytes per second give the same aggregate bandwidth as a monolithic 100 megabyte per second subsystem, but the network no longer has to carry that bandwidth, because selection and other reductive work happens at the drive before anything is transmitted. The design also lets I/O bandwidth grow incrementally, one processor-disk pair at a time, and makes it easy to absorb improvements in disk technology. This is the architecture later given the name shared-nothing, and Gamma is its first measured implementation.

Horizontal partitioning of every relation

All relations in Gamma are horizontally partitioned across all disk drives, so there is no such thing as an unpartitioned table and every scan is automatically a parallel scan. The query language offers four declustering strategies: round robin, hashed, range partitioned with user-specified key ranges per site, and range partitioned with uniform distribution computed by a parallel merge sort. Critically, and unlike VSAM or the Tandem file system, Gamma imposes no relationship between the partitioning attribute of a file and the order of tuples at a site, so a banking relation can be declustered on account number for throughput while carrying a clustered index on branch number for the aggregate queries. Separating declustering from local clustering is what lets physical design serve two access patterns at once.

The split table as the parallelism primitive

Every operator is written as if it ran on a single processor: it consumes a stream of tuples and emits a stream of tuples. Parallelism is injected entirely by a small data structure at the output, the split table, which maps a value derived from each output tuple to the address of a destination process. Because the operator code knows nothing about the split table's contents, the same sequential join or select can be run at any degree of parallelism simply by installing a different table, and repartitioning between operators costs nothing beyond routing. This factoring of parallelism out of operator code and into a routing table is the paper's most reusable idea.

Self-scheduling dataflow execution

After a scheduler activates an operator process, the process identifies itself and then runs without further supervision, reading its input stream, applying its function, and routing results through the split table until it detects end of stream. Closing its output streams sends end-of-stream messages to the downstream processes, and a single control message reports completion to the scheduler. That is three control messages per operator per processor, two to initiate and one to terminate, independent of how many tuples flow, which is the direct answer to DIRECT's message cost that grew with the product of the relation sizes. Everything else is data flowing between processes with no centralized control.

Partitioned hash join in two phases

Applying identical hash-based split tables to both source relations sends every tuple with a given join attribute value to the same site, so a join of two large relations decomposes into independent joins of many smaller buckets. The join operator runs a building phase that consumes the first relation into an in-memory hash table, reports completion, waits for the scheduler to confirm that every site has finished building, then runs a probing phase over the second relation. The scheduler barrier between the phases is the only extra control interaction, bringing the total cost of running a hash join to five messages per site. If the outer relation is already partitioned on the join attribute, it need not be transmitted at all and only the inner relation is redistributed to match.

Bit vector filters in the split table

While each join process builds its hash table from the outer relation, it also hashes the join attribute values into a bit vector filter. When the building phase finishes, each process ships its filter to the scheduler, which collects them all and forwards them to the processes producing the inner relation, where they are installed as an array inside the split table. Inner tuples that cannot possibly find a match are then discarded at the producer, before they are ever put on the network. This is a semijoin performed with a cheap approximate set, and it converts join selectivity directly into saved communication.

How it works — the mechanism, concretely

Hardware and system software base

The prototype is 20 VAX 11/750 processors, each with two megabytes of memory, connected by an 80 megabit per second token ring built for the group by Proteon, plus a further VAX running Berkeley UNIX that acts as the host. Eight of the twenty processors have a 160 megabyte Fujitsu disk attached and hold the database; the rest are diskless and available for join and spool work. The processors run NOSE, an operating system written specifically for database work, which provides lightweight processes with shared memory, non-preemptive scheduling to avoid convoys, and a timer-based one-bit stop-and-wait positive acknowledgement protocol with a deltaT mechanism for re-establishing sequence numbers. File, record, index and scan services come from the Wisconsin Storage System, and the WiSS page format embeds the NOSE interprocessor message header so a page can be read from disk and sent to another processor without copying tuples into an outgoing message template.

Storage organization and the multiprocessor index

A partition command names the declustering strategy and, for range partitioning, the key boundaries: partition employee on emp_id (100, 300, 1000) puts emp_id up to 100 on processor 1, 100 to 300 on processor 2, 300 to 1000 on processor 3, and the rest on processor 4. If the user cannot supply ranges, Gamma loads round robin, sorts the relation with a parallel merge sort on the partitioning attribute, redistributes to equalize tuple counts per site, and returns the maximum key at each site to the host. Ordinary clustered and non-clustered indices can then be built on each fragment. When either range strategy is used, Gamma additionally constructs a multiprocessor index in which the disks and their processors are the nodes of a primary clustered index whose root page lives in the schema on the host; the optimizer reads that root page to decide that a query such as q between A and C need only be sent to processor 1.

Process structure

The Catalog Manager is a host daemon that holds all conceptual and internal schema information in UNIX files, loads it into memory when a database is opened, and keeps the copies cached by individual users consistent, with an internal lock manager guarding the catalog. One Query Manager exists per active user and does schema caching, parsing, optimization and compilation. Each multisite query is controlled by a Scheduler process which activates the operator processes for the nodes of the compiled query tree; schedulers run inside the machine rather than on the host because a message between two query processors is twice as fast as a message to the host, which must cross UNIX. An Operator Process runs each operator at each participating processor, while a centralized Deadlock Detection Process collects wait-for graph fragments from the lock managers and picks victims, and a Log Manager collects log fragments from the query processors for commit, abort and rollback.

Split tables and tuple routing

Gamma uses three kinds of split table. The hashed form applies a hash function to the join or partitioning attribute to yield an index, for example 0 through 3 on four processors, and looks up a destination pair of processor number and port number. The range form keys each entry by the upper bound of a partition range and is used both when a permanent relation is range partitioned and when the split attribute at a leaf of the query tree is the relation's horizontal partitioning attribute, in which case the table is initialized with the source relation's own boundaries so each fragment is processed locally with no transmission. The third form distributes tuples round robin, ignoring key values entirely, and is the default for result relations. An array of bit vector filters is inserted into the split table to drop tuples that cannot join before they are routed.

Hash join control flow

A join begins like any other operator, but with one extra control interaction. In the building phase each join process consumes tuples of the first source relation into in-memory hash tables and bit vector filters, then messages the scheduler that building is done. Only when the scheduler has heard from every join process does it send the message that starts the probing phase, during which each process reads tuples of the second relation and probes its hash table for matching join attribute values; a final message per process reports completion. Effectively the building and probing phases are treated as separate operators for control purposes, which is why the net cost of activating and controlling a hash join is five messages per site, and why the scheduler must also time the operators producing the two input streams to coincide with the two phases.

Hash table overflow

If buckets grow too large during building, the in-memory hash table overflows. The local join operator responds by narrowing the dimensions of the tuple partition used to build the table, creating two subpartitions: one continues to feed the hash table and the other is dumped to an overflow file on disk, which may be remote, and tuples already inserted that now belong to the overflow subpartition are removed from the table. When it reports completion of the building phase, the operator tells the scheduler which repartitioning scheme it used, so the scheduler can rewrite the split tables of the probing relation to spool the matching overflow subpartitions straight to disk, bypassing the join operators entirely. After the non-overflow subpartitions have been joined, the scheduler applies the join recursively to the spooled overflow subpartitions. This fails only when the tuples sharing a single join attribute value exceed available memory, in which case a hash-based variant of nested loops is used.

Selection and update operators

Selection throughput is treated as the gating factor for everything downstream, since an operator that cannot feed the tree limits the parallelism the rest of the plan can use. Gamma applies three complementary techniques: use indices whenever possible, compile selection predicates into machine language procedures so predicate evaluation is not interpreted, and use a limited form of read-ahead so processing of one page overlaps the I/O for the next. Update operators, replace, delete and append, use standard techniques with a single exception: a replace that modifies the partitioning attribute cannot write the tuple back into the local fragment, so the modified tuple is instead pushed through a split table to determine which site should now hold it.

What the paper showed — measurements and proofs

The prototype is fully operational: 20 VAX 11/750 processors with two megabytes of memory each, an 80 megabit per second token ring, and 160 megabyte Fujitsu drives on eight of the processors, running the NOSE operating system over the Wisconsin Storage System.
On synthetic relations of ten thousand 208-byte tuples (thirteen four-byte integers plus three 52-byte strings) with 10 percent selectivity, non-indexed selections on the partitioning attribute (S1) and on a non-partitioning attribute (S3) both showed speedup fairly close to linear from one to eight disks; S3 was in fact slightly faster than S1 because all processors share the cost of distributing the round-robin result relation rather than one processor producing it all.
Single-site indexed selection on the partitioning attribute (S2) improved only from one to about three processors and then flattened, because adding disks reduces index traversal levels and result-storage time but not the number of leaf data pages read, and the single producing processor becomes the bottleneck.
Against the commercial IDM500 with a database accelerator and an equivalent disk, Gamma's single-processor configuration was competitive: the IDM500 took 22.3 seconds for S1 selections, 5.2 seconds for S2, 84.3 seconds for J2 joins and 14.3 seconds for J4 joins, while Gamma retrieved a single tuple through the multiprocessor index in 0.14 seconds.
Joins run entirely on diskless processors (remote joins) were slightly faster than joins run on the disk processors, not slower, because locally the join and select operators compete for the same CPU and because Gamma transfers sequential tuple streams between processors at almost the same rate as between processes on one machine; this shows a complex operator can be offloaded off the storage nodes.
The paper reports and diagnoses a negative result: indexed selection on a non-partitioning attribute (S4) barely improves from four to eight processors, and it is not bandwidth, since redistributing 750 result tuples of 208 bytes is 1.2 million bits or about two hundredths of a second on an 80 megabit ring, but network interface congestion, because tuple-by-tuple round robin fills all 64 output buffers at nearly the same instant while the interface buffers only two incoming packets, forcing five packets per site to be retransmitted and colliding with acknowledgements.

Limits and trade-offs — conceded and discovered

Conceded by the paper: the evaluation is explicitly preliminary and single-user only, with no multiuser tests, no evaluation of the update operators, no aggregate operations or aggregate functions implemented yet, and only uniform attribute value distributions measured; correctness rather than absolute speed had dominated development.
Conceded: the network interface congestion behind the flat S4 curve was left unfixed in the paper, with a page-by-page round robin plus randomized first destination proposed but not measured, and the seventh and eighth disks used in the tests were older 14-inch drives with only 82 percent of the performance of the other six, further distorting the speedup curves.
Conceded: the building and probing phases of the join are not overlapped, so join response time is bounded by the sum of the two phases, and the ratio of control messages to data messages per processor rises as processors are added, which already showed up at eight disks where each join operator handled only about fourteen data pages per relation against five control messages.
Conceded but understated: the overflow scheme fails outright when the tuples sharing one join attribute value exceed available memory, falling back to a hash-based nested loops join, and the authors defer non-uniform attribute distributions to future work; data skew is the central weakness of hash-partitioned parallelism and later work on skew handling and adaptive repartitioning grew directly out of this gap.
Exposed by later work: Gamma deliberately has no site autonomy, a centralized schema and a single point for initiating all queries, with the deadlock detector and the scheduler pool also centralized, so the design has no story for node failure or elastic membership; the follow-on Gamma papers had to abandon the VAX and token ring hardware for a 32-node Intel iPSC/2 hypercube before scaling could be studied seriously.

What it became — the systems that inherited it

Gamma is the paper that turned shared-nothing from an argument into a measurement, and its architecture became the standard template for parallel relational systems, codified six years later in DeWitt and Gray's Parallel Database Systems paper in CACM. Teradata's DBC/1012, Tandem NonStop SQL, IBM DB2 Parallel Edition and Informix XPS all ship the same core recipe of declustered tables, partitioned hash joins and per-node operators. Goetz Graefe, a Gamma co-author, generalized the split table into the Volcano exchange operator, which made parallelism a single pluggable operator in an otherwise sequential plan and is now how SQL Server, Greenplum, Vertica, Spark SQL, Presto and Trino, Snowflake and Dremel-lineage engines all execute in parallel. The two-phase build-and-probe hash join with bit vector filters shipped back to the producers is the ancestor of today's broadcast and shuffle hash joins with runtime Bloom filter pushdown in Spark, Impala and Snowflake, and Gamma's recursive overflow subpartitioning is the ancestor of grace and hybrid hash join spilling. Its four declustering strategies, round robin, hashed and two flavors of range, are exactly the partitioning options every modern warehouse and sharded store still offers. Even the paper's honest failure has descendants: the network interface congestion analysis is an early instance of the shuffle-buffer and incast problems that MapReduce, Spark and every modern exchange implementation must manage.

In the paper’s words — verbatim

“the Gamma prototype shows how parallelism can be controlled with minimal control overhead through a combination of the use of algorithms based on hashing and the pipelining of data between processes.”

Abstract

“With the exception of these three control messages, execution of an operator is completely self-scheduling. Data flows among the processes executing a query tree in a dataflow fashion.”

§3.4

“To utilize the I/O bandwidth available in such a design, all relations in Gamma are horizontally partitioned across all disk drives.”

§6 Conclusions

Vocabulary — as this paper uses it

Horizontal partitioning (declustering): Spreading the tuples of a single relation across all disk drives in the system so that every relation is stored in fragments and every scan is inherently parallel. Gamma applies it to all relations without exception, using round robin, hashing, or one of two range schemes.
Split table: A small table at the output of an operator process that maps a value derived from each result tuple, a hash value, a range bound, or nothing at all for round robin, onto the processor and port of a destination process. It is the only place in Gamma where parallelism is expressed, which is why operator code can be written as if it were sequential.
Horizontal partitioning attribute (HPA): The attribute on which a relation was declustered. When a query's qualified attribute is the HPA and range partitioning was used, the optimizer can route the query to just the relevant sites; when it is not the HPA, the query must be sent to every site.
Multiprocessor index: An index built when a relation is range partitioned, in which the disks and their processors are themselves the nodes of a primary clustered index and the root page is kept with the schema on the host machine. The optimizer consults that root page to direct selections on the partitioning key to the right processors.
Building phase and probing phase: The two halves of Gamma's partitioned hash join: building consumes the first source relation into in-memory hash tables and bit vector filters at each site, probing then streams the second relation against those tables. They are separated by a scheduler barrier and are treated as separate operators for control purposes.
Bit vector filter: An approximate set built by hashing the join attribute values of the outer relation during the building phase, then collected by the scheduler and installed in the split tables of the processes producing the inner relation. Inner tuples that miss the filter are discarded before they reach the network.
Local join and remote join: The paper's terms for running a join entirely on processors that have disks attached versus entirely on diskless processors. Remote joins measured slightly faster, establishing that a complex operator can be offloaded away from the storage nodes.
NOSE and WiSS: NOSE is the purpose-built operating system underneath Gamma, giving lightweight processes with shared memory, non-preemptive scheduling to avoid convoys, and a reliable timer-based message protocol. WiSS, the Wisconsin Storage System, supplies file, record, index and scan services, with page formats that embed NOSE message headers so pages can be shipped without copying.

On the timeline — where this sits in the story

Year1986–1990s

MilestoneGamma and shared-nothing parallel DB research

Problem solvedRelational query execution needed to scale beyond a single server without centralized bottlenecks.

How it solved itGamma showed how data partitioning and parallel relational operators could run on shared-nothing architectures.

Why it matteredInfluenced parallel databases, MPP warehouses, and later distributed analytical engines.

DDIACh. 7: Sharding

View on the timeline