Efficient Sorting, Duplicate Removal, Grouping, and Aggregation

2.1 Interesting Orderings

From early relational database management systems, sort-based algorithms and sort order have been central to query processing engines. Early research into query evaluation and grouping algorithms [11] discussed duplicate removal within sort operations and in-stream grouping for sorted inputs: “…first project the needed domains and then sort on the by-list being careful not to remove duplicates …Since the tuples are sorted in order of the by-list, each tuple read will have either the same by-list as the previous tuple, or it will be an entirely new by-list and there will be no more references to any previous by-lists.” The same research also considered grouping using in-memory hash indexes: “…the assumption that \(B \ge P\) [memory size \(\ge\) output size] is commonly true in practice. To the extent that this holds, the best structure to use is hash, and sorting does not help. If \(B \lt P\) and U [output row count] is large, then sorting clearly wins.”

Early research into query optimization crystallized the concept of interesting orderings and their effect on query evaluation plans [35]. Sort-based algorithms such as merge join have obvious positive interactions with sorted storage structures such as b-tree indexes as well as queries with “order by” clauses. Multiple joins on the same primary key and foreign keys are common in re-assembly of complex objects after relational normalization in the database. Grouping on foreign keys is common, because it computes an aggregate property of the larger entity, e.g., the total value of all line items in a purchasing order. Thus, grouping operations before or after primary key-foreign key joins are found in many queries and query evaluation plans.

In queries with a single “select…from…where…group by…” block, the default evaluation plan executes grouping and aggregation after all joins. It has long been understood that query optimization may modify this sequence order in many queries, e.g., to reduce join input sizes [40]. Another advantage might be pushing the grouping and aggregation logic into the sort required for the join algorithm. While an important and powerful set of optimizations, the present article focuses on run-time efficiency and ignores these and many other query planning techniques.

2.2 Applications of Sort-based Grouping and Aggregation

The algorithms discussed in this article support sort-based duplicate removal, grouping, and aggregation. These discussions go beyond earlier descriptions of sorting and duplicate removal in relational database management systems [4, 11, 16, 24]. A related operation avoids redundant invocations of expensive operations such as (correlated) nested queries and user-defined functions [25].

A typical example of a large duplicate removal operation is counting distinct users in a popular web service. Logs generated by web servers may produce billions of log records per day. A dataflow pipeline or a SQL query extracts user identifiers and then removes duplicates, i.e., multiple log records pertaining to the same user. For a popular web service, this reduces many billions of rows to many millions of rows.

If counts are desired per hour or per country, then the required grouping operation can use the same algorithm. In hash-based query processing, one operation (with hash table and hash-partitioning to overflow files on temporary external storage) removes duplicate user ids and another operation (with its own hash table and hash-partitioning) counts users per hour and country. In sort-based query processing, a single sort operation (on country, hour, and user identifier) serves both duplicate removal and subsequent grouping. Queries of the form “select count (distinct…)… group by…,” i.e., the combination of duplicate removal and grouping, can always benefit from interesting orderinga but should, of course, also benefit from the most efficient available algorithms for duplicate removal and grouping.

“Integrated join” [15, 20] and “group-join” [31] use a single hash table for both grouping and join. They are particularly effective when grouping and joining on the same foreign key. Due to their asymmetry, they inhibit role reversal, whereas in sort-based query processing, in-stream grouping naturally applies to both inputs of merge join as well as its output. For unsorted join inputs, the sort logic can apply duplicate removal, grouping, and aggregation.

“Generalized join” [17, 19] includes merge logic that consumes many runs with a single buffer page, using most of memory for an in-memory index. In that sense, it is similar to wide merging (Section 4). The prior work [19] stipulates an algorithm for duplicate removal, grouping, and aggregation based on generalized join with output candidate rows serving as build input and input rows as probe input, whereas wide merging does not distinguish between build and probe inputs. Nonetheless, wide merging as implemented in F1 Query may be interpreted as a first industrial implementation and deployment of “generalized distinct.” Similarly, early aggregation (Section 3) may be seen as an implementation of the input phase or of a hybrid phase of generalized distinct. Due to a lack of understanding and appreciation of offset-value coding, those earlier attempts might achieve competitive I/O volumes, i.e., overflow partitions in hash join and runs in external merge sort, but they do not achieve equivalent column value accesses and thus CPU effort (Section 5.1).

Rollup functionality has existed for a long time in programming environments such as Cobol and been suggested for database queries [14]. Sort-based aggregation can compute multiple levels of aggregation with a single sort operation, e.g., for a query of the form “select …group by rollup (year, month, day).” In contrast, hash-based aggregation requires separate computations for each level of aggregation. Each level requires a hash table and possibly partitioning to temporary storage.

Log-structured merge-forests and stepped-merge forests [27, 32] are nearly ubiquitous in key-value stores. In this context, runs are often called deltas and merging is often called compaction, because merging includes aspects of aggregation and compression. The individual merge steps are similar to those of external merge sort, but their merge policies (what to merge when) are quite different for multiple reasons. First, their input is assumed endless. For example, it is not possible to delay merging until run generation is complete; merging must be concurrent to run generation. Second, inputs include traditional insertions, which are mapped to append operations, as well as updates, which are mapped to insertions of replacement rows, and deletions, which are mapped to insertions of “tombstone” rows. The merge logic aggregates insertions, updates, and deletions either into a final state or into a history of versions, including removal of out-of-date versions (also known as garbage collection). Third, individual runs are formatted as b-trees, not flat files, to permit search and queries over recent as well as historical information. Alternative formats include a single partitioned b-tree, with runs mapped to partitions. Bit vector filters can enable a query to skip some partitions and thus improve performance. Fourth, the merge fan-in and the frequency of merge steps are controlled not by the memory size but by the desire for good query performance, i.e., searching few partitions. Many designs and deployments of log-structured merge-forests employ low-fan-in merge steps, even binary merging.

Decades ago, Gray suggested sorting recovery log records on the database page identifier to which they pertain [22]: “This compressed redo log is called a change accumulation log. Since it is sorted by physical address, media recovery becomes a merge of the image dump of the object and its change accumulation tape.” It seems a small step from sorting recovery log records to building b-tree indexes, another step to building indexes incrementally and continuously (in the manner of log-structured merge-forests), and yet another small step to using such indexes for page-by-page incremental, on-demand, seemingly instant recovery from single-page failures, from system failures (software crashes), and from media failures [21].

2.3 Optimizing “Group by” and “Order by”

Functional dependencies enable many interesting optimizations for “group by” and “order by” queries [38]. Functional dependencies are implied by primary key integrity constraints and by prior “group by” operations.

More specifically, a “group by” clause requires a set of columns (expressions) and an “order by” clause requires a list of columns. Functionally dependent columns can be removed anywhere in a set but only in subsequent positions within a list. For example, in two database tables for purchase orders and their line items, with o_orderdate functionally dependent on o_orderkey, the first three among the following four clauses permit simplification but the last one does not:

Below is a (first) example of using functional dependencies in an unusual way. The first query seems to require grouping and aggregation after the join, but the second and third queries are essentially equivalent to the first one due to the functional dependency of order date on the grouping key. Adding a functionally dependent column to a “group by” clause applies the insights of Reference [38] in the reverse direction. As grouping key and join key are the same, order date is a constant within each group of line items. The fourth query variant is equivalent to the first query and most conducive to efficient execution. Note that the many-to-one join changes into a one-to-one join.

View Figure

In many queries, query rewriting such as this example is required to enable integrated join or group-join. In integrated join, optimizing grouping and join on the same column (set) applies only to the build input; in group-join, it applies only to the probe input. In sort-based query processing, grouping and join on the same column (set) enjoy the benefits of interesting orderings if grouping is applied to either of the two join inputs, or even to the join output. In other words, interesting orderings benefit query performance whether or not query optimization applies all kinds of clever and uncommon rewrites.

In sum, sort-based duplicate removal, grouping, and aggregation can benefit from proper use of functional dependencies, because they permit optimizations of both grouping and ordering, but it seems that sort-based query evaluation plans are somewhat more forgiving and flexible than hash-based query execution.

2.4 Optimizations of Sort Operations

High-performance sorting requires efficiency, scalability, and robustness of performance. Efficiency may benefit from tree-of-losers priority queues, normalized keys, offset-value coding, and hardware support. Among the techniques mentioned, normalized keys encode key values in order-preserving strings such that all subsequent key comparisons use intrinsics or hardware instructions for binary strings. Further hardware support may focus on tree-of-losers priority queues and offset-value coding in normalized keys, e.g., the UPT “update tree” instruction and the CFC “compare and form codeword” instruction [26]. Iyer [26] sums up that “Together, the UPT and CFC instructions do the bulk of sorting in IBM’s commercial DBMS DB2 running on z/Architecture processors.”

Offset-value coding [7] encodes one row’s key value relative to another row’s key value that is earlier in the sort sequence. Offset and value are combined into an integer such that a single machine instruction can compare two offset-value codes and decide a comparison of two rows encoded relative to the same base row. Offset-value codes can decide many comparisons in tree-of-losers priority queues.

Table 3 illustrates the derivation of descending and ascending offset-value codes in a stream of rows in ascending sort order on all columns. With four sort columns, the arity of the sort key is 4; the example assumes that the domain of each column is 0…99. Descending offset-value codes take the actual offset but the negative of the column value. Ascending offset-value codes take the negative offset but the actual column value. Table 3 ignores that small key domains permit encoding multiple key columns together.

Finally, a tree-of-losers priority queue [13, 28], also known as a tournament tree, embeds a balanced binary tree in an array, with the tree’s unary root in array slot 0. It is efficient due to leaf-to-root passes with one comparison per tree level; root-to-leaf passes with two comparisons per tree level are not required. Run generation and merging with tree-of-losers priority queues can guarantee sort operations with near-optimal comparison counts.

Scalability is principally about parallelism—twice the resources should process the same data in half the time (also known as speed-up) or twice the data in the same time (scale-up). Robustness of performance is about performance cliffs and graceful degradation—for example, the transition from an in-memory quicksort to an external merge sort should be gradual rather than a binary switch, such that a single additional byte or input row cannot force spilling the entire memory contents. The techniques introduced in Sections 3 and 4 are orthogonal to both scalability and robustness of performance: the new techniques do not offer improvements in those directions but they also do not impede or hinder existing or future techniques for scalability and for graceful degradation.

2.5 Early Results in Join-by-grouping

Complementing optimizations of sort-based grouping, there is a join algorithm based on grouping. It requires that the implementation of external merge sort can interleave multiple record types within memory and within each run on temporary storage. Sorting a mixed stream of records on join key values produces mixed “value packets” [29], i.e., sets of rows with equal sort keys. In the context here, equal sort keys means equal join keys. Forming or combining value packets is a kind of aggregation. The join output is computed from the final sorted stream by computing a cross product within each mixed value packet. Alternatively, when the sort and merge logic forms or combines mixed value packets, it can produce join results as an immediate side effect. In other words, early aggregation in this context enables early and incremental join results. Variants of this algorithm can compute semi-join, anti-semi-join, all forms of outer join, set and bag intersection and difference (e.g., “intersect all” in SQL). Anti-semi-join and equivalent result rows of outer joins cannot be produced early.

Figure 4 illustrates one merge step within this algorithm applied to an inner join. Two unsorted input tables (far left top and bottom) are joined on column A. Rows from both inputs are scanned concurrently and run generation creates initial sorted runs (center left). These runs contain multiple record types, one for each join input. Merging runs (from center left to center right) relies on the standard merge logic known from external merge sort. As a side effect, this merge step combines value packets (in this example, for \(A = 13\)) and produces new join results (far right top). Before this merge step, when run generation assembled the merge inputs (center left), it produced early join results (far right middle and bottom) while forming the value packets in these runs.

Fig. 4.

View Figure

Once two records have been joined, they remain in the same value packet until the sort finishes. Hence, there is no danger of duplicate (redundant, wrong) output. For example, in Figure 4, the two original inputs (far left) contain 3 and 2 rows with key value \(A = 13\) for \(3 \times 2 = 6\) rows in the join result; the three partial results (far right top to bottom) contain precisely these 6 rows. When the sort finishes, the final value packets are dropped; the operation’s output is the join result computed incrementally as side effect of forming and combining mixed value packets.

This join algorithm is an alternative to more complex sort-based join algorithms with early output [10]. Its output rate and memory requirements mirror those of symmetric hash join [39] if early aggregation and wide merging are enabled, which are the topics of the next two sections.

2.6 Summary of Related Work

To summarize our observations on related prior work, duplicate removal, grouping, and aggregation occur in a large variety of contexts, from data warehouse queries and business intelligence to analysis of web logs. Substantial research and development effort have been invested in both query optimization and query execution specifically for duplicate removal, grouping, and aggregation. A remaining thorny problem is that traditional sort- and hash-based algorithms are optimal in different circumstances (relative sizes of input, output, and memory; sort orders interesting for subsequent operations), rendering a choice during compile-time query optimization difficult and error-prone. Instead, the next two sections offer a single algorithm that, assuming equal care in algorithm implementation, always matches the best traditional algorithm for duplicate removal, grouping, and aggregation, at least in terms of data movement (including I/O) and of column accesses and comparisons.

3.1 Example 1

As an archetypal example of a grouping query with a small output, consider Query 1 of the TPC-H benchmark [1]. In scale factor \(SF = 1\) of the benchmark, the query scans a database table of about 6M rows, selects about \(90\%\) of those, and then reduces them to four rows with counts, sums, and averages. For scale factor \(SF =\) 1,000, the query reduces 6B rows to four output rows. Clearly, an external merge sort of 6M or even 6B rows with subsequent grouping and aggregation, i.e., the two techniques of Figure 2, cannot compete with an algorithm that uses a hash table to simply accumulate the final query result within a small memory allocation of 4 rows.

There are many real-world queries in which a grouping result fits within the readily available memory allocation. For example, if each operator within a query evaluation plan is allotted 100 MB by default, then any grouping operation with a result smaller than 100 MB should remain an in-memory operation and algorithm. This is, of course, true for hash aggregation but it can also be true for sort-based duplicate removal, grouping, and aggregation.

Figure 5 illustrates this case. The in-memory index can grow until it fills memory. Skew in the key value distribution does not matter as an ordered index adapts to the actual distribution. Only if the output size exceeds the available memory, spilling to runs on temporary storage as shown in Figure 6 is required.

3.2 Example 2

As another example, imagine the “group by” clause of Example 1 extended such that the final output is larger than memory, i.e., \(O \gt M\) or more specifically \(O = 2M\). Even with early duplicate removal, grouping, and aggregation, this example requires runs on temporary storage. As key values in the in-memory index are unique, the in-memory index immediately matches and absorbs a fraction of the input rows. With run generation by replacement selection and memory always full, about \(M/O = \frac{1}{2}\) of all input rows are absorbed immediately. Ignoring the effects of an in-memory run for graceful degradation from in-memory sorting to external merge sort, the total size of all initial runs is about \(M + (1 - M/O) \times I\) (for input size I) or in the specific example \(M + \frac{1}{2}I\). With only unique key values in the in-memory index, the traditional logic for duplicate removal, grouping, and aggregation [4] while writing runs on temporary storage is not required.

Figure 7 shows the predicted spill volume for input size \(I =\) 1,000,000 rows and memory size \(M =\) 100,000 rows. For the left-most point, the output size equals the memory size (\(O=M\)) and no spilling is required. If the output is many times larger than the available memory allocation, then practically all input rows spill. The calculation in this prediction assumes replacement selection using an in-memory index, even though our implementation uses read-sort-write cycles. (Recall that replacement selection is usually implemented using a priority queue and for random input produces runs twice the size of memory, whereas read-sort-write cycles are usually implemented using quicksort and produce runs the size of memory; recall also that an ordered in-memory index permits either read-sort-write cycles or replacement selection, according to the implementer’s preference.)

Fig. 7.

View Figure

3.3 Algorithms and Data Structures

One design combining in-memory run generation with early aggregation uses two data structures. For example, Boncz et al. [5] mention “hash-based early aggregation in a sort-based spilling approach.” If the in-memory hash table fails to absorb (i.e., to aggregate) an input row, then the row is added to both in-memory data structures, i.e., the hash table as well as an array for quicksort or a priority queue for replacement selection.

An alternative design employs a single data structure for two purposes, searching and sorting. A suitable data structure is an ordered in-memory index, e.g., an in-memory b-tree [3, 18]. Note that comparisons are required only during the search. If no match is found, then the search for a match has already found the insertion point as a side effect.

If the search employs a binary search within each tree node, then the count of comparisons per input row is equal to that in a tree-of-losers priority queue, which is 10–20% lower than the count of comparisons in quicksort and very close to the theoretical minimum. (Sorting N items is equivalent to selecting one of \(N!\) permutations, which requires \(\log _2 (N!)\) comparisons.)

Interpolation search within each tree node is even more efficient if the key value distribution is nearly uniform, which is likely the case if the sort key is a hash value. Note that sorting on hash values permits exploiting interesting orderings if other algorithms and storage structures also sort on hash values. Merge joins and b-trees on hash values are attractive for database columns with no real-world “\(\lt\)” comparison, e.g., practically all artificial identifiers and thus many primary keys and foreign keys in real-world databases.

The row format within the in-memory index is the same as in runs on temporary storage. It may differ from the row formats in both the input and the output. For example, in addition to a grouping key, input rows may contain a value, intermediate rows a sum and a count, and output rows an average. Similar considerations apply when computing variance, standard deviation, co-variance, correlation, regression slope and intercept, and so on.

In traditional run generation, read-sort-write cycles may use quicksort to produce runs the size of memory. Run generation by replacement selection using a priority queue can produce runs the size of memory or, with an additional comparison for each new input row as well as a flag within each row in memory, twice the size of memory. Run generation using an in-memory index can produce runs twice the size of memory without an additional comparison and without a flag in each row in memory. Eviction from memory to temporary storage repeatedly scans the in-memory index as shown in Figure 6. Advancing the scan evicts rows and frees index nodes whenever the in-memory index needs to split a node and thus allocate a free node.

3.4 Analysis

Three questions suggest themselves for analysis:

In run generation using an efficient tree-of-losers priority queue, the count of comparisons per input record is \(\log _2 (M/R)\) for memory size M and record size R. This is correct for run size equal to memory size M; one additional comparison is required for expected run size 2M. Using quicksort, the expected count of comparisons is 10–25% higher; the worst case for quicksort is much higher. In run generation using an in-memory index with binary search, the count of comparisons per input record is again \(\log _2 (M/R)\). Using interpolation search or batches of search keys, it can be substantially lower.

If the final output fits in memory, i.e., \(O \le M\), then the count of comparisons per input record is \(\log _2 (O/R)\). Again, it can be substantially lower with interpolation search or batches of search keys. Linear interpolation is effective if the key value distribution in the output is practically uniform. This is probable if the keys are hash values, i.e., when sorting and grouping on hash values, at least as the leading sort key.

In a striking similarity, hash aggregation requires a search in the hash table for each input record, i.e., a hash calculation plus a scan through a hash bucket. Those are comparable to the interpolation calculation and the local search near the interpolated position. If memory remains full all the time during run generation, then each input row has a probability of \(M/O\) (memory size over output size) of finding a matching key value in memory and of being absorbed in the index without insertion. If \(M \ge O\), then this probability is a certainty and spilling to runs on temporary storage is not required. If this probability is very small, then practically all input rows spill into runs on temporary storage.

While the count of comparisons is a traditional metric of algorithm complexity, it is neither the only metric nor the most meaningful one in practice. For example, in a table with 10 or 100 columns, hash-based duplicate removal must locate, access, decompress, and hash every column in every row to compute hash values, whereas sort-based duplicate removal might never access more than the first few columns if those are sufficient to decide comparisons. Section 5.1 analyses these effects in more detail.

3.5 Summary of Early Aggregation

To summarize, early aggregation uses an in-memory index to match and absorb input rows during run generation in duplicate removal, grouping, and aggregation. In the ideal case, it entirely avoids spilling rows to temporary storage. A typical example is TPC-H Query 1 with only four output rows even for very large databases and input tables.

The in-memory index can be a simple b-tree or it can be optimized in many ways. A binary search guarantees \(\log _2 (M)\) comparisons per input row. Replacement selection can produce runs twice as large as memory. An alternative design combines an in-memory hash table (for immediate detection of duplicates) with a priority queue (for replacement selection). This alternative suffers from twice the space overhead and twice the maintenance effort as well as the complexity of coordinated incremental insertions and deletion of rows, e.g., during graceful degradation from an in-memory operation to an external algorithm spilling to temporary storage.

Whatever the algorithm for run generation, runs require merging with the traditional merge logic known from external merge sort or, in many cases, an optimized merge logic to be known as wide merging. Interestingly, both early aggregation and wide merging replace priority queues, the traditional data structure of choice for sorting, with ordered in-memory indexes, e.g., in-memory b-trees.

4.1 Example 3

Consider a specific example of wide merging and its benefits: single-threaded duplicate removal with input size \(I = \text{750,000}\) rows, memory size \(M = \text{1,000}\) rows, partitioning fan-out and merge fan-in (in traditional merge logic) and partitioning fan-out \(F=6\), and final output size \(O = \text{32,000}\) rows. Importantly, the memory is much smaller than the final output and the final output is much smaller than the original input, or \(M\ll O\ll I\).

In this example, hash aggregation invokes \(L=2\) partitioning levels to divide all input rows into \(F^2 = 36\) partitions of about \(I / F^2 = \text{21,000}\) rows each. During these partitioning steps, the output buffers are too small to enable much early (opportunistic) duplicate removal. After two partitioning levels, each partition contributes about \(O / F^2\) rows to the final output. As the output per partition is smaller than memory (\(O / F^2 = 900 \lt \text{1,000} = M\)), duplicate removal can occur in memory in spite of input partitions much larger than memory (\(I / F^2 = \text{21,000} \gt \text{1,000} = M\)). The total size of all temporary partitions in both partitioning levels is \(2I = \text{1,500,000}\) rows, each written and read once. More generally, hash aggregation can aggregate the remaining partitions in memory after \(L \ge \log _F (O/M)\) partitioning levels, which is here \(\log _6 (32) = 2\) partitioning levels.

In contrast, in-sort duplicate removal, grouping, and aggregation starts with run generation by replacement selection. Each run holds about \(2M = \text{2,000}\) rows; thus, this example requires about \(I / (2M) = 376\) runs. The first merge level produces \(376/F = 63\) runs of about \(F \times \text{2,000} = \text{12,000}\) rows. The second merge level produces \(63/F = 11\) runs. Aggregation within runs [4] cuts their size from \(F \times \text{12,000} = \text{72,000}\) rows to \(O = \text{32,000}\) rows. The penultimate merge step combines 6 of these 11 runs into another run of \(O = \text{32,000}\) rows and the last merge step produces the final output. The total size of all runs spilled to temporary storage during run generation, full merge, optimized merge, and partial merge is \(\text{750,000} + \text{750,000} + 11 \times \text{32,000} + \text{32,000} = \text{1,884,000}\) rows, each written and read once. This is about 25% more than the temporary partitions in hash aggregation.

Wide merging enables further savings. In this example, a single final merge step can aggregate the 63 runs after the first merge level. Merging 63 runs with memory for only a few input buffers requires an in-memory index for immediate duplicate removal.

A traditional merge step merging 6 of these 63 runs has its output cut to \(O = \text{32,000}\) rows. This is true whether the merge logic uses traditional single-page buffers and a priority queue or an in-memory index for wide merging. If such an index holds practically all distinct key values over the course of the merge step, then it can (with the appropriate pacing and I/O schedule) absorb rows and key values not only from 6 but any number of runs, e.g., all 63 runs in this example.

With wide merging, i.e., the final merge step immediately after run generation and one full merge level, the total size of all temporary runs is \(\text{750,000} + \text{750,000} = \text{1,500,000}\) rows and thus perfectly competitive with hash aggregation. As in the cost calculation for hash aggregation, the size of the original input determines the cost of each partitioning or merge level yet the size of the final output (together with memory size and partitioning fan-out or merge fan-in) determines the count of required partitioning or merge levels.

Wide merging with duplicate removal, grouping, and aggregation using an in-memory index proceeds in \(O/M\) steps. Each step produces a memory load of candidate output rows, with gradual progression from one step to the next. The runs being merged must have more than \(O/M\) data pages such that the key range of each data page is no larger than the key range of the in-memory index. After run generation, in each row’s first temporary run on storage, the size of runs is M or 2M. In each merge level, the size increases by the fan-in F. The count of merge levels L must ensure that \(F^L \ge O/M\) or \(L \ge \log _F (O/M)\), in a striking similarity to the expression for partitioning levels required in hash aggregation.

Figure 10 illustrates this gradual progress through the key domain as required for wide merging as shown in Figure 9. A priority queue guides the page reads. For key values earlier than the top value of the priority queue, aggregates such as counts and sums are final. These can be produced as final output and removed from the in-memory index. As additional pages are read and their key values are inserted, the in-memory index grows, mostly on the right edge but occasionally also elsewhere.

Fig. 10.

View Figure

4.2 Example 4

Perhaps a similar example might clarify further. Compared to Example 3, this example uses more realistic values for memory size, merge fan-in, and partitioning fan-out: duplicate removal with input size (per thread) \(I = \text{100,000,000}\) rows, memory size (per thread) \(M = \text{100,000}\) rows, partitioning fan-out and (traditional) merge fan-in \(F = 100\), and final output size (per thread) \(O = \text{8,000,000}\) rows. Despite more realistic values for M and F (compared to Example 3), it remains true that \(M \lt O \lt I\).

Hash aggregation starts with a full partitioning level, which produces 100 partitions of about 1,000,000 rows each. Each partition contributes about 80,000 rows to the final output, which can be aggregated entirely in memory. The total spill volume (partitions on temporary storage) is equal to the input size or 100,000,000 rows. The algorithm requires practically the entire available memory throughout, i.e., during both partitioning and in-memory aggregation. Actually, 90% of memory would suffice: memory size \(M = \text{90,000}\) rows and partitioning fan-out \(F = 90\) suffice if the hash value distribution is perfectly uniform.

In-sort duplicate removal, grouping, and aggregation starts with run generation, which produces about 500 runs of about 200,000 rows each. Traditional merging requires five intermediate merge steps to reduce the count of runs from 500 to 100 as required for the final merge step.¹ The output size of these steps is limited to the final output size \(O = \text{8,000,000}\) rows. Thus, the total spill volume is I + 1 \(\times\) \(\text{1,000,000}\) \(+\) 4 \(\times\) \(\text{8,000,000}\) \(=\) \(\text{133,000,000}\) rows, which is 33% worse than the total spill volume in hash aggregation. Nonetheless, the algorithm requires the entire available memory throughout, i.e., during run generation, intermediate merge steps, and the final merge step.

In contrast, wide merging can merge and aggregate the initial 500 runs in a single step (even if memory and page sizes limit traditional merging to fan-in 100). Thus, the spill volume is equal to the input size or 100,000,000 rows, matching the performance of hash aggregation. In this algorithm, run generation requires the entire available memory (M = 100,000 rows) but the final (wide) merge step and its in-memory index require only about 40,000 rows (40% of M).

Alternatively, a memory allocation for run generation of \(M = \text{64,000}\) rows reduces the size of initial runs to about 128,000 rows and produces about 782 runs. Wide merging can combine all these runs in a single final merge step using a memory allocation of about 63,000 rows. In other words, wide merging permits sort-based duplicate removal, grouping, and aggregation with equal I/O volume but with a lower memory allocation than hash aggregation.

4.3 When to Apply Wide Merging

In most practical situations, hash-aggregation can process its input in memory either immediately or after one level of partitioning. In the same cases and situations, wide merging can process its initial runs, no matter their count, without the in-memory index exceeding its size limit. In other words, the situation of Section 4.2 is much more frequent than that of Section 4.1. If, however, the in-memory index with candidate output rows occupies fraction p of its maximal permissible size, then this index and its existing rows absorb the fraction q of rows in pages read from runs, and if \(q \lt p\), then wide merging cannot absorb all rows within its memory limit.

In those cases, traditional merge steps are required before wide merging, because wide merging only applies to the final merge step. Those merge steps use their page buffers in the traditional way (one per merge input) and traditional limits on the fan-in apply. Instead of an in-memory index, the recommended in-memory data structure for these merge steps is a tree-of-losers priority queue.

Pang et al. [33] investigated and experimented with techniques that let an external merge sort react gracefully to growing and shrinking memory grants. For shrinking during the merge phase, their principal technique interrupts the current merge step and introduces a preliminary merge step that reduces the count of remaining runs and thus permits resuming the interrupted merge step with a lower fan-in and thus with fewer buffers. This technique also applies to wide merging: If wide merging is not yet applicable, either from the start of a merge step or merely for a key range, then a preliminary merge step can reduce the run count.

In contrast to the situations Pang et al. investigated, wide merging in the final merge step does not require reducing the run count to the buffer count during the final merge step. Instead, it is sufficient to enable wide merging. A preliminary merge should focus on the smallest runs, which is also the cheapest way to reduce the run count. In the context of wide merging, the significant effect of this policy is to increase the smallest size among the remaining runs, to increase the count of pages in the smallest remaining run, and thus to decrease the key range of those pages. Those smaller key ranges per page enable wide merging, i.e., an in-memory b-tree smaller than the permissible memory might suffice for immediate duplicate removal or aggregation. Preliminary merge steps are required until this becomes possible. As mentioned above, however, this problem should be rare, in fact precisely as rare as hash-aggregation with recursive or multi-level partitioning.

4.4 Analysis

A few questions suggest themselves for further analysis.

In response, it seems that run generation and final merge step using an in-memory index offer performance advantages only in queries that require duplicate removal, grouping, or aggregation. Ordinary sorting, e.g., for “order by” queries and for merge join operations, does just as well with traditional algorithms and data structures, e.g., quicksort or tree-of-losers priority queues.

Wide merging is useful only in the final merge step; it might require earlier merge levels like traditional external merge sort. The number of traditional merge levels is a function of initial run size (memory size), merge fan-in, and final output size. All merge steps in which the step’s total input size is smaller than the operation’s final output size should use traditional merge logic. This analysis assumes that a merge step’s individual inputs are of similar size.

Wide merging applies when merging with a traditional fan-in processes effectively all distinct key values, i.e., when the total merge input is larger than the operation’s final output. This is precisely the first merge step (or merge level) in which traditional early aggregation [4] first becomes effective. The difference is that wide merging immediately produces the operation’s final output by consuming all remaining runs, whereas traditional early aggregation still might require multiple merge steps and levels.

If traditional merge steps are required prior to wide merging in sort-based duplicate removal, grouping, and aggregation, then these merge steps must create runs at least as large as the operation’s final output divided by the fan-in of traditional merge steps (\(O/F\)). It appears that there is little benefit in creating larger intermediate runs. Runs of size \(O/F\) enable traditional early aggregation [4] and, better yet, wide merging. In other words, wide merging replaces (rather than augments) traditional early aggregation. Creating runs of size \(O/F\) requires \(\log _F (O/M)-1\) merge levels after run generation creates runs of memory size M. With the final merge step (merge level) using wide merging, sort-based duplicate removal, grouping, and aggregation requires \(\log _F (O/M)\) merge levels.

4.5 Combining Early Aggregation and Wide Merging

If the final output is only somewhat larger than the available memory, e.g., \(O~=~2M\) or \(O~=~3M\), then early aggregation during run generation and its in-memory index can absorb some of the input rows without growing the index or spilling rows from memory to runs on temporary storage. For example, if \(O~=~2M\), then the rows in memory can absorb half of all input rows; if \(O~=~3M\), then the in-memory index matches and absorbs a third of all input rows; and so on. Nonetheless, a large input can force many runs on temporary storage. In those cases, wide merging can eliminate one or even two merge levels. In other words, a single sort can benefit from both early aggregation and wide merging. With those two techniques and their combined effects, sort-based duplicate removal, grouping, and aggregation always performs very similarly to hash-based alternatives, as discussed further in Section 6.

4.6 Example 5

For another example that differs from Example 4 only in the final output size, consider duplicate removal with input size (per thread) I = 100,000,000 rows, memory size (per thread) M = 100,000 rows, (traditional) merge fan-in and partitioning fan-out F = 100, and final output size (per thread) O = 150,000 rows. In other words, \(I \gg O = 1.5 M\).

In hash aggregation with hash-partitioning, about half of all input rows find a match in memory: hybrid hashing [36] is quite effective in this case. However, the total spill volume is about \(\frac{1}{2}I = \text{50,000,000}\) rows. Sort-based aggregation with early aggregation matches the same fraction of input rows during creation of the initial runs. With replacement selection and a run size of about \(2M = \text{200,000}\) rows, about 250 initial runs are required. With run generation in read-sort-write cycles and a run size of \(M = \text{100,000}\) rows, about 500 initial runs are required. This is too much for traditional merging with fan-in \(F = 100\), but nonetheless wide merging can finish the aggregation in a single merge step. Thus, this example benefits from both early aggregation and wide merging; with these techniques, sort-based aggregation can match the spill volume and performance of hash aggregation.

4.7 Summary of Wide Merging

To summarize, wide merging uses its in-memory index and a single input buffer for all runs on temporary storage. It enables the final merge step in duplicate removal, grouping, and aggregation to consume and to combine many more runs than a traditional merge step using an input buffer for each run. Wide merging applies when traditional merging would produce runs larger than the final output of the grouping operation. Matching the performance and I/O volume of hash aggregation in all cases requires both early aggregation and wide merging.

5.1 Column Accesses

Common wisdom holds that operations on hash values are faster than equivalent operations on complex sort keys. For example, comparing two hash values is much faster than comparing two rows with multi-column keys, in particular if some of these columns are strings or even international strings with specific locales and their sort order.

Offset-value coding [7, 16, 26] applied to columns can put this common wisdom into question, however. On the one hand, offset-value codes are integer values compiled into the system’s source code just like hash values, meaning that a comparison of two offset-value codes is precisely as fast as a comparison of two hash values. On the other hand, sort-based query execution algorithms sometimes require far fewer column accesses than equivalent hash-based algorithms. For example, in duplicate removal for a table with \(K = 20\) or even \(K = 200\) columns, computing the hash values for a table of many rows, e.g., \(N = 10^6\) rows, must access all \(N \times K\) column values. In contrast, sort-based duplicate removal and its many comparisons access only the leading columns required to distinguish each row from its neighbors in the final output. In an extreme case, if the first column happens to be unique, sort-based duplicate removal never accesses the remaining \(K - 1\) columns. Thus, in this extreme case, sort-based duplicate removal accesses K times fewer column values than any hash-based equivalent. In another extreme case, if all rows differ from their neighbors only in the last (\(K{\text{th}}\)) key column, the count of column accesses in sort-based duplicate removal is twice that of hash-based duplicate removal. In other words, with offset-value coding, the count of column accesses in a sort-based algorithm is at worst twice that of a hash-based algorithm but possibly K times lower.

Another extreme case has no distinct rows, e.g., the output of duplicate removal is a single row. In that case, a hash-based algorithm computes a hash value for each row and then compares all columns with the one and only row in the hash table. This requires \(3 K\) column accesses for each input row after the first one. An equivalent sort-based algorithm performs \(N - 1\) row comparisons, each with \(2 K\) column accesses, or less than a hash-based algorithm by a factor of \(1.5\).

Table 4 summarizes these extreme cases. Common cases fall in between these extreme examples. Note that \(N - 1\) was simplified to (approximated as) N in Table 4. Note also that the analysis of column accesses in Table 4 does not account for hash collisions, i.e., different rows mapping to the same hash value. More importantly, the sort order can be chosen freely; otherwise, hash-based algorithms would not apply. If columns with many distinct values are assigned to early positions in the sort key, then hash-based duplicate removal requires more column value comparisons and column value accesses than sort-based duplicate removal. This difference is close to a factor of \(K / 2\), which is a substantial difference for \(K = 20\) or even \(K = 200\) columns. Previous analyses of sort- and hash-based query execution algorithms have not considered the column count as a significant variable although Table 4 shows that it is.

The analysis of sort-based duplicate removal in Table 4 permits another improvement by a factor two, summarized at the bottom of Table 4. If each row has a cache of all offset-value codes ever computed for that row, i.e., if no offset-value codes are ever computed redundantly, then only \(N \times K\) offset-value codes can ever be computed and the maximum number of column accesses in sort-based duplicate removal is \(N \times K\), i.e., equal to the minimum number of column accesses in hash-based duplicate removal due to the calculation of hash values.

If, due to offset-value coding, most comparisons in a sort-based algorithm are as fast as comparing hash values, then the efforts to locate, access, extract, decompress, and decode \(K=20\) or even \(K=200\) column values in each row dwarf the effort to perform \(log_{2}(\text{1,000,000}) = 20\) comparisons for each row. In other words, the \(log_{2}N\) component in the formula of comparison-based sorting matters little, because the count of column accesses dominates the overall performance of duplicate removal. Of course, as in any comparison-based sort, there are \(O(NlogN)\) row comparisons, but many or most of the row comparisons are decided by offset-value codes alone, with comparisons as simple and fast as comparisons of hash values. In fact, with offset-value coding applied to columns, the number of column value comparisons is bounded to \(N \times K\), i.e., it is linear in both the row count and the column count. (The alternative applies offset-value coding to bytes within a normalized key [26], i.e., the entire key encoded in a single order-preserving binary string. Computing such a normalized key is just as expensive as computing a hash value from all key columns, although bytes in a normalized key could conceivably be computed on demand even if that is not possible for hash values.)

5.2 Spilling in a Memory Hierarchy

In addition to comparisons of rows, keys, and column values, the other significant cost in duplicate removal, grouping, and aggregation is moving intermediate rows up and down within the memory hierarchy. The traditional case in point is spilling from memory to temporary storage—terms commonly used in this context are external devices and external algorithms, e.g., external sorting. In many modern query execution engines and their deployments, very similar considerations and optimizations apply to spilling to memory from CPU caches, usually the second- or last-level cache.

As in all external algorithms, the pertinent questions include when to spill, what to spill, how much to spill at a time, and how much spillage is required altogether. Of course, a particularly interesting question is about the conditions for no spillage at all, i.e., when an external algorithm is not required because an internal algorithm suffices. The principal parameters in the relevant formulas are the input size I, the output size O, the memory size M, and the partitioning fan-out or merge fan-in F (when using the entire memory) or f (when using partial memory, e.g., in hybrid algorithms such as hybrid hash join [36]).

The analysis below pursues these questions in detail. It omits spillage (partitioning and merging) in deep (multi-level) memory hierarchies, e.g., CPU cache–memory–external storage devices, memory–flash storage–disk storage, or CPU cache–memory–local storage–disaggregated (distributed, remote) storage. The analysis below also omits parallel algorithms including opportunistic pre-aggregation before a shuffle step (opportunistic = best effort, in-memory only; shuffle = exchange, partitioning). These omitted topics require merely straightforward extensions or repeated application of the analysis below.

The analysis below proceeds in three steps. The first step works out conditions for internal algorithms with no spillage at all. The second step assumes incredibly large inputs and outputs that require external algorithms with recursive partitioning or multi-level merging. This step is closest to a traditional “big O” complexity analysis but adds approximate constants. The third step focuses on conditions and cost functions for hybrid algorithms that divide memory between an internal algorithm for some of the input and an external algorithm for the remainder of the input. Like this last sentence, the analysis focuses on spilling from memory to external devices.

5.3 Summary of Analytical Comparisons

To summarize, this analysis and comparison of sort- and hash-based algorithms for duplicate removal, grouping, and aggregation considers both CPU effort and I/O effort. With respect to CPU effort, sort-based algorithms require more row comparisons than equivalent hash-based algorithms but often fewer column comparisons. Moreover, with offset-value coding, they often require far fewer column accesses. In fact, if all offset-value codes are cached once computed, the worst-case count of column accesses in sort-based duplicate removal equals the best-case count of column accesses in hash-based duplicate removal, and the best-case count of column accesses in sort-based duplicate removal is substantially better. With respect to I/O effort, sort-based algorithms, with the new techniques for early aggregation and wide merging, require no more spilling to temporary external storage than equivalent hash-based algorithms. Thus, this algorithm analysis justifies our belief that a single algorithm is sufficient for all cases of grouping rows from unsorted inputs.

6.1 Product Implementation

This section briefly summarizes the implementation of the new in-sort aggregation techniques for Google’s F1 Query [34, 37]. F1 Query is a federated query processing platform that executes SQL queries against data stored in different storage systems at Google, e.g., BigTable [6], Spanner [8], Mesa [23], and others. Before this work, F1 Query had two aggregation operators. First, for sorted input, in-stream aggregation requires little CPU effort and hardly any memory. The F1 Query optimizer chooses in-stream aggregation whenever possible based on interesting orders in the input [35]. Second, for unsorted input, hash-based aggregation relies on an in-memory hash table. This hash-based operator relies on external merge sort when the output is larger than the available memory allocation.

The new in-sort aggregation algorithm reuses the row-plus-row accumulation component of hash-based and in-stream aggregation. A new order-based indexing component is used for detecting duplicates and groups. For each input batch, the operator first sorts the batch, usually within the CPU cache as these batches are small, to detect duplicates within a batch. Only distinct key values within a batch are looked up in the ordered index with the guided search technique. When running out of memory, the ordered index guides the sequence of rows spilled to intermediate storage, creating sorted runs. These runs are eventually merged and aggregated using wide merging. Contrary to our initial design, the current implementation uses read-sort-write cycles, not replacement selection; therefore, the size of initial runs equals the memory size, even if each run might have absorbed substantially more input with replacement selection in duplicate removal and grouping.

As the new in-sort aggregation produces sorted output as a byproduct of using the ordered index, an optimizer can take advantage of this property. For example, in queries with a “group by” clause followed by an “order by” clause with the sort keys matching the grouping keys, the F1 Query optimizer avoids redundant sorting. Before our new operator, F1 Query planned such aggregation queries using either a hash aggregation followed by a sort or a sort followed by an in-stream aggregation. Plan choices can be sub-optimal due to missing or inaccurate cardinality information. The new in-sort aggregation operator overcomes this problem by always enabling the optimal plan.

6.2 Validation of Examples

Example 1 (Section 3.1) focuses on TPC-H Q1, i.e., a grouping query with a final output smaller than the available memory allocation (\(O \le M\)). Figure 11 shows the performance of in-memory grouping and aggregation using an in-memory b-tree index. From left to right, the output size varies from 4 to 30,000 rows. The input size is constant 6,000,000 rows. As is readily apparent, the CPU effort is low and fairly consistent, because any effects due to the logarithmic depth of the ordered index vanishes compared to other CPU efforts in the query evaluation plan.

Fig. 11.

View Figure

Example 2 (Section 3.2) also varies TPC-H Q1 with output sizes beyond memory size (\(O\gt M\) but \(O\lt F\) \(\times\) M). Figure 12 compares the total size of initial runs to a model that assumes run generation with replacement selection and computes the spill volume as \(M + (1 - M/O) \times I\). In contrast, our implementation relies on run generation by read-sort-write cycles. Given this difference, the distance between these curves seems acceptable.

Fig. 12.

View Figure

Example 3 (Section 4.1) assumes tiny memory, merge fan-in, and partitioning fan-out. Therefore, all algorithms incur multiple partitioning or merge levels. In contrast to traditional sorting and merging, wide merging limits the merge depth \(\log _F (O/M)\) versus \(\log _F (I/M)\). Figure 13 shows the performance advantage of wide merging over aggregation while writing runs [4] as baseline: entire merge levels can be avoided, whereas earlier method merely reduce the size of intermediate runs on temporary storage.

Fig. 13.

View Figure

Example 4 (Section 4.2) assumes realistic memory size, merge fan-in, and partitioning fan-out. Therefore, a single level of merging suffices if wide merging is available. Figure 14 shows that even for modest input sizes, traditional early aggregation (the baseline) requires multiple merge levels. Note that an algorithm can spill more rows than its input size only if it spills some rows multiple times. In contrast, each input row spills only once when wide merging combines all runs in the first (and only) merge step immediately after run generation.

Fig. 14.

View Figure

Example 5 (Section 4.6) shows that in some cases, early aggregation and wide merging are both required for best performance of sort-based duplicate removal, grouping, and aggregation. The experiment in Figure 18 (Section 6.4) confirms the example calculations.

In summary, the examples and the related experiments demonstrate that early aggregation and wide merging, by using in-memory ordered indexes instead of the traditional priority queues, derive substantial benefits for duplicate removal, grouping, and aggregation.

6.3 Early Aggregation During Run Generation

The next set of experiments and diagrams focuses on the hypotheses that, for any input size, output size, row size, page size, and memory size:

The experiments cannot claim to cover all sizes and key values distributions, but they may nonetheless helpful in understanding the performance and scalability of in-sort aggregation with early aggregation during run generation.

Figure 15 shows the performance of in-memory aggregation using either a hash table (hash aggregation) or a b-tree (in-sort aggregation). None of these experiments spill to temporary storage. As is readily apparent, the performance of hash table and in-memory b-tree, both properly optimized, is quite similar. Their remaining performance differences are so minor that other query execution costs such as aggregation arithmetic dominate them. In that sense, in-memory b-trees in sort-based grouping are as fast as hash tables in hash-based grouping, supporting Hypothesis 1.

Fig. 15.

View Figure

Figure 16 shows the performance of three implementations of run generation. Two of these use tree-of-losers priority queues; one of them is optimized with normalized keys for fast comparisons and poor man’s normalized keys for cache efficiency [16]; the other priority queue is optimized with offset-value coding [7]. The third implementation of run generation uses an in-memory b-tree. Again, other query execution costs such as aggregation arithmetic dominate these differences and the experiment supports Hypothesis 2.

Fig. 16.

View Figure

Figure 17 shows the number of initial runs spilled to temporary storage. Each run is the size of memory, either a hash table sorted and written as run or a b-tree written in total when memory is full. Recall that the hash aggregation in F1 Query uses what Boncz et al. [5] call “hash-based early aggregation in a sort-based spilling approach,” which sorts rows in an overflowing hash table, writes them as initial runs on temporary storage, merges those runs, and applies duplicate removal, grouping, and aggregation only during the final merge step. Recall also that our implementation of in-sort aggregation uses its in-memory index for run generation in read-sort-write cycles, not replacement selection. Thus, the counts of initial runs are practically equal, supporting Hypothesis 3.

Fig. 17.

View Figure

6.4 Wide Merging in the Final Merge Step

The next experiments test the hypotheses that, for any input size, output size, row size, page size, and memory size:

Figure 18 reports on the total size of all runs for the experiment of Figure 17. Note that this experiment compares optimized in-sort aggregation with the original hash aggregation of F1 Query. Both algorithms spill from memory to sorted runs on temporary storage. The two algorithms achieve the same amount of in-memory aggregation during this phase, and thus merging in the two algorithms starts with the same counts and sizes of partially aggregated runs. The original algorithm of F1 Query relies on traditional merging, which requires multiple merge steps with intermediate merge results. Thus, the total spill volume exceeds the input size for all input sizes in Figure 18. In contrast, the new algorithm employs wide merging for duplicate removal, grouping, and aggregation. A single merge step suffices and no intermediate merge steps create any additional spill volume. For all input sizes in Figure 18, the total spill volume is much less than the input. One of the data points precisely matches Example 5 (Section 4.6) and all data points support Hypothesis 4.

Fig. 18.

View Figure

Figure 19 shows the cost of a query with matching “group by” and “order by” clauses over a table of \(I = \text{6,000,000}\) rows. If the output of an initial duplicate removal is small, in particular no larger than memory \(M = \text{1,000,000}\) rows, then the cost of a subsequent sort operation barely matters. If, however, the intermediate result is large, then satisfying both clauses with a single operation is very beneficial, supporting Hypothesis 5.

Fig. 19.

View Figure

Figure 20 shows the cost of duplicate removal with subsequent grouping. With hash-based algorithms, two hash tables (and possibly overflow to temporary storage) are required. With a sort-based algorithm, a single sort can perform the duplicate removal using an interesting ordering for the subsequent grouping. Thus, only one memory-intensive operation is required with savings up to a factor of two, supporting Hypothesis 6.

Fig. 20.

View Figure

6.5 Effects of Interesting Orderings

The next experiment tests the hypotheses that:

Figure 21 shows a query evaluation plan for a very simple SQL query computing the intersection of two tables, e.g., “select B from T1 intersect select B from T2.” If column B is not a primary key in tables T1 and T2, then correct execution requires duplicate removal for each input plus a join algorithm to find intersection result. Without the work described in this article, F1 Query chooses a hash-based physical plan with three blocking operators: two hash aggregation operators for duplicate removal and the hash join for set intersection. In contrast, with the new in-sort aggregation operator, F1 Query chooses a sort-based plan physical plan with only two blocking operators: both are in-sort aggregation operators for duplicate removal. Since the output of in-sort aggregation operators are sorted on column B, F1 Query exploits interesting ordering and uses merge join for set intersection, supporting Hypothesis 7.

Fig. 21.

View Figure

Figure 22 shows the performance of sort- and hash-based plans for this query. Each input table has \(I =\) 100,000,000 rows; the memory for each operator is \(M =\) 10,000,000 rows. In the hash-based plan, both duplicate removal operations and the join might spill to temporary storage; each input row is spilled twice. In contrast, the sort-based plan spills each input row at most once. Thus, the effort for spilling is cut in half due to interesting orderings, supporting Hypothesis 8. With in-sort aggregation, set intersection and its most efficient query evaluation plans benefit not only users’ “intersect” queries but also star queries and snowflake queries in relational data warehousing.

Fig. 22.

View Figure

6.6 A Belated Correction

Section 4.4 and Figure 11 of Reference [14] compare sort- and hash-based duplicate removal, grouping, and aggregation. The overall conclusions are that sorting the input for subsequent in-stream aggregation is not competitive and that both sort- and hash-based aggregation exploit strong data reduction and small output sizes.

Figure 23 is a copy of Figure 11 of Reference [14]. As perhaps appropriate at the time, the experimental parameters are input size \(I = 100\text{ MB}\), memory size \(M = 100\text{ KB}\), page size \(P = 8\text{ KB}\), merge fan-in and partitioning fan-out \(F = 10\), and output size O varying from 100 MB to 100 KB, or from input size I to memory size M. The “group size or reduction factor” is the quotient of input and output sizes, \(I \div O\). “Early aggregation” in this diagram means duplicate removal within runs on temporary storage [4]. The I/O volume reflects both writing and reading on temporary storage, i.e., the values in Figure 23 are \(2 \times\) higher than the “total run size” metric used in the present article.

Fig. 23.

View Figure

Figure 24 augments Figure 23 and maybe should replace Figure 11 of Reference [14], if that were possible. Compared to Figure 23, Figure 24 omits the curves for sorting without early aggregation and for hash aggregation without hybrid hashing, but it reproduces two of the curves using the same cost functions and parameters as in earlier work [14]. In addition to sorting with traditional early aggregation [4] and hash aggregation with hybrid hashing, Figure 24 shows a new curve for sort-based aggregation with both early aggregation (run generation using an in-memory index—Section 3) and wide merging (a final merge step using an in-memory index—Section 4). The essential observation is that the gap between sort- and hash-based aggregation, clearly visible in Figure 23, practically disappears in Figure 24. The curves are particularly close in the operating range towards the right with only a single merge or partitioning step. With today’s memory sizes, most grouping operations in production workloads require only one merge or partitioning step. Put differently, sort- and hash-based aggregation algorithms perform very similarly for unsorted inputs. While a single diagram cannot prove it, this is true for any combination of input and output sizes. Moreover, sort-based aggregation is less susceptible to skew in the key value distribution than hash aggregation is to skew in the hash value distribution.

Fig. 24.

View Figure

6.7 Summary of Hypotheses and Observations

Together, our experiments confirm the calculations in our earlier examples, support our claims and hypotheses about the effectiveness of sort-based aggregation with early aggregation and wide merging, and belatedly correct an algorithm comparison published a quarter century ago.