Snapshot isolation query transactions in distributed systems

Abstract

Methods for snapshot isolation query transactions in distributed systems are performed by systems and devices. Distributed executions of queries are performed in a processing system according to an isolation level protocol for data management and data versioning across one or more data sets, one or more compute pools, etc., within a logical server via a single transaction manager that oversees the isolation semantics and data versioning. Read transactions of queries are performed lock-free via the isolation semantics, and instant rollbacks, point-in-time queries, single-phase commits in the distributed systems are also provided. Abort and clean up operations are performed based on a distributed abort protocol and a determined oldest active transaction for the system in which the single transaction manager does not track read-only transactions, and client nodes do not maintain commit tables for transactions.

What is claimed is: 
     
         1 . A system comprising:
 a transaction manager node, a control node, and a plurality of compute nodes;   the control node and each of the plurality of compute nodes being configured to:
 provide a periodic heartbeat communication, to the transaction manager node, comprising:
 an oldest local query identifier of queries that are active and a prior read-only query counter received from the transaction manager node, and 
 a system activity list comprising active identifiers of data-modifying transactions; and 
 
   the transaction manager node configured to:
 store locally, in a globally-unique table for the system, the system activity list and completed identifiers of data-modifying transactions that are completed; 
 determine an oldest active transaction identifier of the system based at least on the periodic heartbeat communication from the control node and said each of the plurality of compute nodes, respectively, and a prior oldest active transaction identifier; and 
 initiate a clean up operation based on the oldest active transaction identifier. 
   
     
     
         2 . The system of  claim 1 , wherein the completed identifiers include at least one of commit identifiers of successfully completed queries or abort identifiers of unsuccessfully completed queries;
 the system further comprising a garbage collector configured to:
 receive the oldest active transaction identifier from the transaction manager node; and 
 perform the clean up operation. 
   
     
     
         3 . The system of  claim 2 , wherein the garbage collector, to perform the clean up, is configured to:
 determine first items for clean up from at least one of an abort identifier in the system activity list, a file comprising data stored by the system, metadata, or a binary large object (blob) of the system that is older than a transaction associated with the oldest active transaction identifier;   determine second items for clean up from at least one commit identifier in the system activity list that is older than a transaction associated with the oldest active transaction identifier and that is older than a point-in-time retention interval; and   clean up the determined first items and the determined second items.   
     
     
         4 . The system of  claim 1 , wherein the transaction manager node, to determine the oldest active transaction identifier, is configured to determine one of the prior read-only query counters from the control node and the plurality of compute nodes that is older than each other one in the system; and
 wherein the transaction manager node is configured to store the oldest active transaction identifier locally.   
     
     
         5 . The system of  claim 4 , wherein the transaction manager node is configured to:
 update the oldest active transaction identifier that is stored based on a subsequently-received periodic heartbeat communication.   
     
     
         6 . The system of  claim 1 , the transaction manager node is configured to:
 abort a query identified in the system activity list by an active identifier that lacks a corresponding identifier in the node-based list; or   abort a query associated with at least one of the control node or one of the plurality of compute nodes based on a forget message provided therefrom at a startup or a restart thereof.   
     
     
         7 . The system of  claim 1 , wherein the oldest local query identifier and the prior read-only query counter each comprise respective local counters maintained respectively by the control node and each of the plurality of compute nodes; or
 wherein the transaction manager node is configured to provide, to each of the control node and the plurality of compute nodes, a respective periodic heartbeat communication response that includes an oldest query identifier and a local system activity list that comprises locally-maintained active identifiers of data-modifying transactions that are active.   
     
     
         8 . A method performed by a computing system that comprises a transaction manager node, a control node, and a plurality of compute nodes, the method comprising:
 scheduling, respectively by the compute node and each of the plurality of compute nodes, a periodic heartbeat communication to the transaction manager node, the periodic heartbeat communication comprising:
 an oldest local query identifier of queries that are active and a prior read-only query counter received from the transaction manager node, and 
 an activity list comprising active identifiers of data-modifying transactions; 
   storing locally, by the transaction manager node in a globally-unique table for the system, the system activity list and completed identifiers of data-modifying transactions that are completed;   determining, by the transaction manager node, a query associated with one of the active identifiers having met an abort condition based at least on the periodic heartbeat communications;   transmitting, by the transaction manager node, indicia of aborting the query to a lock manager service enabling the lock manager service to release locks associated with execution of the query;   modifying, by the transaction manager node and subsequent to the locks being released, an active identifier associated with the query to an abort identifier associated with the query; and   storing the abort identifier, by the transaction manager node, in a master abort cache at the transaction manager.   
     
     
         9 . The method of  claim 8 , further comprising:
 providing, by the transaction manager node to the control node and subsequent to receiving by the system a next query against a data set that has a plurality of versions for data thereof, a token associated with the next query, the token including a transaction start identifier of the next query, an active query list, and a partial abort list of recently aborted queries, in the master abort cache, comprising the abort identifier associated with the query; and   storing respectively in local abort caches, by each of the control node and the plurality of compute nodes, the abort identifier associated with the query that is included in the token.   
     
     
         10 . The method of  claim 8 , further comprising:
 providing, by the transaction manager node to the control node and subsequent to receiving by the system a next query against a data set that has a plurality of versions for data thereof, a token associated with the next query, the token including a transaction start identifier of the next query, an active query list, and a partial abort list of recently aborted queries that includes the abort identifier associated with the query;   identifying, by one or more of the plurality of compute nodes, a version of the plurality of versions of the data in the data set that excludes an aborted version of the data associated with the query based on the partial abort list in the token; and   executing, distributively by the one or more of the plurality of compute nodes, respective portions of a task of the next query based on the data having the version.   
     
     
         11 . The method of  claim 8 , wherein the abort condition comprises at least one of:
 the transaction manager node or a lock manager service having been unavailable;   an abort message being lost and a transaction of the query being completed;   an instance of a database server under which the query executes, a process of the control node, or a process of one of the plurality of compute nodes being restarted;   a compute pool comprising the control node and the plurality of compute nodes being paused; or   a begin-transaction message being successfully provided by the transaction manager node and unsuccessfully received by the control node.   
     
     
         12 . The method of  claim 8 , wherein the periodic heartbeat communication also comprises a node-based list of active identifiers of data-modifying transactions;
 wherein the abort condition comprises a mismatch between the node-based list of active identifiers of data-modifying transactions and the system activity list;   the method further comprising:
 aborting, by the transaction manager node, one or more transactions identified in the system activity list by active identifiers that lack a corresponding identifier in the node-based list. 
   
     
     
         13 . The method of  claim 8 , wherein the abort condition comprises a transaction start identifier of the query indicating an earlier time than the prior read-only query counter of the control node;
 the method further comprising:
 providing, by the transaction manager node to the control node and subsequent to said storing locally the system activity list, a token associated with the query, the token including the transaction start identifier of the query; 
 aborting the query by the control node based on the transaction start identifier of the query indicating an earlier time than the prior read-only query counter of the control node; and 
 providing, by the control node to the transaction manager node, an abort indication for the query. 
   
     
     
         14 . The method of  claim 8 , further comprising:
 updating by the transaction manager node an oldest active transaction identifier of the computing system based at least on the query being aborted; and   initiating by the transaction manager node a clean up operation that includes truncating an entry associated with the abort identifier in the master abort cache based on the oldest active transaction identifier being more recent than the query associated with the abort identifier.   
     
     
         15 . A computer-readable storage medium having program instructions recorded thereon that, when executed by a processing system, perform a method, the method comprising:
 scheduling, by each of a plurality of compute nodes, respectively, and a control node, a periodic heartbeat communication to a transaction manager node, the periodic heartbeat communication comprising:
 an oldest local query identifier of queries that are active and a prior read-only query counter received from the transaction manager, and 
 an activity list comprising active identifiers of data-modifying transactions; 
   storing locally, by the transaction manager node in a globally-unique table for the system, the system activity list and completed identifiers of data-modifying transactions that are completed;   determining, by the transaction manager node, a query associated with one of the active identifiers having met an abort condition based at least on the periodic heartbeat communications;   transmitting, by the transaction manager node, indicia of aborting the query to a lock manager service enabling the lock manager service to release locks associated with execution of the query;   modifying, by the transaction manager node and subsequent to the locks being released, an active identifier associated with the query to an abort identifier associated with the query; and   storing the abort identifier, by the transaction manager node, in a master abort cache at the transaction manager.   
     
     
         16 . The computer-readable storage medium of  claim 15 , wherein the method further comprises:
 providing, by the transaction manager node to the control node and subsequent to receiving by the system a next query against a data set that has a plurality of versions for data thereof, a token associated with the next query, the token including a transaction start identifier of the next query, an active query list, and a partial abort list of recently aborted queries, in the master abort cache, comprising the abort identifier associated with the query; and   storing respectively in local abort caches, by each of the control node and the plurality of compute nodes, the abort identifier associated with the query that is included in the token.   
     
     
         17 . The computer-readable storage medium of  claim 15 , wherein the method further comprises:
 providing, by the transaction manager node to the control node and subsequent to receiving by the system a next query against a data set that has a plurality of versions for data thereof, a token associated with the next query, the token including a transaction start identifier of the next query, an active query list, and a partial abort list of recently aborted queries;   identifying, by one or more of the plurality of compute nodes, a version of the plurality of versions of the data in the data set that excludes an aborted version of the data associated with the query based on the token that includes the abort identifier associated with the query; and   executing, distributively by the one or more of the plurality of compute nodes, respective portions of a task of the next query based on the data having the version.   
     
     
         18 . The computer-readable storage medium of  claim 15 , wherein the abort condition comprises at least one of:
 the transaction manager node or a lock manager service having been unavailable;   an abort message being lost and a transaction of the query being completed;   an instance of a database server under which the query executes, a process of the control node, or a process of one of the plurality of compute nodes being restarted;   a compute pool comprising the control node and the plurality of compute nodes being paused; or   a begin-transaction message being successfully provided by the transaction manager node and unsuccessfully received by the control node.   
     
     
         19 . The computer-readable storage medium of  claim 15 , wherein the periodic heartbeat communication also comprises a node-based list of active identifiers of data-modifying transactions;
 wherein the abort condition comprises a mismatch between the node-based list of active identifiers of data-modifying transactions and the system activity list; and   wherein the method further comprises:
 aborting, by the transaction manager node, one or more queries identified in the system activity list by active identifiers that lack a corresponding identifier in the node-based list. 
   
     
     
         20 . The computer-readable storage medium of  claim 15 , wherein the abort condition comprises a transaction start identifier of the query indicating an earlier time than the prior read-only query counter of the control node; and
 wherein the method further comprises:
 providing, by the transaction manager node to the control node and subsequent to said storing locally the system activity list, a token associated with the query, the token including the transaction start identifier of the query; 
 aborting the query by the control node based on the transaction start identifier of the query indicating an earlier time than the prior read-only query counter of the control node; and 
 providing, by the control node to the transaction manager node, an abort indication for the query.