WO2022006794A1

WO2022006794A1 - Routing directives for partitioned databases

Info

Publication number: WO2022006794A1
Application number: PCT/CN2020/100948
Authority: WO
Inventors: Chen Fu; Le CAI; Feifei Li
Original assignee: Alibaba Group Holding Ltd
Current assignee: Alibaba Group Holding Ltd
Priority date: 2020-07-08
Filing date: 2020-07-08
Publication date: 2022-01-13
Anticipated expiration: 2023-01-08
Also published as: CN115917525A

Abstract

A current node may receive at least one query included in a distributed database transaction, and extract a name of a data table and a partition key from the query. The current node may then obtain an address of a particular node that includes a data shard of the data table corresponding to the partition key, and determine whether the current node is the particular node and whether the distributed database transaction including the query has been processed locally in one or more nodes. Based on a result of the determining, the current node may subsequently return an exception to a client device that submits the query for redirecting the query to the particular node, start to process a transaction associated with the query locally in the current node, or send the query to the particular node.

Description

ROUTING DIRECTIVES FOR PARTITIONED DATABASES

BACKGROUND

With the continuously increasing volume of data that is stored in every walk of life, such as company data, shopping data, personal data, etc., partitioned and distributed database management systems (PDBMS) running on the cloud or a network of servers have become increasingly popular. In a PDBMS, data tables are partitioned horizontally into multiple parts, commonly known as data shards. Each data shard is a collection of rows, and is independently hosted and replicated. Shards can be moved, split, or merged to improve performance and elasticity.

Transactions that involve data hosted on a single node (i.e. a database server) are called local transactions. These transactions are essentially no different from transactions in traditional monolithic database management systems. On the other hand, transactions involving data on multiple nodes, i.e., global transactions, need to go through a complex process called 2 phase commit (2PC) when a commit operation is performed. As a result, global transactions are significantly slower than local transactions.

To achieve better performance, database administrators (DBAs) often specify data partition and placement policies, in such a way so that most transactions can be performed locally. For instance, a database of an e‐commerce application may include warehouse and customer tables, etc. If most orders submitted to the e‐commerce application can be fulfilled by local warehouses, a database administrator may choose to partition the tables based on geographic locations, so that rows representing warehouses and customers in the same location are in the same shard of their respective tables. The database administrator may also specify data shards of the customer table and data shards of the warehouse table of the same location to be hosted on the same node. In this way, most orders can be served by executing local transactions to achieve better performance.

This effort of database partitioning can achieve the best performance if a PDBMS can bring computations to data, i.e., executing a transaction directly on a node that hosts or stores the data. However, it turns out to be difficult to do so for the following reasons: 1) a transaction is best executed by a single node, and aborting and restarting a transaction on another node is feasible, but often costly; 2) it is difficult to predict a node where relevant data is located before the transaction is started, especially when most transactions involve multiple queries. As a result, a node executing the transaction often does not host the data relevant to the transaction, and has to forward query requests to actual hosting nodes, thus incurring high communication costs. Worse still, since the transaction has been assigned to the node, the node would need to act as a control node to coordinate execution or processing of the multiple queries included in the transaction by other nodes, send a commit instruction to these other nodes to complete a 2 phase commit, and collect query results associated with execution or processing of the multiple queries from the other nodes, and send the query results to a client device of a database administrator for presentation or review, thus further incurring communication costs and time due to transmission of query results and instructions between the control node and the other nodes.

SUMMARY

This summary introduces simplified concepts of routing directives for partitioned databases, which will be further described below in the Detailed Description. This summary is not intended to identify essential features of the claimed subject matter, nor is it intended for use in limiting the scope of the claimed subject matter.

This disclosure describes example implementations of routing directives for partitioned databases. In implementations, a current node may receive at least one query included in a distributed database transaction, and extract a name of a data table and a partition key from the query. In implementations, the current node may then obtain an address of a particular node that includes a data shard of the data table corresponding to the partition key, and determine whether the current node is the particular node and whether the distributed database transaction including the query has been processed locally in one or more nodes. In implementations, based on a result of the determining, the current node may subsequently return an exception to a client device that submits the query for redirecting the query to the particular node, start to process a transaction associated with the query locally in the current node, or send the query to the particular node.

BRIEF DESCRIPTION OF THE DRAWINGS

The detailed description is set forth with reference to the accompanying figures. In the figures, the left‐most digit (s) of a reference number identifies the figure in which the reference number first appears. The use of the same reference numbers in different figures indicates similar or identical items.

FIG. 1 illustrates an example environment in which a database system may be used.

FIG. 2 illustrates an example computing node in more detail.

FIG. 3 illustrates an example load balancing node in more detail.

FIG. 4 illustrates an example method of processing a distributed database transaction.

FIG. 5 illustrates another example method of processing a distributed database transaction.

FIG. 6 illustrates an example scenario of processing a distributed database transaction.

DETAILED DESCRIPTION

Overview

As noted above, existing partitioned and distributed database systems adopt loading balancing strategies that fail to deterministically or strategically assign a distributed database transaction to a particular node that includes one or more data tables involved in the database transaction, and a node that is assigned with the database transaction may need to act as a control or coordinate node to coordinate other nodes that include data tables involved in the distributed database transaction to process the distributed database transaction and to commit synchronously the distributed database transaction. This not only increases communication costs and time due to data and instructions transmitted among the nodes, but also wastes processing resources of the control or coordinate node that could be needless for the distributed database transaction.

This disclosure describes an example database system. In implementations, a load balancing node or server of the example database system may receive a data packet including a distributed database transaction from a client device. In implementations, the distributed database transaction may include multiple queries that involve one or more data tables included in one or more computing nodes of the example database system. In implementations, the distributed database transaction may include a routing directive indicating a particular data table and a partition key. The routing directive may be located at a particular position (such as at the beginning of the distributed database transaction and before the multiple queries) in the distributed database transaction and may include a designated or reserved term indicating the nature of the routing directive. In implementations, the load balancing node may read the routing directive from the distributed database transaction, and determine a computing node that include a data section of the particular data table corresponding to the partition key based on a predetermined mapping relationship. The load balancing node may then assign the distributed database transaction to the determined computing node, and send the distributed database transaction to the determined computing node for subsequent processing and communication with the client device.

Additionally, in implementations, the load balancing node of the example database system may receive a data packet including a distributed database transaction. The distributed database transaction may include a plurality of queries directing to one or more data tables included in one or more computing nodes of the database system. In implementations, the load balancing node may assign and send the distributed database transaction to a computing node based on a predetermined load balancing strategy. By way of example and not limitation, the predetermined load balancing strategy may include randomly assigning the distributed database transaction to a computing node, assigning the distributed database transaction to a computing node in a round‐robin manner, assigning the distributed database transaction to a computing node that currently has the least workload, assigning to the distributed database transaction to a computing node based on a mapping relationship between an IP address of the client device and the computing node, etc.

In implementations, the computing node that receives the distributed database transaction may first determine whether a corresponding data section of a data table involved in the first query or any one of the plurality of queries of the distributed database transaction exists in the computing node. If affirmative, the computing node may process the distributed database transaction. Otherwise, the computing node may send an exception to the client device, the exception including a routing directive to redirect the client device to a correct computing node that includes a data section of a data table involved in the first query or any one of the plurality of queries of the distributed database transaction for processing the distributed database transaction. In implementations, the exception sent to the client device may include information of the correct computing node, such as an identifier of the correct computing node, a virtual address of the correct computing node, such as the load balancing node, for example, of the database system may assign the distributed database transaction to the correct computing node after receiving a new request for processing the distributed database transaction from the client device.

As described above, the example database system can deterministically or strategically assign or redirect a request for processing a distributed database transaction from a client device to a computing node that includes a data section of a data table involved in at least one query of the distributed database transaction, thus avoiding the use of a control or coordinate node and hence reducing the communication costs and resource waste.

In implementations, functions described herein to be performed by the database system may be performed by multiple separate units or services. Moreover, although in the examples described herein, the database system may be implemented as a combination of software and hardware implemented and distributed in multiple devices, in other examples, the database system may be implemented and distributed as services provided in one or more computing devices over a network and/or in a cloud computing architecture.

The application describes multiple and varied embodiments and implementations. The following section describes an example framework that is suitable for practicing various implementations. Next, the application describes example systems, devices, and processes for implementing a database system.

Example Environment

FIG. 1 illustrates an example environment 100 usable to implement a database system. The environment 100 may include a database system 102. In implementations, the database system 102 may include a plurality of servers or computing nodes 104‐1, 104‐2, …, 104‐N (which are collectively called as computing nodes 104) . The computing nodes 104 may communicate data with one another via a network 106. In implementations, the database system 102 may further include at least one load balancing node 108 for allocating workload to the server or computing nodes 104. In implementations, one or more servers or computing nodes 104 may be designated or used as the at least one load balancing node 108.

In implementations, each of the servers or computing nodes 104 may be implemented as any of a variety of computing devices, but not limited to, a desktop computer, a notebook or portable computer, a handheld device, a netbook, an Internet appliance, a tablet or slate computer, a mobile device (e.g., a mobile phone, a personal digital assistant, a smart phone, etc. ) , a server computer, etc., or a combination thereof.

The network 106 may be a wireless or a wired network, or a combination thereof. The network 106 may be a collection of individual networks interconnected with each other and functioning as a single large network (e.g., the Internet or an intranet) . Examples of such individual networks include, but are not limited to, telephone networks, cable networks, Local Area Networks (LANs) , Wide Area Networks (WANs) , and Metropolitan Area Networks (MANs) . Further, the individual networks may be wireless or wired networks, or a combination thereof. Wired networks may include an electrical carrier connection (such a communication cable, etc. ) and/or an optical carrier or connection (such as an optical fiber connection, etc. ) . Wireless networks may include, for example, a WiFi network, other radio frequency networks (e.g.,

Zigbee, etc. ) , etc.

In implementations, the environment 100 may further include a client device 110. The client device 110 may be implemented as any of a variety of computing devices, but not limited to, a desktop computer, a notebook or portable computer, a handheld device, a netbook, an Internet appliance, a tablet or slate computer, a mobile device (e.g., a mobile phone, a personal digital assistant, a smart phone, etc. ) , a server computer, etc., or a combination thereof.

In implementations, the database system 102 may receive a request for processing a distributed database transaction from the client device 110. For example, a user 112 (such as a database administrator, etc. ) of the client device 110 may submit a plurality of queries involving one or more data tables as a distributed database transaction to the database system 102. In response to receiving the request, the database system 102 may determine at least one computing node including a data section of a data table involved in the distributed database transaction based on a routing directive, send the distributed database transaction to the at least one computing node for processing, and return a result of processing the distributed database transaction to the client device 110.

Example Partitioned Database

In implementations, the database system 102 may further include one or more databases that are partitioned or distributed among the plurality of computing nodes 104. By way of example and not limitation, a partitioned or distributed database may include one or more data tables that are divided or partitioned, with each data table being divided horizontally into multiple parts called data shards or simply shards. In implementations, each shard may include a collection of rows, and may be independently hosted and replicated in one or more computing nodes 104. Furthermore, shards can be moved, split, or merged to improve performance and elasticity of the database system 102.

In implementations, transactions that involve data hosted on a single computing node or server 104 (such as a database server) may be called local transactions. They are essentially no different from the transactions in the traditional monolithic DBMSs. On the other hand, transactions involving data on multiple computing nodes, i.e.global transactions, need to go through a complex process called 2 phase commit (2PC) when committing. As a result, processing of global transactions is significantly slower than processing of local transactions.

In implementations, a data table may be partitioned horizontally based on a value of a column, or a combination of multiple columns. In the latter case, values from each of the columns may form a tuple, and these columns may be treated as a single logical column for the purpose of partitioning. Without the loss of generality and for the sake of simplicity, tables are assumed to be partitioned based on a partition column.

In implementations, rows having a same value in a partition column may be placed in a same data section or shard. For example, an e‐commerce application may include a database having a customer table, which includes columns such as street, city, state, zip code. These columns may form an address of a customer. A user 112 (such as a database administrator) may partition the customer table using a column associated with the zip code alone, so that all rows having a same zip code in the customer table are located in a same data section or shard.

In implementations, the database system 102 may further include or provide a partition function for each data table, part _table (k) →shard_id, where k is called a partition key, which is a value of a partition column. In implementations, an output of this function may an identifier (i.e., ID) of a shard including all the rows having a value of a partition column to be equal to the partition key (i.e., k) .

In implementations, the database system 102 may allow the user 112 to specify one or more partition and placement policies for data stored in the partitioned databases. Continuing the above example of the e‐commerce application, a warehouse table may further be included in addition to the customer table. As most orders for this e‐commerce application are probably fulfilled from local warehouses, the user 112 may decide to partition a database based on geographic locations. In this case, zip codes of locations of warehouses and zip codes of customer addresses may be used as corresponding partition columns for the warehouse table and the customer table respectively. Furthermore, the user 112 may instruct the database system 102 to allow rows that represent warehouses and customers in the same zip code to be hosted or stored in the same node, so that most order processing may incur local transactions. In other words, both functions part _warehouse (k) and part _customer (k) takes a zip code as a partition key, and

wherein place (. ) is a placement function.

In implementations, the partition function and the placement function may be implemented by querying metadata that includes a mapping from partition keys to shard IDs, and a mapping from the shard IDs to computing nodes when corresponding shards are hosted or stored. This metadata may be replicated to some or all of the computing nodes 104 and/or the load balancing node 108 in the database system 102.

Example Computing Node

FIG. 2 illustrates the computing node 104 in more detail. In implementations, the computing node 104 may include, but is not limited to, one or more processors 202, an input/output (I/O) interface 204, and/or a network interface 206, and memory 208. In implementations, some of the functions of the computing node 104 may be implemented using hardware, for example, an ASIC (i.e., Application‐Specific Integrated Circuit) , a FPGA (i.e., Field‐Programmable Gate Array) , and/or other hardware.

In implementations, the processors 202 may be configured to execute instructions that are stored in the memory 208, and/or received from the I/O interface 204, and/or the network interface 206. In implementations, the processors 202 may be implemented as one or more hardware processors including, for example, a microprocessor, an application‐specific instruction‐set processor, a physics processing unit (PPU) , a central processing unit (CPU) , a graphics processing unit, a digital signal processor, a tensor processing unit, etc. Additionally or alternatively, the functionality described herein can be performed, at least in part, by one or more hardware logic components. For example, and without limitation, illustrative types of hardware logic components that can be used include field‐programmable gate arrays (FPGAs) , application‐specific integrated circuits (ASICs) , application‐specific standard products (ASSPs) , system‐on‐a‐chip systems (SOCs) , complex programmable logic devices (CPLDs) , etc.

The memory 208 may include computer readable media in a form of volatile memory, such as Random Access Memory (RAM) and/or non‐volatile memory, such as read only memory (ROM) or flash RAM. The memory 208 is an example of computer readable media.

The computer readable media may include a volatile or non‐volatile type, a removable or non‐removable media, which may achieve storage of information using any method or technology. The information may include a computer readable instruction, a data structure, a program module or other data. Examples of computer readable media include, but not limited to, phase‐change memory (PRAM) , static random access memory (SRAM) , dynamic random access memory (DRAM) , other types of random‐access memory (RAM) , read‐only memory (ROM) , electronically erasable programmable read‐only memory (EEPROM) , quick flash memory or other internal storage technology, compact disk read‐only memory (CD‐ROM) , digital versatile disc (DVD) or other optical storage, magnetic cassette tape, magnetic disk storage or other magnetic storage devices, or any other non‐transmission media, which may be used to store information that may be accessed by a computing device. As defined herein, the computer readable media does not include any transitory media, such as modulated data signals and carrier waves.

Although in this example, only hardware components are described in the computing node 104, in other instances, the computing node 104 may further include other hardware components and/or other software components such as program units to execute instructions stored in the memory 208 for performing various operations. For example, the computing node 104 may further include a local or partitioned database 210 for storing data tables and other program data 212. By way of example and not limitation, the computing node 104 may store data sections or shards of one or more data tables in the local or partitioned database 210. In implementations, the one or more data tables may be divided and distributed according to respective partition keys among different computing nodes 104.

Example Load Balancing Node

FIG. 3 illustrates the load balancing node 108 in more detail. In implementations, the load balancing node 108 may include, but is not limited to, one or more processors 302, an input/output (I/O) interface 304, and/or a network interface 306, and memory 308. In implementations, some of the functions of the load balancing node 108 may be implemented using hardware, for example, an ASIC (i.e., Application‐Specific Integrated Circuit) , a FPGA (i.e., Field‐Programmable Gate Array) , and/or other hardware.

In implementations, the processors 302 may be configured to execute instructions that are stored in the memory 308, and/or received from the I/O interface 304, and/or the network interface 306. In implementations, the processors 302 may be implemented as one or more hardware processors including, for example, a microprocessor, an application‐specific instruction‐set processor, a physics processing unit (PPU) , a central processing unit (CPU) , a graphics processing unit, a digital signal processor, a tensor processing unit, etc. Additionally or alternatively, the functionality described herein can be performed, at least in part, by one or more hardware logic components. For example, and without limitation, illustrative types of hardware logic components that can be used include field‐programmable gate arrays (FPGAs) , application‐specific integrated circuits (ASICs) , application‐specific standard products (ASSPs) , system‐on‐a‐chip systems (SOCs) , complex programmable logic devices (CPLDs) , etc.

The memory 308 may include computer readable media in a form of volatile memory, such as Random Access Memory (RAM) and/or non‐volatile memory, such as read only memory (ROM) or flash RAM. The memory 308 is an example of computer readable media as described in the foregoing description.

Although in this example, only hardware components are described in the load balancing node 108, in other instances, the load balancing node 108 may further include other hardware components and/or other software components such as program units to execute instructions stored in the memory 308 for performing various operations. For example, the load balancing node 108 may further include a mapping table 310 and other program data 312. In implementations, the mapping table 310 may include mapping from a combination of information of a data table and information of a partition key to information of a computing node that includes a data section or shard of the data table corresponding to the partition key. By way of example and not limitation, given a name of a data table and a value of a partition key, the load balancing node 108 may determine an identifier or address of a computing node that includes a data section or shard of the data table corresponding to the value of the partition key.

In implementations, the load balancing node 108 may obtain this mapping table 310 in advance, for example, by receiving broadcasting information from the plurality of computing nodes 104. In implementations, broadcasting information of each computing node 104 may include, but is not limited to, information about data section (s) or shard (s) of data table (s) that correspond to certain values of partition key (s) is/are included or stored in the respective computing node 104.

Additionally or alternatively, in some implementations, the load balancing node 108 may be associated with a mapping device (which may be a server or computing device provided in the database system 102) . The mapping device may collect information about data section (s) or shard (s) of data table (s) that correspond to certain values of partition key (s) is/are included or stored in each computing node 104 from the plurality of computing nodes 104, for example, by broadcasting information from the plurality of computing nodes 104 as described above. In this case, the load balancing node 108 may send information (such as a name) of a data table and information (such as a value) of a partition key obtained from a distributed database transaction to the mapping device, which maps the information of the data table and the information of the partition key to information (e.g., an identifier or address) of a computing node that includes a data section or shard of the data table corresponding to the partition key. The load balancing node 108 may then receive the information of the computing node from the mapping device, thus reducing the workload and complexity of the load balancing node while enabling the load balancing node to continuously process load balancing of incoming requests or data packets.

In implementations, the load balancing node 108 may include one or more predetermined load balancing strategies. By way of example and not limitation, the one or more predetermined load balancing strategies may include assigning a distributed database transaction received (e.g., from the client device 110) to a computing node in a random manner, assigning the distributed database transaction to a computing node in a round‐robin manner, assigning the distributed database transaction to a computing node that currently has the least workload, assigning to the distributed database transaction to a computing node based on a mapping relationship between an IP address of the client device and the computing node, etc.

Example Methods

FIG. 4 shows a schematic diagram depicting an example method of processing a distributed database transaction. FIG. 5 shows a schematic diagram depicting another example method of processing a distributed database transaction. The methods of FIG. 4 and FIG. 5 may, but need not, be implemented in the environment of FIG. 1 and using the computing node and the load balancing node of FIG. 2 and FIG. 3. For ease of explanation,

methods

400 and 500 are described with reference to FIGS. 1‐3. However, the

methods

400 and 500 may alternatively be implemented in other environments and/or using other systems.

The

methods

400 and 500 are described in the general context of computer‐executable instructions. Generally, computer‐executable instructions can include routines, programs, objects, components, data structures, procedures, modules, functions, and the like that perform particular functions or implement particular abstract data types. Furthermore, each of the example methods are illustrated as a collection of blocks in a logical flow graph representing a sequence of operations that can be implemented in hardware, software, firmware, or a combination thereof. The order in which the method is described is not intended to be construed as a limitation, and any number of the described method blocks can be combined in any order to implement the method, or alternate methods. Additionally, individual blocks may be omitted from the method without departing from the spirit and scope of the subject matter described herein. In the context of software, the blocks represent computer instructions that, when executed by one or more processors, perform the recited operations. In the context of hardware, some or all of the blocks may represent application specific integrated circuits (ASICs) or other physical components that perform the recited operations.

Referring back to FIG. 4, at block 402, the load balancing node 108 or the database system 102 may receive a request for processing a distributed database transaction from the client device 110.

In implementations, the load balancing node 108 may receive a request for processing a distributed database transaction from the client device 110. In implementations, the distributed database transaction may include one or more queries that refer to one or more data tables located or stored in one or more computing nodes 104 in the database system 102. In implementations, the distributed database transaction may further include a designated statement indicating a data table (e.g., a name of the data table) and a partition key (e.g., a value of the partition key) . In implementations, the designated statement may include a reserved term (e.g., “HINT” ) indicating the nature of the designated statement.

In implementations, the load balancing node 108 may attempt to read the distributed database transaction, and determine whether a designated statement including a reserved term is included in the distributed database transaction. In response to determining that a designated statement including a reserved term (such as “HINT” ) is included in the distributed database transaction, the load balancing node 108 may further extract a name of a data table and a value of a partition key from the designated statement.

Alternatively, the load balancing node 108 may determine that the distributed database transaction does include a designated statement (such as a statement including a reserved term “HINT” ) .

At block 404, the load balancing node 108 may determine a computing node that includes a data shard of a data table corresponding to a value of a partition key included in the distributed database transaction.

In implementations, if the designated statement is included in the distributed database transaction, after extracting the name of the data table and the value of the partition key from the designated statement, the load balancing node 108 may determine a computing node including or storing a data section or shard of the data table corresponding to the value of the partition key. By way of example and not limitation, the load balancing node 108 may determine the computing node including or storing the data section or shard of the data table corresponding to the value of the partition key based on the mapping table 310. For example, the mapping table 310 may store mapping relationships between partition keys and computing nodes in which individual data shards of a data table separately corresponding to different values of partition keys.

Additionally or alternatively, a partition function and a placement function may be determined and stored in the load balancing node 108 in advance. The load balancing node 108 may apply the partition function on the name of the data table and the value of the partition key to obtain an identifier (i.e., ID) of a data shard of the data table corresponding to the value of the partition key. The load balancing node 108 may then apply the placement function on the identifier of the data shard to obtain a location (e.g., an address) or an identifier of a computing node storing or including the data shard.

Additionally or alternatively, the load balancing node 108 may send an inquiry including the name of the data table and the value of the partition key that are extracted from the designated statement to the mapping device that is associated with the load balancing node 108, and receive a location (e.g., an address) or an identifier of a computing node storing or including a data shard of the data table corresponding to the value of the partition key from the mapping device.

In implementations, in response to determining that no designated statement (such as a designated statement having a reserved term “HINT’ ) is included in the distributed database transaction, the load balancing node 108 may read and parse the distributed database transaction to determine a first query. By way of example and not limitation, the load balancing node 108 may determine the first query by detecting an occurrence of a reserved term (such as “SELECT” , “UPDATE” , etc. ) for a query statement (such as a “SELECT” query statement, a “UPDATE” query statement, etc. ) , and determine a name of a data table and a value of a partition key from the first query. The load balancing node 108 may then determine a computing node including or storing a data section or shard of the data table corresponding to the value of the partition key as described in the foregoing description.

In implementations, depending on the processing capabilities and workload, the load balancing node 108 may send the distributed database transaction to another load balancing node for parsing the first query from the distributed database transaction and determining a computing node including or storing a data section or shard of the data table corresponding to the value of the partition key as described in the foregoing description. The load balancing node 108 may or may not receive a result of which computing node that includes or stores a data section or shard of the data table corresponding to the value of the partition key from the other load balancing node.

Alternatively, in response to determining that no designated statement (such as a designated statement having a reserved term “HINT’ ) is included in the distributed database transaction, the load balancing node 108 may determine a computing node for processing the distributed database transaction based on one or more predetermined load balancing strategies. By way of example and not limitation, the one or more predetermined load balancing strategies may include assigning the distributed database transaction to a computing node in a random manner, assigning the distributed database transaction to a computing node in a round‐robin manner, assigning the distributed database transaction to a computing node that currently has the least workload, etc.

At block 406, the load balancing node 108 may assign and send the distributed database transaction to the determined computing node.

In implementations, depending on whether a computing node including or storing a data section or shard of a data table corresponding to a value of a partition key is determined (e.g., a designated statement indicating a name of the data table and the value of the partition key is included in the distributed database transaction) , the load balancing node 108 may assign or send the distributed database transaction to the computing node including or storing the data section or shard of the data table corresponding to the value of the partition key. In implementations, the load balancing node 108 may send or assign the distributed database transaction to the computing node to enable the computing node to act as a coordinate node to manage local transactions of the plurality of queries in one or more computing nodes. Alternatively, the load balancing node 108 may assign or send the distributed database transaction to the computing node that is determined based on the one or more load balancing strategies.

Referring to FIG. 5, at block 502, a first computing node 104‐1 of the database system 102 may receive at least one query included in a distributed database transaction.

In implementations, depending on an input source, the first computing node 104‐1 of the database system 102 may receive at least one query included in a distributed database transaction or the entire distributed database transaction. For instance, the first computing node 104‐1 may receive a distributed database transaction from the load balancing node 108 of the database system 102. In this case, the first computing node 104‐1 may be assigned or selected by the load balancing node 108 to process the distributed database transaction based one or more load balancing strategies as described above. Alternatively, the first computing node 104‐1 may be assigned or selected by the load balancing node 108 to process the distributed database transaction after the load balancing node 108 determines that the first computing node 104‐1 includes or stores a data section or shard of a data table corresponding to a value of a partition key involved in a query (e.g., the first query) in the distributed database transaction as described in the foregoing description.

In implementations, the first computing node 104‐1 may receive at least one query included in a distributed database transaction or the distributed database transaction from another computing node 104 (such as a second computing node 104‐2 which is different from the first computing node 104‐1) . In implementations, the second computing node 104‐2 may process a certain query included in the distributed database transaction, and determine that a data section or shard of a data table involved in that query is included or stored in the first computing node 104‐1. The second computing node 104‐2 may then send that query of the distributed database transaction to the first computing node 104‐1 for processing.

At block 504, the computing node 104 may obtain a name of a data table and a value of a partition key that are involved in the query.

In implementations, the first computing node 104‐1 may parse the query, and obtain a name of a data table and a value of a partition key from the query. In implementations, the value of the partition key may include respective values of one or more columns of the data table.

At block 506, the first computing node 104‐1 may obtain an identifier or an address of a particular computing node that includes a data shard of the data table corresponding to the value of the partition key.

In implementations, the first computing node 104‐1 may determine a particular computing node including or storing a data section or shard of the data table corresponding to the value of the partition key. By way of example and not limitation, the first computing node 104‐1 may determine the particular computing node including or storing the data section or shard of the data table corresponding to the value of the partition key based on a mapping table included in the first computing node 104‐1. In implementations, such mapping table may be similar to the mapping table 310, and may store mapping relationships between partition keys and computing nodes in which individual data shards of a data table separately corresponding to different values of partition keys.

Additionally or alternatively, a partition function and a placement function may be determined and stored in the first computing node 104‐1 in advance. The first computing node 104‐1 may apply the partition function on the name of the data table and the value of the partition key to obtain an identifier (i.e., ID) of a data shard of the data table corresponding to the value of the partition key. The first computing node 104‐1 may then apply the placement function on the identifier of the data shard to obtain a location (e.g., an address) or an identifier of the particular computing node storing or including the data shard.

Additionally or alternatively, the first computing node 104‐1 may send an inquiry including the name of the data table and the value of the partition key that are extracted from the designated statement to a mapping device that is associated with the first computing node 104‐1, and receive a location (e.g., an address) or an identifier of the particular computing node storing or including a data shard of the data table corresponding to the value of the partition key from the mapping device.

Additionally or alternatively, the first computing node 104‐1 may obtain a location (e.g., an address) or an identifier of the particular computing node that includes the data shard of the data table corresponding to the value of the partition key by obtaining an identifier of the data shard based on a first mapping relationship between the data shard and a combination of the data table and the value of the partition key, and obtain the address of the particular node based on a second mapping relationship between the address of the particular node and the identifier of the data shard.

At block 508, the first computing node 104‐1 may determine whether the first computing node 104‐1 is the particular computing node and whether the distributed database transaction including the query has been processed locally in one or more computing nodes.

In implementations, the first computing node 104‐1 may compare an address or identifier of the first computing node 104‐1 with the address or identifier of the particular computing node to determine whether the first computing node 104‐1 is the particular computing node. Additionally, the first computing node 104‐1 may also determine whether the distributed database transaction including the query has been processed locally in the first computing node and/or one or more other computing nodes in the database system 102.

At block 510, the first computing node 104‐1 may return an exception to a client device that submits the query or the distributed database transaction to redirect the query or the distributed database transaction to the particular computing node.

At block 512, the first computing node 104‐1 may start to process a transaction associated with the query locally in the first computing node

At block 514, the first computing node 104‐1 may send or relay the query or the distributed database transaction to the particular computing node.

In implementations, the first computing node 104‐1 may return the exception to the client device that submits the query or the distributed database transaction to redirect the query or the distributed database transaction to the particular computing node in response to determining that the first computing node is not the particular computing node and the distributed database transaction including the query has not been processed locally in the one or more computing nodes. In implementations, the exception may include a virtual address or an identifier of the particular computing node.

In implementations, the first computing node 104‐1 may start to process the distributed database transaction including the query locally in the first computing node 104‐1 in response to determining that the first computing node 104‐1 is the particular computing node and the distributed database transaction including the query has not been processed locally in the one or more computing nodes. For example, the first computing node 104‐1 may read the query to determine a type of the query (e.g., a “SELECT” statement, a “UPDATE” statement, etc. ) , and perform a corresponding operation on a data section or shard of the data table stored or hosted in the first computing node according to the query.

In implementations, the first computing node 104‐1 may send the query to the particular computing node in response to determining that the transaction including the query has been processed locally in the one or more computing nodes.

At block 516, the first computing node may commit a local transaction that is processed for the query in coordination with one or more other computing nodes that process other queries included in the distributed database transaction.

In implementations, the distributed database transaction may include a plurality of queries, and at least some of the plurality of queries may involve data shard (s) that is/are stored in one or more other computing nodes. In this case, the first computing node 104‐1 may need to wait for the one or more other computing nodes to complete processing of other queries included in the distributed database transaction before committing a local transaction that the first computing node 104‐1 performs for the query.

By way of example and not limitation, the first computing node 104‐1 may act as a coordinate or control node that manages and/or coordinate processing of local transactions of the plurality of queries included in the distributed database transaction. In implementations, upon receiving signals indicating that the other queries has been processed by the one or more other computing nodes from the one or more other computing nodes, the first computing node may signal a commit instruction to the one or more computing nodes, and commit the local transaction that has been processed for the query by the first computing node 104‐1 in coordination with the one or more other computing nodes that process the other queries included in the distributed database transaction, thus performing a 2 phase commit (2PC) .

Alternatively, one of the other computing nodes, which is different from the first computing node, may act as a coordinate or control node that manages and/or coordinate processing of local transactions of the plurality of queries included in the distributed database transaction. After the first computing node 104‐1 completes processing a local transaction associated with the query that involves a data shard of the data table included or stored in the first computing node 104‐1, the first computing node 104‐1 may send a signal to the coordinate or control node, indicating that the corresponding local transaction for the query is completed and waiting for a commit instruction to commit the local transaction. Upon receiving a commit instruction from the coordinate or control node, the first computing node may commit the local transaction in coordination with the one or more other computing nodes that process the other queries included in the distributed database transaction, thus performing a 2 phase commit (2PC) .

At block 514, after committing the local transaction, the first computing node 104‐1 may send out a result of processing the local transaction.

In implementations, depending on whether the first computing node 104‐1 is a coordinate or control node, the first computing node 104‐1 may send a result of processing the local transaction to another computing node that acts as the coordinate or control node, or a client device from which the distributed database transaction is initiated or sent. For example, if the first computing node 104‐1 is a coordinate or control node, the first computing node 104‐1 may collect results of processing local transactions associated with other queries of the plurality of queries by the one or more other computing nodes from the one or more other computing nodes, and send the collected results and its result of processing the local transaction as an aggregated or final result to the client device that initiates or sends the distributed database transaction. Alternatively, if the first computing device is not a coordinate or control node, the first computing node may send its result of processing the local transaction to another computing node that acts as the coordinate or control node, which then send an aggregated or final result to the client device from which the distributed database transaction is initiated or sent, after collecting all results of processing local transactions associated with the plurality of queries in the distributed database transactions.

Although the above method blocks are described to be executed in a particular order, in some implementations, some or all of the method blocks can be executed in other orders, or in parallel.

Example Scenarios

In implementations, routing directives may be messages that are exchanged between a client (such as the client device 110) and a server (such as a computing node 104) of a partitioned and distributed database management system (such as the database system 102) , and actions the client and the server may perform in association with these messages. Two example routing directives, namely, REDIRECT and HINT are used introduced herein for illustration. Apparently, other routing directives may be introduced and used by the database system 102, which are not limited in the present disclosure.

In implementations, a REDIRECT message is a message that is sent from a server (e.g., a first computing node 104‐1) to a client (e.g., the client device 110) . By way of example and not limitation, a format for a REDIRECT message may include REDIRECT <node address>, which indicates to the client to reconnect to another server (e.g., a second computing node 104‐2) of a partitioned database system (such as the database system 102) , and restart a current distributed transaction that is initially assigned to the server (i.e., the first computing node) .

In implementations, arbitrarily restarting an ongoing transaction may cause performance degradation, as rolling back the current distributed transaction would cause time and resource wastes. Therefore, restarting a current distributed transaction initiated by a client device and instructing the client device to redirect the current distributed transaction to another computing node may be selectively performed based on a timing of the current distributed transaction. By way of example and not limitation, a restarting or redirection of a current distributed transaction may (only) happen at the very beginning of a distributed transaction including the current distributed transaction, as shown in the following algorithm described in Table 1 as follows.

Table 1: Redirecting Algorithm

As can be seen from the above table, the computing node may first parse the distributed transaction to obtain a name of a data table and a value of a partition key that appear in the distributed transaction (e.g., a first query of the distributed transaction) . After obtaining the name of the data table and the value of the partition key, the computing node may locate a particular computing node that includes or stores a data section or shard of the data table corresponding to the value of the partition key by inputting the name of the data table and the value of the partition key as parameters to a partition function to obtain an identifier of a data section or shard, and inputting the identifier of the data section or shard as a parameter to a placement function to obtain an address or identifier of the particular computing node that includes or stores the data section or shard. The computing node may then determine whether the computing node is the particular computing node and whether the distributed transaction has been processed locally in one or more computing nodes. Based on a result of such determination, the computing node may return or raise an exception to the client device that initiates the distributed transaction to instruct the client device to redirect or resend the distributed database transaction to the particular computing node, or start to process a local transaction associated with the distributed transaction locally in the computing node, or send the distributed transaction to the particular computing node.

In implementations, a function raise_exception () represents a database functionality that abort a current distributed transaction, and inform a client device to know about such abortion, with a detailed error message given in a parameter list. In implementations, a REDIRECT message may be included or piggybacked on this exception function by the computing node (e.g., the first computing node 104‐1) , and sent to the client device when the computing node aborts the current distributed transaction. In this example, since no local transaction associated with the current distributed transaction has been started (i.e., no lock is obtained, or no modification is made) in the computing node and other computing nodes of the database system, an overhead for aborting is minimal. Therefore, a data structure representing the current distributed transaction may be released by the computing node, and no computing node needs to be informed to abort any associated local transactions, as no local transactions associated with the current distributed transactions are started and processed by other computing nodes.

In implementations, the client device may be augmented or extended to recognize the REDIRECT message, and recognize that the distributed current transaction has been aborted from the REDIRECT message. In implementations, the client device may extract information of the other computing node (e.g., the second computing node that includes or stores data shard (s) involved in the first query of the distributed transaction) , and connect to the other computing node specified by the REDIRECT message, to restart the distributed transaction.

For example, FIG. 6 illustrates a connection process 600 with a REDIRECT message. An example distributed transaction including a plurality of queries is given as follows:

BEGIN

UPDATE warehouse

SET ytd += in_payment_amount

WHERE warehouse_id = in_w_id AND warehouse_zip = in_w_zip;

UPDATE customer

SET balance ‐= in_payment_amount

WHERE customer_id = in_c_id AND customer_zip = in_c_zip;

COMMIT

At S602, after load balancing by a load balancer (e.g., the load balancing node 108) , a client device (e.g., the client device 110) may establish a connection channel with a first computing node 104‐1, and sends over a “BEGIN” statement to the first computing node 104‐1 via the connection channel, at which point a distributed transaction is started. Upon receiving a first “UPDATE” statement, the first computing node 104‐1 may finds that the first “UPDATE” statement is the very first query of the current distributed transaction. In this example, the first computing node 104‐1 is not one of hosting nodes that include data shards to respond or answer such query. Therefore, at S604, the first computing node 104‐1 raises an exception, sends back “REDIRECT Second Computing Node” , and aborts the current distributed transaction. At S606, upon receiving the exception, the client device 110 may connect to the second computing node 104‐2, and resends the “BEGIN” statement to the second computing node, which starts a new distributed transaction. Upon receiving the first “UPDATE” statement, the second computing node determines and finds that data needed by this first “UPDATE” statement is available locally in the second computing node. The second computing node may therefore start a local transaction, and then proceed with executing the query of such first “UPDATE” statement locally in the second computing node.

In the above example scenario of FIG. 6, the first computing node 104‐1 does not really participate in the distributed transaction initiated by the client device 110, as no query of the distributed transaction is executed by the first computing node 104‐1. Therefore, first computing node 104‐1 does not need to relay query results associated with such distributed transaction. As most distributed transactions include multiple queries, REDIRECT messages may therefore save communication cost in a vast majority of situations.

In implementations, a second type of routing directive is a routing HINT message that is sent from a client (e.g., the client device 110) to a load balancer (e.g., a load balancing node 108 of the database system 102) . In implementations, a routing hint message may be started with a BEGIN statement, with a format: BEGIN HINT <table name> <partition key>.

Unlike the REDIRECT message that is a system level optimization with no participation from a user (such as database administrator or programmer) , a HINT message is a tool for the database administrator or programmer to optimize the performance of specific databases, just like partitioned and distributed database management systems (such as the database system 102) allow a database administrator to specify partition and placement policies. By providing a routing HINT message in the BEGIN statement, a programmer may suggest how a current distributed transaction is to be routed.

In order for enable the use of a HINT message, the load balancer (e.g., the load balancing node 108) may be augmented or configured to compute a partition function and a placement function as described in the foregoing description, or a combined version of the partition function and the placement function, i.e., place (part _{table_name} (partition_key) ) , so that the load balancer can compute a hosting node (i.e., a computing node 104 in the database system 102) that stores data shard (s) of a data table referred in the HINT message. In implementations, the load balancer may issue an RPC (i.e., a remote procedure call) and delegate this functionality to any one server in a partitioned and distributed database management system (i.e., any computing node 104 in in the database system 102) . Alternatively, a mapping table may be stored locally in the load balancer (e.g., the load balancing node 108) , or stored in a mapping device associated with the load balancer as described in the foregoing description.

In implementations, a database administrator or programmer may need to modify their distributed transactions or queries, adding a reserved term “HINT” after a BEGIN statement. At the time of connection, a BEGIN statement with a HINT message is transported to the load balancer (e.g., the load balancing node 108) , which may then extract a name of a data table and a value of a partition key from the HINT message. The load balancer may then use the function place (part _{table_name} (partition_key) ) to compute a hosting node, and may then forward the connection from the client device to the hosting node.

In implementations, a HINT message may suppress a REDIRECT message. If a distributed transaction is designated with a reserved term “HINT” , a computing node that processes the distributed transaction may never need or may not be allowed to raise a REDIRECT exception that is described in the above description for REDIRECT messages, due to a pre‐designation of a computing node through the HINT message in the distributed transaction.

Using the foregoing example distributed transaction as further shown as follows:

BEGIN

UPDATE warehouse

SET ytd += in_payment_amount

WHERE warehouse_id = in_w_id AND warehouse_zip = in_w_zip;

UPDATE customer

SET balance ‐= in_payment_amount

WHERE customer_id = in_c_id AND customer_zip = in_c_zip;

COMMIT

The first query in the distributed transaction targets a warehouse table. Since a database administrator may know that the warehouse table is partitioned by values of zip codes, the database administrator may modify the above BEGIN statement as: BEGIN HINT warehouse in_w_zip, which essentially indicates to the load balancer to forward the current distributed transaction to a hosting node that includes data relevant to the first query (e.g., the second computing node 104‐2 as shown in FIG. 6) , thus achieving the same goal as a REDIRECT message. Furthermore, in this case, using the HINT message can further avoid any extra data transmitted back and forth between the client device and the first computing node as described in the above description for the example scenario of using a REDIRECT message, thus further saving communication costs and time between the client device and the first computing node.

Furthermore, a distributed transaction may include multiple queries, and each query may go to a different computing node. For example, a distributed transaction may include ten queries, and most of these ten queries, except the first query, involve data of one or more data shards in a same computing node. If the first query needs to be processed by a first computing node, while remaining nine queries need to be processed by a second computing node, the database system 102 or the load balancing node 108 may better forward the distributed transaction to the second computing node and use the second computing node as a control or coordination node to coordinate processing of local transactions associated with the queries in the distributed transaction by different computing nodes, collect query results of the local transactions from other computing nodes, and send a combined or aggregated query result to a client device that initiates the distributed transaction, thus further reducing communication costs and time that are caused by transmitting the query results (which may include data rows that be selected or updated in the distributed transaction) between the computing nodes. This can be achieved using a HINT message, not a REDIRECT message.

Conclusion

Although implementations have been described in language specific to structural features and/or methodological acts, it is to be understood that the claims are not necessarily limited to the specific features or acts described. Rather, the specific features and acts are disclosed as exemplary forms of implementing the claimed subject matter. Additionally or alternatively, some or all of the operations may be implemented by one or more ASICS, FPGAs, or other hardware.

The present disclosure can be further understood using the following clauses.

Clause 1: A method implemented by a current node, the method comprising: receiving at least one query included in a distributed database transaction; obtaining a name of a data table and a partition key that are involved in the query; obtaining an address of a particular node that includes a data shard of the data table corresponding to the partition key; determining whether the current node is the particular node and whether the distributed database transaction including the query has been processed locally in one or more nodes; and based on a result of the determining, performing one of: returning an exception to a client that submits the query for redirecting the query to the particular node, starting to process a transaction associated with the query locally in the current node, or sending the query to the particular node.

Clause 2: The method of Clause 1, wherein returning the exception to the client that submits the query for redirecting the query to the particular node is performed in response to determining that the current node is not the particular node and the distributed database transaction including the query has not been processed locally in the one or more nodes.

Clause 3: The method of Clause 2, wherein the exception comprises a virtual address or an identifier of the particular node.

Clause 4: The method of Clause 1, wherein starting to process the transaction associated with the query locally in the current node is performed in response to determining that the current node is the particular node and the distributed database transaction including the query has not been processed locally in the one or more nodes.

Clause 5: The method of Clause 4, wherein the distributed database transaction comprises a plurality of queries, at least some of the plurality of queries involving data that is stored in different nodes.

Clause 6: The method of Clause 5, further comprising committing the transaction associated with the query locally in coordination with other nodes that process other queries of the plurality of queries.

Clause 7: The method of Clause 1, wherein sending the query to the particular node in response to determining that the transaction including the query has been processed locally in the one or more nodes.

Clause 8: The method of Clause 1, wherein obtaining the address of the particular node that includes the data shard of the data table corresponding to the partition key comprises: obtaining the data shard based on a first mapping relationship between the data shard and a combination of the data table and the partition key; and obtaining the address of the particular node based on a second mapping relationship between the address of the particular node and the data shard.

Clause 9: The method of Clause 1, wherein the partition key comprises key values of one or more columns of the data table.

Clause 10: One or more computer readable media storing executable instructions that, when executed by one or more processors of a computing device, cause the one or more processors to perform acts comprising: receiving a request for processing a distributed database transaction, the distributed database transaction being started with a designated statement and including a plurality of queries after the designated statement, and the designated statement indicating a name of a data table and a partition key; determining a computing node that includes a data shard of the data table corresponding to the partition key; and assigning the distributed database transaction to the computing node.

Clause 11: The one or more computer readable media of Clause 10, wherein determining the computing node that includes the data shard of the data table corresponding to the partition key comprises determining the computing node that includes the data shard of the data table corresponding to the partition key based at least in part on mapping relationships between partition keys and computing nodes that separately store data shards of the data table corresponding to the partition keys.

Clause 12: The one or more computer readable media of Clause 11, wherein the acts further comprise storing the mapping relationships between the partition keys and the computing nodes that separately store the data shards of the data table corresponding to the partition keys.

Clause 13: The one or more computer readable media of Clause 10, wherein determining the computing node that includes the data shard of the data table corresponding to the partition key comprises: sending an inquiry including the name of the data table and the partition key to a mapping device that stores the mapping relationships; and receiving information of the computing node that includes the data shard of the data table corresponding to the partition key.

Clause 14: The one or more computer readable media of Clause 10, wherein the acts further comprise sending the distributed database transaction to the computing node to enable the computing node to act as a coordinate node to manage local transactions of the plurality of queries in one or more computing nodes.

Clause 15: The one or more computer readable media of Clause 10, wherein the designated statement comprises a reserved term that indicates a nature of the designated statement to the computing device.

Clause 16: The one or more computer readable media of Clause 10, wherein at least some of the plurality of queries involves data that is stored in different computing nodes.

Clause 17: A current node comprising: one or more processors; and memory storing executable instructions that, when executed by the one or more processors, cause the one or more processors to perform acts comprising: receiving at least one query included in a distributed database transaction; obtaining a name of a data table and a partition key that are involved in the query; obtaining an address of a particular node that includes a data shard of the data table corresponding to the partition key; determining whether the current node is the particular node and whether the distributed database transaction including the query has been processed locally in one or more nodes; and based on a result of the determining, performing one of: returning an exception to a client that submits the query for redirecting the query to the particular node, starting to process a transaction associated with the query locally in the current node, or sending the query to the particular node.

Clause 18: The current node of Clause 17, wherein returning the exception to the client that submits the query for redirecting the query to the particular node is performed in response to determining that the current node is not the particular node and the distributed database transaction including the query has not been processed locally in the one or more nodes, the exception comprising a virtual address or an identifier of the particular node.

Clause 19: The current node of Clause 17, wherein starting to process the transaction associated with the query locally in the current node is performed in response to determining that the current node is the particular node and the distributed database transaction including the query has not been processed locally in the one or more nodes, the distributed database transaction comprising a plurality of queries, at least some of the plurality of queries involving data that is stored in different nodes.

Clause 20: The current node of Clause 17, wherein obtaining the address of the particular node that includes the data shard of the data table corresponding to the partition key comprises: obtaining the data shard based on a first mapping relationship between the data shard and a combination of the data table and the partition key; and obtaining the address of the particular node based on a second mapping relationship between the address of the particular node and the data shard.

Claims

A method implemented by a current node, the method comprising:

receiving at least one query included in a distributed database transaction;

obtaining a name of a data table and a partition key that are involved in the query;

obtaining an address of a particular node that includes a data shard of the data table corresponding to the partition key;

determining whether the current node is the particular node and whether the distributed database transaction including the query has been processed locally in one or more nodes; and

based on a result of the determining, performing one of: returning an exception to a client that submits the query for redirecting the query to the particular node, starting to process a transaction associated with the query locally in the current node, or sending the query to the particular node.
The method of claim 1, wherein returning the exception to the client that submits the query for redirecting the query to the particular node is performed in response to determining that the current node is not the particular node and the distributed database transaction including the query has not been processed locally in the one or more nodes.
The method of claim 2, wherein the exception comprises a virtual address or an identifier of the particular node.
The method of claim 1, wherein starting to process the transaction associated with the query locally in the current node is performed in response to determining that the current node is the particular node and the distributed database transaction including the query has not been processed locally in the one or more nodes.
The method of claim 4, wherein the distributed database transaction comprises a plurality of queries, at least some of the plurality of queries involving data that is stored in different nodes.
The method of claim 5, further comprising committing the transaction associated with the query locally in coordination with other nodes that process other queries of the plurality of queries.
The method of claim 1, wherein sending the query to the particular node comprises sending the query to the particular node in response to determining that the transaction including the query has been processed locally in the one or more nodes.
The method of claim 1, wherein obtaining the address of the particular node that includes the data shard of the data table corresponding to the partition key comprises:

obtaining the data shard based on a first mapping relationship between the data shard and a combination of the data table and the partition key; and

obtaining the address of the particular node based on a second mapping relationship between the address of the particular node and the data shard.
The method of claim 1, wherein the partition key comprises key values of one or more columns of the data table.
One or more computer readable media storing executable instructions that, when executed by one or more processors of a computing device, cause the one or more processors to perform acts comprising:

receiving a request for processing a distributed database transaction, the distributed database transaction being started with a designated statement and including a plurality of queries after the designated statement, and the designated statement indicating a name of a data table and a partition key;

determining a computing node that includes a data shard of the data table corresponding to the partition key; and

assigning the distributed database transaction to the computing node.
The one or more computer readable media of claim 10, wherein determining the computing node that includes the data shard of the data table corresponding to the partition key comprises determining the computing node that includes the data shard of the data table corresponding to the partition key based at least in part on mapping relationships between partition keys and computing nodes that separately store data shards of the data table corresponding to the partition keys.
The one or more computer readable media of claim 11, wherein the acts further comprise storing the mapping relationships between the partition keys and the computing nodes that separately store the data shards of the data table corresponding to the partition keys.
The one or more computer readable media of claim 10, wherein determining the computing node that includes the data shard of the data table corresponding to the partition key comprises:

sending an inquiry including the name of the data table and the partition key to a mapping device that stores the mapping relationships; and

receiving information of the computing node that includes the data shard of the data table corresponding to the partition key.
The one or more computer readable media of claim 10, wherein the acts further comprise sending the distributed database transaction to the computing node to enable the computing node to act as a coordinate node to manage local transactions of the plurality of queries in one or more computing nodes.
The one or more computer readable media of claim 10, wherein the designated statement comprises a reserved term that indicates a nature of the designated statement to the computing device.
The one or more computer readable media of claim 10, wherein at least some of the plurality of queries involves data that is stored in different computing nodes.
A current node comprising:

one or more processors; and

memory storing executable instructions that, when executed by the one or more processors, cause the one or more processors to perform acts comprising:

receiving at least one query included in a distributed database transaction;

obtaining a name of a data table and a partition key that are involved in the query;

obtaining an address of a particular node that includes a data shard of the data table corresponding to the partition key;

determining whether the current node is the particular node and whether the distributed database transaction including the query has been processed locally in one or more nodes; and

based on a result of the determining, performing one of: returning an exception to a client that submits the query for redirecting the query to the particular node, starting to process a transaction associated with the query locally in the current node, or sending the query to the particular node.
The current node of claim 17, wherein returning the exception to the client that submits the query for redirecting the query to the particular node is performed in response to determining that the current node is not the particular node and the distributed database transaction including the query has not been processed locally in the one or more nodes, the exception comprising a virtual address or an identifier of the particular node.
The current node of claim 17, wherein starting to process the transaction associated with the query locally in the current node is performed in response to determining that the current node is the particular node and the distributed database transaction including the query has not been processed locally in the one or more nodes, the distributed database transaction comprising a plurality of queries, at least some of the plurality of queries involving data that is stored in different nodes.
The current node of claim 17, wherein obtaining the address of the particular node that includes the data shard of the data table corresponding to the partition key comprises:

obtaining the data shard based on a first mapping relationship between the data shard and a combination of the data table and the partition key; and

obtaining the address of the particular node based on a second mapping relationship between the address of the particular node and the data shard.