2006-03-07 Assigned to APPLIED MINDS, INC. reassignment APPLIED MINDS, INC. ASSIGNMENT OF ASSIGNORS INTEREST (SEE DOCUMENT FOR DETAILS). Assignors: BAX, ERIC, CALLEJAS, AUGUSTO, HILLIS, W. DANIEL, KAO, HARRY, KOLEHMAINEN, MATHIAS L.
Disclosed are a method and apparatus for limiting the number of results returned by a data store in response to a query. Upon receiving an initial query, the data store returns a page of results that includes a subset of the data items within the data store satisfying the conditions of the query. The data store also provides a marker indicating the extent of the set of data items. If a subsequent query that requests additional results which satisfy the same query conditions and that includes the marker is received, the data store returns a page of results that includes a subset of data items that are disjoint from the initial subset, and provides an updated marker which indicates the extent of the union of the initial and subsequent subsets. If still further results are desired from the data store, an additional query containing the updated marker may be submitted.
This application claims priority to U.S. provisional patent application Ser. No. 60/636,822, entitled "Method and Apparatus for Dynamically Generating Assertion-Based Pre-sentations of Information" and filed 2004 Dec. 15, the entirety of which is incorporated herein by this referenced thereto. The application is also a continuation-in-part of U.S. utility patent application Ser. No. 11/177,789, entitled "Anti-Item for Deletion of Content in a Distributed Datastore" and filed 2005 Jul. 8, now U.S. Pat. No. 7,774,308 the entirety of which is incorporated herein by this reference thereto.
The invention relates to data stores, and in particular to large scale distributed data stores.
In recent years, the need for reliable and efficient storage of large amounts of data has increased dramatically. Indeed the need for extensive storage has even outpaced the remarkable advances in storage technology, e.g. the increasing capacity and decreasing cost of hard disks, and processing power. As a result, storage of truly large amounts of data on a single server is in many cases inpractical, and a distributed approach is desirable. Furthermore, even in those cases where storage can be handled by a single machine, a distributed data store may offer superior reliability and more efficient load handling.
At a recent conference of database users, Adam Bosworth envisioned a massive distributed data store through which "anyone can get to any data anywhere anytime". (Database Requirements in the Age of Scalable Services; O'Reilly MySQL Users Conference; Santa Clara, Calif.; (Apr. 18-21, 2005)). The data store would be completely scalable, use inexpensive, commodity components, and be capable of handling billions of transactions a day. Most importantly, it would allow even novice computer users to serve structured data to a world wide community of users conveniently, leading to an explosion of accessible information.
Partitioning. Data storage and retrieval and the associated workloads would be distributed across many nodes within a storage network.
Caching. Data would be stored locally on a temporary basis to minimize the effects of surges ill requests for particular items within the data store.
Stateless Nodes. Responding to queries would require a node to maintain only a minimum amount of state information.
Coarse-Grained Interactions. Clients, servers, and other network nodes would interact relatively infrequently, sending, receiving, and operating upon chunks of data at a time.
To date, substantial effort has been directed at developing very large databases (VLDBs) that exhibit some of these characteristics. Notable examples include the Mariposa wide-area distributed database (Michael Stonebreaker, Paul M. Aoki, Witold Litwiu, Avi Pfeffer, Adamn Sah, Jeff Sidell, Carl Staelin, Andrew Yu; Mariposa: a wide area distributed database system, VLDB Journal: 5(1):48-63; (1996)) and the PIER massively distributed query engine (Ryan Huebsch, Joseph M Hellerstein. Nick Lanham, Boon Thau Loo, Scott Shenker, Ion Stoica; Querying the Internet with PIER; Proceedings of the 29th VLDB Conference; Berlin, Germany; (2003)).
Many VLDB systems, including the PIER, query engine, are based upon content addressable networks CANs). A CAN is based upon a multi-dimensional Cartesian coordinate space that spans the entirety of the data to be stored. Each computer that stores data is considered as a node within a graph. As a new node is added to the graph, it is dynamically assigned a subspace of the Cartesian coordinate space within the subspace of the node to which it is connected. The newly added node is henceforth responsible for storing an(d returning data items within the assigned subspace.
Inserting data into a distributed system and searching for data within, i.e. submitting queries to, a distributed system present well known challenges. The difficultly of responding to such requests may be simplified greatly if the algorithms used allow each node to consider only local information in responding to the request. In the case of a CAN, each node stores information indicating the subspace assigned to each of its nearest neighbors within the graph, and a, greedy forwarding algorithm is used to propagate insert and query requests. Specifically, a node passes each request to the neighboring node with the assigned subspace closest to the location of the inserted or requested item within the Cartesian space. Each node thus acts upon purely local information, and does not need to coordinate with neighboring nodes to make routing decisions.
Nonetheless, difficulties may arise in distributed data stores when responding to search requests. Specifically, if a query associated with a search is propagated in an outwardly fanning manner to a number of nodes within the data store, and the data returned from the nodes are aggregated along the return path to the user, a bottleneck may ensue. This effect is particularly acute for those queries that entail the return of large amounts of data. As a data store grows larger and stores more data, this is increasingly the case.
It would therefore be desirable to provide a system for responding to queries of a distributed data store that minimizes the likelihood of a bottleneck should the act of responding to a query entail the return of large amounts of data. Furthermore, it would be desirable for the system to operate in a manner that is consistent with the goal of stateless nodes. Finally., the system should not require a user to anticipate a potential bottleneck but, rather, function in an automatic manner.
The invention provides a method and apparatus for limiting the number of results returned by a data store in response to a query. Upon receiving an initial query, the data store returns a page of results that includes a subset of the data items within the data store satisfying the conditions of the query. The data store also provides a marker indicating the extent of the set of data items. If a subsequent query that requests additional results which satisfy the same query conditions and that includes the marker is received, the data store returns a page of results that includes a subset of data items that are disjoint from the initial subset, and provides an updated marker which indicates the extent of the union of the initial and subsequent subsets. If still further results are desired from the data store, an additional query containing the updated marker may be submitted.
The invention thus provides a solution to the bottlenecking problem that arises in distributed data stores when responding to queries having very large result sets. By providing a paging mechanism to limit the number of results returned the invention restricts the amount of node resources used by a single query in a given period of time, thus preserving sufficient bandwidth for the execution of other data store operations.
The number of data items included in each successive page of results may be specified upon sub-mission of the query. Alternatively, the number of data items included in each successive page may be based upon a default page size specified for the data store.
The data items within the data store are characterized by a total ordering that provides an ordinal ranking for each data item. The extent of the results returned may be defined by a greatest ordinal ranking. In the preferred embodiment of the invention, the ordinal ranking is based on a globally unique identifier (GUID). In responding to an initial query, the data store returns a page of results including the data items within the data store with GUIDs having lowest ordinal ranking. The marker provided is the GUID having the greatest ordinal ranking among the GUIDs of the data items within the page of results. In responding to subsequent queries, the data store retrieves from storage the data items having GUIDs having lowest ordinal ranking among the data items with GUIDs having greater ordinial ranking than that of the marker.
In one embodiment of the invention the GUID includes a timestamp. In another embodiment the GUID includes a unique machine identifier. In the preferred embodiment of the invention the GUID includes both a timestamp and a unique machine identifier, and it is stored within the data store.
Also in the preferred embodiment of the invention, the data store is a distributed, ordered tuple store composed of a number of nodes which are connected in a directed acyclic graph. Data items are stored by the nodes as tuples containing values stored in one or more fields. Preferably, each of the tuples within the data store includes a GUID stored as a field within the tuple, thus providing a basis for a total ordering of the tuples. Alternatively, the total ordering may be based on the values stored in one or more of the fields of each tuple.
The distributed, ordered tuple store may store data items on both a permanent and temporary basis, using disk- or RAM-based storage. The nodes within the distributed, ordered tuple store include servers that are capable of receiving a query from a client, respositories that are capable of storing data items on a permanent basis, and relays that do not store data items on a permanent basis.
A query submitted to the distributed, ordered tuple store includes a range of values specified for at least one of the fields. The range of values specified for a field may be a wildcard, indicating that the values of the field are not considered in returning the page of results.
To obtain results to a query, a node partitions the query into one or more subquries; retrieves an internal page of tuples satisfying the conditions of the subquery from the local store of tuples within the node for each of a portion of the subqueries; forwards each of the remaining subqueries, as a query, to one or more other nodes within the distributed, ordered tuple store; receives a return page from each node to which a subquery was forwarded; merges the return pages and the internal pages to create a, merge(l return) page; and returns the merged return page as a return page. In the case of the node receiving the query from the client, the return page is returned to the client.
Preferably, the determination of which subqueries to forward and which subqueries to retrieve from the local store is based upon a storage range of the node that indicates which tuples are stored within the local store. The forwarding operation is based upon a coverage of each of the nodes to which the node is connected, indicating which tuples are accessible to each of the nodes. Finally, the merging operation preferably involves selecting the data items having lowest ordinal ranking from among the return pages and internal pages.
FIG. 12 shows a subsequent page of results returned to the client for a subsequent query following the query of FIG. 10.
The invention provides a method and apparatus for limiting the number of results returned by a data store in response to a query. The invention is best understood with reference to a detailed description of a preferred embodiment of the invention in which the data store is a distributed, ordered tuple store.
FIG. 1 shows the structure of a distributed, ordered tuple store according to the invention. The distributed ordered tuple store consists of multiple interconnected nodes. In the preferred embodiment of the invention, the nodes are connected in a directed acyclic graph as shown in FIG. 1, in which six interconnected nodes (Na, Nb, Nc, Nd, Ne, and Nf) are illustrated.
FIG. 2 shows the notions of superior nodes, inferior nodes, and rank within a distributed., ordered tuple store according to the invention. As illustrated in the FIG. 2, nodes may be characterized by a rank. Nodes of lower rank are termed inferior, and nodes of higher rank are termed superior. In the preferred embodiment of the invention nodes may only seek data from superior nodes, as indicated by the direction of the arrows in FIG. 2. That is, requests for data may be passed to superior nodes but may not be passed to inferior nodes. In this sense, superior nodes are said to lie upstream of inferior nodes, and inferior nodes are said to lie downstream of superior nodes. Each node that is directly connected to a downstream node by an upstream edge is an upstream neighbor of the downstream node. The set of upstream neighbors of a node is the termed the upstream neighborhood of the node.
The nodes within a distributed ordered tuple store may be further classified as one or more of several node types. A repository is a node that stores a portion of the data within the distributed, ordered tuple store on a permanent basis, e.g. using disk-based storage. A relay is an intermediate node that connects to an upstream node, but does not store data on a permanent basis. Finally, a server is a node that facilitates communication with a computing devices outside the distributed, ordered tuple store, called a client. A node may, at its discretion, store a portion of the data within the distributed, ordered tuple store on a temporary basis, e.g. using a RAM-based cache. Also, a single node may be more than one of a repository, relay, and server.
FIG. 3 shows a distributed, ordered tuple store incorporating nodes of several types according to the invention. In FIG. 3, node Na of FIG. 1 is a server S that supports a connection to a client. Node Na is also a repository R. Node Na is connected to two additional repositories R (Nb and Nc of FIG. 1). One of these repositories is further connected to an intermediate relay node I (Nd of FIG. 1). The relay is in turn connected to two more repositories R (Ne and Nf of FIG. 1).
Note that a single distributed, ordered tuple store may contain data within a single tuple space or it may contain data within multiple tuple spaces.
representing, for example, the age, name, and pet of a particular person.
indicates that a node is responsible for storing tuples (2, 9), (2,10), (3, 9), and (3, 10).
FIG. 4 shows a set of hosting ranges for the distributed, ordered tuple store of FIG. 1.
As noted above, nodes may also store data on a temporary basis, e.g. by using a RANM-based cache. The storage range of a node is the union of its hosting range and the tile or tiles of data within its temporary storage.
FIG. 5 shows the coverages resulting from the hosting ranges of FIG. 4. The coverage of a node is the union of its hosting range and the hosting ranges of the nodes in its upstream neighborhood. In the preferred embodiment of the invention in which the nodes are connected in a directed acyclic graph the coverage is defined by the superior nodes to which the node is connected.
A node is said to have full coverage of a tuple space if its coverage spans the entire tuple space. For any tuple within the tuple space, a, node with full coverage either hosts the tuple itself, or is connected to a superior node that provides coverage of the tuple. By definition, a node is of rank zero if and only if it has full coverage.
So that each node may calculate its coverage, each node advertises its coverage to the inferior nodes to which it is connected. In the simplest embodiments of the invention, an administrator configures the network of nodes, explicitly indicating the connectivity between nodes, and the hosting ranges of each. This calculation must be performed once upon system configuration. In more sophisticated embodiments of the distributed, ordered tuple store, the connectivity of nodes and the hosting ranges of the nodes may be adjusted dynamically in response to measured system loads. In such systems, the advertising of coverage occurs on all ongoing basis.
In the preferred embodiment of the invention, a distributed, ordered tuple store supports insert, delete, and query operations. As noted above, a client communicates with a server to initiate an operation and to receive any data returned by the operation, e.g. in the case of a query operation. Briefly, an insert operation adds a specified set of tuples to the distributed, ordered tuple store, a delete operation removes a specified set of tuples from the distributed ordered tuple store, and a query operation retrieves a specified set of tuples from within the distributed, ordered tuple store.
Insertion of data into the distributed, ordered tuple store is performed by a client that connects to a server within the distributed, ordered tuple store. In the preferred embodiment of the invention, requests to insert data are only passed from inferior nodes to superior nodes. Further, to ensure that any tuple within the tuple space can be inserted properly, a client connects only to nodes of rank zero.
FIG. 6 shows the insertion of a tuple according to the invention. The client inserts the tuple (5, 6) at node Na. Because node Na is of rank zero, the inserted tuple is within its coverage. However, the inserted tuple is not within the hosting range of node Na, so node Na does not store the tuple locally. Node Na then considers the coverage of the superior nodes to which it is connected and passes the insertion request to one or more of those superior nodes that provide coverage for the inserted tuple. The number of nodes to which the tuple is passed is based on the amount of redundancy desired in the distributed, ordered tuple store. In the distributed, ordered tuple store of FIG. 5, both nodes Nb and Nc provide coverage for the tuple (5, 6), and we consider the case in which node Na passes the insertion request to both nodes.
Nb receives the insertion request. The inserted tuple is within the hosting range of node Nb, so node Nb stores the tuple locally. After storing the tuple locally, node Nb considers the coverage of the superior nodes to which it is connected. The inserted tuple is within the coverage of node Nc but not within the coverage of node Nd. Node Nb therefore passes the insertion request to node Nc but not to node Nd.
Thus, node NC receives two insertion requests for the tuple (5, 6). Upon receipt of the first insertion request, node NC determines that the tuple is within its hosting range. The tuple is therefore stored locally. Upon receipt of the second insertion request, node Nc again determines that the tuple is within its hosting range. However, node Nc notes that it has already received the insertion request and that the inserted tuple is already within its local storage. Node Nc therefore ignores the second insertion request. Because node Nc is not connected to any superior nodes, the insertion operation terminates.
The deletion field Fd is added solely to differentiate tuples from anti-tuples. All tuples are of the form (a, b, 0), and all anti-tuples are of the form (c, d, 1). The deletion field is not used in determining where within the data store a particular tuple is stored. That is, all nodes must store both tuples and anti-tuples.
Actual deletion from the distributed, ordered tuple store proceeds in a manner highly analagous to the insertion operation. A client connects to a node of rank zero, and requests, for example, the deletion of a tuple (5,6, 0) by inserting the corresponding anti-tuple (5, 6, 1). The insertion of the anti-tuple proceeds as the insertion of the tuple in FIG. 6. Arrival of the anti-tuple at the same location as the tuple results in deletion of the tuple. Actual removal of the tuple from storage may be performed in several ways. Most simply, upon arrival of an anti-tuple, the storing node checks for the presence of the corresponding tuple. If the corresponding tuple is present, it is removed from storage. Alternatively, the anti-tuple may be stored alongside the tuple, and optionally removed at a later tuple through a garbage collection process. In this approach, or in a, approach in which the tuple is stored indefinitely, the presence of the anti-tuple alongside the tuple results in the exclusion of the tuple from any subsequent query operations.
A user wishing to obtain data items from the distributed, ordered tuple store initiates a search by specifying a set of search conditions. Preferably, the search conditions are submitted via a client. To retrieve the desired data items form the distributed ordered tuple store, the client submits a query to the distributed, ordered tuple store with query conditions corresponding to the search conditions. Preferably, the search conditions and query conditions are expressed as field ranges. The result set of a query, and by extension the search, is the zero or more tuples within the distributed, ordered tuple store that satisfy the query conditions. A tuple satisfies the conditions if, for each specified field range, the value of the field within the tuple is within the specified range. That is, the result set of a query is all tuples within the distributed, ordered tuple store that lie within the subspace of the tuple space specified in the query.
returns a result set containing the tuples (6, 7), (6, 8), and (6, 9), if they are present in the distributed, ordered tuple store.
The distributed, ordered tuple store handles a query by fanning out the query over a query tree. The query is fanned out through division into subqueries that are forwarded as queries to upstream nodes. The root of the query tree is the server at which the query was submitted. A leaf of the query tree is a node that handles a received query without forwarding subqueries to nodes in its upstream neighborhood. Often, the leaves of the query tree are repositories that retrieve the requested data from permanent storage, though other types of nodes may serve as leaves by retrieving cached data.
Upon receiving a query, a node partitions the query into a set of subqueries. The node retrieves a set of tuples from storage for those subqueries that are within its hosting range and, optionally, those subqueries that are within its storage range but not its hosting range, i.e. temporarily stored items. Each remaining subquery is forwarded, as a query, to one or more nodes in its upstream neighborhood that provide coverage for the subquery. The node then waits for sets of tuples from any upstream nodes to which it has forwarded a subquery. Upon receiving the set of tuples, the node merges the received sets of tuples with the sets of tuples retrieved from storage and returns the merged set of tuples to the downstream node from which it received the query. Generally, the merged set of tuples returned by a node, either to a downstream node or to the client, is returned in an ordered arrangement.
Note that in the case of a leaf, no subqueries are forwarded, and the set of triples retrieved from storage is returned immediately. In the case of a root, i.e. a server, the merged set of tuples is returned to the client. Because clients may only connect to servers, i.e. nodes of rank zero with full coverage, and because nodes only forward subqueries to nodes with a hosting range covering the subquery, a node never receives a query for which it offers no coverage. Finally, a node does not forward queries downstream, that is, towards peers of lower rank.
As noted above, merging set of tuples near the root of the query tree can lead to bottlenecking, especially in distributed, ordered tuple store exhibiting a high degree of branching and in response to queries that generate a large number of results. More detail about the merging and ordering processes and a mechanism for limiting the number of results returned to relieve the bottlenecking problem are provided in the discussions of total ordering and paging below.
The specification of a tuple may also allow for a wildcard or "don't care" field value for one or more of the fields within the tuple space. For example, the insertion operation propagates the anti-tuple (*, 6, 1) to all nodes with a hosting range encompassing a field value of 6 for the second field. Arrival at these nodes results in the deletion of all tuples of the form (f1, 6,0 ), where f1 is any field value selected from the field F1. Thus, in the network of FIG. 5, the tuple propagates to, and deletes, an), matching tuples stored at nodes Nb and Nc.
The symmetry of the insertion and deletion operation suggests that a wildcard field value may also be used in the insertion process. After insertion, a tuple containing a wildcard propagates to all nodes for which the explicitly specified field values are within the hosting range. Upon arrival at these hosting nodes, all tuples within the node's hosting range that match the tuple containing the wildcard are stored. However, because this operation generally results in a broad, untargeted insertion, it may not be appropriate for many distributed, ordered tuple stores.
is satisfied by all tuples within the distributed, ordered tuple store that have a value of 3 in the first field, regardless of the value in the second field. In some embodiments of the invention, the query syntax may not require the user to specify ranges for all fields, and those fields that are not specified are treated as if a wildcard value were specified.
The tuples within the distributed, ordered tuple store are characterized by a total ordering. The purpose of the total ordering is to provide an ordinal ranking for each data item that facilitates a deterministic ordering of the data items within the data store and within sets of tuples. Generally, the ordinal ranking is based upon one or more of the ordered lists of values from which the values of a tuple are chosen, i.e. fields described by Equation 1.
In the preferred embodiment of the invention, the ordinal ranking is based upon a globally unique identifier.
In one embodiment of the invention, a single field within the tuple space serves as a GUID. The ordinal ranking provided by a single field GUID is based upon the ordered list of values from which the field value within the tuple is chosen. e.g. the ordered list of values described in Equation 3. Often, the GUID is a field of numeric values with increasing ordinal rankings corresponding to increasing numeric values. For example, suppose the fields within a tuple correspond to the first name, last name, and social security number of an individual, with the social security number unique to each tuple. In this case, the social security number field may be used as a GUID, with the total ordering of the tuples corresponding to an ordering of individuals by increasing social security number.
A particularly useful single field GUID is a time-based GUID in which a server places a unique time-based value into a field within each tuple upon insertion into the distributed, ordered tuple store. The GUID may be a timestamp, for example a Unix timestamp indicating the number of seconds since the Unix Epoch on 1970 Jan. 1. Alternatively, if greater precision is desired, an NTP timestamp may be used. To ensure consistency among widely separated nodes, the timestamps is preferably based on a common time zone, such as GMT.
While the increased temporal resolution offered by timestamps such as the NTP timestamp make the likelihood of truly simultaneous insertions at disparate clients exceedingly unlikely, an identifier based on a timestamp alone may not guarantee global uniqueness. Therefore the preferred embodiment of the invention uses a GUID incorporating both a timestamp and a unique machine identifier indication the server at which the entry is inserted into the distributed ordered tuple store. The GUID may be formed by catenating the local server time at the time of insertion and the unique machine identifier, with the local time providing the higher order information.
The machine identifiers may be assigned to servers at the time of manufacture, on a random basis as the servers are started, or based on the physical location of the servers. If the server clock providing the local server time is not sufficiently precise to ensure that multiple operations do not occur at the same time, the value of a counter that increases with each successive operation of the server could be combined with the GUID containing the time and machine identifier.
The distributed method of generating a GUID described above avoids the network congestion that results in a system assigning GUIDs based on a counter maintained by a single machine. Furthermore, a GUID containing both time and server information can provide important functionality in the insertion operation described above and illustrated in FIG. 6. As noted, the topology and hosting ranges of some distributed, ordered tuple stores may result in a single insertion request arriving at a particular node more than once. By inspecting the GUID of an insertion request upon arrival a node may determine whether it has already handled the particular insertion quest. If so, the no(de may ignore the insertion request.
The unique nature of a GUID does not necessarily imply that a particular GUID appears only once within the distributed, ordered tuple store. As noted in the discussion of FIG. 6, it is possible in some embodiments of the invention for a particular tuple to be stored in multiple nodes. Such redundant storage does not corrupt the unique nature of the GUID, because it is still correct to say that no two tuples are characterized by the same GUID. Rather, the same tuple has merely been stored in more than one location.
In still other embodiments of the invention, the total ordering is based on multiple fields within the tuple. The particular fields incorporated into the total ordering are chosen such that no two tuples within the tuple space are characterized by the same ordinal ranking. In this approach, different operations, in particular different queries, may use different total orderings, with a user specifying a particular method of computing the ordinal rankings upon posing a query. The user is thus able to specify an ordering of the query results based on a successive sorting of tuples using the multiple fields. If the fields specified by the user do not contain a subset of fields, such that no two tuples have the same values for each of the fields in the subset, then fields are added to the specified fields until the specified fields include such a subset. The addition of fields may be performed automatically or by the user in response to a prompt. To ensure that the additional fields do not disrupt the intended ordering of the user, the additional fields are added as the least significant fields within the total ordering.
Returning to the example of a tuple with fields corresponding to the first name, last name, and social security number of an individual, suppose a user wishes to obtain results ordered by last name then first name, as in a phone book. The user may specify a total ordering based on the last name and first name fields, with the last name serving as the more significant field. If there exist tuples within the distributed, ordered tuple store that contain the same first and last name, the social security number field may be added to the total ordering as the least significant field. The addition of the social security number as the least significant field yields an ordering that is both deterministic and consistent with the intent of the total ordering initially specified by the user.
In many instances, a search submitted to a client by a user contains a set of search conditions satisfied by a large number of tuples within the distributed, ordered tuple store. For example, a set of search conditions that specifies within its set of ranges an entire tuple space produces a result set consisting of every tuple within the tuple space stored in the distributed, ordered tuple store. Search conditions incorporating wildcards may also yield very large result sets. When responding to queries with query conditions based on such search conditions, merging sets of tuples as they are forwarded downstream towards the server requires substantial resources at a number of nodes over a substantial period of time. Providing a complete result set when responding to such queries would render the distributed ordered tuple store unable to handle other operations while responding to the query. As a result, pending operations would accumulate at servers and overflow any buffers used to store pending operations. This bottlenecking problem merely worsens as the number of nodes and tuples in) a distributed, ordered tuple store increases. Clearly, there is a need to restrict the amount of node resources that can be consumed by a single query in a given period of time, preserving sufficient bandwidth for the execution of other operations.
To address this potential problem, the invention provides a paging mechanism in which the distributed, ordered tuple store successively returns subsets of a complete result set, termed pages, in response to searches with large result sets. Each successive page is produced in response to a query with query conditions corresponding to the search conditions. If, upon return of a page, more results are desired a subsequent query is submitted to the distributed, ordered tuple store to produce another page. A single search may thus entail the submission of multiple queries.
Paging allows each node to restrict the number of results returned in responding to a query, thus imparting a granular nature to the work performed and the data handled by the distributed, ordered tuple store. This allows multiple operations to be interleaved more readily. Consequently, paging allows comparatively simple operations, such as single tuple insertions and deletions, to complete while searches with large result sets continue to progress.
If a search in progress were to require information about the state of the search to be stored within the distributed, ordered tuple store between successive pages, having many searches in progress at once could potentially overload the nodes within the distributed, ordered tuple store. It is therefore desirable that the approach to paging be stateless, i.e. requiring little or no state to be stored within the distributed, ordered tuple store between the submission of successive queries and the return of successive pages. Furthermore, if a search in progress were to require that a range of the data within the distributed, ordered tuple store be locked, denying other operations such as insertions or deletions access to the data between the return of successive pages, then insertions and deletions would accumulate at nodes affected by the locks, potentially leading to a buffer overflow. It is therefore also desirable that the approach to paging be lock-free, i.e. operating without need for locking data within the distributed, ordered tuple store.
Accordingly, the proposed approach to paging operates without storing the state of searches or locking data in the distributed, ordered tuple store between the successive return of pages. This allows insertion and deletion operations to proceed interleaved with the handling of the successive query operations of a search, yet avoids the potential data overflow associated with a high volume of query operations, and avoids the potential buffer overflow associated with insertions and deletions attempting to access locked data.
a In response to an initial query received from a client, the distributed, ordered tuple store returns an initial page containing the subset of p tuples having lowest ordinal ranking.
To request a subsequent page, the client submits a subsequent query including a marker indicating the greatest ordinal ranking among the ordinal rankings of the results in the initial page.
In response to the subsequent query received from a client, the distributed, ordered tuple store returns a subsequent page containing the subset of p tuples having lowest ordinal ranking among those tuples having ordinal ranking greater than that of the marker.
Thus, the marker returned to the client by the distributed, ordered tuple store in response to the initial query in the form of the greatest ordinal ranking indicates the extent of the results returned in the response to the initial query. If the ordinal rankings are based on a GUID, the marker returned with the initial query is the GUID having the greatest ordinal ranking among the GUIDs of the tuples within the initial page.
By submitting the marker to the distributed, ordered tuple store within the subsequent query, the client ensures that the subsequent page of results is disjoint from the initial page of results. Specifically, the results returned in each subsequent page have ordinal ranking greater than the greatest ordinal ranking among the results in any previous page. Results are thus not repeated from one page to the next. In contrast to a cursor in a conventional database, which is a data structure stored within the database to indicate the current entry within a list of results, the marker is a stateless, virtual cursor in that it need not be stored within the distributed, ordered tuple store store between successive pages. Additionally, the use of a marker as described in the procedure above obviates the need for locking data between successive pages.
The number p of results returned in each page is termed the page size. A set of tuples of this size is termed a full page. The page size may be configured, e.g. by a system administrator, to be the same for every node, initial page, and subsequent page. Alternatively, the page size may be specified on a case by case basis, e.g. by a system administrator when configuring individual nodes within the system or by a user when submitting an initial or subsequent query. Such specification allows the page size to be varied based on any combination of node, query, and page. As noted above, the total ordering upon which paging is based may also be specified by the user upon submitting a query. In the preferred embodiment of the invention, if no total ordering or page size is specified, a default page size is implemented using ordinal rankings based on a time- and machine-based GUID. By always enforcing a default page size, the data store bounds the amount of computation and storage required of each node per query, and the benefits of paging are thus provided without requiring user knowledge of the paging mechanism.
If a node responding to a query can retrieve from within its local store of tuples at least enough results to produce a full page, the node returns a full page. Otherwise, the node returns as many results as possible. While page size may vary from node to node, the page size for each node in the query tree must be at least as large as the page size for any node along the path back to the root node to ensure proper functioning of the gathering and merging operations discussed below.
Partitions the query into one or more subqueries.
Retrieves from the local store a page of results for each of a subset of the subqueries. For each such subquery, the node retrieves from the local store the p tuples having lowest ordinal ranking among those tuples having ordinal ranking greater than that of the marker, if any marker was received within the query.
Forwards each of the remaining subqueries, as a query, to one or more nodes in its upstream neighborhood that provide coverage for the subquery.
Receives a return page from each upstream node to which it has forwarded a subquery.
Gathers a set of pages including those pages containing tuples obtained from the local store and those pages received in response to forwarded subqueries.
Merges the gathered pages to create a return page.
Returns the return page to the downstream node from which it received the query.
Several special cases of the above generalized procedure should be noted. In the case of a leaf of the query tree, because there is no upstream neighborhood to which the node may forward queries, the node itself retrieves a page of results for all of the subqueries. Effectively, then, a leaf may bypass the partitioning and forwarding operations and return results immediately. In the case of a root of the query tree. i.e. a server, the return page is returned to the client that submitted the query. Finally, it is important to note that the page of results is only limited by a marker if one is received within the query.
Generally, the partitioning operation is based on the storage range of the node and the coverages of the nodes in its upstream neighborhood. Preferably, the subqueries are constructed such that a maiximum amount of data can be retrieved from within the storage of the partitioning node. However, the node may also consider its current loading levels when partitioning. For example, a busy node may elect to forward on subqueries for which it could have retrieved results from the local store.
Orders the results within each of the gathered pages.
Allocates a set of pointers one pointer for each gathered page, each pointing to the tuple within the page with the lowest ordinal ranking.
Chooses from among the tuples designated by the pointers the tuple with the lowest ordinal ranking.
Adds the chosen tuple to the return page.
Advances the pointer corresponding to the chosen tuple to the tuple within the corresponding gathered page with the next lowest ordinal ranking. If there is no such tuple., the pointer is destroyed.
Because upstream nodes preferably provide return pages that are themselves ordered, the ordering operation that begins the pointer- based merging procedure mats in many instances be skipped. Notably, the merging procedure above itself yields an ordered return page. Thus, if the pointer-based merging procedure is implemented consistently across all nodes within a distributed ordered tuple store, it may be implemented without the initial ordering operation provided the leaves within the query tree return ordered results.
In GUID-based distributed, ordered tuple stores providing redundant storage, it may be desirable to ensure that the return page does not contain duplicate results, i.e. two copies of a tuple with a particular GUID. The merging process may additionally include a comparison operation to ensure that no two results within a return page have the same GUID. For example, in the pointer-based merging procedure, upon advancing the pointer to the tuple with the next lowest ordinal ranking the GUID of the next tuple may be compared to that of the tuple added to the return page. If the GUIDs are the same, the pointer is again advanced to the tuple with the next lowest ordinal ranking. Elimination of duplicate tuples may not be desirable in some instances, however, such as in queries posed to assess the degree of redundant storage within a distributed, ordered tuple store.
in which the final field F7 is a GUID. The distributed, ordered tuple store implements a default page size of two tuples.
FIG. 7 shows a set of hosting ranges for the distributed ordered tuple store of FIG. 1, and FIG. 8 shows the coverages resulting from the hosting ranges of FIG. 7. FIG. 9 shows a set of tuples stored in each node within the distributed ordered tuple store of FIG. 1 with the hosting ranges of FIG. 7.
Node Na retrieves results for subquery Qa a from the local store and forwards subquery Qa b oil to node Nb, as a query. As subquery Qa b is also within the coverage of node Nc, node Na could have partitioned the incoming query into three subqueries, sending one subquery to node Nb and one subquery to node Nc. However, because the coverage of node Nc is a subset of the coverage of node Nb, node Nc, in this example elects to forward a single subquery to node Nb in an effort to minimize the number of forwarded subqueries.
Node Nb retrieves results for subquery Qb b from the local store and forwards subqueries Qb c and Qb d as queries to nodes Nc and Nd, respectively.
Nodes Nc and Nd receive the respective queries. The partitioning and forwarding procedure continues in a similar manner at node Nd, which forwards subqueries to nodes Ne and Nf. Nodes Nc, Ne, and Nf are leaf nodes, and therefore do not forward any subqueries. In the case of the leaf nodes, the received queries are partitioned into a single subquery that is retrieved from the local store, as shown in FIG. 10.
containing the same tuples to node Nd. Note that the tuples retrieved and provided by node Ne are the p tuples within storage with GUIDs having lowest ordinal ranking. As noted, in this example p=2 and two tuples are provided in the return page.
Node Nd receives the return page from node Ne and similarly receives a return page from node Nf. Node Nd merges the two received return pages with the results it has retrieved from the local store. However, because node Nd is a relay node with no storage there are no such results. Node Nd therefore merely merges the two received return pages to create a return page containing the tuples with the lowest ordinal ranking among the received return pages based on the GUID. The resulting return page is provided to node Nb.
Node Nb merges pages Pd b, Pc b, and Pb b to create a return page containing the two tuples with the lowest ordinal ranking among the three pages. The resulting return page is provided to node Na.
is provided to the client as a response to the submitted query.
FIG. 12 shows a subsequent page of results returned to the client for a subsequent query following the query of FIG. 10. To request the subsequent page of results, the client submits a subsequent query similar to the initial query, but additionally including a marker indicating the GUID having highest ordinal ranking among the GUIDs of the tuples returned in the initial return page.
Consequently, those nodes that retrieved the tuples in the initial return page, i.e. nodes Nb and Ne, retrieve a different page of results from the local store in responding to the subsequent query.
Thus, as desired, the two tuples returned in response to the initial query are the two tuples with GUIDs having lowest ordinal ranking throughout the distributed, ordered tuple store, while the two tuples returned in response to the subsequent query are the tuples with GUIDs of third and fourth lowest ordinal ranking.
If additional storage is available at nodes between the return of successive pages, the speed of the paging process may be increased using the pointer-based merging procedure. Specifically, it is possible to reduce the number of results that must be provided in gathered pages for creation of return pages subsequent to the first return page.
Stores the ordered results within each gathered page that were not included in the return page as a partial gathered page.
depending on whether the gathered page was obtained from the local store or received from an upstream node.
Orders the results within the truncated page of results.
Appends the truncated page of results to the truncated gathered page to complete the gathered page.
The completed gathered page may then be merged, as described in the pointer-based merging procedure. Note that if no results from a gathered page are included in a return page, then no new results need be obtained from the source providing the gathered page, whether from local storage or an upstream node, in creating a subsequent return page. Similary, if the number of results remaining in a gathered page after removal of results for inclusion in the return page is at least as large as the number of results in the return page, then there is no need to obtain more results for the subsequent return page. Finally, if additional computational resources are available at a node after providing a return page, the node may use the resources to obtain and merge gathered pages in anticipation of a subsequent query, either from the local store or from nodes in its upstream neighborhood.
wherein submitting said updated particular marker to said data store within a subsequent query ensures that a subsequent page of results is disjoint from a previous page of results.
2. The method of claim 1, wherein the number of data items in any of said initial page and said subsequent page is specified for said query upon submission of said query.
3. The method of claim 1, wherein the number of data items in any of said initial page and said subsequent page is determined by a page size specified by an administrator of said data store.
5. The method of claim 1, wherein said ordinal ranking is based on a globally unique identifier.
a unique machine identifier identifying a server receiving said query.
7. The method of claim 5, wherein said globally unique identifier is stored within said data store.
8. The method of claim 1, wherein a methodology for computing said ordinal ranking is specified with said query.
9. The method of claim 1, wherein said data store comprises a distributed data store.
10. The method of claim 9, wherein said distributed data store comprises one or more nodes.
11. The method of claim 10, wherein said nodes are connected in a directed acyclic graph.
a relay that cannot store data items on a permanent basis.
wherein said return page comprises any of said initial page and said subsequent page of results for said receiving node.
14. The method of claim 13, wherein said internal pages, said return pages, and said merged return page each comprises a common number of data items.
15. The method of claim 13, wherein said merging step comprises the step of selecting data items having greatest ordinal ranking among a union of said return pages and said internal pages.
16. The method of claim 1, wherein said data store comprises a distributed, ordered tuple store comprising one or more nodes; and wherein said data items within said data store comprise tuples having values stored in one or more fields.
17. The method of claim 16, wherein said query conditions comprise a range of values specified for at least one of said fields.
18. The method of claim 16, wherein said query conditions comprise a wildcard specified for at least one of said fields for values in said at least one of said fields that are not considered in returning any of said initial page and said subsequent page of results.
19. The method of claim 16, wherein said ordinal ranking is based on a globally unique identifier and wherein said globally unique identifier is stored within said distributed, ordered tuple store as a value in one of said fields.
20. The method of claim 16, wherein said ordinal ranking is based on a globally unique identifier and wherein said globally unique identifier is calculated based on the values stored in one or more of said fields.
22. The method of claim 21, wherein said internal pages, said return pages, and said merged return page each comprise a common number of tuples.
23. The method of claim 21, wherein said portion and said remaining portion are determined based upon a storage range of said node, wherein said storage range indicates which of said tuples are stored within said local store of tuples.
24. The method of claim 21, wherein said forwarding step is based upon a coverage of each of said nodes to which said node is connected, wherein said coverage indicates which of said tuples are accessible to each of said nodes to which said node is connected.
27. The computer of claim 25, wherein said data store comprises a distributed data store comprising one or more nodes.
wherein said data items within said data store comprise tuples having values stored in one or more fields.
30. The computer of claim 29, wherein each of said tuples has an ordinal ranking within said distributed, ordered tuple store based on a globally unique identifier.
wherein submitting said marker to said data store within a subsequent query ensures that a subsequent page of results is disjoint from said page of results.
wherein said previous marker indicates an extent of a union of said previous page and said page.
wherein submitting said updated marker to said data store within a subsequent query ensures that a subsequent page of results is disjoint from a previous page of results.
wherein submitting said subsequent marker to said data store within a subsequent query ensures that a subsequent page of results is disjoint from a previous page of results.
"Content-Addressable Network Model", Information Visualization Cyberinfrastructure, SLIS Indiana University, Jul. 10, 2004, 3 pages.
Abiteboul, et al.; "Compact Labeling Schemes for Ancestors Queries"; 2001 Symposium on Discrete Algorithms; ISBN 0-89871-4907, pp. 547-556, specifically p. 554.
Adam Bosworth; IT Conversations New Ideas Through Your Headphones; Santa Clara, CA Apr. 18-25, 2005.
Cesa-Bianchi, et al.; "Minimizing regret with label efficient prediction"; Jun. 2005; IEEE, vol. 51, pp. 2152-2162.
Crespo, et al.; Semantic Overlay Networks for P2P systems; Jan. 2003; Technical report, Stanford University.
David Patrick Reed; Naming and Synchronization in a Decentralized Computer System; Sep. 1978; Massachusetts Institute of Technology.
Deng, Chao et al., "Multi-Agent Framework for Distributed Systems", Aug. 26-28, 2004; IEEE, vol. 1, 22-25.
Duong, et al.; "LSDX: a new labelling scheme for dynamically updating XML data"; Jan. 2005; ACM, pp. 185-193.
http://en.wikipedia.org/wiki/Content-addressable-network; Content Addressable Network; May 2005.
http://en.wikipedia.org/wiki/Content—addressable—network; Content Addressable Network; May 2005.
Jeff Dean; Google's Big Table; University of Washington Oct. 18, 2005; Google Blogoscoped 2003-2005 by Philipp Lenssen.
Larson, P.A.; Dynamic Hash Tables; Apr. 1988; Communications of the ACM, vol. 31, No. 4, pp. 446-457.
M. Stonebraker, P. Aoki, W. Litwin, A. Pferffer, A. Sah, J. Sidell, C. Sttttaelin, and A. Yu; Mariposa: A Wide-Area Distributed Database System; Department of Electrical Engineering & Computer Sciences, Univ. of Calif Berkeley, CA,: The VLDB Journal 1996.
Okubo, Takao et al., "An Optimistic Method for Updating Information in Distributed Collaborative Work", Aug. 19-21, 1998; IEEE; Kawasaki, Japan, Dec. 27, 2012, 400-405.
R. Huebsch, J. Hellerstein, N. Lanham, B. Loo, S. Shenker, and I. Stoica; Querying the Internet with PIER; Proceedings of the 29th VLDB Conference, Berlin, Germany, 2003.
S. Ratnasamy, P. Francis, M. Handley, R. Karp, and S. Shenker; A Scalable Content-Addressable Network. Proc. ACM SIGCOMM 2001, pp. 161-172, Aug. 2001.
Wade, S , "An Investigation into the Use of the Tuple Space Paradigm in Mobile Computing Environments", Ph.D Dissertation Lancaster University, England., Sep. 1999, 236 Pages.
Gupta et al. 2003 Approximate Range Selection Queries in Peer-to-Peer systems.