EA027808B1 - Database management system - Google Patents

Abstract

The present invention is related to a method and apparatus for implementation of a multi-user, distributed database management system that meets simultaneously the requirements of transactionality, high availability, fail safety and scalability. The invention comprises horizontal sectioning of data by means of distribution and persistent storage of database objects at several network nodes with the possibility of intersection of data subsets of any two nodes where none of the nodes is obliged to store the complete set of all database objects. Efficient horizontal scaling of the system is based on a method of issuing unique primary key values to database objects independently of other nodes.

Description

This invention relates to the field of database management systems. More specifically, this invention relates to a method and apparatus for implementing a multi-user distributed database management system that simultaneously satisfies the requirements of transactionality, high availability, fault tolerance, and scalability.

State of the art

Over the past years, database management systems (DBMS) have become well-established and widely used as part of applied software systems for many branches of human activity. Initially, database management systems were created with the expectation of serving many users, called clients, as one physical server. This is due to the relative simplicity with which a single-machine system can ensure database transactionality by satisfying four standard requirements for a transaction system: atomicity, consistency, isolation, and reliability of data operations (eng. ЛСГО - ЛЮписПу.

Atomicity ensures that the transaction is always complete or not complete at all. Consistency ensures that transactions only gain access to the final results of other transactions; intermediate data states due to transaction operation are not visible to other transactions. Isolation ensures that competitive transactions do not affect each other, and in the event that transactions intersect according to data, they are ordered by the time they work with the data so as not to violate the principles of atomicity and consistency. Reliability guarantees that a successfully completed transaction will not be already canceled, and its result will not be lost.

There are a number of approaches to creating a DBMS with LSGO transactions in a single-machine version.

Early approaches are based on the use of various means of blocking access to data: exclusive and non-exclusive locks, blocking of read operations, blocking of write operations, locks with multiple concurrent reading and exclusive writing, and locks can be applied to individual objects or records, as well as to a memory page or range of records.

Later approaches are based on concurrent access control using multi-versioning (eng. MUSS - MiSUegChop Sopsiggepsu Sop1to1), the essence of which is to use timestamps or increasing values of transaction identifiers associated with consecutive versions of each record. Each transaction is allowed to read the latest version from a variety of records for which the timestamp or identifier is less than the timestamp or identifier of the reading transaction. Thus, each client is provided with a logical snapshot of the database for the duration of each transaction, which has the property that the changes made by the client to the database are invisible to other clients until the transaction is committed. This control method is more effective than locks in terms of overhead, it allows you to ensure that reading transactions do not block writing, and writing transactions do not block reading.

Recently, due to the development of the global Internet, there has been a steady tendency to increase the number of clients served and the amount of data to limit values that a single server can handle. In addition, a new requirement has emerged for high availability databases. To cope with the growing demands, various vertical and horizontal scaling options are used. Vertical scaling involves increasing system performance by increasing the performance of components of one machine, for example, by increasing the number of processors, RAM, throughput of I / O controllers and the number of disk storage devices involved in parallel. Horizontal scaling involves increasing the performance of a multi-machine system by increasing the number of machines called nodes that perform the same function. Horizontal scaling is more convenient and efficient than vertical scaling, as well as more economically viable, allowing you to extensively increase system performance by adding cheap consumer-level nodes. However, horizontal scaling requires significantly more sophisticated software.

In the widely used relational DBMSs, the cluster approach is often used when the logical server consists of several nodes, each node has a processor and RAM, and all nodes share a common redundant array of independent disks connected to nodes by a high-performance I / O controller. With this approach, all nodes share common data in external memory through which the synchronization of node states occurs. Thanks to shared external memory in a cluster of nodes, a single time management and logical ordering of all transactions for all nodes is provided. The positive side of this approach is the provision of LSGO transactions, but it is economically costly and has a low scalability ceiling, especially in relation to the growth in the number of database clients.

Duplication of data (Eng. GerPSaBop) between several DBMS servers is another approach to scaling DBMSs. At the same time, one node is considered the main one and contains the main copy of the data, with which copies of data on other nodes are synchronized. When the main node fails, its function is transferred to another node. An example of this approach is a DBMS, consisting of a pair of nodes, where one node performs rarely occurring write transactions, and the other node performs frequently occurring read transactions. Duplication of data occurs in one direction, and nodes perform different functions. The obvious problems of this approach are the lack of versatility and scalability, as well as a violation of data consistency - the client continues to receive outdated data while reading until the updated data is replicated.

An approach based on the federalization of data looks economically feasible when a federated database is created as a mediator add-in that integrates several separate databases located on independent nodes connected by a computer network. The mediator is responsible for creating a common logical representation (model) of data and isolates clients from the distributed physical nature of the data. The positive side of this approach is the possibility of using ready-made single-machine transactional DBMSs for implementing databases of individual nodes. However, the extremely high complexity of the mediator, the need to use an additional transaction coordinator with a slow and unstable to the coordinator's failures two-phase fixation protocol, network congestion, which become especially noticeable with an increase in the number of clients and the size of the database, make this approach resource-intensive and poorly scalable.

Other approaches that provide good horizontal scaling are associated with weakening requirements for atomicity and data consistency in favor of high availability, scalability and system performance.

Partitioning (English ratyutid) is another approach used to create distributed database management systems. It is based on dividing the logical database into separate parts stored on several physical nodes. Partitioning can be horizontal when tables are divided into rows in a relational database, or vertically when tables are divided into columns.

One of the types of horizontal partitioning is sharding (from English kyatbtd), which is also applied to object and other database management systems of non-relational type. The essence of sharding is the distribution of logical database objects among physical databases on nodes in accordance with the value of the object key or the value of some hash function from the object key. An example is the separation of all mailing addresses into disjoint ranges of postal code values and the distribution of these ranges together with the addresses included in them among the nodes. Thus, geographically close addresses are more likely to fall on the same node. When it is necessary to avoid imbalances between the number of objects on the nodes, a hash function is used, for example, the function of summing all digits of the passport number with dividing the result by the number of nodes allows you to get the target node number for each object and distribute records with passport data evenly among all nodes.

The sharding or partitioning approach has several drawbacks. Among them, the need to implement a two-phase transaction fixation protocol, which, as you know, is slow and unstable to coordinator failures, as well as a violation of LSGS requirements for transactions involving objects on several nodes. An insurmountable problem of this approach is the ordering of transactions of different clients that come simultaneously to different nodes and cover the same data that are distributed between several nodes. Ordering transactions using timestamps requires matching the system clocks of all nodes with a level of accuracy that in practice is not achievable. That is why in this approach it is difficult to implement competitive access control using multiversion. Unsuitable are also locks leading to endless expectations in a distributed system. In addition, in this approach, the operation of joining records tables (operation или relational database) becomes difficult to implement or unacceptably slow. In terms of object databases, it becomes difficult to implement the query of a set of main and subordinate objects located on different nodes. Especially in relation to queries with the condition that the values of the attributes of records or objects fall within the specified range.

Sharding or database partitioning approaches are applicable for a limited range of tasks when scalability and high availability of the system are more important than full-fledged transaction support.

Another approach is based on the use of a distributed transactional cache. Its essence lies in the fact that all database objects, including directories, table entries, index fragments, as well as other data and metadata (so-called atoms), are distributed in several copies among many transactional nodes that play the role of cache memory . At the same time, at least one of the nodes is not a cache (the so-called archive node), but provides permanent storage of the database. Object editing operations affect the archive node and all transactional nodes, where at the time of editing there were cache copies of objects. To do this, a list of links to

- 2 027808 nodes where there are copies of it. Any of the transactional nodes can fulfill the user’s request, ensuring that all data used to execute the request flows over it. Copying objects to transactional nodes and matching their copies are performed as necessary. To ensure data consistency, each node maintains a counter of the last transaction commit. This approach is closest to the proposed invention in terms of the achieved technical result, since it allows to implement an elastically scalable database management system with support for ASGO transactions, in which nodes can even be geographically distributed. However, the need to have an archive node on which the entire database is located limits the scalability and versatility of this approach. In fact, the problem of efficient storage and quick editing of a large database is transferred to the archive node. An insurmountable problem of this approach is also full-text indexing and keyword search, when the volume of index data is too large for one node.

In well-known approaches, a separation is often made between the logical and physical schemas of a database, when the logical schema used to design the database is translated into the physical schema used to implement the database using a specific DBMS. The logical scheme operates with high-level concepts, for example, for the EC model (from the English ΕηΙίΙνКе1айопвЫршойе1) - these are entities, attributes of entities, relationships between entities. The physical scheme operates with low-level concepts, for example, for a relational database - these are tables, fields, indexes. Moreover, in well-known approaches, the concept of a relationship between entities is not a concept of an interface with a DBMS, but is merely modeled in one way or another, which does not allow you to efficiently manage direct and feedback connections between objects distributed among nodes and automatically perform the necessary data denormalization to reduce internal communication between nodes and fast query execution.

To summarize, we can state that the well-known approaches meet many, but not all, of the requirements for modern databases. Actually, a database management system is required that provides good horizontal scaling with an increase in the number of users and data volumes, while supporting data sharding, when the functions of permanent data storage are distributed between nodes, i.e. none of the nodes contains the entire copy of the database. Such a system should ensure atomicity, consistency, isolation and reliability of transactions, i.e. satisfy ASGO-requirements. The system should be decentralized and provide high availability and fault tolerance due to backup and storage of redundant copies of objects and metadata on several nodes. In addition to this, such a database management system should provide the user with tools for working with structured data (objects), large binary data (files), as well as means of high-performance full-text data search for keywords.

SUMMARY OF THE INVENTION

The objective of the invention is to obtain a multi-user distributed data processing system in which none of the nodes contains a complete set of all database objects.

Another objective of the invention is to obtain a multi-user distributed system of transactional data processing, providing atomicity, consistency, isolation and reliability of data operations.

Another objective of the invention is to obtain a multi-user distributed data processing system that provides a high degree of availability.

Another objective of the invention is to obtain a multi-user distributed data processing system that provides a high degree of fault tolerance.

Another objective of the invention is to provide a scalable multi-user distributed data processing system.

In accordance with one aspect of the present invention, all of the above problems are solved by horizontal partitioning of data by distribution (sharding) and permanent storage of database objects on several nodes of a computer network with the ability to intersect subsets of objects of any two nodes, when no node is required to store the full the set of all database objects, as well as by managing forward and backward logical links between objects, when each object is created and maintained logical list of backlinks to objects containing direct links to this object is relevant, the node that stores the keys of all objects referencing this object is stored on the node, and a list of backlinks to objects located on this node is associated with each of these keys, and direct links objects located on other nodes are replaced by links to nodes from which target objects can be obtained by the key of the object referencing them, as well as by ordering objects according to the values of one or more fields and distribution of copies of objects among several database nodes in such a way that the whole set of field values involved in ordering is divided into ranges, and each range is assigned a database node where copies of objects included in ordering in this range are stored, as well as the transmission account between nodes of the database of 3,027,808 messages with commands that allow you to save and retrieve objects at the request of customers with checking the conditions of the request at those nodes where copies of objects are stored, without transferring all required s to verify the conditions for requesting index data and copies of objects on one node coordinating the execution of a client request. An effective horizontal scaling of the database management system is based on the method of issuing unique primary key values to database objects, in which unique primary key values are issued by each node independently of other nodes and without resorting to a third party, unique primary key values are distributed by each node in the entire range possible primary key values in evenly small groups. To ensure data consistency, each node contains an increasing logical time counter used for issuing time stamps, and for any two values of the time counter, a strict order relation is defined, the time counter is incremented before each internal event of the node, each node transmits the current value of the time counter to other nodes when sending any messages, each node when receiving a message from another node increases its time counter so that it is greater than the largest of the current value and the time counter value received with the message, the time counter value is returned in each response to the database client’s request, it is stored by the client while working with the database, and transmitted with each request for coordinated access to the data.

In accordance with another aspect of the present invention, the objectives of the invention are achieved by horizontal partitioning of data by distribution (sharding) and permanent storage of relational database records on nodes of a computer network with the ability to intersect subsets of records of any two nodes, when no node is required to store the complete set of all database records, as well as by managing direct and inverse logical relationships between database entities, when for each direct relationship is created and maintained logical feedback is relevant, feedback is stored on the node storing the record with this record, and direct communications with records located on other nodes are replaced by mapping to nodes from which the associated records can be obtained by the record key on the source node, as well as sorting records by the values of one or more fields and distributing copies of records among several database nodes in such a way that the whole set of field values involved in the ordering is divided into ranges, and each range with the database node is kept in correspondence, where copies of records are stored that are included in this range during ordering, as well as by transmitting messages between the database nodes with commands that allow you to create, update and read data at the request of customers by checking the query conditions on those nodes where copies of records are stored, without transferring all the conditions for requesting index data and copies of records required for verification to one node coordinating the execution of the client’s request. An effective horizontal scaling of the database management system is based on the method of issuing primary key values to entities of a relational database, in which unique values of primary keys are issued by each node independently of other nodes and without resorting to a third party, unique values of primary keys are distributed by each node in the entire range possible primary key values in evenly small groups. To ensure data consistency, each node contains an increasing logical time counter used for issuing time stamps, and for any two values of the time counter, a strict order relation is defined, the time counter is incremented before each internal event of the node, each node transmits the current value of the time counter to other nodes when sending any messages, each node when receiving a message from another node increases its time counter so that it is greater than the largest of the current value and the time counter value received with the message, the time counter value is returned in each response to the database client’s request, it is stored by the client while working with the database, and transmitted with each request for coordinated access to the data.

List of drawings

In FIG. 1 schematically presents one of the options for the DBMS topology.

In FIG. 2 shows a diagram of the hardware components of one node.

In FIG. 3 shows a diagram of the software components of the node in the context of the roles of the coordinator and participant in the transaction.

In FIG. Figure 4 shows an example of a graph of objects that contain links to each other and are distributed in some way among database nodes.

In FIG. 5 shows a detailed structure of objects and links for the example of FIG. 4.

In FIG. 6 shows a detailed structure of object revisions for the example of FIG. 5.

In FIG. 7 shows a diagram that allows you to understand how to place objects among database nodes.

In FIG. Figure 8 shows a flowchart that helps to understand the way in which transactions to retrieve objects from a database are performed.

In FIG. Figure 9 shows a flowchart that helps to understand the way in which transactions to save objects to a database are performed.

- 4 027808

In FIG. 10 is a block diagram in addition to FIG. 9 and helping to understand the way in which objects are saved on a participating node to a database fragment.

In FIG. 11 shows the first part of a flow chart that helps to understand the way in which transactions to search for objects in a database are performed.

In FIG. Figure 12 shows the second part of a flowchart that helps to understand the way in which transactions to search for objects in a database are performed.

In FIG. 13 is an example of some initial database management system scalable in various ways to the options shown in FIG. 14.

In FIG. 14 shows scaling options for the database management system of FIG. 13. Information confirming the possibility of carrying out the invention

Information confirming the possibility of implementing the invention is set forth further in the sections Hardware-software device, Data Structures, Data Processing Methods.

Hardware and software device.

A part of the present invention is shown in FIG. 1 variant of a multi-user scalable distributed transactional database management system 18. DBMS 18 includes a number of nodes N1 to N8, interacting with each other on an equal basis through computer network 19. So, for example, node N1 can directly exchange network messages with nodes from N2 to N8. Messaging occurs in asynchronous mode to ensure maximum system throughput when processing client requests.

The system architecture is described in more detail in FIG. 2, which shows the hardware components of a typical assembly 20. Referring to FIG. 2, a typical node 20 comprises a central processor 25 that communicates with one or more clients 21 via a network interface 27 and a computer network 22. At the same time, a node 20 communicates with other same nodes through a network interface 28 and a computer network 24. The central processor 25 also interacts with random access memory 26, which stores a copy of the database management program 29, which implements the preferred embodiment of the present invention. When this program is running, operational storage of database objects 30 is provided, customer requests are executed, communication with the same programs of other nodes is ensured, and related tasks are solved. The central processor 25 additionally interacts via an input-output device 31 with a non-volatile read-only memory 32, for example, a data storage device on hard drives. The read-only memory device 32 stores sequentially recorded (so-called serialized) object data 33 and binary database files 34.

Typically, computer networks 22 and 24 are physical subnets of computer network 19 or logical routes within computer network 19. Separation of computer network 19 into independent subnets is usually performed for security and performance reasons, although it is not required. Computing network 22 should provide high throughput and allow relatively large delay times when transmitting messages due to the remoteness of the clients. Computing network 22 usually uses global Internet technologies and contains a request balancing device 23 that routes and evenly distributes client requests among all nodes connected to network 22. Computing network of nodes 24 should provide high throughput and low latency of messages in the network during interworking communication and is usually based on LAN technology and fiber optic communication lines.

As a protocol for transmitting messages, computer networks 22 and 24 can use the TCP protocol (English Tgapvt188up Οοηίτοί Рго1ео1 - transmission control protocol), which is the protocol of the transport layer of the Internet network and ensures reliable transmission of the data stream through the established connection. In addition, application-level protocols can be used as the message transfer protocol in networks 22 and 24, for example, the HTTP protocol (English NuregTech! TgapzGeg Рgo1ео1 - the hypertext transfer protocol), which is the main protocol of the world wide web (the so-called web), it works on top of the TCP protocol, provides dialogue interaction in the request-response mode, has the means of typing data and setting contextual data (so-called cookies), the ability to forward requests and multiplex them within the same connection, and also allows so easy to overcome firewalls. The application-level protocol §ОАР (English §1р1е ОБ) ес1 Асееве Рго1ео1 - a simple protocol for accessing objects) used by web services for exchanging structured messages and working on top of the HTTP protocol, as well as other similar protocols, can also be used. The DBMS option under consideration supports several protocols as the basic message transfer protocol, in particular, the TCP and HTTP protocols selected through the system configuration parameters, the TCP protocol being preferably used for interaction between nodes (for performance reasons), and the HTTP protocol for interacting with clients 21 (in order to overcome firewalls and provide access to a database of web clients).

In a preferred embodiment, an object-oriented programming paradigm (OOP) is used. According to the OOP paradigm, objects are dynamically created instances of classes that combine data (so-called properties) and their routines

- 5,027,808 processing (so-called methods). Classes are descriptions of dynamic modules, instances of which objects can be created in unlimited quantities, in contrast to static modules of structural programming. Classes as data types can extend other classes (the so-called inheritance) and redefine the work of methods by using the mechanism of indirect call of subprograms (the so-called polymorphism). It will be apparent to those skilled in the art that the invention can be practiced both with and without using OOP tools.

In FIG. 3 shows the composition of the software components of a typical node 20, which can play two roles: 1) a transaction coordinator, 2) a transaction participant. In the considered embodiment, any node is capable of performing both one of the roles and simultaneously both of these roles. According to FIG. 3, node 40 acts as a transaction coordinator, providing clients with an interface and providing access to the database, and node 60 acts as a participant in transactions, providing fast and permanent data storage. Thus, the nodes 40 and 60 are functionally identical, and when working in one role or another, one or another set of functional modules is involved in them. Modules and communications that are enabled are marked with a continuous line, and modules and communications that are not involved with a dashed line.

The transaction coordinator node 40, using the network service for receiving client requests 41, receives requests from clients of the database 21 through the computer network 22.

The network service for receiving client requests 41 is an interface for communication between a client and a DBMS and is a web service. In a preferred embodiment, the service 41 is based on the HTTP protocol. Service 41 performs two tasks: deserialization - sequentially reading client data from an incoming network stream and restoring request objects from this data; serialization - the conversion of objects obtained as a result of a query into a stream of serial data with their transmission through the network to the client. As a data format for serialization and deserialization, text in some markup language of documents with a simple formal syntax is usually used, for example, in the language XMB (from English exMagMe Magkir Lapdiade - an extensible markup language).

The network service for receiving client requests 41 transmits deserialized requests to the kernel for executing client requests 42, converting the commands of a higher level of a client into a sequence of commands of a lower level of the system. In general terms, the kernel for executing client requests provides parsing, compilation and optimization of client requests, and then executing control commands on other objects of the system. Commands are executed on such objects as the system transaction manager 43, the list of databases 44, a fragment of the database 45, the communication channel with another node 54, as well as on their subordinate objects.

Client requests can be presented in the form of objects of some DBMS software interface or in the form of text descriptions in some language, such as, for example, the language of the GOITER (from English 81gis1igeb Onegu Lapdiade - the language of structured queries). The SCA language is used in relational database management systems.

In the present embodiment, the following types of client queries to the database are proposed: Save objects, Extract objects, Find and extract objects by condition. Behind each type of request there are corresponding classes that describe the request object and the response object transmitted between the client 21 and the query execution core 42.

According to FIG. The 3 database fragments 45 and 65 store and serve subsets of the logical database data. There can be many such fragments on one node, since in the preferred embodiment of the invention under consideration the node supports working simultaneously with several independent databases within the same node infrastructure. Fragments of all independent databases of one node are collected in lists 44 and 64. Each database is identified by its name, and database lists 44 and 64 provide access to the desired fragment by the database name.

The kernel for executing client requests 42 on the transaction coordinator node 40 uses only those modules of the database fragment 45 that are necessary for coordinating transactions, namely the database configuration 46, the transaction manager 48 and the key generator 50. Data fragments are provided by the database fragments on the nodes- participants, for example, fragment 65 on the participating node 60. In the fragment of database 65, only those modules are used that are necessary to ensure data storage and transactional work with them, namely, the configuration of database 66, edzher transaction 68, the primary index 67, secondary codes 69, reference codes 71, 72 and a persistent storage file storage 73.

To execute a client request, the kernel 42 turns the client request into many internal requests for fragments on other nodes where the requested data is located. Internal requests are transmitted to the internal query execution core 74 of each participant node 60 involved in the transaction through the logical communication channel 55 of the coordinator node and through the corresponding logical communication channel 75 of the participant node. A bi-directional logical communication channel is created between each pair of nodes. It provides transparent interaction between the kernel for executing client requests 42 and

- 6 027808 internal query execution kernel 74 with a database fragment on another node as if it is located on the same node as the query core.

Sending messages with internal requests and receiving messages with answers is performed by the communication channel using queues of outgoing and incoming messages. The outgoing message queue 56 is serviced by the network message sending service 58, which removes requests from the queue and transfers them from its node, in particular node 40, to the network message receiving service 79 at the member node 60. The network message receiving service 79 puts incoming requests in the incoming queue Messages 77. The kernel for executing internal requests 74 removes requests from this queue and executes them. Resultant responses to internal requests fall into the outgoing message queue 76. The network message sending service 78 removes the outgoing replies from this queue and transfers them from the participating node 60 to the coordinator node 40, where they fall into the network message receiving service 59, which puts the replies to the incoming queue messages. Incoming responses are associated with communication channel 55 with outgoing requests and are returned to the client request execution core 42.

So that the response message can be matched with the request message, each message contains the identifier of the request-response pair. The identifier is issued to the request message, travels with it to the called node, then it is copied from the request message to the response message and returned to the calling node.

In the implementation of the network message sending service 56 and the network message sending service 76, it is recommended to use buffering and multiplexing - if possible, transmit several messages of several computing streams with one network packet of the basic transport protocol selected for node interaction. Buffering and multiplexing can increase network bandwidth, reduce network traffic and reduce the average delay time when transmitting messages in the computing network of nodes 24.

According to FIG. 3, in the present embodiment, the node database fragment contains the following modules: database configuration, transaction manager, key generator, primary index, secondary indexes, reference indexes, file storage, persistent storage. The listed modules are discussed below.

The database configuration 46 (on the coordinating node) and 66 (on the participating node) contains objects that describe the structure of the database. In this embodiment of the invention, the database configuration includes determining the types of objects, determining ways to sort objects by the values of one or more fields (the so-called database indexes), a list of database allocation nodes, index allocation maps among nodes, file allocation maps among nodes The accepted backup level for the database. The definition of the type of objects (the so-called database entity) consists of the definitions of scalar and reference fields of objects of this type. For reference fields (the so-called relations), the fields with backlinks corresponding to them in the addressable objects are indicated. Backlink values are monitored and, if necessary, automatically added, deleted, or adjusted by the database management system when the user sets direct links. Reference fields can be list (the most common case) and single. A list link field contains many links, a single link field contains one link. The backward link for any reference field can be a single or list field in the addressed object. Methods of ordering objects according to the values of one or more fields in the considered embodiment of the invention are classified as follows: primary index, secondary indices, reference indices. All types of indices and index placement maps among nodes are discussed below.

According to FIG. The 3rd fragment of the host database contains a transaction manager. Transaction Manager 48 (on the coordinating node) and 68 (on the participating node) provide management of distributed transactions. The distributed transaction mechanism is discussed in detail below in the description of the invention. The transaction manager provides the means to create, start, commit, and roll back transactions. It includes a list of current transactions, an active client registry, a transaction timestamp generator, and a garbage collector that removes obsolete revisions of transactional and state objects.

In addition to a set of transaction managers — one for each database fragment — the node contains a system transaction manager, shown in FIG. 3 blocks 43 and 63 at nodes 40 and 60, respectively. The system transaction manager provides atomic and consistent execution of client administrative requests, in particular requests for creating, deleting, and configuring databases. Functionally, the system transaction manager is no different from the transaction manager of a database fragment, since it is an instance of their common class. The difference is that system transactions always cover all DBMS nodes, both the transaction coordinating node and the transaction participating nodes.

The key generator 50 is used by the coordinator node to create unique primary keys for database objects. In this embodiment, the primary key is a 64-bit integer. So that keys issued by different nodes do not repeat,

- 7 027808 part of the bits in the key value is used to encode the number of the node that issued the key. A method for issuing primary keys is described below in the description of the data structures of the present invention.

The primary index 67 on the member node is a fragment of a dictionary that maps the primary keys of objects to the object storage cells. Adding a new key to the dictionary and the corresponding object storage cell occurs on the node atomically for client requests. In the present embodiment, atomicity is ensured by a locking mechanism. An object storage cell is created once when a new primary key is added to the dictionary and is not recreated when editing the object. It maintains a list of object revisions and uses the ICCM mechanism to control competitive access using multi-versioning. The primary index is distributed among the DBMS nodes based on the database 46 (on the node coordinator) and 66 (on the participant node) of the primary index allocation map. The primary index allocation map is used by the coordinating node to determine the nodes involved in the transaction. In an embodiment of the invention, a primary index allocation map may be constructed as follows. The set of possible values of the primary key of the object is divided into continuous ranges. Each range is assigned a list of nodes on which copies of objects with keys from this range are stored. The presence of redundant copies of objects is necessary to ensure a high degree of readiness and system stability to node failures. The primary index location map is used to access an object by its key. The value of the object key determines the range number of the primary index. The range number selects a list of nodes where copies of the object are stored. From the list of nodes, any node is selected to read the object. On the selected node, by the key value in the primary index, the object storage cell is searched and from it, in accordance with the client’s time stamp, the object’s revision is selected. In more detail, the structure of the primary index placement map and the method of its application are described below in the description of the present invention.

Secondary indexes 69 on the member node are fragments of secondary indexes in this database fragment. Each secondary index has an identifier assigned in the database configuration 46 (on the coordinating node) and 66 (on the participating node) and used in client requests. The identifier selects the desired secondary index. The secondary index is a set of objects ordered by the values of one or several fields. The fields by which ordering occurs form a secondary key. If at any given moment in time no more than one object corresponds to the secondary key, the secondary key is considered unique. A non-unique secondary key can correspond to several objects. The secondary index is distributed among the DBMS nodes based on the database map 46 (on the coordinating node) and 66 (on the participating node) of the secondary index location map. The secondary index location map is similar to the primary index location map with the only difference being that the secondary index keys divided into ranges can consist of several fields of various types. The secondary index placement map can be made up so that several ranges of the same secondary index appear on one node. All ranges of the secondary index stored on one node make up a fragment of the secondary index. Physically, each fragment of the secondary index is implemented through well-known data specialists with a typical set of operations — an ordered dynamic array and / or binary tree. To ensure multi-version access to objects in the present implementation of the invention, revisions of objects together with storage cells of objects are included in the secondary index. A unique secondary key in such a data structure can correspond to several revisions of the same object. Moreover, the uniqueness of the secondary key is ensured for any given timestamp at the level of object storage cells.

The reference indices 71 on the member node are logical ordering of links in list object reference fields. Ordering is performed in accordance with the values of one or more scalar fields of objects addressed by links. For example, suppose an object of the User class contains a list link field of a User group in a database of some social network. The reference index allows you to arrange in this field links to objects of the Group class by the value of the Group name field. In terms of relational databases, a reference index corresponds to a secondary index for a table of relationships between entities. Each reference index has an identifier assigned in the database configuration 46 (on the coordinating node) and 66 (on the participating node) and used in client requests.

Permanent storage 72 on a participating node is a means of storing (serializing) objects of a database fragment into one or more files on an external data medium, such as a hard disk, as well as a means of loading (deserializing) objects from these files into RAM. Saving objects to persistent storage occurs periodically. The moment of saving can be chosen in different ways: fixing the transaction, garbage collection in the database, an explicit request from the user to save objects in the permanent storage. Downloading objects from persistent storage is performed during the launch of the DBMS node, as well as during the reboot of the node, which is equivalent to the node launch scenario. Permanent storage requirements are fast sequential

- 8 027808 write data to an external medium and quickly sequentially read data from an external medium. The implementation of persistent storage does not require cumbersome means of virtualization and projecting files into RAM, since database objects are not ejected from RAM to external media, they are always present in memory and stored on external media only for saving and restarting nodes.

The file storage 73 on the participating node provides storage of data of the type BOB (from English Βίpagu lagde OB) uc1 is a binary large object) in the file system on external media of the DBMS nodes, for example, on hard disks. In the present invention, a BOB-type data set is represented as a particular database file object. File objects, like ordinary objects, can have scalar and reference fields for storing data, as well as participate in transactions along with ordinary objects. One of the predefined scalar fields of a file object is the name of a logical file in the database. A secondary index with a unique key is built on this field so that you can search for file objects by name. Unlike other objects, each file object has a logical field for storing its own data set of the BOB type. The data set is not stored in RAM, but on an external medium, usually on a file system. The following operations are provided with a data set of a file object: Read whole, Read a block, Add a block to the end, Overwrite the whole. Various implementations of the invention may additionally support the operation Overwrite the block, but this seriously complicates the management of revisions of the file object and slows down the shared access to the data set from different clients due to the need to use additional locks.

A data set of type BOB of each file object (file data) is distributed among DBMS nodes based on the rules for placing files in file storage specified in the database configuration 46 (at the coordinator node) and 66 (at the participant node). The rule defines the set of nodes among which the files are placed, and the size of the continuous recording block per node. File data is written to the nodes in the set in blocks in a circle taking into account the backup level specified in the database configuration (if it is greater than zero, the same data block gets to several adjacent nodes). At the time of recording, a file location map is built or updated. It is stored in one of the internal system fields of the file object revision. A file allocation map is used to read file data. Using it, for any section of the file, you can determine the location of blocks of this section on the database nodes.

Data structures.

In FIG. 4 shows an example of an object graph. Objects contain links to each other and are distributed in some way among database nodes. In the example, it is assumed that the objects belong to two different classes: and (from the English. Ikeg - user) and C (from the English. Sgoir - group) and are respectively designated I1, I2, IZ and C1, C2, C3. Class objects refer to objects of class C. So object I1 refers to objects C1 and C2, object I2 refers to objects C2 and C3, and object IZ refers to objects C1 and C3. These links are considered direct from the point of view of objects of class AND. They are shown in FIG. 4 solid lines. Direct links correspond to backlinks: from objects C1 and C2 to object I1, from objects C2 and SZ to object I2 and from objects C1 and C3 to object IZ. Backlinks are shown in FIG. 4 dashed lines. Note that for objects of class C, direct links are shown by dashed lines, and backlinks are shown by solid lines. In the example, it is assumed that, according to the primary index allocation map, objects I1 and I2 were at node N1, objects C1 and C2 were at node N2, and objects IZ and SZ were at node No. At the same time, links between objects from IZ and SZ connect objects of one node, and the remaining links connect objects of different nodes.

In FIG. 5, for the above example, a detailed method of storing objects and links is shown. Objects I1 and I2 are stored on node N1, objects C1 and C2 are stored on node N2, objects IZ and SZ are stored on node No. The named objects are distributed among the nodes in accordance with the value of their primary key and the primary index allocation map. Objects I1, I2, IZ, C1, C2, SZ store the values of scalar fields. In each object, links to other objects are replaced by links to nodes from which other objects can be obtained by the primary key of the object referencing them. To do this, the nodes contain auxiliary objects marked with * and designed to store backlinks to their main objects. Auxiliary objects of the example: I1 *, I2 *, IZ *, C1 *, C2 *, SZ *.

Objects I1, I2, IZ, C1, C2, SZ have primary keys, respectively, KI1, KI2, KIZ, KS1, KS2, KSZ, by which they can be found in the primary index. In addition to scalar fields, these objects contain some list link field showing how storage of links is organized. For objects of class And this is a field And ^ C, and for objects of class C a field C ^ And. In terms of a hypothetical social network that includes the classes User and Group, the field I ^ C corresponds to the relation of the User Group, and the field C ^ I to the relation of the Members of the group.

In the objects I1, I2, IZ, the field I ^ C stores links to the corresponding nodes where the objects C1, C2, C3 are located. In particular, in the object I1 in the field I ^ C the reference I1 ^ No. to the node N2 is stored, in the object I2 C in the field I ^ C the links υ2 ^ Ν2 and υ2ΑΝ3 to the nodes N2 and N3 are stored, and in the object IZ to field And ^ C stored links υ3ΑΝ2 and υ3ΑΝ3 to the nodes N2 and N3. Similarly, in the objects C1, C2, C3 in the field C ^ And links are stored to the corresponding nodes where the objects C1, C2, C3 are located. In particular, in the object С1 in the field С ^ И the links С1 ^> N1 and С1 ^ N3 are stored to the nodes N1 and N3, in the object С2 in the field С ^ И the reference С2 ^ Ш is stored to the node N1, and in the object С3 in the field С3 C ^ And references C3 ^ W and C3 ^ N3 to the nodes N1 and N3 are stored. Thus, direct references to objects in objects I1, I2, I3, C1, C2, C3 are replaced by links to nodes.

To encode links to nodes, you can use the numbers, names or identifiers of nodes. However, in practice, it is better to use indirect addressing through the numbers of the ranges into which the primary index is divided. Primary key ranges are mapped to nodes, forming a map of the placement of objects among nodes. All objects included in the same range by the value of their primary key are thus placed on the same node. Instead of links to nodes, the numbers of the corresponding ranges are written in the reference field, into which the whole set of potential values of the primary key is divided. The number of ranges can significantly exceed the number of nodes and provides the necessary flexibility in managing the placement of objects. Note that even if direct links are encoded by range numbers, the latter are ultimately mapped to nodes during system operation, which is used as an assumption in FIG. 5.

Auxiliary objects I1 *, I2 *, I3 *, C1 *, C2 *, C3 * have primary keys KI1, KI2, KI3, KS1, KS2, KS3, respectively, by which they can be found in the primary index. These objects contain a list link field that shows how backlink storage is organized. For the objects I1 *, I2 *, I3 * this is the field I ^ C, and for the objects C1 *, C2 *, C3 * it is the field C ^ I, which in terms of the hypothetical social network mentioned above corresponds to the relations of the User Group and the Members of the group.

At each node where objects I1, I2, I3 are located, auxiliary objects C1 *, C2 *, C3 * are stored, containing backlinks of the main objects within this node. Similarly, at each node where the objects C1, C2, C3 are distributed, auxiliary objects I1 *, I2 *, I3 * are stored, containing backlinks of the main objects within this node.

In particular, on the node N1, the auxiliary object C1 * contains in the field C ^ AND a link C1 ^ I1 to the object I1. At the same node, the auxiliary object C2 * contains in the field C ^ I links C2 ^ I1 and C2 ^ I2 to the objects I1 and I2. At the same node, the auxiliary object C3 * contains in the field C ^ And a link C3 ^ I2 to the object I2.

At node N2, the auxiliary object I1 * contains in the field I ^ C references I1 ^ C1 and I1 ^ C2 to objects C1 and C2. At the same node, the auxiliary object I2 * contains in the field I ^ C a link I2 ^ C2 to the object C2. At the same node, the auxiliary object I3 * contains in the field I ^ C a link I3 ^ C1 to the object I1.

At node N3, the auxiliary object C1 * contains in the field C ^ AND a link C1 ^ I3 to object I3. At the same node, the auxiliary object C3 * contains in the field C ^ AND a link C3 ^ I3 to the object I3. At the same node, the auxiliary object I2 * contains in the field I ^ C a link I2 ^ C3 to the object C3. At the same node, the auxiliary object I3 * contains in the field I ^ C a link I3 ^ C3 to the object C3.

In a preferred embodiment of the invention, the method for storing objects and links is supplemented by competitive access control tools based on the multi-version mechanism of the ICSS. According to the refined data structure, the object is replaced by the storage cell of the list of revisions of the object. Each revision is associated with a timestamp corresponding to the moment of committing the transaction within which this revision was created. The time stamp is used to filter data changes that are visible to the client. The reader transaction selects the latest revision from the list in which the timestamp does not exceed the timestamp of the reader transaction. Old revisions that clients can no longer work with are periodically removed from the list and disposed of by the database garbage collector.

The augmented data structures necessary to ensure the operation of the multi-version mechanism are shown in FIG. 6 as extensions of the ordinary object I1 and the auxiliary object C1 * of FIG. 5 to the level of representation of object storage cells and revision list.

Object I1 at node N1 is represented as storage cell NI1, in which the link O ^ I1B addresses a linked list of revisions of the object. The first revision on the list stores the latest revision, I1B, created as part of transaction T2. The revision contains the primary key KI1 (common for all revisions of the object), a TAG2 link to transaction T2, a link P ^ I1A to the previous revision I1A, a backward link H ^ NI 1 to the storage cell NI 1 (through it you can get to any revision), values the scalar fields RI 1B stored in the audit and the values of the reference fields stored in the audit, of which FIG. 6 shows only the field I ^ C with reference I1 ^ No. The I1B revision timestamp is stored in transaction T2 in the TSh field. The previous revision I1A contains the primary key KI1 (the same as in revision I1B), the TAG1 link to transaction T1 (within the framework of which the revision was created), an empty link to

- 10 027808 previous revision (previous revision is absent), backward link H ^ NI 1 to storage cell Ni1, the values of the RFL scalar fields stored in the revision and the values of the reference fields stored in the revision, of which FIG. 6 shows only the field I ^ O without links.

The auxiliary object O1 * is represented as a storage cell HO1 *, in which the link O ^ O1 * A addresses a single revision O1 * A (in the example, it is assumed that the object O1 * has only one revision). Revision O1 * A contains the primary key KO1, a link T ^ T2 to transaction T2 (within the framework of which the revision was created), an empty link P ^ L1b to the previous revision (no previous revision), a reverse link Н ^ НО1 * to the storage cell НО1 * , the values of reference fields stored in the revision, of which only the O ^ And field is shown with the link 01 ^ -NS.

Shown in FIG. 6, the structure of storing revisions of objects suggests that revision timestamps are not stored in revisions themselves, but in transaction objects, in particular T1 and T2. A transaction object is first created on the coordinator node and duplicated on each participating node, where objects are updated within the transaction. A transaction object is identified by a timestamp, for example, transaction T1 is identified by a timestamp ΤΟΝ1, transaction T2 is identified by a timestamp ΤΟΝ2. In FIG. 6 it is believed that transaction T2 is completed after transaction T1 and the timestamp of the TSH is greater than the timestamp of the TSH. Each transaction object contains a list of T ^ H links to storage cells of objects updated on the node as part of the transaction. In FIG. 6 it is assumed that in transaction T1 on node N1 only the I1 object was updated, therefore the object T1 contains in the field T ^ H a single link T1 ^ HI1 to the storage cell HI1. In transaction T2, two objects were updated on node N1 - I1 and O1 *, therefore, object T2 contains in the field T ^ H a link T1 ^ HI1 to the storage cell HI1 and a link T1 ^ HO1 * to the storage cell HHO1 *.

Shown in FIG. 6, the storage structure of revisions of objects I1 and O1 * can be expanded to any number of revisions, objects, transactions, and nodes by the usual repetition of elements.

In FIG. 7 shows a diagram that allows one to understand the method of placing objects among database nodes proposed in the invention and implemented in the system. The key generator 50 comprises an internal unsigned integer counter 80. The counter cell width in a preferred embodiment of the invention is 64 bits or more. A part of the cell in the range of bits from O to B is atomically incremented by one each time the key generator 50 is accessed for a new value. Part of the cell in the range of bits from C to O is reserved for the node number. In them, the key generator 50 enters the same unique number of its node. This ensures the uniqueness of the numerical values issued by the generator. The numerical value of the counter 80 is converted to the unsigned numerical value of the primary key 81 as follows. Part of bits A to B in counter 80 is symmetrically displayed in such a way that bits A and B, bits A plus one and B minus one, bits A plus two and B minus two, and so on, until all bits are symmetrically swapped, part bits from O to A in the counter 80 are not involved in symmetric permutation. The primary key 81 resulting from this generation is unique in the database, evenly distributed in the range of its possible values on a large sample and grouped with a number of other sequentially issued primary keys on a small sample. The method of placing the object addressed by the key on the database nodes is based on this property of the primary key — the value of the primary key 81 is translated on the primary index 82 allocation map to one of the object's nodes.

The map of the primary index 82 may include several object distributions among database nodes, of which the most recent distribution is used to place newly created objects on the nodes, and all other distributions are historical and are used to determine the location of objects created in the past. For example, in FIG. Figure 7 shows three distributions: the distribution of objects 83, the distribution of objects 84 and the last current distribution of objects 85. Each new distribution of objects in the primary index allocation map is created by the database administrator as necessary, for example, when adding new member nodes on which the primary index is planned to be placed.

The distribution of objects consists of a set of non-overlapping ranges of unsigned numerical values of the primary key, from the minimum zero value to the maximum possible MAX value. For each range, a list of nodes to which objects with primary keys from this range are stored is indicated. The number of nodes in the range depends on the backup level specified in the database configuration. If the redundancy level is zero (operation without redundancy), the list of nodes in the range consists of one node. If the reservation level is equal to one (single hot standby mode), the list of nodes in the range consists of two nodes. If the redundancy level is two (two hot standby mode), the list of nodes in the range consists of three nodes. The reservation level can be arbitrarily large. In a preferred embodiment, it is limited to three (operating mode with three hot reserves) for economic reasons. An example of a primary index allocation map in FIG. 7 is created for the operation mode with one hot reserve of each node located on the neighboring node, and in the distribution 11 027808 division 83 three equal ranges on three nodes are specified, in distribution 84 two equal equal ranges on two nodes are set, in distribution 85 five equal sizes are set ranges on five nodes.

The distribution of objects contains the initial and final values of the counter 80 of the key generator, in the range of which the objects were issued or issued primary keys. If the numerical value of the counter 80 obtained by the inverse transformation from the primary key 81 of a certain object is between these values, then the placement of this object is determined by this distribution. In the example of FIG. 7, the initial and final values of counter 80 for distribution 83 are zero and one billion, respectively, the initial and final values of counter 80 for distribution 84 are one billion plus one and a half billion, respectively, the initial and final values of counter 80 for distribution 85 are one and a half billion plus one and the maximum possible value of the MAX counter of 80. For a 64-bit counter, the value of MAX is two to sixty-fourth.

When the database administrator creates a new distribution for primary index objects, it applies only to new objects - a new distribution is created in the primary index allocation map in which the initial value of counter 80 is set to one greater than the maximum value known in the database and the final value of counter 80 is not limited to. Moreover, in the previous distribution of objects, the final value of the counter 80 is fixed at the level of the maximum value known in the database (the initial value does not change). As a result, a new generation of objects is created with a new location on the database nodes. This method of creating generations of placement of objects assumes that the values of the counter 80 are numerical, are issued to objects sequentially from lower to larger values and can be restored from the values of the primary key 81 by symmetric permutation of bits from B to A.

In other embodiments, the counter 80 and / or primary key 81 may be any type of data, for example a line of text, an array of numbers, a structure with fields, etc. In all embodiments of the invention, the following requirements are met for the counter 80 and the primary key 81: the ability to work with values as sets of bits and fix the bit width.

Data processing methods.

In a highly distributed system, it is difficult or impossible to ensure absolute time synchronization of all system customers and the global ordering of their transactions. In the present invention, each client exists in its time. If it is necessary to provide a causal relationship between the actions of some clients, the system can synchronize clients in time. This is done at the request of one of the synchronized clients. From the moment of synchronization, the client again exists in its time.

Time management in the system is performed at the level of each individual database and its clients, independently of other databases.

Each database node has its own separate time counter, which is a large-sized integer. In the present embodiment, this is a 64-bit integer. The time counter provides, upon request, time stamps, which are a pair of numbers: the node number and the value of the time counter for this node. Of the two time stamps, the larger one has the larger counter value, and if the counters are equal, that time stamp is larger with the larger node number. Two timestamps with the same node number and time counter are considered equal.

There is no single synchronous time for all nodes in the system, each database node lives in its own relative time, and the time value for the entire system is represented by an array of timestamps collected from all database nodes. During the communication of the nodes with each other, the time counters of the interacting nodes are synchronized, and during synchronization, time always moves forward. To ensure time synchronization, the current time stamp of the sending node is transmitted with each message. The promotion of the time counter of the database node is also performed for each internal event of the node, namely when creating a writing transaction, when the transaction is put in a ready state and when the transaction is committed.

The client can request any of the database nodes due to the fact that the system does not store the state of the client, and the concept of a session is absent. The context of client connection to the database is stored on the client side and transmitted along with each request to the database. The client connection context, or simply the client context, contains authentication data and a timestamp. The timestamp in the context of the client advances each time the client accesses the database. For writing transactions, the client’s time stamp is synchronized with the node’s time stamp, and for reading transactions, it moves forward closer to the node’s time stamp. This logic ensures data consistency from the point of view of an arbitrary client and, if possible, conflict-free editing and reading of data simultaneously by several clients.

When a client accesses a node, it may turn out that the time of the node is behind the time of the client, which indicates that the client has completed its last request to another node with which this node has not yet synchronized. In this case, the node time advances towards the client time, i.e. the client essentially provides indirect time synchronization of nodes.

- 12 027808

Thus, each client works with the database in its relative time. If some client needs to read data modified by another client, then for the consistent reading of the correct data, the first client must first match its timestamp with the timestamp of the second client.

It will be apparent to those skilled in the art that such a time management method in a distributed system is an improvement to Lamport watches. The first improvement is that the time counter contains a node number, due to which a strict order relation is determined for all values of the node counters and, therefore, all events in the system are strictly ordered. The second improvement is that the client does not have its own time counter, but at the same time it is a carrier of time counter values, indirectly participating in the time synchronization of nodes. All this allows for data consistency when routing client requests to different system nodes.

In a preferred embodiment, each client request is executed as a transaction. The system manages transactions, and the transaction mechanism is hidden from clients. Transactions are divided into reading and writing. Reading transactions are used to execute data retrieval and retrieval requests, and writing transactions are used to execute data storage queries.

A reader transaction is created by the client query engine 42 in transaction manager 48 and is deleted immediately after the request is executed. Upon creation, the reading transaction receives a timestamp, which is used to retrieve revisions of objects fixed at this point in time and to skip later revisions. The timestamp value for the reading transaction is selected so that it is in the range between the client's timestamp and the current node's timestamp. It is assumed that the client’s time stamp is not greater than the node’s time stamp; otherwise, the node’s time is synchronized with the client’s time, so that this assumption is always true.

In a preferred embodiment, each database node stores a timestamp value that the database node had N seconds ago, where N is the database configuration parameter. The timestamp value for the reading transaction is selected as the maximum from the client's timestamp value and the node's timestamp value N seconds ago. Thus, the relevance of the data visible by the client is no more than N seconds behind the actual node data. Other, more adaptive ways to select a timestamp for reading transactions may also be used.

A writing transaction is created by the client query execution engine 42 in transaction manager 48, placed in the list of active transactions and remains in it after the request is executed. Collective commit is performed for writing transactions, so the writing transaction remains in the list of active transactions until the information about its commit or rollback is distributed to all nodes involved in the transaction, after which the transaction is disposed of by the garbage collection process running in the background. Each writing transaction has a unique identifier, called a key, by which a transaction can be quickly found in the list of active transactions. To ensure uniqueness, transaction keys are timestamps issued independently by the coordinator node, which, together with the issuance of the next key, advances its time counter. In addition to the key, the writing transaction stores a list of node numbers that are involved in the transaction. This list is used to address the sending of commit or rollback commands, as well as to find out the status of a transaction if the coordinator node could not send out commit or rollback commands. After committing, the writing transaction also stores the commit timestamp, which is used by reading transactions to access the database snapshot at a specific point in time, as described above.

When the request is executed, the attributes of the writing transaction, including the transaction key and the list of nodes involved in the transaction, are transferred to the internal query execution kernel 74 of each involved participant node. The core of the execution of internal queries 74 in the transaction manager 68 creates a copy of the transaction with the specified attributes, which is placed in the list of active transactions. After the query is completed by the internal query execution engine 74, the transaction is put into a state of readiness for commit. In this state, both commit and rollback of the transaction can occur, which is determined by the results of the transaction on other nodes. When a transaction is put in a ready state, the current timestamp of the database node is calculated, which is stored in the attributes of the transaction and returned to the kernel for executing client requests 42 at the coordinator node.

After receiving the successful results of the transaction from the participating nodes, the kernel for executing client requests 42 puts the transaction in a ready state on the coordinating node, and the current timestamp of the database node is calculated, which is stored in the attributes of the transaction. For all timestamps collected from the nodes involved in the transaction, the maximum value is calculated, which is used as the timestamp of the transaction commit.

The transaction is committed first at the coordinator node, after which the commit request at the participating nodes is put in a special queue and the transaction result is returned to the client. A commit request contains a transaction key, a commit timestamp, and a list of nodes involved in the transaction. The queue of requests for committing transactions is processed by a parallel process, which periodically selects accumulated requests and sends batch commands to the database nodes in batch mode. The list of nodes to which commit commands are sent is formed as a union of the sets of all nodes from all accumulated requests. Transactions are rolled back in the same way, only rollback commands are sent to the participating nodes instead of commit commands. Thus, the commit or rollback of transactions on the participating nodes occurs asynchronously in batch mode, which significantly reduces the time delay for the client between sending a request and receiving a response, and also reduces the internal network traffic of the system. This approach compares favorably with the two-phase fixation mechanism.

When performing multiple write transactions in those cases when transactions simultaneously update several objects located on several nodes, a transaction conflict may occur. Let us illustrate it with an example. Assume that transactions T1 and T2 update objects A and B located at nodes N1 and N2, respectively. Each of these transactions updates both objects. The scenario may be as follows. At the same time, transaction T1 updated object A, and transaction T2 updated object B. At the next time, transaction T1 tries to update object B, which has already been changed by transaction T2 and has not yet been committed, and transaction T2 is trying to update object B, which Modified by transaction T1 and also not yet committed. There is a transaction conflict. If both transactions wait for a commit, an infinite lock will occur. Therefore, when a conflict is detected, the transaction is rolled back, and the client receives a result with a sign of conflict. In a preferred embodiment, measures are taken to overcome the conflict. The database client automatically tries to retry the request after a short delay in the hope that the next attempt will be conflict-free. After several unsuccessful attempts, the amount of which is determined by the configuration of the client, the client reports the conflict to the user. The user decides to continue trying to fulfill the request or to postpone the action until other clients finish working with the data that caused the conflict.

In the event of a failure of the coordinator node 40 or a failure of the computing network of nodes 24 when transmitting commit or rollback commands, there is a queue of requests in transaction manager 68 for clarifying the results of transactions that have entered a state of readiness for commit, but for some reason have not been committed. Requests are placed in this queue by the transaction manager 68 on the participating node, together with the transaction being placed in a ready state for commit and immediately before the transaction result is returned to the calling kernel for executing client requests on the coordinating node. The queue of requests for clarifying the results of transactions is processed by a parallel process that periodically selects accumulated requests and sends them in batch mode to the database nodes involved in the transaction. For each transaction, the request is sent first to the coordinator node, and then to the participating nodes. The coordinator node is calculated among the nodes involved in the transaction by the value of the transaction key (the transaction key is a time stamp with the number of the coordinator node encoded in it). If, as a result of a survey of all nodes involved in a transaction, it is discovered that a transaction was committed on one of the nodes or was rolled back, then the transaction is transferred to the same commit or rollback state on the node performing the survey. If the final state of the transaction after polling all the nodes involved in the transaction is not determined, the request to find out the result of the transaction is again placed in the processed queue so that after some time the attempt to find out the result of the transaction is repeated. For optimal operation, the time interval between servicing the request queue for clarifying the results of transactions should be significantly larger than the time interval between servicing the request queue for committing and rolling back transactions.

Based on the above structures for storing objects and links (Figs. 5 and 6) and the method of transactional work with objects, we consider the method of extracting objects from the database. We will do this by the example of fulfilling some request to retrieve the I2 User along with all Groups O to which the I2 User refers through the I ^ O field. A block diagram of the execution of the request is shown in FIG. 8.

At step 90, the kernel for executing client requests 42 of some coordinating node N4 receives a request from the client of the form Retrieve objects whose parameters are the client’s timestamp,, the primary key of the KI2 object, the class names of the requested objects and and O, as well as the name of the reference field O> 0 . through which subordinate objects are selected.

In step 91, the transaction manager 48 creates a reading transaction T with a time stamp ΤΟΝ in the list of active transactions. The timestamp value is selected so that it is not less than the client’s timestamp, but not more than the node’s timestamp. Timestamp ΤΟΝ is used to read revisions of objects fixed by this time and to discard later data changes. At this step, the client’s logical time is actually moving forward, and the client is slightly behind in time from the system, which in most cases ensures conflict-free reading and editing of data. The amount of client time lag from the system time may

- 14 027808 dynamically adjusted depending on the current system load.

At step 92, the kernel for executing client requests 42 requests an audit of the object with the primary key KI2 at time ΤΟΝ. To do this, based on the value of the primary key KI2 and the primary index location map, the node where the object is located is determined. In this example, this is node N1. Next, a request 93 containing a time stamp ΤΟΝ and a key KI2 is transmitted to node N1.

At step 94, the core of the execution of internal queries 74 of the N1 node selects the object storage cell by the primary key KI2 in the primary index 67 of the database fragment 65. In this cell, the last recorded revision of the I2 object at the time moment ΤΟΝ is selected, for example, the revision of I2L. The audit of IIL is returned in response 95 to the kernel for executing client requests 42 on node N4.

At step 96, the client query engine 42 queries class C subordinate objects that are referenced by the I2A revision. For this, by referring to the field I ^ C of the revision I2A, the nodes where the subordinate objects are stored are determined. In the example of FIG. 5 are nodes N2 and N3. Further, two subordinate requests 97 and 98 are addressed in parallel to the core of internal queries 74 at node N2 and the core of internal queries 74 at node N3. The parameters of each subordinate request are the timestamp TC ^ primary key KI2 and the field I ^ C.

At step 99, the core of performing internal queries 74 of node N2 searches for the cell of auxiliary object I2 * using the primary key KI2 and selects from it the last committed revision at the moment of time TOM. This revision contains in the field I ^ C references to storage cells of objects of class C located in the primary index 67 of database fragment 65 on node N2. An array of backlinks consisting of FIG. 5 from one link I2 ^ C2, pointing to the storage cell of the object C2. Using this link on node N2, the last recorded revision of the object C2 is selected at the moment TOM, let’s say C2A, and a copy of this revision is returned in response 101 to the subordinate request 97.

Similarly, at step 100, the core of performing internal queries 74 of node N3 searches for the cell of auxiliary object I2 * using the primary key KI2 and selects from it the last committed revision at time TOM. This revision contains links to all cells of objects of class C in the field I ^ C, which according to the map, the primary index locations are stored on node N3. An array of backlinks consisting of FIG. 5 from one link I2 ^ C3 to the storage cell of the object C3. Using this link on node N3, the last recorded revision of the C3 object is selected at the moment TOM, let’s say C3A, and a copy of this revision is returned in the response 102 to the subordinate request 98.

At step 103, the kernel for executing client requests 42 completes the reciprocal links between all received object revisions (revisions I2A, C2A, C3A in the considered example).

At step 104, the reading transaction T is excluded from the list of active transactions and transferred for disposal. The memory utilization of all completed transactions is performed periodically by the transaction manager. The disposal mechanism is discussed below in the description of the invention.

At step 105, the received object graph is returned to the client along with the new TOM timestamp. The client must pass this timestamp in the next server call.

It should be obvious to a person skilled in the art that the considered example can be easily generalized for an arbitrary set of objects, classes, reference fields and subordinate queries by repeating the above steps as many times as necessary.

Based on the above structures for storing objects and links (Figs. 5 and 6) and the method of transactional work with objects, we consider the method of storing objects in a database. We will do this by the example of fulfilling some request to update the scalar data of Group C1 and User I2 and add User I2 to Group C1. A block diagram of the execution of the request is shown in FIG. nine.

At step 110, the kernel for executing client requests 42 of a certain coordinating node N4 receives a request from the client of the form Save objects, the parameters of which are the client’s time stamp CCN, primary key KC1 of object C1, values of scalar fields PC1 of object C1, primary key KI2 of object I2, values of scalar fields RI2 of the object I2, the reference C1 ^ I2 from the object C1 to the object i2.

At step 111, backlinks are added to the objects, in particular, the link I2 ^ C1 is built from the I2 object to the C1 object.

At step 112, the kernel for executing client requests 42 of node N4 creates a write transaction T with a key equal to the increased value of the time stamp of node N4. A transaction is transferred to the execution phase of operations and placed in the list of active transactions of the transaction manager 48.

At step 113, the kernel for executing client requests 42 of node N4 requests the latest fixed revisions of objects C1 and I2 from the nodes. This is done according to the method discussed above and as shown in FIG. 8 block diagram of the extraction of objects. As a result of the execution of the scenario of this scheme, at step 114, from the primary index 67 of the database fragment 65 of the N1 node, a certain revision of the IIX of the I2 object is extracted and returned to the kernel for executing client requests 42 on the N4 node. Similarly on

- 15 027808 in step 115, from the primary index 67 of the database fragment 65 of the N2 node, a certain revision С1Х of the object 01 is extracted and returned to the kernel for executing client requests 42 on the N4 node.

At step 116, the kernel for executing client requests 42 creates objects 01Υ and υ2Υ as copies of revisions 01X and i2X, respectively, in which scalar fields P01 and Pu2 are updated and increased by unit of revision number. In this step, auxiliary objects 01 * Υ and υ2 * Υ are additionally created, containing updates for backlink lists, which, according to FIG. 5 are stored in objects 01 * and υ2 * at nodes N1 and N2, respectively. Object 01 * Υ is an update for object 01 * on node N1 and contains the link 01 ^ -Sh in the additional list of links for the field 0 ^ П. The object υ2 * Υ is an update for the object υ2 * on the node N2 and contains the link Ш ^ -01 in the additional list of links for the field b> 0.

At step 117, the kernel for executing client requests 42 on the secondary index allocation cards calculates the storage nodes of revisions 01X, 01Υ, υ2Χ, υ2Υ in the secondary indices. In the example scenario shown in FIG. 9, it is believed that two secondary indexes are constructed — the secondary index for some scalar fields of objects of class 0 and the secondary index for some scalar fields of objects of class υ. In this case, it is believed that the following placement of revisions of objects 01 and υ2 in the secondary indices occurs: revision 01X - at node N1, revision 01Υ - at node N3, revisions υ2Χ and υ2Υ - at node N2. The resulting placement means that the new revision 01Υ of object 01 turned out to be in a different range of the secondary index compared to the previous revision 01X and should be placed on another node, the new revision υ2Υ of object υ2 is located on the same node as the previous revision of SH.

Given the placement of revisions 01Υ, υ2Υ, 01Х, ШХ in the primary and secondary indices, objects are distributed among the participating nodes as follows. At step 118, node N1 receives: revision υ2Υ for inclusion in the primary index, revision 01 * Υ for inclusion in the primary index as an auxiliary object, revision number 01X for removing the object from the secondary index on this node. At step 119, node N2 receives: revision 01Υ for inclusion in the primary index, revision υ2 * Υ for inclusion in the primary index as an auxiliary object, revision υ2Υ for inclusion in the secondary index. In step 120, the node N3 receives a 01Υ revision for inclusion in the secondary index. Each participating node also receives a transaction identifier T, within which objects are saved.

Further, each of the participating nodes performs the same sequence of operations shown in FIG. 9 in steps 121, 124, 127, and 130. At node N1, the internal query execution engine 74 performs the following actions. At step 121, a writing transaction T is registered. The transaction is transferred to the execution state of transactions and placed in the list of active transactions of transaction manager 68. At step 124, in the database fragment 65, revisions υ2 рев and 01 * фраг are stored in the primary index 67 and the attribute is stored in the secondary index 69 withdrawals of an object with revision 01X from the secondary index. The operations of this step are explained in more detail in FIG. 10. At step 127, transaction T is put into a ready-to-commit state and the timestamp ΚCN1 of this transaction is calculated. The time stamp of the PCS is calculated as the unit time value of the node increased by one. At step 130, a request to clarify the result of transaction T is queued for transactions awaiting a result. The PCS timestamp is returned in response 133 to the kernel for executing client requests 42 to coordinator node N4.

Similarly, at node N2, the internal query execution engine 74 performs the following actions. At step 122, the writing transaction T is recorded. The transaction is transferred to the execution state of the transactions and placed in the list of active transactions of transaction manager 68. At step 125, revisions 01Υ and υ2 * Υ are stored in the database fragment 65 in the primary index 67 and the revision is stored in the secondary index 69 υ2Υ. At step 128, transaction T is put into a ready-to-commit state, and the timestamp of the KSS of this transaction is calculated. The time stamp of the KSSh is calculated as the unit time value of the node increased by one. At step 131, a request to clarify the result of transaction T is queued for transactions awaiting a result. The time stamp of the KSSh is returned in the response 134 to the kernel for executing client requests 42 to the coordinator node N4.

Similarly, on node N3, the internal query execution engine 74 performs the following actions. At step 123, a writing transaction T is registered. The transaction is transferred to the execution state of operations and placed in the list of active transactions of transaction manager 68. At step 126, the 01Υ revision is saved in the database fragment 65 in the secondary index 69. At step 129, transaction T is put into a ready-to-commit state, and the PCS timestamp of this transaction is calculated. The time stamp of the PCS is calculated as the unit time value of the node increased by one. At step 132, a request to clarify the result of transaction T is queued for transactions awaiting a result. The timestamp of the RSL is returned in the response 135 to the kernel for executing client requests 42 to the coordinator node N4.

At step 136, the kernel for executing client requests at node N4, the transaction T is put into a state of readiness for fixation and the time stamp K.С№) of this transaction is calculated. The time stamp K.С№) is calculated as the unit time value of the node increased by one.

- 16 027808

At step 137, a new timestamp ΊΌΝ is calculated, equal to the maximum of the timestamps ΚΗΝΟ, Κ £ Ν1, ΚΗΝ2, ΚΤΝ3.

At step 138, transaction T is committed with the timestamp Τί '.' Ν. The fixation affects only the coordinating node Ν4.

The transaction is committed to the participating nodes later in asynchronous mode. To do this, at step 139, a request to complete the commit of transaction T on the participating nodes Ν1, Ν2, Ν3 is placed in the queue of transactions awaiting completion of the commit. Requests from this queue are fetched by an asynchronous process that sends commit commands to the participating nodes. The transaction fixation mechanism is almost single-phase and does not require the client to wait for the transaction to be committed to all nodes. It is enough to switch the transaction to the ready state on all nodes in order to consider the transaction logically fixed. Participation in the transaction of two or more nodes provides the necessary guarantee of the reliability of the transaction mechanism and its resistance to failure of one of the nodes. An additional optimization during commit is sending out commands to the nodes in batch mode, when in one message the node receives several commit commands for several completed transactions.

At step 140, a new TSI timestamp is sent to the client, which it subsequently uses to perform operations of extracting and searching for objects. It is guaranteed that the client will read the latest object revisions by the time of ΤCN and not see past data regarding this timestamp.

In FIG. 10 is a block diagram in addition to FIG. 9 and helping to understand the way in which objects are saved on a participating node to a database fragment as part of a transaction to save objects.

The list of objects 141 intended for storage is transferred to the processing cycle 142, in which, for each object O from the input list 141, operations are performed starting from 143 and ending with 152.

At step 143, the storage mode of object O specified in the input list is determined — creating a new object or updating an existing one. If the creation of a new object is requested, then in step 144, a new HO cell is created in the primary index to store object O. This operation is performed atomically. If updating of an existing object is requested, then in step 145, the existing HO cell of object O is searched in the primary index.

At step 146, the O object is atomically added to the revision list of the cell BUT. The structure of the audit list was discussed above in the description of FIG. 6. The list of revisions is unidirectional, the elements follow from the latest revision to the oldest revision, a link to the beginning of the list is contained in the object storage cell. With each element of the list, a transaction is associated within which an object revision was created. The transaction object stores the timestamp with which the transaction was committed. The transaction reference is empty if the transaction has been disposed of, in which case the value of the commit timestamp does not matter anymore, since only those transactions whose commit time is guaranteed to be in the past relative to the timestamp of any client are utilized. Object O is added to the beginning of the list of revisions of the cell BUT, and this only happens if the previous revision is fixed at the time of the operation and does not exceed the revision of object O by number. If the specified condition is not met, this indicates a conflict of the same object updated simultaneously by multiple clients.

At step 147, it is checked whether a transaction conflict has occurred, and if there is no conflict, then proceeds to the next step 148. In the case of a transaction conflict, the cycle 142-153 is interrupted and at step 154 the sign of transaction rollback is set.

In cycle 148-152, object O is added to the secondary indexes defined for object O in the source list. The request to add the object O to some secondary index 8 comes from the node coordinator only if the secondary key of the object O in index 8 falls on the range located on this node. From the coordinating node, a request may be made to exclude object O from secondary index 8 on this node if a newer revision of the object is created in the transaction and placed in a different range of secondary index 8 on another participating node.

The secondary index is of two types - with unique and non-unique keys. At step 149, a check is made of the type of secondary index 8. If it is an index with unique keys, at step 150 it is checked whether there is a conflict of the secondary key of object O with the secondary keys of existing objects in index 8. If a conflict of the secondary key in index 8 is detected, the cycle 148-152 is aborted and at step 154 a sign of transaction rollback is set. If a conflict of the secondary key is not detected, or if the secondary index 8 is built with non-unique keys, at step 151 object O is added to the secondary index 8.

Loop 148-152 is executed for each secondary index 8 requested for object O in source list 141.

At step 155, the sign of successful saving of objects to the database fragment, or the sign of transaction rollback, is returned to the internal query execution kernel 74. In the case of a transaction rollback, the internal query execution kernel 74 completely cancels the operation of saving the list of objects to the database fragment.

Based on the above storage structures of objects and links (Fig. 5 and 6) and the method

- 17 027808 transactional work with objects, consider the method of searching for objects in the database by the values of scalar fields. We will do this by the example of searching and retrieving a certain number of Users and for each User - a certain number of Groups associated with it. A block diagram of the execution of the request is shown in FIG. 11 and FIG. 12.

In this embodiment, the search for objects by the values of scalar fields is index-sequential and requires a secondary index built on these fields in the database. If the required secondary index is built, a search can be performed on it.

At step 160, the kernel for executing client requests 42 of a certain coordinating node N4 receives a request of the form “Find and extract objects” from the client, the parameters of which are the client’s CCN, class and objects to be searched, secondary index δ (in which you should search for objects), the secondary key δυ of the initial object (from which sequential filtering should be performed), which specifies the primary key Ki of the passed object, the filtering condition of the C objects of the class and by the values of scalar fields, the maximum quantity Νυ is extracted objects of class u, the name of the reference field and> 0 for selecting subordinate objects, class O of subordinate desired objects, reference index K (constructed from the field u ^ O and providing the necessary order of the objects of class O in the objects of class i), the condition for filtering CO objects class O according to the values of scalar fields, the limit number N0 of recoverable objects of class O in each object of class and.

At step 161, the transaction manager T creates a reading transaction T with the time stamp Τί ',' в in the list of active transactions 48. The timestamp value is selected so that it is not less than the client’s timestamp, but not more than the node’s timestamp. ВремениCN timestamp is used to read revisions of objects fixed by this time and to discard later data changes. At this step, the client’s logical time is actually moving forward, and the client is slightly behind in time from the system, which in most cases ensures conflict-free reading and editing of data. The amount of client time lag from the system time can be dynamically adjusted depending on the current system load.

At step 162, the kernel for executing client queries 42 selects the secondary index location map δ from the database configuration 46, finds in this map the corresponding secondary key δ and the secondary index start node on which to perform binary search and filtering of objects. In the case of FIG. 11 example, it is assumed that the starting node becomes the node N1. From this node, from the secondary index δ, objects of the class fixed at time ΤCN are requested and satisfying the C condition, in an amount of no more than ΝΗ starting from the object with the secondary key δi and the primary key Ki, skipping this object. To do this, in request 163, the TSS timestamp is used to transfer the class of retrieved objects and, the secondary index δ, the secondary key δ ^ the primary key Ki, the filtering condition C, the limit number of objects ΝΗ

At step 164, the kernel executing internal queries 74 of the participating node by a binary search looks for the first object with the secondary key δ in the secondary index δ. If the primary key Ki is not specified, then filtering of objects using the начинаетсяСN timestamp and condition C starts. If the primary key Ki is specified, by sequential viewing of objects with the key δ and in the secondary index δ, an object with the primary key Ki is found. Immediately after this object, filtering of objects begins using the метCN timestamp and the C condition. The secondary index fragment on this node contains all revisions of all objects in which secondary keys are included in the range of this node. This means that when passing through the secondary index, several revisions of the same object may fall into the sample. To eliminate this, the following filtering method is used. For each revision selected from the secondary index, a check is made to see if it is the revision closest to the ΤCN timestamp. According to FIG. 6, each revision (for example, 1B) contains a link to the storage cell of the object (for example, Ni1), a link to the transaction object (for example, T1) with a time stamp (for example, Τί'Ν1) and a link to the next revision (for example, and 1A). Through the cell and the linked list, all revisions of the object are available. Having a revision, the kernel for executing internal queries accesses the object storage cell, passes through a linked list of revisions and looks for the revision with the closest time stamp to ΤCN, less than or equal to Τί ',' Ν. Then, a check is made to see if this revision matches the revision taken from the secondary index. If not, then the revision taken from the secondary index is skipped. If - yes, it checks that the audit is committed. If the revision is not committed, an attempt is made to commit on the fly. Failure at the fixation stage on the fly causes early termination of operation 164 with a return to the participating node and then to the client of the sign of conflict with another parallel request. In response, the client can wait a while and repeat the request in the hope that there will be no conflict with the new attempt. If the revision is committed, the C filtering condition is applied to it.

The C filtering condition is a Boolean expression over the fields of an object. According to the results of computing this expression, the revision is either included in the result set {No.}, or is skipped. Filtering objects using the метCN timestamp and the C condition stops if the number of elements in the set {Щ} becomes equal to ΝΗ or the end of the fragment is reached

- 18 027808 of secondary index 8 on this node. The resulting set {No.} of the selected objects, the number ΝΝυ of the selected objects, the secondary key Ν8υ of the last element and the primary key ΝΚυ of the last element are returned in the answer 165 to the coordinating node.

At step 166, the client query engine 42 of the coordinator node checks whether the required number of objects has been selected. For this, the value of ΝΝυ is compared with the value of Νυ. If ΝΝυ is less than Νυ, filtering of objects should be continued on the next node in the next range of secondary index 8. If you want to continue the search, in step 167 the value of Νυ decreases by наυ, the secondary key 8υ is assigned the value Ν8υ, the primary key Κυ is assigned the value ΝΚυ and the search query repeated on the next node in the next range of the secondary index (if any). In the example of FIG. 11, the next node for secondary index 8 is node узел2. A search query 168 is transmitted to it. The query parameters 168 fully correspond to the query parameters 163.

At step 169, the next participant node searches in its range of secondary index 8 for the object following the secondary key 8υ and the primary key Κυ. This object corresponds to the beginning of the secondary index range of 8 on this member node. Next, at step 169, filtering of objects is performed, similar to filtering of step 164. As a result, a response 170 is generated that is equivalent to the response 165 described above. This response is transmitted to the coordinator node, which again checks 166 and decides whether to continue searching for objects on next node in the next range of secondary index 8.

If the coordinating node selected the necessary number of objects from the secondary index 8 or all the ranges of the secondary index are scanned, the range polling cycle is interrupted, at step 171 all the selected objects merge into one set {υί} and the one shown in FIG. 12 selection of subordinate objects.

At step 172, the kernel for executing client requests 42 of the coordinator node selects the set of primary keys {Κυί} from the set {υί} of selected objects. For each key from this set, auxiliary objects with such a key will be found on the nodes, subordinate objects will be extracted from them through direct links, transferred to the coordinating node, combined and limited in number. As a result, for each object υί many subordinate objects {О_)} will be formed, which is denoted as υί: {Ο_)}. The set of all objects υί together with their subordinate objects is denoted as {υί: {0 (}}.

At step 173, the kernel for executing client requests 42 of the coordinator node requests on the nodes of the participants objects of class O fixed at time моментCN, to which auxiliary objects with keys {Κυί} are referenced through the field υ ^ Ο. Participating nodes are determined based on the values of the keys {Κυί} and the map for placing the primary index on the nodes. The parameters of the query 174 sent to each relevant participating node include: timestamp ΤΟΝ, keys {Κυί} of auxiliary objects located on the node, name of the reference field υ ^ Ο for selecting subordinate objects, class O of selected objects associated with the field υ ^ Ο reference index K (which should provide the necessary order of viewing and retrieving objects), the condition of SO extraction of objects and the maximum number of ΝΟ recoverable objects.

Next, the internal query engine 74 of each relevant member node performs the same sequence of operations as shown in FIG. 12 in the form of a cycle 175-177. The body of the cycle is represented by block 176. In it, for each key Κυί from the input set {Κυί}, a revision of the auxiliary object υί * fixed at time ΤCN is searched. In this revision, the reference index K is extracted from the field E> C. If this is the first use of this reference index in this revision of the object, then the reference index K is constructed by sorting the primary set of links of the field υ ^ Ο based on the values of the scalar fields of the addressed objects. Through references of the index K, objects of class Ο fixed at the time времениCN that satisfy the condition CΟ are extracted in an amount of no more than ΝΟ. As a result, for each key Κυί, a set of subordinate objects {Ο_ |} is created, which is denoted as Κυί: {Ο_)}. The set of all Κυί keys together with their associated subordinate objects is denoted as {Κυί: {Ο_ |}}. This set is returned to the coordinator node in the response 178 of the member node.

At step 179, the kernel for executing client requests 42 of the coordinating node collects the responses {Κυί: {Ο_ |}} of all involved participating nodes. For each key Κυί, the sets {Ο_ |} obtained from all nodes are combined into one set of objects in the order of objects Ο_) in the index K, then no more than ΝΟ objects are selected from the resulting set, and they are associated with the object υί. This creates the resulting set of objects {υί: {Ο_ |}}. At step 180, reciprocal links are completed in all of these objects.

At step 181, the transaction manager 48 in the list of active transactions is deleted by the transaction manager 48 of the coordinating node T.

At step 182, the resulting object graph {υί: {Ο_)}} is returned to the client along with the new timestamp Τί '.' Ν. The client should pass this timestamp in the next server call.

- 19 027808

In order to ensure long-term storage of objects in a database that is resistant to power outages, fixed objects are stored in a permanent storage 72 (Fig. 3). Since disk storage is usually the storage device in persistent storage, the objects in it are in files, and the objects are stored in persistent storage using the serialization method. There are several modes for saving objects to persistent storage: synchronous saving of objects within a transaction, asynchronous saving of objects outside of transactions as additional insurance in case of a node failure, creating a database snapshot at the request of the database administrator. The mode of saving objects to persistent storage is selected by the database administrator.

In this embodiment, the same method of writing objects to files is used in all modes. In the framework of this method, each database node serializes its fragment of the database into its local persistent storage. A set of logical files containing the entire set of available database objects consists of a snapshot of the database at some point in time (the first logical file), increment to this snapshot to the current point in time (second logical file) and metadata about the snapshot and increment (third logical file). Each logical file can consist of one or more physical files. Metadata is the extreme timestamps of logical data files, the number and size of objects in them, and other statistical data about the objects.

Saving objects to persistent storage is carried out mainly by serializing objects into an increment file. Only committed transactions are serialized. This is done in synchronous or asynchronous mode depending on the installation of the database configuration parameters. If synchronous mode is selected, the serialization of objects on the nodes is carried out as part of the transaction immediately before the transaction is in ready mode. This is the most reliable and least productive mode. It provides reliable long-term storage of objects even if there is no database backup at the node level. If asynchronous mode is selected, objects are serialized periodically along with garbage collection. This is the most productive and less reliable mode. It is used together with database backup at the node level and provides better performance for long-term storage of objects with sufficient reliability.

Saving objects in permanent storage is carried out in parallel by various processes that process customer requests. In order to reduce losses associated with synchronizing access to a single logical increment file, in the considered embodiment of the invention, a separate serialization process and a queue of requests to it are used. Other processes initiate the storage of objects in persistent storage when atomically queuing the serialization requests for saving objects. The serialization process wakes up when requests appear in its queue. It reads all requests and serializes the objects of these requests until the queue is empty. Then it is put on hold until a new request appears in the queue.

Over time, the increment logical file becomes very large, and this can slow down the recovery of the database from files when the node is rebooted. To reduce the time it takes to load a database from files, you need to periodically create a new logical database snapshot file and start creating the incremental logical file again. This can be done manually at the command of the database administrator or in automatic mode upon the occurrence of some event, for example, when the logical file reaches the increment of the specified limit length.

A new logical database snapshot file and a new logical increment file are created in the background parallel to the recording of the previous increment file. At the time of creating a new logical file for the database snapshot, the process of serializing objects simultaneously writes two logical increment files - the old and the new. As soon as a new database snapshot is created, the serialization process switches to writing only the new logical increment file. All actions are reflected in the logical metadata file. When the old logical database snapshot and increment files are no longer needed, they are deleted. The deletion mode can be manual or automatic, depending on the database configuration. The result is a smooth reduction in the size of the serialized database in persistent storage and disk utilization.

In the present embodiment, the so-called garbage is constantly accumulating in the database during the operation of the DBMS - unnecessary objects of completed writing transactions and obsolete revisions of database objects. Among the completed writing transactions, only those objects in which the state of fixation or rollback is set at all nodes of the participants are counted as garbage. Such objects can no longer be needed to determine the status of transactions on any database nodes. Among obsolete revisions of database objects, only those revisions are considered to be garbage in which the commit timestamp is less than the timestamp of any reading client. Such revisions are no longer available to any reading client in the present and future.

Disposal of unnecessary objects is done by the batch process of garbage collection, launched in

- 20 027808 each database fragment of each DBMS node. The parameters of the garbage collection process are two time stamps, denoted as CCNT and C'C'NO. The CCNT timestamp sets the cutoff border for completed writing transactions, and the CCNO timestamp sets the cutoff border for obsolete object revisions. The value of the CCNT timestamp is calculated as the minimum value of the commit or rollback timestamp among all active writing transactions. The CCNO timestamp value is calculated as the minimum value among all timestamps of reading clients.

The garbage collection process works as follows.

From the list of active transactions, committed or canceled transactions are selected in which the timestamp is less than the CCNT timestamp. As a result, a list of utilized transactions is generated. Then, for each transaction from this list, a cycle of operations is performed with the objects included in the transaction. For each object included in the transaction, a search is made for the nearest revision preceding the CCNO timestamp. This revision is considered the oldest available revision of the object; all older revisions are deleted. The garbage collection process is completed by deleting the list of utilized transactions along with the objects included in it.

The timestamps CCNT and CCNO of a database fragment periodically move forward to the minimum values of timestamps CCNT and CCNO in other database fragments at other nodes. The nodes using the auxiliary system process periodically exchange time synchronization messages with each other. In the synchronization process, the values of CCNT and CCNO are calculated for the database fragment of its node as described above. Then, the values of CCNT and CCNO are collected from other nodes from other fragments of the database and placed in two arrays: SSETSC] and CCNO [N], where N is the number of the database node. In the SSETSH] and SSNO [N] arrays, the minimum values are searched. The CCNT and €. '€' N0 timestamps of the native database fragment move forward to these minimum values. The obtained values of SSCG and CCNO are used in the next garbage collection.

A node may contain fragments of several databases. This means that a pair of arrays ССКГЩ] and ССNО [N] exists in each fragment of each database, and there can be many of these pairs on the node. The auxiliary system process of the node updates all pairs of arrays of all fragments of all databases.

The subject invention is a scalable database management system. Scaling means increasing or decreasing the number of database nodes in order to control the availability and size of the database. In the present invention, scaling is performed in the following ways: adding or removing a transaction coordinator, adding a node to a primary index allocation map, adding or removing a node to a secondary index allocation map, adding a node to a file allocation scheme.

All of these methods can be used in any combination. Let us consider them using the example of scaling the system shown in FIG. 13, to the options shown in FIG. 14. General details of all the drawings of the system details: clients 21 sending requests to the client computer network 22 to the request balancing device 23, coordinating nodes 184 receiving client requests from the request balancing device 23 through the client computer network 22, member nodes 183, connected to coordinating nodes 184 through a computer network of nodes 24 and processing commands of coordinating nodes 184. FIG. 13, nodes N4 and N5 work as transaction coordinators, nodes N1 and N2 work as participants in transactions.

Scaling option 185 in FIG. 14 is a system to which the N6 coordinator has been added. The new node N6 is not involved in data storage, so adding a node is very fast - just turn on the new node in the client computer network 22, the computer network of nodes 24 and register the node on the query balancing device 23. Adding a node as a transaction coordinator allows you to redistribute the load created by clients reduce the response time of the system and, as a result, increase the availability of the database. Removing the coordinator node allows you to free up excess resources when the load on the system decreases.

Scaling option 186 in FIG. 14 is a system to which an N3 subassembly has been added.

Scaling option 187 in FIG. 14 is a system to which node N3 has been added, acting simultaneously as a coordinator and a participant in transactions. Adding a member node allows you to increase the size of the database and redistribute the load created by the internal queries of the system. The new node can participate in storing primary index data, secondary index data, file data. Any combination of this application of the new assembly is permissible.

If a node is added to store primary index data, a new distribution of objects is created in the primary index allocation map. A new distribution can only involve a new node or a new node in combination with other nodes. In any case, the redistribution of existing objects among nodes does not occur; the new distribution is valid only for newly created objects.

If the node is added to or removed from the secondary index allocation map,

- 21 027808 redistribution of the secondary index among the database nodes. Depending on the size of the secondary index and the volume of data being moved, the redistribution of the secondary index may take some noticeable time for database clients. During this time, the database may be read-only or unavailable to clients.

If a node is added to store file data, a new rule for placing new files is created in the file allocation scheme. The file allocation scheme is similar to that shown in FIG. 7 of the primary index allocation map with the only difference being that instead of distributions with ranges in the file allocation scheme, rules for distributing files among a set of nodes are set. Each rule includes a list of nodes and a data block size that is continuously recorded on each node and duplicated on a neighboring node (or neighboring nodes) in accordance with the backup level specified in the database configuration. There can be many such rules. The most recent rule is applied to newly created files. The remaining rules are historical and apply to reading and writing files that have already been created.

Thus, all of the above methods for scaling the database in the present invention are implemented.

As a result, it should be obvious to an ordinary person skilled in the art that the system constructed in accordance with the proposed invention is a multi-user distributed data processing system. This system provides atomicity, consistency, isolation and reliability of data operations. At the same time, it provides a high degree of availability, fault tolerance and scalability.

The invention is disclosed in certain embodiments. It should be obvious that the disclosed device can be implemented with modifications without departing from the invention. The appended claims cover all the many variations and modifications that are within the spirit of the present invention.

Claims (1)

CLAIM 1. A database management system consisting of objects distributed among several nodes of a computer network, said system performing transactions on objects and comprising: A) means of distribution and permanent storage of database objects on several nodes in such a way that a separate node contains a subset of objects, moreover, subsets of objects of any two nodes may or may not intersect, and no node is required to store the complete set of all database objects; B) means of managing forward and backward logical links between objects based on the following principle of encoding logical links: ί) for each object, a logical list of backlinks to objects containing direct links to this object is created and maintained up to date; ϊϊ) on the object-storing node, the keys of all objects that refer to the given object are stored, and with each of these keys a list of backlinks to objects located on this node is associated; iii) direct links to objects located on other nodes are replaced by links to nodes from which target objects can be obtained by the key of the object referencing them; C) means of ordering objects according to the values of one or more fields and distributing copies of objects among several nodes in such a way that the whole set of field values involved in ordering is divided into ranges, and each range is assigned a node or several nodes where copies of objects are stored included in ordering in this range; Ό) means for storing and retrieving objects at the request of database clients by sending messages with commands between nodes, and checking whether objects correspond to the query condition is performed on those nodes where copies of objects are stored, without transferring all the index data and copies of objects required to check the query conditions one node coordinating the execution of a client request.