HK1228057B - Database management system - Google Patents

Description

Database Management System

Description of the case

This application is a divisional application of Chinese invention patent application No. 201180014604.6, filed on March 18, 2011.

Technical Field

The present invention relates generally to database management systems and, more particularly, to a method and apparatus for implementing a multi-user, elastic, on-demand distributed relational database management system characterized by atomicity, performance, and scalability.

Background Art

Over the past few years, the use of databases for storing and retrieving information has become an important tool in a wide range of business applications. Initially, many database systems operated on a single server device with multiple users. However, over the past few years, various factors have arisen that have required fundamental changes in database architecture. First, database storage requirements have become extremely large. Second, the number of users attempting to access such databases has also increased. Third, the use of databases for retrieving relatively stable data with minimal updates has been replaced by transaction processing.

A transaction is a unit of work that must be completed in its entirety. A single transaction can include multiple data manipulations. For example, a single transaction might include a read operation followed by a write operation. In recent years, significant effort has been directed toward enabling relational databases to support increasing transaction processing rates.

Databases are now judged by the standards defining the ACID properties: atomicity, consistency, isolation, and durability. Atomicity ensures that all transaction tasks are completed in their entirety. Consistency ensures that only valid data is written to the database. Isolation ensures that other operations cannot access or "see" data in an intermediate state during a transaction. Durability ensures that once a transaction has been successfully processed, it cannot be reversed.

Consistency is particularly important in multi-user systems, where two or more users may seek simultaneous access to shared, volatile data. Early multi-user systems used locking operations to ensure consistency. Locks can be exclusive or write locks or non-exclusive or read locks, and can be applied to individual records or to pages. However, as databases have grown in size and transaction rates have increased, the overhead of managing locks has become significant and, in some cases, prohibitive.

Multi-version concurrency control (MVCC) is an alternative process for ensuring concurrency. For complex databases, MVCC can be more efficient than locking. MVCC uses timestamps or incremental transaction identifiers (IDs) to serialize different versions of a record. Each version allows a transaction to read the most recent version of the object before the timestamp or ID. With this control method, other users will not see any changes to a record, for example, until the change is committed. MVCC also eliminates locking, which has other additional overhead, and creates a system where read operations cannot block write operations.

In addition to meeting the ACID test, there is now a demand for continuous availability to users. Some database systems dedicate one computer system to transaction processing and another to decision support and other reporting processing. They are interconnected so that other functions can be supported simultaneously. As databases grow in size and complexity, existing data processing systems are replaced with more powerful data processing systems. Another means of adapting to growth involves a replication system, in which one machine is designated as the "head" machine, which keeps all the replica machines synchronized. If the head machine fails, the process assigns that function to another replica machine. Different replica machines can be used for specific users. This approach is not scalable because all machines must have the same capabilities.

Alternatively, multiple autonomous database systems can be consolidated into a single "federated" database, using a computer network that interconnects the individual databases. A federated database requires "middleware" to synchronize the constituent databases. This "middleware" can become quite complex. As the database size increases, the resources required to operate the middleware can impose significant overhead, potentially degrading overall system performance.

Partitioning is another method for implementing a database, in which the logical database or its constituent elements are divided into distinct, independent parts. In a distributed database management system, each partition can be spread across multiple nodes. Users at a given node can execute local transactions for that partition. Partitioning can also be achieved by forming smaller databases or by splitting only selected elements of a table.

There are two general approaches to partitioning. In horizontal partitioning, also known as "sharding," different rows are placed in different tables and on different servers. Typically, they have certain characteristics in common, such as ranges of postal codes or surnames divided into different tables by range. For example, a first database might include all records for surnames in the range A to M; a second database for those in the range N to Z. Sharding, as a form of horizontal partitioning, involves locating the rows of the database on separate servers. Sharding does not count the number of rows in each table and improves search performance. However, sharding uses hash codes at the application level, which makes it difficult to implement. It also involves a two-phase commit. The complexity of sharding makes it suitable for specific applications because the basis for defining the shards is well defined.

Vertical partitioning involves creating a table with fewer columns and splitting the columns across the table. Like federated databases, vertical partitioning requires middleware to determine how to route any request for a specific field to the appropriate partition. Additionally, these systems use a two-phase commit sequence that is complex to implement.

In another approach, known as a "shared-nothing" architecture, each node is independent and self-sufficient. Shared-nothing architecture is popular for web development because it can be scaled up simply by adding nodes in the form of inexpensive computers. This approach is also popular in data warehouse applications, where updates tend to occur less frequently than transactions. However, processing joins is complex when processing large data sets from different partitions or machines.

Some database systems are referred to as "distributed." One implementation of a distributed system consists of a "cluster" and two communication paths. A high-speed Internet path carries data between the clusters. Various control functions, such as lock management, require a high-speed, dedicated communication path. While this approach addresses database redundancy and availability, lock management, as previously mentioned, can limit system performance.

In a "shared-all" system, ultra-high-speed communication keeps the system synchronized. However, lock management can require a large amount of bandwidth resources. To avoid this, such systems include point-to-point communication channels and very complex disk controllers.

Collectively, those prior art systems meet some, but not all, of the known requirements for database systems. What is needed is a database architecture that is scalable, meets the ACID properties of atomicity, consistency, isolation, and durability. What is also needed is a database system that operates over the Internet without requiring dedicated high-speed communication paths, provides transaction processing, and is operable over a large geographic area.

Summary of the Invention

Therefore, an object of the present invention is to provide a flexible, scalable, on-demand, distributed data processing system.

Another object of the present invention is to provide a flexible, scalable, on-demand, distributed data processing system that is fault-tolerant.

Yet another object of the present invention is to provide a flexible, scalable, on-demand, distributed data processing system with high availability.

Yet another object of the present invention is to provide a flexible, scalable, on-demand, distributed data processing system that is platform independent.

Yet another object of the present invention is to provide a resilient, scalable, on-demand, distributed data processing system that is atomic, consistent, isolated, and durable.

It is a further object of the present invention to provide a flexible, scalable, on-demand, distributed data processing system that operates over the Internet without requiring dedicated high-speed communication paths.

It is a further object of the present invention to provide a resilient, scalable, on-demand, distributed data processing system that provides transaction processing and is suitable for implementation over a large geographic area.

According to one aspect of the present invention, a database management system that enables users to interact with a database consisting of data and metadata includes a plurality of nodes and persistent storage with communication paths therebetween. Each node includes an interface between high-level input and output commands at the user level and input and output commands at the system level, the input and output commands at the system level controlling the sequence of operations used to interact with the database, wherein, in response to specific system-level commands, an atomic object generates atoms, each atom containing a specified fragment of data or metadata, whereby the set of all instances of atoms collectively defines all of the metadata and data in the database. Each node further includes: communication control for establishing a communication path with each node in the system; a method for requesting a copy of an atom that is relevant to a query but not present in that node from a selected node in response to a system command from the interface; a method for replicating the requested atom for transmission to the requesting node in response to a request for an atom from another node, whereby only the atoms required to complete the query need be present in any transaction node at any given time; and a method for replicating that atom in response to a change in the atom at that node for transmission to every other node in the system where the atom resides. The persistent storage contains a collection of atoms of all the data and metadata collectively contained in the database.

According to another aspect of the present invention, a database management system enables users to interact with a database consisting of data and metadata. The system includes at least one transaction node that provides user access to the database and at least one archive node that maintains an archive of the entire database. Each transaction node includes a database request engine that provides an interface between high-level input and output query commands at the user level and input and output commands at the system level. The input and output commands at the system level control the sequence of operations used to interact with the database. In response to specific system-level commands, atomic objects generate atoms. Each atom contains a specified fragment of data or metadata, whereby the set of all instances of atoms collectively defines all metadata and data in the database. A database system network interconnects all of the nodes. Communication control in each of the nodes establishes a communication path with each node in the system. Methods in each transaction node, in response to system commands from the database request engine, request a copy of an atom that is relevant to the query but does not exist in that node. Another method in each node, in response to a request for atoms from another node, replicates the requested atoms for transmission to the requesting node, whereby only the atoms required to complete the query command need be located in any transaction node at any given time. Another method in each transaction node, in response to a change in an atom at that node, replicates that change to every other node in the system that contains a replica of that node.

According to another aspect of the present invention, a database management system for a logical database comprises data records organized into tables to be accessed from a plurality of transaction nodes, the plurality of transaction nodes processing transactions associated with the logical database. The database is parsed into fragments, each fragment storing a portion of the metadata and/or data associated with the logical database for transmission as serialized messages within the database management system and stored as non-serialized messages. The system includes at least one archive node storing all of the fragments in non-serialized form in persistent storage, thereby forming a single repository for the entire database. Each transaction node responds to queries from a user by establishing a sequence of low-level commands for identifying the fragments associated with the query, and responds to the low-level commands by obtaining copies of existing fragments associated with the query being processed, whereby a given fragment may be located at some other node or only at the archive node. Each transaction node replicates any changed fragment to the at least one archive node and to each transaction node where a copy of that fragment resides, whereby changes to fragments in other nodes are made on a peer-to-peer basis, and whereby any transaction node contains only those fragments that are relevant to queries made by users accessing the database through that transaction node.

BRIEF DESCRIPTION OF THE DRAWINGS

The subject matter of the invention is particularly pointed out and distinctly claimed in the appended claims. Various objects, advantages and novel features of the invention will become more fully apparent from the following detailed description when read in conjunction with the accompanying drawings in which like reference numerals indicate like parts, and in which:

FIG1 is a diagram in schematic form of one embodiment of a resilient, scalable, on-demand, distributed data processing system incorporating the present invention having interconnected transaction and archival nodes;

Figure 2 describes the organization of transaction nodes;

Figure 3 describes the organization of archive nodes;

4A and 4B depict the logical organization of “atomic” objects generated by the atomic classes shown in FIGs. 2 and 3 , which are useful in implementing the present invention and which may appear in a transaction node at any given time;

Figure 5 describes the information in the main directory atom;

Figure 6 describes the information in the transaction manager atom;

Figure 7 describes the information in the database atom;

Figure 8 describes the information in the schema atom;

Figure 9 describes the information in the table atom;

Figure 10 describes the information in the table directory atom;

Figure 11 describes the information in the index atom;

Figure 12 describes the information in the record state atom;

Figure 13 describes the information in the data atom;

FIG14 depicts the information in a large binary object (Blob) state atom;

Figure 15 describes the information in a large binary object atom;

FIG16 depicts the syntax of exemplary asynchronous messages transmitted between transactions and archive nodes in the database system of FIG1 ;

FIG17 depicts various message types by which information is transferred between transaction and archive nodes in the database system of FIG1 ;

FIG18 is a flowchart useful for understanding the method by which nodes are joined to the database system in FIG1 ;

Figure 19 describes the information in a node object;

FIG20 is a flowchart useful for understanding a method for creating atoms through their nodes according to the present invention;

FIG. 21 is a flowchart useful in understanding the method by which unique atom identifiers are assigned during the method of FIG. 20 ;

FIG. 22 is a flowchart useful in understanding a method by which one node obtains a copy of an atom from another node; and

FIG. 23 is a flow chart useful in understanding the method by which the present invention commits a transaction.

DETAILED DESCRIPTION

FIG1 illustrates one embodiment of a resilient, scalable, on-demand, distributed database system 30 having multiple data processing nodes incorporating the present invention. Nodes N1 through N6 are "transaction nodes" that provide user application access to the database; nodes A1 and A2 are "archive nodes" that maintain a disk archive of the entire database at each archive node. While archive nodes typically store the entire database, a single transaction node contains only the portion of the database required to support the transactions executing at that node at that time.

Each node in Figure 1 can communicate directly with other nodes in the system through the database system network 31. For example, node N1 can establish a communication path with each of nodes N2 to N6, A1 and A2. Communication between any two nodes is by means of serialized messages. In a preferred embodiment, message transmission is performed in an asynchronous manner to maximize the bandwidth used by the system, thereby performing various operations in a timely and rapid manner. Typically, the database system network 31 will operate with a combination of high-bandwidth, low-latency paths (e.g., Ethernet) and high-bandwidth, high-latency paths (e.g., wide area networks). Each node has the ability to limit the use of low-latency paths to time-critical communications (e.g., obtaining atoms). High-latency paths can be used for non-critical communications (e.g., requesting to update table information). Moreover, and preferably, the data processing network of the present invention includes a message transmission protocol such as the Transmission Control Protocol (TCP) and ensures that each node processes messages in the same order in which they are sent to each node by other nodes.

FIG2 depicts a representative transaction node 32 connected to a database system network 31 and various end users 33. Transaction node 32 includes a central processing system (CP) 34 that communicates with database system network 31 via a network interface 35 and with various users via a user network interface 37. Central processing system 34 also interacts with RAM memory 38, which contains a copy of a data flow management program that implements a preferred embodiment of the present invention. This program is used to provide a remote interface 40, a database request engine 41, and a set of classes and objects 42.

The database request engine 41 is located only on the transaction node and is the interface between the high-level input and output commands at the user level and the system-level input and output commands at the system level. Generally speaking, 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 set 42.

For purposes of explaining the present invention, the set of classes/objects 42 is divided into a subset of "atomic classes" 43, a subset of "message classes" 44, and a subset of "helper classes" 45. Additional details of these classes are described below.

As will become apparent and in accordance with the present invention, at any given time, a transaction node contains only those portions of the total database that are relevant to active user applications. Furthermore, the various features of the present invention enable all portions of the database in use to reside in random access memory 38. It is not necessary to provide supplemental storage, such as disk storage, at the transaction nodes during operation of this system.

Referring to FIG3 , each archive node 50, such as archive node A1 or A2 in FIG1 , is also connected to a database system network 31. However, instead of end users 33 associated with transaction nodes 32 in FIG2 , the archive nodes are connected only to persistent storage 51, which is typically a disk-based storage system or a key-value store. Archive nodes 50 include a central processing system 54, which communicates with persistent storage 51 via I/O channels 52 and with database system network 31 via network interface 55. Central processing system 54 also interacts with RAM memory 57, which contains a set of classes or objects 62. Similar to transaction nodes 32 in FIG2 , the class/object set 62 in FIG3 includes a set 63 of "atomic classes," a set 64 of "message classes," and a set 65 of "helper classes."

A preferred embodiment of the present invention uses object-oriented programming (OOP), in which, as is known in the art, classes and subclasses, such as those shown in Figures 2 and 3, define methods, data structures, and processes by which "instances" or objects of that class or subclass can be generated. "Inheritance" and/or "polymorphism" can be used to generate "instances." It will become apparent to those skilled in the art that implementations that do not use object-oriented programming or variations of the specifically disclosed embodiments are possible.

The present invention will now be described in several stages. The "Atoms" section defines the hierarchy and functionality of objects generated by the Atom classes 43 and 63 in Figures 2 and 3, respectively. The "Messages" section describes the message set that provides communication between transactions and archive nodes that may be generated by the message classes 44 and 64 in Figures 2 and 3, respectively. The "Methods" section describes the basic operations with respect to database management. The "Examples" section describes the interaction of atoms, messages, and methods, thereby achieving the objectives of the present invention in response to a specific database query to the database request engine 41.

atom

As previously mentioned, each of the atomic classes 43 in Figure 2 and 63 in Figure 3 produces an "atom". More specifically, an atomic class defines one or more "atom types" or "atom objects". Each "atom type" or "atom object" produces itself, an "instance" of an "atom". As will become clear when the purpose of each particular atomic object is understood in more detail, each "atom" contains a specific piece of database information. Some atoms contain part of the database metadata; others contain data records; and others serve as directories for creating and tracking other atomic types. Some "atom types" may instantiate only one atom that is replicated to all nodes. Other "atom types" may instantiate multiple atoms that are replicated to other nodes on an on-demand basis.

Atoms have specific characteristics. In a transaction node, atoms exist only in non-persistent memory and take the form of non-serial messages that have been populated with a specific atom type to provide an efficient memory-resident format for that atom. Each atom has a means for encoding its contents as a serial message and a means for decoding the serial message to retrieve the atom's contents. Such serial messages are used in conjunction with a number of operations, as will be described later.

Each serial message sent from a node replicating an atom includes the contents of that atom with the node identifier appended and the most recent transaction commit sequence number for that node. When the archive node receives that serial message, it deserializes the message (i.e., atom) before placing the remaining contents in permanent storage, removing the node list and commit sequence number.

A number of rules apply to atoms according to the present invention. The reasons and implications of those rules will become clearer. First, each atom must have a unique identifier to provide reliable identification of that atom anywhere in the database processing network 30 in Figure 1. Second, any atom must exist in two nodes at the same time to maintain redundancy, except that a single atom can exist after the single atom is created and before the archive node has requested a copy. Third, transaction nodes will only install atoms on demand. Fourth, each time a change is made to an atom in a node, that node must replicate that changed atom on a "peer" basis; that is, replicate that changed atom to all archive nodes and only those transaction nodes that contain that same atom.

A "garbage collection" process, described in more detail below, may occur on archive and transaction nodes. This process removes inactive atoms from transaction and archive nodes, with the result that a transaction node may store those atoms that are currently relevant to the user application in random access memory at that node. Thus, the database request engine 41 "sees" the entire database as local and is unaware that it is operating in a multi-node environment and does not have a complete copy of the database at its node. Archive nodes have the option of clearing the contents of an atom after it has been serialized to disk, thereby reducing the size of the memory required for storage. If an archive node receives a replication message from such an atom, the archive node must obtain the content from disk memory and then apply the replicated information from the atom.

Against this general background, each atom type will now be described at a "logical" or functional level. This information, combined with a further discussion of the operation of the invention, will enable one of ordinary skill in the art to make and use the invention in any of a variety of implementations, including implementations based on object-oriented programming.

4A and 4B illustrate a typical set of atoms that may reside in a transaction node 32 and the database engine 41 at any given time. In this example, the transaction node hosts a master directory atom 70, a transaction manager atom 71, a database atom 72, a schema atom 73, a table atom 74, and a table directory atom 75. Each database has only one master directory atom 70, one transaction manager atom 71, and one database atom 72. When a database requests that the engine 41 create a new database, the transaction manager atom 71, the database atom 72, and the schema atom 73 are created.

4A , the master directory atom 70 tracks the state of transactions and archive nodes in the database system 30 of FIG 1. It can also be viewed as an active index that creates and monitors the transaction manager atom 71, the database atom 72, each schema atom 73, each corresponding set of table atoms 74, and the table directory atom 75 along with the sequence ID manager 82.

The table directory atom 75 acts as an active index and creates and monitors the index atom 76, record status atom 77, data atom 78, large blob status atom 80, and large blob atom 81 associated with a single table. That is, there is one table directory atom 75 for each table.

Figure 4B is useful for understanding the interaction and management of different atom types. In this context, neither the master directory atom 70 nor the table directory atom 75 performs any management functions. Relative to the remaining atoms, the master directory atom 70 manages each schema atom 73. Each schema atom 73 manages each associated table atom 74 and sequence ID manager 82. Each table atom 74 manages its corresponding table directory atom 75, index atom 76, record status atom 77, data atom 78, large binary object status atom 80, and large binary object atom 81.

Still referring to FIG4B , the database request engine 41 communicates with the master catalog atom 70, the transaction manager atom 71, the database atom 72, each schema atom 73, each table atom 74, and the sequence ID manager 82. The database request engine 41 acts as a 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 to form various pieces of database information.

Each atom has certain common elements and other elements specific to its type. Referring to Figure 5, a master directory atom 70 includes common elements 70A to 70I. Element 70A is the unique identifier of the atom. Because there is only one instance of the master directory atom that is replicated to all nodes, the master directory atom ID 70A is given a fixed number, usually "1". As a general rule, pointers 70B and 70C identify the master directory atom and the create directory atom, respectively. For the master directory atom, two pointers identify the master directory atom itself.

Each atom must have a chairperson. The chairperson performs the functions described below. Element 70D is a pointer to the node where the chairperson for that atom resides.

Each time a copy of an atom is changed at any transaction node, it receives a new change number. Element 70E records that change number.

Whenever a node requests an atom from another node, there is an interval during which the requesting node will not be known to other transaction nodes. Element 70F is a list of all nodes for which the supplying node must relay messages from all other nodes containing that atom to the requesting node until the request is fulfilled.

The operations of the database system are also divided into cycles. Cycle reference element 70G provides the cycle number of the last access to the atom. Element 70H is a list of all active nodes containing the atom. Element 70I includes several status indicators.

Still referring to FIG5 , the global node IDS entry 70J contains an ID manager for assigning a unique identifier to each active node in the system. As is known, such identifiers are long strings. The local node IDS entry 70K includes a range of numbers up to the total number of nodes that can be attached to the system. Together, these entries provide a correspondence between the two types of identification. The use of local node IDS improves efficiency.

When a transaction node joins the database system 30 in FIG1 , the connection manager 70L performs that process. A specific implementation for enabling a transaction node to join the database system is described below. The joining node uses the pending node status entry 70M to indicate that it does not have a global address for a responding node until it receives further communication from that responding node. The database UUI entry 70N contains the universally unique identifier of the database.

The entries at 70P are important because they link all atoms for which the master directory atom 70 acts as an active index. As previously described, these include the database atom 72 and each of the schema atom 73, table atom 74, and table directory atom 75.

Password entry 70Q represents the means for authenticating connections into the database. Actual and Software Version entries 70R and 70S allow the system to operate with backward compatibility when newer versions of the software are installed. Actual Software entry 70R identifies the software version currently in use; Software Version entry 70S is the number corresponding to the most recently installed version. This allows individual nodes to be updated to newer versions without requiring updates to other nodes and without shutting down the database accessed by all nodes.

Still referring to FIG5 , the master directory atom 70 also includes a pointer 70T to the transaction manager atom 71, a pointer 70U to the configuration object, a pointer 70V to the garbage collection thread, and a pointer 70W to the ping thread. The ping thread operates periodically and independently of other operations at the node. It "pings" each node to provide information that can be used in determining the communication efficiency of the corresponding path. For example, if node N1 in FIG1 has the option of communicating with node N2 or N5, node N1 can use the ping information in the most efficient selection of the communication path to nodes N2 and N5 for that communication. Other selection processes can also be replaced or added.

6 , a transaction manager atom 71 exists for each database and is created during the same process as the master directory atom 70. The transaction manager atom 71 creates, tracks, and ends database transactions in response to database commands from the database request engine 41. The transaction manager atom 71 includes elements 71A-71I corresponding to similar elements in the master directory atom. However, element 71A is the identifier of the transaction manager atom 71. Elements 71B and 71C both point to the master directory atom 70.

The ID manager 71J provides unique transaction sequence identification and maintains a list of active transactions 71K, a list of committed transactions 71L, and a list of failed transactions 71M. Element 71N stores commit sequence information. The ID manager 71J assigns a transaction ID at the beginning of each transaction. Each transaction ID is unique, but not necessarily sequential. The local transaction manager atomically assigns a commit sequence number to element 71N when a transaction commits. The sequence numbers are sequential and each is specific to the node requesting the transaction. A transaction transition event counter 71P identifies discrete events that occur during each transaction, such as the start of a transaction and the successful commit of a transaction. Such a counter is useful when multiple transactions involving the same information overlap.

7 , a database atom 72 is created at the same time as the master directory atom 70 and the transaction manager atom 71. The database atom 72 identifies each of the schema atoms 73. The database atom 72 may be involved in the authentication process when a new user attempts to join the database. It may also include other data regarding the authorization level.

In general, database atom 72 includes elements 72A-72I corresponding to similar elements in Figure 5. Element 72A is a database atom identifier. Pointers 72B and 72C each identify a master directory atom 70. Schema name-schema ID registry 72J associates schema names with schema atom identifiers.

Referring to Figure 8, a schema atom 73 creates and tracks table atoms for that schema. The database atom 72 can manage multiple schema atoms, and each schema atom can interact with multiple table atoms. The schema atom 73 includes elements 73A-73I corresponding to elements 70A-70I in Figure 5. Element 73A is a unique schema atom identifier 73A, and elements 73B and 73C are pointers to the master directory atom 70. Tables have unique names in a schema. The table name-table atom ID registry 73J provides a correspondence between each table name and the corresponding table atom. Each schema sequence has a name. The sequence name-sequence ID manager registry 73K provides a relationship between those names and the corresponding sequence ID manager, such as the sequence ID manager 82 in Figures 4A and 4B, associated with each schema atom.

FIG9 provides a logical view of a table atom 74 that contains metadata related to fields, formats, indexes, and types and manages each of the index atom 76, record status atom 77, and large binary object status atom 80 of that table. It also creates and tracks data within the table. The table atom 74 includes elements 74A-74I corresponding to elements 70A-70I in FIG5 . Element 74A includes a unique table atom identifier, and elements 74B and 74C both point to the main directory atom. Pointer 74J identifies the corresponding table directory atom. Element 74K includes a list of all fields of the table.

Each table atom has several ID managers. Pointer 74L points to the ID manager, which provides a unique identifier for each field. Pointers 74M, 74N, 74P, and 74Q identify independent ID managers for assigning identifiers to index atoms, data atoms, large binary object atoms, and subtypes, respectively. Element 74R is a list of existing subtypes. Arrays 74S and 74T provide locations for recording status atoms and large binary object status, respectively.

Referring now to FIG. 10 , there is a table directory atom for each table atom. Each table directory atom 75 is created when the table atom is created. The table directory atom then creates and tracks atoms specific to a table including index, record status, data, large binary object status, and large binary object atoms. Each table directory atom 75 includes elements 75A-75I corresponding to elements 70A-70I in FIG. 6 . Element 75A is a unique table directory atom identifier as assigned by the master directory atom. Two elements 75B and 75C point to the master directory atom 70. An ID manager 75J provides unique atom identifiers for each of the index, record status, data, large binary object status, and large binary object atoms. A list 75K identifies all atoms associated with the corresponding table atom. A pointer in element 75L identifies the location of each atom at the local node associated with the corresponding table atom. A series of lists 75M identify, for each atom, a list of nodes that have a copy of that atom. A bitmap 75N provides a convenient means for identifying other objects and directories when atoms are in an archive node.

Referring to Figure 11, each index in the table has an index atom 76, and each table can have multiple index atoms. Each index atom includes elements 76A-76I corresponding to elements 70A-70I in Figure 5 respectively. Element 76A is a unique index atom identifier as assigned by the corresponding table directory atom. Pointers 76B and 76C identify the main directory atom and the table directory atom respectively. Element 76J contains a binary tree of index nodes to provide traditional indexing functions. Element 76K contains index levels. Such index structures and operations are known to those skilled in the art.

Returning to Figure 12, the record status atom 77 manages the record versions and status for a fixed range of record numbers within a single table. Therefore, a given table atom can manage multiple record status atoms. Each record status atom includes elements 77A-77I corresponding to elements 70A-70I in Figure 5. Element 77A includes the record status atom ID assigned by the create table directory atom. Pointers 77B and 77C identify the main directory and table directory atoms, respectively. Element 77J is an array for locating all data atoms managed by the record status atom 77. Element 77K contains the record number for the "basic record". That is, each data atom stores multiple records. Element 77 is a pointer to the corresponding table atom.

In database applications targeted by the present invention, multiple users can generate multiple versions of the same record. A preferred embodiment of the present invention uses multi-version concurrency control (MVCC) to ensure that transactions never have to wait for the database by allowing several versions of a record or other object to exist in the database simultaneously. Consequently, each record state atom 77 includes metadata about each record version. Entry 77M is a bitmap identifying the location of each versioned record that is useful in garbage collection.

The record status atom 77 includes a transaction ID 77P for each version 77N of the record to identify the transaction that generated the version. The format version entry 77Q identifies the version number of the table subtype that existed when the record was inserted. This format identifies the physical order of the records and identifies the subtype to which the record belongs for the database program in use when the record version was created. Element 77R includes the record version sequence number; element 77S, the position of the next older or previous version of the record. The index 77T and the base record identifier 77K to the data atom array 77J together provide the address of the actual time slot 77U in the data atom that has the record version.

FIG13 illustrates a data atom 78 having elements 78A-78I corresponding to elements 70A-70I in FIG5 . In data atom 78, element 78A is a data atom identifier 78A assigned by a table directory atom. Elements 78B and 78C are pointers to a master directory atom and a corresponding table directory atom, respectively. ID manager 78J assigns a record slot identifier to each record in data atom 78. Element 78K identifies the address and length of each record in data atom 78. Element 78C represents a data record and its version.

Now, referring to Figure 14, the database also stores "large binary object records". A "large binary object record" is typically a collection of binary data stored as a single entity in the database. Large binary object records do not exist in the version. The large binary object state atom includes elements 80A-80I corresponding to elements 70A-70I in Figure 5. Element 80A has a large binary object state atom unique atom identifier. Elements 80B and 80C are pointers to the main directory and table directory atoms, respectively. List 80J identifies all large binary object atoms managed by a single large binary object state atom 80. Entry 80K provides the identifier of the basic large binary object record. Element 80L points to the corresponding table atom. For each large binary object record, the large binary object state atom includes an index 80M for the large binary object atom. Element 80N identifies the time slot in the large binary object atom used for the large binary object record.

FIG15 depicts a large binary object atom 81 having elements 81A-81I corresponding to elements 70A-70I in FIG5 , respectively. Element 81A is an atom identifier assigned by a table directory atom. Elements 81B and 81C are pointers to a master directory atom and a corresponding table directory atom, respectively. An ID manager 81J assigns a large binary object slot identifier to each large binary object in the large binary object atom 81. Element 81K identifies the address and length of each large binary object in the large binary object atom 78. Element 81L represents all large binary object records assigned to the large binary object atom.

In summary, each atom has a relationship to only a fragment of the database. For example, the database atom 72 contains metadata identifying the schema for the database. Each schema atom 73 contains metadata identifying all tables associated with that schema. For each table, a table atom 74 and a corresponding table directory atom 75 provide metadata about the table including information such as the identification of the fields and their attributes. The record status atom includes metadata about a set of records. The data atom includes information about each data record with a pointer to the slot containing the records and their versions. The large binary object state and large binary object atoms contain similar information about large binary object records.

information

As previously described, communication between any two nodes is by means of serialized messages that are sent asynchronously using TCP or another protocol with controls for maintaining the sequence of message transmission. FIG16 illustrates the basic syntax of a typical message 90 comprising a variable-length header 91 and a variable-length body 92. The header 91 includes a message identifier code 93 that specifies the message and its function. Because the present invention contemplates situations where different nodes may operate using different software versions, the header 91 also includes an identification 94 of the software version that created the message. The remaining elements in the header include a local identification 95 of the sender (i.e., from the main directory atom in FIG5 ) and the destination of the message, i.e., information 96 of the directory atom (e.g., table directory ID 75A in FIG10 ) and an atom identification 97 (e.g., record status ID 77A in FIG12 ). From this information, the receiving node can deserialize, decode, and process the message.

Figure 17 depicts a set of messages having the syntax of Figure 16 for use with a particular embodiment of the invention. Each performs a specific function which will now be described.

As briefly described above, when a message is to be sent, there are different communication paths to different nodes. For example, if a node acting as a requesting node needs to obtain an atom, a copy of that atom may be located in multiple other nodes. In this embodiment, "pinging" provides selection information useful for selecting the node corresponding to the best path. As is known, pinging includes determining the time it takes for a "ping" command to reach its destination and receive a confirmation message. In this embodiment of the invention, each node periodically uses a helper class to send a ping message 110 to each of the other nodes it is connected to. Each receiving node uses a helper class to return a ping confirmation message 111 containing the ping time. Each node accumulates this information about the time of sending and receiving these messages in a node object described below with respect to FIG. 8. When a node is ready to send a message to one of multiple nodes, the sending node analyzes factors including, but not limited to, the accumulated ping data to select one of the nodes as the receiving node for that message.

The next set of messages involves connecting a new node into the database system 30 in FIG. 1 or connecting a previously inactive node back into the database system 30 in FIG. 1 . When such a node, such as transaction node N2, wishes to connect to the database system 30, it initiates the connection process described in detail below with respect to FIG. 19 . Referring to FIG. 17 , once that process identifies an active node for receiving messages, the joining node sends a connect message 112 to the selected node. The selected node returns a welcome message 113 and a new node message 114 to all other connected nodes in the database system 30. Each of the other connected nodes sends its own welcome message 113 to the joining node. When this message sequence is complete, the joining node can then proceed to further steps to obtain various atoms.

Archive nodes can operate in active mode or synchronization mode when they are synchronized with another node.A helper class in an archive node sends node status messages 115 to alert all other nodes of any status changes in that archive node.

The next set of messages is involved when a node, acting as a requesting node, retrieves a copy of an atom from another node. For example, after a node joins the database system 30 in FIG1 , it typically requests a copy of the master directory atom. This process is described in more detail in conjunction with the explanation of FIG20 and FIG21 .

Still referring to FIG17 , the requesting node sends an object request message 116 to the selected node, which typically returns an object message 117 with the requested atom. The selected node also sends an object available message 118 to all other nodes with the requested atom. Each node that receives the object available message 118 from the selected node returns an object confirmation message 119 to the selected node. After the selected node has received all object confirmation messages 119, the selected node sends an object completion message 120 to the requesting node.

In some cases, the selected node sends an object unavailable message 121 to declare that the selected node has discarded the requested atom. The bounce object message from the selected node indicates that the requested atom was not found in one of the main directory atom or the table directory atom. This may occur when an update transaction is in progress and the selected node does not respond to the object request message 116 because the garbage collection process has collected the atom before the object request message was received. In response, the requesting transaction node can select another node in the database system that has the atom.

The database request engine 41 in FIG1 and FIG4A and 4B can periodically generate a register object message 123 or a deregister object message 124. These messages are for atoms comprising a register, such as database and schema atoms. When a user at a node issues a command to delete an item, such as a table, an object delete message 125 is sent from that node.

Whenever a local node is updated or modified, its change number is incremented. Each replication message contains the local change number for that atom. Each atom keeps track of the latest change number for each node. When an archive node receives a copy of a modified atom, it copies the atom's change number, clears the change number, and then serializes the atom to disk. If changes have been made, the change number will be non-zero. The archive node sends an object write message 126 with the change number written to each node. Each receiving node compares its own change number with the change number in the message. The node can then update the state of the atom to note that the atom has been archived and is a potential candidate for garbage collection.

As previously indicated, each atom must have a unique identity. When the first atom of a particular atom type is created (e.g., a new table atom), the creating transaction node is designated as the chairperson of that table atom. The rules for controlling the "chairperson identity" are described below. Whenever a transaction node needs to create a new atom type, if it does not have an identity available, it sends an ID request message 127 to the chairperson. The chairperson typically returns an ID delegation message 128, which includes a block of at least one unique identity from its allocation of free identity values. This process will be described in more detail with respect to the process of Figure 20.

4A and 4B , the table atom 74 can send any one of a number of messages. If the database request engine 41 initiates processing for adding a new field, the table atom at that corresponding transaction node generates the new structure and sends a table field add message 129, which replicates the change to all other nodes that include that table atom. If a transaction node updates a field, subtype, or field attribute that changes the table format, that node issues a table format message 130. Whenever a node needs to create a new record state or large binary object state atom, a table request record message 131 is generated. Only the chairperson can create this atom; and, when this occurs, the chairperson broadcasts a table record object message 132.

Each time a transaction node inserts a new record into a table, it generates a table record message 133. Whenever a new index needs to be created, such as when adding an indexed field to a table, the create table atom copies the new table atom. In this case, the index is set to a write-only index. After all the relevant processing involved has completed, the node sends a table index add message 134.

Any time a table atom creates a large binary object state atom for a large binary object, the table atom generates a table large binary object message 135. The table large binary object message 136 indicates that a new large binary object has been created.

In a database that utilizes types and subtypes, the database request engine 41 in Figure 1 will generate a command for causing a table atom to be assigned a new table type identifier. When this occurs, a table type message 137 is replicated to all nodes that have similar table atoms.

A table record erase message 138 provides the record number associated with a particular record status atom. When it is determined that a record within that table contains an unusually long chain of reversed versions or other criteria, the chairperson of the table atom generates a table garbage collection message 139. The result is that "unused" atoms are "purged" of data.

The record status atom 77 in Figures 4A and 4B also generates several special messages. If it becomes necessary to update a particular record, the database management system of the present invention creates a new version of that record. Referring back to Figure 17, the corresponding record status atom generates a record update request message 140, which is directed to the chairman of that atom to seek permission to update that particular record. The chairman responds by generating a record update response message 141, which grants or denies permission to update that record. If the chairman grants permission, the record status atom is requested to perform the update, and a record update message 142 with the new version of the atom is sent to each node that has a copy of that atom.

Data atoms store up to a certain maximum number of records and their versions. The corresponding record status atom monitors the size of the data atoms it manages. If the record status atom determines that the managed data atom has exceeded that size, it generates a new data atom and copies the new data atom with the help of a record data object message 143.

The record status atom generates a record record message 144 for replicating any new record versions. Periodically, the table atom chair initiates a process by which the record status atom identifies record versions older than the oldest active transaction. When this occurs, the chair's record status atom sends a record delete message 145, which enables deletion of those older record versions during a subsequent garbage collection process. If a situation arises in which a transaction recovery becomes necessary, the record status atom generates an undo record message 146 that updates the undo record.

There is also an index-specific set of messages. As is known, an index has an optimal maximum size; and if the index exceeds that size, the index should be split. According to the present invention, only the chairman can split an index atom. The chairman can do so unilaterally or in response to an Index Split Request message 147 from another copy of that atom. When the chairman causes the split, the chairman generates an Index Split message 148 containing the split index. After the split, the chairman sends an Index Delete message 149 to truncate the original index atom after the split. Each time an index node is added to the index, it generates an Index Node Add message containing the index key, record identifier, and other information of the new index. When the index has been fully populated and is therefore ready to be used during retrieval operations, a Table Index Ready message 151 is generated. As previously shown, adding a table index generates an index atom, which is a write-only index and is not readable. The Table Index Ready message 151 makes such a write-only index readable.

The large binary object state atom identifies a time slot in the large binary object atom by generating a large binary object large binary object message 152.

The data record message 153 contains the time slot number and the record length. It also includes the data record.

When the state of a transaction changes, the transaction manager atom 71 for a given transaction node generates transaction status messages 154. These indicate whether the current state of the transaction is active, pre-commit, committed, or recovered. In certain circumstances, it is possible that a request associated with one transaction will be blocked by another transaction on another node. In that case, the transaction manager atom receives a transaction blocked message 155. If a transaction is stuck and must be recovered, the transaction manager atom associated with the node causing the deadlock generates a transaction deadlock message 156, which causes the transaction to be recovered.

The time of the last write to an atom can be requested for the home directory or any table directory atom on an archive node. This occurs when the requesting node sends a Request Write Time message 158. The recipient of that message then returns those requested times in a Write Time message 159.

The aforementioned messages constitute a set by which the various processes required to maintain a database management system incorporating the present disclosure can be properly handled. As will be apparent, each message has minimal overhead in the header. Each message can be reasonably short. When used with TCP or other messaging protocols, the messages must be sent in sequence, and at any given node, upon receipt, they must be processed in the same order in which they were sent, even by different nodes.

method

It will be useful to further understand the invention describing some basic methods associated with various aspects of the operation of the invention. Variations with respect to each will be apparent to those of ordinary skill in the art.

FIG18 is a flow chart of the operations that occur in various nodes in database system 30 of FIG1 when a node joins the network. For the purposes of this description, assume that node 5 in FIG1 is joining the database system; it is designated as joining transaction node 170. As a first step, joining transaction node 170 uses step 171 to establish a TPC connection with a selected node. Essentially, joining transaction node 170 sends a message to a fixed location identifying the database. A connection agent, not shown but known in the art, responds to this request by either denying or granting access to the database system. If the connection agent grants access, it selects a node, such as transaction node N1, as selected node 172. The connection agent then sends a message to joining transaction node 170, designating selected node 172 as the port number. FIG18 also represents all other active transaction and archive nodes in the database system as group 173.

Once the connection is established in step 171, the joining transaction node 170 uses step 174 to send a connect message to the selected node 172. The selected node 172 responds to the connect message in step 175 by using its connection manager to update its home directory atom, assigning a local node ID to the joining transaction node 170, and adding that node to the array of local nodes in its node object, an example of which is shown in FIG.

FIG19 depicts a node object such as that described in conjunction with step 175 of FIG18. It contains a pointer 400A pointing to the socket for the node, a pointer 400B pointing to the home directory atom at the node, and a pointer 400C pointing to the connection manager in the home directory atom. Pointer 400D identifies the thread for listening for incoming messages, and pointer 400E identifies the socket buffer for receiving messages. The message listener waits for a message, determines the message type, and then processes the message to completion.

Node object 400 comprises the global node ID 400F and local node ID 400G of the node that this node is listening to as all atoms.Element 400H is the pointer to the queue that points to and waits for the message that will send from node.Element 400I and 400J comprise the identification of local port and remote port.Element 400K comprises the version number of the software that is used to operate at remote node place.Node type element 400L indicates whether remote node is transaction node, archive node that carries out synchronous processing or online archive node.

Element 400M contains the name of the local node; element 400N contains the name of the remote node. Element 400P is a pointer to the current message being processed. Elements 400Q and 400R identify the time and ping time of the last ping operation. As previously shown, each node generates a sequential submission sequence number in response to each commit operation for a transaction initiated at that node. Element 400S contains this number. Element 400T indicates whether this node object is a node object for this node.

18, at step 176, the selected node 172 sends a welcome message to the joining transaction node 170 containing the global node ID. The selected node then broadcasts a new node message to the group 173 of all other transaction and archive nodes in step 177.

Each node in group 173 responds to the new node message by recording the global ID and assigning the local ID of the joining archive node and updating the local list of all connected nodes in their respective home directory atoms in step 180. Each then uses step 181 to send a welcome message to the joining transaction node 170. Upon completion of this process, the joining transaction node 170 has a complete list of all active nodes, including the selected node 172 and all nodes in group 173.

When the joining transaction node 170 receives the welcome message from the selected node in step 182, it sends an object request message to the selected node 172 (step 183) to request a copy of the master directory atom. After the selected node 172 has updated the various information items in its master directory atom, the selected node implements step 184 to serialize its master directory atom in the object message sent to the joining transaction node 170 and broadcast the object available message to all other nodes in the system. Thus, the master directory atom in each node is updated and synchronized.

Without waiting for receipt of an object message, the joining transaction node may also begin processing at step 185 for retrieving copies of database atoms and transaction manager atoms from the selected node 172 .

Each other node that receives the object available message responds in step 186 by sending an object confirmation message back to the selected node.

The database request engine in the joining node initiates this sequence. Thus, when the method of FIG. 18 has been completed, the joining transaction node 170 is connected to the database system and, as shown in FIG. 4A and 4B , includes copies of the master directory atom 70, the transaction manager atom 71, and the database atom 72. The joining transaction node 170 will thereafter create or obtain copies of the other atoms on an as-needed basis.

During some operations, a transaction node may create a new table, necessitating the creation of a new table atom. FIG20 discloses a process 190 in which node A is the requesting node and atom X is the atom that requests and will manage a new atom Y. To create a new table atom Y, atom X will be a schema atom and local directory Z will be a master directory atom. Step 191 represents the preparation function for the request to obtain an instance of atom Y. Each time an atom is changed, it is assigned a change number; the initial value is usually "0".

At step 192, local directory Z creates an instance of atom Y with no content and designates the local node as the chairperson of atom Y. Process 193 then allows local directory Z to assign an object ID to the new atom Y. Details of such processing are shown and described with respect to FIG.

Next, local directory Z sets an exclusive lock to allow changes to it to occur without any influence from external automation. While the lock is in place, local directory Z sets the state of the new atom Y to a "dirty" state and a "non-open" state. The "dirty" state indicates that the new atom Y has not yet been copied to the archive. The "non-open" state indicates that the new atom Y is not yet available to other nodes. At step 195, local directory Z updates itself and then releases the exclusive lock. At step 196, directory Z broadcasts an object available message identifying atom Y to all other instances of directory Z in the transaction and archive nodes.

In step 197, Atom X, acting as the management atom for the new Atom Y, populates an instance of Atom Y and sets the state of Atom Y to "open," indicating that the new Atom can be replicated. Sometime thereafter, the archive node will store a copy of the new Atom to persistent storage. That is, the archive node will respond to the receipt of the Object Available message by requesting a copy of Atom Y, thereby providing redundancy. Upon completion, the receipt of the Object Confirmation message from the archive node will cause the "dirty" state to change, and this change will then be reflected to all other nodes that have a copy of Atom Y.

The allocation process 193 in Figure 20 and similar allocation processes are used in various stages of the operation of the present invention for allocating object IDs. Figure 21 describes this allocation process in more detail. Specifically, when a directory atom wants to allocate an object ID, it determines in step 200 whether it has a local ID available for it. If it does, control passes to step 201, which allocates an available ID in response to the request. If the local ID is not available, control passes to step 202 to determine whether the directory atom at this node has "chairman" status. If so, the chairman has authorization to directly identify an available ID, and control passes to step 201.

As will be recalled, when there is at least one other node for a particular atom, each atom contains a list of nodes that include a copy of the atom being requested. If this is the first request for that atom after the atom has been created according to the process of Figure 20, the corresponding directory in the first node on the list is the chairman. Step 203 represents the process of selecting a node and identifying the chairman. For this selection process, it is preferred that communication is first established with the transaction node. Step 204 represents the sending of an ID request message to the chairman.

When the chairperson receives the ID request message in step 205, it obtains the block of available ID numbers (step 206). The chairperson identifies the transaction node making the request as the location of this block of available ID numbers (step 207). The chairperson then sends an ID delegation message to the requesting node in step 210. When the requesting node receives the ID delegation message from the chairperson in step 211, it stores the block of ID numbers in step 212 and then selects the first available ID for allocation in step 201. Referring back to FIG. 20 , in this case, process 193 proceeds directly from steps 200 to 202 to step 201 because the chairperson was designated in step 192 of FIG. 20 .

Whenever the database engine 41 in Figures 4A and 4B makes a request for an atom, the response 220 in Figure 22 is processed. For example, assume that the database request engine 41 in the requesting node 221 requests a copy of atom Y (e.g., a table atom) from a selected node 222 in the database with other nodes 223. When the database request engine 41 makes that request, step 224 determines whether atom Y exists in the requesting node 221. If so, step 225 ends the process because the requested atom exists. If not, control passes to step 226, whereupon the local directory Z creates an empty instance of atom Y and declares atom Y as "unpopulated". If the database engine 41 is requesting a table atom 74, the master directory atom 70 will perform this step. Step 230 then uses the aforementioned selection process to identify the selected node 222. The preference is to select any one of the transaction nodes before selecting the most responsive archive node as the selected node 222.

In step 231, the requesting node 221 sends an object request message for atom Y to the selected node 222. In response, the selected node 222 uses step 232 to send an object message with the requested atom Y in its serialized form with the node and sequence number.

At the same time, the selected node 222 broadcasts an object available message to all other nodes 223. It also creates a repeater list for all other nodes that have copies of atom Y. At this point in the process, the other nodes do not communicate directly with atom Y in the requesting node because they do not know that they should send a replication message for that atom to the requesting node. Therefore, when any of the other nodes replicates its atom Y, the selected node 222 will relay the message to the requesting node 221.

When the requesting node 221 receives the object message from the selected node 222, it performs accessibility analysis of the message in step 233. If the message contains the current atom, atom Y processes the message in step 236 and sends an object confirmation message to the selected node in step 237.

All other nodes 223 respond to the object available message using step 240 by sending an object confirmation message to the selected node 222. The selected node 222 monitors the object confirmation messages using step 241. Specifically, it removes each of the other nodes from its repeater list in response to each object confirmation message and stops relaying to that node. When all other nodes 223 have been removed from the list, the selected node stops all relaying and broadcasts an object completion message.

It is possible that another node sends a Duplicate message that arrives at the requesting node 221 during the time between steps 226 and 236 in Figure 22. This can interrupt message processing. As a result, when a Duplicate message for atom Y in the above example is received at the requesting node while atom Y is not populated, it is placed on the pending message list as part of each atom, although not shown in any of Figures 5 to 13.

As previously indicated, the present invention is particularly well-suited for databases interfaced with transaction processing technology. In this regard, appropriate means for "commit" messages should be included. FIG. 23 illustrates one means for ensuring data consistency in such an environment. Specifically, FIG. 23 depicts four nodes: transaction node A 250, transaction node B 251, transaction node C 252, and archive node 253. Assume that transaction node B 251 sends a pre-commit message at step 254 with a transaction ID atomically supplied by the transaction manager at that node. This message is routed to the archive node at step 255. When all conditions for committing that transaction have been met, archive node 255 issues a commit transaction message at step 256 and, at step 257, broadcasts that commit message. Each of the transaction nodes responds by updating its corresponding transaction number at step 258.

As previously described with respect to the process for requesting a copy of an atom in FIG21 , the requesting node performs an accessibility analysis on any received message in step 233. This test provides assurance that any transaction node is always operating with valid information and involves analysis of transaction ID numbers and commit sequence numbers. This analysis and other features of the present invention will be facilitated by further understanding transaction IDs and commit sequence numbers, by analyzing the relative order of transactions, and by understanding "atom skew."

With respect to transaction IDs and commit sequence numbers, and as previously described, each transaction identifier is unique across the entire database system of FIG. 1 . The purpose of a transaction ID is to be a unique, permanent, system-wide marker for indicating which transaction created a particular record version. Each transaction ID is assigned by a local copy of a transaction manager atom 71, and one such transaction manager atom will be the chairperson, which assigns a block of identifier numbers to the transaction manager atom on each transaction node in the database system. The transaction manager atom at a given transaction node sequentially assigns an unused number in the assigned block to each transaction that receives a start. As a result, a newer transaction started at a given node can have a higher transaction ID than an older transaction started at that same node. However, on a system-wide basis, a transaction ID does not imply anything about the relative start times of different transactions.

As a result, some method must be provided to ensure that a transaction can read a record version only if the transaction that created the record version has been committed before the reading transaction begins. In the present invention, such a method for determining which record version to read is the transaction commit time of the transaction that created the record version, rather than the start time. The transaction ID does not contain its commit time, so the transaction manager on each node atomically assigns a commit sequence number to each transaction based on the actual commit operation. Each transaction node generates its commit sequence number in an increasing sequence. If a transaction from a given node has a commit sequence number of 467, then all transactions from nodes with lower commit sequence numbers are guaranteed to have been committed. All nodes generate the same commit sequence number, so interpreting the number requires the number and identity of the corresponding node.

When a transaction commits, its transaction manager atomically sends a commit message to all nodes that includes the corresponding transaction ID and commit sequence number. When an atom serializes itself, it includes the highest commit sequence number it has seen from each node that has a copy of that atom. For example, suppose nodes A, B, and C contain copies of atom Z and that node A generates an object message for atom Z. The serialized form of that message will include the highest commit sequence number that node A has seen from itself and the highest commit sequence number it has seen from nodes B and C. The serialized message from node A describes the state of the transaction on all nodes that share a copy of atom Y as node A sees it. It is possible that node B or node C may have actually issued a higher commit sequence number for a transaction that node A has not yet received or processed.

The transaction manager on each node maintains a transaction object for each node in the database system. Each transaction object reflects the status of transactions on all nodes, and each object is assigned two or more local, sequentially increasing numbers. When the transaction manager atomically starts a transaction or receives a transaction transition message indicating that another node has started a transaction, the transaction manager atomically creates a local transaction object with a new transaction ID and assigns it a start number. When the local transaction commits or when the transaction manager atomically receives a transaction transition message indicating that a transaction on another node has been committed, the transaction object is assigned a transaction end number along with the commit sequence number. The start and end numbers are from the same sequence. For example, if transaction 123 has a higher start number than the end number of transaction 453, transaction 123 can read the record version created by transaction 453 from any node executing transaction 453.

Transaction start and end numbers are local; that is, they reflect the state of the transaction as seen on the local node. Different nodes assign different values and see different orderings of transactions on other nodes. As will be apparent, there can be delays in receiving and processing transaction status messages. Each transaction runs only on its local transaction node. This database system prevents a transaction node from "seeing" a transaction as committed before it commits on its local transaction node. For example, if node A begins transaction 123 after receiving and processing the commit message from transaction node B for transaction 453, transaction 123 can read changes from transaction 453, even if transaction node C, which received the commit and begin messages for the transactions in another order, sees the two transactions as contemporaneous.

When no transaction is running on transaction node C, the information is not relevant to transaction node C. However, the difference between the views of the database system at each individual transaction node may cause problems in certain situations.

In contrast to "atom skew," according to one aspect of the present invention, each node must process messages from another node in the order in which it sent the messages. However, all nodes operate independently. Therefore, at any given time, some nodes will have already received and processed messages that other nodes have not yet processed. Some copies of atoms will include changes that other copies do not. These differences do not affect any node with a copy, because each copy of the atom is in a consistent state with that node. As previously described, whenever a transaction node changes an atom, it replicates the changed atom to all other nodes that contain a copy of that atom. If transaction nodes B and C each contain a copy of that atom, and transaction node B changes that atom, transaction node B sends a replication message to transaction node C. While transaction node B is processing the transaction, it sends a transaction status message to transaction node C, replicating any atoms that transaction node B changed. When transaction node B commits the transaction through the process shown in Figure 23, transaction node C should have received the transaction status message indicating that the transaction has been committed. Transaction node C will process all messages related to the transaction, and then it will receive a commit message because all messages will be processed in the order in which they were sent. The copies of the atoms on different transaction nodes may be different, but all changes made by a transaction on any transaction node will be on every other node, which then sees the transaction as committed.

When viewing transactions from a system-level perspective, different transaction nodes act asynchronously. The time it takes for a single message to be processed by different nodes can vary due to differences in the efficiency of communication paths and the number of messages each transaction node needs to process. Differences in processing time at different nodes must be accounted for. Consider three transaction nodes: Node A, Node B, and Node C. Assume that Node B executes transaction 768, which changes Atom X, and Node C has a copy of Atom X. Assume that Atom X at Node B sends a change message to Node C, and some time passes before Node C processes those changes. Assume also that Node B has assigned a commit sequence number of 47 to transaction 768. The transaction manager on Node B atomically sends a transaction status message to all nodes in the database system. Assume also that Node A requests a copy of Atom X from Node C after receiving the commit message from Node B but before Node C has finished processing the transaction. Node A receives and processes the transaction status message from Node B. From Node A's perspective, transaction 768 is committed, and the highest commit sequence number for Node B is 47. However, for this timing, the atom returned from node C does not reflect the changes made by the transaction at node B.

According to the present invention, when atom X on node C serializes itself to send to node A, it includes the highest commit sequence number from every node that has a copy of atom X. In this case, it is assumed that the message will include commit sequence number 97 for node C and commit sequence number 46 for node B.

Referring to Figure 22, node A, acting as a requesting node, creates an empty instance of Atom X and then requests its contents from node C. The accessibility analysis in Figure 22 obtains the current highest committed sequence number for each node as seen by node A. It also extends the serialized message with Atom X to obtain the current highest committed sequence number for each node as seen by node B. In this comparison, the accessibility analysis will determine that the serialized commit sequence number in the message is 46, while its local highest commit sequence number for B is 47. Therefore, node B cannot use the serialized version of Atom X in its current state and must wait for Atom X on node C to send it an object completion message. If the commit sequence numbers are the same or greater, node A can continue processing the atom.

As previously described, when Node C sends a serialized copy of Atom X to Node A, it sends Object Available messages for the other replicas of Atom X, in this case, the replica on Node B. Atom X on Node B sends an Object Confirmation message to Atom X on Node C and adds the replica on Node A to its own replica list. Thereafter, Atom X on Node B sends Change messages for its replicas on both Nodes A and C. Simultaneously, Atom X on Node C processes messages from Node B, relaying them to Node A. Atom X on Node A processes the messages, making itself increasingly up-to-date. When Atom X on Node C processes Object Confirmation messages from Atom X on Node B and all other replicas of Atom X on other nodes, it sends an Object Completed message for Atom X on Node A.

According to the present invention, each transaction node should operate without having to transfer any atoms to disk or other permanent storage. This requires that inactive atoms be removed from each transaction node in a rapid manner. To achieve this goal, a helper class in each node periodically provides garbage collection functionality that is called by the cycle manager to initiate new garbage collection cycles independently of other operations. More specifically, the cycle manager performs an aging cycle whereby any atoms referenced in the previous cycle are moved to the front of the least recently used (LRU) list in the master directory atoms for that node. The cycle manager obtains an exclusive lock to ensure that no other threads are active with respect to the previous cycle.

The cycle manager then starts a record garbage collection thread for the table atoms that were marked for record garbage collection. If the current amount of memory at the node exceeds the specified amount defined by the configuration object identified by pointer 70U at Figure 5, the garbage collection thread processes the atoms in the cycle from least recently used to most recently used through the main directory LRU. It classifies an atom as a candidate for garbage collection if that atom (1) is not active, (2) has been referenced in the current aging cycle, (3) is not an "object not complete" atom, (4) has been changed locally and has not yet been written to the archive node, and (5) is a directory with no memory resident atoms. The atom in the archive node must also not have been serialized to disk.

If the garbage collection process determines that all of the foregoing conditions have been met, there are two options. If the node is an archive node and there are other instances of the atom at other nodes, the contents of the atom are "cleared". Otherwise, the requesting atom discards itself, which it does by first looking at the directory atom it created for a final test. If that test is passed, the directory atom clears its pointer to the candidate atom, broadcasts an object unavailable message to other nodes containing instances of the candidate atom, and then deletes itself. This cycle continues, atom by atom, until working memory is within acceptable limits.

Two mechanisms are involved in this process. When a requesting directory atom finds an atom for classification, the directory atom sets its cycle reference entry to the current cycle number. Additionally, the transaction manager, database, schema, sequence, and table atoms acquire a shared lock on the current cycle so that methods in those atoms can hold onto the atom pointer without having to increment the atom usage count to prevent the object or atom from being garbage collected.

Figures 19 and 20 describe "chairpersonship." As shown in Figure 20, when an atom is created, the created node is designated as the atom chairperson, and an ordered list of nodes for that atom is established. When a node creates an atom, it is the only entry in the ordered list. The first node in the list is the chairperson.

When another node later requests a copy of that atom, the selected node places the requesting node's identifier just after itself in the ordered list, regardless of whether it is the chairperson. Consequently, if the node designated as chairperson for any atom becomes inactive for any reason, every other copy of that atom has an ordered list of nodes. Using this ordered list, the next chairperson is the first remaining transaction node on the list. If there are no transaction nodes on the list, the first unsynchronized archive node is designated.

It may now be useful to discuss a simple database query to help understand the interactions between different nodes and atoms within the nodes. Assume that the database request engine at the transaction node receives a database query to select all records in the table "members" that have the zip code "01944." Assume that the transaction manager atom and the table atom are located at the local node, that all query processing has already completed, and that an index atom exists for the "zipcode" field in the "members" table.

Initially, the database request engine 41 in FIG2 issues a command to the transaction manager to begin a new transaction. At this point, the transaction manager atomically assigns a transaction ID. Next, the database request engine uses the table atom for the "Member" table to determine whether an index atom exists for the "ZipCode" field. If both atoms are present at the local node, they are processed. If not, copies of these atoms are obtained from other nodes, and processing continues.

The table atom for the "Members" table scans the index using information from the corresponding table directory atom. This produces a bitmap that identifies every "record" in the "Members" table that has a specified postal code. As will be apparent, this bitmap can be limited to a specific field or can be the result of a logical combination of indexes from multiple fields.

Next, a loop is established based on the resulting bitmap. For each iteration of the loop, the database request engine issues a call to the table atom to process the get record method specified by the transaction ID in the record. The table atom selects the appropriate record status atom associated with the record atom by dividing the record number by a fixed number corresponding to the maximum number of record IDs managed by the record status atom. Next, the table atom uses the transaction ID and the identified record number to use the get method in the selected record status atom. For multi-version records, the record status atom loops through any record versions to find the correct version and the appropriate pointer to the corresponding data atom. The record status atom calls that data atom with the data atom number pointing to the record and allows it to be retrieved. When this processing is completed, the database request engine provides the user with a record set that lists all records with the specified postal code.

In summary, it will be apparent to those skilled in the art that a database management system constructed in accordance with the present invention provides a resilient, scalable, on-demand distributed data processing system. The system is fault-tolerant and highly available. It is platform-independent and operates to provide a resilient, consistent, isolated, and durable database. Furthermore, it can operate over the Internet without requiring high-speed communication paths and is suitable for transaction processing that can be implemented over a wide geographic area.

The present invention achieves all of these objectives by implementing one or more of the following features. The present invention partitions a database into distributed objects that are replicated on a peer-to-peer basis. Each transaction and archive node determines which atoms to reside in memory on a local basis. Directory atoms track the location of local and remote replicas of each atom and identify the directories of which they are members. Furthermore, each node can determine the best of multiple nodes from which to request replicas of atoms for a geographically distributed system.

The present invention has been disclosed in terms of specific embodiments. It will be apparent that many modifications may be made to the disclosed apparatus without departing from the invention. It is therefore intended that the appended claims cover all such changes and modifications as come within the true spirit and scope of the invention.

Claims (35)

1. A distributed database system, comprising: Multiple nodes, operatively coupled to each other via a communication network, to provide users with access to data in the database, each of the multiple nodes comprising: A memory for storing sets of atoms, each atom in the set containing a specified fragment of data or metadata; and A database request engine is configured to parse user queries into query commands interpreted by the atomic set, retrieve metadata and data from the atomic set stored in the memory in response to the query commands, and transmit requests from at least three of the plurality of nodes to another node for copies of atoms associated with the query commands and omitted from the memory. Each of the at least three of the plurality of nodes includes a corresponding copy of the atom associated with the query command. 2. The distributed database system of claim 1, wherein each of the plurality of nodes is configured to measure latency relative to each other of the plurality of nodes. 3. The distributed database system of claim 2, wherein each of the plurality of nodes is configured to measure the latency by checking each of the other nodes in the plurality of nodes. 4. The distributed database system of claim 1, wherein the atom set includes directory atoms, the directory atoms tracking the position of all atoms in the atom set and the position of each of the plurality of nodes, each node having a copy of each atom in the atom set. 5. The distributed database system of claim 1, wherein the database request engine is configured to receive a serialized message encoding the content of the copy of the atom in response to a request for the copy of the atom. 6. The distributed database system of claim 5, wherein the database request engine is further configured to decode the serialized message to produce a copy of the atom. 7. The distributed database system of claim 1, wherein the database request engine is configured to transmit the requests asynchronously. 8. A method for obtaining data stored in a distributed database system, the distributed database system comprising a plurality of transaction nodes and at least one archive node operatively coupled to each other via a communication network, the method comprising: The user query is received at the first transaction node among the plurality of transaction nodes. The first transaction node has a first memory for storing a first set of atoms, and each atom in the first set of atoms contains a specified fragment of data or metadata. The user query is parsed into a query command through the first transaction node; The first transaction node determines that there are no atoms related to the query command in the first atom set; The first transaction node requests a copy of the atom associated with the query command from a second transaction node among the plurality of transaction nodes, wherein the second transaction node and the at least one archive node include copies of the atom associated with the query command; and In response to a message from the second transaction node, the atoms in the first set of atoms are copied at the first transaction node. 9. The method of claim 8, wherein the set of atoms comprises catalog atoms, and the method further comprises: The catalog atoms are used to track the position of all atoms in the atom set and the position of each of the plurality of transaction nodes, each of which has a copy of each atom in the atom set. 10. The method of claim 8, wherein requesting a copy of the atom comprises asynchronously transmitting the request. 11. The method of claim 8, wherein replicating the atom comprises: In response to the request for the copy of the atom, a serialization message encoding the contents of the copy of the atom is received. 12. The method of claim 11, wherein replicating the atom comprises: The serialized message is decoded to produce the copy of the atom. 13. The method of claim 8, further comprising: Measure the delay from the first transaction node to at least one other transaction node among the plurality of transaction nodes. 14. The method of claim 13, wherein measuring the latency includes verifying the at least one other transaction node. 15. A method for adding a first node to a distributed database system, the distributed database system comprising a plurality of nodes operatively coupled to each other via a communication network, the plurality of nodes storing a plurality of atoms, each of the plurality of atoms containing a specified fragment of data or metadata, the method comprising: Establish a communication connection between the first node and a selected node among the plurality of nodes; Send a request to connect to the distributed database system from the first node to the selected node; Update the directory atoms in the selected node to indicate that the first node is the active node among the plurality of nodes, wherein the directory atoms track the position of each copy of each atom among the plurality of atoms in the plurality of nodes; and Transfer a copy of the directory atom from the selected node to the first node. 16. The method of claim 15, further comprising: A message is sent to the first node, the message including an identification of the selected node. 17. The method of claim 15, further comprising: Broadcast a message to the other nodes among the plurality of nodes, the message indicating the availability of the first node to the other nodes among the plurality of nodes. 18. The method of claim 15, further comprising: In response to a user query, the first node requests a copy of one of the plurality of atoms. 19. The method of claim 18, further comprising: In response to a request for a copy of the atom, the first node receives a serialized message encoding the contents of the copy of the atom. 20. A distributed database system, comprising: At least one archive node, the at least one archive node being used to store multiple atoms, each of the multiple atoms containing a specified fragment of data or metadata stored in the distributed database system; Multiple transaction nodes, communicatively coupled to the at least one archive node, each of the multiple transaction nodes comprising: A memory for storing a subset of the plurality of atoms; An interface for receiving a first request for an atom that does not exist in the subset of the plurality of atoms, the atom being stored in the archive node; and A central processing system, operatively coupled to the interface and the memory, is configured to transmit a second request for the atom to the archive node, receive a copy of the atom from the archive node, and store another atom in the memory. 21. The distributed database system of claim 20, wherein each of the plurality of transaction nodes is configured to measure latency relative to each other of the plurality of transaction nodes. 22. The distributed database system of claim 21, wherein each of the plurality of transaction nodes is configured to measure the latency by checking each of the other transaction nodes among the plurality of transaction nodes. 23. The distributed database system of claim 20, wherein the plurality of atoms includes directory atoms that track the location of all atoms in the plurality of atoms and the location of each of the plurality of transaction nodes, each of the transaction nodes having a copy of each of the plurality of atoms. 24. The distributed database system of claim 20, wherein the interface is configured to receive a serialized message encoding the contents of the copy of the atom in response to the request for the copy of the atom. 25. The distributed database system of claim 24, wherein the central processing system is further configured to decode the serialized message to produce a copy of the atom. 26. The distributed database system of claim 20, wherein the interface is configured to asynchronously transmit the second request. 27. The distributed database system of claim 20, wherein the archive node is configured to transfer a copy of the atom to each of the plurality of transaction nodes. 28. A method for obtaining data stored in a distributed database system, the distributed database system comprising a plurality of transaction nodes and at least one archive node, the plurality of transaction nodes and the at least one archive node being operatively coupled to each other via a communication network, the method comprising: A request for an atom is received at a first transaction node among the plurality of transaction nodes. The first transaction node has a first memory for storing a first set of atoms, each atom in the first set of atoms containing a specified fragment of data or metadata. It is determined that the atom does not exist in the first set of atoms; It is determined that the atom does not exist in other transaction nodes among the plurality of transaction nodes; A copy of the atom is requested from the at least one archive node, and the at least one archive node stores a copy of each atom in the distributed database system; In response to a message from the archive node, the atoms in the first set of atoms are copied at the first transaction node. 29. The method of claim 28, wherein the set of atoms comprises catalog atoms, and the method further comprises: The directory atoms are used to track the location of each atom in the distributed database system and each of the plurality of transaction nodes, each of which has a copy of each atom in the distributed database system. 30. The method of claim 28, wherein requesting a copy of the atom includes asynchronously transmitting the request. 31. The method of claim 28, wherein replicating the atom comprises: In response to the request for the copy of the atom, a serialization message encoding the contents of the copy of the atom is received. 32. The method of claim 31, wherein replicating the atom comprises: The serialized message is decoded to produce the copy of the atom. 33. The method of claim 28, further comprising: Measure the delay from the first transaction node to at least one other transaction node among the plurality of transaction nodes. 34. The method of claim 33, wherein measuring the latency includes verifying the at least one other transaction node. 35. The method of claim 34, further comprising: At least one communication path is selected based on the measured delay.