Asynchronous deletion of a range of messages processed by a parallel database replication apply process

Mechanisms are provided for replicating transactions at a source database node on a target database node of the data processing system. A transaction message, for a transaction, is stored in a receive queue data structure of the target database node. An agent thread applies a change to data of the target database node based on a specification of the change in the transaction message. An identifier of the transaction message is stored in a done message data structure of the target database node and the identifier of the transaction message is also stored in a prune queue data structure of the target database node. A prune thread determines if a contiguous range of identifiers of transaction messages equal to or greater than a predetermined batch size is present in the prune queue data structure. If so, then a range of transaction messages is deleted from the done message data structure.

BACKGROUND

The present application relates generally to an improved data processing apparatus and method and more specifically to mechanisms for asynchronously deleting ranges of messages processed by a parallel database replication Apply program that does not require coordinated commits between the message delivery mechanism and a target database.

In a relational database management system, data is stored in a multiplicity of tables having a multiplicity of rows (records), the rows having a multiplicity of columns (fields). A subset of the columns are designated as key columns and the combination of values of the key columns of the rows of a single table must be distinct. It is frequently desired to maintain copies (replicas) of a first table residing in a first database of the relational variety in one or more other databases of the relational variety. Furthermore, it is desired that changes (inserts, deletes, and updates) to rows of the table in the first database be copied (replicated) to the table copies residing in the other databases. Additionally, it is sometimes desired that the changes made to any of the table copies residing in any of the several relational databases be copied (replicated) to all the other table copies.

The propagation of changes made to one copy of the table may be synchronous or asynchronous to the original change. Synchronous propagation makes changes at all copies as part of the same transaction (i.e. an atomic unit of work) that initiates the original changes. Asynchronous propagation copies the original changes to the other table copies in separate transactions, subsequent to the completion of the transaction initiating the original changes. Synchronous change propagation requires that the database management systems maintaining all (or most) copies be active and available at the time of the change. Also, synchronous change propagation introduces substantial messaging and synchronization costs at the time of the original changes.

The means of detecting changes to be propagated asynchronously can be active or passive. Active change detection isolates the changes, at the time of the change, for later processing using database triggers or a similar mechanism. Passive change detection exploits information from the database recovery log, where changes are recorded for other purposes, to deduce what rows of which tables were changed as well as both the old and new values of changed columns.

In a typical database environment, there are varying levels of parallel transactional processing involving concurrent transactions that execute read write actions against database information. Fundamental to the nature of a data replication process is the choice of how to move, order and apply that stream of parallel database event changes to a target database.

One conventional approach provides a certain degree of “apply” parallelism by grouping related tables into distinct sets and having each set of tables applied by a completely separate program. However, this approach places a heavy burden on the user. The user may have difficulty knowing which tables are logically related and must be grouped together.

In another conventional approach, parallelism is provided but without preserving the source data event order. Thus, to provide data integrity, a “shadow” table is used to track and maintain each individual data row change. This approach, however, has a significant overhead cost in both making updates and in performing lookups against the shadow table. Other conventional approaches provide parallelism but by using a very proprietary way that has no or limited applicability outside of a specific system.

SUMMARY

In one illustrative embodiment, a method, in a data processing system, is provided for replicating transactions at a source database node on a target database node of the data processing system. The method comprises receiving, in a receive queue data structure associated with a browser thread, a transaction message from the source database node and applying, by an agent thread, a change to data of the target database node based on a specification of the change in the transaction message in the receive queue data structure. The method further comprises storing an identifier of the transaction message in a done message data structure associated with the target database node and storing, by the browser thread, an identifier of the transaction message in a prune queue data structure associated with the target database node, in response to the transaction message being placed in the done message data structure. Moreover, the method comprises deleting, by a prune thread executing on the target database node, a batch of one or more completed transaction messages from the receive queue regardless of an order or sequence of the completed transaction messages in the receive queue. Furthermore, the method comprises determining, by the prune thread, whether a contiguous range of identifiers of transaction messages equal to or greater than a predetermined batch size is present in the prune queue data structure. In addition, the method comprises deleting a range of transaction messages from the done message data structure in response to a contiguous range of identifiers of transaction messages equal to or greater than the predetermined batch size being present in the prune queue data structure.

In other illustrative embodiments, a computer program product comprising a computer usable or readable medium, such as a computer usable or readable device, having a computer readable program is provided. The computer readable program, when executed on a computing device, causes the computing device to perform various ones, and combinations of, the operations outlined above with regard to the method illustrative embodiment.

DETAILED DESCRIPTION

The illustrative embodiments provide an efficient mechanism for asynchronously deleting ranges of messages processed by a parallel database replication Apply program that does not require coordinated commits between the message delivery mechanism and a target database. The illustrative embodiments detect contiguous ranges of messages that have been processed by a parallel Apply program to aggressively delete the entries from the done message (DONEMSG) data structure (hereafter referred to as the DONEMSG table), thereby keeping the DONEMSG table small in size, i.e. having a small number of entries (or rows if one considers the DONEMSG table to be a table data structure) and thus, providing improved database performance. Moreover, the illustrative embodiments provide mechanisms for immediately deleting messages from the receive queue after they have been processed, thereby keeping the receive queue at its minimum size for better performance of the queuing system. Except during restart, the Apply program does not need to query the DONEMSG table to determine which messages have been processed. The mechanisms of the illustrative embodiments allow pruning of messages from queue structures without introducing additional overhead into the database replication process.

In one illustrative embodiment, mechanisms are provided for replicating transactions at a source database node on a target database node of the data processing system. The mechanisms capture one or more transactions at a source database node of the data processing system and send the transactions to the target database node using a queuing mechanism. A browser thread reads the transaction messages in order and queues non-dependent transactions for parallel apply in a work queue data structure where a pool of agents apply the transactions to the target database node in parallel. Each agent inserts a unique message identifier in the DONEMSG table in the same commit-scope as the transaction applied to the target database. An agent sends each committed transaction message in a done queue data structure to the browser thread and the browser thread sends an identifier of the transaction message in a prune queue data structure to the prune thread. The prune thread immediately deletes the messages from the queue mechanism resulting in messages being deleted in the order in which they were committed at the target (which is different than the order in which these messages arrived from the source). The deletes are batched for optimal performance. A prune thread executing on the target database node determines if a contiguous range of identifiers of transaction messages equal to or greater than a predetermined batch size is present in the prune queue data structure. A contiguous range of transaction messages are deleted from the DONEMSG table in response to a contiguous range of identifiers of transaction messages equal to or greater than the predetermined batch size being present in the prune queue data structure.

An operating system runs on processing unit206. The operating system coordinates and provides control of various components within the data processing system200inFIG. 2. As a client, the operating system may be a commercially available operating system such as Microsoft Windows 7 (Microsoft and Windows are trademarks of Microsoft Corporation in the United States, other countries, or both). An object-oriented programming system, such as the Java programming system, may run in conjunction with the operating system and provides calls to the operating system from Java programs or applications executing on data processing system200(Java is a trademark of Oracle and/or its affiliates).

As a server, data processing system200may be, for example, an IBM® eServer™ System p® computer system, running the Advanced Interactive Executive (AIX®) operating system or the LINUX operating system (IBM, eServer, System p, and AIX are trademarks of International Business Machines Corporation in the United States, other countries, or both, and LINUX is a registered trademark of Linus Torvalds in the United States, other countries, or both). Data processing system200may be a symmetric multiprocessor (SMP) system including a plurality of processors in processing unit206. Alternatively, a single processor system may be employed.

Referring again toFIG. 1, one or more of the depicted computing devices inFIG. 1, e.g., the server computing device104and/or106, may provide a relational database system having one or more relational databases that are managed by one or more relational database management systems that, among other things, manages replication of changes to databases within the relational database system, i.e. implements a database replication system. The relational database system may store in a plurality of tables having a plurality of rows (records), the rows having a plurality of columns (fields). The relational database management system(s) maintain copies (replicas) of a first table residing in a first relational database of the relational database system, in one or more other second relational databases. Furthermore, the relational database management system(s) replicates changes (inserts, deletes, and updates) to rows of the table in the first relational database, to the copies of the table residing in the other second databases. Moreover, the relational database management system(s) replicate changes made to any of the table copies residing in any of the several relational databases to all the other table copies.

In accordance with the mechanisms of the illustrative embodiments, in addition to replicating changes to databases in other databases as mentioned above, the illustrative embodiments provide mechanisms for managing messages used to perform the replication of changes to databases. These mechanisms for managing messages include mechanisms for asynchronously performing deletion of ranges of messages processed by a parallel database replication Apply mechanism. The mechanism for asynchronously performing the deletion does not require coordinated commits between the message delivery mechanism and a target database. To better understand the way in which the mechanisms of the illustrative embodiments operate, it is best to first understand an example of a relational database system that implements a database replication system using one or more relational database management system(s).

FIG. 3illustrates an example of a system for providing parallel “apply” operations in asynchronous data replication in a database system. The example system shown inFIG. 3is a database replication system as described in commonly owned and co-pending U.S. patent application Ser. No. 12/049,196 entitled “Parallel Apply Processing in Data Replication with Preservation of Transaction Integrity and Source Ordering of Dependent States,” filed Mar. 14, 2008. The system includes a source node301and a target node306. These nodes may be computing devices, such as servers104and106inFIG. 1. At the source node301are one or more source table copies302, a recovery log303, a Capture program304(“Capture”), and a send queue305. At the target node306are a receive queue307, an Apply program (“Apply”)308and one or more target table copies312. Apply308includes a browser thread309, a work queue310, a done queue311, and one or more agent threads312. Capture304reads changes of committed transactions from the recovery log303and sends them to Apply308running on the target node306. Apply308eventually re-executes the changes of the transactions.

The tabular data at the source table copies302whose changes are to be replicated to the target table copies313reside in a Relational Database Management System (RDBMS) such as the DB2® RDBMS product offered by International Business Machines Corporation. The RDBMS maintains a recovery log303and a means to query its contents. The entries of the recovery log303describe changes to rows of the source tables302at source node301. More specifically, the entries in the recovery log303contain information defining (1) the table being changed, (2) the value of the key column in the row being changed, (3) the old and new values of all columns of the changed row, and (4) the transaction (unit of work) containing the change. Recovery log entries for inserted rows contain only new column values while recovery log entries for deleted rows contain only old column values. Recovery log entries for updated rows contain the new and old values of all row columns. The order of entries in the recovery log303reflect the order of change operations within each transaction and the order of transaction commit records reflects the order in which transactions are completed. The format of a row change log record can be abstracted as follows:

To control the propagation of changes to table copies, copy control tables (not shown) designating table copies and their target table copies are used by the replication system. The control information includes, but is not limited to: (1) the name of the copied table, (2) a list of the table copies' key columns, (3) filtering and projection information, and (4) the message channels on which to send descriptions of changes to the target table copies.

The list of key columns defined for a replication definition will be hereafter referred to as the “replication key”. The replication key should not be confused with other attributes of source or target table copies which may use primary key columns or foreign key columns. However, it is possible that the primary key of a source or target table copy may be comprised of the same set of columns as are specified for the replication key. The replication key uniquely identifies a row entity in a target table copy so that it can be located by Apply, in applying an update or delete change operation. Because the replication key uniquely identifies a row entity, it is used in the serialization of changes made to these unique row entities.

The type of row operation in change log records can be delete, insert, update, or key update. Updates that do not modify the replication key (update) are distinguished from updates that do modify the replication key (key update).

The changes made to table copies are determined by reading the recovery log303. Changes are saved in memory until a transaction commit record is seen on the recovery log303. Only committed transactions at the source node301are moved and applied to target nodes306. Change records are grouped into their originating source transaction units and written as one logical message unit. Because a logical transaction message can be quite large, it may be broken down into a plurality of physical messages. In this specification, a “transaction message” refers to a logical transaction message. Changes to be sent to the other table copies are sent via logical message units on the recoverable queues (e.g., send queue305and receive queue307) designated in the copy control tables for the table copies of the log records.

The transactions messages are put on the recoverable queue in the source commit order. Within each transaction, the change records are arranged in the order in which they occurred within the source transaction. In such a case, there is no inherent parallelism in the movement of the committed transactional data. The queuing of the transactional data is serialized such that data is moved to the target node306in the source transactional commit order.

In capturing the information for individual change records, the type of change operation for each change determines what replication key column values will be sent as part of that change record. For insert and update types of change records, the new replication key column values are sent as part of the change records within the transaction message. By definition, an insert is a new record and therefore has no old values. By definition, the new replication key column values of an update type of change record must be the same as the old replication key column values. For delete type change records, there is by definition no new record, only an old record, and therefore the old replication key column values are sent. For key update records, the old replication key column values are sent in addition to the new replication key column values.

For any given receive/recoverable queue307that is populated with transactions from a given source node301and is to be used as the source of changed data to be applied to a given target node306, Apply308has a browser thread309and one or more agent threads312, where the number of agents is determined through user input. The work queue310and the done queue311are structures internal to Apply308, and are created for the purpose of communication between the browser thread309and the agent threads312.

FIG. 4is a flowchart illustrating an example operation for providing parallel apply in asynchronous data replication in a database system. First, the browser thread309examines the next transaction message in the receive queue307(step401). The values of the replication key columns for each row change in the transaction message are stored in a logical data structure, such as in memory or the like (step402). In this case, information describing the transaction, including the values of the replication key columns, is stored in a logical data structure and tracked. Other information concerning the transaction may also be stored. The logical data structure may also be used to track any preceding non-completed transaction messages, including any subsequent transaction messages that are dependent upon it.

The browser thread309determines if the transaction message has dependencies (step403). A transaction message has a dependency if the preservation of the integrity of the data requires that one or more preceding non-completed transaction messages be applied prior to the application of the current transaction message. If the transaction message has dependencies, then the browser thread309checks the transaction messages on the done queue311to see if the completion of any of those transaction messages clears the dependencies (step404). If not, then non-completed transaction messages upon which the transaction message is dependent are marked to indicate the transaction message's dependency (step405). The current transaction message is also marked with its dependencies and held (step406), and not allowed to be applied. If it does not have any dependencies, then the transaction message can be applied in parallel with the preceding transaction(s) currently being applied, and is thus placed on the work queue310(step407). Once placed on the work queue310, the transaction message becomes eligible to be applied by any available agent thread312. The more agent threads312that are made available to be used, the more transaction messages which are eligible for application can be applied in parallel.

Each of a plurality of agent threads312look on the work queue310, each removes a transaction message from the work queue (step408). Each agent thread312then applies the row changes in the transaction message to the target table copies313in parallel with each other (step409). All row changes from a transaction message are applied as a transaction unit, and are committed as a unit. In this case, committed as part of this transaction is an update of a control table to indicate that this transaction has been successfully committed at the target table copy313(step410). The update is an insert of an entry into the control table (also referred to as the DONEMSG table) for the completed transaction. When the logical transaction message comprises a plurality of physical transaction messages, a plurality of entries, one for each physical transaction message, can be inserted. A control table in the same relational database as the target table copies313is used in order to provide for the best performance of this transaction application, while at the same time, keeping a permanent record of the successful application of the transaction. The insert to the control table is important for message cleanup of the receive queue307, as described hereafter.

In one case, application of the changes is performed using generated Structured Query Language (SQL) statements of a non-proprietary nature. These SQL statements may or may not be exactly the same as the originating SQL statements made at the source node301. However, the net effect of these changes is typically identical to the net effect of the changes made by the originating SQL statements. For example, an originating SQL statement such as “DELETE FROM SOURCE.TABLE” could be made. This statement would have the effect of deleting all rows from the table named SOURCE.TABLE. If there were five rows in the table at this point in time, then there would be five rows deleted, and five log records would be generated on the recovery log. Each log record would indicate the delete operation of one of the five rows. From the inspection of the recovery log, the five operations would be used to capture the information of five distinct data events, all of which occurred during a single transaction. This transaction would be queued and moved to the target node306, and the application of these changes would be made as five distinct SQL statements, with each SQL statement targeting one of the individual rows of the corresponding target table copy. At the commit point of this applied transaction, the functional equivalence point is then reached, such that the same five rows have been deleted from the corresponding source and target table copies. Thus, a non-proprietary implementation of Apply is provided with this mechanism. This mechanism can be extended for use in any database that accepts standard SQL and has the general database property of atomicity.

Once the application is complete, the transaction message is placed on the done queue311(step411). The indicators of held transaction messages dependent on this now completed transaction message, if any exist, which were previously marked (via step405) can now be checked (step412). These held transaction messages will be changed to remove the dependency or dependencies that existed regarding the now completed transaction message (step413). After removal of these dependencies, each of the held transaction messages are checked to see if any other dependencies remain (step414), against other preceding still non-completed transaction messages. Any held transaction message that is now determined to be dependency free (step414), can be safely applied in parallel with the other transaction messages currently being applied, and thus placed on the work queue310(step407). For held transaction messages with remaining dependencies, they remain as held transaction messages.

FIG. 5is a flowchart outlining, in more detail, the determination of dependencies in the operation for providing parallel apply in asynchronous data replication in a database system. For every transaction message that the browser thread309examines, critical pieces of information regarding that transaction are assessed and tracked. For each row change that makes up the transaction message, information regarding the values of the replication key columns is noted and tracked as part of that transaction. From the time of the initial examination of a transaction by the browser thread309until the eventual placement of that transaction message on the done queue311after successful application, the replication key column information for every row change within this transaction message is used to assess newly arriving transactions, to determine their eligibility for placement on the work queue310. If a newly assessed transaction message contains row changes with replication key column values that match the values of the replication key columns from row change of any preceding transaction messages that have not yet completed, then this newly assessed transaction message is not eligible yet for application and must not yet be placed on the work queue310.

As illustrated inFIG. 5, the browser thread309examines a transaction message in the receive queue307(step501). The transaction message can contain a plurality of row changes. For each of the row changes, steps502through512are performed. The browser thread309examines the next change in the transaction message (step502). If the type of change is an insert or key update (step503), then the browser thread309determines if the new replication key value of the insert or key update change is the same as the old replication key value of any preceding non-completed transaction messages (step504). If they are the same, then the preceding non-completed transaction message is marked to indicate the transaction message's dependency, and the transaction message is marked to indicate the preceding non-completed transaction message upon which it depends (step505).

The new replication key column values of an insert or key update type of row change represent the introduction of a new row entity. Either of these row actions could have been preceded by a delete of that row entity (carrying old replication key column values) or by a key update which had the net effect of a delete followed by an insert, where it would be the delete aspect of the prior row action that could potentially have commonality with this row action and is therefore of interest. Therefore, the new replication key column values of an insert or key update row change are compared to the old replication key column values of all preceding non-completed transaction messages.

The method by which it is determined that a new or old replication key value is the same as another new or old replication key value can be relaxed so long as the same replication key values are not determined to be different. Those with ordinary skill in the art at the time of the invention will recognize that the comparison of the result of any deterministic function (e.g., a hash code function) can be used to insure that identical replication key values are matched, while differing replication key values may be incorrectly matched. The performance benefits of simplified comparing can outweigh the loss of parallelism due to incorrectly matched replication key values.

If the type of change is a delete or a key update (step506), then the browser thread309determines if the old replication key value of the delete or key update change is the same as the new replication key value of any preceding non-completed transaction message (step507). If they are the same, then the preceding non-completed transaction message is marked to indicate the transaction message's dependency, and the transaction message is marked to indicate the preceding non-completed transaction message upon which it depends (step508).

The new replication key column values of an update type of row change represent the change of non-replication key column values of an existing row entity. This row action could have been preceded by an insert of that row entity (carrying new replication key column values), or by a key update which had the net effect of a delete followed by an insert, where it would be the insert aspect of the prior row action that could potentially have commonality with this row action and is therefore of interest. Therefore, the new replication key column values of an update row change are compared to the new replication key column values of all preceding non-completed transaction messages.

If the type of change is an update (step509), then the browser thread309determines if the new replication key value of the update change is the same as the new replication key value of any preceding non-completed transaction message (step510). If they are the same, then the preceding non-completed transaction message is marked to indicate the transaction message's dependency, and the transaction message is marked to indicate the preceding non-completed transaction message upon which it depends (step511).

The old replication key column values of a delete or key update type of row change represent the deletion of an existing row entity. Either of these row actions could have been preceded by an insert of that row entity (carrying new replication key column values), by an update of that row entity (carrying new replication key column values), or by a key update which had the net effect of a delete followed by an insert, where it would be the insert aspect of the prior row action that could potentially have commonality with this row action and is therefore of interest. Therefore, the old replication key column values of a delete or key update row change are compared to the new replication key column values of all preceding non-completed transaction messages.

Once the last change in a transaction message has been examined (step512), and the transaction message is determined to have dependencies (step513), the process continues with step404inFIG. 4. If the transaction message is determined to have no dependencies, then the process continues with step407inFIG. 4.

With these mechanisms, whole source transactions are executed as whole target transactions, and changes to any individual table row entity, as determined by the specified and required replication key column values, are serialized to the same degree that those changes were serialized at the source database. Transactions with no dependencies are likely to be committed in a different order from the source commit order.

FIG. 6illustrates an example of the method for providing parallel apply in asynchronous data replication in a database system. The transaction data found in the recovery log303is grouped by transaction and those transactions are sent to the send queue305in source commit order. For example, transaction1(Tx1), transaction2(Tx2), and transaction3(Tx3) were started in Tx1-Tx2-Tx3order, but were committed in Tx1-Tx3-Tx2order. Thus, they are sent to the receive queue307in committed Tx1-Tx3-Tx2order.

When Tx1arrives on the receive queue307, the browser thread309examines Tx1(step401). Information concerning Tx1is stored (step402). Such information includes the fact that Tx1involves an insert into table T1of a row with replication key value=1. Since there are no preceding transactions, Tx1has no dependencies (step403). Tx1is thus placed on the work queue (step407).

As Tx1is removed from the work queue (step408), and being applied (step409), the browser thread309examines Tx3(step401). Information concerning Tx3is remembered (step402). Such information includes the fact that Tx3involves a delete from table T1of a row with replication key value=1 and an insert into table T1a row with replication key value=2. The browser thread309determines that Tx3has a dependency for table T1delete, since the old replication key value of the delete (key=1) is the same as the new replication key value for the insert in Tx1(step507). Assuming that Tx1has not yet completed, there are no transaction messages on the done queue311so steps404and405are not performed. Tx1is thus marked to indicate the dependency of Tx3, and Tx3is marked to indicate it is dependent upon Tx1(step508). Tx3is held (step406).

The browser thread309next examines Tx2after it arrives on the receive queue307(step401). Information concerning Tx2is remembered (step402). Such information includes the fact that Tx2involves an update in table T2of a row with replication key=1, and an update in table T2of a row with replication key=3. The browser thread309determines that Tx2has no dependencies (step403) (and step510), and places Tx2on the work queue310(step407).

When application of Tx1completes (step409), the control table is updated to indicate its completion (step410). Tx1is also placed on the done queue311(step411). From the marks added to Tx1above, the browser thread309knows to remove from Tx3its dependency upon Tx1. The browser thread309then checks if Tx3is now dependency free (step412). Since Tx3is now dependency free, it is placed on the work queue (step407).

In this case, the receive queue307is a persistent queue, while the work queue310and the done queue311are not. The persistence of the receive queue307is to protect the integrity of the data in case of a system failure or some other interruption during the transaction application process. However, the persistent nature of the receive queue307requires that messages in the receive queue307be removed after transactional messages have been successfully applied. Otherwise, if the process is interrupted, the system upon restart will attempt to apply the changes in the transaction messages on the receive queue307again, leading to errors.

One possible method of removal is a two-phase commit approach, where the delete of the message from the receive queue307is committed as part of the same transaction at the target node306that applies the changes. Another method is to use an asynchronous “cleanup” approach, as described below. The asynchronous cleanup approach has the advantage of defraying the delay and overhead costs associated with the two-phase commit approach.

In the asynchronous cleanup approach, it is noted that a control table is updated and committed as part of the transaction that applies the changes associated with a logical replication transaction message at a target node306. This allows for a background task to be executed on a periodic basis which deletes messages from the receive queue307based on the existence of an entry in the control table indicating that this message has been successfully applied. After the delete of one or more logical transaction messages from the receive queue307has been committed, entries for the logical transmission message from the control table can be safely removed. If the logical transaction message comprises a plurality of physical transaction message, then each physical transaction has its own entry in the control table. Each entry for the physical messages is individually removed. This approach avoids the cost of a two-phase commit since the control table rows are deleted after the committed delete of the messages on the receive queue307. If entries in the control table exist without corresponding queue messages because those messages have already been deleted due to some process interruption, this poses no possible harm to the system, and such extra control table rows can be safely removed at anytime.

Referring again toFIG. 3, using the database replication system described above as an example, message queues, e.g., send queue305and receive queue307, are used to deliver database transactions that are captured at a source database system301and applied at a target database system306. The message queue structures305and307reside in a queuing mechanism that is outside of the database management system (DBMS). The queuing mechanism provides the ability to stage the messages.

The messages in the message queue structures must be deleted after the changes are committed at the target database. For performance reasons, as noted above, a coordinated two-phase commit protocol is not used between the queuing mechanism and the target database. For example, when a two-phase commit protocol is used, it has been observed that the Apply program308can take up to 10× more time to apply the replicated transactions. Instead, the done message (DONEMSG) data structure, also known as the DONEMSG table,314is used to maintain the identifiers of the message already processed. Messages are processed in parallel by the Apply program308using multiple Apply agent threads312, each processing one or more messages. If the Apply program308is stopped and restarted, it is possible to determine which messages have been processed by reading the entries in the DONEMSG table314. The Apply program308tolerates messages that have an entry in the DONEMSG table314, but have already been deleted from the receive queue307.

The Apply program308processes can be aborted at any time. However, a message that was previously processed by the Apply program308must not be processed again if and when the Apply program308is restarted. The Apply program308can only delete a message from the receive queue307after the changes this message contained have been committed to the target database's table copies313.

Messages arrive in the receive queue307in the commit order of the changes at the source database system301. However, because messages are processed in parallel using the multiple Apply agent threads312, and the transactions take variable amount of time to complete, and non-dependent transactions can be completed in any order, the messages to delete from a queue, such as receive queue307, are not contiguous.

The messages contain a sequence identifier that is generated by the capture program304and which is used by the Apply program308to detect lost messages. The sequence identifier includes at least a sequence time (timestamp) and a monotonically increasing sequence number. A new sequence time is generated by the Capture program on the source database system301whenever replication is reinitialized. Replication may be reinitialized when the user wants to reset both sides (the source database system301and target database system306) and start-over. When a new sequence time is generated, the sequence number is reset to 1. The Apply program308checks that messages are contiguous and increasing based on the sequence identifier. A maximum message identifier is retained in the DONEMSG table314for each receive queue307. The maximum message identifier is read by the browser thread309on startup of the browser thread309. This assists the browser thread309to start processing from a correct point in the receive queue307.

When pruning from the DONEMSG table314, i.e. when deleting messages from the DONEMSG table314, the Apply program308can delete all rows from the DONEMSG table314except the row containing the maximum message identifier (Max MSGID). This is done so that the Apply program308knows the restart point for reading from the DONEMSG table314after the Apply program308has successfully deleted the corresponding messages from the receive queue307.

In a high volume database replication system, such as that described above, it is not uncommon to replicate hundreds of millions of transactions a day. There is very high activity on the receive queue307and the DONEMSG table314. A delay of a few minutes in pruning messages from these queues may result in the accumulation of millions of messages as well as the DONEMSG table314growing to millions of entries, resulting in performance degradation that can be compounding with the constant arrival of new messages. Slow pruning can lead to the receive queue307filling up, which can force the replication process to stop. Performance degrades as the queues grow larger because it introduces more processor, memory, and input/output overhead in the queuing mechanism.

Queries against the DONEMSG table314to determine if a message has already been applied become more expensive when the DONEMSG table314grows larger. Pruning the DONEMSG table314is costly if done one row at a time. In addition, a low-overhead mechanism is desirable for efficiently and aggressively keeping the receive queue307and DONEMSG table314to a minimum size without introducing overhead into the replication process. It is also highly desirable to keep a minimum number of processed messages in the receive queue307.

The illustrative embodiments provide such mechanisms by providing an efficient method of pruning the DONEMSG table314, and other queue structures, which leverages database set operations. This pruning id done asynchronously using batch operations so as to avoid introducing overhead into the replication process. Moreover, to keep a minimum number of processed messages in the receive queue, messages are pruned from the receive queue307immediately after they have been processed. It can be determined that a message has been processed after the insertion of its message identifier (msgid) has been committed into the DONEMSG table314.

Keeping the DONEMSG table314small allows faster restart of the replication process. On restart, the Apply program308determines where to start reading the receive queue307, skipping already processed messages. The Apply program308on restart first deletes all already processed messages and their corresponding entry in the DONEMSG table314. The larger the DONEMSG table314, the longer the restart time.

Operations against the target database system306are costly and should be minimized. The Apply program308should not have to query the DONEMSG table314to determine which messages can be deleted except during recovery upon restart. Pruning of the DONEMSG table314should be done using a minimum number of Structured Query Language (SQL) operations. For example, a single SQL statement that deletes a large number of contiguous rows, for which the database management system will be able to use an index to retrieve the data, will be much more efficient than issuing an individual SQL statement for each row that is to be deleted.

As mentioned above, the illustrative embodiments provide mechanisms for efficiently detecting contiguous ranges of messages that have been processed by a parallel Apply program to aggressively delete the entries from the DONEMSG table, thereby keeping the DONEMSG table small in size, i.e. having a small number of entries (or rows if one considers the DONEMSG table to be a table data structure) and thus, providing improved database performance. Moreover, the illustrative embodiments provide mechanisms for immediately deleting messages from the receive queue after they have been processed, thereby keeping the receive queue at its minimum size for better performance of the queuing system. Except during restart, the Apply program does not need to query the DONEMSG table to determine which messages have been processed.

FIG. 7is an example block diagram of an improved database replication system in accordance with one illustrative embodiment. With the exception of the prune thread715, the ordered list data structure716, the prune queue718, and counter720, the elements shown inFIG. 7are similar to corresponding elements shown inFIG. 3described above. The similar elements, however, may be modified to operate in conjunction with the new elements715-720as described hereafter. It should be appreciated that, similar toFIG. 3, the source database system701and target database system706may be computing devices, such as servers or the like, with the elements shown within the source database system701and target database system706being implemented as software instructions executed by one or more processors of the respective systems and/or data structures stored in storage devices, such as a memory, hard disk, or the like, of the respective systems.

With reference toFIG. 7, the Apply program's prune thread715asynchronously prunes the DONEMSG table714and the receive queue707for all browser threads709. The prune queue718is an in-memory queue used for communicating between the browser threads709and the prune thread715. It is important to note that no SQL SELECT statements are issued against the DONEMSG table714to identify messages to prune during the Apply operation. Instead a protocol is established between the prune thread715and the browser thread709as described hereafter.

In accordance with a prune handshake protocol, each browser thread709, on startup or shutdown, communicates with the prune thread715to register or de-register its corresponding receive queue707for pruning. For example, on startup, a browser thread709registers its receive queue707for pruning with the prune thread715by sending a START_PRUNING message that contains the maximum message identifier (MSGID) from the DONEMSG table714. The maximum message identifier represents the largest sequence number for a received message received in the Receive Queue707. If the DONEMSG table714is empty, then instead of the maximum message identifier, a null value is provided in the START_PRUNING message.

The prune thread715records the browser thread's receive queue identifier and the maximum MSGID as the starting point for pruning. If the DONEMSG table714was empty, then the prune thread715records that a maximum MSGID was not seen and waits for a maximum MSGID with a sequence number of 1 before starting pruning. In addition, the prune thread715creates a new in-memory ordered list data structure716to record the message identifiers (MSGIDs) for messages received by the receive queue707, adds the MSGID information to this list data structure716, and opens a connection with the receive queue707, if needed, so that the prune thread715can starting pruning the receive queue707.

On the other hand, when a browser thread709is stopped, the browser thread709de-registers its receive queue707from the prune thread715by sending a STOP_PRUNING message to the prune thread715over the prune queue718. The prune thread715immediately stops pruning the corresponding receive queue707, closes the connection to the corresponding receive queue707, if needed, and deletes the ordered list data structure716associated with the browser thread709.

In accordance with a message protocol between the browser thread709and the prune thread715, the browser thread709informs the prune thread715that a message has been processed and is eligible for pruning by sending a MSGID notification to the prune thread715via the prune queue718. Each browser thread709receives completed transactions from its corresponding agent threads712via the done queue711. For each transaction received, the browser thread709sends the corresponding MSGIDs to the prune thread715via the prune queue718.

When a transaction is too large to fit into the memory of the target database system, it is referred to as a “monster” transaction. A monster transaction is delivered using several messages. When a monster transaction is received, parallelism is turned off because the transaction is directly applied by the browser thread709. It is not handed to the agents for parallel apply.

The browser thread709processes a monster transaction in chunks such that at any given point in time, only a portion of the transaction is buffered in memory. The browser thread709applies the monster transaction by reading enough transaction messages to fill the memory, applying the partially read transaction, discarding all message to free up the memory, and repeating this process until a last message has been processed at which time a database commit is issued. The browse thread709waits for a database commit operation before sending the monster transaction's MSGIDs to the prune thread715. The browser thread709maintains a list of all the MSGIDs of the monster transaction in the list data structure716. After the database commit operation, the browser thread sends the MSGIDs to the prune thread715.

Control messages are sent by the capture program704to make changes to the replication environment. For example, a control message is sent to subscribe a new database table702for replication. These messages are processed directly by the browser thread709and are also subject to pruning after they have been processed. The browser thread709sends a notification, to the prune thread715via the prune queue718, for every control message processed after its corresponding MSGIDs have been committed to the DONEMSG table714.

In addition to the handshake protocol and message protocol described above, the illustrative embodiments provide pruning mechanisms for pruning the receive queue707and DONEMSG table714. The pruning mechanisms implement a pruning operation to prune both the receive queue707and the DONEMSG table714. With this pruning operation, the browser thread709prunes the receive queue707and DONEMSG table714only during a startup and shutdown operation to determine the starting point from which messages should be processed from the receive queue. However, during normal operation the browser thread709is only responsible for reading the transaction messages from the receive queue707and scheduling them for parallel apply. This allows the system to achieve optimal performance. While one thread (browser) reads the transaction messages without performing expensive deletes, another thread (prune) handles the expensive task of pruning messages. The pruning performed by the browser thread709on startup is similar to what is described above with regard to asynchronous cleanup approach inFIG. 3.

During Apply708operations, the prune thread715performs these tasks asynchronously for optimal performance. The browser thread709is responsible for registering with the prune thread715the receive queue707to be pruned and sends MSGID notifications for completed transactions or control messages to the prune queue718. Messages are deleted from the receive queue707under transactional control and the deletion of the messages in the receive queue707are committed when a batch_size is reached. That is, certain queuing mechanisms, such as IBM's MQ Series, allow applications to delete messages from a queue, but the messages are not deleted until the application issues a COMMIT operation. This allows applications to reduce the overhead of I/O needed for a COMMIT operation by batching several operations together and issuing a single COMMIT for all the operations. This is referred to as transactional control.

A determination that the batch_size has been reached may be made by the prune thread715by comparing a counter720value to a pre-defined batch_size value to see if the counter720value is equal to or greater than the pre-defined batch_size value. The counter720may be incremented with each message deleted from the receive queue707by the prune thread715. If the batch_size is not reached during the Apply operation, the commit will take place during shutdown by the browser thread709.

The prune thread715also checks the ordered list data structure716to determine if a contiguous range of messages, i.e. a range of messages having consecutive MSGIDs with no gaps, having a size equal to or greater than the batch_size value is present in the ordered list data structure716. If a contiguous range of messages is found in the ordered list data structure716, then a single range-based delete command, e.g., a SQL DELETE statement, is executed against the DONEMSG table714. Upon a successful execution of the delete command, any outstanding changes on the receive queue are committed first and database changes are committed after. This order allows for recovery in the event that the application crashes after the messages have been deleted from the receive queue but before the entries are deleted from the DONEMSG table714. In such cases, the browser thread709on startup individually deletes all entries in DONEMSG table714and the corresponding messages in the receive queue707, as described with regard to the asynchronous cleanup approach inFIG. 3, before processing any new messages in the receive queue707. This guarantees that messages are processed exactly once. The just deleted range of messages are then removed from the in-memory ordered list data structure716and if the highest MSGID in the deleted range of messages is bigger than the maximum MSGID, then the maximum MSGID is updated to equal the highest MSGID. If the delete command does not complete successfully, then the operation stops.

If a contiguous range of messages is not found in the in-memory ordered list data structure716, then the point up to which MSGIDs are contiguous is stored, i.e. the highest MSGID of the contiguous range is stored. This becomes the starting point of the search for a contiguous range of messages equal in size to the batch_size in a next iteration of the pruning operation. The pruning thread715then waits for more messages from the browser thread709before determining to inspect the ordered list data structure716again. There is also a time-based criteria to check for a range of messages. This is done so that if multiple batches become eligible at the same time, due to multiple notifications, the prune thread can issue the delete-by-range without waiting for a new notification.

FIG. 8is a flowchart outlining an example pruning operation in accordance with one illustrative embodiment. As shown inFIG. 8, the operation of the pruning thread starts with the receive of a MSGID notification on the prune queue (step802). This MSGID notification is sent from a browser thread in response to a transaction being completed by the agent thread on a table copy in the target database system. In response to receiving the MSGID notification, the prune thread switches to the receive queue that contains the message with the MSGID identified in the MSGID notification and deletes the message from the receive queue (step804). The deletion of the message is not committed on the receive queue until the batch_size is reached, as described hereafter. While changes to queues are not committed, they are present in a recovery log data structure (not shown) in case the message needs to be processed again in the event of a stop and restart of the database replication process.

The ordered list data structure corresponding to the receive queue is searched backwards and the MSGID received in the MSGID notification is added to the list of MSGIDs for this receive queue (step806). A determination is made as to whether it is time to inspect the ordered list data structure for a contiguous range of messages having a total number, or size, equal to a pre-established batch_size (step808). The determination as to whether to inspect the ordered list data structure may be performed based on various criteria. For example, the determination may be based a number of MSGIDs having been received in MSGID notifications that equals some multiple of the batch_size. This determination may be made based on a counter value that is maintained by the prune thread that increments when a MSGID notification is received and whose value may be compared to the batch_size or multiple of the batch_size. The counter value may be reinitialized in response to an inspection of the ordered list data structure is performed. As another example, a timer may be used such that when the timer expires, the search of the ordered list data structure for a contiguous range of MSGIDs equal in size to the batch_size may be performed. Other criteria may likewise be used to determine when to search the ordered list data structure for the contiguous range of MSGIDs without departing from the spirit and scope of the illustrative embodiments.

If it is not time to perform a search of the ordered list data structure for the contiguous range of MSGIDs, then the operation waits for a next MSGID notification (step810) and in response to receiving a next MSGID notification, returns to step802. If it is determined that it is time to perform a search of the ordered list data structure, then it is determined whether the maximum MSGID is known or not (step812). If the maximum MSGID is not known, then a check is made as to whether the first element in the ordered list data structure has a sequence number of 1 (step814). If so, then the sequence number 1 becomes the maximum MSGID (step816). If not, then the prune thread waits for a MSGID to be added to the ordered list data structure that has a sequence number of 1. This is done by going to step810.

If the maximum MSGID is known (step812), and the first MSGID in the ordered list data structure is lower than the maximum MSGID (step818), then a single delete command is executed against the DONEMSG table (step820). On a successful execution of the delete command, any outstanding changes on the receive queue are committed first, database changes are committed next, and the deleted MSGID is removed from the in-memory ordered list data structure (step822). If the delete command is not executed successfully, then an error occurs and the operation stops.

If the maximum MSGID is known, and the first MSGID in the ordered list data structure is higher than the maximum MSGID, then a determination is made as to whether a contiguous range of MSGIDs equal or greater in size to a batch_size is found in the ordered list data structure (step824). If a contiguous range of MSGIDs is not found in the ordered list data structure, then a highest MSGID up to which MSGIDs are contiguous is stored (step826). This becomes the starting point of the search in the next iteration. The operation then continues to step810where more MSGID notifications are received and MSGIDs are added to the ordered list data structure.

If a contiguous range of MSGIDs equal or greater in size than the batch_size is found in the ordered list data structure, then a single range-based delete command is executed against the DONEMSG table (step828). If the single range-based delete command executes successfully, then any outstanding changes on the receive queue are committed first and database changes are committed next (step830). The just deleted range of messages are removed from the in-memory ordered list data structure and the maximum MSGID is updated if necessary (step832). The maximum MSGID is updated if the highest MSGID in the deleted range is larger than the old maximum MSGID. The operation then goes to step810and awaits the next MSGID notification. If the single range-based delete command does not execute successfully, then the operation terminates with an error.

It should be noted that the prune thread can receive MSGIDs in a random order since the Apply program executes transactions in parallel. This may result in gaps in the message sequence in the ordered list data structure. Gaps can occur, for example, when an agent thread applies non-dependent transactions in parallel. In such a case, it is possible that a transaction coming later in the receive queue is processed before a transaction ahead in the receive queue. In this case, the MSGIDs are out-of-order, resulting in gaps.

In another example, the Apply's parallelism can cause gaps in the MSGIDs on startup. For example, with 8 Apply agents applying transactions TXS1through TXS8, if only TXS2and TXS6were committed in a prior shutdown, the receive queue has transactions TXS1, TXS3, TXS4, TXS5, TXS7, and TXS8. The maximum MSGID in the DONEMSG table corresponds to TXS6, but Apply must still process TXS1, TXS3, TXS4, and TXS5before processing TXS7and TXS8.

The first scenario is addressed by maintaining the ordered list of MSGIDs. Starting with the maximum MSGID, the prune thread waits for all MSGIDs in the sequence before issuing the range-based delete command on the DONEMSG table. For example, if the maximum MSGID on startup is MSGID10and the batch size is 5, the prune thread waits for MSGID11, MSGID12, MSGID13, MSGID14, and MSGID15before issuing a range-based delete command from MSGID11to MSGID15, regardless of the order in which the transactions corresponding to these MSGIDs are committed.

The second scenario set forth above is addressed by storing the maximum MSGID on startup and issuing a single delete command for any MSGID that is lower than the maximum MSGID. For example, if a maximum MSGID on startup is MSGID10and the batch_size is 5, and the prune thread receives a notification for a lower MSGID, i.e. MSGID4, the prune thread issues a single delete command for MSGID4.

It should also be appreciated that it is possible for the browser thread to not find any MSGIDs in the DONEMSG table on startup and a null may be sent as part of the START_PRUNING notification to the prune thread. Here, the prune thread waits for a MSGID with sequence number 1 before issuing any range-based deleted command.

It should further be noted that a user can reinitialize the database replication process, thereby reloading the target database table. In this scenario, the MSGID sequence number is also reset to 1. However, the MSGID will have a sequence time, or timestamp, that is higher than the previous MSGID sequence. The prune thread automatically handles a reset in the sequence number as follows. When a MSGID has a new sequence time, the ordered list data structure is inspected by the prune thread for a contiguous range of MSGIDs with an older sequence time. If there is a single MSGID, then a single delete command is executed on the DONEMSG table to thereby delete this message. If there is a contiguous range of MSGIDs in the ordered list data structure, then a range-based delete command is issued without satisfying the batch size. On successful execution of single delete command or the range-based delete command, any outstanding changes on the receive queue are committed and any database changes are committed. The just deleted single MSGID or a range of MSGIDs is removed from the in-memory ordered list data structure. The maximum MSGID is updated so that any new MSGIDs that are lower and still using the old sequence time would still be deleted by issuing a single delete command on the DONEMSG table. If the single delete command or range-based deleted command does not execute successfully, then the operation stops with an error.

Thus, the illustrative embodiments provide mechanisms for pruning messages in a DONEMSG table of a replication database system asynchronously using range-based delete commands that delete a contiguous range of messages using a single range-based delete command. The illustrative embodiments further provide mechanisms for immediately deleting messages from a receive queue of the replication database system so as to minimize the size of the receive queue. The pruning mechanisms of the illustrative embodiments provide efficient pruning of messages in the queues without introducing additional overhead in the replication database system.