CN100565460C - Be used for method of managing data - Google Patents

Abstract

The invention describes the various technology of the performance that is used to improve shared-nothing database system, wherein, at least two nodes that move in the node of this shared-nothing database system can be shared the ground accessing disk.Especially, be provided in the proprietorial technology that changes the data in the shared-nothing database under the situation that does not change the position of data on long-time memory.Because the persistent storage position of data is not changed during the proprietorial transfer of data, therefore can more freely passes ownership, and have than physics and rearrange the other littler performance loss that causes by data.The various technology that are used to provide fast run-time reassignment of ownership have also been described.Since can run time between carry out reallocation, therefore needn't make no shared system off line carry out reallocation.In addition, these technical descriptions carry out reallocation as how relative fine granulation, avoid the entitlement of the minority data item on the node of reallocation in node and need to carry out a large amount of reallocation through the mass data of all nodes.

Description

Methods used to manage data

technical field

The present invention relates to techniques for managing data in a shared-nothing database system running on shared disk hardware.

Background technique

Multiprocessing computer systems generally fall into three categories: shared-everything systems, shared-disk systems, and shared-nothing systems. In all resource sharing systems, programs on all processors can directly access all volatile storage devices (hereinafter generally referred to as "memory") and all nonvolatile storage devices (hereinafter generally referred to as "disks") in the system. "). Therefore, advanced wiring between different computer components is required to provide all resource sharing functions. Additionally, there are scalability limitations as with all resource sharing structures.

In a shared disk system, processors and memory are grouped into nodes. Each node in the shared disk system itself can constitute a system for sharing all resources including multiprocessors and multimemory. Programs on all processors can access all disks in the system, but only programs on processors belonging to a particular node can directly access memory within a particular node. Shared disk systems generally require less wiring than all resource sharing systems. Shared disk systems can also easily accommodate unbalanced workload conditions because all nodes have access to all data. However, shared disk systems are susceptible to coherence overhead. For example, if a first node has modified data and a second node wants to read or modify that same data, several steps must be taken to ensure that the correct version of the data is provided to the second node.

In a shared-nothing system, all processors, memory, and disks are grouped into nodes. As in a shared-disk system, in a shared-nothing system, each node can itself constitute a shared-everything system or a shared-disk system. Only programs running on a specific node can directly access memory and disks within a specific node. Shared-nothing systems of the three general types of multiprocessing systems generally require a minimal amount of wiring between the various system components. However, shared-nothing systems are most susceptible to unbalanced workload conditions. For example, all data to be accessed during a particular task may exist on a particular node's disk. Therefore, only programs running within that node can be used to execute work granules, even if programs on other nodes remain idle.

Databases running on multi-node systems generally fall into two categories: shared-disk databases and shared-nothing databases.

shared disk database

Shared disk databases work together based on the assumption that all data managed by the database system is visible to all processing nodes available to the database system. Thus, in a shared disk database, the server can assign any job to a program on any node, regardless of the location of the disk containing the data to be accessed during the job.

Because all nodes have access to the same data, and each node has its own private cache, multiple versions of the same data item can exist in the caches of any number of multiple nodes. Unfortunately, this means that when a node requests a particular version of a particular data item, that node must coordinate with other nodes to have the particular version of the data item delivered to the requesting node. Thus, shared disk databases are said to operate on a "data transfer" principle, where data must be transferred to nodes that have been designated to process the data.

Such a data transfer requirement may result in a "ping". In particular, pinging occurs when a copy of a data item required by one node exists in another node's cache. Checking may require data items to be written to disk and then read from disk. The performance of the disk operations necessary to check can significantly reduce the performance of the database system.

Shared-disk databases can run on both shared-nothing and shared-disk computer systems. To run a shared-disk database on a shared-nothing computer system, software support can be added to the operating system or other hardware can be provided to allow programs to access remote disks.

no shared database

Shared-nothing databases assume that programs can only access data if the data is contained on disks belonging to the same node as the program. Therefore, if a particular node wants to perform an operation on a data item owned by another node, the particular node must send a request to the other node asking the other node to perform the operation. Thus, a shared-nothing database is said to perform "function transfer," rather than transfer data between nodes.

Because any given block of data is owned by only one node, only this one node (the "owner" of the data) will permanently have a copy of the data in its cache. Thus, there is no need for a cache coherency mechanism of the type required in a shared disk database system. In addition, the shared-nothing system does not suffer from the performance penalty associated with pinging since the node owning the data item is not required to save a cached version of the data item to disk so that another node can then deposit the data item into its cache.

Shared-nothing databases can run on both shared-disk multiprocessing systems and shared-nothing multiprocessing systems. To run a shared-nothing database on a shared-disk machine, a mechanism can be provided for partitioning the database and assigning ownership of each partition to a specific node.

The fact that only owning nodes can operate on data blocks means that the workload in a shared-nothing database can become wildly unbalanced. For example, in a system of ten nodes, 90% of all work requests may involve data owned by one of the nodes. Therefore, the one node is overworked, while the computing resources of the other nodes are underutilized. To "rebalance" the workload, the shared-nothing database can be taken offline, and the data (and its ownership) can be redistributed among the nodes. However, this process involves potentially moving large amounts of data, and may only temporarily resolve workload imbalances.

Contents of the invention

In order to solve one of the above-mentioned technical problems, the present invention provides a method for managing data, the method comprising the steps of: maintaining multiple persistent data items, the persistent data items including specific data items stored at specific locations on the persistent memory; assigning exclusive ownership of each of the persistent data items to one of the nodes, wherein, a particular node of said plurality of nodes is assigned exclusive ownership of said particular data item; when any node wants to perform an operation involving said particular data item, since said particular data item exists at said particular location, said node desiring said operation to be performed communicates said operation to said specific node for said specific node to perform said operation on said specific data item; when said specific node continues to operate, without sending reassigning ownership of the particular data item from the particular node to another node in the event that the particular data item is moved from the particular location on the persistent storage; after the reassignment, when any When a node wants to perform an operation involving said particular data item, said node expecting said operation to be performed, since said particular data item exists at said particular location, communicates said operation to said other nodes, with performing said operation on said particular data item by said other node; continuing to have a set of data items owned by said first node after said first node has been removed from said multi-node system; A request for an operation on the data item reassigns ownership of the data item from the first node to one or more other nodes. With this technical solution, the shared-nothing database system can be rebalanced and restored more effectively, ownership can be transferred more freely, and there is less performance loss than the physical rearrangement of data, and the reconfiguration can be performed at a relatively fine-grained level. Allocation, avoiding the need to perform massive redistribution of large amounts of data across all nodes just to redistribute ownership of a few data items on one of the nodes.

Description of drawings

The invention is described, but not limited to, by way of example in the accompanying drawings, in which like reference numerals indicate similar elements, in which:

Figure 1 is a block diagram illustrating a cluster comprising two shared disk subsystems according to an embodiment of the invention; and

Figure 2 is a block diagram of a computer system on which embodiments of the present invention may be implemented.

Detailed ways

Various techniques for improving the performance of shared-nothing database systems including shared disk storage systems are described below. In the following description, for purposes of explanation, numerous specific details are set forth in order to provide a thorough understanding of the present invention. It is evident, however, that the invention may be practiced without these specific details. In other instances, well-known structures and devices are shown in block diagram form in order to avoid unnecessarily obscuring the present invention.

Functional Overview

Various techniques for improving the performance of a shared-nothing database system in which at least two of the nodes running the shared-nothing database system have shared access to a disk are described below. As determined by the shared-nothing structure of database systems, each data block is still owned by only one node at any given time. However, the fact that at least some of the nodes running a shared-nothing database system have shared access to disk is exploited to more efficiently rebalance and restore a shared-nothing database system.

In particular, techniques are provided for changing the ownership of data in a shared-nothing database without changing the location of the data on memory. Since the persistent storage location of the data is not changed during the transfer of ownership of the data, ownership can be transferred more freely and with less performance penalty than a physical rearrangement of the data would cause.

Various techniques for providing fast run-time reallocation of ownership are also described. Because the reallocation can be performed during runtime, it is not necessary to take the shared-nothing system offline to perform the reallocation. In addition, these techniques describe how to perform redistribution at a relatively fine granularity, avoiding the need to perform extensive redistribution of large amounts of data across all nodes just to redistribute ownership of a few data items on one of the nodes. distribute.

Exemplary cluster including shared disk systems

Figure 1 is a block diagram illustrating a cluster 100 in which embodiments of the present invention may be practiced. Cluster 100 includes five nodes 102, 104, 106, 108, and 110 connected by an interconnect 130 that allows the nodes to communicate with each other. Group 100 includes two disks 150 and 152 . Nodes 102 , 104 , and 106 have access to disk 150 , and nodes 108 and 110 have access to disk 152 . Thus, the subsystem including nodes 102, 104, and 106 and disk 150 constitutes a first shared disk system, while the subsystem including nodes 108 and 110 and disk 152 constitutes a second shared disk system.

Farm 100 is an example of a relatively simple system that includes two shared disk subsystems with no overlapping membership between the shared disk subsystems. The actual system may be more complex than the group 100, with hundreds of nodes and hundreds of shared disks, and a many-to-many relationship between nodes and shared disks. In such a system, for example, a single node with access to many disks may be a member of several different shared-disk subsystems, where each shared-disk subsystem includes one of the shared disks and has access to All nodes that share the disk.

A shared-nothing database on a shared-disk system

For purposes of illustration, it will be assumed that a shared-nothing database system is running on cluster 110 , wherein the databases managed by the shared-nothing database system are stored on disks 150 and 152 . Based on the shared-nothing nature of the database system, data can be divided into five groups or partitions 112, 114, 116, 118, and 120. Each partition is assigned to a corresponding node. A node assigned to a partition is considered the sole owner of all data present in that partition. In this example, nodes 102, 104, 106, 108, and 110 own partitions 112, 114, 116, 118, and 120, respectively. Partitions 112 , 114 , and 118 owned by nodes (nodes 102 , 104 , and 106 ) that have access to disk 150 are stored on disk 150 . Similarly, partitions 118 and 120 owned by nodes with access to disk 152 (nodes 108 and 110 ) are stored on disk 152 .

As dictated by the shared-nothing nature of database systems running on the swarm 100, any data block is owned by at most one node at any given time. Additionally, access to shared data is coordinated through function transfers. For example, in the context of a database system supporting the SQL language, a node that does not own a particular data block can cause an operation on that data by sending a fragment of an SQL statement to a node that does own the data block.

ownership mapping

In order to perform function transfer efficiently, all nodes need to know which node owns which data. Therefore, an ownership map is established, wherein the ownership map indicates the ownership assignment of data to nodes. During runtime, different nodes refer to the ownership map to send SQL fragments to the correct node at runtime.

According to one embodiment, the data-to-node mapping need not be decided at compile time of the SQL (or any other database access language) statement. Instead, as will be described in more detail below, data-to-node mappings can be established and modified during runtime. Using the techniques described below, when ownership changes from one node with access to the disk on which the data resides to another node with access to the disk on which the data resides, it is possible to case to perform a change of ownership.

locking

A lock is a structure used to coordinate access to a resource among multiple entities that can access the resource. In the case of a shared-nothing database system, no global locking is required to coordinate access to user data in a shared-nothing database because any given block of data is only owned by a single node. However, because all nodes of a shared-nothing database require access to the ownership map, some locking may be required to prevent inconsistent updates to the ownership map.

According to one embodiment, a two-node locking scheme is used when the ownership of a data block is being redistributed from one node (the "old owner") to another node (the "new owner"). Additionally, a global locking mechanism can be used to control access to metadata associated with a shared-nothing database. Such metadata may include, for example, an ownership map.

Transfer ownership without moving data

According to one aspect of the invention, the ownership of data can be changed from one node (the original owner) to another node associated with the data (the new owner) without moving the data. For example, assume that a particular data item currently exists in partition 112 . The data item is owned by node 102 because it exists in partition 112 . In order to change ownership of the data to node 104 , the data must no longer belong to partition 112 but instead belong to partition 114 . In a traditional implementation of a shared-nothing database system, such a change of ownership would typically result in the data item being physically moved from one physical location on disk 150 corresponding to partition 112 to another physical location on disk 150 corresponding to partition 114. Location.

Conversely, according to embodiments of the present invention, partitions 112 and 114 are not necessarily physical partitions at specific locations on disk 150 . In contrast, partitions 112 and 114 are location-independent partitions that simply represent the collection of data items currently owned by nodes 102 and 104 , respectively, regardless of where a particular data item exists on disk 152 . Thus, since partitions 112 and 114 are location independent, data items can be moved from one partition to another (i.e., assigned from one owner to another) without any actual movement of the data on disk 150. an owner).

While changing the ownership of a data item does not require movement of the data item, it does require a change in the ownership mapping. Unlike user data in a shared-nothing database, the ownership map is shared across nodes. Therefore, parts of the ownership map can be cached in private caches of different nodes. Thus, in response to a change in ownership of the data item, the ownership map is changed and the cached copy of the affected portion of the ownership map is invalidated.

According to an alternative embodiment, ownership changes are performed similarly to schema changes of base objects. In particular, after a change is made to the ownership mapping, compiled statements referring to the ownership mapping are invalidated and recompiled to use the new ownership mapping.

Adding and removing nodes

During operation of cluster 100 , nodes may need to be added or removed from cluster 100 . In a traditional shared-nothing system, such an operation would involve frequently moving large amounts of data from a file or one physical partition of a disk to another file or another physical partition of a disk. By using location-independent partitioning, the only data that must be physically rearranged is that whose ownership is transferred to nodes that do not have access to the disks on which the data currently resides.

For example, suppose a new node X is added to cluster 100 , and node X has access to disk 152 but not disk 150 . Some data currently owned by nodes 102-110 may be redistributed to node X in order to rebalance the workload among the nodes. Since the data whose original owners were nodes 102 to 106 exists on disk 150 which is not accessible to node X, the data must be physically moved to disk 152 which is accessible to node X. However, since the data whose original owners were nodes 108 and 110 already exists on disk 152 to which node X has access, ownership of that data can be transferred to node X by updating the ownership map without moving the actual data .

Similarly, when a node is removed from the cluster 100, only data items that need to be physically rearranged need to be transferred to nodes that currently do not have access to the disks on which the data items reside. Data items whose ownership is transferred to nodes that have access to the disks on which the data resides need not be moved. For example, if node 102 is removed from swarm 100, and ownership of all data items previously held by node 102 is transferred to node 104, no data items need to be physically rearranged in response to the change in ownership.

gradual transfer of ownership

According to one embodiment, by performing ownership transfers gradually rather than abruptly, the performance penalty associated with large reallocations of data ownership in response to added or removed nodes may be mitigated. For example, when a new node is added to the swarm, the system may begin transferring ownership of few or no data items to the new node, rather than transferring ownership of enough data to keep the new node as busy as existing nodes. According to one embodiment, ownership is gradually transferred based on workload needs. For example, the transfer of data ownership can be triggered when the system detects that the workload of one of the nodes becomes excessive. In response to detecting an overworked node, some data items belonging to the overworked node may be assigned to previously added nodes. Gradually, more and more data items may be allocated from the overworked node to the new node until it is detected that the overworked node is no longer overworked.

On the other hand, reassignment of ownership can be triggered when the workload of existing nodes drops below a certain threshold. In particular, it is desirable to transfer some ownership responsibilities from other busy nodes to new nodes when the workload of busy nodes lightens, to avoid reallocation operations degrading the performance of already overworked nodes.

As for the gradual transfer of ownership from removed nodes, ownership transfers may, for example, be triggered when necessary. For example, if data item X is owned by the removed node, data item X may be reassigned to another node when it is detected that some nodes have requested operations involving data item X. Similarly, transferring ownership from a removed node to an existing node may be triggered when the workload of the existing node drops below a certain threshold.

Bucket-based partitioning

As mentioned above, the data managed by a shared-nothing database is partitioned, and the data in each partition is owned exclusively by one node. According to one embodiment, partitions are established by assigning data to logical storage segments, and then assigning each storage segment to a partition. Therefore, the mapping of data to nodes in the ownership mapping includes the mapping of data to storage segments and the mapping of storage segments to nodes.

According to one embodiment, the mapping of data to storage segments is established by applying a hash function to the name of each data item. Similarly, a bucket-to-node mapping can be established by applying another hash function to the identifier associated with the bucket. Alternatively, one or both of these maps can be built using range-based partitioning, or by simply enumerating each individual relationship. For example, one million data items can be mapped to fifty buckets by dividing the namespace of the data items into fifty extents. Fifty buckets can then be mapped to five nodes by storing a record for each bucket that (1) identifies the bucket and (2) identifies the node that is currently assigned the bucket.

The use of buckets significantly reduces the size of the ownership map relative to a map in which a separate map record is stored for each data item. Additionally, in embodiments where the number of buckets exceeds the number of nodes, the use of buckets makes it relatively easy to reallocate ownership to subsets of the data owned by a given node. For example, a new node may be allocated a single memory segment from a node currently allocated ten memory segments. Such reallocation would simply involve modifying, for that bucket, a record indicating the bucket-to-node mapping. The data-to-segment mapping of data being reallocated does not have to be changed.

As noted above, the mapping of data to buckets may be established by using any of a variety of techniques including, but not limited to, hash partitioning, range partitioning, or column values. If range-based partitioning is used and the number of ranges is not significantly larger than the number of nodes, the database server can use more Fine (narrower) scope to achieve desired number of buckets. If the range key is a value that can be changed, then in response to a change in the range key value for a particular data item, the data item is removed from its original storage segment and added to a new value corresponding to the data item's range key storage segment.

tree-based partitioning

Another way to partition data items managed by a database system into subsets is to use a hierarchical storage structure (e.g., a BTree) such that the upper level of the tree structure (e.g., the root) is owned by all nodes, and the lower levels (e.g., leaf nodes) are partitioned among the nodes. According to one embodiment, the tree structure comprises a plurality of subtrees, wherein each subtree is assigned to a specific node. In addition, each subordinate tree node corresponds to a set of data. A set of data associated with subordinate tree nodes is owned by nodes associated with subtrees that include tree nodes.

In such an embodiment, when the ownership of a subtree changes, the superior is invalidated through a locking/broadcasting scheme. Subordinate pointers are modified to move ownership of subtrees under different nodes.

Handle dirty versions during redistribution

As described above, when the new owner of the data has access to the disk on which the data resides, the ownership of the data is changed by reallocating storage segments to nodes without physically moving the physical location of the data on the disk . However, it is possible that the original owner has a "dirty" version of one or more reallocated data items in its volatile memory. A dirty version of a data item is a version that includes changes that do not affect the version currently existing on disk.

According to one embodiment, dirty versions of data items are written to shared disk as part of an ownership transfer operation. Thus, when a new owner reads from disk a data item that it has newly acquired ownership of, the version of the item read by the new owner will reflect the latest changes made by the previous owner.

Optionally, to avoid the overhead associated with writing dirty versions of data items to disk, if redo is forced and not rewritten, dirty data items can be read from the original owner's volatile memory before writing them to shared disk. Clean up dirty versions of data items. In particular, when an ownership node makes a change to a data item, a "redo" record is generated that reflects the change. As long as the redo records for the change are forced to disk at or before the ownership of the data item changes, the original owner can clean up dirty versions of data items without first saving the dirty versions to disk. In this case, the new owner can determine what changes must be made to the on-disk version of the data item by (1) reading the redo records and (2) making the indicated changes to the on-disk version of the data item, to reconstruct the latest version of the data item.

Another option is to transfer dirty data items to the new owner automatically (on the initiative of the original owner) or on demand (in response to the new owner's request) when a transaction requests data in the new owner's node cache.

Changes to a data item may be reflected in multiple recovery logs if dirty versions of the data item were not flushed to disk prior to the ownership change. For example, suppose a first node makes a change to a data item, after which ownership of the data item is transferred to a second node. A first node may dump redo logs to disk, but transfer dirty versions of data items directly to a second node without first storing them to disk. The second node may then make a second change to the data item. Assume that the second node dumps the second changed redo record to disk, and then the second node fails before storing the dirty version of the data item to disk. In these cases, changes to the on-disk version that must be applied again to the data item are reflected in both the first node's redo log and the second node's redo log. According to one embodiment, online recovery is performed by merging redo logs to recover data items.

According to one embodiment, ownership of a data item may be transferred without waiting for the transaction that is modifying the data item to commit. Therefore, changes made to data items by a single transaction can be spread across multiple redo logs. In these cases, the database server's transactional rollback mechanism is set to undo changes to multiple logs, where the undo operations are performed on the data blocks in the reverse order in which the changes were made to the data blocks. In addition, a media recovery mechanism is provided that can merge the redo logs of all owners, where the merge process includes redo records of all changes made since the time the data was backed up.

Redistribution without blocking update

According to one aspect of the invention, reallocation of ownership of data items is performed without blocking updates to the data being reallocated. According to one embodiment, by having the database server wait for all transactions that access any data items belonging to the reallocated bucket to commit, and wait for all dirty data items belonging to the bucket to be flushed to disk, the Assignment of ownership is performed in case of update. In these cases, the data belonging to the reallocated bucket can be updated immediately (without waiting for dirty versions to be dumped to disk) if the original owner cannot access the data in exclusive or shared mode. If the original owner did have exclusive access to the data, the original owner may have a dirty version of the data in its cache, so updates are delayed until the original owner removes the dirty page (or redo for related changes) Write to shared disk.

The database server may be configured to wait for an already requested in-progress update by the original owner to complete before allowing a new update to a data item whose ownership has been newly transferred. On the other hand, the database server can be set to abort operations in progress and then reissue the transaction to the owner. According to one embodiment, the decision as to whether to wait for a given in-progress operation to complete is made based on various factors. Such factors may include, for example, how much work has been done for the operation.

In some cases, waiting for the original owner's requested updates to complete may cause spurious deadlocks. For example, assume row A is in bucket 1, and rows B and C are in bucket 2. Suppose transaction T1 updates row A, and another transaction T2 has updated row B. Suppose at this point, the ownership of bucket 2 is remapped to the new node. If at this point T1 wants to update row C, T1 will wait for the remap to complete. Therefore, T1 will wait for T2. If T2 wants to update row A, there is a deadlock between T1 and T2.

According to one embodiment, even when ownership of several buckets is required, the transfer of ownership is only performed one bucket at a time to minimize the amount of time a transaction will have to wait to access data in a reallocated bucket.

Technology Used to Transfer Ownership

The following examples illustrate techniques for transferring ownership of data within a shared-nothing database executing on a shared disk system, according to various embodiments of the invention. In the following example, it is assumed that ownership of data is changed while a transaction that has modified the data is still in progress. That is, the database system does not wait for an in-flight transaction to stall that has accessed data to be reallocated.

One technique for transferring ownership will be described below with reference to an example where ownership of a subset of hypothetical objects ("bucket B") changes from node X to node Y. According to one embodiment, the database system initially marks bucket B as being "in transition" from node X to node Y. Changes to the ownership map are then broadcast to all nodes, or invalidated via a global lock.

According to one embodiment, query execution plans involving data in storage segment B are regenerated in response to a change in the ownership map. Optionally, the cached mapping is invalidated or reloaded in response to a change in ownership mapping.

After reallocation, any new subqueries/data manipulation language (dml) fragments that access data in bucket B will be transmitted to node Y. Optionally, the SQL fragment currently running in X may be rolled back before the bucket is marked as in transition from X to Y. These fragments can then be republished to node Y after reallocation. It should be noted that the transaction to which these fragments belong is not itself rolled back, only the current call is rolled back and resent to the new owner. In particular, changes made by node X to data in bucket B in previous calls are not affected.

According to one embodiment, node X is able to detect that there is no call in progress that is accessing data in bucket B through a simple local locking scheme. The locking scheme may involve, for example, having each program that accesses data in a memory segment acquire a shared lock/latch on that memory segment. When the memory segment is to be reallocated, the program performing the reallocation waits until it can acquire an exclusive lock/latch on the memory segment. By obtaining an exclusive lock/latch on the memory segment, the reallocator ensures that no other program is currently accessing the memory segment.

According to one embodiment, node X does not wait for all calls to complete successfully before marking the bucket as in transition because of a potential deadlock. The following is an example of how such a deadlock can occur. Consider lines 1, 2, and 3 in the memory segment to be remapped.

The following sequence of events can lead to a deadlock:

(a) T1 updates row 2.

(b) T2 performs a multi-row update on row 1 followed by row 2.

T2 now waits for T1.

(c) Deciding that the memory segment is to be remapped.

(d) T1 wants to update row 3. T1 now waits for T2.

According to one embodiment, a forced abort of an ongoing call at node X is avoided by allowing the ongoing call to continue executing normally as long as the data in memory segment B accessed by the ongoing call is in the cache. In other words, X cannot read blocks from disk in segment B without an inter-node lock. If there is a cache miss and the storage segment is in transition, then X must send the message to Y, either retrieve the block from Y, or read the block from disk. A cache coherency protocol is used between X and Y while buckets are in transition.

Process new owner's request

After bucket B has been reassigned to node Y, any new SQL fragments that need to access data in bucket B start executing in node Y. The technique used by node Y when reading a newly transferred data item from disk may vary based on what the previous owner node X did before the storage segment was marked as in transfer. The following cases are examples illustrating different situations that may be handled by the new owner node Y.

Case A: Assume node X aborts all ongoing calls and writes all dirty blocks mapped to the storage segment to the shared disk. For efficiency, each node can link dirty data items to each bucket object sequence. In these cases, node Y is able to read directly from disk. No cache dependencies are required. Node Y is immediately marked as the new owner of the bucket.

Case B: Suppose node X aborts all in-progress calls, but dirty data items are not written out. In these situations, node Y will need to retrieve or verify that node X has no dirty copies before reading the block from disk. If X has a dirty copy, the before image is left on node X for recovery and to ensure that the checkpoint did not advance past changes made in node X that have not yet advanced on node Y The block record (block write) is reflected to the disk. After all dirty data items in X have been written (either by itself or by a pre-image (PI) cleared by records in Y), the bucket state changes from in transition to as owner the Y. Y can now access the disk blocks in that bucket without checking X.

If node X fails while the bucket state is in transition, then if node Y does not have a current copy of the data item, then the recovered node (node Y, if it continues to exist) will need to apply the Generated redo (and possibly generated redo in node Y if node Y also fails).

Case C: Assume that node X aborts the in-progress call and cleans up dirty data items. In these cases, node Y is able to read blocks directly from disk without cache dependencies. However, if there is unapplied redo in X, the block may need to be updated. A bucket will be considered in transition until all redo generated in X has been applied by Y and written to disk. This is needed to prevent X from making its checkpoints past redo segments that have not been reflected on disk.

If node X fails while the bucket state is in transition, then if node Y does not have a current copy of the data item, then the recovered node (node Y, if it continues to exist) will need to apply the Generated redo (and possibly generated redo in node Y if node Y also fails).

Case D: Assume that node X continues to execute the call in progress. Node X and Node Y will need to check each other out before accessing blocks from disk. Y is marked as the new owner when all in-progress calls have completed and all blocks have been written out or transferred to Y. From this moment on, there is no need to cache dependencies.

If node X or node Y fails while the bucket state is in transition, recovery must be performed. Restoration techniques used in this environment may be similar to those described in US Patent No. 6,353,836, filed March 4, 2002, and US Patent Application No. 10/092,047, each of which is incorporated herein in its entirety. If both nodes fail, the redo logs from X and Y need to be merged.

As described in case D, allowing node X to continue executing calls in progress may have various benefits. In particular, node X is allowed to continue executing calls in progress such that ownership will be reassigned with minimal impact on ongoing transactions. However, it requires a cache coherency and recovery scheme to be performed between node X and node Y for bucket B.

One method for providing cache coherency in these situations involves having node X acquire locks for all data items it currently caches for bucket B. A home/directory node for all data items in B can be assigned as node Y. A notification is then sent to all nodes that bucket B will be moved from X to Y. This notification invalidates/updates the ownership map so that further accesses to B will be sent to Y.

If an invalidation occurs before then, the ownership reallocation operation is aborted. Otherwise the map is updated to indicate that Y is the new owner and the cache dependency is valid.

X and Y then execute a cache coherency protocol for all data items in bucket B, such as the protocols described in US Patent No. 6,353,836 and US Patent Application No. 10/092,047. When there are no more dirty data items in B, the cache coherency protocol can be stopped. Y can release any locks it may have acquired for data items in B.

According to one embodiment, the cache coherence protocol is always in effect such that the node owning the bucket is also the master of these locks. In most cases, each node will only allocate local locks (since it is the dominator) and cache coherency messages will only be needed when ownership changes. When ownership changes, locks opened in that storage segment can be dynamically reassigned to the new owner.

Modifications prior to ownership change

According to an embodiment, sub-transactions that modified data in node X prior to the ownership change will still be considered active, since transaction rollback will require these changes. If the transaction backs out and it has made changes to that bucket in node X, then node Y will need to apply the undo log by reading it from X's portion of the shared disk log.

A subtransaction that modifies data in node X before the ownership change may update the same data in node Y. This requires that requests for transactional locks such as row locks or page locks in node Y need to be coordinated with node X. If a child transaction requests a lock and the lock is already held by its sibling transaction in node X, the lock request is granted. However, if the lock is held by an unrelated transaction, the lock request is blocked. Wait - The graph reflects this wait as a wait on the parent transaction, allowing detection of global deadlocks. Once the ownership change is complete, and all transactions that have acquired locks to access data in bucket B in node X have ended (committed or aborted), only local locks are required for transactional locks.

Requests for transactional locks can always be coordinated locally by having the database server wait for the transaction to complete before initiating an ownership change.

hardware overview

FIG. 2 is a block diagram illustrating a computer system 200 on which an embodiment of the present invention may be implemented. Computer system 200 includes a bus 202 or other communication means for communicating information and a processor 204 coupled with bus 202 for processing information. Computer system 200 also includes main memory 206 , such as a random access memory (RAM) or other dynamic storage device, connected to bus 202 for storing information and instructions to be executed by processor 204 . Main memory 206 may also be used for storing temporary variables or other intermediate information during execution of instructions to be executed by processor 204 . Computer system 200 further includes a read only memory (ROM) 208 or other static storage device coupled to bus 202 for storing static information and instructions for processor 204 . A storage device 210, such as a magnetic or optical disk, is provided and connected to bus 202 for storing information and instructions.

Computer system 200 may be connected via bus 202 to a display 212, such as a cathode ray tube (CRT), for displaying information to a computer user. An input device 214 including alphanumeric and other keys is connected to bus 202 for communicating information and command selections to processor 204 . Another type of user input device is cursor control 216 , such as a mouse, trackball, or cursor direction keys, for communicating direction information and command selections to processor 204 and for controlling cursor movement on display 212 . The input device typically has two degrees of freedom in two axes, a first axis (eg, X-axis) and a second axis (eg, Y-axis), enabling the device to specify a position on a plane.

The present invention is directed to the use of computer system 200 for performing the techniques described herein. These techniques are implemented by computer system 200 in response to processor 204 executing one or more sequences of one or more instructions contained in main memory 206 , according to one embodiment of the invention. Such instructions may be read into main memory 206 from other computer-readable media, such as storage device 210 . Execution of the sequences of instructions contained in main memory 206 causes processor 204 to perform the process steps described herein. In alternative embodiments, hard-wired circuitry may be used in place of or in combination with software instructions to implement the invention. Thus, embodiments of the invention should not be limited to any specific combination of hardware circuitry and software.

The term "computer-readable medium" is used herein to refer to any medium that participates in providing instructions to processor 204 for execution. Such a medium may take many forms, including but not limited to, non-volatile media, volatile media, and transmission media. Non-volatile media include, for example, optical or magnetic disks, such as storage device 210 . Volatile media includes dynamic memory, such as main memory 206 . Transmission media include coaxial cables, copper wire, and fiber optics, including the wires that make up bus 202 . Transmission media can also take the form of acoustic or light waves, such as those generated during radio-wave and infrared data communications.

Common forms of computer readable media include, for example, floppy disks, floppy disks, hard disks, magnetic tape, or any other magnetic media, CD-ROMs, any other optical media, punched paper, paper tape, or any physical media with a pattern of holes, RAM, PROM, EPROM, FLASH-EPROM, or any other memory chips or cartridges, or carrier waves mentioned below, or any other medium readable by a computer.

Various forms of computer readable media may be involved in carrying one or more sequences of one or more instructions to processor 204 for execution. For example, the instructions may initially be carried on a disk in a remote computer. The remote computer can load the instructions into its dynamic memory and then send the instructions over a telephone line using a modem. A modem local to computer system 200 can receive the data on the telephone line and use an infrared transmitter to convert the data to an infrared signal. An infrared detector can receive the data carried in the infrared signal and appropriate circuitry can place the data on bus 202 . Bus 202 carries the data to main memory 206 , from which processor 204 retrieves and executes the instructions. The instructions received by main memory 206 can optionally be stored on storage device 210 either before or after execution of the instructions by processor 204 .

Computer system 200 also includes a communication interface 218 connected to bus 202 . A communication interface 218 , which provides bi-directional data communication, is connected to a network link 220 connected to a local area network 222 . For example, communication interface 218 may be an Integrated Services Digital Network (ISDN) card or a modem for providing a data communication connection to a corresponding type of telephone line. As another example, communication interface 218 may be a local area network (LAN) card for providing a data communication connection to a compatible local area network (LAN). Wireless links may also be used. In any such implementation, communication interface 218 sends and receives electrical, electromagnetic, and optical signals that carry digital data streams representing various types of information.

Network link 220 may generally provide data communication to other data devices through one or more networks. For example, network link 220 may connect to host computer 224 through local area network 222 or to data equipment operated by Internet Service Provider (ISP) 226 . ISP 226 in turn provides data communication services through a global packet data communication network currently known as the "Internet" 228 . Local area network 222 and Internet 228 both use electrical, electromagnetic or optical signals that carry digital data streams. The signals through the various networks and the signals on network link 220 and the signals through communication interface 218, both carrying digital data to and from computer system 200, are exemplary forms of carrier waves carrying the information.

Computer system 200 is capable of sending messages and receiving data (including program code) over a network, network link 220 , and communication interface 218 . In the example of the Internet, the server 230 may transmit the requested program code for the application through the Internet 228, the ISP 226, the local area network 222, and the communication interface 218.

The received code may be executed by processor 204 as it is received, and/or stored in storage device 210 or other non-volatile medium for subsequent execution. In this manner, computer system 200 can obtain the application code in the form of a carrier wave.

The above descriptions are only preferred embodiments of the present invention, and are not intended to limit the present invention. For those skilled in the art, the present invention may have various modifications and changes. Any modifications, equivalent replacements, improvements, etc. made within the spirit and principles of the present invention shall be included within the protection scope of the present invention.

Claims (24)

1. A method for managing data, said method comprising the steps of: maintaining a plurality of persistent data items on persistent storage accessible by multiple nodes in a multi-node system, the persistent data items including specific data items stored at specific locations on the persistent storage; assigning exclusive ownership of each of the persistent data items to one of the nodes, wherein a particular node of the plurality of nodes is assigned exclusive ownership of the particular data item; When any node wants to perform an operation involving said specific data item, said node expecting said operation to be performed, since said specific data item exists in said specific location, transmits said operation to said specific node , for the specific node to perform the operation on the specific data item; reassigning ownership of the particular data item from the particular node to another without moving the particular data item from the particular location on the persistent storage as the particular node continues to operate a node; After the reallocation, when any node wants to perform an operation involving the specific data item, since the specific data item exists in the specific location, the node expecting the operation to be performed will send the an operation is communicated to the other node for the other node to perform the operation on the particular data item; continuing to have a set of data items owned by the first node after the first node has been removed from the multi-node system; and Responsive to detecting a request for an operation involving the data item, reassigning ownership of the data item from the first node to one or more other nodes. 2. A method for managing data, said method comprising the steps of: maintaining a plurality of persistent data items on a persistent storage accessible by a plurality of nodes, the persistent data items including a specific data item stored at a specific location on the persistent storage; assigning exclusive ownership of each of the persistent data items to one of the nodes, wherein a particular node of the plurality of nodes is assigned exclusive ownership of the particular data item; When any node wants to perform an operation involving said specific data item, said node expecting said operation to be performed, since said specific data item exists in said specific location, transmits said operation to said specific node , for the specific node to perform the operation on the specific data item; reassigning ownership of the particular data item from the particular node to another without moving the particular data item from the particular location on the persistent storage as the particular node continues to operate node, wherein said particular node stores a dirty version of said particular data item in volatile memory when said particular data item is to be redistributed to said other nodes, and wherein said step of reallocating ownership include: forcing one or more redo records associated with the dirty version to persistent storage; and After the reallocation, when any node wants to perform an operation involving the specific data item, since the specific data item exists in the specific location, the node expecting the operation to be performed will send the The operation is communicated to the other nodes for the other nodes to perform the operation on the particular data item. 3. A method for managing data, said method comprising the steps of: maintaining multiple persistent data items on persistent storage that can be accessed by multiple nodes; Assigning ownership of each of said persistent data items to one of said nodes by: assign each data item to one of the plurality of buckets; and assigning each storage segment to a node of the plurality of nodes; wherein said node to which the storage segment is assigned is established as the owner of all data items assigned to said storage segment; wherein the step of assigning each data item to one of the plurality of storage segments is performed by (a) applying a hash function to a name associated with each data item or (b) using range-based partitioning; When a first node wants to perform an operation involving a data item owned by a second node, the first node communicates the operation to the second node for the second node to perform the operation. 4. The method of claim 3, wherein the step of assigning each storage segment to one of the plurality of nodes is performed by applying a hash function to an identifier associated with each storage segment. 5. The method of claim 3, wherein the step of allocating each storage segment to one of the plurality of nodes is performed using range-based partitioning. 6. A method of managing data, said method comprising the steps of: A plurality of persistent data items are maintained on persistent storage accessible by multiple nodes, wherein the persistent data items are rows assigned to one or more storage segments, and wherein the persistent data items include a first persistent a data item and a second persistent data item; Assigning ownership of each of said persistent data items to one of said nodes by: assigning each data item to one of a plurality of storage segments: wherein said first persistent data item is assigned to a first storage segment and said second persistent data item is assigned to a second storage segment, wherein said said first memory segment is different from said second memory segment; and assigning each storage segment to a node of the plurality of nodes; wherein said node to which the storage segment is assigned is established as the owner of all data items assigned to said storage segment; Wherein, the step of assigning each data item to one of the plurality of storage segments is performed by enumerating the relationship of a single data item to the storage segment; When a first node wants to perform an operation involving a data item owned by a second node, the first node communicates the operation to the second node for the second node to perform the operation. 7. The method of claim 3, wherein the step of assigning each bucket to one of the plurality of nodes is performed by enumerating a single bucket-to-node relationship. 8. The method of claim 3, wherein the number of buckets is greater than the number of nodes, and the bucket-to-node relationship is a many-to-one relationship. 9. The method of claim 3 , further comprising the step of converting all data items mapped to buckets by modifying bucket-to-node mappings without modifying the mapping of data items to buckets. Ownership is redistributed from the first node to the second node. 10. The method of claim 3, wherein the data item is allocated to a first storage segment, the first storage segment is allocated to the first node, further comprising allocating the first storage segment Steps for different nodes. 11. The method of claim 2, further comprising flushing the dirty version from the volatile memory without writing the dirty version of the particular data item to the persistent memory. 12. The method of claim 11 , wherein the other nodes rebuild the dirty data by applying the one or more redo records to the version of the particular data item present on the persistent store. Version. 13. The method of claim 2, wherein the other nodes rebuild the dirty data by applying the one or more redo records to the version of the particular data item present on the persistent store. Version. 14. The method of claim 2, further comprising transferring the dirty version of the particular data item from volatile memory associated with the particular node to volatile memory associated with the other node. 15. The method according to claim 14, wherein the step of transferring the dirty version is actively performed by the specific node without requesting the dirty version through the other nodes. 16. The method of claim 14, wherein the step of transferring the dirty version is performed by the specific node in response to a request for the dirty version from the other node. 17. The method of claim 2, wherein said step of reassigning ownership of said particular data item from said particular node to said other nodes is performed as ownership of a data item transfers from said particular node to said other node. Part of a gradual transfer of one or more nodes in a node group, wherein the node group does not include the particular node. 18. The method of claim 17, wherein the gradual transfer is initiated in response to detecting that the other node is processing fewer work requests relative to one or more nodes in the multi-node database system. 19. The method of claim 18 , wherein terminating said gradual transfer. 20. The method of claim 2, wherein the step of reallocating ownership of the particular data item is performed without waiting for a transaction that is modifying the particular data item to commit. 21. The method of claim 20, wherein the transaction modifies the specific data at a first time when the specific data item is owned by the specific node, and when the specific data item is owned by the Said transaction modifies said particular data at a second time while owned by said other node. 22. The method of claim 21 , further comprising: undoing changes made when the transaction changed the particular data at the second time by based on undo records in undo logs associated with the other nodes and undoing the changes made by the transaction at the first time based on an undo record in an undo log associated with the particular node, thereby rolling back the changes made by the transaction. 23. The method of claim 2, further comprising: said other node receives a request to update said particular data item; in response to receiving the request, determining whether the particular node accesses the particular data item in exclusive mode or shared mode; If the specific node does not access the specific data item in exclusive mode or shared mode, the other nodes do not wait for the specific node to dump any dirty version or redo of the dirty version of the specific data item In the case of persistent storage, update the particular data item. 24. The method of claim 23, further comprising the step of: in response to transferring ownership of the particular data item to the other node, suspending ongoing operations involving the particular data item; The ongoing operation is automatically re-executed after the ownership of the particular data item has been transferred to the other node.

Images