Abstract:
Disclosed is a data processing system implemented method, a data processing system, and an article of manufacturing for use with the data processing system. The data processing system implemented method is used for directing the data processing system to join rows associated with a column of a source table with rows associated with a column of a target table. The data processing system includes a source node containing the source table and including a target node containing the target table. The method includes generating a reduced representation of selected rows associated with the column of the source table, and generating a representation of the column of the target table, filtering the generated reduced representation of selected rows associated with the column of the source table through the generated representation of the column of the target table, the filtered generated reduced representation of selected rows identifying source table rows that do not have to be joined with the target table, and joining, to the rows associated with the column of the target table, the rows associated with the column of the source table minus the filtered generated reduced representation of selected rows.

Description:
FIELD OF THE INVENTION  
       [0001]     The present invention relates generally relates to database management systems, and more particularly, the present invention relates to a data processing system implemented method, a data processing system and an article of manufacture for joining rows associated with a column of a source table with rows associated with a column of a target table.  
       BACKGROUND  
       [0002]     In order to handle an ever growing amount of information, database management systems have recently been implemented on what are known as distributed or parallel architectures. One form of distributed architecture is the shared-nothing architecture. This architecture is called shared-nothing because the architecture does not include a single point of failure, such as a shared disk or shared memory.  
         [0003]     A typical shared nothing system includes a plurality of database partitions coupled together by a network. The database partitions generally include a set of storage devices, such as disk drives, and one or more processors. The network may comprise various types of networks, such as a local area network, and physically diverse paths. Data in a shared nothing system is partitioned using a partitioning scheme such as a hash partitioning scheme and stored across the various database partitions that are connected together through the network. Thus, each database partition keeps a part of the overall database and various operations on parts of the database in different database partitions may be performed in parallel.  
         [0004]     Data in the database is kept as tuples in relations that comprises of logically grouped attributes. Tuples and attributes in relations are also referred to as rows and columns in tables. An important feature of any database system, including shared-nothing systems, is the ability to process operations between rows in tables, such as relating data in one table to data in another table. These types of operations are particularly useful when querying related data from multiple tables such that the data needs to be retrieved in a meaningful way. For example, structured query language (“SQL”) supports what are known as join operations for querying data from multiple tables. Indeed, many consider the join operation as the cornerstone of database management systems.  
         [0005]     A join operation may be expensive to perform in that it may require a large amount of processing and communication resources. One form of join operation in a distributed database system is the semi-join operation. A semi-join operation involves two tables R and S that are stored on different sites within a distributed system. A semi-join from R to S is performed by first projecting Ron the join attributes at the site of R. The join attributes typically include one or more keys for selecting various tuples from tables R and S. Site R sends the resulting projection to the site of S, where a join operation is performed. The projection of R essentially acts as a filter of S, and thus, the result of the semi-join is usually smaller than S. Site S then sends to site R the semi-join result to complete the join operation. Hence, semi-joins can require creating and sending projections of one of the tables involved in the join across a network when the tables involved in the join are stored at different sites. This could be expensive and hence techniques to reduce semi-join processing are needed.  
         [0006]     Using bit filters is one technique used to reduce the amount of communication during the execution of semi-joins. For example, bit filters can be used to reduce the size of the projection of the joining keys, and thus, reduce the amount of data communication needed. A bit filter operates by hashing the projection data into a set of bits. This set of bits serves as a compressed representation of the hashed data. Bit filters are also known as Bloom filters or Hash filters by those skilled in the art. Prior art that looked at semi-joins in the context of distributed systems essentially dealt with joining tables R and S located at two sites where each site contained a single table.  
         [0007]     In general, data in tables in a partitioned environment may be distributed over multiple database partitions. One way to improve the performance of a database system is to divide the join operation or query so that each database partition can work in parallel with little inter-node communication. If data within two tables is partitioned in a similar way, a collocated join would be the most efficient join method. This can be done if the join attributes are the same as the partitioning attributes of the tables. No data would need to be moved to do the join. More often than not, this match of the partitioning keys and the join attributes is not possible and tuples of one or both tables need to be moved across the database partitions in order to perform the join operation. To minimize movement of tuples, bit filters might be used. In known systems, bit filters are broadcast between the various database partitions during various stages of a database operation. However, broadcasting bit filters themselves can consume a large amount of communication resources of the system. For example, when a particular database partition of one table is performing operations for a join, it must broadcast its updated bit filter to all the database partitions of the other table participating in the join. Some systems might use compressed bit filters to improve their efficiency. Unfortunately, even in a compressed form, bit filters can become a tremendous burden particularly when there are a large number of database partitions involved. Each database partition associated with one table requires enough memory space to store all the bit filters of all the other table&#39;s database partitions. In addition, database partitions may update their data frequently, and thus, may frequently need to broadcast/re-broadcast their bit filters. This can consume significant communication resources in the database system.  
         [0008]     Therefore, it would be desirable to provide methods and systems that efficiently uses processing and communications resources when performing database operations. It would also be desirable, among other things, to provide methods and systems that efficiently use memory and data communications resources to perform various types of join operations across database partitions.  
       SUMMARY  
       [0009]     In an aspect of the present invention, there is provided a data processing system implemented method of directing a data processing system to join rows associated with a column of a source table with rows associated with a column of a target table, the data processing system including a source node containing the source table and including a target node containing the target table, the method including generating a reduced representation of selected rows associated with the column of the source table, and generating a representation of the column of the target table, filtering the generated reduced representation of selected rows associated with the column of the source table through the generated representation of the column of the target table, the filtered generated reduced representation of selected rows identifying source table rows that do not have to be joined with the target table, and joining, to the rows associated with the column of the target table, the rows associated with the column of the source table minus the filtered generated reduced representation of selected rows.  
         [0010]     In another aspect of the present invention, there is provided a data processing system for joining rows associated with a column of a source table with rows associated with a column of a target table, the data processing system including a source node containing the source table and including a target node containing the target table, the data processing system including a generating module for generating a reduced representation of selected rows associated with the column of the source table, and generating a representation of the column of the target table, a filtering module for filtering the generated reduced representation of selected rows associated with the column of the source table through the generated representation of the column of the target table, the filtered generated reduced representation of selected rows identifying source table rows that do not have to be joined with the target table, and a joining module for joining, to the rows associated with the column of the target table, the rows associated with the column of the source table minus the filtered generated reduced representation of selected rows.  
         [0011]     In yet another aspect of the present invention, there is provided an article of manufacture for directing a data processing system to join rows associated with a column of a source table with rows associated with a column of a target table, the data processing system including a source node containing the source table and including a target node containing the target table, the article of manufacture including a program usable medium embodying one or more instructions executable by the data processing system, the one or more instructions including data processing system executable instructions for generating a reduced representation of selected rows associated with the column of the source table, and generating a representation of the column of the target table, data processing system executable instructions for filtering the generated reduced representation of selected rows associated with the column of the source table through the generated representation of the column of the target table, the filtered generated reduced representation of selected rows identifying source table rows that do not have to be joined with the target table, and data processing system executable instructions for joining, to the rows associated with the column of the target table, the rows associated with the column of the source table minus the filtered generated reduced representation of selected rows.  
         [0012]     Additional features of the invention will be set forth in part in the description which follows, and in part will be obvious from the description, or may be learned by practice of the invention. The features of the invention will be realized and attained by means of the elements and combinations particularly pointed out in the appended claims.  
         [0013]     It is to be understood that both the foregoing general description and the following detailed description are exemplary and explanatory only and are not restrictive of the invention, as claimed. 
     
    
     BRIEF DESCRIPTION OF THE DRAWINGS  
       [0014]     The accompanying drawings, which are incorporated in and constitute a part of this specification, illustrate embodiments of the invention and together with the description, serve to explain the principles of the invention.  
         [0015]      FIG. 1  illustrates a system  100  implemented in the form of a shared nothing distributed architecture;  
         [0016]      FIG. 2  illustrates a software architecture for the system  100  of  FIG. 1 ;  
         [0017]      FIG. 3  illustrates an architecture of an operator node which is part of the system  100  of  FIG. 1 ;  
         [0018]      FIG. 4  illustrates an example of a routing table and buffer space in a memory of an operator node, such as the operator node of  FIG. 3 ; and  
         [0019]      FIGS. 5A and 5B  illustrate a process flow related to a join operation of a query involving multiple operator nodes of the system of  FIG. 1 . 
     
    
     DETAILED DESCRIPTION OF THE EMBODIMENTS  
       [0020]     Reference will now be made to exemplary embodiments of the invention, which are illustrated in the accompanying drawings. Wherever possible, the same reference numbers will be used throughout the drawings to refer to the same or like parts.  
         [0021]     An embodiment provides a method of communicating data between partitions of a database system. Data related to queries of the database system are received at a first database partition designated as the source node. Data that is destined to a second database partition is identified. The second partition is designated as the target node. The target node has a bit filter created based on the tuples of the relation relevant to the query at this partition. The bit filter may be one that is created as part of a join operation such as a hash join or a merge join. These methods are known methods to those skilled in the art. The identified data at the source node is stored in a set of data buffers assigned to the target node. The identified data is encoded into hash codes and stored into a hash code buffer also assigned to the target node. The data buffers and the hash code buffer are filled until a threshold is reached. The threshold is determined based on the number of partitions involved in each relation of the join, the communication layer&#39;s buffer window count, memory allocated for the operation and a measure of reduction factor of the bit filter. A separate asynchronous process transmits the hash code buffer to the target node. The target node checks the hash codes against the bit filter and creates a bitmap that indicates tuples in the data buffer at the source node that will not qualify the join operation . . . The bitmap is transmitted from the target node to the source node where the data from the corresponding data buffers assigned to the target node can be discarded. The data buffers are compacted to eliminate unused space freed up by the discarded tuples if any. The potentially reduced set of data buffers is then sent from the source node to the target node to perform the join operation at the target node.  
         [0022]      FIG. 1  illustrates a system  100  implemented in the form of a shared nothing distributed architecture. System  100  can include a host node  102 , a set of operator nodes  104   a - c , storage devices  106   a - c , and a network  108 . In general, host node  102  and operator nodes  104   a - c  may communicate with each other, for example, by passing messages over network  108 . In addition, operator nodes  104   a - c  may be coupled to one or more of storage devices  106   a - c . For example, operator nodes  104   a - c  may be connected to storage devices  106   a - c  through known types of interfaces, such as small computer system interface (“SCSI”), and integrated drive electronics (“IDE”) interfaces. The components of system  100  will now be further described.  
         [0023]     Host node  102  coordinates the operation of system  100 , such as query parsing, optimization, and compilation. For example, host node  102  may maintain the schema information for databases stored by system  100 . In addition, host node  102  may serve as an interface to other devices and systems on behalf of system  100 . For example, host  102  may provide an interface for ad-hoc queries, and embedded queries. Host node  102  may be implemented using known general purpose computers and processors.  
         [0024]     Operator nodes  104   a - c  serves as a site for performing various operations of system  100 . As shown in  FIG. 1 , operator nodes  104   a - c  can be implemented in a parallel, shared-nothing architecture. That is, operator nodes  104   a - c  are connected to their own disk drives (e.g., storage devices  106   a - c ) or random access memory and communicate with each other by sending messages through network  108 . Accordingly, one skilled in the art will recognize that the embodiment shown in  FIG. 1  can scale any number of operator nodes. Operator nodes  104   a - c  can be implemented using known general purpose computers and processors, such as those manufactured by the Intel Corporation. For example, operator nodes  104   a - c  can include an Intel Pentium™ processor, and several megabytes of memory.  
         [0025]     As noted, operator nodes  104   a - c  can be configured to communicate with each other based on passing messages over network  108 . Small messages, such as those less than 100 bytes can be sent as datagram packets. For larger messages, operator nodes  104   a - c  may open communications circuit with each other over which messages can be asynchronously transmitted. In some embodiments, after a message has been transmitted, operator nodes  104   a - c  may release the circuit. In addition, in some embodiments, operator nodes  104   a - c  may limit the maximum length of a message based on the physical memory available to each processor in operator nodes  104   a - c.    
         [0026]     Storage devices  106   a - c  provide local storage for their respective operator nodes, i.e., operator nodes  104   a - c . In some embodiments, system  100  may partition tuples of database across storage devices  106   a - c . Storage devices  106   a - c  can be implemented using known devices, such as disk drives or optical disk drives having embedded disk controllers. Storage devices  106   a - c  may also have its own random access memory to serve as a disk cache.  
         [0027]     Network  108  can be any number of interconnections, series of points or nodes, or communication paths for interconnecting the elements of system  100 , such as host  102  and operator nodes  104   a - c . Network  108  can be configured for communications over various spatial distances and can accommodate various types of data transmission technologies, whether it is public or private. Network  108  can be implemented using hardware and software that are well known to those skilled in the art. For example, network  108  can include a local area network, such as an Ethernet or token ring network. Other types of networks are also consistent with the principles of the present invention.  
         [0028]      FIG. 2  illustrates a software architecture for system  100 . As shown, host  102  can include a catalog manager  200 , a query manager  202 , and have access to schema data  204 . These components of host  102  will now be described.  
         [0029]     Catalog manager  200  can serve as a central repository of all conceptual and internal schema information for each database maintained by system  100 , such as schema data  204 . Since multiple users may have the same database open at once and a user may reside on a machine (not shown) other than host  102 , catalog manager  200  can also ensure consistency among copies of the database cached by each user. In some embodiments, catalog manager  200  can be implemented as a daemon process.  
         [0030]     Schema data  204  is a collection of meta-data that describes the relations of the databases maintained by system  100 . That is, schema data  204  describes the “layout” of a database that outlines the way data is organized into tables. In some embodiments, schema data  204  is configured using a series of SQL statements, such as “CREATE” statements. Schema data  204  can be loaded into the memory of host  102  when a database is first opened.  
         [0031]     Query manager  202  serves as an interface to a database and manages the execution of queries. For example, query manager  202  may cache information from schema data  204  and provide an interface for queries. Query manager  202  can be configured to process ad hoc queries entered by a user (not shown) as well as embedded queries requested by a program executing on another machine.  
         [0032]     In addition, query manager  202  can be configured to parse queries, determine a query execution plan, optimize the execution plan, and compile the query execution plan. Query manager  202  can use known relational techniques for query parsing, optimization, and code generation for executing a query. In some embodiments, query manager  202  employs hash-based algorithms for joins and other complex operations.  
         [0033]     Query manager  202  may assign scheduler processes, such as scheduler process  208 , to a query, and pass the query execution plan in query information message  206  to one or more of operator nodes  104   a - c . For example, query manager  202  can recognize that certain queries can be directed to only a subset of the nodes in system  100 . In the case of a single site query, query manager  202  may directly send the query in message  206  to the appropriate operator node for execution.  
         [0034]     A query may also require the involvement of multiple operator nodes, e.g., operator nodes  104   a - c . In order to control the execution of a multi-site query, one of operator nodes  104   a - c  may be designated a coordinator that is responsible for making global decisions related to the query. A node may be designated as a coordinator based on a variety of factors including, hardware capacity, location, user or administrator configuration, etc. For example, as shown in  FIG. 2 , operator node  104   a  has been designated as a coordinator site. However, one skilled in the art will recognize that any of operator nodes  104   a - c  can function as a coordinator site.  
         [0035]     In order to control the execution of a query among multiple nodes, a scheduler process  208  may be activated in a coordinator, e.g., operator node  104   a . Scheduler process  208  can run on any of operator nodes  104   a - c  and multiple instances of scheduler process  208  can run on a single processor or node of system  100 . In some embodiments, scheduler process  208  is activated when query manager  202  establishes a connection. Once scheduler process  208  has been activated, query manager  202  sends the compiled query in message  206 . In response to receipt of message  206 , scheduler process  208  in turn activates operator processes, such as operator processes  210   a - c , in operator nodes  104   a - c.    
         [0036]     One or more of operator nodes  104   a - c  may then work in conjunction to process the query specified in query information message  206 . Operator nodes  104   a - c  obtain results for the query. For example, operator nodes  104   a - c  may access their respective storage devices  106   a - c  to identify tuples or data that satisfy the query. Operator processes  210   a - c  then collects these results and passes them back to query manager  202  in the form of one or more messages. Query manager  202  reads the results of the query, compiles them, and returns the results through the query interface to the user or program from which the query was initiated.  
         [0037]      FIG. 3  illustrates an architecture of an operator node  104   a  which is part of the system  100  of  FIG. 1 . In particular,  FIG. 3  illustrates some of the components of operator node  104   a . As shown, operator node  104   a  may comprise a processor  300 , a communications interface  302 , and a memory  304 .  
         [0038]     Processor  300  controls the operation of operator node  104   a . In particular, processor  300  interprets and executes instructions provided to operator node  104   a , such as instructions from operator process  210   a . Processor  300  can be implemented as any type of processor, such as a general purpose processors manufactured by the Intel Corporation.  
         [0039]     Communications interface  302  couples operator node  104   a  to network  108 . Communications interface  302  can be implemented using known hardware and software. For example, communications interface  302  can be configured as an Ethernet or token ring interface. Other types of communications interfaces are also consistent with the principles of the present invention.  
         [0040]     Memory  304  serves as a local main memory for operator node  104   a . Memory  304  can be implemented using known types of memory, such as a random access memory. In addition, as shown in  FIG. 3 , memory  304  may include a hash code buffer  306 , a routing table  308 , and a buffer space  310 . These portions of memory  304  will now be further described.  
         [0041]     The hash code buffer space  306  serves as a temporary storage location for hash codes used by operator node  104   a . As noted, in some embodiments, operator node  104  may use various hash codes to perform actions related to a query. Such hash codes and their associated algorithms are known to those skilled in the art.  
         [0042]     Routing table  308  indicates where operator node  104   a  should route or send messages for portions of a query. Buffer space  310  serves as a temporary location for the data in these messages. One example of the relationship between routing table  308  and buffer space  310  will now be described with reference to  FIG. 4 .  
         [0043]      FIG. 4  illustrates an example of a routing table and buffer space in a memory of an operator node, such as the operator node  104   a  of  FIG. 3 . As shown, memory  304  may comprise routing table  308  and buffer space  310 . In addition, buffer space  310  may be partitioned into buffer groups  312   a  and  312   b . Buffer groups  312   a  and  312   b  may further comprise data buffers  314   a - d  and  314   e - h  respectively. One skilled in the art will recognize that buffer space  310  can include any number of buffer groups and each buffer group can include any number of data buffers.  
         [0044]     During query processing, results for a query may be temporarily stored in data buffers  314   a - h . When the results are to be sent to another node, the results are routed based on information in routing table  308 . For example, as shown, data buffers  314   a - d  are routed to operator node  104   b  and data buffers  314   e - h  are routed to operator node  104   c . Of course one skilled in the art will recognize that memory  304  can be configured in a variety of ways consistent with the principles of the embodiment.  
         [0045]      FIGS. 5A and 5B  illustrate a process flow related to a join operation of a query involving multiple operator nodes of the system of  FIG. 1 . In general, when a source node retrieves a tuple for a join operation, it is stored in a data buffer assigned to a target node that is also involved with the join operation. Also, the source node determines one or more hash codes from the tuple and stores these hash codes in a separate buffer, i.e., a hash code buffer. The source operator node then continues processing with other query operations and allows data to accumulate in the data buffers and the hash code buffer corresponding to each target operator node. However, when one or more of the data buffers is full, the first operator node sends the hash code buffer to the target node.  
         [0046]     The target operator node creates bit filters from the tuples in the relation prior to or during the initial phase of the join operation. On receiving the hash code buffer from the source operator node, the target operator node creates a bitmap based on the result of checking each hash code against its set of bit filters. The target node then sends the bitmap back to the source node. The source node uses the bitmap to filter out data tuples that will not participate in the join operation. The source node then sends the potentially reduced set of tuples to the target node, where the join operation may be completed. The result of the join operation may then be provided by the target operator node.  
         [0047]     In some embodiments, the size of the hash codes and bitmap buffers is smaller than the size of the tuples. For example, in some embodiments, the size of a hash code is 4 bytes, and the size of a bitmap entry per tuple is one bit.  
         [0048]     In addition, in some embodiments, communications between operator nodes  104   a - c  are asynchronous, so that the sending of messages and the processing of tuples can be overlapped. Furthermore, in some embodiments, the data buffers are partitioned into groups of buffers and the hash code buffer are sent after a certain number of data buffers have been filled for a given destination or target operator node.  
         [0049]     For purposes of explanation, the following description relates a join operation involving data at operator nodes  104   a  and  104   c . Either operator node  104   a  or  104   c  may serve as the coordinator this operation. The various stages for executing the join operation will now be described.  
         [0050]     In stage  600 , operator nodes  104   a  and  104   c  receive information in message  206  related the operations for a query. For example, operator nodes  104   a  and  104   c  may receive information in message  206  related to a join operation. This information may be passed to operator nodes  104   a  and  104   b  from scheduler process  208 , which in turn activates operator processes  210   a  and  210   c . Operator processes  210   a  and  210   c  may then activate operator nodes  104   a  and  104   c  to perform the operations requested in the information of message  206 . Processing may then proceed in parallel in nodes  104   a  and  104   c . The processing in node  104   a  will now be described.  
         [0051]     In stage  602 , operator node  104   a  may perform a projection related to the join operation. In particular, operator node  104   a  may access storage device  106   a  and select various tuples from tables in storage device  106  based on predicates of the requested query operation. Storage device  106   a  may then provide selected tuples  604  to operator node  104   a . Processing then flows to stage  606  in operator node  104   a.    
         [0052]     In stage  606 , source operator node  104   a  buffers tuples  604  into a data buffer designated for target operator node  104   c . For example, operator node  104   a  may store tuples from tuples  604  that are destined for node  104   c  into data buffer  314   e  of buffer group  312   b . Processing then flows to stage  608  in operator node  104   a.    
         [0053]     In stage  608 , operator node  104   a  encodes tuples  604  into hash codes. For example, operator node  104   a  may encode tuples  604  for a join operation based on the well known hybrid hash join algorithm. In particular, the hash codes may be obtained by a hash function that maps the joining key to a hash code. Operator node  104   a  may then store the hash codes into hash code buffer  306  of memory  304 . Processing then flows to stage  610 .  
         [0054]     In stage  610 , operator node  104   a  determines whether a threshold level has been reached in buffer space  310 . In some embodiments, the threshold level is based on buffer level data  612  from the data buffers in buffer space  310 . For example, the threshold level may be reached when a single one of data buffers  314   a - d  or  314   e - h  is full or substantially full. Alternatively, the threshold level may be based on a predetermined number of data buffers  314   a - d  or  314   e - h  are full or substantially full. If the threshold has not been reached, processing in operator node  104   a  repeats at stage  606 . That is, operator node  104  continues to accumulate tuples into its data buffers and build up the hash code buffer corresponding to each target operator node.  
         [0055]     However, if the threshold has been reached, then processing flows to stage  613  that asynchronously triggers stage  614  and also goes back to stage  606 . In stage  614 , operator node  104   a  transmits the hash codes buffer to the corresponding target operator node  104   c . In particular, operator node  104   a  retrieves the hash codes buffer from hash code buffer space  306  and references routing table  310 . Operator node  104   a  may then send one or more messages, such as hash code message  616 , through network  108  to operator node  104   c . Asynchronously, source node  104   a  continues processing at  606  to fill data buffers and hash code buffers for other target nodes. On receiving the hash code buffer, processing at target operator node  104   c  is active at stage  618 .  
         [0056]     At stage  618 , operator node  104   c  scans the hash codes based on its bits filters built after receiving relevant data tuples from the second relation received at stage  600  or possibly from the storage device  106   c  based on the query information. In particular, operator node  104   c  may scan bit filter data  622  in its local main memory. Bit filter data  622  serves as a summary representation of the tables stored in storage device  106   c  and may be determined based on well known algorithms. Processing then flows to stage  620 .  
         [0057]     In stage  620 , operator node  104   c  generates a bit map that indicates the data in bit filter data  622  which matches the hash codes in message  616 . Processing then flows to stage  624  in  FIG. 5B .  
         [0058]     In stage  624 , operator node  104   c  sends the bit map back to the source operator node  104   a . In particular, target operator node  104   c  refers to its routing table  310  and formats a bit map message  626 . Operator node  104   c  then sends bit map message  626  through network  108  back to operator node  104   a . Processing then continues at operator node  104   a  in stage  628 .  
         [0059]     In stage  628 , operator node  104   a  probes for non-matches to data in bit map message  626 . In particular, operator node  104   a  scans data buffer for example,  314   e  based on bit map message  626  and filters out non-matching tuples. Processing then flows to stage  632 .  
         [0060]     In stage  632 , operator node  104   a  sends potentially reduced non-filtered data tuples  630  to operator node  104   c  to complete the join operation. In particular, source operator node  104   a  sends the reduced data tuple message  634  to operator node  104   c  through network  108 . Message  634  may comprise one or more of data buffers  314   e - h . In addition, in some embodiments, the number of data buffers sent in message  634  is smaller than the threshold level depending on the selectivity of the bit filters and the match with the hash codes. Processing then continues at operator node  104   c  in stage  636 .  
         [0061]     In stage  636 , operator node  104   c  performs the join operation based on matching tuple message  634  and tuples retrieved from storage device  106   c . Processing then flows to stage  638 , where operator node  104   c  determines a result of the join operation and may then return the matched tuples for further processing in the query or store the result, for example, in storage device  106   c . Alternatively, operator node  104   c  may send the result to host  102 . In addition, operator nodes  104   a  and  104   c  and host  102  may exchange one or more control messages to indicate to each other that processing for the join operation has been completed. Processing is complete after all the tuples relevant to the query at the source nodes and the tuples at all the target nodes have been matched.  
         [0062]     Other embodiments of the invention will be apparent to those skilled in the art from consideration of the specification and practice of the invention disclosed herein. It is intended that the specification and examples be considered as exemplary only, with a true scope and spirit of the invention being indicated by the following claims.