Parallel data sorting

Techniques for high-performance parallel data sorting are provided. K, M, and N exceed 1. In a first phase, a plurality of unordered data elements to be sorted is divided into K unordered lists each preferably having approximately M elements. Each of these K unordered lists can be independently sorted in parallel using any algorithm, such as quicksort, to generate K ordered lists. In a second phase, N balanced workloads are determined from the K ordered lists by using an iterative converging process capped by a maximum number of iterations. Thus, any non-uniform or skewed data distribution can be load balanced with minimal processing time. Once the N balanced workloads are determined, they can be independently sorted in parallel, for example by using a merge sort, and then combined with a fast concatenation to provide the final sorted result. Thus, sorting operations are fully parallelized while avoiding expensive data scanning steps.

FIELD OF THE INVENTION

The present disclosure relates to data sorting, and more specifically, to improved, high-performance parallel data sorting suited to multi-threaded and multi-node environments.

BACKGROUND

Sorting data is a classic optimization problem with practical application in a wide variety of academic and industrial fields. Computer applications may require high-performance sorting methods to conduct business intelligence analytics, to provide presentation rendering, to respond to external requests from users and applications, and for other tasks. For example, a database may be queried for a list of records that is sorted according to user or application defined criteria. Since the overall processing time to answer these queries is directly impacted by the sort execution time, a high performance sort is needed to provide results in a timely manner. Sort performance is especially important for applications working with big data sets such as database management systems (DBMSs) for large enterprises or high-performance computing (HPC), as the large number of data records may magnify the execution time of any sorting operation.

Multi-threaded processing may be utilized to provide suitable response times for these data intensive applications, wherein resources such as processor cores and/or processing nodes are added according to the data processing workload. With a highly parallelizable workload, multi-threaded processing has the potential to provide optimized performance scaling in a cost efficient and practical manner. Since sorting may contribute a large proportion of the data processing workload, sorting becomes a prime target for parallelization to reduce query latency times and to improve data processing throughput in multi-threaded environments.

Serial sorting techniques such as quicksort are readily available, providing sufficient performance for applications with low to moderate data processing needs. However, these serial sorting methods are less applicable for multi-threaded applications with high data processing needs. While various approaches for parallelizing serial sorting methods have been suggested, these approaches may break down when attempting to process a large number of elements that need to be sorted in a data intensive application, which may number in the billions or more, or when attempting to distribute the workload to a large number of parallel processing threads in a highly multi-threaded environment, which may number in the hundreds or more.

Furthermore, a given data set to be analyzed may include any kind of data distribution, and thus a sort must be able to process a data set regardless of its particular data distribution. Any parallelization approach that requires a lengthy pre or post-processing step to cope with non-uniform data distributions may impose an unacceptable performance penalty by reducing or negating performance gains obtained from parallelization. For example, while radix-sort may be amenable to parallelization as each partition can be independently sorted, the partitioning of data according to most significant bits provides poor workload balancing for non-uniform or skewed data distributions. Thus, a computationally expensive pre-processing step is required for radix-sort to cope with non-uniform data distributions, for example by conducting a serial data scan to determine balanced workload partitions. While a parallel data scan is also possible, this would impose significant processing overhead due to the inter-process communication required to resolve write contention, which only grows worse as the number of threads increases. In either case, the performance penalty from the pre-processing step may outweigh any performance gains from parallelizing the radix-sort.

Based on the foregoing, there is a need for a method to provide high-performance parallel data sorting suited to multi-threaded and multi-node environments.

DETAILED DESCRIPTION

General Overview

In an embodiment, an improved method for high-performance parallel data sorting is provided. In a first phase, a plurality of unordered data elements to be sorted is divided into K unordered lists each preferably having approximately M elements. Each of these K unordered lists can be independently sorted in parallel using any suitable algorithm, such as quicksort, to generate K ordered lists.

In a second phase, N balanced workloads are determined from the K ordered lists received from the first phase. Each of the N balanced workloads is bounded by a particular range of values from the K ordered lists, allowing each of the N balanced workloads to be fully sorted in parallel and quickly concatenated at a final combining step. To determine the particular range of values or the index splits for a particular workload, candidate index splits are chosen and refined up to a maximum number of iterations such that a size of the particular workload converges towards a balanced workload size, or 1/N of the K ordered lists. In this manner, the N balanced workloads can be quickly determined without scanning the actual data. Once the N balanced workloads are determined, they can be independently sorted in parallel, for example by using a K-way merge sort. These sorted workloads can then be combined to provide the final sorted result.

This improved high-performance parallel data sorting method provides several technical advantages to achieve higher performance in highly multi-threaded and multi-core environments. First, since the parallel sorting at each phase proceeds independently, no inter-process communication is required, greatly simplifying implementation and eliminating expensive overhead. Second, since N balanced workloads are determined, load balancing for N threads is optimized regardless of the particular data distribution of the elements to be sorted. Third, because the index splits of the N balanced workloads are determined by iteratively converging towards a balanced size, no expensive data scan steps are required. Accordingly, this improved high-performance parallel data sorting method has particular relevance for applications that need to process big data sets, numbering in the billions of elements or more, on hardware with many parallel processing units, numbering in the hundreds or more.

Database Systems

Embodiments of the present invention are used in the context of DBMSs. Therefore, a description of a DBMS is useful.

A DBMS manages a database. A DBMS may comprise one or more database servers. A database comprises database data and a database dictionary that are stored on a persistent memory mechanism, such as a set of hard disks. Database data may be stored in one or more data containers. Each container contains records. The data within each record is organized into one or more fields. In relational DBMSs, the data containers are referred to as tables, the records are referred to as rows, and the fields are referred to as columns. In object-oriented databases, the data containers are referred to as object classes, the records are referred to as objects, also referred to herein as object records, and the fields are referred to as attributes. Other database architectures may use other terminology.

Users interact with a database server of a DBMS by submitting to the database server commands that cause the database server to perform operations on data stored in a database. A user may be one or more applications running on a client computer that interact with a database server. Multiple users may also be referred to herein collectively as a user.

A database command may be in the form of a database statement that conforms to a database language. A database language for expressing the database commands is the Structured Query Language (SQL). There are many different versions of SQL, some versions are standard and some proprietary, and there are a variety of extensions. Data definition language (“DDL”) commands are issued to a database server to create or configure database objects, such as tables, views, or complex data types. SQL/XML is a common extension of SQL used when manipulating XML data in an object-relational database.

A multi-node database management system is made up of interconnected nodes that share access to the same database. Typically, the nodes are interconnected via a network and share access, in varying degrees, to shared storage, e.g. shared access to a set of disk drives and database blocks stored thereon. The nodes in a multi-node database system may be in the form of a group of computers (e.g. work stations, personal computers) that are interconnected via a network. Alternately, the nodes may be the nodes of a grid, which is composed of nodes in the form of server blades interconnected with other server blades on a rack.

Resources from multiple nodes in a multi-node database system can be allocated to running a particular database server's software. Each combination of the software and allocation of resources from a node is a server that is referred to herein as a “server instance” or “instance”. A database server may comprise multiple database instances, some or all of which are running on separate computers, including separate server blades.

Improved Parallel Sorting System Overview

To support data intensive applications with large numbers of records and transactions, server node120utilizes a multi-core architecture to execute multiple threads concurrently, thereby reducing latency times and increasing data processing throughput. While only a single processor130and four processing cores132A-132D are shown inFIG. 1, embodiments may include any number of processors and processing cores, numbering in the hundreds or more. Processing cores132A-132D may be independent physical or logical cores that are capable of executing respective threads134A-134D concurrently. Additionally, while only a single server node120and a single client110are shown inFIG. 1, embodiments may include multiple server nodes and/or multiple clients. Further, while system100is shown in the context of networked client-server architecture, system100may be flexibly configured according to specific application requirements. For example, in the context of system-on-a-chip or embedded applications, client110and server node120may be combined into a single monolithic device.

In an embodiment, system100may be used in the context of databases. However, system100is not necessarily limited to database contexts and service142may correspond to any application or service that requires data to be sorted. In the context of databases, server node120may correspond to a database server with service142corresponding to a DBMS for database170, enabling client applications such as application112to interact with database170. Accordingly, application112may send a database query to service142over network180, wherein the database query requests records that are sorted according to certain criteria. In an embodiment, this database query may correspond to a SQL SELECT query that includes an ORDER BY clause.

When service142receives the database query from application112, service142may retrieve the requested database records of the SELECT query from database170, which are then stored in memory140as a list of unsorted records, or unsorted data elements160. Unsorted data elements160may contain a large number of elements, for example billions or more. For simplicity, it may be assumed that memory140is large enough to avoid any I/O overhead from swapping to disk.

Service142may utilize sorting module150to sort unsorted data elements160according to the criteria in the database query, or the ORDER BY clause, to generate sorted result168. As shown in sorting module150, various processing steps are carried out to convert unsorted data elements160into sorted result168, as described below in conjunction withFIG. 2A. Note that while each processing step of sorting module150may appear to create a new data element in memory140, embodiments may create or modify structures such as pointers, linked lists, or other data structures as appropriate to avoid expensive memory copy operations.

To accelerate sorting module150, parallel sort152and parallel merge156utilize multiple threads concurrently, or threads134A-134D inFIG. 1. Each of these threads operates on independent portions of data in memory140, bypassing concurrency issues such as write contention and overhead from inter-process communication. In embodiments with multiple nodes, sorting module150may distribute parallel processing steps and associated data to threads on remote nodes (not shown inFIG. 1) via network180. The remote nodes may correspond to multi-core server nodes similar to server node120. Once sorted result168is available, it may be returned to application112via network180to respond to the original database query.

Improved Parallel Sorting Process

With a basic outline of system100now in place, it may be instructive to review a high level overview of the processing steps carried out by sorting module150. Turning toFIG. 2A,FIG. 2Ais a block diagram that depicts a process for improved parallel data sorting, according to an embodiment.

Receiving K Ordered Lists

At block202of process200, referring toFIG. 1, server node120receives K ordered lists, or ordered lists162, wherein M represents a number of elements in each of the ordered lists162. In other words, the ordered lists162should be preferably balanced in size, with each of the K ordered lists including approximately M elements. However, in some embodiments, one or more lists of ordered lists162may include a number of elements that deviates from M to some extent.

Block202may correspond to a first phase of process200. In some embodiments, ordered lists162may be provided by a program or service other than service142. For example, a program from a different server node may send an external sort request with ordered lists162over network180. In another embodiment, server node120may create ordered lists162in response to a request from service142to sort unsorted data elements160. In this case, the receiving of ordered lists162will be from transforming unsorted data elements160, as illustrated inFIG. 1andFIG. 2B.

Referring toFIG. 2B,FIG. 2Bis a block diagram that depicts a process for receiving K ordered lists each preferably including approximately M elements, according to an embodiment. Process222ofFIG. 2Bincludes split151, parallel sort152, unsorted data elements160, unordered lists161, and ordered lists162. Process222may correspond to block202fromFIG. 2A. With respect toFIG. 2B, numbered elements may correspond to like numbered elements fromFIG. 1.

As shown inFIG. 2B, process222may begin by receiving unsorted data elements160. For example, referring toFIG. 1, service142may populate unsorted data elements160with records retrieved from database170to answer a SQL SELECT query from application112. Sorting module150may then receive a request from service142to receive and sort unsorted data elements160according to the GROUP BY clause in the SQL SELECT query.

For illustrative purposes, unsorted data elements160is populated with T=64 integer elements having a normal distribution with a mean of 40 and a standard deviation of 20. Thus, unsorted data elements160have a non-uniform distribution. However, unsorted data elements160may correspond to any number of elements T, such as billions or more elements, having any data distribution, including uniform, non-uniform, and highly skewed distributions. Furthermore, for simplicity, the elements in unsorted data elements160are integers to be sorted in ascending numerical order. However, embodiments may include elements as records to be sorted according to one or more sort keys, wherein each sort key can be in ascending or descending order.

Process222may utilize split151to divide unsorted data elements160approximately evenly into unordered lists161. Referring toFIG. 1andFIG. 2B, split151is configured with K=4 since there are 4 threads available for use at server node120, or threads134A-134D. Thus, unordered lists161include 4 lists each having M=T/K=64/4=16 elements. When T cannot be exactly evenly divided by K, each list in unordered lists161may include approximately M elements. Additionally, in some embodiments, split151may be configured to split only at particular split boundaries, such as between blocks of a fixed size that contain multiple elements, rather than splitting at the granularity of individual elements. In this case, unordered lists161may include lists having a number of elements that deviate from M.

As shown inFIG. 2B, each of lists 1-4 in unordered lists161directly corresponds to a contiguous segment of unsorted data elements160. Accordingly, split151can avoid any computationally expensive data analysis steps by dividing unsorted data elements160directly into contiguous segments based on the number of elements (T) within unsorted data elements160, the number of threads (K) available for use at server node120, and any split boundaries, if applicable.

With unordered lists161now available, process222may continue with the step of parallel sort152to sort each of the unordered lists161in parallel. For example, referring toFIG. 1, each of threads134A-134C may sort a respective list 1-4 of unordered lists161in parallel. Any sorting method may be utilized in parallel sort152, such as quicksort. Since the lists in unordered lists161are roughly the same size with approximately M elements each, threads134A-134D should complete closely in time to minimize any blocking from waiting for a particular thread to finish. After parallel sort152is finished, ordered lists162may be received, populated as shown inFIG. 2B. Of course, since ordered lists162are only sorted with respect to each individual list, further processing is needed to generate the final sorted result168. Thus, process200may move from the first phase of block202to a second phase, beginning with block204.

Defining a Target Size Range

At block204of process200, referring toFIG. 1, server node120defines, for N balanced workloads164, a target size range that does not exceed a predetermined threshold from a target size of KM/N. Since 4 threads are available, or threads134A-134D, N is set to 4. Note that in this example K=N, since both parallel sort152and parallel merge156execute on the same 4 threads134A-134D. Thus, the target size is KM/N=4(16)/4=16.

However, K does not necessarily need to equal N. K may not match N if the number of threads utilized is different between parallel sort152and parallel merge156. For example, one or more threads134A-134D might have been unavailable during parallel sort152but made available during parallel merge156, leading to K<N. This may occur, for example, if one or more processing cores132A-132D are busy during parallel sort152, for example by executing threads from a different or higher priority process. In embodiments where ordered lists162is received from an external source, K may also differ from N since the external source may have used a different number of threads to create ordered lists162. Regardless, both parallel sort152and parallel merge156operate as fully parallel steps, utilizing respectively K and N threads, which may or may not be the same number.

Since balanced workloads164are determined using split point converge154, or a process that iteratively converges a workload size towards a balanced size, defining a target size range is useful to reduce processing time. More specifically, split point converge154may finish early if the workload size is within the target size range prior to a maximum allowable number of iterations. Thus, the predetermined threshold can be set larger to reduce the execution time of split point converge154, or the predetermined threshold can be set smaller to reduce size variance in balanced workloads164. The predetermined threshold may be set manually or automatically. For example, the predetermined threshold may be set by using a ratio, such as 50% of M/N, or 0.50(16/4)=2. Based on this predetermined threshold, the target size range includes sizes that do not exceed 2 from the target size 16, or [14, 18], which includes sizes 14 through 18 inclusive.

Determining Balanced Workloads

At block206of process200, referring toFIG. 1, server node120determines the N balanced workloads164as including, from each of the K ordered lists162, a subset that is bounded by a particular index range, wherein the determining adjusts, up to a maximum number of iterations, the particular index range for each said subset to converge a size of each said N balanced workloads164towards the target size range defined in block204. InFIG. 1, block206corresponds to split point converge154.

Referring toFIG. 2C,FIG. 2Cis a block diagram that depicts a process for determining N balanced workloads from the K ordered lists ofFIG. 2B, according to an embodiment. Process226ofFIG. 2Cincludes converging parameters153, split point converge154, ordered lists162, and balanced workloads164. Balanced workloads164includes workload165A, workload165B, workload165C, and workload165D. Process226may correspond to block206ofFIG. 2A. With respect toFIG. 2C, numbered elements may correspond to like numbered elements fromFIG. 1.

Converging parameters153specifies the parameters that are used by split point converge154. As discussed with the example in conjunction with block204, the predetermined Threshold value may be set to 2. Further, as discussed above, split point converge154may iterate for a maximum number of iterations MaxIterations=3. Similar to the Threshold value, the MaxIterations value may be set manually or automatically, with a larger number of iterations reducing size variance in balanced workloads164and a smaller number of iterations reducing the execution time of split point converge154.

After executing split point converge154, it can be seen that workload165A includes 16 elements, workload165B includes 19 elements, workload165C includes 15 elements, and workload165D includes 14 elements. Thus, each of workloads165A-165D is balanced to include approximately M elements, or 16 elements. Further, as shown inFIG. 2C, each of workloads165A-165D include a subset of values from each list 1-4 of ordered lists162, which are bounded by particular index ranges as indicated by the shaded lines inFIG. 2C. The determination of these index ranges is described in further detail below in conjunction withFIG. 2D-2F, which respectively describe a process for converging workloads165A-165C towards the target size range of [14, 18].

Iterative Split Point Converging

Before discussing the specific examples shown inFIG. 2D-2F, each iteration in split point converge154for a particular workload may be described using the following steps:

1. If the current iteration is the first iteration, then select a particular list from the ordered lists and select an element with the index M/N as the candidate split point. The particular list that is chosen is not vital; for simplicity the particular list is always selected to be the last list in the examples, or list 4.

2. If the current iteration is after the first iteration, then the candidate split point is adjusted from a previous split point based on the current size of the current workload (determined in step 4 below). If the current size is smaller than the target size range, then the particular list is the list with the smallest split point in the ordered lists. If the current size is larger than the target size range, then the particular list is the list with the largest split point in the ordered lists. Ties may be resolved by arbitrarily picking one of the tied lists. Set the candidate split point for the particular list by using the converging formula i=┌(i+M/N)/2┐, wherein the left side i corresponds to the index for the candidate split point and the right side i corresponds to the index for the previous split point in the particular list. Note that this converging formula is only one example; the converging formula may be adjusted to increase or decrease the distance moved for each converging step.

3. Find the split points for the other lists in the ordered lists based on the value at the candidate split point in the particular list. For example, the split point for the other lists may be chosen such that values prior to the split point do not exceed the candidate value at the candidate split point. In some embodiments, the values prior to the split point may include one or more values that equal the value at the candidate split point. This may provide some flexibility in adjusting the split points when a large amount of skew in the data results in a large number of identical values. Since the ordered lists are already sorted with respect to each ordered list, the split point may be quickly found in each of the other lists, for example by using a converging sampling algorithm or another method to avoid a full data scan. Furthermore, if the current iteration is after the first iteration, the last split point may be used as a starting reference point to find the current split point.

4. A current size of the current workload is determined by summing the number of elements prior to the split point (including the candidate split point) of each of the ordered lists. This current size is compared to the target size range. If the current size is within the target size range, then the converging process ends early. If the current iteration has reached the maximum number of iterations permissible or MaxIterations, then the converging process also ends early. In this case, if a large number of identical values are adjacent to one or more of the split points, then the split points may be adjusted accordingly to attempt to reach the target size range. If none of the above applies and the current iteration has not yet reached MaxIterations, then a new iteration begins with step 1 above.

5. After the above steps 1-4 have completed, a single set of split points for the sorted lists is now known, which also defines a particular index range for each of the list subsets that make up a particular workload, the particular index range starting from a beginning offset and not exceeding the split point. The beginning offset for each of the ordered lists may be initialized to zero (0) when determining the first workload. To determine the next set of split points, the beginning offset of each ordered list may be moved to the split point, and steps 1-4 as described above may be repeated to determine the next workload. This process may be repeated to determine all of the workloads, with the exception that the final workload simply corresponds to all of the elements from the beginning offset to a last element of each of the ordered lists after the penultimate workload (or N−1 workload) is determined.

Iterative Split Point Converging—First Workload

Having described the iterative split point converging process in general, it may be instructive to examine the process applied to specific example data. Turning toFIG. 2D,FIG. 2D, is a block diagram that depicts a process for converging a workload165A towards the target size range [14, 18] by iteratively adjusting index splits for the K ordered lists, according to an embodiment. The first iteration begins with split point iteration155A. As described above in step 1, the particular list is selected as the last list, or list 4, with the index M/N, or 16/4=4 as the candidate split point. Thus, the candidate split point is indicated in list 4 with the pipe or | symbol, just prior to the element at index 4 having a value of 20, indicated in bold in split point iteration155A. Note that this example uses the convention wherein the first index is index 0; thus, index 4 actually refers to the 5thelement.

Since split point iteration155A is the first iteration, step 2 is skipped and step 3 begins. In step 3, the split points are found for all other lists, or list 1, list 2, and list 3. Since the candidate split point has a candidate value of 20, the split points are chosen such that values prior to the split points do not exceed 20. Thus, as shown in split point iteration155A, the split point for list 1 is prior to index 2 (18≦20), the split point for list 2 is prior to index 2 (13≦20), and the split point for list 3 is prior to index 1 (13≦20). These split points are indicated by the | or pipe symbol in lists 1, 2 and 3, with the largest values not exceeding 20 indicated in bold.

A current size of workload165A is determined by summing the number of elements prior to the split point of each of the ordered lists. Thus, the current size of workload165A is 2+2+1+4=9. This current size is compared to the target size range [14, 18]. Since the current size is smaller than the target size range, the converging process continues to the next iteration.

The second iteration begins with split point iteration155B. Starting with step 2, since the current size is smaller than the target size range, the particular list is the list with the smallest split point, or list 3 with a split point at index 1. Thus, the candidate split point for list 3 is set using the converging formula i=┌(i+M/N)/2┐ or i=┌(1+16/4)/2┐=┌5/2┐=┌2.5┐=3. This moves the split point for list 3 forward from index 1 to index 3 having a value of 33, indicated in bold in split point iteration155B.

Next, in step 3, the split points are found for all other lists, or list 1, list 2, and list 4. Since the candidate split point has a value of 33, the split points are chosen such that values prior to the split points do not exceed 33. Thus, as shown in split point iteration155B, the split point for list 1 is prior to index 3 (29≦33), the split point for list 2 is prior to index 9 (29≦33), and the split point for list 4 is prior to index 7 (27≦33). These split points are indicated by the | or pipe symbol in lists 1, 2 and 4, with the largest values not exceeding 33 indicated in bold.

A current size of workload165A is determined by summing the number of elements prior to the split point of each of the ordered lists. Thus, the current size of workload165A is 3+9+3+7=22. This current size is compared to the target size range [14,18]. Since the current size is larger than the target size range, the converging process continues to the next iteration.

The third iteration begins with split point iteration155C. Starting with step 2, since the current size is larger than the target size range, the particular list is the list with the largest split point, or list 2 with a split point at index 9. Thus, the candidate split point for list 2 is set using the converging formula i=┌(i+M/N)/2┐ or i=┌(9+16/4)/2┐=┌13/2┐=┌6.5┐=7. This moves the split point for list 3 backwards from index 9 to index 7 having a value of 26, indicated in bold in split point iteration155C.

Next, in step 3, the split points are found for all other lists, or list 1, list 3, and list 4. Since the candidate split point has a value of 26, the split points are chosen such that values prior to the split points do not exceed 26. Thus, as shown in split point iteration155C, the split point for list 1 is prior to index 2 (18≦26), the split point for list 3 is prior to index 1 (13≦26), and the split point for list 4 is prior to index 6 (20≦26). These split points are indicated by the | or pipe symbol in lists 1, 3 and 4, with the largest values not exceeding 26 indicated in bold. Note that in the case of list 2, the values prior to the split point include a value equal to the split point, or 26.

A current size of workload165A is determined by summing the number of elements prior to the split point of each of the ordered lists. Thus, the current size of workload165A is 2+7+1+6=16. This current size is compared to the target size range [14, 18]. Since the current size is within the target size range, the converging process finishes. However, even if the current size was outside the target size range, the converging process would still finish since the maximum number of iterations has been reached (MaxIterations=3).

With the first set of split points now known as indicated in split point iteration155C, the contents of workload165A can be defined from subsets of each of ordered lists162, wherein the subsets have index ranges starting from the beginning offset up to the split point of each list. Since workload165A is the first workload, the beginning offset is 0 for each list. Thus, workload165A is populated as shown inFIG. 2C, with list 1 subset having an index range of 0 to 1 inclusive (2 elements), list 2 subset having an index range of 0 to 6 inclusive (7 elements), list 3 subset having an index range of 0 to 0 inclusive (1 element), and list 4 subset having an index range of 0 to 5 inclusive (6 elements). Since workload165A is the first workload, there is no lower bound to the possible values, but the upper bound is 26, or the value of the final split point candidate at list 2, or index 7. Thus, the range of possible values in workload165A is indicated by [−∞, 26].

Iterative Split Point Converging—Successive Workloads

As discussed above, once steps 1-4 have been completed for a particular workload, step 5 may be carried out to continue with the next workload. Accordingly, the beginning offset of each sorted list is moved forward to the split point of each sorted list. Thus, the beginning offsets of lists 1-4 are set to 2, 7, 1, and 6, respectively. This is shown in split point iteration155D ofFIG. 2E, wherein list 1 now starts at index 2 with a value of 29, list 2 now starts at index 7 with a value of 26, list 3 now starts at index 1 with a value of 30, and list 4 now starts at index 6 with a value of 27. Accordingly, the completed workload is removed from consideration when determining the next set of split points.

The above process of steps 1-4 as described with workload165A is carried out for workload165B, as shown in split point iteration155D, split point iteration155E, and split point iteration155F ofFIG. 2E. For brevity, a full discussion of each iteration step is omitted. In this example, the current size never reaches the target size range of [14, 18], but the process nevertheless ends after 3 iterations in accordance with MaxIterations. The range of possible values in workload165B is indicated by [26, 40]. Step 5 is again applied to move the beginning offsets forward to the second set of split points, and the process is repeated for workload165C.

As shown inFIG. 2F, workload165C is determined after only two iterations, or split point iteration155G and split point iteration155H. The range of possible values in workload165C is indicated by [40, 55]. Since workload165C is the penultimate workload, workload165D is simply the leftover elements after the split points in split point iteration155H. The range of possible values in workload165D is indicated by [55, ∞], since workload165D is the final workload with no upper bound of values. Accordingly, all of the workloads165A-165D in balanced workloads164have been determined, thereby completing block206in process200ofFIG. 2A.

Merging the Balanced Workloads

At block208of process200, referring toFIG. 1, server node120sorts each of the N balanced workloads164in parallel. Referring toFIG. 2G,FIG. 2Gis a block diagram that depicts a process for parallel merging and combining the N balanced workloads164, according to an embodiment.FIG. 2Gincludes parallel merge157, combine158, workload165A, workload165B, workload165C, workload165D, sorted workload167A, sorted workload167B, sorted workload167C, sorted workload167, and sorted result168. With respect toFIG. 2G, workloads165A-165D may correspond to workloads165A-165D fromFIG. 2C, sorted workloads167A-167D may correspond to sorted workloads166fromFIG. 1, and numbered elements may correspond to like numbered elements fromFIG. 1.

As shown inFIG. 2GandFIG. 1, workloads165A-165D are sorted by parallel merge156using respective threads134A-134D to output sorted workloads167A-167D, which may correspond to sorted workloads166ofFIG. 1. Since the subsets within each workload165A-165D are already sorted with respect to each subset, the second phase using parallel merge156may utilize a faster sorting algorithm compared to the first phase using parallel sort152. For example, parallel merge156may utilize a method that takes advantage of the ordered properties of the subsets in each workload, such as by utilizing a K-way mergesort or a 2-way mergesort.

Combining the Sorted Workloads

At block210of process200, referring toFIG. 1, server node120combines the sorted workloads166to output sorted result168. Referring toFIG. 2C, each successive workload of workloads165A-165D includes adjacent value ranges without any overlap (besides the edge bounding values). Accordingly, the sorted workloads167A-167D will also have this property of non-overlapping adjacent value ranges. As a result, combine158may use a simple concatenation of sorted workloads166, or sorted workloads167A-167D, to generate the final sorted result168, as shown inFIG. 2G. The sorted result168may then be returned to application112to answer the original SQL SELECT query, duly sorted according to the GROUP BY clause. Application112may utilize sorted result168to output to a display, to create an output file or report, or to use for further processing and analysis.

Process200is thus complete, with the most time consuming steps of parallel sort152and parallel merge156carried out fully in parallel. An optimized utilization of all available threads is achieved regardless of the particular data distribution or skew of unsorted data elements160by ensuring approximately equal workloads, or unordered lists161and balanced workloads164, for each respective parallel processing step, or parallel sort152and parallel merge156. Additionally, the other steps of split151, split point converge154, and combine158can be carried out quickly without any expensive data scan operations. Thus, process200can provide efficient scaling for the massive number of elements and the large amount of threads demanded by the most data intensive applications, such as enterprise grade databases.

Hardware Summary

For example,FIG. 3is a block diagram that illustrates a computer system300upon which an embodiment of the invention may be implemented. Computer system300includes a bus302or other communication mechanism for communicating information, and a hardware processor304coupled with bus302for processing information. Hardware processor304may be, for example, a general purpose microprocessor.

Computer system300further includes a read only memory (ROM)308or other static storage device coupled to bus302for storing static information and instructions for processor304. A storage device310, such as a magnetic disk or optical disk, is provided and coupled to bus302for storing information and instructions.

Computer system300may implement the techniques described herein using customized hard-wired logic, one or more ASICs or FPGAs, firmware and/or program logic which in combination with the computer system causes or programs computer system300to be a special-purpose machine. According to one embodiment, the techniques herein are performed by computer system300in response to processor304executing one or more sequences of one or more instructions contained in main memory306. Such instructions may be read into main memory306from another storage medium, such as storage device310. Execution of the sequences of instructions contained in main memory306causes processor304to perform the process steps described herein. In alternative embodiments, hard-wired circuitry may be used in place of or in combination with software instructions.

The received code may be executed by processor304as it is received, and/or stored in storage device310, or other non-volatile storage for later execution.