Abstract:
A system and method for processing a group and aggregate query on a relation are disclosed. A database system determines whether assistance of a heterogeneous system (HS) of compute nodes is beneficial in performing the query. Assuming that the relation has been partitioned and loaded into the HS, the database system determines, in a compile phase, whether the HS has the functional capabilities to assist, and whether the cost and benefit favor performing the operation with the assistance of the HS. If the cost and benefit favor using the assistance of the HS, then the system enters the execution phase. The database system starts, in the execution phase, an optimal number of parallel processes to produce and consume the results from the compute nodes of the HS. After any needed transaction consistency checks, the results of the query are returned by the database system.

Description:
FIELD OF THE INVENTION 
       [0001]    The present application relates generally to the processing of SQL queries and in particular to the processing of a group-and-aggregate query with the assistance of a heterogeneous system. 
       BACKGROUND 
       [0002]    One of the core SQL operations is a group-and-aggregation operation. As the name suggests, two operations are involved. The group operation groups together all rows in a relation that share the same keys (columns). The aggregate operation aggregates values of non-key columns of the relation within each group. Some group-and-aggregate operations specify a set of filters to be applied on the relations before the grouping operation on the relation, which can be materialized. 
         [0003]    An example of a group and aggregate operation is: 
         [0000]    
       
         
               
               
             
               
               
               
             
               
               
             
               
               
               
             
           
               
                   
                   
               
             
             
               
                   
                 select d.DEPTNO 
               
               
                   
                    ,d.NAME 
               
             
          
           
               
                   
                    ,count(e.EMPNO) 
                 as NUM_EMP 
               
               
                   
                    ,nvl(sum(e.MSAL),o) 
                 as SUM_MSA 
               
             
          
           
               
                   
                 from DEPT d 
               
               
                   
                    ,EMP e 
               
             
          
           
               
                   
                 where 
                 d.DEPTNO = e.DEPTNO  (+) 
               
               
                   
                 group by 
                 d.DEPTNO, d.DNAME 
               
               
                   
                   
               
             
          
         
       
     
         [0004]    This SQL statement joins a department table d and an employee table e, groups the rows by department number DEPTNO and department name DNAME, and counts the number of employees NUM_EMP and the sum of their salaries e.MSAL into SUM_MSA. This query has the form of select AggFunc from R group by K, where AggFunc includes the count and sum functions, R is the relation with the department and employ tables, and K includes the two columns department name and department number. 
         [0005]    Currently database implementations of group-and-aggregate queries, such as the one above, use the classical iterator-based technique for serial evaluation of a query. The iterator technique includes opening a row source iterator on the relation, fetching rows, and filtering the rows. If the grouping includes sorting, the rows that pass the filter tests are sorted on the group-by keys. If the grouping includes hashing, a hash is computed based on the group-by key values. The sorted or hashed rows are organized into sort or hash run structures. After all of the rows in the relation have been consumed, the row source iterator is closed and the grouped and aggregated rows are returned. 
         [0006]    Large relations cause problems with current implementations of queries such as group-and aggregate. One problem is that the relation is so large that it does not fit in available memory, thus requiring many trips to disk to process portions that do fit in the available memory. The multiple trips to disk limit performance of the system to that of the disk system. 
         [0007]    Another problem is that the cost of applying the filters on each of the rows in the relation may be prohibitive. If the selectivity of the filters is low, the number of rows returned by the operation is large, leading to cases in which some aggregation operations do not fit in available memory. 
         [0008]    Yet another problem is that, if the number of groups resulting from the grouping operation is large, then constructing large hash or sort runs stresses the memory hierarchy of on-chip caches and memories. 
         [0009]    One approach to solving the above problems is to execute portions of the group-and-aggregate query in parallel, by taking advantage of multi-threaded CPU cores, pools of server processes, or multi-node clustered configurations. Executing portions of the query in parallel also requires some technique for merging these operations into a final result. 
         [0010]    Another approach is to off-load the processing of some of the operations involved in the group-and-aggregate operation to another system that is likely to perform the operations at a lower cost or to reduce the amount of data that the server process needs to process. 
       Heterogeneous Systems 
       [0011]    For large relations, database systems can benefit from Heterogeneous Systems (HS). These systems are ones with a large number of disk-less compute nodes, each with its own main memory, and a high-speed interconnect among the nodes. As the number of nodes is very large, the amount of memory aggregated over all of the nodes is also very large. The database system using the HS has access to an in-memory representation of the relation in the HS and to persistent storage where the relation is stored. 
         [0012]    Heterogeneous Systems are often organized in the form of a set of clusters of hierarchies, each cluster having a tree-like structure. Each leaf in the tree has a compute node and memory and is connected via switches that reside at multiple levels in the tree. Compute nodes in the hierarchy are built for both very efficient processing of a well-defined set of query primitives and low power consumption. The types of processors at each of the compute nodes can be different from processors elsewhere in the hierarchy or from processors in a database system that connects to the heterogeneous system. 
         [0013]    In one embodiment, a hierarchy has 200 compute nodes and a total of 3 terabytes (TB) of memory distributed over the nodes. A cluster of four such hierarchies provide about 12 TB of working memory, which is sufficiently large for holding an in-memory copy of a large relation. 
         [0014]    A heterogeneous system offers many benefits, such as a very high degree of parallelism, high throughput, and low power for operations, such as group-and-aggregate, on large relations. However, a heterogeneous system may have some functional limitations and cost-benefit tradeoffs in its use. One functional limitation is the inability to perform certain underlying functions needed by the group-and-aggregate operation. These functions include fetching the underlying row sources, supporting functions that use the key and column data types, and those that perform the particular aggregation specified. Lacking the ability to perform these underlying functions reduces the performance benefit of the heterogeneous system. Cost-benefit tradeoffs include comparison of the cost of loading portions of the relation into the heterogeneous system and collecting the results with the benefits of any improvement in the time and power consumed when the heterogeneous system assists in the group-and-aggregate operation. Additionally, because the heterogeneous system has no persistent storage for storing redo logs, the database system incurs a cost to assure transactional consistency. 
         [0015]    The approaches described in this section are approaches that could be pursued, but not necessarily approaches that have been previously conceived or pursued. Therefore, unless otherwise indicated, it should not be assumed that any of the approaches described in this section qualify as prior art merely by virtue of their inclusion in this section. 
       SUMMARY 
       [0016]    An embodiment for performing the operation has two phases, a compile phase and an execution phase. During the compile phase, the database system determines the size of the operation on a given relation and whether the HS is capable of performing functions needed to assist in the operation. Also during this phase, the embodiment determines the costs of performing the operation with the assistance of the HS compared to the database system alone performing the operation. If the cost with the assistance of the HS is low enough compared to the cost of using the database system by itself, then the HS assists in the execution phase. In the execution phase, the database system activates a special row source during which the HS produces partial results, which are aggregated and merged and provided to the database system. After all of the partial results are merged, the database system performs checks and adjustments to assure transactional consistency, closes the row source, and returns the query result. 
     
    
     
       BRIEF DESCRIPTION OF THE DRAWINGS 
         [0017]    In the figures: 
           [0018]      FIG. 1  depicts a flow chart of the major steps in processing a group-and-aggregate query in a heterogeneous system, such as the example system depicted in  FIG. 10 ; 
           [0019]      FIG. 2  depicts a flow chart of the check Op On HS step in  FIG. 1 ; 
           [0020]      FIG. 3  depicts a flow chart of the select Query Execution Plan step in  FIG. 1 ; 
           [0021]      FIG. 4  depicts a flow chart of the determine strategies step in  FIG. 1 ; 
           [0022]      FIG. 5  depicts a flow chart of the select Ph1 strategy step in  FIG. 4 ; 
           [0023]      FIG. 6  depicts a flow chart of the select Ph2 strategy step in  FIG. 4 ; 
           [0024]      FIG. 7  depicts a flow chart of the select Ph3 strategy step in  FIG. 4 ; 
           [0025]      FIG. 8  depicts a flow chart of the select Ph4 strategy step in  FIG. 4   
           [0026]      FIG. 9  depicts a flow chart of loop step in  FIG. 1 ; 
           [0027]      FIG. 10  depicts an example system in which an embodiment operates; and 
           [0028]      FIG. 11  depicts an example computer system. 
       
    
    
     DESCRIPTION 
     General Overview 
       [0029]    Assuming that the database had previously loaded the relation into the HS, which requires that the database system partition the relation and distribute the relation in a balanced manner among the compute nodes in the HS, an embodiment performs a sequence of checks to determine whether the HS is capable of assisting and whether the HS would improve the performance of the operation. These checks include the cost of performing the operation in the database system alone, the cost of performing the operation in a hierarchy of compute nodes, and the cost of merging the results from multiple hierarchies into a final result. If the checks indicate that the costs are sufficiently low, then the database system uses the HS in the operation. Otherwise, the database system performs the operation by itself. 
         [0030]    If the database system does decide to use the HS, the database system prepares processes to produce and collect results from the HS. The database system then starts a special row source, which is an iterator over the relation on which the operation is to be performed. Producing results from the HS requires a set of processes be started in the HS under control of a scheduler. Collecting the results requires that the database system start a number of consumer processes, where the number of consumer processes depends on the degree of parallelism (DOP) supported by the database system and is adjusted to account for the degree of parallel execution in the HS. While active, each of the consumer processes expects to receive a certain number of payloads produced from the HS after which the consumer process completes. After all of the consumer processes finish, the database system determines whether any blocks are out of sync with the blocks in the database system. If so, the database system takes care of the transactional semantics to assure that transactional consistency for the out of sync blocks, after which the database system closes the special row source. 
       DETAILED DESCRIPTION 
       [0031]    In the following description, for the purposes of explanation, numerous specific details are set forth in order to provide a thorough understanding of the present invention. It will be apparent, however, that the present invention may be practiced without these specific details. In other instances, well-known structures and devices are shown in block diagram form in order to avoid unnecessarily obscuring the present invention. 
         [0032]    Referring to  FIG. 1 , the group-and-aggregate operation on a relation has two phases, the compile phase and the runtime phase, during which a number of steps are executed. During the compile phase, steps are performed that assist in determining whether the HS can be used and whether it is worth using. The steps performed are checkOpOnHS  110 , selectQryExPln  112 , determine strategies  114 , and computeFinalCost  116 . Each of these steps is described below. During the runtime phase, steps are performed that manage the operation in the HS, if the HS is assisting, as determined in step  118 . Those steps include loadPartitions  120 , startGroupAndAggregatePushdownRowSource  122 , collect and merge results loop  124 , run discreteGroupByRowSource  126 , and closeGroupAggregatePushDownRowSource  128 . If the HS is not assisting as determined in step  118 , then the operation is performed in the database system in step  130 . 
       Compile Phase 
       [0033]      FIG. 2  depicts steps involved in the checkOpOnHS step  110  of  FIG. 1 . The system checks, in step  210 , to determine if the operation can run on the HS. In this step, the system checks recursively to decide whether the underlying row sources can be fetched from the HS. In step  216 , the system checks whether the HS supports data types of the group-by keys and columns that are aggregated. In step  214 , the system checks whether the HS can compute the aggregation functions, after a possible query rewrite. In step  212 , the system checks to determine if relations are all loaded in the HS. For example, in the most simple case, when the underlying row source is a table scan, the compile-time checks determine whether HS can support data types in the filtered and projected columns, whether the filters can be evaluated in the system, and whether the relation, or the columns that are projected, selected, grouped-by and aggregate, are all loaded in the HS. 
         [0034]      FIG. 3  depicts steps involved in the selectQryExPln step  112  of  FIG. 1 . The system gathers various statistics on the relation to be executed. The statistics include the size of the group (group-by cardinality), obtained in step  310 , for the entire group-by relation, the size of the groups in each hierarchy H of the HS obtained in step  312 , the size of the group for each node in the HS obtained in step  314 . Statistics on group-by cardinality are needed to best determine the strategies used by both the database system and the HS, because group-by operations are very sensitive to the number of groups produced. Statistics also include the number of blocks in the relation, obtained in step  316 , that are loaded in the HS but are not in sync with the database system. This statistic is vital for handling any transactional semantics after the execution of the operation. It is important to maintain a certain level of accuracy regarding the number of out-of-sync blocks. The statistics should indicate whether it is very likely that a majority of blocks, as determined in step  316 , in database persistent storage have changed since the relation was last loaded into the HS. If so, as determined in step  318 , the database system can decide that the HS should not be used, as in step  320 , because the cost of maintaining transactional consistency is too high. In that case, the user can reload consistent data into the HS. 
       Strategies 
       [0035]      FIG. 4  depicts steps involved in the determineStrategies step  114  in  FIG. 1 . The system performs a number of calculations in sequence. These calculations are then used to compute a final cost based on which a final determination is made to use or not use the HS. 
       Ph1 Strategy 
       [0036]    In the selectPh1Strategy step  410  in  FIG. 4 , the system decides on the best strategy for performing the group and aggregate operation in the database system only (i.e., without the assistance of the HS).  FIG. 5  depicts steps involved in step  410 . The step  510  in  FIG. 5  computes the costDB value. If the cost is low as determined in step  512  of  FIG. 5 , the HS is not used as depicted in step  514 . In one embodiment, 
         [0037]    costDB(tDB,powerDB)=tDB/powerDB, 
         [0038]    where tDB is the estimated time of executing the operation and powerDB is the power required to execute the operation. 
         [0039]    The following Phase2-4 strategies determine the cost benefit tradeoffs of having the HS assist the database system. 
       Ph2 Strategy 
       [0040]    In the step selectPh2Strategy  412  of  FIG. 4 , the system decides on the best strategy for performing the group-aggregate operation within each individual hierarchy of the HS. In an HS with multiple hierarchies, the strategy for each can be different. The phase2 strategy can be altered, so the strategy is considered a hint. The best plan is selected for each node, because the optimizer has information as to which portions of the HS have hardware accelerators for efficient operations such as sorting. For example, if a hierarchy has an accelerator for sorting, the optimizer can indicate to the hierarchy that it perform a sort-based aggregation, even though the database system chooses a hash based aggregation for itself. These plans and estimations on potential group-by cardinality reduction from intermediary aggregation (i.e., aggregation at non-leaf nodes from the second phase) are used to drive the selection of plans for merging results from hierarchies to produce the final result for the HS in phase 3. 
         [0041]      FIG. 6  depicts, in one embodiment, an algorithm for phase2 that includes the steps of, for each node n in the hierarchy as determined in step  610 , obtaining the size sz of the partition in the node n in step  612 , reading the expected cardinality statistics c for the node n in step  614 , and for each candidate strategy s in Phase 2 as determined in step  616 , computing the cost C(n,s,c, powern) for node n in step  618 , and comparing all of the costs to select, in step  620 , the strategy s* that minimizes the cost C(n,s,c, powern). 
         [0042]    In one embodiment, the cost 
         [0000]    
       
         
           
             
               
                 C 
                  
                 
                   ( 
                   
                     n 
                     , 
                     s 
                     , 
                     c 
                   
                   ) 
                 
               
               = 
               
                 
                   T 
                   - 
                   
                     Phase 
                      
                     
                         
                     
                      
                     2 
                      
                     
                       
                         ( 
                         
                           s 
                           , 
                           c 
                         
                         ) 
                       
                       · 
                       sz 
                     
                   
                 
                 powern 
               
             
             , 
           
         
       
     
         [0000]    where powern is the power requirement for the node n and T−Phase2(s,c) is the time of phase 2 assuming strategy and cardinality c. The algorithm then returns the optimized strategy, s*. 
       Ph3 Strategy 
       [0043]    In the selectPh3Strategy step  414  of  FIG. 4 , the system decides on the best strategy for merging results produced from the nodes in each hierarchy within the HS based on the results in the second phase to create a single result from each hierarchy. This phase considers the very high degree of parallelism inherent in the large number of nodes in a HS, the number of producer processes that the database system can handle, and the number of cores in multi-core compute nodes. 
         [0044]      FIG. 7  depicts, in one embodiment, how Phase 3 evaluates a Centralized Aggregation strategy, a centralized multi-phase aggregation strategy, and a Repartition Aggregation strategy and then selects among them depending on the selectivity of the group-and aggregate operation. If the group-and-aggregate is estimated to be highly selective as determined in step  710 , then the Centralized Aggregation strategy is selected in step  720 . If the selectivity is medium as determined in step  722 , then the Centralized Multi-Phase Aggregation strategy is selected in step  724 . If the selectivity is low as determined in step  726 , then the Repartition Aggregation strategy is selected in step  728 . 
         [0045]    In the Centralized Aggregation Strategy, each leaf node in the HS aggregates its data and sends the aggregated data to its parent node. Intermediate nodes in HS just relay the data and the root node acts as the final merging stage. 
         [0046]    In the Centralized Multi-phase Aggregation strategy, each leaf node in HS aggregates its data and sends the aggregated data to its parent node. Intermediate nodes aggregate data from their child nodes. This algorithm has a potential advantage over the Centralized Aggregation algorithm if the intermediate aggregate reduces the group cardinality. 
         [0047]    In the Repartition Algorithm, each leaf node aggregates its data and then partition and redistributes the results to the leaf nodes based on an agreed-upon scheme such that no groups with the same keys from different leaf nodes are merged by separate leaf nodes. Each leaf node then sends its final results to the expected consumer. 
       Ph4 Strategy 
       [0048]    In step  416  of  FIG. 4 , the optimizer considers plans for merging results coming from nodes in different hierarchies, where the number of nodes is much higher than: the typical number of nodes in a distributed cluster, the typical number of nodes in producer/consumer processes that execute in parallel in the database system along, and the typical number of cores in multi-core systems tuned for scalable aggregation. 
         [0049]    More specifically, the system uses the results from phase 3 to decide on the best overall strategy for merging results from each hierarchy of the HS to produce the total HS result.  FIG. 8  depicts steps involved in step  416 . In step  810  of  FIG. 8  determines whether the HS can merge the results into a single hash table. If so, then the system does so. If not, the system creates a private hash table for each server process, as depicted in steps  814  and  816 . These tables can be adjusted in the dispatchConsumerProc step, discussed below. 
         [0050]    Breaking the selection and cost functions into phases described above observes the natural dependencies among these phases and helps to modularize the optimizer code that runs during compilation. 
       Compute Final Cost 
       [0051]    In the computeFinalCost step, the system computes the cost, costHet, to perform the operation with the assistance of the HS, determined by the Ph2-Ph4 strategy steps, and compares it with the cost of computing the operation in the database system alone determined in the Ph1 strategy step. If the cost of operating with the HS is lower than the cost using the database system alone, then the HS assists in the execution of the operation. The details of the final cost calculation are described below. If the cost costHet is less than the costDB, then the system proceeds with the assistance of the HS, otherwise it uses only the database system. 
       Execution 
       [0052]    During the execution phase, steps are executed to manage the production and consumption of results when the HS is used. The steps include loadPartitions  120  in  FIG. 1 , startGroupAggPushDownRowSource  122 , run loop to Collect And Merge Results from Nodes  124 , runDiscreteGroupByRowSource  126 , and closeGroupAggPushDownRowSource  128 . Each of these steps is described below. 
         [0053]    In the loadPartitions step  120 , the system partitions and loads the relation among the nodes in the HS if the relation has not already been loaded into the HS in step  212  of  FIG. 2 . The step  120  uses statistics to determine the optimum range-based partitioning of the data. Partitioning among the node is based on the group-by keys. If histogram statistics on the group-by relation are not available, the statistics are gathered. Even though gathering the statistics can be costly, it is not done that often because the query has low rates of data insertion or updates. The HS loads the data in the same range on leaf-nodes that share the same immediate switch or parent node. Loading data in this manner minimizes the number of groups that are common to distinct switch nodes, which minimizes the number of merges at intermediary levels in the HS, before data is sent back to the database system. 
         [0054]    In the startGroupAggPushDownRowSource step  122  of  FIG. 1 , the system activates the new group and aggregate row source. After the row source is started, the database system enters a loop  124  during which payloads (i.e., messages consisting) of groups and corresponding aggregates are requested from the HS and consumed in the database system. This flow of payloads from the HS to the database system is governed by a set of producer processes in the HS that produce the payloads and a set of consumer processes in the database system that consume the payloads. The set of producer processes in the HS is further governed by the partitioning of the relation in the HS during the loadRoutine step and controlled by a scheduler in the HS. In one embodiment, producer processes in the HS are based on priority queues such that each group-and-aggregate operation is broken into multiple producer processes, which are all queued at potentially different priorities. In the embodiment, the number of consumer processes in the database system is set to match the partitioning of the relation in the HS. The new row source guarantees that the consumer processes can process a certain number of groups, along with their aggregates. The load balancing between the producer processes and the consumer processes is dynamic so that the database system can adjust to the high rate and possibly unbalanced sizes of the payloads from the different producer processes in the HS. 
         [0055]    The steps that govern the flow of payloads from the HS to the database system are depicted in  FIG. 9  and include the following requestFetchNewPayLoadFromHS  912 , waitForNewPayloadOrSelectFirstBuffed  914 , selectConsumerProcForMerge  916 , dispatchConsumerProc  918 , adjustRowSourceForNextFetch  920 . 
         [0056]    The requestFetchNewPayLoadFromHS step  912  in  FIG. 9  makes a request for a new payload from the HS, which produces payloads for the row source. 
         [0057]    The step waitForNewPayloadOrSelectFirstBuffed  914  in  FIG. 9  waits for the production of a new payload from the HS as the payloads become available. When a buffered payload is available, the step then selects the buffered payload. 
         [0058]    The step selectConsumerProcForMerge  916  in  FIG. 9  determines which of the consumer processes running on the database system should merge the payload into the global results. 
         [0059]      FIG. 10  depicts steps involved in the dispatchConsumerProc step  918  in  FIG. 9 . The database system dispatches the consumer processes for fetched payload processing. In the step, the system monitors the run time to determine whether the performance expectations of the HS are occurring. If the run time statistics indicate that the performance is low as in step  1010  in  FIG. 10 , then the system creates new tables in step  1012 . If the size of the private hash table ph is too large as determined in step  1014 , then the partition distribute module is run in step  1016 . If the size of the private hash table is greater than a threshold as determined in step  1018 , then the private hash table is synchronized with the shared table in step  1020  and the hit frequency is maintained in step  1022 . If the single hash table is used (hs is true), then the system creates a shared table in step  1026 . 
         [0060]    The adjustRowSourceForNextFetch step  920  in  FIG. 9  adjusts the row-source guarantee for the next fetch to manage flow control between the producers in the HS and the consumers in the database system. 
         [0061]    After all of the payloads have been produced and consumed as determined in step  910 , the loop ends and the step discreteGroupByRowSource step  126  in  FIG. 1  checks to determine the blocks that are not in sync with persistent storage in the database system. For those blocks that are not in sync, the function starts a row source to handle these blocks. 
         [0062]    After the database system has processed all of the payloads and merged into the result any out of sync blocks, the closeGroupAggPushDownRowSource step  128  in  FIG. 1  is executed to close the new row source. 
       Cost Model 
       [0063]    The cost model is built upon the parameters in the table below and  FIG. 10  depicts various costs incurred in the hierarchical system. In  FIG. 10 , the database system  1010  is coupled via link  1008  to the hierarchical system, which includes nodes n0  1012 , n10-14 ( 1016   a - e ), nodes n20-24 ( 1020   a - e ), nodes n25-29 ( 1024   a - e ), nodes n30-34 ( 1028   a - e ), and nodes n35-n39 ( 1032   a - e ). As depicted, nodes n30-34 are coupled via interconnect  1026  to parent node n20. Nodes n35-n39 are coupled via interconnect  1030  to parent node n25. Nodes n20-24 are coupled via interconnect  1018  to parent node n10; and node n24-n29 are coupled via interconnect  1022  to parent node n14. Other nodes, such as n21-24, n26-29 and n11-n13 can be parent nodes as well. 
         [0000]    
       
         
               
               
               
             
           
               
                   
               
               
                 Parameter 
                 Definition 
                 Meaning 
               
               
                   
               
             
             
               
                 R 
                   
                 the relation 
               
               
                 N 
                   
                 number of nodes in HS 
               
               
                 N2 
                   
                 number of intermediate nodes in merge 
               
               
                 Rn 
                   
                 relation partition on node n 
               
               
                 card(R) 
                   
                 size of relation R 
               
               
                 card(R)/N 
                   
                 size of relation per node 
               
               
                 card(Rn) 
                   
                 cardinality of partitioned relation 
               
               
                 DOP 
                   
                 degree of parallelism among producers 
               
               
                   
                   
                 and consumer processes 
               
               
                 records 
                 card(R)/DOP 
                 size of relation per process 
               
               
                 f 
                 card(groupsK in 
                 reduction factor (selectivity of query) 
               
               
                   
                 R) ÷ (card(R)/N) 
               
               
                 xDB 
                   
                 rate of processing records in DB 
               
               
                 LDB 
                   
                 latency of communication between 
               
               
                   
                   
                 producers and consumers 
               
               
                 L 
                   
                 interconnect latency 
               
               
                 L2 
                   
                 latency of communication between 
               
               
                   
                   
                 nodes 
               
               
                 powerDB 
                   
                 estimated power requirement for DB 
               
               
                 powerHS 
                   
                 estimate power requirement for HS 
               
               
                 c1 
                 xDB · records 
                 time to process in DB 
               
               
                 c2 
                 LDB · f · records 
                 latency to transfer data between 
               
               
                   
                   
                 producers and consumers 
               
               
                 c3 
                 xDB · f · records 
                 time for consumers to produce final 
               
               
                   
                   
                 results 
               
               
                 X 
                   
                 processing rate in node 
               
               
                 Y 
                   
                 merging rate of a node 
               
               
                 c4 
                 X/N · card(R) 
                 time for a node to process its partition 
               
               
                 c5 
                 Y · f · card(R) 
                 time for a node to merge results if 
               
               
                   
                   
                 merging with a single merger node 
               
               
                 c6 
                 Y/N2 · f · card(R) 
                 time for a node to merge results if 
               
               
                   
                   
                 merging across multiple (N2) inter- 
               
               
                   
                   
                 mediate nodes 
               
               
                 c7 
                 L · f · card(R) 
                 time to transmit results over 
               
               
                   
                   
                 interconnect 
               
               
                 c8 
                 xDB · f · card(R) 
                 time to merge in DB 
               
               
                 c9 
                 L2/N2 · f 
                 alternative latency 
               
               
                   
               
             
          
         
       
     
         [0064]    The computeFinalCost step calculates the cost of performing the operation in the database system alone or with the assistance of the HS, based on a number of primary inputs and derived values. The time to produce results in the database system,  1010  in  FIG. 10 , alone is 
         [0000]        t DB= c 1 +c 2 +c 3. 
         [0065]    The cost for performing the operation with the assistance of the HS is 
         [0000]    tHS=c4+c6+c7+c8, as depicted for node n0  1012 . Alternatively, the cost with the assistance of the HS is
 
tHS1=c4+c6+c7+c9, as depicted for node n0  1012 , if the alternative latency c9 is used.
 
The cost costHS of using the assistance of the HS is then
 
         [0000]      costHS= t HS or  t HS1/powerHS. 
         [0000]    The optimizer decides to use the HS if costHS&lt;costDB. Thus, the decision is based on performance per unit of power. 
         [0066]    In practice the DOP for database execution, is on the order of 10 to 100 times smaller than the degree of parallelism in the HS and 10 times smaller than the degree of parallelism among the merging nodes. If the HS exhibits a 10 times improvement in performance per unit of power to process the group-and-aggregate operation, i.e., if 
         [0000]    
       
         
           
             
               xDB 
               powerDB 
             
             / 
             
               
                 X 
                 powerHS 
               
               ~ 
               10 
             
           
         
       
     
         [0067]    then the cost optimizer chooses the HS if 
         [0000]    
       
         
           
             
               
                 L 
                  
                 
                     
                 
                  
                 D 
                  
                 
                     
                 
                  
                 B 
               
               
                 D 
                  
                 
                     
                 
                  
                 O 
                  
                 
                     
                 
                  
                 P 
               
             
             &gt; 
             
               
                 ( 
                 
                   
                     Y 
                     
                       N 
                        
                       
                           
                       
                        
                       2 
                     
                   
                   + 
                   L 
                 
                 ) 
               
               * 
               
                 powerDB 
                 powerHS 
               
             
           
         
       
       
         
           
             
               
                 and 
                  
                 
                   
 
                 
                 ( 
                 
                   1 
                   + 
                   f 
                 
                 ) 
               
               * 
               
                 xDB 
                 
                   D 
                    
                   
                       
                   
                    
                   O 
                    
                   
                       
                   
                    
                   P 
                 
               
             
             &gt; 
             
               
                 x 
                 N 
               
               * 
               
                 powerDB 
                 powerHS 
               
             
           
         
       
     
         [0068]    The first inequality is highly dependent on the specific interconnect latencies, and the second inequality is highly dependent on the parameter f. 
         [0069]    For small f, i.e., for the case of very selective group-and-aggregate queries, it more likely that the second inequality is met. This is the case when the overall latency of merging a small number of groups across the N2 nodes and transmitting the final result over a fast interconnect is expected to be no larger than the overhead of merging final results across DOP potentially more powerful nodes. 
         [0070]    For large f, i.e., for the case of low selectivity queries, the latency over the interconnect is likely to dominate the latency LDB, because there is no data reduction between nodes N and N2 or between N2 and the database system. In this case, the optimizer should make the decision not to use the assistance of the HS. 
       Hardware Overview 
       [0071]    According to one embodiment, the techniques described herein are implemented by one or more special-purpose computing devices. The special-purpose computing devices may be hard-wired to perform the techniques, or may include digital electronic devices such as one or more application-specific integrated circuits (ASICs) or field programmable gate arrays (FPGAs) that are persistently programmed to perform the techniques, or may include one or more general purpose hardware processors programmed to perform the techniques pursuant to program instructions in firmware, memory, other storage, or a combination. Such special-purpose computing devices may also combine custom hard-wired logic, ASICs, or FPGAs with custom programming to accomplish the techniques. The special-purpose computing devices may be desktop computer systems, portable computer systems, handheld devices, networking devices or any other device that incorporates hard-wired and/or program logic to implement the techniques. 
         [0072]    For example,  FIG. 11  is a block diagram that depicts a computer system  1100  upon which an embodiment may be implemented. Computer system  1100  includes a bus  1102  or other communication mechanism for communicating information, and a hardware processor  1104  coupled with bus  1102  for processing information. Hardware processor  1104  may be, for example, a general-purpose microprocessor. 
         [0073]    Computer system  1100  also includes a main memory  1106 , such as a random access memory (RAM) or other dynamic storage device, coupled to bus  1102  for storing information and instructions to be executed by processor  1104 . Main memory  1106  also may be used for storing temporary variables or other intermediate information during execution of instructions to be executed by processor  1104 . Such instructions, when stored in non-transitory storage media accessible to processor  1104 , convert computer system  1100  into a special-purpose machine that is customized to perform the operations specified in the instructions. 
         [0074]    Computer system  1100  further includes a read only memory (ROM)  1108  or other static storage device coupled to bus  1102  for storing static information and instructions for processor  1104 . A storage device  1110 , such as a magnetic disk or optical disk, is provided and coupled to bus  1002  for storing information and instructions. 
         [0075]    Computer system  1100  may be coupled via bus  1102  to a display  1112 , such as a cathode ray tube (CRT), for displaying information to a computer user. An input device  1114 , including alphanumeric and other keys, is coupled to bus  1102  for communicating information and command selections to processor  1104 . Another type of user input device is cursor control  1116 , such as a mouse, a trackball, or cursor direction keys for communicating direction information and command selections to processor  1104  and for controlling cursor movement on display  1112 . This input device typically has two degrees of freedom in two axes, a first axis (e.g., x) and a second axis (e.g., y), that allows the device to specify positions in a plane. 
         [0076]    Computer system  1100  may 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 system  1100  to be a special-purpose machine. According to one embodiment, the techniques herein are performed by computer system  1100  in response to processor  1104  executing one or more sequences of one or more instructions contained in main memory  1106 . Such instructions may be read into main memory  1106  from another storage medium, such as storage device  1110 . Execution of the sequences of instructions contained in main memory  1106  causes processor  1104  to perform the process steps described herein. In alternative embodiments, hard-wired circuitry may be used in place of or in combination with software instructions. 
         [0077]    The term “storage media” as used herein refers to any non-transitory media that store data and/or instructions that cause a machine to operation in a specific fashion. Such storage media may comprise non-volatile media and/or volatile media. Non-volatile media includes, for example, optical or magnetic disks, such as storage device  1110 . Volatile media includes dynamic memory, such as main memory  1106 . Common forms of storage media include, for example, a floppy disk, a flexible disk, hard disk, solid state drive, magnetic tape, or any other magnetic data storage medium, a CD-ROM, any other optical data storage medium, any physical medium with patterns of holes, a RAM, a PROM, and EPROM, a FLASH-EPROM, NVRAM, any other memory chip or cartridge. 
         [0078]    Storage media is distinct from but may be used in conjunction with transmission media. Transmission media participates in transferring information between storage media. For example, transmission media includes coaxial cables, copper wire and fiber optics, including the wires that comprise bus  1102 . Transmission media can also take the form of acoustic or light waves, such as those generated during radio-wave and infra-red data communications. 
         [0079]    Various forms of media may be involved in carrying one or more sequences of one or more instructions to processor  1104  for execution. For example, the instructions may initially be carried on a magnetic disk or solid-state drive of a remote computer. The remote computer can load the instructions into its dynamic memory and send the instructions over a telephone line using a modem. A modem local to computer system  1100  can receive the data on the telephone line and use an infra-red transmitter to convert the data to an infra-red signal. An infra-red detector can receive the data carried in the infra-red signal and appropriate circuitry can place the data on bus  1102 . Bus  1102  carries the data to main memory  1106 , from which processor  1104  retrieves and executes the instructions. The instructions received by main memory  1106  may optionally be stored on storage device  1110  either before or after execution by processor  1104 . 
         [0080]    Computer system  1100  also includes a communication interface  1118  coupled to bus  1102 . Communication interface  1118  provides a two-way data communication coupling to a network link  1120  that is connected to a local network  1122 . For example, communication interface  1118  may be an integrated services digital network (ISDN) card, cable modem, satellite modem, or a modem to provide a data communication connection to a corresponding type of telephone line. As another example, communication interface  1118  may be a local area network (LAN) card to provide a data communication connection to a compatible LAN. Wireless links may also be implemented. In any such implementation, communication interface  1118  sends and receives electrical, electromagnetic or optical signals that carry digital data streams representing various types of information. 
         [0081]    Network link  1120  typically provides data communication through one or more networks to other data devices. For example, network link  1120  may provide a connection through local network  1122  to a host computer  1124  or to data equipment operated by an Internet Service Provider (ISP)  1126 . ISP  1126  in turn provides data communication services through the world wide packet data communication network now commonly referred to as the “Internet”  1128 . Local network  1122  and Internet  1128  both use electrical, electromagnetic or optical signals that carry digital data streams. The signals through the various networks and the signals on network link  1120  and through communication interface  1118 , which carry the digital data to and from computer system  1100 , are example forms of transmission media. 
         [0082]    Computer system  1100  can send messages and receive data, including program code, through the network(s), network link  1120  and communication interface  1118 . In the Internet example, a server  1130  might transmit a requested code for an application program through Internet  1128 , ISP  1126 , local network  1122  and communication interface  1118 . 
         [0083]    The received code may be executed by processor  1104  as it is received, and/or stored in storage device  1110 , or other non-volatile storage for later execution. 
         [0084]    In the foregoing specification, embodiments of the invention have been described with reference to numerous specific details that may vary from implementation to implementation. The specification and drawings are, accordingly, to be regarded in an illustrative rather than a restrictive sense. The sole and exclusive indicator of the scope of the invention, and what is intended by the applicants to be the scope of the invention, is the literal and equivalent scope of the set of claims that issue from this application, in the specific form in which such claims issue, including any subsequent correction.