Distributed database management system with dynamically split B-tree indexes

A distributed data processing system for a database composed of data records organized into tables for responding to a request to split an index in a consistent and concurrent fashion. A request to split an index atom at any given node is processed by a designated “chairman” for that index atom. The chairman splits its index by maintaining a lower portion thereof and by moving the contents of an upper portion to a newly created index atom as the right sibling. Each other node takes appropriate steps to assure orderly transfer of this information from the chairman consistently and concurrently across the distributed database processing system.

BACKGROUND OF THE INVENTION

Field of the Invention

This invention generally relates to database management systems and more specifically to a methodology for splitting indexes in a distributed database management system.

Description of Related Art

Databases typically incorporate indexes for enabling the efficient retrieval of certain information. A B-tree data structure is a popular indexing structure that is optimized for use in databases that read and write large blocks of data and that enables efficient database searching. A B-Tree data structure includes a root and a plurality of leaves. The root uses a key value to identify a corresponding leaf. Each leaf points to any records that contain the key value. The key values are sorted in order, typically in a plurality of leaves thereby to form a sorted list. Specifically, a given leaf includes a “left sibling” and a “right sibling” that identify a leaf to the left of and a leaf to right of the given leaf thereby to maintain a list in sorted order. The first or left-most leaf and last or right-most leaf include entries denoting the ends of the list of leafs for that root.

Typically each leaf has a fixed memory size. When a size threshold is reached, it becomes necessary to define a key value and to “split” that leaf into “left” and “right” leafs. The “left” leaf receives values that are less than the defined key value and the “right” leaf receives the remaining values with appropriate modifications to the root. In centrally based and non-shared databases, the splitting process is efficient because generally there is only one copy of the index in the database system. The split is easy to effect by quiescing the data processing system during the actual splitting operation.

Recently there has been interest in the development of distributed databases. The above-identified U.S. Pat. No. 8,224,860 discloses an implementation of a distributed database wherein user access is provided through a network of transactional nodes and complete copies of the information in the database are only stored on archival nodes that act as storage managers and are not directly accessed by users. More specifically, a user connects to a transactional node to perform operations on the database by generating high-level queries that the transactional node processes. In this implementation a given transactional node need only contain that data and metadata as required to process queries from users connected to that node. The data and metadata are defined by an array of atom classes, such as an index class, and atoms where each atom corresponds to a different instance of the class, such as index atom for a specific index. Replications or copies of an atom may reside in multiple nodes wherein the atom copy and a given node are processed in that node at least in part independently of each other. When an atom is created at a node, it is designated as a “chairman.” The designation of a node as a chairman can be transferred to a replication of that atom in another node.

In the implementation of U.S. Pat. No. 8,224,860 asynchronous messages transfer atoms and information about atoms among the different nodes to maintain the database in a consistent and a concurrent state. Specifically each node in the database network has a unique communications path to every other node. When one node generates a message involving a specific atom, it can communicate as necessary with those other nodes that contain replications of that specific atom. Each node generates these messages independently of other nodes. So it is possible that, at any given instant, multiple nodes will contain copies of a given atom and different nodes may be at various stages of processing them. As these operations in different nodes normally are not synchronized, it is necessary to operate the database so each user is assured that interaction is with a concurrent and consistent database.

Splitting an index in a distributed database such as disclosed in the above-identified U.S. Pat. No. 8,224,860 involves splitting replications of an index atom that performs as a leaf at the transaction node requesting the split and each archival node plus any other transactional node that has a copy of that index atom. It is possible for multiple nodes to request a split of a given index whereupon a race problem can exist with an erroneous outcome. Prior methods, such as those involving quiescence, are not readily applicable to implementations of a distributed database of the type discussed above without introducing unacceptable system performance degradation. What is needed is a method for handling requests for splitting an index in a distributed database wherein copies of the index are located in multiple locations.

SUMMARY

Therefore it is an object of this invention to provide an implementation of a distributed database that processes requests to split an index in a consistent and concurrent fashion.

Another object of this invention is to provide an implementation of a distributed database that processes requests to split an index in consistent and concurrent fashion without any significant performance degradation.

Yet another object of this invention to provide an implementation of a distributed database that processes a requested split of an index and eliminates the involvement of nodes that do not include that specific index.

In accordance with one aspect of this invention a distributed database processing system includes a plurality of nodes, each of which includes means for establishing communications with every other node wherein the database has an atom class for each category of metadata and data including an index atom class that provides an index atom for each index in the database and each index atom includes a range of key values. An index atom can be replicated to other nodes. An index atom is split when a node detects a need to split the index atom based upon a split key value that defines lower and upper portions of the index. Another node identifies a location in its index atom based upon the key value for defining the approximate lower and upper portions of the keys for the index atom, creating a second index atom as a right sibling to the first index atom, transferring the key values in the upper portion of the first index atom to the to the lower portion of the second index atom and transmitting to all other nodes with the identified index atom an index split message including the split key value. Each other node responds to the receipt of the index split message by deleting corresponding to the key values in the upper portion of the first index atom being split, retrieving the populated second index atom copy from the one node with the right sibling and sending a split done message to the one node chairman whereupon the one node broadcasts an index split done message when all other nodes have generated the split done message.

In accordance with another aspect of this invention, a distributed database management system includes a plurality of transactional and archival nodes wherein each transactional node responds to queries by interacting with a portion of the database thereat and wherein an archival node stores a version of the entire database. Communications are established at each node and every other node. The system has an atom class for each category of metadata and data including an index atom class that provides a B-Tree index atom for each index in the database. Each index atom can be replicated on demand to the archival node and at least one transactional node. One index atom in one node is designated as a chairman for that index atom. The process of splitting an index atom includes detecting at one of the transactional nodes a need to split a B-Tree index at that node and, if the node does not contain the chairman, transmitting a split index request message to the chairman node including the identification of the requesting node and a split key value. The chairman responds to its internal request for splitting the index or the receipt of the split index request message from another node by defining the contents of lower and upper portions for the index in response to the split key value, creating a second index atom as a right sibling of the index atom being split, moving the upper portion of the index for that index atom to the second index atom, and transmitting to all nodes that contain that index atom, an index split message including the split key value that defines the split for the lower and upper portions. Each non-chairman node responds by deleting from its existing index atom in response to the split key value the contents in the upper portion of the index atom being split and retrieving from the chairman the populated second index atom copy as the right sibling for the index atom being split. Each node transmits a split done message to the chairman. The chairman sends an index split done message when all nodes involved in the index split operation have reported completion to the chairman.

DESCRIPTION OF ILLUSTRATIVE EMBODIMENTS

FIG. 1depicts an implementation of an elastic, scalable, on-demand, distributed database system30that operates over a plurality of nodes. Nodes N1through N6are “transactional nodes” that provide user access to the database; nodes A1and A2are “archival nodes” that act as storage managers and function to maintain a disk archive of the entire database at each archival node. While an archival node normally stores the entire database, a single transactional node contains only that portion of the database it determines to be necessary to support transactions being performed at that node at that time.

Each node inFIG. 1can communicate directly with each other node in the system30through a database system network31. For example, node N1can establish a communications path with each of nodes N2through N6, A1and A2. Communications between any two nodes is by way of serialized messages. In one embodiment, the messaging is performed in an asynchronous manner to maximize the bandwidth used by the system thereby to perform various operations in a timely and prompt manner. Typically the database system network31will operate with a combination of high-bandwidth, low-latency paths (e.g., an Ethernet network) and high-bandwidth, high-latency paths (e.g., a WAN network). Each node has the capability to restrict use of a low-latency path to time-critical communications (e.g., fetching an atom). The high-latency path can be used for non-critical communications (e.g. a request to update information for a table). Also and preferably, the data processing network of this invention incorporates a messaging protocol, such as the Transmission Control Protocol (TCP) and assures that each node processes messages in the same sequence in which they were sent to it by other nodes.

FIG. 2depicts a representative transactional node32that links to the database system network31and various end users33. The transactional node32includes a central processing system (CP)34that communicates with the database system network31through a network interface35and with the various users through a user network interface37. The central processing system34also interacts with RAM memory38that contains a copy of the database management program that implements this invention. This program functions to provide a remote interface40, a database request engine41and a set42of classes or objects. The database request engine41only exists on transactional nodes and is the interface between the high-level input and output commands at the user level and system level input and output commands at the system level. In general terms, its database request engine parses, compiles and optimizes user queries such as SQL queries into commands that are interpreted by the various classes or objects in the set42.

In this system, the classes/objects set42is divided into a subset43of “atom classes,” a subset44of “message classes” and a subset45of “helper classes.” At any given time a transactional node only contains those portions of the database that are then relevant to active user applications. Moreover, all portions of database in use at a given time at any transactional node are resident in random access memory38. There is no need for providing supplementary storage, such as disk storage, at a transactional node during the operation of this system.

Referring toFIG. 3A, a Master Catalog atom70tracks the status of transactional and archival nodes in database system30ofFIG. 1. It also can be considered as an active index that creates and monitors the Transaction Manager atom71, the Database atom72, each Schema atom73, each corresponding set of Table atoms74and Table Catalog atoms75, and Sequence ID Managers82. The Table Catalog atom75acts as an active index that creates and monitors Index atoms76. Record States atoms77. Data atoms78, Blob States atoms80and Blob atoms81associated with a single table. There is one Table Catalog atom75for each table.

FIG. 3Bis useful in understanding the interaction and management of different atom types. In this context, neither the Master Catalog atom70nor the Table Catalog atom75performs any management functions. With respect to the remaining atoms, the Database atom72manages each Schema atom73. Each Schema atom73manages each related Table atom74and Sequence ID Manager atom82. Each Table atom74manages its corresponding Table Catalog atom75, Index atoms76, Record States atoms77. Data atoms78, Blob States atom80and Blob atoms81. Still referring toFIG. 3B, the database request engine41communicates with the Master Catalog atom70, Transaction Manager atom71, the Database atom72, each Schema atom73, each Table atom74and the Sequence ID Managers82. The database request engine41acts as compiler for a high-level language such as SQL. As a compiler, it parses, compiles and optimizes queries and obtains metadata and data from atoms for the formation of the various fragments of data base information.

Each atom has certain common elements and other elements that are specific to its type. For purposes of describing this invention,FIG. 4depicts an index atom76that is implemented as a B-Tree index and is split according to this invention. Element76A is a unique identification for the index atom76. Pointers76B and76C identify a master catalog atom and the creating catalog atom, respectively. Each atom must have a chairman that performs functions as described later. Element76D points to the node where the chairman for that atom resides.

Each time a copy of an atom is changed in any transactional node, it receives a new change number. Element76E records that change number. Whenever a node requests an atom from another node, there is an interval during which time the requesting node will not be known to other transactional nodes. Element76F is a list of all the nodes to which the supplying node must relay messages that contain the atom until the request is completed.

Operations of the database system are also divided into cycles. A cycle reference element76G provides the cycle number of the last access to the atom. Element76H is a list of the all active nodes that contain the atom. Element76I includes several status indicators. Elements76contains a binary tree of index nodes to provide a conventional indexing function. Element76K contains an index level. Such index structures and operations are known to those in skilled in the art.

As previously indicated, communications between any two nodes is by way of serialized messages which are transmitted asynchronously using the TCP or another protocol with controls to maintain messaging sequences.FIG. 5depicts the basic syntax of a typical message90that includes a variable-length header91and a variable-length body92. The header91includes a message identifier code93that specifies the message and its function. As this invention envisions a scenario under which different nodes may operate with different software versions, the header91also includes identification94of the software version that created the message. The remaining elements in the header include a local identification95of the sender and information96for the destination of the message and atom identification97. From this information, a recipient node can de-serialize, decode and process the message.

FIG. 6depicts a set of messages that are helpful in implementing this invention. An Index Split Request message147is sent to the chairman by another (non-chairman) atom to institute an index split operation. An Index Split message148contains an index atom identifier and the key on which to split the index atom (i.e., a “split key”). An Index Split Start message160indicates that the chairman has begun processing the split and is broadcast to all nodes with a copy of the index being split. An Index Split Done message161indicates that a node has completed the split operation.

FIG. 7broadly discloses a process200for implementing an index splitting function in accordance with this invention. This disclosure can be better understood by referring toFIG. 8in whichFIGS. 8A through 8Ddepict the state of the index at various stages of the process as initiated by a non-chairman node. Specifically.FIG. 8Adepicts an un-split index201C where “C” designates an index atom located at the chairman's node. All other copies of the index atom are designated as201N to indicate that they are located in a node that does not contain the chairman.FIG. 8Aalso discloses the index201C at the chairman node as comprising a lower portion202C and an upper portion203C. Similarly, the index201N in a non-chairman node comprises a lower portion202N and an upper portion203N.

If either the chairman or a non-chairman index node needs to be split, only the chairman controls the splitting operation. If the chairman determines that the atom201C requires splitting, step204transfers control to step205whereupon the chairman selects a key value upon which to base the split. InFIG. 8this value is represented by a vertical dashed line206C and typically would be chosen so that a lower portion202C and an upper portion203C have approximately the same size. If the non-chairman determines that its index atom needs to be split, the non-chairman generates an Index Split Request message at step208with a split key value corresponding to a split position206N. As will be apparent, either request is directed to step207whereupon the chairman takes control of the operation.

At step207, the chairman broadcasts an Index Split Started message to all nodes having a copy of the index atom to be split. Each receiving node responds to the Index Split Started message by buffering and processing subsequent incoming messages in the nodes that include the index atom to be split. As the chairman now controls split, further operations will involve the index atom201C.

Next, the chairman creates a new index atom at step210shown as an empty atom211C inFIG. 8B. This new index atom then becomes the right sibling of the index atom being split by the chairman and the left sibling of the index atom that had been the right sibling prior to the split. Then the chairman transfers control to an index split process212to split the index as shown inFIG. 9wherein the chairman uses step213to determine a key value of the index upon which to base the splitting operation; i.e., a split key value. As previously indicated, that value is established by the chairman or non-chairman that originates the request and defines a boundary in the index atom between the lower and upper key value ranges. The chairman uses this information to place all the keys in the upper key value portion into the new index atom and to truncate the original index by deleting the keys with values in the lower portion as shown inFIG. 8C.

In step214ofFIG. 9the chairman broadcasts an Index Split message to all other nodes that contain an existing index atom including the chairman and non-chairman nodes inFIG. 7and all other non-chairman nodes. At the chairman atom, this message initiates a process for assuring that messages received for and generated by the chairman atom have correct addresses for the left and right siblings for certain messages. An Index Node Added message150inFIG. 6indicates that a new index node is added and contains an index key, record identification and other information and an Index Node Removed message contains corresponding information concerning the removal of an index node. From the occurrence of the split in the chairman node in step214until the end of the splitting operation there exists the possibility that non-chairman nodes may not have processed the split. Each time one of these messages is received during this interval, messages received at the left sibling that should be processed on the right sibling are rebroadcast from the right sibling. Messages received at the right sibling are rebroadcast from the left sibling.

This process continues until the chairman receives Index Message Done messages from all the other nodes/atoms involved in the split operation. When this occurs, step217terminates the rebroadcasts of step215and broadcasts an Index Split Done message161to all the nodes with a copy of the index atom.

When a non-chairman node receives an Index Split message from the chairman, an Index Split process220inFIG. 7and shown in more detail inFIG. 10begins. Each non-chairman node responds to that message by truncating its existing index atom201N at step222to obtain a copy of the chairman's index atom as shown inFIG. 8Cthat becomes the left sibling of the new index atom. Step223obtains a copy of the populated new index211C as shown inFIG. 8Dwhich becomes the lower portion of the new index atom and prevents any processing on the keys from the right sibling.

Next step224“prunes” the right sibling by adding all its local keys from the upper portion copy213N as also show inFIG. 8D. This occurs before allowing any message processing on the right sibling213N and assures that all messages are processed against the same indexes as if no split had occurred. Next step225allows the non-chairman to begin to process any messages directed to it. Step227broadcasts an Index Split Done message to all the nodes with an existing index atom copy. This signifies that foregoing processing has been completed at the respective node. Control then transfers from process220back to the chairman Index Split Process214shown inFIG. 9. Specifically as shown inFIG. 9, at this point the chairman broadcasts an Index Split Message to all of the nodes with an existing atom copy whereupon control transfers to a non-chairman tracked message process230as shown inFIG. 11.

Referring now toFIGS. 7 and 11, a stored message process230uses step231to divert control to step232to process each message received at that node for the index atom. When an Index Split Start message is received, step231diverts control to step233that responds to the receipt of the Index Split Done message in step233. During the interval between these two messages, steps231and233divert control to step234to buffer each incoming message at step234. Each message includes its atom ID, sender ID represented by a node number and a change number. If step234receives a first message or new message not previously received, step235diverts control to record the message in some manner, as by entry in a bit map at step236. Then the message is processed. If step235determines that the bit map has recorded the receipt of that message from another node, control transfers to step237to drop or ignore the message because it is a duplicate. Consequently redundant processing does not occur.

As will now be apparent, this invention insures that an index in a distributed database can be split in a concurrent and consistent fashion even though copies of that index may exists at multiple nodes at which data processing occurs. This occurs without any need to quiesce the system so there is no significant performance degradation. Also this invention eliminates the involvement of nodes that do not include that specific index.

This invention has been disclosed in terms of certain implementations that are directed to a specific implementation of a distributed database. Although the invention is disclosed for in one specific implementation, the application to other implementations will be apparent to others without departing from the invention. Therefore, it is the intent of the appended claims to cover all such variations and modifications as come within the true spirit and scope of this invention.