HK1228057A1 - Database management system - Google Patents

Abstract

The invention relates to database management system. Particularly, the invention relates to a multi-user, elastic, on-demand, distributed relational database management system. The database is fragmented into distributed objects called atoms. Any change to a copy of an atom at one location is replicated to all other locations containing a copy of that atom. Transactional managers operate to satisfy the properties of atomicity, consistency, isolation and durability.

Description

Database management system

Description of the cases

The application belongs to divisional application of Chinese patent application 201180014604.6 with the application date of 2011, 3, month 18.

Technical Field

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

Background

The use of databases for storing and retrieving messages has emerged as an important tool in a large number of commercial applications over the past years. Initially, many database systems operated on a single server device with multiple users. However, over the past years, various factors have arisen that require the basic characteristics of the database architecture to change. As a first factor, database storage requirements have become extremely large. Second, the number of users attempting to access such databases has also grown. Third, the use of databases for retrieving relatively stable data with minimal updates has been replaced with transactions.

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

The database is now judged by criteria that define the ACID properties, namely atomicity, consistency, isolation, and persistence. Atomicity ensures that all transactional 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 "view" the data in the intermediate state during the transaction. Persistence guarantees that once a transaction has been successfully processed, it cannot be recovered.

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. The locks may be exclusive or write locks or non-exclusive or read locks, and may be applied to individual records or to pages. However, when databases have increased in size and when transaction rates have increased, the overhead for managing locks has become significant, and in some cases prohibitive.

Multi-version concurrency control (MVCC) is an alternative process for ensuring simultaneity. For complex databases, MVCC may be more efficient than locking. MVCC uses timestamps or added transaction Identifications (IDs) to serialize different versions of records. Each version allows the 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, for example, to the record until the change is committed. MVCC also eliminates locks with other additional overhead and establishes a system where read operations cannot block write operations.

In addition to meeting the ACID test, there is now a requirement 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 increase in size and complexity, existing data processing systems are being replaced with more powerful data processing systems. Another approach for accommodating growth involves replication systems in which one machine is designated as the "head" machine, which keeps all of the replication machines synchronized. If the head machine fails, the process assigns that function to another replication machine. Different replication machines may be available to a particular user. This approach is not scalable because all machines must have the same capabilities.

As another approach, multiple autonomous database systems may be integrated into a single "federated" database having a network of computers that interconnect the individual databases. Federated databases require "middleware" to keep the constituent databases in sync. This "middleware" can become very complex. As the database size increases, the resources required to operate the middleware may impose a large overhead sufficient to degrade overall system performance.

"partitioning" is another means for implementing a database in which a logical database, or its constituent elements, are divided into distinct, independent portions. In a distributed database management system, each partition may be spread across multiple nodes. A user at a given node may perform a local transaction for the partition. Partitioning can also be achieved by forming a smaller database or by splitting selected elements of only one table.

There are two general approaches to partitioning. In a horizontal partition, also referred to as a "shard," different rows are placed in different tables and different servers. Generally, they have a certain commonality, such as a range of zip codes or last names divided into different tables by range. For example, a first database might include all records for last names in the range A through M; the second database is for data in the range N to Z. Fragmentation, which is one form of horizontal partitioning, involves locating rows of a database on separate servers. Sharding does not count the number of rows in each table and improves search performance. However, fragmentation uses hash codes at the application level, which makes it too difficult to implement. It also comprises a two-stage delivery. The complexity of a shard makes it suitable for a particular application, as the basis for defining a shard is well defined.

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

In yet another approach, known as a "shared-nothing" architecture, each node is independent and self-contained. The shared-nothing architecture is popular for web development because it can be expanded upward simply by adding nodes in the form of inexpensive computers. This approach is popular in data warehouse applications where updates tend to occur less frequently than transactions. However, the process of joining is complex on large data sets from different partitions or machines.

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

In a "full share" system, ultra-high speed communications keep the system synchronized. However, lock management may require a significant amount of bandwidth resources. To avoid this, such systems contain 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, satisfies the ACID properties atomicity, consistency, isolation, and durability. What is also needed is a database system that operates over the internet without the need for dedicated high-speed communication paths, that provides for transactions, and that is operable over large geographic areas.

Disclosure of Invention

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

It is a further object of this invention to provide a flexible, scalable, on-demand, distributed data processing system that is fault tolerant.

It is a further object of this invention to provide a flexible, scalable, on-demand, distributed data processing system that has a high degree of usability.

It is a further object of this invention to provide a flexible, scalable, on-demand, distributed data processing system that is platform independent.

It is a further object of this invention to provide a flexible, scalable, on-demand, distributed data processing system that is atomic, consistent, isolated, and durable.

It is a further object of this 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 this invention to provide a flexible, scalable, on-demand, distributed data processing system that provides transactions and is suitable for implementation over large geographic areas.

According to one aspect of the present invention, a database management system that enables a user to interact with a database of data and metadata includes a plurality of nodes and persistent storage having 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 that control a sequence of operations for interacting with the database, wherein, in response to a particular system-level command, an atomic object generates atoms, each atom containing data or a specified piece of metadata, whereby all sets of instances of an atom collectively define all of the metadata and data in the database. Each node additionally comprises: a communication control for establishing a communication path with each node in the system; a method of requesting a copy of an atom from a selected node that is relevant to a query but does not exist in that node in response to a system command from the interface; a method of replicating a requested atom for transmission to a requesting node in response to a request for the atom from another node, whereby only the atoms required to complete a query need be located in any transactional node at any given time; and, a method of replicating that atom for transmission to each other node in the system in which that atom resides, in response to changes in the atom at that node. The persistent memory contains a collection of atoms for all of the data and metadata collectively contained in the database.

According to another aspect of the invention, a database management system that enables a user to interact with a database comprised of data and metadata, the system comprising at least one transactional node that provides a user with access to the database and at least one archival node that maintains an archival of the entire database. Each transactional 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 that control a sequence of operations for interacting with the database, an atomic object generating an atom in response to a particular system-level command. Each atom contains a designated piece of data or metadata, whereby all instances of an atom collectively define all of the 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. The method in each transactional node requests a copy of the atom relevant to query building but not present in that node in response to a system command from the database request engine. Another approach in each node replicates the requested atom for transmission to the requesting node in response to a request for the atom from another node, whereby only the atoms needed to complete the query command need be located in any transactional node at any given time. Another method in each transactional node is to replicate a change in the atom at that node to each other node in the system that contains a copy of that node in response to that change.

According to yet another aspect of the invention, a database management system for a logical database, the logical database being comprised of data records organized as tables to be accessed from a plurality of transactional nodes that process transactions related to the logical database, wherein the database is parsed into segments, wherein each segment stores a portion of the metadata and/or data related to the logical database for transmission as serial messages in the database management system and as non-serial messages. The system includes at least one archive node that stores all of the fragments in non-serial form in persistent storage, thereby forming a single repository for the entire database. Each transactional node responds to a query from a user by establishing a sequence of low-level commands identifying segments relevant to the query, and responds to the low-level commands by obtaining those copies of existing segments relevant to the query being processed thereat, whereby a given segment may be located at some other node or only at an archival node. Each transactional node replicates that segment to each transactional node where a copy of the at least one archive node and any changed segments reside, whereby changes are made to segments in other nodes on a peer-to-peer basis, and whereby any transactional node contains only those segments relevant to queries made by users accessing the database through that transactional node.

Drawings

The appended claims particularly point out and distinctly claim the subject matter of this invention. The various objects, advantages and novel features of this invention will be more fully apparent from a reading of the following detailed description in conjunction with the accompanying drawings in which like reference numerals refer to like parts, and in which:

FIG. 1 is a diagram in schematic form of one embodiment of a resilient, scalable, on-demand, distributed data processing system containing the present invention with interconnected transaction and archive nodes;

FIG. 2 depicts an organization of a transaction node;

FIG. 3 depicts an organization of archival nodes;

FIGS. 4A and 4B depict the logical organization of "atomic" objects generated from the atomic classes shown in FIGS. 2 and 3, which are useful for implementing the present invention and which may appear in a transactional node at any given time;

FIG. 5 depicts information in a home directory atom;

FIG. 6 depicts information in a transaction manager atom;

FIG. 7 depicts information in a database atom;

FIG. 8 depicts information in a schema atom;

FIG. 9 depicts information in a table atom;

FIG. 10 depicts information in a table directory atom;

FIG. 11 depicts information in an index atom;

FIG. 12 depicts information in a record status atom;

FIG. 13 depicts information in a data atom;

FIG. 14 depicts information in a big binary object (Blob) state atom;

FIG. 15 depicts information in a blob atom;

FIG. 16 depicts syntax of an exemplary asynchronous message transmitted between transactional and archival nodes of the database system of FIG. 1;

FIG. 17 depicts various message types by which information is communicated between transactional and archival nodes of the database system of FIG. 1;

FIG. 18 is a flow chart useful for understanding a method by which nodes join the database system in FIG. 1;

FIG. 19 depicts information in a node object;

FIG. 20 is a flow diagram useful for understanding a method for creating atoms through its nodes in accordance with the present invention;

FIG. 21 is a flow chart useful for understanding a method by which unique atomic identifications are assigned during the method of FIG. 20;

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

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

Detailed Description

FIG. 1 depicts one embodiment of a flexible, scalable, on-demand, distributed database system 30 having a plurality of data processing nodes that incorporate the present invention. Nodes N1 through N6 are "transactional nodes" that provide user application access to the database; nodes A1 and A2 are "archive nodes" that are used to maintain disk files for the entire database at each archive node. While archive nodes typically store the entire database, a single transactional node contains only that portion of the database needed to determine the transactions that are to be supported at that node at that time.

Each node in fig. 1 may communicate directly with other nodes in the system through database system network 31. For example, node N1 may establish a communication path with each of nodes N2 through N6, a1, and a 2. Communication between any two nodes is by means of serialized messages. In a preferred embodiment, messaging is performed in an asynchronous manner to maximize the bandwidth used by the system, thereby performing various operations in a timely and expeditious manner. Typically, 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 network). Each node has the capability to restrict the use of low-latency paths to time-critical communications (e.g., acquire atoms). High latency paths may be used for non-critical communications (e.g., information requesting an update to a table). Also and preferably, the data processing network of the present invention incorporates a messaging 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 it by other nodes.

Fig. 2 depicts a representative transactional node 32, the representative transactional node 32 linked to a database system network 31 and various end users 33. The transactional node 32 includes a central processing system (CP)34, the CP 34 communicating with the database system network 31 through a network interface 35 and communicating with individual users through a user network interface 37. The central processing system 34 also interacts with a RAM memory 38, the RAM memory 38 containing a copy of the data stream management program implementing the preferred embodiment of the present invention. The 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 transactional node and is the interface between high-level input and output commands at the user level and system-level input and output commands at the system level. Generally, 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 class/object set 42 is divided into a subset 43 of "atomic classes," a subset 44 of "message classes," and a subset 45 of "helper classes. Additional details of these classes are described below.

As will become apparent and in accordance with the present invention, at any given time, the transactional node contains only those portions of the overall database that are relevant to the active user application. Moreover, the various features of the present invention enable all portions of the database in use to reside in the random access memory 38. Supplemental memory, such as disk memory, need not be provided at the transactional node during operation of this system.

Referring to FIG. 3, each archival node 50, such as archival node A1 or A2 in FIG. 1, is also connected to the database system network 31. However, instead of the end user 33 associated with the transactional node 32 in FIG. 2, the archival node is only connected to a persistent storage 51, which persistent storage 51 is typically a disk-based storage system or a key-value repository. Archival node 50 includes a central processing system 54, the central processing system 54 communicates with persistent storage 51 through an I/O channel 52 and with database system network 31 through a network interface 55. The central processing system 54 also interacts with a RAM memory 57, the RAM memory 57 containing a set 62 of classes or objects. Similar to the transactional node 32 in FIG. 2, the class/object set 62 in FIG. 3 includes a set 63 of "atom 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 classes and subclasses, such as those shown in fig. 2 and 3, define methods, data structures, and processes by which "instances" or objects of that class or subclass can be generated, as is known in the art. "instances" can be generated using "inheritance" and/or "polymorphic transformations". It will become apparent to those skilled in the art that implementations are possible that do not use object-oriented programming or variations in the specifically disclosed embodiments.

The invention will now be described in several stages. The "atom" portion defines the level and function of the object produced by the atom classes 43 and 63 in FIGS. 2 and 3, respectively. The "messages" section describes a set of messages that provide communication between transaction and archival nodes that may be generated by the message classes 44 and 64 in fig. 2 and 3, respectively. The "methods" section describes basic operations with respect to database management. The "examples" section describes the interaction of atoms, messages and methods, whereby the objectives of the present invention are achieved in response to a particular database query to the database request engine 41.

Atom(s)

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

Atoms have specific characteristics. In a transactional node, an atom exists only in non-persistent memory and is in the form of a non-serialized message that has been populated with a particular atom type to provide an efficient memory resident format for the atom. Each atom has means for encoding its content as a serial message and means for decoding the serial message to retrieve the atom's content. Such a serial message is used in connection with a number of operations as will be described later.

Each serialized message sent from a node for replicating an atom includes the contents of that atom with the appended node identification and the latest transaction commit sequence number for that node. When the archival node receives that serial message, it de-serializes the message (i.e., atom) before placing the rest of the message in persistent 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 identification to provide a reliable identification of that atom anywhere in the database processing network 30 in fig. 1. Second, any atom must exist in both nodes simultaneously to maintain redundancy, except that a single atom may exist after the single atom is created and before the archive node has requested a copy. Third, the transactional node will only install atoms as needed. Fourth, each time a change is made to an atom in a node, that node must replicate that changed atom on a "peer-to-peer" basis; that is, that changed atom is replicated to all archive nodes and those transactional nodes that contain only that same atom.

The "garbage collection" process described in more detail below can occur on both the archive and transactional nodes. The process removes inactive atoms from the transaction and archival nodes, and as a result, the transaction node can store those atoms currently associated with the user application in random access memory at that node. Thus, the database request engine 41 "sees" the entire database as being local and unaware that it is operating in a multi-node environment and that there is no complete copy of the database at its nodes. The archive node has the option to clear the contents of the atom after it has been serialized to disk, thereby reducing the size of memory required for storage. If an archive node receives a copy message from such an atom, the archive node must retrieve the content from disk storage and then apply the copied information from the atom.

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

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

Referring to FIG. 4A, master catalog atom 70 tracks the state of transactional and archival nodes in database system 30 of FIG. 1. It can also be thought of 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 table catalog atoms 75, and the sequence ID manager 82.

The table catalog atom 75 acts as an active index and creates and monitors an index atom 76, a record state atom 77, a data atom 78, a blob state atom 80, and a blob atom 81 associated with a single table. That is, each table has a table catalog atom 75.

FIG. 4B is useful for understanding the interaction and management of different atom types. In this context, neither master catalog atom 70 nor table catalog atom 75 perform any administrative functions. The master catalog atom 70 manages each schema atom 73 with respect to the remaining atoms. Each schema atom 73 manages each associated table atom 74 and sequence ID manager 82. Each table atom 74 manages its corresponding table catalog atom 75, index atom 76, record status atom 77, data atom 78, blob status atom 80, and blob atom 81.

Still referring to FIG. 4B, the database request engine 41 communicates with a master catalog atom 70, a transaction manager atom 71, a database atom 72, each schema atom 73, each table atom 74, and a sequence ID manager 82. The database request engine 41 acts as a compiler for high level languages such as SQL. As a compiler, it parses, compiles, and optimizes queries and obtains metadata and data from the atoms to form various pieces of database information.

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

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

Each time any transactional node changes a copy of an atom, 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 by other transactional nodes. Element 70F is a list of all nodes for which the provisioning node must relay messages to the requesting node from all other nodes containing that atom until the request is completed.

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

Still referring to FIG. 5, a 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 of characters. Local node IDS entries 70K include a range of numbers up to the total number of nodes that can be attached to the system. Together, these entries provide correspondence between the two types of identification. The use of local node IDS improves efficiency.

When a transactional node joins database system 30 in fig. 1, connection manager 70L carries out that process. A particular implementation for enabling a transactional node to join a 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 the responding node until it receives further communications from that responding node. Database UUI entry 70N contains a universally unique identification of the database.

The entries at 70P are important because they link all atoms for which the master catalog 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 catalog atom 75.

Password entry 70Q represents a means for authenticating a connection into a database. The actual and software version entries 70R and 70S allow the system to operate with backward compatibility when newer versions of software are installed. The actual software entry 70R identifies the version of software that is in use at the time; software version entry 70S, a 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 fig. 5, the master catalog 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. ping threads operate periodically and independently of other operations at the node. It "pings" each node to provide information that may be used in the determination of the communication efficiency of the corresponding path. For example, if node N1 in fig. 1 has the option of communicating with node N2 or N5, node N1 may use the ping information in the most efficient selection of communication paths to nodes N2 and N5 for that communication. Other selection processes may be substituted or added.

Referring to FIG. 6, there is one transaction manager atom 71 for each database and it is created during the same process that creates the master catalog 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 catalog atom. However, element 71A is an identification of the transaction manager atom 71. Elements 71B and 71C both point to master catalog atom 70.

The ID manager 71J provides a unique transaction sequence identification and maintains an active transaction list 71K, a committed transaction list 71L, and a failed transaction list 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 atom assigns a commit sequence number to element 71N when the transaction commits. The sequence numbers are sequential and each is specific to the node requesting the transaction. The transaction transition event counter 71P identifies discrete events that occur during each transaction, such as the beginning of a transaction and the successful commit of a transaction. Such counters are useful when multiple transactions involving the same information overlap.

Referring to FIG. 7, a database atom 72 is created at the same time that a master catalog atom 70 and a transaction manager atom 71 are created. The database atoms 72 identify each of the schema atoms 73. The database atom 72 may involve an authentication process when a new user attempts to join the database. It may also include other data regarding the authorization level.

In general, the database atoms 72 include elements 72A-72I corresponding to similar elements in FIG. 5. Element 72A is a database atom identification. Each of pointers 72B and 72C identifies a master catalog atom 70. The schema name-schema ID registry 72J correlates schema names with schema atom identifications.

Referring to FIG. 8, schema atom 73 creates and tracks the table atoms for that schema. The database atom 72 may manage multiple schema atoms, and each schema atom may interact with multiple form atoms. Schema atoms 73 include elements 73A-73I corresponding to elements 70A-70I in fig. 5. Element 73A is a unique schema atom identification 73A and elements 73B and 73C are pointers to the master catalog atom 70. Tables have unique names in the drawings. Table name-table atom ID registry 73J provides a correspondence between each table name and the corresponding table atom. Each sequence of drawings has a name. The sequence name-sequence ID manager registry 73K provides a relationship between those names and the corresponding sequence ID manager associated with each schema atom, such as the sequence ID manager 82 in fig. 4A and 4B.

FIG. 9 provides a logical view of the table atom 74 containing metadata relating to fields, formats, indices, and types and managing each of the index atom 76, record state atom 77, and blob state atom 80 of that table. It also creates and tracks data within the table. The table atoms 74 include elements 74A-74I corresponding to elements 70A-70I in FIG. 5. Element 74A includes a unique table atom identification, and both elements 74B and 74C point to the master catalog atom. Pointer 74J identifies the corresponding table directory atom. Element 74K contains a list of all fields of the table.

Each table atom has several ID managers. The pointer 74L points to an ID manager that provides a unique identification to each field. Pointers 74M, 74N, 74P, and 74Q identify separate ID managers for assigning identifications to index atoms, data atoms, blob atoms, and subtypes, respectively. Element 74R is a list of existing subtypes. Arrays 74S and 74T provide locations for recording the status atoms and the status of the blob, respectively.

Referring now to FIG. 10, there is one table catalog atom for each table atom. Each table catalog atom 75 is created when a table atom is created. In turn, the table catalog atom creates and tracks atoms specific to one table that includes an index, record state, data, blob state, and blob atom. Each table catalog atom 75 includes elements 75A-75I corresponding to elements 70A-70I in FIG. 6. Element 75A is a unique table directory atom identification as assigned by the master directory atom. Two elements 75B and 75C point to the master catalog atom 70. ID manager 75J provides a unique atom identification for each of the index, record status, data, blob status, and blob atoms. List 75K identifies all atoms associated with the corresponding table atom. The 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. Bitmap 75N provides a convenient means for identifying other objects and directories when an atom is in an archive node.

Referring to FIG. 11, there is one index atom 76 per index in a table, and there may be multiple index atoms per table. Each index atom includes elements 76A-76I corresponding to elements 70A-70I, respectively, in FIG. 5. Element 76A is a unique index atom identification as assigned by the corresponding table directory atom. Pointers 76B and 76C identify the master catalog atom and the table catalog atom, respectively. Element 76J contains a binary tree of index nodes to provide conventional indexing functionality. Element 76K contains the index level. Such indexing structures and operations are known to those skilled in the art.

Returning to fig. 12, the record status atom 77 manages record versions and statuses for a fixed range of record numbers within a single table. Thus, a given table atom can manage multiple record state atoms. Each record status atom includes elements 77A-77I corresponding to elements 70A-70I in FIG. 5. Element 77A includes a record state atom ID that creates a table directory atom assignment. Pointers 77B and 77C identify the master catalog and table catalog 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 the database application to which the present invention is directed, multiple users may generate multiple versions of the same record. The preferred embodiment of the present invention uses multi-version concurrency control (MVCC) to ensure that transactions never have to wait for a database by: allowing several versions of a record or other object to exist in the database at the same time. As a result, each record status atom 77 includes metadata about each record version. Entry 77M is a bitmap identifying the location of each versioned record 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. Format version entry 77Q identifies the version number of the table subtype that exists when the record is inserted. This format identifies the physical order of the records and identifies the subtype of the database program to which the record belongs that was in use when the record version was created. Element 77R includes a record version sequence number; element 77S, the location of the next older or previous version of the record. The index 77T to the data atom array 77J and the base record identification 77K together provide the address of the actual time slot 77U in the data atom with the record version.

FIG. 13 depicts a data atom 78 having elements 78A-78I corresponding to elements 70A-70I in FIG. 5. In the data atom 78, the element 78A is a data atom identification 78A assigned by a table catalog atom. Elements 78B and 78C are pointers to the master catalog atom and to the corresponding table catalog atom, respectively. The ID manager 78J assigns a record slot identification to each record in the data atom 78. Element 78K identifies, for each record in data atom 78, the address and length of that record. Element 78C represents the data record and its version.

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

FIG. 15 depicts a blob atom 81 having elements 81A-81I corresponding, respectively, to elements 70A-70I in FIG. 5. Element 81A is an atom identification assigned by a table directory atom. Elements 81B and 81C are pointers to the master catalog atom and the corresponding table catalog atom, respectively. ID manager 81J assigns a blob slot identification to each blob in blob atom 81. Element 81K identifies the address and length of each blob in blob atom 78. Element 81L represents all blob records assigned to the blob atom.

In summary, each atom has a relationship only to a fragment of the database. For example, the database atom 72 includes metadata that identifies a schema for the database. Each schema atom 73 contains metadata that identifies all tables associated with that schema. For each table, table atom 74 and corresponding table catalog atom 75 provide metadata about the table containing information such as identification of fields and their attributes. The record status atom includes metadata about a set of records. The data atom includes information about each data record having a pointer to the time slot containing the record and the respective version. The blob state and blob atom contain similar information about the blob record.

Message

As previously described, communication between any two nodes is by means of serialized messages sent asynchronously using TCP or another protocol with control for maintaining messaging sequences. Fig. 16 depicts the basic syntax of an exemplary 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 that different nodes may operate using different software versions, header 91 also includes an identification 94 of the software version that created the message. The remaining elements in the header include the sender's local identification 95 (i.e., from the master catalog atom in fig. 5) and the destination of the message, i.e., the catalog atom's information 96 (e.g., table catalog ID 75A in fig. 10) and atom identification 97 (e.g., record status ID 77A in fig. 12). From this information, the receiver node can de-serialize, decode, and process the message.

FIG. 17 depicts a message set having the syntax of FIG. 16 for a specific embodiment of the present invention. Each performing a specific function that will now be described.

As briefly mentioned above, when a message is to be sent, there are different communication paths to different nodes. For example, if one node that is the 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 best path correspondent node. As is known, pinging includes determining when a "ping" command reaches its destination and an acknowledgement message is received. In this embodiment of the invention, each node periodically uses a helper class to send ping messages 110 to each of the other nodes to which it is connected. Each receiving node uses the helper class to return a ping acknowledgement message 111 containing the ping time. Each node accumulates this information about the time at which the messages were sent and received in the node object described below with respect to fig. 8. When a node is ready to send a message to one of the 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 database system 30 in fig. 1 or connecting a previously inactive node back into database system 30 in fig. 1. When such a node, e.g., transactional node N2, wishes to connect to 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 the message, 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 sequence of messages is complete, the joining node may then perform further steps to obtain various atoms.

The archival nodes can operate in an active mode or a synchronization mode when they synchronize with another node. A helper class in an archive node sends a node status message 115 to alert all other nodes of any state changes in that archive node.

When a node that is a requesting node retrieves a copy of an atom from another node, the next set of messages is involved. For example, after a node joins the database system 30 in FIG. 1, it typically requests a copy of the master catalog atom. This process is described in more detail in connection with the explanation of fig. 20 and 21.

Still referring to FIG. 17, the requesting node issues 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 that have the atom. Each node that receives the object available message 118 from the selected node returns an object acknowledge message 119 to the selected node. After the selected node receives all object acknowledge messages 119, the selected node sends an object complete message 120 to the requesting node.

In some cases, the selected node sends an object unavailable message 121 to state that the selected node has discarded the requested atom. The bounce object message from the selected node indicates that the requested atom is not found in one of the master catalog atom or the table catalog 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 that atom before receiving the object request message. In response, the requesting transactional node can select another node in the database system with the atom.

The database request engine 41 in fig. 1 and fig. 4A and 4B may periodically generate a register object message 123 or a deregister object message 124. These messages are directed to atoms, such as database and schema atoms, that comprise a registry. 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 of 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 a change has been made, the change number will not be 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 may then update the state of the atom to note that the atom has been archived and is a potential candidate for garbage collection.

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

Referring to fig. 4A and 4B, the table atom 74 can send any of a number of messages. If the database request engine 41 initiates processing for adding a new field, a new structure is generated at the table atom of that corresponding transactional node and a table field addition message 129 is sent, which table field addition message 129 replicates the change to all other nodes including that table atom. If a transactional node updates a field, subtype, or field property that changes the table format, that node issues a table format message 130. Table request record message 131 is generated whenever a node needs to create a new record state or blob state atom. Only the chairman can create this atom; and, when this occurs, the chairman broadcasts the table record object message 132.

Each time a transactional node inserts a new record in the table, it generates a table record message 133. The create table atom duplicates a new table atom whenever it becomes necessary to create a new index, as when fields of an index are added to a table. At this time, the index is set to the write-only index. After all relevant processing involved has been completed, the node sends a table index addition message 134.

Any time a table atom creates a blob state atom for a blob, the table atom generates a table blob message 135. Table blob message 136 indicates that a new blob has been created.

In a database utilizing types and subtypes, the database request engine 41 in FIG. 1 will generate a command to cause a table atom to assign a new table type identification. When this occurs, the table type message 137 is replicated to all nodes with similar table atoms.

The table record erasure message 138 provides the record number for the particular record status atom. When it is determined that the records within that table contain an unusually long chain of inverted versions or other criteria, the chairman of the table atom generates a table garbage collection message 139. The result is that the "unused" atoms are "emptied" of data.

The record status atom 77 in fig. 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 fig. 17, the corresponding record state atom generates a record update request message 140, and the record update request message 140 is directed to the chairman of that atom for permission to update that particular record. The chairman responds by generating a record update response message 141, which record update response message 141 grants or denies permission to update that record. If the chairperson permits permission, the record state atom is requested to perform an update and a record update message 142 is sent to each node that has a copy of that atom with a new version of the atom.

Data atoms store up to some maximum number of records and versions thereof. The corresponding record state atom monitors the size of the data atom it manages. If the record status atom determines that the managed data atom has exceeded that size, it generates a new data atom and replicates the new data atom by means of the record data object message 143.

The record state atom generates a record message 144 for copying any new record versions. Periodically, the form atom chairman initiates a process by which the record state atom identifies a version of the record that is older than the oldest active transaction. When this occurs, the record status atom of the chairperson sends a record delete message 145, the record delete message 145 enabling the deletion of those older record versions during the subsequent garbage collection process. If a situation arises in which it becomes necessary to recover the transaction, the record state atom generates an undo record message 146 that updates the undo record.

There is also a set of index specific messages. As is known, the index has an optimal maximum size; and, if the index exceeds that size, the index should be partitioned. According to the invention, only the chairman can split the index atom. The chairman may do so unilaterally or in response to an index split request message 147 from another copy of that atom. When the chairman causes the segmentation, the chairman generates an index segmentation message 148 that contains the segmentation index. After the split, the chairman sends an index delete message 149 to puncture the original index atom after the split. Each time an inode is added to an index, it generates an inode addition message containing the index key, record identification, and other information for the new index. When the index has been completely filled and is therefore ready for use during a retrieval operation, a table index ready message 151 is generated. As previously indicated, adding a table index generates an index atom that is write-only and unreadable. The table index ready message 151 makes such write-only index readable.

The blob state atom identifies the slot in the blob atom by generating the blob message 152.

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

When the state of a transaction changes, the transaction manager atom 71 for a given transactional node generates a transaction state message 154. These indicate whether the current state of the transaction is in the active state, the pre-commit state, the commit state, or the recovery state. In certain situations, 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 transaction blocking message 155. If the transaction is pinned and must be recovered, the transaction manager atom associated with the node causing the pinning generates a transaction pinning message 156 that causes the transaction to recover.

The master directory or any table directory atom on the archive node may request the time of the last write atom. This occurs when the requesting node sends a request write time message 158. The recipient of that message then returns the time of those requests 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 may be reasonably short. When used with TCP or other messaging protocols, the messages must be sent in order and, at any given node, they must be processed in the same order in which they were even sent by different nodes when received.

Method of producing a composite material

It will be useful to further understand the present invention to describe some basic methods related to various aspects of its operation. Variations on each will be apparent to those of ordinary skill in the art.

Fig. 18 is a flow chart of operations occurring in different nodes in database system 30 of fig. 1 when a node joins the network. For the purposes of this description, assume that node 5 in FIG. 1 is to join a database system; it is designated as the joining transactional node 170. As a first process, the joining transactional node 170 uses step 171 to establish a TPC connection with the selected node. Basically, the joining transactional node 170 sends a message to a fixed location that identifies the database. A connection broker, not shown but known in the art, responds to this request by denying or granting access to the database system. If the connection broker grants access, it selects a node, such as transactional node N1, as the selected node 172. The connection broker then sends a message to the joining transactional node 170 with the designation of the selected node 172 as a port number. Fig. 18 also represents all other active transaction and archive nodes in the database system as a group 173.

Once the connection is established in step 171, the joining transactional node 170 sends a connect message to the selected node 172 using step 174. The selected node 172 responds to the connect message at step 175 by: update its master catalog atom with its connection manager, assign a local node ID to the joining transactional node 170, and add that node to the array of local nodes in its node object, an example of which is shown in FIG. 19.

FIG. 19 depicts a node object such as described in conjunction with step 175 of FIG. 18. It contains a pointer 400A to the socket for the node and a pointer 400B to the master directory atom at the node and a pointer 400C to the connection manager in the master 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 the message, determines the message type, and then processes the message to completion.

The node object 400 includes, as all atoms, the global node ID 400F and the local node ID 400G of the node to which this node is listening. Element 400H is a pointer to a queue waiting for a message to be sent from a node. Elements 400I and 400J contain the identification of local and remote ports. Element 400K contains the version number for the software operating at the remote node. The node type element 400L indicates whether the remote node is a transactional node, an archival node for synchronization, or an archival node that is online.

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 of the last ping operation and the ping time. As indicated previously, each node generates a sequential commit 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.

Referring back to FIG. 18, at step 176, the selected node 172 sends a welcome message to the joining transactional 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 archival nodes in step 177.

Each node in group 173 responds to the new node message at step 180 by: the global ID is recorded and the local IDs of the joining archival nodes are assigned and the local list of all connected nodes in their respective master catalog atom is updated. Each then sends a welcome message to the joining transactional node 170 using step 181. Upon completion of this process, the joining transactional node 170 has a complete list of all active nodes including the selected node 172 and all nodes in the group 173.

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

Rather than waiting for the receipt of the object message, the joining transactional node may also begin the process of step 185 for retrieving copies of the database atom and the transaction manager atom from the selected node 172.

Each of the other nodes that receive the object available message responds in step 186 by sending an object acknowledge message back to the selected node.

The database in the joining node requests the engine to initiate this sequence. Thus, when the method of FIG. 18 has been completed, the joining transactional node 170 connects to the database system and, as shown in FIGS. 4A and 4B, includes copies of the master catalog atom 70, the transaction manager atom 71, and the database atom 72. The joining transactional node 170 will then create or obtain copies of other atoms on an as needed basis.

During some operations, a transactional node may create a new table, necessitating the creation of a new table atom. FIG. 20 discloses a process 190 in which node A is the requesting node and atom X is the atom that requests and will manage new atom Y. To create a new table atom Y, atom X will be a schema atom and local catalog Z will be a master catalog atom. Step 191 represents the prepare 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 typically "0".

At step 192, local directory Z creates an instance of atom Y without content and designates the local node as the chairman of atom Y. The process 193 then allows the 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. 21.

Next, the local directory Z sets an exclusive lock to allow changes to it to occur without any impact from external automation. When the lock is in place, the local directory Z sets the state of the new atom Y to a dirty state and an unopened state. The dirty status indicates that the new atom Y has not yet been copied to the archive. The "not open" state indicates that a new atom Y is not yet available for 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 transactional and archival nodes.

At step 197, atom X, which is a management atom for new atom Y, populates the instance of atom Y and sets the status of atom Y to "open" to indicate that the new atom may be copied. Sometime thereafter, the archive node will store a copy of the new atom to persistent storage. That is, the archive node will respond to receipt of the object available message by: a copy of atom Y is requested, thereby providing redundancy. When this is done, the receipt of an object acknowledgement message from the archive node will cause the "dirty" state to change, and this change is then reflected to all other nodes that have copies of atom Y.

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

As will be recalled, when at least one other node exists for a particular atom, each atom contains a list of nodes that include copies of the atom being requested. If this is the first request for an atom after that atom has been created according to the process of FIG. 20, then the corresponding directory in the first node on the list is the chairman. Step 203 represents the process of selecting a node and identifying a 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 chairman receives the ID request message in step 205, it obtains a block of available ID numbers (step 206). The chairman identifies the transactional node making the request as the location of this block of available ID numbers (step 207). The chairman then sends an ID delegation message to the requesting node at step 210. When the requesting node receives an ID delegation message from the chairman in step 211, it stores a block of ID numbers in step 212, and then selects the first available ID for assignment in step 201. Referring back to FIG. 20, in this case, process 193 transitions from step 200 through step 202 directly to step 201 because the chairman is specified in step 192 of FIG. 20.

The response 220 in fig. 22 is processed each time a request for an atom is made by the database engine 41 in fig. 4A and 4B. For example, assume that the database request engine 41 in requesting node 221 requests a copy of atom Y (e.g., a table atom) from a selected node 222 in a database with other nodes 223. When the database request engine 41 makes that requirement, step 224 determines if atom Y is present in the requesting node 221. If so, step 225 ends the process because there are atoms requested. If not, control passes to step 226, whereupon local directory Z creates an empty instance of atom Y and declares atom Y as "unfilled". If database engine 41 is requesting a table atom 74, then master catalog atom 70 will perform this step. Step 230 then identifies the selected node 222 using the selection process described previously. The preference is to select any of the transactional nodes before selecting the most responsive archival 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 the 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 with 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 duplicate message for that atom to the requesting node. Thus, 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 an accessibility analysis of the message at step 233. If the message contains the current atom, atom Y processes the message in step 236 and sends an object acknowledgement message to the selected node in step 237.

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

It is possible that another node sends a duplicate message to the requesting node 221 during the time between steps 226 and 236 in figure 22. This may 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 in the pending message list as part of each atom, although not shown in any of fig. 5-13.

As indicated previously, the present invention is particularly suited to databases that interface with transaction processing techniques. In this connection, appropriate means for "submitting" the message should be included. FIG. 23 describes one means of ensuring consistency of data in such an environment. Specifically, fig. 23 depicts four nodes, namely: transactional node a 250, transactional node B251, transactional node C252, and archival node 253. Assume that transactional node B251 sends a pre-commit message with the transaction ID supplied by the transaction manager atom at that node at step 254. The message is routed to the archival node at step 255. When all conditions for committing that transaction have been completed, the archival node 255 issues a commit transaction message at step 256 and broadcasts that commit message at step 257. Each of the transactional nodes updates its corresponding transaction number in response at step 258.

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

Each transaction identifier is unique across the database system of fig. 1 with respect to the transaction ID and commit sequence number, and as previously described. The purpose of the transaction ID is a unique, permanent, system-wide flag that indicates which transaction created a particular record version. Each transaction ID is assigned by a local copy of the transaction manager atom 71 and one such transaction manager atom would be a chairman that assigns a block of identifier numbers to the transaction manager atom on each transactional node in the database system. The transaction manager atom at a given transactional node assigns, in order, to each transaction that receives a start, an unused number in the assigned block. As a result, the transaction ID of a newer transaction starting at a given node may be higher than the transaction ID assigned to an older transaction starting on that same node. However, on a system-wide basis, the 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 may read a record version only if the transaction that created the record version has committed before the read transaction started. In the present invention, such a method for determining which record version to read is the transaction commit time, not the start time, of the transaction used to create the record version. The transaction ID does not contain its commit time, so the transaction manager atom on each node assigns a commit sequence number to each transaction based on the actual commit operation. Each transactional node generates its commit sequence number in increasing sequence. If the transaction from a given node has a commit sequence number 467, then all transactions from nodes with lower commit sequence numbers are confident that they have been committed. All nodes generate the same commit sequence number, so interpreting this number requires the number and identification of the corresponding node.

When a transaction commits, its transaction manager atom sends a commit message to all nodes including the corresponding transaction ID and commit sequence number. When an atom serializes itself, it includes the highest commit sequence number that it has seen from each node that owns a copy of that atom. For example, assume that nodes A, B and C contain a copy 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 that 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 when node a sees that all nodes. It is possible that node B or node C may have actually issued a higher commit sequence number for transactions that node a has not yet received or processed.

A transaction manager on each node maintains a transaction object for each node in the database system. Each transaction object reflects the state of the transaction on all nodes and each object is assigned two or more numbers that increase locally and continuously. When the transaction manager atom starts a transaction or receives a transaction transition message indicating that another node has started a transaction, the transaction manager atom creates a local transaction object with a new transaction ID and assigns a start number to it. When a local transaction commits or when the transaction manager atom receives a transaction transition message indicating that a transaction on another node has been committed, a transaction end number is assigned to the transaction object along with the commit sequence number. The start and end numbers are from the same sequence. As an example, if transaction 123 has a start number that is higher than the end number of transaction 453, then transaction 123 may read the record version created by transaction 453 from any node that executed transaction 453.

The 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 ordering of transactions on other nodes. As will be apparent, there may be delays in receiving and in processing transaction status messages. Each transaction runs only on its local transactional node. This database system prevents a transactional node from "seeing" a transaction as it committed before it committed on its local transactional node. For example, if node A starts transaction 123 after receiving and processing a commit message from transaction node B for transaction 453, transaction 123 may read the changes from transaction 453 even though transaction node C received the transaction commits and starts the message in another order and sees the two transactions as contemporaneous.

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

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

Different transactional nodes act asynchronously when viewing transactions from the system level. The time that a single message is being processed by different nodes may vary due to differences in the efficiency of the communication path and the number of messages that each transactional node needs to process. Differences in processing time at different nodes must be considered. Consider three transactional nodes: node a, node B and node C. Assume node B executes a transaction 768 that 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 that some time has elapsed before node C processes those changes. Also assume that node B has assigned a commit sequence number 47 to transaction 768. The transaction manager atom on node B sends a transaction status message to all nodes in the database system. Also assume that node a requests a copy of atom X from node C after receiving the commit message from node B but before node C completes processing the transaction. Node a receives and processes the transaction status message from node B. From the perspective of node a, 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 for transmission to node A, it includes the highest commit sequence number from each node that has a copy of atom X. In this case, assume that the message will include a commit sequence number 97 for node C and a commit sequence number 46 for node B.

Referring to FIG. 22, node A, as the requesting node, creates an empty instance of atom X and then requests its contents from node C. The accessibility analysis in FIG. 22 obtains the current highest commit sequence number for each node as seen by node A. It also extends the serialized message with atom X to obtain the current highest commit 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, and its locally highest commit sequence number for B is 47. Thus, 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 an object complete message to it. If the commit sequence number is the same or greater, node A may continue to process atoms.

As previously described, when node C sends a serialized copy of atom X to node A, it sends an object available message for the other copies of atom X, in this case the copy on node B. Atom X on node B sends an object acknowledge message to atom X on node C and adds the copy on node a to its own copy list. Atom X on node B will then send a change message for its copy on both nodes a and C. At the same time, atom X on node C processes messages from node B, relaying them to node a. The atom X on node a processes the message so that it is itself more and more up to date. When atom X on node C processes object acknowledge messages from atom X on node B and all other copies of atom X on other nodes, it sends an object complete message for atom X on node a.

According to the present invention, each transactional node should operate without having to transfer any atoms to disk or other persistent storage. This requires that inactive atoms be removed from each transactional node in a expeditious manner. Without achieving this goal, helper classes in each node periodically provide garbage collection functions that are invoked by the loop manager to initiate new garbage collection loops independent of other operations. More specifically, the loop manager executes an aging loop whereby any atom referenced in the previous loop is moved to the front on the Least Recently Used (LRU) list in the master catalog atom for that node. The loop manager gets an exclusive lock to ensure that no other threads are active with respect to the previous loop.

The loop manager then starts a record garbage collection thread for the table atom that records the garbage collection flag. If the current amount of memory at the node exceeds the specified amount defined by the configuration object identified by pointer 70U at FIG. 5, garbage collection thread processing cycles through the main directory LRU from least recently used to most recently used atoms. If that atom is (1) inactive, (2) has been referenced in the current aging cycle, (3) is not an "object incomplete" atom, (4) has been locally changed and has not been written to an archival node, and (5) is a directory with no memory resident atoms, it classifies the atom as a candidate for a garbage collection candidate. The atoms in the archive node must otherwise not have been serialized to disk.

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

Two mechanisms involve this process. When the request catalog atom finds an atom for classification, the catalog 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 loop so that methods in those atoms can hold atom pointers without having to increment the atom usage count to prevent objects or atoms from being garbage collected.

Figures 19 and 20 depict "chairman identity". As shown in FIG. 20, when an atom is created, the created node is designated as the atom chairperson, and an ordered list of nodes for the atom is created. When a node creates an atom, it is the only entry in the ordered list. The first node in the list is the chairman.

When another node later requests a copy of that atom, the selected node places the identity of the requesting node just after itself in the ordered list, regardless of whether it is a chairman or not. As a result, if a node designated as the chairman for any atom becomes inactive for any reason, every other copy of that atom has an ordered list of nodes. With this ordered list, the next chairman is the first remaining transactional node on the list. If there are no transactional nodes on the list, a first asynchronous archive node is designated.

Discussing simple database queries may now be useful to help understand the interactions between different nodes and atoms within the nodes. Assume that a database request engine at a transactional node receives a database query to select all records in a table "member" with zip code "01944". Assuming that the transaction manager atom and the form atom are located at local nodes, all query processing has been completed and there is an index atom for the "zip code" field in the "members" form.

Initially, the database request engine 41 in FIG. 2 issues a command to the transaction manager to start a new transaction, at which point the transaction manager atom assigns a transaction ID. Next, the database request engine utilizes the form atom for the "Member" form to determine if an index atom exists for the "postal code" field. If the two atoms are at a local node, they are processed. If they are not at the local node, copies of these atoms are obtained from other nodes, and processing continues.

The table atom for the "member" table scans the index with information from the corresponding table catalog atom. This results in a bitmap that identifies each "record" in the "member" table with the assigned zip code. As will be apparent, this bitmap may be limited to a particular field or may be the result of a logical combination of indices 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 used in the record. The table atom selects the appropriate record state atom associated with the record atom by: the record number is divided 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 acquisition method in the selected record status atom. For multi-version records, the record state atom loops through any record version to find the correct version and the appropriate pointer to the corresponding data atom. The record state atom calls that data atom with a data atom number that points to the record and allows it to be retrieved. When this is done, the database request engine provides the user with a record set that lists all records having the specified zip code.

In summary, it will be apparent to those of ordinary skill in the art that a database management system constructed in accordance with the present invention provides a flexible, scalable, on-demand distributed data processing system. The system is fault tolerant and has a high degree of availability. It is platform independent and operates to provide a flexible, consistent, isolated and durable database. Moreover, it can operate over the internet without the need for high-speed communication paths and is suitable for transactions that can be implemented over a wide geographic area.

The present invention accomplishes all these objects by achieving 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 transactional and archival node determines which atoms are to reside on memory on a local basis. Catalog atoms track the location of local and remote copies of individual atoms and identify the catalog of which they are a member. Further, each node may determine an optima of the plurality of nodes from which to request a copy of the atom that enables the geographically dispersed system.

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

Claims (3)

1. A database management system for enabling a user to interact with a database of data and metadata, the system comprising:

A) a plurality of nodes, each node comprising:

i) an interface between high-level input and output commands at a user level and input and output commands at a system level, the input and output commands at the system level controlling a sequence of operations for interacting with the database, wherein, in response to a particular system-level command, an atomic object generates atoms, each atom containing data or a specified piece of metadata, whereby a set of all instances of an atom collectively define all of the metadata and data in the database;

ii) a communication control for establishing a communication path with each other node in the system;

iii) a method of requesting a copy of an atom from a selected node that is relevant to the query but not present in that node in response to a system command from the interface;

iv) a method of replicating a requested atom for transmission to a requesting node in response to a request for the atom from another node, whereby only the atoms required to complete a query need be located in any transactional node at any given time; and

v) a method of replicating an atom at the node for transfer to each other node in a system in which the atom resides in response to a change in the atom,

B) means connected to said plurality of nodes for providing a communication path between all of said nodes, an

C) Means connected to said communication means to provide persistent storage for information in the atoms, whereby the collection of atoms in persistent storage collectively contains all of said data and metadata in said database.

2. A database management system for enabling a user to interact with a database of data and metadata, the system comprising:

A) at least one transaction node providing a user with access to the database and at least one archive node maintaining an archive of the entire database, each transaction node including a database request engine providing an interface between high-level input and output query commands at a user level and input and output commands at a system level, the input and output commands at the system level controlling a sequence of operations for interacting with the database, wherein, in response to a particular system-level command, an atom object generates atoms, each atom containing a specified piece of data or metadata, whereby a set of all instances of an atom collectively defines all of the metadata and data in the database,

B) a database system network interconnecting all of said nodes;

C) a communication control in each of the nodes for establishing a communication path with each node in the system,

D) a method in each transactional node for requesting a copy of an atom related to the query command but not present in the node in response to a system command from the database request engine,

E) a method in each node for replicating a requested atom for transmission to a requesting node in response to a request for the atom from another node, whereby only atoms required to complete a query command need be located in any transactional node at any given time, and

F) in each transactional node, a method of replicating a change in an atom at the node to each other node in a system containing a copy of the node in response to the change.

3. A database management system for a logical database comprised of data records organized as tables to be accessed from a plurality of transactional nodes that process transactions related to the logical database, the system comprising:

A) means for parsing the database into segments, wherein each segment stores a portion of metadata and/or data related to the logical database for transmission as a serial message in the database management system and as a non-serial message,

B) at least one archival node storing all of the fragments in non-serial form in persistent storage, thereby constituting a single repository for the entire database, wherein each transactional node comprises:

i) means for responding to a query from a user by establishing a sequence of low-level commands for identifying segments relevant to the query,

ii) means for responding to said low-level command by obtaining only those copies of existing segments relevant to said query being processed at it, whereby a given segment may exist at some other node or only at an archival node, and

iii) means for replicating the segment to each transactional node where a copy of the at least one archive node and any changed segments reside, whereby changes are made to segments in other nodes on a peer-to-peer basis, and whereby any transactional node contains only those segments relevant to queries made by users accessing the database through that transactional node.