Executing transactions on distributed databases

A distributed database system executes transactions on a distributed database. A received transaction includes statements describing modifications of records stored in the distributed database. The distributed database system executes the transaction at a query server by obtaining copies of records corresponding to the statements of the transaction and performing the modifications specified by the statements of the transaction on the record copies. The distributed database system stores the modified record copies at the query server during execution of the transaction. After the transaction has successfully been executed at the query server, the distributed database system attempts to perform a commit process to update the records stored in the distributed database based on the modified record copies.

BACKGROUND

Field of Art

This disclosure relates in general to distributed databases, and in particular to executing transactions in distributed databases.

Description of the Related Art

Enterprises store data in various types of data stores such as relational databases, object-oriented databases, graph databases, document-oriented databases and so on. A large amount of data is stored in non-distributed database (e.g., centralized databases), such as traditional Structured Query Language (SQL) databases. Some database management systems (DBMSs) for such non-distributed databases represent transactions using a declarative query language (e.g., SQL), and maintain the “ACID” properties for transactions (i.e., atomicity, consistency, isolation, and durability).

However, there is an increase in using distributed databases to store data. An example of such distributed databases includes NoSQL databases, such as document-oriented databases, key-value stores, and graph databases. Due to the distributed nature of these databases, executing transactions which maintain the ACID properties and provide effective performance is challenging. Furthermore, NoSQL databases do not represent transactions using a declarative query language capable of representing transaction statements of arbitrary complexity, such as SQL.

SUMMARY

Embodiments of a disclosed system, method and computer readable storage medium execute transactions in a distributed database. A distributed database system receives transactions from a client device that can include statements describing arbitrarily complex modifications of records stored in the distributed database (e.g., documents), and may include a single statement (i.e., single statement transactions) or multiple statements (i.e., multi-statement transactions). Furthermore, the received transactions can be represented using a declarative query language or other query languages (e.g., procedural query languages). The distributed database system executes the transaction at a query server, where the execution includes obtaining copies of indexes and records corresponding to the statements of the transaction (e.g., by retrieving the indexes and records from the distributed database) and performing the modifications described by the one or more statements of the transaction on the record copies. The distributed database system stores modified record copies at the query server during execution of the transaction. Further, during execution of a statement of the transaction, if the statement is associated with a record copy stored at the query server during execution of a previous statement, the query server uses the stored record copies to execute the statement. By executing transactions at the query server, according to some embodiments, the distributed database system applies optimistic concurrency control for transactions by concurrently executing transactions at individual query servers without locking some or all of the processes of the distributed database system prior to performing a commit. After the transaction has successfully been executed at the query server, the distributed database system performs a commit process to update the records stored in the distributed database based on the modified record copies, such as committing the modified record copies to one or more data servers of the distributed database. In particular, the distributed database system performs a commit process which prevents conflicts with other transactions or other modifications of records executed by the distributed database system (e.g., a transaction executed concurrently).

A distributed database system receives a transaction including a statement including instructions for modification of a record stored at a data server of a distributed database. After receiving the transaction, the distributed transaction system executes the transaction at a query server. During execution of the transaction at the query server, the distributed database system obtains a copy of the record at the query server. Using the obtained record copy, the distributed database system performs the modification described by the statement on the record copy at the query server. The distributed database system stores the modified record copy at the e query server. After executing the received transaction at the query server, the distributed database system updates the record stored at the data server based on the modified record copy. In particular, the distributed database system updates the record without conflicting with any other transactions executed by the distributed database system.

In some embodiments, the distributed database system maintains one or more delta tables at the query server during execution of a transaction to store modified record copies. The query server can use the delta tables to execute statements (e.g., accessing the modified records), applying predicate logic in some statements of a transaction, or committing modified record copies to the distributed database.

In some embodiments, the distributed database system maintains transaction logs corresponding to statements of a transaction at the query server during execution of a transaction. The distributed database system can use the transaction logs to fully or partially rollback transactions which fail during execution at the query server.

DETAILED DESCRIPTION

System Environment

FIG.1is an embodiment of a block diagram of a distributed database system environment100for performing transactions. In the embodiment shown, the system environment includes a distributed database system110, a client device120, and a network130. Other embodiments may use more or fewer or different systems than those illustrated inFIG.1. Functions of various modules and systems described herein can be implemented by other modules or systems than those described herein.

The distributed database system110manages a distributed database. The distributed database system110includes distributed query servers112, distributed index servers114, and distributed data servers116(e.g., database nodes). The distributed database system110receives transactions from the client device120that can include one or more statements (i.e., single or multi-statement transactions). In an exemplary embodiment, the statements of the received transactions are represented using a declarative query language, such as the Structured Query Language (SQL). A declarative query language refers to a query language which describes requests (e.g., transaction statements) in terms of what data to process and the desired results of the processing (e.g., what data to retrieve or what updates should be performed on data), but does not specify how the processing should be executed. Instead, a declarative query language relies on the underlying database system (e.g., the distributed database system110) to determine how the requests should be executed. As such, declarative query languages allow for users (e.g., the client device120) to submit arbitrarily complex requests, such as describing an arbitrary number of modifications to an arbitrary number of records or record fields. Example declarative query languages include structured query language (SQL), SQL++, Non-First Normal Form Query Language (N1QL), XML Query (XQuery), Cypher, SPARQL Protocol and RDF Query Language (SPARQL), and Gremlin. In other embodiments, the statements of received transactions can be represented using other types of query languages, such as procedural query languages. The distributed database system110executes received transactions using a query server of the distributed query servers112through a process which ensures the ACID properties of database transactions. In particular, according to some embodiments, the distributed database system110employs optimistic concurrency control for transactions by concurrently executing transactions at individual query servers112without locking some or all of the processes of the distributed database system110prior to performing a commit. AlthoughFIG.1shows a single element, the distributed database system110broadly represents a distributed database including the distributed query servers112, the index servers114, and the data servers116which may be located in one or more physical locations. The individual elements of the distributed database system110(e.g., the query servers112, the index servers114, and the data servers116) may be any computing device, including but not limited to: servers, racks, workstations, personal computers, general purpose computers, laptops, Internet appliances, wireless devices, wired devices, multi-processor systems, mini-computers, cloud computing systems, and the like. Furthermore, the elements of the distributed database system110depicted inFIG.1may also represent one or more virtual computing instances (e.g., virtual database nodes), which may execute using one or more computers in a datacenter such as a virtual server farm. In particular, the query servers112, index servers114, and data servers116may each be virtual database nodes executed on one or more computing devices.

In an exemplary embodiment, the statements of the received transactions describe manipulations of stored data (e.g., SQL data manipulation language (DML) statements, such as SELECT, INSERT, UPDATE, UPSERT, DELETE, and MERGE). DML statements can describe modifications of one or more records, such as INSERT, UPDATE, UPSERT, DELETE, and MERGE. In the same or different embodiment, the statements of the received transactions describe new data to store (e.g., SQL data definition language (DDL) statements), permissions of the distributed database system110(e.g., SQL data control language (DCL) statements), or configuring the processing of transactions by the distributed database system110(e.g., SQL transaction control language (TCL) statements). In various embodiments, the distributed database system may or may not permit transactions including certain types of statements (e.g., DDLs, DCLs, or TCLs).

The query servers112receive and process transactions in the distributed database system110. The query server112may receive the transaction in the form of statements received from a client device, for example, database statements that include instructions such as update, insert, delete operations that modify records of the distributed database. In embodiments, a transaction received by the distributed database system110(e.g., from the client device120) is routed to a particular query server112which is used to execute the transaction. For instance, the distributed database system110can select one of the query servers112to execute a received transaction (i.e., an execution query server), where the selected query server executes all of the statements included in the received transaction after execution begins. Each statement of a transaction may be assigned a transaction identifier of the transaction, and the distributed database system110may forward all statements with the transaction identifier to the same execution query server. A query server112maintains the ACID properties of transactions (atomicity, consistency, isolation, and durability) while concurrently executing one or more transactions in the distributed database system110. Furthermore, the query servers112do not need to communicate with each other in order to coordinate concurrent execution of transactions in the distributed database. In maintaining the ACID properties during execution of a transaction in a distributed database, the execution query server executes the one or more statements that represent instructions of the transaction by communicating with one or more of the index servers114and one or more of the data servers116. In particular, the execution query server executes the one or more statements of a received transactions by generating or accessing local copies of records (e.g., documents in a document-oriented database or rows in a relational database) corresponding to the one or more statements. The execution query server further locally performs modifications corresponding to the one or more statements on the local copies of the records. After locally executing the one or more statements, the execution query server commits the local copies to the data servers116. For example, the execution query server may begin committing the local copies to the data servers116based on reaching a “COMMIT” statement of the transaction. As described above with reference to the distributed database system110, a query server112may be a virtual database node executed on one or more computing devices (e.g., a server computer or server cluster), where each of the one or more computing devices can include one or more virtual database nodes. Execution of ACID transactions by an execution query server is described in greater detail below with reference toFIGS.2-3.

In some embodiments, the execution query server caches local copies records as the statements of a transaction are executed. In this case, the execution query server can use cached records to execute a statement if records relevant to the statement were cached during execution of a previous statement. In doing so, the execution query server improves transaction execution efficiency by quickly accessing cached records, rather than repeatedly performing the relatively expensive operation of retrieving the records from the data servers116.

In some embodiments, the execution query server generates a query execution plan (QEPs) for one or more statements (e.g., DML statements) of a received transaction in order to execute the transaction. A QEP for a statement can be represented by an ordered sequence of operators, where each operator describes instructions for a specific operation on one or more indexes (e.g., stored on one or more index servers114) or records (e.g., stored on one or more data servers116). Example operations performed by QEP operators include SQL fetch, index scan, key scan, union scan, intersect scan, nested or fetch loop joins, hash joins, and any other operation usable to execute a transaction statement. In particular, the query servers112can generate QEPs including operators configured to fetch indexes or data from an index server114or data server116, respectively, or to use cached data retrieved for a previous statement, depending on whether the relevant data has been cached or not. Generating QEPs for query statements which fetch data or use cached data to execute transactions is described in greater detail below with reference to the transaction execution module210andFIG.2.

In some embodiments, the query servers112determine an optimal QEP from a set of possible (i.e., logically equivalent) QEPs for a given query statement using a set of optimization criteria. In one embodiment, the optimization criteria include a set of rules for generating QEPs, such as an order in which query filters are applied to fields, which logical operators to use, and any other applicable rules used to optimize query execution. In the same or different embodiment, the optimization criteria may identify an optimal QEP based on execution costs determined for individual QEPs in the set of possible QEPs. For example, QEPs may be costed, and an optimal QEP may be selected, using any of the methods described in co-pending U.S. patent application Ser. No. 16/788,923, filed Feb. 12, 2020, which is incorporated herein by reference in its entirety.

The index servers114manage indexes for data stored in the distributed database system110. In various embodiments, the index servers114can receive requests for indexes (e.g., during execution of a transaction) from the query servers112(e.g., from an execution query server), directly from the client device120, or some other element of the system environment100. The index servers114can generate indexes for one or more fields of records stored by the data servers116. Indexes stores by the index servers can include B-tree indexes, inverted tree indexes, hash indexes, R-tree indexes, GIST indexes, or any other suitable type of database index. The index servers114may automatically generate or update indexes for one or more records stored in the data server116based on transactions performed by the to the distributed database system110. Additionally, or alternatively, the index servers114may automatically generate or update indexes for one or more records stored in the data server116based on a request (e.g., an instruction associated transaction) received from another component of the distributed database system110(e.g., the query servers112). In some embodiments, a given index server of the index servers114manages indexes stored on a corresponding index storage server or server cluster. In the same or different embodiments, a given index server of the data servers116manages indexes stored locally on the given data server. As described above with reference to the distributed database system110, an index server114may be a virtual database node executed on one or more computing devices (e.g., a server computer or server cluster), where each of the one or more computing devices can include one or more virtual database nodes.

The data servers116manage data (e.g., records) stored in a distributed database of the distributed database system110. In various embodiments, the data servers116can provide requested data to other elements of the system environment100(e.g., the query servers112) and store new or modified data in the distributed database. In particular, the data servers116can perform commits of new or modified data to the distributed database which maintain the ACID properties of transactions. The distributed database may be one of various types of distributed databases, such as a document-oriented database, a key-value store, a graph database, a relational database, a wide-column database, or a search index. In some embodiments, a given data server of the data servers116manages data stored on a corresponding data storage server or server cluster. In the same or different embodiments, given data server of the data servers116manages data stored locally on the given data server. As described above with reference to the distributed database system110, a data server116may be a virtual database node executed on one or more computing devices (e.g., a server computer or server cluster), where each of the one or more computing devices can include one or more virtual database nodes.

In an exemplary embodiment, in order to maintain the ACID properties in committing a transaction, the data servers116maintain Active Transaction Records (ATRs) which describe the status of transactions actively being executed by one or more query servers112. The ATRs are accessible to the query servers112in order to provide awareness to the query servers112of other transactions being executed and prevent transaction conflicts from occurring. Furthermore, the data servers116maintain data associated with individual records usable to stage modifications of the records (i.e., virtual attributes). Similarly, to the ATRs, the virtual attributes of each record are accessible to the query servers112in order to provide awareness of whether a record is being modified by another active transaction. Staged modifications in the virtual attributes of a record can include an identifier of a transaction corresponding to the modification, such as an identifier of an entry for the transaction in an ATR. Furthermore, the virtual attributes can include information allowing a first query server112to determine whether a record has been modified by a second query server112during execution of a transaction locally at the query server112. In particular, the virtual attributes can include a check-and-set (CAS) value which is received by a query server112when reading the virtual attributes of a record. The CAS value for the virtual attributes for a record can be updated each time the virtual attributes are modified. As such, a query server112can prevent write-write transaction conflicts by determining whether a CAS value for the virtual attributes of a record changes between the time the CAS is first received and the query server112attempts to modify the virtual attributes (e.g., to stage new modifications). This exemplary embodiment and its various processes are described in further detail by co-pending U.S. Provisional Application No. 63/029,325, which is incorporated by reference herein in its entirety. Furthermore, performing commits of new or modified data to the distributed data servers116which maintain the ACID properties of transactions, and particularly using ATRs and virtual attributes, is described in greater detail below with reference toFIGS.2and3.

The client device120provides transactions to the distributed database system110. In particular, the client device120sends single or multi-statement transactions to the distributed database system110. The transactions can include statements represented using a declarative programming language, procedural query language, or other type of query language used by the distributed database system110. In embodiments, the client device120sends transactions to the distributed database system110over the network130. The transactions may be generated via the client application125or other process executing on the client device120. Furthermore, the client device120can receive data from the distributed database system110, such as data requested in a transaction. In some embodiments, the client device120provides transactions to the distributed database system110through one or more transaction coordination servers, which then route the transactions to an executing query server of the query servers112. In the same or different embodiments, the client device120provides transactions directly to an execution server of the query servers112. In still another same or different embodiment, the client device120executes transactions locally by communicating directly with the index servers114or data servers116. Example client devices include personal computers (PCs), mobile phones, additional server computers, etc. The client device120may communicate with the distributed database system110through an Application Programming Interface (API). An example API the distributed database system110might provide is a Representation State Transfer (REST) API.

In some embodiments, the client application125communicates with the distributed database system110via software integrated with a software development kit (SDK) associated with the distributed database system110. In this case, the client application125may submit transactions using software tools provided by the SDK, such as transaction execution functions, stored procedures (e.g., PL/SQL), or eventing functions (e.g., Couchbase Eventing Functions). In this case, the client application125may submit transactions to the distributed database system using software tools of the SDK. The SDK may be implemented using any programming language (e.g., Java, C++, Python, etc.). The SDK may communicate with the distributed database system110via an Application Programming Interface (API) associated with the distributed database system110(e.g., using Representational State Transfer (REST) over an application protocol (e.g., HTTP)).

The interactions between the client device120and the distributed database system110are typically performed via a network130, for example, via the Internet. In one embodiment, the network uses standard communications technologies or protocols. Example networking protocol include the transmission control protocol/Internet protocol (TCP/IP), the user datagram protocol (UDP), internet control message protocol (ICMP), etc. The data exchanged over the network can be represented using technologies and/or formats including JSON, the hypertext markup language (HTML), the extensible markup language (XML), etc. In another embodiment, the entities can use custom or dedicated data communications technologies instead of, or in addition to, the ones described above. The techniques disclosed herein can be used with any type of communication technology, so long as the communication technology supports receiving a web request by the distributed database system110from a sender, for example, a client device120and transmitting of results obtained by processing the web request to the sender.

FIG.2is a block diagram of an embodiment of a query server200. The query server200may be an embodiment of one of the query servers112. In the embodiment shown, the query server200includes a transaction execution module210, a transaction commit module220, a transaction rollback module230, and a cached record store240. In other embodiments, the query server200may include different or additional components than those shown inFIG.2. Furthermore, some or all of the operations described for the query server200may be performed by other components of the distributed database system110, or another suitable device.

The transaction execution module210executes transactions received by the query server200(i.e., when the query server200is used as an execution query server). In embodiments, the transaction execution module210executes each of one or more statements included in a received transaction. The received transaction can be a single statement transaction or a multi-statement transaction. If the received transaction is a multi-statement transaction, the statements of the transaction can describe modifications to multiple distinct groups of records stored in the data servers116(e.g., relational tables or document collections). The transaction execution module210can receive individual statements of a single transaction together or individually. In one embodiment, the client application125or distributed database management system110assign a transaction identifier to each statement associated with the same transaction provided by the client application125. In doing so, the distributed database management system110can provide each statement with the same transaction identifier to the transaction execution module210of the same query server200to execute the transaction. As part of executing a statement, the transaction execution module210can communicate with one or more index servers114(e.g., in order to retrieve index keys for data relevant to a given statement of the transaction) and with one or more data servers116(e.g., in order to retrieve data relevant to a given statement of the transaction). For instance, if a statement identifies data stored by one or more of the data servers116, the transaction execution module210may retrieve indexes corresponding to the data (e.g., based on a filter included in the statement) and use the retrieved indexes to retrieve the data from the one or more data servers116. Furthermore, the transaction execution module210can locally cache the data in the cached record store240and if the cached data is identified by a subsequently executed statement the transaction execution module210can retrieve the cached data instead of communication with a data server116. After retrieving data identified by a statement, the transaction execution module210performs any modifications to the data corresponding to the statement (e.g., INSERT, UPDATE, MERGE, DELETE) and stores a local copy of the modified data. In one embodiment, the transaction execution module210executes the statements of a received transaction serially in order to account for modifications corresponding to previously executed statements in subsequently executed statements. In doing so, the transaction execution module210accounts for the cumulative modifications corresponding to multiple statements of the received transaction. The transaction execution module210can provide the local copies of modified data corresponding to a transaction being executed to the transaction commit module220in order to commit the modifications to the data servers116, as described below. Embodiments of execution of a transaction by the transaction execution module210is described in greater detail below with reference toFIGS.3and4A-B.

In some embodiments, the transaction execution module210maintains a set of data structures to locally store (i.e., cache) indexes or data describing the execution of a transaction. In particular, the transaction execution module210can store the local copies of modified data for a transaction in one or more delta tables (e.g., in the cached data store240). In one embodiment, the transaction execution module210maintains one or more delta tables for each transaction executed by the query server200. If a transaction describes modifications to multiple groups of records (e.g., document collections), the transaction execution module210can maintain a distinct delta table for each group of records corresponding to the transaction. In this case, each delta table includes local copies of modified records from the corresponding group of records. In the same or different embodiments, the transaction execution module210can maintain transaction logs for some or all of the statements describing the modification performed for the corresponding statement. The transaction logs can be used by the transaction rollback module230to partially rollback modifications performed for statements of transactions, as described in greater detail below with reference to the transaction rollback module230. In various embodiments, the local data structures corresponding to a transaction being executed by the transaction execution module210are private such that they are inaccessible to other processes on the transaction execution module210executing other transactions. Delta tables and transaction logs are described in further detail below with reference toFIGS.3and4B. In some embodiments, the transaction execution module210maintains additional delta table or transaction logs for a transaction corresponding to previously executed statements of the transaction, which can be used to roll-back the modifications of a transaction in current delta table or transaction logs to the previous modifications (i.e., “save points”). Additionally, or alternatively, the transaction execution module210can maintain data structures for each statement of a transaction describing the data prior to any manipulations performed for the statement. The transaction execution module210can further maintain additional or different local data structures than those described above to describe the execution of a transaction.

In some embodiments, the transaction execution module210uses cached modified records to apply filter logic included in some statements of a received transaction. For example, if a received transaction includes a SELECT statement (e.g., SELECT a FROM x WHERE a<10) coming after one or more statements modifying the records corresponding to the SELECT statement, the transaction execution module210may apply the filter logic to the cached modified records to generate a projection table for the SELECT statement. In particular, if the transaction execution module210stores the modified records in one or more delta tables, the transaction execution module210can apply the filter logic to the one or more delta tables to generate the projection table. For instance, the transaction execution module210can efficiently combine the local copies of records in delta tables for previously executed statements and records retrieved from one or more data servers116to select records to modify for a current statement. This technique ensures correctness of modifications while improving performance and avoiding intermediate writes, index updates, and coordinating statement execution. Projection tables are described in greater detail below with reference toFIG.4B.

During the process of executing a transaction by the transaction execution module210, various issues can arise preventing processing of the transaction by the query server200at the current time. For instance, in various embodiments, a received transaction may not be executable by the transaction execution module210, such as a transaction including malformed statements (e.g., invalid syntax of a query language used to represent the statements) or statements of a type not permitted by the transaction execution module210(e.g., DDLs). In this case, the transaction execution module210can validate the statements of a received transaction before beginning execution of the transaction. If an issue is identified in the transaction, the transaction execution module210can terminate processing of the transaction, and may further notify a client device120which submitted the transaction that the transaction was not executed. As another example, a technical error may occur during execution of the transaction preventing further execution of the transaction, such as the query server200crashing or experiencing some other error. In this case, the transaction execution module210aborts the transaction and provides the transaction to the transaction rollback module230to roll back the transaction. Transaction rollbacks are described in greater detail below with reference to the transaction rollback module230andFIG.3. In some embodiments, the transaction execution module210does not abort a transaction if a failure occurs during retrieval of indexes from one or more index servers114in executing a statement of a received execution. In this case, the transaction execution module210may automatically retry retrieving the indexes, either immediately or after a time period has elapsed.

In some embodiments, the transaction execution module210executes some statements of a transaction by generating a QEP for the individual statements, as described above with reference to the query server112andFIG.1. A QEP can include QEP operators describing an operation on data stored on one or more data servers116, respectively, or cached on the query server200. The transaction execution module210may generate different QEPs in order to account for different statement execution scenarios depending on whether relevant data are cached on the query sever200. In particular, the transaction execution module210can generate a QEP for a statement that uses operators (e.g., scan or join operators) which retrieve data from the data servers116, access local copies of data stored on the query server200(e.g., in a delta table corresponding to the transaction), or perform any combination thereof.

In some embodiments, a component of the distributed database system110executes a background process to update indexes stored on the index servers114to be consistent with updates to data stored on the data servers116(e.g., based on modifications to data resulting from an executed transaction). In this case, at the point in time the transaction execution module210is executing a statement of a received transaction the indexes used to execute the statement may not be consistent with modifications to data corresponding to transactions previously executed by the query server200. In order prevent conflicts or other issues resulting from index inconsistency, the transaction execution module210can wait to retrieve indexes for a statement of a received transaction currently being executed until it determines that the indexes have been updated to be consistent with a previously execute transaction. For example, indexes stored in the index servers114may be associated with one or more values indicating the most recent transaction or other operation performed on the indexes (e.g., a sequence or version number). In this case, the transaction execution module210can check the one or more values to determine whether the indexes are consistent with modifications corresponding to previously executed transaction. As a result of waiting until indexes are consistent, the transaction execution module210can ensure a “read-your-writes” consistency level across two or more transactions executed serially.

In some embodiments, the transaction execution module210considers and implements integrity constraints for the distributed database system110. In particular, the integrity constraints for the distributed database system place restrictions on data stored in the data servers116. For instance, the distributed database system110may implement various SQL integrity constraints such as domain constraints, key constraints, referential integrity constraints, and entity integrity constraints. The transaction execution module210considers existing integrity constraints during execution of transactions, such as determining whether modifications corresponding to a transaction conform to existing integrity constraints. The transaction execution module210further implements new constraints for the distributed database system110, such as constraints specified by a received transaction.

In the same or different embodiments, the transaction execution module210considers and implements database triggers for the distributed database system110. In particular, the database triggers for the distributed database system enforces restrictions on manipulations of transactions executed by the transaction execution module210in response to certain transaction events (i.e., triggers). Similar to the integrity constraints described above, the transaction execution module210considers existing database triggers during execution of transactions. The transaction execution module210further implements new database triggers the distributed database system110, such as database triggers specified by a received transaction.

The transaction commit module220commits local copies of modified data corresponding to a transaction executed by the transaction execution module210to one or more of the data servers116. In embodiments, the transaction commit module220individually commits local copies of modifications for each record included in the modified data (i.e., record modifications). In particular, the commit process is divided into two stages: a pending stage during which the record modifications are staged in one or more data servers116storing the relevant records, and a committed stage during which the records are updated to persistently reflect the staged modification. The pending stage and the committed stage are configured to maintain the ACID properties of transactions during the commit process for record modifications in a distributed database. The transaction commit module220can further communicate with the transaction rollback module230in order to rollback a transaction in the pending stage in response to a failure or a conflict with another pending transaction being executed by the query server200, a different query server112, or a client device120communicating directly with one or more data servers116. Transaction rollbacks are described in greater detail below with reference to the transaction rollback module230andFIG.3. After the record modifications for a transaction are fully committed to one or more corresponding data servers116, the transaction commit module220can communicate with the transaction execution module210or directly with the client device120which submitted the transaction to indicate the transaction was committed or provide data requested in the transaction.

In an exemplary embodiment, during the commit process by the transaction commit module220a transaction entry is added to an ATR corresponding to one or more records corresponding to a transaction. As described above with reference toFIG.1, the transaction entry in the ATR provides visibility of the status of the transaction to other query servers112executing transactions or a client device120submitting transactions to the distributed database system110. In one embodiment, the transaction commit module220manually selects an ATR for the transaction and adds the transaction entry to the transaction. In an alternative embodiment, the transaction entry is automatically added to the ATR by a data server116, such as based on a first request to stage record modifications by the transaction commit module220. Initially, the transaction entry for the transaction indicates that the transaction is pending and has not yet been committed (e.g., the transaction entry includes a field “transaction status”: “pending”).

During the pending stage, the transaction commit module220stages each of the record modifications for each record in the virtual attributes corresponding to the record. While in the pending stage, the transaction can be aborted for various reasons, such as encountering conflicting changes corresponding to a different pending transaction in the virtual attributes of a record (e.g., identifying a mismatching CAS value or identifying staged modifications in virtual attributes for another transaction), or a data server116crashing. If the transaction is aborted during the pending stage, the transaction commit module220updates the transaction entry in the ATR to indicate that the transaction has been aborted (e.g., the transaction status field is updated to “transaction status”: “aborted”). Furthermore, the transaction commit module220communicates with the transaction rollback module230to rollback the aborted transaction. If the transaction commit module220successfully stages each of the record modifications, the transaction commit module220enters a committed stage and fully commits the transaction. In particular, the transaction commit module220persistently updates each record to reflect the staged record modifications. Before fully committing the transaction, the transaction commit module220updates the transaction entry in the ATR to indicate that the transaction is committed (e.g., the transaction status field is updated to “transaction status”: “committed”).

During the committed stage, the transaction commit module220considers the transaction to have been committed and the query server200does not roll back the transaction. To address failed transactions which have entered the committed stage, the distributed database system110maintains a cleanup process which completes the commit process for transactions which failed while being fully committed. In particular, the transaction commit module220may participate in the cleanup process by periodically querying ATRs stored by the data servers116and, if a failed transaction is identified, finish the process of committing the transaction. After successfully fully committing the modified records for a transaction, the transaction commit module220updates the transaction entry in the ATR to indicate that the transaction has successfully been fully committed (e.g., the transaction status field is updated to “transaction status”: “completed”).

The transaction rollback module230rolls back aborted transactions partially executed by the query server200. In embodiments, the transaction rollback module230rolls back transactions which have been aborted prior to reaching the committed stage of the commit process executed by the transaction commit module220, as described above. If a transaction is aborted or fails prior to the committed stage of the commit process, the transaction rollback module230frees up local memory resources used to execute the transaction locally (e.g., delta tables, transaction logs, or other data structures generated to execute a transaction at the query server). In some embodiments, the transaction rollback module230adds the transaction to an aborted transaction queue maintained by the distributed database system110. In this case, the transaction execution module210of the query server200or another query server112attempts to execute the transaction again at some time after the transaction is added to the aborted transaction queue. For example, the transaction execution module210may execute one or more transactions in the aborted transaction queue on a periodic basis, or in response to an event, such as a request to retry the transaction from a client device120.

In embodiments, the transaction rollback module230rolls back transactions aborted during local execution of the transactions at the transaction execution module210(e.g., prior to the commit process), as described above with reference to the transaction execution module210. In this case, the transaction rollback module230removes local data at the query server200used to execute the transaction prior to the transaction being aborted (i.e., a local rollback). In the same or different embodiments, the transaction execution module210can abort execution of a statement of a transaction without aborting the entire transaction (i.e., a partial abort). In this case, the transaction rollback module230can use the transaction logs for the aborted statements to undo modifications corresponding to the individual statement. Example scenarios in which the transaction rollback module230may perform a rollback of a transaction during local execution include the client device120requesting a partial rollback of the transaction to a save point, the client device120requesting a rollback of the entire transaction, or automatically performing a partial or full rollback due to invalid transaction statements.

In the same or different embodiments, the transaction rollback module230can roll back transactions which are aborted or otherwise fail during the pending stage of the commit process at the transaction commit module220(i.e., a local rollback). For example, the transaction commit module220may determine that another transaction in the pending or committed stage conflicts with the transaction being committed by the transaction commit module220. As another example, one or more of the data servers116may crash during the pending stage of the commit process. In these cases, the transaction rollback module230can communicate with the one or more data server116in order to rollback data modifications staged in the data servers116(i.e., a remote rollback). The transaction rollback module230can further perform a local rollback before, after, or concurrently to performing of a remote rollback.

In some embodiments, the transaction commit module220cannot reliably determine the state of a transaction in the process of being committed (i.e., the transaction is in an ambiguous state). For example, an ATR may include an entry for the transaction indicating that the transaction is still in the pending stage when some or all of the records corresponding to the transaction have been updated to reflect the staged record modifications, suggesting the transaction is in the committed stage. As another example, the transaction commit module220may not be able to determine the status of a transaction in a timely manner (e.g., an attempt by the transaction commit module220to read the ATR timed out). In such cases, the transaction commit module220can notify a client device120that submitted the transaction (e.g., via the client application125) that the transaction is in an ambiguous state and can further provide information describing the ambiguous state to the client device120. The client device120can then submit further transactions or other instructions in order to resolve the ambiguous state of the transaction. Additionally, or alternatively, the transaction commit module220can perform additional processing steps in response to encountering a transaction in an ambiguous state, such as retrying execution of the transaction or proceeding to use a record associated with an ambiguous transaction despite the ambiguous state.

FIG.3is a flow diagram illustrating an embodiment of an execution300of a distributed transaction310by the query server200. In the embodiment shown, the query server200receives the transaction310at the transaction execution module210. The transaction310may be a multi-statement transaction or a single statement transaction. In particular, the transaction310can be a single statement transaction describing a large volume of data (e.g. thousands or millions of records). If the transaction310is a single statement transaction represented using SQL syntax it may be an UPDATE or DELETE statement of multiple records, a MERGE INTO statement, an INSERT INTO statement, or a SELECT FROM statement. Additionally, the transaction310can be a multi-statement transaction (e.g., as depicted inFIG.4A) including multiple DML statements. The statements of the transaction310may further describe records corresponding to different groups of records (e.g., relational tables or document collections). The transaction execution module210executes each of the one or more statements of the transaction310locally, and depending on the embodiment can execute some or all of the statement of the transaction310concurrently and in various orders.

During execution300of the transaction310, the transaction execution module210retrieves transaction indexes320from one or more index servers114. The transaction indexes320include a set of distinct indexes which together can be used to execute the one or more statements of the transaction310. As such, the transaction execution module210can retrieve various indexes within the transaction indexes320corresponding to different statements of the transaction310during different or concurrent time intervals. The particular transaction indexes320used to execute the transaction310may depend on a set of one or more QEPs generated by the transaction execution module210to execute the statements of the transaction310. Furthermore, some statements of the transaction310can be executed without the transaction indexes320, such as statements including one or more identifiers of one or more records stored in the data servers116, allowing the transaction execution module210to access the one or more records directly.

Using at least in part the transaction indexes320, the transaction execution module210retrieves transaction data330from one or more data servers116. The transaction data330includes data corresponding to the statements of the transaction310(e.g., data modified or requested by the transaction310). As such, the transaction execution module210can retrieve various subsets of data within the transaction data330corresponding to different statements of the transaction310during different or concurrent time intervals. The transaction execution module210caches the transaction data360on the query server200during execution of the one or more statements of the transaction310(e.g., in the cached record store240). If cached data of the transaction data330corresponds to a statement of the transaction310executed subsequently to the caching (e.g., after the corresponding data was cached during execution of a previous statement), the transaction execution module210can use the cached data to execute the statement.

Using the retrieved transaction data330, the transaction execution module210performs the modifications corresponding to the one or more statements of the transaction execution module210on the corresponding data and stores local copies of the modified data on the query server200(e.g., in the cached record store240). The transaction execution module210may store the local copies of the modified data or other data describing the modifications using various data structures, such as one or more delta tables or transaction logs, as described above with reference to the transaction execution module210andFIG.2.

If the transaction execution module210successfully executes the one or more statements of the transaction310and performs the corresponding modifications on the transaction data330, the transaction execution module210provides the modified transaction data340to the transaction commit module220. If the transaction execution module210does not successfully execute the one or more statements of the transaction310, the transaction execution module210aborts execution of the transaction310and communicates with the transaction rollback module230to roll back the transaction310. In this case, the transaction rollback module230may perform a local rollback to remove any data describing the execution of the transaction310prior to aborting the transaction (e.g., delta tables or transaction logs). If the transaction can be retried, the transaction rollback module230adds the transaction310to an aborted transaction queue maintained by the query server200and the query server200tries to execute the transaction310again at a later time. If the transaction cannot be retried, such as if the transaction is malformed, then the transaction execution process by the query server200ends.

After receiving the modified transaction data340, the transaction commit module220then attempts to commit the modified transaction data340to one or more relevant data servers116(e.g., the data servers116storing the transaction data330). If the transaction commit module220successfully commits the modified transaction data340to the one or more relevant data servers116, the execution300end. If the transaction commit module220does not successfully commit the modified transaction data340to the relevant one or more relevant data servers116during the pending stage, the transaction commit module220aborts the execution300of the transaction310and communicates with the transaction rollback module230to perform at least a remote roll back of the transaction310. In this case, the transaction rollback module230adds the transaction to an aborted transaction queue so that the transaction310can be retried at a later time. If the transaction commit module220does not successfully commit the modified transaction data340to the one or more relevant data servers116during the committed stage, the transaction commit module220of the query serve200, or another query sever112, finishes committing the modified transaction data340during a cleanup process at a later time.

FIG.4Aillustrates an embodiment of a transaction400represented using a declarative query language. In the embodiment shown, the transaction400is a multi-statement transaction including statements represented using a declarative query language with SQL syntax. In other embodiments, the statements of the transaction400are represented using declarative query languages with other syntax or other query languages (e.g., procedural query languages). In the example depicted inFIG.4A, the transaction400includes five statements (i.e., statements405,410,415,420, and425). In other examples, the transaction400can include different statements than those depicted inFIG.4A, and furthermore can include additional or fewer statements.

As depicted inFIG.4A, the statements405and425of the transaction400signify the start and the committing of the transaction, respectively. In embodiments, the transaction400is assigned a transaction identifier during execution of the BEGIN WORK statement405. The statements410,415, and420are DML statements describing modifications of fields “a,” “b,” and “c” for a set of records “x1” (e.g., a collection of documents stored on one or more data servers116). In particular, the statements410and415are UPDATE statements which describe modifications (in particular, additions) to a value for the field “a” of records within “x1” where a value for the field “b” is less than ten and fifteen, respectively. The statement420is a SELECT statement which describes retrieving values for the fields “a,” “b”, and “c” from records of “x1” where the value for the field “b” is less than twenty. Depending on the embodiment, some statements of the transaction400can be executed in various orders or concurrently (e.g., by the transaction execution module210). For example, the additions corresponding to statement410and415can be performed in any order.

FIG.4Billustrates an embodiment of internal data structures maintained by a query server112during execution of the transaction400. In the embodiment shown, a query server (e.g., the query server200) maintains a delta table including delta tables entries430and440and a transaction log including transaction log entries435and445for the UPDATE statements410and415, respectively. The query server further maintains a projection table450and a transaction log entry455for the SELECT statement450. Further still, the query maintains a commit table460including the data to be committed based on execution of the statements410,415, and420. In other embodiments, the query server maintains other data structures during execution of the transaction400.

The delta table entries430and440include local copies of the records modified by the UPDATE statements410and415, respectively. The delta table entries430and440correspond to entries in a single delta table for the transaction310, where the entry for each record is added as statements describing modifications to the record are executed. In particular, the delta table entries430includes local copies of the values for the fields “a,” “b” and “c” of the records “k1,” “k2,” and “k3” (i.e., the records in “x1” where the value for the field “b” is less than ten) after the modification corresponding to the statement410is applied. Similarly, the delta table entries440includes local copies of the values for the fields “a,” “b” and “c” of the records “k1,” “k2,” “k3,” and “k12” (i.e., the records in “x1” where the value for the field “b” is less than fifteen) after the modification corresponding to the statement410is applied. As depicted, the delta table entries430and440indicate that the statement410was executed before the statement415, as the local copies of the values in the delta table entries440reflect the cumulative modifications of the statements410and415. In an alternative embodiment, the statement415is executed before the statement410. The query server112may have retrieved the records “k1,” “k2,” and “k3” from one or more data servers116during execution of the statement410. In this case, rather than re-retrieving the records, the query server112can use the local copies of the records “k1,” “k2,” and “k3” stored in the delta table entries430(or otherwise cached on the query server112) to execute the statement415. The delta table entries430and440additionally include CAS values for each of the records “k1,” “k2,” and “k3,” which are eventually transferred to the commit table460and can be used to prevent conflicts during the commit process, as described above with reference to the data servers116andFIG.1.

The transaction log entries435,445, and455are entries in a transaction log for the transaction400corresponding to the statements410,415, and420, respectively. The transaction log entries435,445, and455describe the specific modifications performed on a value of a field for each record in a delta table entry of a respective statement. The transaction log entries435,445, and455can be used by the transaction rollback module230to rollback aborted transactions, as described above with reference to the transaction rollback module230. In particular, the transaction log entry435includes the values for the records “k1,” “k2,” and “k3” before and after the modification corresponding to the statement410was applied. In the case of the transaction log entry435, the values for the records before the modification of the statement410was applied are the values stored by the distributed database system110before the transaction400was executed. Similarly, the transaction log entry445includes the values for the records “k1,” “k2,” “k3,” and “k12” before and after the modification corresponding to the statement410was applied. In the case of the transaction log445, the values for the records “k1,” “k2,” and “k3” before the statement415was applied are values reflecting the modifications corresponding to the statement410which was executed prior to the statement415. The transaction log entry455does not include any modifications of values for records in “x1” because the statement420does not describe any modifications. In the embodiment shown, the transaction log entry455is an empty data structure, while in other embodiments the query server112only creates transaction log data structures (e.g., allocate the requisite memory space) for statements describing modifications.

The projection table450includes local copies of the records selected by the SELECT statement420. In particular, the projection table450includes local copies of the values of the fields “a,” “b” and “c” of the records “k1,” “k2,” “k3,” “k5,” and “k12” (i.e., the records in “x1” where the value for the field “b” is less than twenty). In creating the projection table450, rather than re-retrieving the records “k1,” “k2,” “k3,” and “k12”, the query server112can use the local copies of the records “k1,” “k2,” “k3,” and “k12” stored in the delta tables430and440(or otherwise cached on the query server112) to execute the statement420. In particular, the query server112can apply the filter logic of statement420(i.e., WHERE b<20) to the delta table entries440in order to generate the projection table450.

The commit table460includes local copies of the records to be committed by the query server112. In particular, the commit table460includes local copies of the modified values of the fields “a,” “b” and “c” of the records “k1,” “k2,” “k3,” and “k12” (i.e., the records modified by the transaction400). The commit table460can be used to commit the modified records to one or more data servers116, such as using the commit process described above in reference to the transaction commit module220. The commit table460additionally includes the CAS values for the modified records, which can be used to avoid transaction conflicts during the commit process of the local copies of records in the commit table460, as described above with reference to the data servers116andFIG.1.

FIG.5is a flow chart illustrating an embodiment of a process500for executing a transaction in a distributed database. As described herein, the process500is performed by the distributed database system110. In other embodiments, other entities may perform some or all of the steps inFIG.5. Embodiments may also include different and/or additional steps or perform the steps in different orders.

In the embodiment shown inFIG.5, the method500begins with the distributed database system110receiving510a transaction including a statement describing a modification of a record stored at a data server of a distributed database (e.g., a data server116). For example, the distributed system110may receive a transaction from a client device120and select one of the query servers112to execute the transaction. The received transaction may be a single statement or multi-statement transaction, and the statement of the transaction be represented using a declarative query language or another type of query language. After receiving the transaction, the distributed database system110executes the transaction at s query server (e.g., a query server112). During execution of the transaction at the query server, the distributed database system110obtains520a copy of the record at the query server. For example, the transaction execution module210may retrieve the record from a data server116and generate a local copy of the retrieved record. As another example, if the record has previously been retrieved during execution of another statement of the transaction and a local copy of the record is cached at the query server (e.g., in a delta table corresponding to the transaction), the transaction execution module210can retrieve the cached local copy of the record. Using the obtained record copy, the distributed database system110performs530the modification corresponding to the statement on the record copy at the query server. For example, the statement may describe an update to the value of a field of the record, and the transaction execution module210may modify the value of the field of the record copy to reflect the update. The distributed database system110stores540the modified record copy at the query server. For example, the transaction execution module210may store the modified record copy in a delta table. After executing transaction at the query server, the distributed database system110updates550the record stored at the data server based on the modified record copy. In particular, the distributed database system110updates the record without conflicting with any other transactions executed by the distributed database system110(e.g., transactions being executed concurrently by the distributed database system110). For example, the transaction commit module220may update the record by performing a commit process for the transaction including a pending stage and a committed stage.

Computer Architecture

FIG.6is a high-level block diagram illustrating a functional view of a typical computer system for use as one of the entities illustrated in the system environment100ofFIG.1according to an embodiment. Illustrated are at least one processor602coupled to a chipset604. Also coupled to the chipset604are a memory606, a storage device608, a keyboard610, a graphics adapter612, a pointing device614, and a network adapter616. A display618is coupled to the graphics adapter612. In one embodiment, the functionality of the chipset604is provided by a memory controller hub620and an I/O controller hub622. In another embodiment, the memory606is coupled directly to the processor602instead of the chipset604.

The storage device608is a non-transitory computer-readable storage medium, such as a hard drive, compact disk read-only memory (CD-ROM), DVD, or a solid-state memory device. The memory606holds instructions and data used by the processor602. The pointing device614may be a mouse, track ball, or other type of pointing device, and is used in combination with the keyboard610to input data into the computer system600. The graphics adapter612displays images and other information on the display618. The network adapter616couples the computer system600to a network.

As is known in the art, a computer600can have different and/or other components than those shown inFIG.6. In addition, the computer600can lack certain illustrated components. For example, a computer system600acting as a server (e.g., a query server112) may lack a keyboard610and a pointing device614. Moreover, the storage device608can be local and/or remote from the computer600(such as embodied within a storage area network (SAN)).

The computer600is adapted to execute computer modules for providing the functionality described herein. As used herein, the term “module” refers to computer program instruction and other logic for providing a specified functionality. A module can be implemented in hardware, firmware, and/or software. A module can include one or more processes, and/or be provided by only part of a process. A module is typically stored on the storage device1008, loaded into the memory606, and executed by the processor602.

The types of computer systems600used by the entities ofFIG.1can vary depending upon the embodiment and the processing power used by the entity. For example, a client device120may be a mobile phone with limited processing power, a small display618, and may lack a pointing device614. The entities of the distributed database system110, in contrast, may comprise multiple blade servers working together to provide the functionality described herein.

Additional Considerations

Some portions of above description describe the embodiments in terms of algorithmic processes or operations. These algorithmic descriptions and representations are commonly used by those skilled in the computing arts to convey the substance of their work effectively to others skilled in the art. These operations, while described functionally, computationally, or logically, are understood to be implemented by computer programs comprising instructions for execution by a processor or equivalent electrical circuits, microcode, or the like. Furthermore, it has also proven convenient at times, to refer to these arrangements of functional operations as modules, without loss of generality.

As used herein, any reference to “one embodiment” or “an embodiment” means that a particular element, feature, structure, or characteristic described in connection with the embodiment is included in at least one embodiment. The appearances of the phrase “in one embodiment” in various places in the specification are not necessarily all referring to the same embodiment. Similarly, use of “a” or “an” preceding an element or component is done merely for convenience. This description should be understood to mean that one or more of the element or component is present unless it is obvious that it is meant otherwise.

Where values are described as “approximate” or “substantially” (or their derivatives), such values should be construed as accurate+/−10% unless another meaning is apparent from the context. From example, “approximately ten” should be understood to mean “in a range from nine to eleven.”

Upon reading this disclosure, those of skill in the art will appreciate still additional alternative structural and functional designs that may be used to employ the described techniques and approaches. Thus, while particular embodiments and applications have been illustrated and described, it is to be understood that the described subject matter is not limited to the precise construction and components disclosed. The scope of protection should be limited only by the following claims.