Multi-level partitioned database system

[00002] Data organization is important in relational database systems that deal with complex queries against large volumes of data. Relational database systems allow data to be stored in tables that are organized as both a set of columns and a set of rows. Standard commands are used to define the columns and rows of tables and data is subsequently entered in accordance with the defined structure. The defined table structure is logically maintained, but may not correspond to the physical organization of the data. For example, the data corresponding to a particular table may be split up among a number of physical hardware storage facilities.

[00003] Users of relational database systems require the minimum time possible for execution of complex queries against large amounts of data. Different physical types of storage, for example random access memory and hard drives, incur different length delays. In addition, writing to memory or a hard drive is often slower than reading an equivalent amount of data from memory or a hard drive. The organization of data corresponding to tables defined in a relational database system may determine the number of writes and reads that need to be performed in order to execute a common query. In some situations, particular portions of the data in a table are searched more often than other portions. If the data is properly organized, performance can be improved by searching a part of the data for queries that can take advantage of that organization. If the data is not organized in any way, it will often need to be searched in its entirety to satisfy a query or copied and restructured into a useful organization.

[00004] In general, in one aspect, the invention features a database system with multi-level partitioning. The system includes a plurality of storage facilities with each storage facility storing data that represents rows of database tables. In each storage facility table rows corresponding to a specific table are logically ordered according to a row identifier, also called the row ID. The row ID includes a first value that is based on one or more columns of the table. The row ID also includes a second value that is based on one or more columns of the table. The row ID also includes a third value. The first value is calculated based on a single-level function. The single-level function is a combination of a plurality of functions. The single-level function produces different values for two rows when any of the plurality of functions produces different values for the two rows. The first value of the row ID is predominate in determining the order of the rows in the storage facilities. The second value determines the order of those rows with identical first values. If any rows have identical first and second values, the third value determines the order of those rows.

[00005] In a more specific implementation of the invention, the plurality of functions are combined into the single-level function using the following formula: P=( . . . ((P₁−mn₁)*(mx₂−mn₂+1)+P₂−mn₂)*(mx₃−mn₃+1)+ . . . +P_(n−1)−mn_(n−1))*(mx_n−mn_n+1)+P_n−mn_n) where P is the single-level function, P_i is the ith one of the plurality of functions, mx_i and mn_i are the maximum and minimum values of P_i, respectively, the initial ellipses represents a number of left parentheses necessary to match each right parenthesis that is included for each of the plurality of functions as shown above, and the second ellipses repeats the pattern to include a term for each combined function.

[00006] The multi-level partitioned database system has the advantage of organizing the data stored by the system. For example, this organization allows access in multiple ways via the multiple partitioning expressions. For instance, if a query gives values for columns of the second partitioning function but not the first, a scan of rows is only needed in each of the first level partitions but only for the designated second level partition within each of those first level partitions. Other features and advantages will become apparent from the description and claims that follow.

[00012] The partitioned table storage technique disclosed herein has particular application, but is not limited, to large databases that might contain many millions or billions of records managed by a database system (“DBS”) 100, such as a Teradata Active Data Warehousing System available from NCR Corporation. FIG. 1 shows a sample architecture for one node 105₁ of the DBS 100. The DBS node 105₁ includes one or more processing modules 110_{1 . . . N}, connected by a network 115, that manage the storage and retrieval of data in data-storage facilities 120_{1 . . . N}. Each of the processing modules 110_{1 . . . N} may be one or more physical processors or each may be a virtual processor, with one or more virtual processors running on one or more physical processors.

[00013] For the case in which one or more virtual processors are running on a single physical processor, the single physical processor swaps bet ween the set of N virtual processors.

[00014] For the case in which N virtual processors are running on an M-processor node, the node's operating system schedules the N virtual processors to run on its set of M physical processors. If there are 4 virtual processors and 4 physical processors, then typically each virtual processor would run on its own physical processor. If there are 8 virtual processors and 4 physical processors, the operating system would schedule the 8 virtual processors against the 4 physical processors, in which case swapping of the virtual processors would occur.

[00015] Each of the processing modules 110_{1 . . . N} manages a portion of a database that is stored in a corresponding one of the data-storage facilities 120_{1 . . . N}. Each of the data-storage facilities 120_{1 . . . N} includes one or more disk drives. The DBS may include multiple nodes 105_{2 . . . N} in addition to the illustrated node 105₁, connected by extending the network 115.

[00016] The system stores data in one or more tables in the data-storage facilities 120_{1 . . . N}. The rows 125_{1 . . . Z} of the tables are stored across multiple data-storage facilities 120_{1 . . . N} to ensure that the system workload is distributed evenly across the processing modules 110_{1 . . . N}. A parsing engine 130 organizes the storage of data and the distribution of table rows 125_{1 . . . Z} among the processing modules 110_{1 . . . N}. The parsing engine 130 also coordinates the retrieval of data from the data-storage facilities 120_{1 . . . N} in response to queries received from a user at a mainframe 135 or a client computer 140. The DBS 100 usually receives queries and commands to build tables in a standard format, such as SQL.

[00017] The rows 125_{1 . . . Z} are distributed across the data-storage facilities 120_{1 . . . N} by the parsing engine 130 in accordance with their primary index. The primary index defines the columns of the rows that are used for calculating a hash value. The function that produces the hash value from the values in the columns specified by the primary index is called the hash function. Some portion, possibly the entirety, of the hash value is designated as a hash bucket. The hash buckets are assigned to data-storage facilities 120_{1 . . . N} and associated processing modules 110_{1 . . . N} by a hash bucket map. The characteristics of the columns chosen for the primary index determine how evenly the rows are distributed.

[00018] FIG. 2 shows the rows of a table be distributed. The table 200 contains a plurality of rows and is stored in a plurality of data storage facilities 120_1-4 by the parsing engine 130 (not shown). For example, two columns 210, 220 can be designated as the primary index when the table is created. The hash function is then applied to the contents of columns 210, 220 for each row. The hash bucket portion of the resulting hash value is mapped to one of the data storage facilities 120_1-4 and the row is stored in that facility. For example, if the primary index indicates a column containing a sequential row number and the hash function is the sum of the remainder when the row is divided by four and the value one, the first eight rows will be distributed as shown in FIG. 2.

[00019] Queries involving the values of columns in the primary index can be efficiently executed because the processing module 110_n having access to the data storage facility 120_n that contains the row can be immediately determined. For example, referring to FIGS. 1 and 2, if values from row 2 are desired, the parsing engine 130 can apply the hashing function to determine that only processing module 110₂ need to be used to access storage facility 120₂. As another example, an equality join between two tables that have the same primary index columns is very efficient. All of the rows that need to be joined are found in the same data storage facility 120_n and no movement of information between the facilities is necessary.

[00020] While the primary index of a table can be chosen for equality joins, for example the order number column of an order table, additional design features can make searches on other columns, for example the date column of the order table, more efficient. Referring to FIG. 3A, an example table 300 is shown. The table 300 includes six columns. In this example, the user designates a first column 302 as the primary index. The user also expects to perform numerous searches corresponding to a particular function of a second column 306 and another function of a third column 304. The table 300 is created with a command that includes a specification of the primary index, as the first column 302, a first partition function that is evaluated using values from the second column 304, and a second partition function that is evaluated using values from the third column 306. Functions are also referred to as expressions, but will be referred to as functions herein. One possible partition function would be a range function. If the second column 304 contained dates between 1996 and 2001, the partition function could be the difference between the year of the date and 1995, resulting in values of 1 through 6. A single partition function, like a primary index, can include more than one column. For example, a partition function could return the difference between the values in two date columns.

[00021] In one implementation, the structure of table 300 has been defined by a command to include a primary index 302 and two different partition functions. The partition functions each including one column, the second column 304 and the third column 306, respectively. The partition functions are combined into a function referred to herein as a single-level partition function such that effectively rows are ordered by the value of first partition function and within such a partition are ordered by the second function. In one implementation, the single-level partition function will produce different values for row data whenever that row data would have produced different values from any of the two or more partition function that were combined to produce the single-level partition function. One formula for generating a single-level partition function, P, from n partition functions is included below wherein P_i represents the i^th partition function entered in the table definition that uses values from one or more columns. The variables mn_i and mx_i represent the minimum and maximum resulting values of P_i, respectively.
P=( . . . ((P₁−mn₁)*(mx₂−mn₂+1)+P₂−mn₂)*(mx₃−mn₃+1)+ . . . +P_(n−1))*(mx_n−mn_n+1)+P_n−mn_n)  [10]
The initial ellipses represents a number of left parentheses necessary to match each right parenthesis that is included for each partition function as shown above. The second ellipses repeats the pattern to include a term for each specified partition function. For example, a table created with three partition functions, P₁, P₂, and P₃, would have the following single-level partition function, P.
P=(((P₁−mn₁)*(mx₂−mn₂+1)+P₂−mn₂)*(mx₃−mn₃+1)+P₃−mn₃)
As a further example, the three partition functions are: P₁(date)=month(date); P₂ (date)=year(date); and P₃ (product number)=category(product #). The dates range from 1995 to 2001 and there are 8 categories of product numbers, 1 through 8. A single-level partition function, P, is then calculated.
P(date, product #)=(((month(date)−1)*(2001-1995+1)+year(date)-1995)*(8-1+1)+category(product #)−1)
P(date, product #)=(((month(date)−1)*7+year(date)-1995)*8+category(product #)−1)
Other formulas can also be used to generate a single-level partition function from the initial partition functions. For example, if partition number zero is reserved, the value one can be added to the above formula to shift the results up by one. In some implementations, partition number zero is reserved for use by the system.

[00029] In FIG. 3B, a multi-level partitioned database storage facility 120₁ is shown. Rows 125_1-15 are allocated to the storage facility 120₁ as discussed above with respect to FIG. 2. The rows 125_1-15 are organized within the storage facility 120₁ in accordance with partitions and the hash value of the primary index. In one implementation, a uniqueness value is assigned to rows to order rows that have identical partition values and hash values. The rows 125_1-15 stored in the storage facility 120₁ are ordered at a top level by the results of the defined partition functions. Two first level partitions are shown 310, 312. For each first level partition 310, 312, two second level partitions are shown 320, 322, 324, 326. For each first level partition, for example first level partition 310, and each second level partition, for example second level partition 322, two third level partitions are shown, for example third level partitions 334, 336. In other words, rows 125_6-8 have identical results for all three partitions functions when they are applied to the values in those rows. On the other hand, row 125₅ has the same result for the first two partition functions as rows 125_6-8, but a different result for the third partition function.

[00030] While the table is created with separate partition functions resulting in a multi-level partition, the partitions can be implemented using a partition function calculated from the separate partition functions as described in the above-disclosed formula or other similar formulas. In one implementation, that change from multi-level to single level partitioning is not revealed to the user. The table definition, when viewed by the user retains the separate functions for each partition level, but the formula is used to convert queries addressing particular partitions of one of the multiple levels to partitions of the single level.

[00031] In one implementation, a row identification (row ID) is stored for each of the rows 125_1-15. The row ID includes a portion that identifies the partition in which that row is located. That partition is determined using the combined partition function, P, calculated from the partition functions specified in the table definition. The length of the partition portion of the row ID determines the number of partitions that can be specified. The total number of partitions for a multi-level partitioning definition is the product of the number of partitions at each level added to the number of reserved partitions. For example, the three partition functions used to illustrate the combination formula above have 12 (the number of months), 7 (the number of possible years), and 8 (the number of product categories) partitions each. If partition zero is reserved, the resulting single-level partition function would require (12*7*8)+1 or 673 partitions. A one-byte partition portion of the row ID would only allow 256 unique partitions and would be inadequate. If the row ID includes two bytes to identify the row ID, the 673 required partitions are easily covered by the 65536 possible values. Some implementations of the invention do not include use of a row ID.

[00032] The rows are also ordered within each partition. For example, the partition 342 contains four rows 125_11-14. Those rows 125_11-14 are stored within that partition 342 in the order of the hash result for each row. In one implementation, the hash result is included in a row ID with the partition in which the row is located. A uniqueness value is also maintained for each row. No two rows with the same partition and hash value in a table can have the same uniqueness value. The uniqueness values are determined when the rows are added to the table. For example, a sequential number (the next uniqueness number after the highest one currently being used) or any currently unused number can be used as the uniqueness value. In one implementation, a uniqueness value is included in a row ID with a hash value and a partition number. For example, a ten-byte row ID can include two bytes for the partition value, four bytes for the hash value, and four bytes for the uniqueness value. If two rows are in the same partition and have the same hash value, their order is determined by their uniqueness values, which by definition cannot be identical. The uniqueness value does not play a role in ordering rows that have different partition or hash values. A larger number of possible hash values increases the number of data storage facilities that can be used to store rows of the database and reduces the chance that multiple rows will have the same hash value. A larger number of possible uniqueness values allow tables with larger numbers of rows with the same hash value to be stored. In an alternate implementation, uniqueness values are not assigned to the rows and the order of rows with identical hash values is not determined.

[00033] FIG. 4 shows a flow chart of an algorithm for creating and storing row data in a multi-level partitioned table in a database system. Initially, a table is created 410. One of the steps in creating a database table is defining the columns that will contain row data for the table 412. The database system can add columns such as a row ID to those specified. One or more columns are chosen as the primary index for the table 414. The primary index is used as the input to the hash function. One or more columns are also chosen as partitioning columns 416 for a partition level. A function is specified that uses the values of the partitioning columns for each row to calculate the partition in that level for that row 418. The selection of a partition function 418 can include specifying minimum and maximum resulting values for that function in some implementations. In an alternate implementation, the minimum and maximum resulting values are determined by applying the selected functions to the data type of the selected columns. The partition function itself may define the partition columns combining steps 416 and 418. More partition columns and functions are selected for each additional partition level 420. Once the partition functions for the multi-level partitioning are selected, a single-level partition function is generated 422. In one implementation the single-level partition function is generated in accordance with formula 10.

[00034] Once a table has been defined, users are allowed to fill the table by creating rows 424. The data in the new row is used to determine the hash value and hash bucket by applying the hash function to the columns of the primary index 426. The data storage facility in which the row is stored is determined by the hash bucket in combination with the hash bucket map 428. The single-level partition function is applied to the values in any columns selected for any of the initial partition functions 430. In one implementation a uniqueness number is also assigned to the row 432. The rows for that table in a particular data storage facility are then located in order of the single-level partitions 434. Within the single-level partitions, the rows are located in order of the result of the hash function, the hash value 436. Any rows in a partition that share that hash value are then ordered by the uniqueness value 438, if there is one.

[00035] In one implementation, the new row is placed in the proper position relative to the already ordered previously-input rows. In another implementation, several rows are input in a session before reordering according to partition, hash, and uniqueness occurs.

[00036] The text above described one or more specific implementations of a broader invention. The invention also is carried out in a variety of alternative implementations and thus is not limited to those described here. For example, while the invention has been described here in terms of a DBMS that uses a massively parallel processing (MPP) architecture, other types of database systems, including those that use a symmetric multiprocessing (SMP) architecture, are also useful in carrying out the invention. Many other implementations are also within the scope of the following claims.