Inverse distribution function operations in a parallel relational database

Inverse distribution operations are performed on a large distributed parallel database comprising a plurality of distributed data segments to determine a data value at a predetermined percentile of a sorted dataset formed on one segment. Data elements from across the segments may be first grouped, either by partitioning keys or by hashing, the groups are sorted into a predetermined order, and data values corresponding to the desired percentile are picked up at a row location of the corresponding data element of each group. For a global dataset that is spread across the database segments, a local sort of data elements is performed on each segment, and the data elements from the local sorts are streamed in overall sorted order to one segment to form the sorted dataset.

BACKGROUND

This invention relates generally to analyzing the data population of a dataset to determine information that characterizes the data population, and more particularly to determining the data values at predetermined percentiles of a data population that is distributed across multiple nodes of a distributed parallel database.

It is frequently desirable to characterize the data in a data population in order to better understand the nature of the data. Important characteristics of the data population include data values which occur at certain percentile levels. For example, determining data values at the median (50thpercentile), the 90thpercentile, or the 99thpercentile levels is important, especially for financial data as to satisfy legal reporting and regulatory requirements, because percentile values allow insight into the underlying data and permit the data to be summarized meaningfully. Percentile values are determined using inverse distribution functions which are different from other types of mathematical calculations that characterize a data distribution as they produce the actual real data values in the data distribution at desired percentiles. The median, for instance, of a data distribution is different from the average because it produces the real value of the middle data element in the distribution. Moreover, it is unaffected by an outlying value that could significantly skew the average value.

While performing inverse distribution operations to determine the data values at selected percentiles on a small dataset is relatively straightforward, doing so on a large parallel database where the data is distributed across clusters of multiple computers is exceptionally difficult. This is because there is an ordering constraint upon the data population which requires getting the data into a particular order before percentile levels can be determined. It is not possible to calculate inverse distribution functions in parallel on separate subsets of data and combine the answers in a way to derive correct results for the overall dataset. The median of a data distribution, for instance, is not equal to the median of medians. Ordering of the data in a large distributed parallel database has not generally been possible in a way that does not revisit the data multiple times or require massive movements of large amounts of data. Accordingly, known approaches to performing inverse distribution function operations on parallel databases are inefficient, costly and difficult.

It is desirable to provide systems and methods which address the foregoing and other known problems of characterizing distributed datasets by enabling inverse distribution operations to determine data values at selected percentile levels of a data population that is distributed across multiple nodes of a parallel database system efficiently and cost effectively, and it is to these ends that the present invention is directed.

DESCRIPTION OF PREFERRED EMBODIMENTS

The invention is particularly well adapted for characterizing the data distribution of a dataset in a large parallel distributed relational database, such as a shared nothing database system, using inverse distribution function processes to determine data values at selected percentile levels, and will be described in that context. It will be appreciated, however, that this is illustrative of only one utility of the invention, and that the invention may be employed with other types of database systems and for other types of operations.

FIG. 1illustrates the architecture of a shared nothing database system100comprising one type of a parallel distributed database system100in which the invention may be employed. The database may include a master node102which connects to a plurality of segment nodes104_A through104_N. Each segment node may comprise one or more database (DB) segments (database instances), including one or more primary databases and one or more mirror databases. For fault tolerance purposes, a primary database segment and its corresponding mirror database segment may be located on different nodes. The master and segment nodes may comprise generally similar server applications having the same process model as a standalone server that is augmented with extensions for a distributed system, such as data distribution, remote process communications, and data replication between primary-mirror pairs.

FIG. 2illustrates an example of the architecture of a node202of the database system that is configured to perform processes and operations in accordance with the invention. The master node and the segment nodes may have similar architectures. The node202may comprise a host computer server system210(which may comprise a single CPU or a multi-processor system comprising a plurality of CPUs) connected to input/output (I/O) devices212by a bus214. The I/O devices may be standard computer system input and output devices. One or more network interface circuits216may also be connected to bus214to allow the node to operate in the networked environment of the parallel distributed database system ofFIG. 1. The node may further have storage220comprising non-transitory physical storage media connected to the bus that stores executable instructions to control the operations of the computer system. Storage220may include a main memory222comprising instructions that control the CPU to operate in accordance with the invention, as will be described, and may contain other storage224for storing a database instance.

SQL databases provide a standard process “percentile_cont” that can be used to apply an inverse distribution function operation to a dataset to determine the data value at a predetermined percentile level (P) in an ordered set of data comprising N rows, i.e., records.FIG. 3illustrates diagrammatically the inverse distribution function operation. A row number RN of a row in an ordered set of data containing the data value corresponding to the desired percentile level P is determined according to the relationship RN=(N−1)*P+1, and the data value V at row RN is determined by linear interpolation between the values in a row above and a row below RN if RN falls between rows.FIG. 3illustrates the process for determining the data value at the P=0.5 (50thpercentile) level, which is known as the median. The median is the data value at the middle of an ordered set of data.FIG. 3illustrates determining the median value of sales within a dataset of branch sales data as shown in table300. The first step is to determine the number N of rows in the dataset. In the example ofFIG. 3, N=4. At step 2 the data is sorted by sales to produce a sort ordered dataset310, and at step 3 row number RN of the median value of sales is determined using the above relationship. In the example shown, RN=2.5, which occurs midway between the second and third rows in table310. Accordingly, the median data value V is determined by linear interpolation, as shown, to be 4900. In the description that follows, for convenience and simplicity, and without the loss of generality, the invention will be described with respect to analyzing a dataset for the “median” or 50thpercentile level. It will be appreciated, however, that the invention is equally applicable to determining data values at any other desired percentile levels.

While the process illustrated inFIG. 3for determining a percentile value is relatively straightforward for a dataset that is centrally located in a one database, it is difficult to apply inverse distribution function processes in a large distributed parallel database system where the data is distributed across multiple computing clusters because of the need to first sort the relevant dataset into a predetermined order. In a large distributed database system, this may require moving massive amounts of data between clusters, which is very costly and inefficient. The invention addresses this problem by providing very efficient and relatively low cost processes for performing inverse distribution operations on a distributed parallel database.

The invention, broadly summarized, determines the number of elements (rows or values) in a relevant dataset, collects or sorts the data into a prescribed order, and determines the data value(s) at the element or row numbers of the ordered data corresponding to the desired percentile(s), as by counting the number of elements or rows into the ordered dataset and picking up the data values at the appropriate row numbers. Depending upon the nature and characteristics of the data, and the information desired, the invention may accomplish this in somewhat different ways, as will be described.

FIG. 4illustrates a process which a database user may perform preliminarily for selecting an appropriate approach for analyzing a dataset based upon the characteristics of the dataset and the information desired. At400, the user may first determine whether no data grouping (402) or whether data grouping (404) will be used. By “grouping” or “grouped” as used herein is meant whether the data is organized into a plurality of separate groups (datasets), each of which is characterized by an inverse distribution process. No grouping means the dataset comprises one large “global” group which is analyzed and characterized as a unit. Whether or not grouping is used may depend upon the nature of data and the query. A query that seeks the median of sales worldwide, for example, will analyze all of the sales data as one large global group, whereas a query that seeks the medians of sales in different regions will analyze separate groups (datasets), one for each region, and obtain a median for each group.

Next, at406and408, determinations may be made as to whether the records in the datasets contain duplicate values. If they contain a significant numbers of duplicates, a de-duplication process (which will be described below in connection withFIG. 8) may be employed to reduce the number of duplicate records before analyzing the dataset. Because the runtime of a database query is determined by the number of records involved, decreasing the number of records decreases the runtime and associated costs, and improves efficiency. In the non-grouping case illustrated inFIG. 4, the global dataset may be analyzed in either a “naïve” (meaning no substantial number of duplicates) pickup global process410or in a de-duplicated pickup global process412. In the grouping case illustrated inFIG. 4, each group may be considered separately and analyzed either as a naïve group (414) or a de-duplicated group (422). Before analysis, however, a determination may be made as to whether the groups are small or large at416and424. Depending upon whether the size of a group, different pickup processes may be used. For the naïve groups, if the groups are large a naïve pickup hash analysis process is preferably employed, as shown at418, whereas if the groups are small a naïve pickup sort analysis process may be employed, as indicated at420. Similarly, for de-duplicated groups, a de-duplicate pickup hash analysis process may be employed for large groups (426), and a de-duplicate pickup sort analysis process may be employed for small groups (428). Preferred embodiments of pickup global, pickup hash, and pickup sort processes in accordance with the invention will be described below.

FIG. 5illustrates a non-group naïve pickup global process410in accordance with a preferred embodiment of the invention. As described above, in a global process one large dataset or group is analyzed. In order to obtain the median, the data values of the records comprising the global dataset must be placed in sorted order, the number of records or rows in the dataset must be counted, and the data value at the designated row number RN selected or determined by interpolation to provide the desired median value. However, in a large parallel distributed database with millions or billions of data records distributed across multiple clusters it has been impractical, inefficient and costly to perform these operations on one large consolidated dataset. A pickup global process in accordance with the invention as illustrated inFIG. 5avoids these difficulties.

As shown inFIG. 5, in an embodiment of the pickup global process in accordance with the invention, a first portion of the process is performed in each of the segments of the database cluster, and a second portion of the process is performed by the master host of the database cluster. The first and second portions of the process that are performed in the segments and the master are indicated by the separating dashed line inFIG. 5. In a first segment of the database, a local sort is performed at502on a relevant dataset in a table500to place the relevant data into sorted order, e.g., numerical order. Simultaneously with performing the local sort, a partial aggregate or count of the number of data elements in the local dataset on the first segment may be performed in parallel at504as the data is sorted. The sorted local data may be supplied to a gather merge module510on the master node. Separately, a partial aggregate or count of the number of data elements or rows in the sorted data may be produced at504and supplied to another gather module512on the master. The gather merge module510receives the sorted local data from the first segment and may place it into a first container on the master corresponding to the first segment. The gather module512may likewise gather or collect on the master the partial aggregate or count of the number of data elements of the sorted data in the first container. These steps may then be repeated for each of the other segments in the database cluster, resulting in a plurality of sort ordered local datasets in separate containers on the master with counts of the numbers of data elements in each dataset. A second aggregate process514may be performed on the master to sum the various counts from the gather module512for the plurality of locally sorted datasets in the gather module510to give the total number of data elements or rows in the global dataset comprising a plurality of local datasets from the various segments. This total may be supplied to a join module520on the master.

After gathering the local data sorts from the segments, the gather merge module510streams a sort ordered global dataset to join module520by selecting data elements in sorted order from the sorted local datasets in the various containers. Since each of the local datasets has already been placed in sorted order in its respective container, the merge function of module510need only scan the plurality of containers and select the next sequentially ordered data element in the global dataset, e.g., the next largest data value for a dataset ordering of smallest to largest values, and stream the selected data element to the join module520. The join module520supplies the total count of data elements or rows in the global dataset from the second aggregate process514to the pickup global module522along with the ordered data elements streamed by the merge operation of module510. The pickup global module may pick up the data value at the row number RN corresponding to the middle row of the ordered global dataset to supply the median value. Since the count of the total number of rows in the global dataset is known as a result of the second aggregate process514, the row number for the median value is readily determined. Accordingly, once the pickup global process522receives the data value corresponding to the calculated row number RN for the median (or the adjacent rows where RN falls between rows), the pickup global process can supply the answer for the median value without needing to receive the remaining half of the ordered global dataset. Thus, the process can be terminated, thereby reducing processing costs by at least one-half and improving efficiency.

For percentile levels other than the median, even greater efficiencies may be achieved. The 90thpercentile level, for example, seeks the data value that is equal to or greater than 90% of the values in the dataset. This may be determined either by counting upwards from the bottom row (smallest value) in the ordered data set to the row that is 90% of the way to top of the dataset, or by counting down from the top 10% of the way to the bottom. Accordingly, by streaming the data in inverse order from the highest value to the lowest, and knowing the total number of rows, data streaming may be terminated at the correct row after streaming only 10% of the data.

FIG. 6is an overview of an embodiment of a pickup hash process in accordance with the invention. In this process, values in a dataset may first be partitioned into groups using queries having partitioning keys to produce partitioned data600. The data may be then subjected to a hashing process610which stores the data values in hash buckets or containers for each group based upon the partitioning keys. A small sort630may be performed of each group to produce a sorted group dataset while simultaneously counting the number of values in the group. The sorted groups may then be streamed to a pickup hash operation620which uses the count and the previously described inverse distribution operation to pick up a data value corresponding to a desired percentile level.FIG. 6illustrates the pickup hash process for three groups. As shown, unordered values 3, 12, 10 and 5 of Group 1 may be placed in a corresponding hash bucket by process610as tuples (group_key1, valueV), and the hash bucket values may be sorted to produce the sort ordered values 3, 5, 10 and12for the group, which may be streamed to the pickup hash operation620for selection of the value at the desired percentile level. The process may then be repeated for the remaining Groups 2 and 3, as indicated. This results in three data values, one for each group, corresponding to the desired percentile level for each group.

FIG. 7illustrates an overview of an embodiment of a pickup sort process in accordance with the invention. In the pickup sort process, records are sorted across groups by a local sort process710, and streamed to a pickup sort operation720. A full dataset730may be sorted to produce a dataset740sorted by group, as shown. The sorted dataset of a group, e.g., Group 1, may then be scanned to determine a count of the number of data elements in the group, as shown at step 1 in the figure. In step 2, the process may return to the beginning of the group, and in step 3 the group may be scanned forward to identify the designated row number, RN, which may be picked up by the pickup sort process720. At step 4, the process fast forwards to the next group, i.e., Group 2, and repeats.

As shown inFIGS. 6 and 7the pickup hash and the pickup sort processes are somewhat similar. They differ in that the pickup hash first hashes data into groups before sorting, and is better suited to a large number of groups since each sort is more likely to fit in memory. In contrast, the pickup sort across groups directly is better suited to a smaller number of groups. If values in a group are duplicated, efficiencies can be gained in any of the grouped and non-grouped pickup processes of the invention by compressing streamed data in a de-duplication process using run length encoding. Run length encoding can significantly reduce the amount of data that is transported and the cost of a runtime operation. Run length encoding may be performed on groups in parallel, where identical values may be packed into one value without moving data, and the results of the compression may be merged after moving.

FIG. 8illustrates the results of an embodiment of a de-duplication process in accordance with the invention. First, a dataset800may be organized into groups using queries and partitioning keys to produce tuples (key1, value1), (key1, value2), . . . (keyN, valueM), where the keys correspond to the groups and the values are the data values in the group. Running counts of identical data values in each group may be produced as the dataset800is formed, and dataset800may then be compressed into a smaller dataset810, where each record (key1, value1, count1), (key1, value2, count2), . . . (keyN, valueM, countNM) of dataset810indicates its group, its value, and a count of the number of identical values in the group. As shown, compressed dataset810is substantially smaller than dataset800. Accordingly, it requires less time, network bandwidth and cost to move dataset810across a network and process it. For a dataset containing, for example, millions of records of which only several thousands have unique values, the efficiencies and savings can be substantial.

FIG. 9illustrates in detail a pickup hash process in accordance with the invention that embodies de-duplication. Process modules918,920and922correspond generally to process modules610,620and630, respectively ofFIG. 6, and process modules912,914and916comprise the de-duplication process. The process operations illustrated inFIG. 9are performed on the different segments, and the results are gathered by a gather process924at the master node. Referring toFIG. 9, a dataset in the form of a plurality of tables910may be distributed across different segments. Each segment may perform a partial aggregation912that collects the data into various groups on each segment and may simultaneously count the number of elements in each group. At914, the segments may redistribute the partially aggregated data from the tables and counts to different segments in order to consolidate data of a group from across the different segments onto one segment. This has the advantage of putting all of relevant data for each group together on one segment. Each segment may then perform a second aggregate operation916to sum the group counts from the various segments to obtain a total count of the group. At918, a hashing operation, as previously described, is performed to put records for the same group in the same container or hash bucket, and pickup hash and small sort processes as previously described may be performed at920and922to sort the values in each hash bucket or group and to pick up the values of the sorted data of each group. The values picked up on the various segments comprising the desired percentile values of each group are gathered at924on the master segment, from which they may be reported.

FIG. 10illustrates in detail a pickup sort process in accordance with the invention that embodies de-duplication. The pickup sort process ofFIG. 10is similar to the pickup hash process ofFIG. 9, except that the pickup hash process first hashes data in a dataset according to group. As with the pickup hash process, portions of the pickup sort process may be performed on different segments.

Referring toFIG. 10, a dataset comprising a plurality of tables1010may be distributed across different segments, and operations1012(partial aggregate),1014(redistribute) and1016(second aggregate) comprising the de-duplication process which may be substantially the same as described in connection withFIG. 9may be performed. Unlike the pickup hash process which hashes data into groups, the local sort operation1018sorts the entire dataset across each group in a manner similar to that described inFIG. 7, and the pickup sort process1020picks up the results for each group in a streaming way and supplies the results to a gather process1022on the master node which may report the results. As described above in connection with the pickup global process, once the master receives the data value corresponding to the desired percentile, the process may terminate.

Finally,FIG. 11illustrates an embodiment of the pickup global process ofFIG. 5that includes a de-duplication process (elements1102,1104and1106) such as described in connection withFIGS. 9 and 10. The remaining operations1108,1110,1114,1116,1118,1120, and1122illustrated in the figure may be substantially the same as the operations502,510,504,512,514,520, and522, respectively, described above in connection withFIG. 5.

From the foregoing it can be seen that the inverse distribution operations of the invention for both group and non-grouped datasets as well as for large and small datasets all involve placing the data values of a dataset into sorted order, determining the row number of the row containing the data value corresponding to the desired percentile level, and picking up the data value using a pickup operation to supply the answer. The invention is unique and highly advantageous in the manner in which it derives the information to characterize a dataset of a large parallel distributed database by arranging data from across multiple segments of a distributed parallel database cluster into small groups at single locations, to sort order of the data and to pickup the relevant data values, or otherwise picking up one or more relative values as data is streamed from distributed segments to a single location. Thus, the invention allows characterization of a large distributed dataset to be performed efficiently and cost-effectively.

While the foregoing has been with respect to preferred embodiments of the invention, it will be appreciated by those skilled in the art that changes in these embodiments may be made without departing from the principles and spirit of the invention, the scope of which is defined in the appended claims.