Patent Description:
Massively parallel processing (MPP) database management systems scale by distributing data partitions to servers and running transactions in parallel. A single transaction can be processed in parallel on multiple servers. Such parallel processing presents challenges to transaction management, multi-version concurrency control (MVCC), and recovery.

A global transaction manager (GTM) supports atomicity consistency isolation duration (ACID) compliant transactions in an MPP database. The GTM provides a global transaction identification number (ID) to uniquely identify a transaction in the system. When a transaction involving multiple servers commits, a two-phase commit is conducted to ensure that the processing of the transaction in all the servers has been completed. The GTM also offers a global snapshot of active transactions to support MVCC, a fundamental mechanism to achieve high concurrency, enabling readers to avoid blocking writers, and writers to avoid blocking readers. In MVCC, when a database record is updated, it is not replaced by the updated record. Instead, a new version of the record is created. Both the old and new versions exist in the system, so readers and writers of the same record avoid blocking each other. They can access the right version based on the snapshot taken when a transaction or statement starts, and the transaction IDs stored in the header of the record, representing transactions performing an update. When those updating transactions, such as insert, update, and delete, commit before the snapshot is taken, their versions are visible.

Taking a snapshot and transferring it to servers for each transaction or statement causes the GTM to become a potential performance bottleneck. The visibility check using transaction IDs and transaction status log, such as Clog in PostgreSQL, is often complicated, because time information is not used to determine the occurrence of events.

<CIT> discloses a method, system and program for merging log entries from multiple recovery log files.

<CIT> discloses a log archive filtering method for transaction-consistent forward recovery from catastrophic media failures.

An embodiment method of performing point-in-time recovery (PITR) in a massively parallel processing (MPP) database) comprising a centralized global transaction manager GTM (<NUM>), a plurality of data nodes (<NUM>) and a coordinator (<NUM>), the MPP configured for processing a transaction in parallel across the plurality of data nodes (<NUM>), where the GTM (<NUM>), coordinator (<NUM>) and each data node (<NUM>) run on different servers, where the method includes receiving, by a data node from a coordinator, a PITR recovery request comprising a recovery target, the recovery target comprising a time or a transaction ID generated by a global transaction manager, GTM; reading a log record of the MPP database, the log record comprising a global timestamp acquired from the GTM; determining a type of the log record; and updating a transaction table (<NUM>) based on the recovery target when the type of the log record is an abort or a commit; wherein updating a transaction table (<NUM>) based on the recovery target comprises: setting a timestamp and a state of the transaction table in accordance with the recovery target and the log record.

The foregoing has outlined rather broadly the features of an embodiment of the present invention in order that the detailed description of the invention that follows may be better understood. Additional features and advantages of embodiments of the invention will be described hereinafter, which form the subject of the claims of the invention. It should be appreciated by those skilled in the art that the conception and specific embodiments disclosed may be readily utilized as a basis for modifying or designing other structures or processes for carrying out the same purposes of the present invention.

For a more complete understanding of the present invention, and the advantages thereof, reference is now made to the following descriptions taken in conjunction with the accompanying drawing, in which:.

Global point-in-time recovery (PITR) is problematic in massively parallel processing (MPP) databases, because the data nodes run with their own timers, and a commit request of the same transactions may arrive at data nodes at different times. Using either time or transaction identification number (ID) as a recovery target in PITR does not achieve a consistent recovery point.

An embodiment method uses global transaction manager (GTM) generated global timestamps to achieve read consistency and PITR for MPP databases. <FIG> illustrates MPP database <NUM>. GTM <NUM>, a centralized component, runs on a server. GTM is coupled to data nodes <NUM> to coordinate transactions. Three data nodes are pictured, but many more data nodes may be present. Coordinator <NUM> communicates with GTM <NUM> and data nodes <NUM>. One coordinator is pictured, but more coordinators may be used. Coordinator <NUM> receives a transaction from application <NUM>. Some example applications are a bank and an automated teller machine (ATM), system logging, phone call billing systems, experimental data, and other applications involving large amounts of data. Transactions and their statements register with GTM <NUM>. Operations include start, end, commit, rollback, and abort. The time information from GTM <NUM>, such as the return from gettimeofday(), may become a globally unique tag for transaction operations. For example, if a transaction sends a commit request, GTM <NUM> returns a commit timestamp. When a reader starts its transaction, GTM <NUM> returns begin timestamp or reader timestamp. Those timestamps are stored in a transaction table. In a visibility check, if the reader timestamp is larger than the commit time of update transactions, the record is valid to access.

In addition to the transaction table, the timestamps are stored in log records for PITR. Using the timestamps in the log records, PITR can recover to any history point without setting barriers in logs and blocking transaction commit processes. If the recovery target is a specific time, the recovery aborts all transactions committed after the target time after replaying the log. If the recovery target is a transaction ID, it is translated into a recovery time using the transaction commit time from the transaction table or the commit log.

<FIG> illustrates an example record configuration in an MPP data node. The transaction identification number (TxId) is <NUM>, and the reader timestamp (TM) is RT1. Multiple versions of a record are maintained in the data node. Page <NUM> contains the current version, record <NUM>. A record contains information <NUM>, which contains transaction number <NUM> and version <NUM>. Transaction number <NUM> and version <NUM> are associated with the record. The version <NUM> is the version number of the record, xmin is the transaction number that created the record, and xmax is the transaction number for the transaction which deleted the record and/or created the next version of the record. The old version of the record is moved to undo area <NUM>. Record <NUM>, version three (V3), was created in transaction <NUM>. Because record <NUM> is the current version, xmax is null. The previous version of record <NUM> is record <NUM>. Record <NUM>, version two (V2) was created in transaction <NUM>, with the next version created in transaction <NUM> (record <NUM>). The first version of the record is record <NUM>, version one (V1), created in transaction <NUM>, with the next version created in transaction <NUM>.

Transaction table <NUM> contains the transactions with their current state, running (i), abort (a), or commit (c), and the corresponding timestamp. Only one timestamp for a transaction is recorded in all the data nodes. When a transaction is running, the timestamp represents its beginning time, the reader timestamp. When a transaction is committed or aborted, the timestamp is the time when the commit or abort process completes, the commit or abort timestamp. When a transaction begins, it acquires a timestamp from the GTM, called the reader timestamp, which is also referred to as the beginning timestamp. This transaction should not access the result of any transaction committed after the corresponding timestamp.

As shown in <FIG>, a transaction with transaction ID (TxId) of <NUM> starts and acquires a reader timestamp (RT1). When the transaction accesses a record, its last updater is the transaction with TxId of <NUM>. In one example, the records indicate a bank account, and the transaction represents an operation on the account, such as a deposit or withdrawal. Transaction table <NUM> shows that the transaction with TxId <NUM> has committed. If the commit timestamp is less than RT1, the latest version is visible to the transaction with TxId = <NUM>. If RT1 is less than the commit timestamp, an older version of the record is checked. The earlier version of the record is transaction TxId = <NUM>. This continues until a visible version is found or the record cannot be accessed.

The transaction table can be built locally on the data nodes of an MPP database. When a transaction is changing a record, such as an insert, update, and delete, the transaction registers itself with local transaction managers on the individual data nodes storing the record. Transactions not changing local records do not affect the visibility check.

<FIG> illustrates message diagram <NUM> for performing a transaction on an MPP database. There are three types of servers in an embodiment MPP database: one or more coordinator(s), a GTM, and multiple data nodes. The coordinator receives requests from applications. Based on the accessed data, the coordinator sends messages, such as modified structured query language (SQL) statements, to the data nodes, where the data is stored. Data nodes maintain their own transaction tables for visibility checks. When a transaction begins, the coordinator requests a global transaction ID (GXID) and a reader (or beginning) timestamp from the GTM. The GTM creates a GXID and a timestamp and transmits them to the coordinator. A GXID is not used in read-only transactions. However, read-only transactions do use reader timestamps. The coordinator transmits the GXID and the reader timestamp to the data nodes. Only the reader timestamp is transmitted in a read-only transaction. The data nodes update their local transaction tables with the new GXID and the reader timestamp.

The data nodes perform the transaction and send the results, for example SQL results, to the coordinator. The coordinator forwards the results to the application. The application issues a commit request to the coordinator. The coordinator initiates a <NUM>-phase commit procedure by sending PREPARE COMMIT messages to the data nodes. After collecting the responses, the coordinator requests a commit timestamp from the GTM. The coordinator transmits the commit timestamp, along with a commit request, to the data nodes, which update their local transaction tables with the commit timestamp. The reader timestamp is replaced with the commit timestamp. The commit timestamp is also recorded in the commit log record for use in recovery. The data nodes transmit commit responses to the coordinator. The coordinator sends a commit report to the GTM, and also notifies the application that the commit has been successful. A similar process is conducted for abort transactions. When the coordinator collects responses from the data nodes for PREPARE COMMIT messages, some of the data nodes report commit failed. Instead of sending COMMIT messages, the coordinator sends an ABORT message, and an abort timestamp to the data nodes to abort the transaction. The data nodes receive the abort request and mark the transaction as aborted in their transaction tables with the abort timestamp. In an example, the two phase process is not performed for an abort.

In an example, for read-only transactions, only one message between the GTM and a coordinator is used to acquire a reader timestamp. For read-write transactions, an additional message is used to obtain the timestamp for commit or abort timestamp to update the transaction table. The messages containing one timestamp, which may be <NUM> bytes, are much smaller than the snapshot of a list of active transactions.

The GTM may generate the time using the gettimeofday call in Linux. The time is translated into a <NUM> bit value, representing the number of the total microseconds elapsed since the epoch. The timestamp may be replaced by an ascending sequence number. If the transaction isolation is read committed, the reader timestamp is acquired when the statement starts.

<FIG> illustrates timeline <NUM> for several transactions in an MPP database and their commit orders in coordinator and data nodes. For one transaction, the coordinator and the data nodes may see different commit orders, because they run on different servers. For example, from the coordinator's point of view, transaction T1 committed after transaction T2. Data node <NUM> also saw transaction T1 committed after transaction T2. However, in data node <NUM>, T1 committed before transaction T2. This is because the commit requests sent from the coordinator may arrive at data node <NUM> in a different order than the order they were generated.

A file system backup may be taken before transactions T1, T2, and T3 start. MPP database PITR may restore the system to targets by replaying write-ahead logging (WAL) based on the file system backup. Table <NUM> shows log records to be replayed for recovery targets rec_time1, rec_time2, and rec_time3. Because there are no transactions committed before rec_time1, to recover to rec_time1, there is no log to replay. Transaction T2 was committed before rec_time2, and to recover to rec_time2, only transaction T2 is recovered. To recover to rec_time3, transaction T1 and transaction T2 are both recovered, because transactions T1 and T2 were both committed before rec_time3. Using the target of rec_time3 as an example, PITR will replay log records until transaction T1 is committed and both data node <NUM> and data node <NUM> should perform the recovery. However, even though T2 was committed before T1, they were committed in different orders on the data nodes. On data node <NUM>, T1 was committed before T2. PITR replays log records on data node <NUM> until transaction T1 is committed and the transaction T2's commit log record is not processed. Transaction T2's change is not committed. Data node <NUM> generates inconsistent results with data node <NUM>: data node <NUM> restores both transaction T1 and transaction T2, while data node <NUM> only restores transaction T1.

Global commit timestamps may be recorded in commit log records. These globally unique timestamps are used in PITR to restore the system to a historic point without inconsistency in the data nodes. <FIG> illustrates three recover targets: rec_time1, rec_time2, and rec_time3. Rec_time1 is after T1 and T2 began, and before they are committed. Both data node <NUM> and data node <NUM> compare the commit timestamps of T1 and T2 in commit records with rec_time1 and make the same conclusion, that there is no need for log replay. The target time rec_time2 is between the times when T2 and T1 committed. PITR should recover only T2. With commit timestamps, data node <NUM> can see that T1 has been committed after T2, even though T1's commit log record is generated before T2's commit log recording. After replaying WAL, data node <NUM> and data node <NUM> marks T1 as aborted and T2 as committed. For the target of rec_time3, both T1 and T2 were committed, and their updates should be recovered as committed. T3 began but has not yet committed, so T3 is set as aborted. For recovery targets using a transaction ID, the recovery target can be translated to the transaction's commit timestamp. The recovery follows the same logic using time as the recovery target.

<FIG> illustrates flowchart <NUM> for a method of PITR. Initially, in step <NUM>, a data node receives a PITR recover request, for example from a coordinator. The PITR recovery request includes the target. The target may be a time or a transaction ID.

In step <NUM>, the data node determines whether the target is a time or a transaction ID. When the target is a time, the data node proceeds to step <NUM>, and when the target is a transaction ID, the data node proceeds to step <NUM>.

In step <NUM>, the data node determines whether there are additional logs to be considered for recovery in the data node. When there are no additional logs, the method ends. When there are additional logs, the data node proceeds to step <NUM>, and reads the next record.

Step <NUM> determines whether the record is a commit or abort record. Step <NUM> is a step of determining a type of the log record, which may be an abort transaction or a commit transaction. When the record is not a commit or abort record, the record is replayed in step <NUM>, and the flow returns to step <NUM> to consider additional targets. When the record is a commit or abort, the data node proceeds to step <NUM>.

In step <NUM>, the data node determines whether the record is a commit record or an abort record. When the record is an abort record, the data node sets the transaction in the transaction table to abort in step <NUM>, and returns to step <NUM>. When the record is a commit record, the data node proceeds to step <NUM>.

In step <NUM>, the data node determines whether the commit TM is before the target timestamp. When the commit timestamp is after the target timestamp, the data node proceeds to step <NUM> to set the transaction entry in the transaction table to abort. When the commit timestamp is before the target timestamp, the data node sets the transaction entry in the transaction table to commit in step <NUM>, and returns to step <NUM> to consider additional records.

In step <NUM>, the data node determines whether there are additional logs to examine for recovery. When there are no more logs, the method ends. When there are more logs, the data node proceeds to step <NUM> and reads the next record.

Then, in step <NUM>, the data node determines whether the record is a commit or abort record. When the record is not a commit or an abort record, the record is replayed in step <NUM>, and the data center returns to step <NUM>. When the record is a commit or abort record, the data node proceeds to step <NUM>.

The data node determines whether the transaction ID is the target transaction in step <NUM>. When the transaction ID is not the target transaction ID, the data node sets the transaction timestamp entry in the transaction table in step <NUM>, and proceeds to step <NUM>. When the transaction ID is the target transaction ID, the data node proceeds to step <NUM>.

In step <NUM>, the target transaction is replaced with a commit timestamp. Then, the entry to the transaction table is set to commit in step <NUM>. Next, committed transactions in the transaction table are changed to aborted transaction if their timestamp is after the target transaction's timestamp in step <NUM>. If their timestamp is before the target transaction timestamp, their status stays the same. Then, the data node proceeds to step <NUM>.

<FIG> illustrates flowchart <NUM> for a method of performing a transaction in an MPP database performed by a coordinator. Initially, in step <NUM>, the coordinator receives a transaction request from an application. The application may be a bank looking up the amount of money in a bank account or an ATM requesting a withdrawal of money.

Next, in step <NUM>, the coordinator requests a GXID from the GTM. The GXID is a global transaction ID, a globally unique identifier for the transaction.

In response, the coordinator receives the GXID from the GTM in step <NUM>.

Then, the coordinator requests a timestamp from the GTM in step <NUM>. The reader timestamp is the timestamp at the beginning of the transaction.

In response, the coordinator receives the reader timestamp from the GTM in step <NUM>.

In step <NUM>, the coordinator transmits a transaction request to the data nodes performing the transaction. The transaction requests include the GXID and the reader timestamp.

The data nodes respond to the coordinator with results of the transaction in step <NUM>. The results may be SQL results. Each data node involved in the transaction transmits its results to the coordinator.

The coordinator transmits the transaction results to the application in step <NUM>. The application examines the results.

Next, in step <NUM>, the coordinator receives a commit or abort request from the application. This begins a two phase commit procedure.

In step <NUM>, the coordinator transmits a prepare commit message to the data nodes. This message is sent to the data nodes that performed the transaction.

In response, the coordinator receives prepare responses from the data nodes in step <NUM>. The prepare responses indicate that the data nodes have prepared to commit.

Then, in step <NUM>, the coordinator requests a commit timestamp or an abort timestamp from the GTM based on the responses from the data nodes.

In response, the coordinator receives the commit timestamp from the GTM in step <NUM>.

Next, in step <NUM>, the coordinator transmits commit requests or abort requests to the data nodes. The commit requests include the commit timestamp, and the abort requests include the abort timestamp.

The coordinator, in step <NUM>, receives commit or abort responses from the data nodes. The commit responses indicate that the commit has been successfully performed, while the abort responses indicate that the abort has been successfully performed.

The coordinator notifies the GTM that the transaction has been successfully committed or aborted in step <NUM>.

Also, in step <NUM>, the coordinator notifies the application that the transaction has successfully been committed or aborted. The transaction is now complete.

<FIG> illustrates flowchart <NUM> for a method of performing a transaction in an MPP database performed by a GTM. Initially, in step <NUM>, the GTM receives a GXID request from a coordinator.

Next, in step <NUM>, the GTM creates a GXID. The GXID is a unique identifier that globally identifies this transaction. In one example, the GXIDs are an ascending sequence.

Then, in step <NUM>, the GTM transmits the GXID to the coordinator.

In step <NUM>, the GTM receives a reader timestamp request from the coordinator. This timestamp request is requesting a reader timestamp.

Next, in step <NUM>, the GTM creates the reader timestamp. This may be created, for example, using a gettimeofday call.

Then, in step <NUM>, the GTM transmits the reader timestamp to the coordinator.

The GTM receives a commit or abort timestamp request in step <NUM>.

In step <NUM>, the GTM creates the commit timestamp or an abort timestamp.

Next, in step <NUM>, the GTM transmits the commit timestamp to the coordinator.

Finally, in step <NUM>, the GTM receives a commit report or an abort report from the coordinator. The commit report indicates that the commit has been successfully performed. The abort report indicates that the abort has been successfully performed.

<FIG> illustrates flowchart <NUM> for a method of performing a transaction in an MPP database performed by a data node. Initially, in step <NUM>, the data node receives a begin transaction message from the coordinator. The begin transaction message contains the GXID and the reader timestamp. The begin transaction message also contains the GXID when the transaction is not a read-only transaction.

Next, in step <NUM>, the data node inserts the GXID and the reader timestamp in the transaction table. The data nodes maintain their own transaction tables.

In step <NUM>, the data node performs the transaction. The data nodes may each perform a portion of the transaction in a share nothing configuration. The transaction may be a SQL transaction.

In step <NUM>, the data node transmits the results of the transaction performed in step <NUM> to the coordinator.

The data node receives a prepare commit message or an abort message from the coordinator in step <NUM>.

Then, in step <NUM>, the data node prepares to commit or abort, based on the prepare commit or abort message. To prepare to commit, the data node completes previous requests from the transaction and writes log records generated by the transaction to permanent storage. An abort message instructs the data node to abort the transaction. The abort message includes the abort timestamp. The data node goes step <NUM> to complete the abort. Next, if receiving a prepare commit message, the data node transmits a prepare response to the coordinator in step <NUM>. The prepare response indicates that the data node has successfully prepared to commit.

In step <NUM>, the data node receives a commit message from the coordinator. The commit message instructs the data node to commit the transaction. The commit message includes the commit timestamp.

Then, in step <NUM>, the data node performs the commit or abort. The data node replaces the reader timestamp in the transaction table with the commit timestamp or the abort timestamp. Also, the data node stores the commit timestamp in the transaction log record or the abort timestamp in the transaction log.

Finally, in step <NUM>, the data node transmits a commit response or an abort response to the coordinator. The commit response indicates that the data node has committed its portion of the transaction, while the abort response indicates that the data node has aborted the transaction.

<FIG> illustrates a block diagram of processing system <NUM> that may be used for implementing the devices and methods disclosed herein. Specific devices may utilize all of the components shown, or only a subset of the components, and levels of integration may vary from device to device. Furthermore, a device may contain multiple instances of a component, such as multiple processing units, processors, memories, transmitters, receivers, etc. The processing system may comprise a processing unit equipped with one or more input devices, such as a microphone, mouse, touchscreen, keypad, keyboard, and the like. Also, processing system <NUM> may be equipped with one or more output devices, such as a speaker, a printer, a display, and the like. The processing unit may include central processing unit (CPU) <NUM>, memory <NUM>, mass storage device <NUM>, video adapter <NUM>, and I/O interface <NUM> connected to a bus.

The bus may be one or more of any type of several bus architectures including a memory bus or memory controller, a peripheral bus, video bus, or the like. CPU <NUM> may comprise any type of electronic data processor. Memory <NUM> may comprise any type of system memory such as static random access memory (SRAM), dynamic random access memory (DRAM), synchronous DRAM (SDRAM), read-only memory (ROM), a combination thereof, or the like. In an embodiment, the memory may include ROM for use at boot-up, and DRAM for program and data storage for use while executing programs.

Mass storage device <NUM> may comprise any type of storage device configured to store data, programs, and other information and to make the data, programs, and other information accessible via the bus. Mass storage device <NUM> may comprise, for example, one or more of a solid state drive, hard disk drive, a magnetic disk drive, an optical disk drive, or the like.

Video adaptor <NUM> and I/O interface <NUM> provide interfaces to couple external input and output devices to the processing unit. As illustrated, examples of input and output devices include the display coupled to the video adapter and the mouse/keyboard/printer coupled to the I/O interface. Other devices may be coupled to the processing unit, and additional or fewer interface cards may be utilized. For example, a serial interface card (not pictured) may be used to provide a serial interface for a printer.

The processing unit also includes one or more network interface <NUM>, which may comprise wired links, such as an Ethernet cable or the like, and/or wireless links to access nodes or different networks. Network interface <NUM> allows the processing unit to communicate with remote units via the networks. For example, the network interface may provide wireless communication via one or more transmitters/transmit antennas and one or more receivers/receive antennas. In an embodiment, the processing unit is coupled to a local-area network or a wide-area network for data processing and communications with remote devices, such as other processing units, the Internet, remote storage facilities, or the like.

While several method embodiments have been provided in the present disclosure, a computer program (product) is provided, which includes program instructions for causing a computer system to perform any of the methods or steps operated in GTM. Corresponding, the apparatus functioning as a GTM is also provided, which includes means operable to perform any of the methods or steps operated in GTM.

While several method embodiments have been provided in the present disclosure, a computer program (product) is provided, which includes program instructions for causing a computer system to perform any of the methods or steps operated in a coordinator. Corresponding, the apparatus functioning as a coordinator is also provided, which includes means operable to perform any of the methods or steps operated in coordinator.

Claim 1:
A method of performing point-in-time recovery (PITR) in a massively parallel processing (MPP) database (<NUM>) comprising a centralized global transaction manager GTM (<NUM>), a plurality of data nodes (<NUM>) and a coordinator (<NUM>), the MPP configured for processing a transaction in parallel across the plurality of data nodes (<NUM>), where the GTM (<NUM>), coordinator (<NUM>) and each data node (<NUM>) run on different servers, the method comprising:
receiving, by a data node (<NUM>) from the coordinator (<NUM>), a PITR recovery request comprising a recovery target, the recovery target comprising a time or a transaction ID generated by the GTM (<NUM>);
reading a log record of the MPP database, the log record comprising a global timestamp acquired from the GTM (<NUM>);
determining a type of the log record; and
updating a transaction table (<NUM>) based on the recovery target when the type of the log record is an abort or a commit; wherein updating a transaction table (<NUM>) based on the recovery target comprises:
setting a timestamp and a state of the transaction table in accordance with the recovery target and the log record.