Abstract:
Systems, methods and computer program product embodiments for providing a locking protocol for partitioned and distributed database tables are disclosed herein. A locking method includes executing, by at least one processor, a first database transaction on a second node, attempting to acquire and acquiring a lock on the second node in intentional exclusive mode, executing, by the at least one processor, a second database transaction on a first node, acquiring a lock on the first node in exclusive mode and waiting to acquire a lock on the second node in exclusive mode, routing, by the at least one processor, the first database transaction to the first node and unsuccessfully trying to acquire a lock on the first node and committing, by the at least one processor, the first database transaction.

Description:
CROSS-REFERENCE TO RELATED APPLICATION 
       [0001]    This application claims the benefit of U.S. Provisional Application No. 61/731,631, “Locking Protocol for Partitioned and Distributed Tables,” filed Nov. 30, 2012, incorporated by reference herein. 
     
    
     BACKGROUND 
       [0002]    1. Field 
         [0003]    The present embodiments are generally related to a locking protocol for partitioned and distributed database tables. 
         [0004]    2. Background 
         [0005]    Conventional database management systems have been optimized to perform on hardware with limited main memory, e.g. random access memory (RAM). These conventional database management systems have slower disk input/output (I/O) that serves as a bottleneck. 
         [0006]    However, computer architecture has advanced so that multi-core parallel processing is possible by processor cores communicating using RAM or a shared cache. In addition, RAM is no longer as limited a resource. Databases may now be stored entirely in RAM and thus disk access is no longer a limiting factor for performance. However, multi-core systems present other challenges. 
         [0007]    Databases of online transaction processing systems have been modified to utilize multi-core parallel processor computer systems efficiently. In particular, these databases support parallel execution of transactions, are now located in-memory and are organized to be cache efficient. In addition, the databases support partitioning over a plurality of nodes. Conventionally, there was a single lock manager used for an entire partitioned database table. Maintaining this single lock manager provides challenges such as deadlock and extra network costs as well as overhead resulting from a master database node. However, conventional locking protocols may be improved to mitigate deadlock and overhead. 
       BRIEF SUMMARY 
       [0008]    Briefly stated, the example embodiments include system, method and computer program product embodiments, and combinations and sub-combinations thereof, for providing a locking protocol for partitioned and distributed database tables. According to embodiments, multi-core parallel processing in-memory partitioned database systems may execute an optimistic intentional exclusive locking protocol. 
         [0009]    In an embodiment, a method includes executing, by at least one processor, a first database transaction on a second node, attempting to acquire and acquiring a lock on the second node in intentional exclusive mode. The method further includes executing, by the at least one processor, a second database transaction on a first node, acquiring a lock on the first node in exclusive mode and waiting to acquire a lock on the second node in exclusive mode. In addition, the method includes routing, by the at least one processor, the first database transaction to the first node and unsuccessfully trying to acquire a lock on the first node. The first database transaction is then committed by the at least one processor. 
         [0010]    In a further embodiment, a method includes attempting, by at least one processor, an intentional exclusive lock for a local database node having a partition of a database table. The method further includes determining, by the at least one processor, that the trying failed and determining whether a remote database node having another partition of the database table has acquired an intentional exclusive lock and acquiring the intentional exclusive lock for the another portion of the database table if not acquired. 
         [0011]    Further features and advantages, as well as the structure and operation of various embodiments thereof, are described in detail below with reference to the accompanying drawings. It is noted that the disclosure is not limited to the specific embodiments described herein. Such embodiments are presented herein for illustrative purposes only. Additional embodiments will be apparent to persons skilled in the relevant art(s) based on the teachings contained herein. 
     
    
     
       BRIEF DESCRIPTION OF THE DRAWINGS/FIGURES 
         [0012]    The accompanying drawings, which are incorporated herein and form a part of the specification, illustrate exemplary embodiments and, together with the description, further serve to explain the principles of the disclosure and to enable a person skilled in the relevant art(s) to make and use the contemplated and disclosed embodiments. 
           [0013]      FIG. 1  illustrates a block diagram of database system hardware architecture according to example embodiments. 
           [0014]      FIG. 2  illustrates a partitioned database according to example embodiments. 
           [0015]      FIG. 3  illustrates an example instance of deadlock between DDL and DML transactions. 
           [0016]      FIG. 4A  illustrates a method based on an optimistic IX locking protocol according to example embodiments. 
           [0017]      FIG. 4B  illustrates the optimistic IX locking protocol according to example embodiments. 
           [0018]      FIG. 5  illustrates a further embodiment involving the optimistic IX locking protocol according to example embodiments. 
           [0019]      FIG. 6  illustrates an even further embodiment involving the optimistic IX locking protocol according to example embodiments. 
           [0020]      FIG. 7  illustrates an additional embodiment involving the optimistic IX locking protocol according to example embodiments. 
           [0021]      FIG. 8  illustrates a method of avoiding deadlock using an optimistic IX locking protocol according to example embodiments. 
           [0022]      FIG. 9  illustrates an example computer system according to example embodiments. 
       
    
    
       [0023]    Features and advantages of embodiments of the present disclosure will become more apparent from the detailed description set forth below when taken in conjunction with the drawings. In the drawings, like reference numbers generally indicate identical, functionally similar, and/or structurally similar elements. Generally, the drawing in which an element first appears is indicated by the leftmost digit(s) in the corresponding reference number. 
       DETAILED DESCRIPTION 
       [0024]    Introduction 
         [0025]    The following detailed description refers to the accompanying drawings that illustrate exemplary embodiments consistent with this disclosure. Other embodiments are possible, and modifications can be made to the embodiments within the spirit and scope of the embodiments. Therefore, the detailed description is not meant to limit the embodiments. Rather, the scope of the embodiments is defined by the appended claims. 
         [0026]    Example Hardware Architecture 
         [0027]      FIG. 1  shows a block diagram of a database system  106  according to example embodiments. The database system may be, but is not limited to, an in-memory column-store database system. 
         [0028]    In conventional database systems, the focus is directed to optimizing disk access, by minimizing a number of disk pages to be read into main memory when processing a query. This bottleneck is shown in  FIG. 1  at  102 . 
         [0029]    However, the performance bottleneck in multi-core parallel processor computer systems is found between a CPU cache and RAM. The processor cores wait for data to be loaded from RAM into the processor cache. This bottleneck is shown in  FIG. 1  at  104 . This bottleneck is addressed by making efficient usage of the CPU cache. As a number of processor cores increase, CPUs will continue to be able to simultaneously process increasingly more data. 
         [0030]    As shown in  FIG. 1 , database system  106  may include a computer  108  having at least one processor (CPU)  110 . As an example, the database system  106  in  FIG. 1  is shown having one processor, but the database system is not limited to having one processor and may have two or more processors. As an example, CPU  110  in  FIG. 1  is shown as having two cores  112 , but the processor  110  may include less than two cores or more than two cores. The cores  112  may have a CPU cache  114  that is shared among the cores. Each core  112  may have a plurality of hardware contexts, e.g. threads. In addition, the computer  108  includes random-access memory (RAM)  116  which may include hundreds of GB or TBs of RAM. According to example embodiments, the database system  106  may be an in-memory column-store database system stored and executed within RAM  116 . Thus, as opposed to conventional database systems stored on disk where disk access and speed presents a bottleneck, the RAM  116  of an in-memory database system presents a bottleneck for the faster cache  114 . The RAM  116  and processor  110  may communicate via a bus  118 . 
         [0031]    Transactions 
         [0032]    According to example embodiments, the database system  106  may execute transactions. A transaction is a logical unit of work that includes one or more SQL statements. A transaction may begin with a first executable statement being DML (data manipulation language) used for inserting, updating, deleting or replacing (upserting) data into partitioned database tables or DDL (data definition language) used for defining partitioned database tables such as creating or dropping a table. In addition, a transaction ends with one of the following events: a COMMIT or ROLLBACK statement issues, a DDL statement executes (e.g. automatic commit) or an error occurs (e.g. a lock timeout error or a deadlock error). 
         [0033]    According to example embodiments, transactions executed are provided full ACID support (atomicity, consistency, isolation and durability). In addition, according to example embodiments, the database provides multi-version concurrency control (MVCC) with statement-level and transaction-level isolation as well as multi-level locking and deadlock detection. Regarding statement-snapshot isolation, a statement may see changes that are committed before a statement is started. This is a default isolation level and is also known as read-committed. Regarding transaction-snapshot isolation, a statement may see changes committed before its transaction is started. This is known as repeatable-read or serializable. 
         [0034]    Locking 
         [0035]    According to example embodiments, the database system  106  may serialize access to shared resources that may change. Serialization may be provided by locks. A lock may be acquired right before changes are made to a database and released at transaction commit or transaction rollback. According to example embodiments, there are three types of transaction locks: a DB lock, e.g. a meta lock, a Table Lock, e.g. an object lock and a record lock. The DB lock may include a shared mode (S) and an exclusive mode (X). The Table Lock may include intentional exclusive (IX) and exclusive (X) modes. The record lock may include exclusive (X) mode. The example embodiments described below are related to Table Locks, but the embodiments are not limited to Table Locks. 
         [0036]    An exclusive lock may be acquired by a Lock Table command explicitly or by a DDL command implicitly. However, a transaction that holds an exclusive lock is the only transaction that may access the table. Lock requests for the table by other transactions are blocked while the exclusive lock is held. 
         [0037]    Intentional exclusive locks may be acquired by DML implicitly. Multiple transactions may acquire an intentional exclusive lock. Exclusive lock requests for the table by other transactions are blocked while the intentional exclusive lock is held. 
         [0038]    According to example embodiments, a lock wait timeout may occur. A lock wait timeout occurs when a commit/rollback is missing, when an update transaction takes a long time to process or a lock wait timeout configuration value is very small. In addition, deadlocks may occur and may be automatically detected. Deadlocks may be resolved by rolling back a database transaction that is involved in the deadlock. However, according to example embodiments, deadlocks may be mitigated. 
         [0039]    Partitioned Tables 
         [0040]    According to example embodiments, database tables may be partitioned into multiple partitions as shown in  FIG. 2 . In particular, a database may be divided or partitioned into independent parts and may be distributed. Thus, each database may be spread amongst a plurality of partitions or nodes and transactions may be performed in parallel on the partitions.  FIG. 2  shows partition P 1   202  and partition P 2   204 . P 1   202  is associated with node 1  and P 2   204  is associated with node 2 . As shown in  FIG. 2 , there is not a partition associated with node 3 , but this is merely an example, and a partition may be associated with node 3 . 
         [0041]    Deadlock in Partitioned Database 
         [0042]    Conventionally, for DML transactions a shared lock is acquired for each partition of a database table. When DDL transactions occur simultaneously, deadlock situations may occur. 
         [0043]      FIG. 3  illustrates deadlock in a partitioned database between DDL and DML transactions. Conventionally, for DDL an X (exclusive) lock was used to lock a master node and all nodes containing the partitions. An X lock may acquire multiple locks on multiple nodes. 
         [0044]    However, according to example embodiments, a single database transaction may move around to multiple connections on multiple nodes using statement routing. Thus, if a DML single transaction also acquires an IX lock on multiple nodes, deadlock may occur between DDL and DML operations. 
         [0045]    In other words, this conventional method of locking causes deadlock  300  between DDL and DML transactions. As an example, a first transaction Tx 1   302  may be DML and a second transaction Tx 2   304  may be DDL. Tx 1   302  may begin first as DML on node 2 . An IX lock may be applied to node 2 . Next, Tx 2   304  may begin DDL. An X lock may be applied to node 1 . Tx 2   304  attempts to apply an X lock on node 2 , but is forced to wait for Tx 1   302 . Next, Tx 1   302  performs DML on node 1  and applies an IX lock on node 1  and waits for Tx 2   304 . At this point, there is deadlock because Tx 1   302  has acquired multiple locks on multiple partitions of the same database. Acquisition of IX locks on multiple nodes is avoided according to the embodiments described below. 
         [0046]    According to embodiments, each node has its own local lock server and there is not a centralized global lock server. Deadlock problems may be solved according to the example embodiments below. 
         [0047]    Optimistic IX Locking Protocol 
         [0048]    According to example embodiments, deadlock may be avoided between an IX lock used for DML in a first transaction and an X lock used tor DDL in a second transaction. Rather than a single lock manager, each node may maintain its own lock manager. 
         [0049]    As shown in  FIG. 4A , an optimistic IX locking protocol  400  is shown as pseudocode. First, IX try_lock is performed by a transaction in step  402  before any locking occurs. If try_lock is successful by the transaction, then a local node is locked in an IX mode. 
         [0050]    If try_lock fails, then it is determined whether the transaction has already acquired an IX lock on any of the remote nodes in step  404 . 
         [0051]    If the transaction has acquired a lock on one of the remote nodes, then locking of the local node is skipped in step  406 . 
         [0052]    However, if no other remote lock exists, then the transaction may wait on the local node in step  408  and the transaction may wait to acquire the local node in IX mode without deadlock occurring. 
         [0053]    In a first embodiment shown in  FIG. 4B , a first transaction Tx 1   410  performs DML and a second transaction Tx 2   420  performs DDL. As shown, DML is performed by Tx 1   410  on node 2 . Next, Tx 1   410  may perform IX try_lock on node 2 , which is successful, e.g. Tx 1  will acquire the lock in IX mode. Simultaneously, DDL is performed by Tx 2   420 . An X lock is acquired by Tx 2   420  and node 1  is locked. Tx 2   420  will acquire and apply an X lock to node 2 , but will wait for Tx 1   410 . A database connection is moved from node 2  to node 1  by Tx 1   410 . Next, DML is performed by Tx 1   410  on node 1 . Tx 1   410  will perform IX try_lock on node 1 , which will fail because Tx 2   420  holds an X lock on node  1 . Tx 1   410  will then check for an IX lock on a remote node. Tx 1   410  has an IX lock on node 2 , and will skip locking. Tx 1   410  will then commit and Tx 2   420  is successful. Thus, there is no deadlock. In other words, DML transactions may only lock a single partition/node of a database at a time, thereby mitigating deadlock. 
         [0054]    In an additional embodiment shown in  FIG. 5 , a first transaction Tx 1   510  performs DML and a second transaction Tx 2   520  performs DDL. As shown, DDL is performed by Tx 2   520 . An X lock is applied to node 1  and then an X lock is applied to node 2 . Next, Tx 1   510  may perform DML on node 2 . Tx 1   510  will perform IX try_lock on node 2 , but this will fail because Tx 2   520  holds an X lock on node 2 . Next, Tx 1   510  will check whether there is an IX lock on a remote node. Tx 1   510  will determine that there are no other IX locks on remote nodes. Tx 1   510  will then wait for Tx 2   520  to complete. Tx 2   520  will commit and then Tx 1   510  may continue. Thus, the scenario shown in  FIG. 5  is successful and no deadlock occurs. 
         [0055]    In a further embodiment shown in  FIG. 6 , a first transaction Tx 1   610  performs DML and a second transaction Tx 2   620  performs DDL. As shown, DML is performed by Tx 1   610  on node 2 . Next, Tx 1   610  may perform IX try_lock on node 2 , which is successful. Tx 1   610  will lock node 2  in IX mode. Simultaneously, Tx 2   620  may add a partition on node 3 . Next, Tx 2   620  will apply an X lock to node 1 , and then attempt to apply an X lock to node 2 . However, Tx 2   620  will have to wait for Tx 1   610 . Tx 1   610  will perform statement routing and apply DML to node 1 . Tx 1   610  will perform IX try_lock to node 1 , but this will fail. Tx 1   610  will then try to check remote IX locks. Tx 1   610  will see that there is an IX lock on node 2  and will skip locking node 1 . Thus, Tx 1   610  will then commit. Tx 2   620  will be able to commit and deadlocks may be avoided. 
         [0056]    In an even further embodiment shown in  FIG. 7 , five transactions may occur simultaneously. As shown, DML is performed by Tx 1   710 , DDL is performed by Tx 2   720 , DML is performed by Tx 3   730 , DML is performed by Tx 4   740  and DML is performed by Tx 5   750 . 
         [0057]    First, Tx 1   710  may perform IX try_lock on node 2 , which is successful. Tx 1   710  will lock node 2  in IX mode. Next, Tx 2   720  performs DDL and applies an X lock to node 1  and then applies an X lock to node 2 . However, Tx 2   720  will wait for Tx 1   710  to apply the X lock on node 2 . Next, Tx 1   710  may perform IX try_lock on node 1 . However, this will fail. Tx 1   710  will then check whether the transaction has already acquired an IX lock on remote nodes. Tx 1   710  will see that it has an IX lock on node 2  and will skip locking node 1 . Thus, Tx 1   710  will commit. 
         [0058]    Next, Tx 3   730  may perform IX try_lock on node 1 , and this will fail. However. Tx 3   730  does not hold any other IX locks on remote nodes. Thus, Tx 3   730  may apply an IX lock on node 1 , and will wait for Tx 2   720 . 
         [0059]    Next, Tx 4   740  may perform IX try_lock on node 2  and this will fail. However, Tx 4   740  does not hold any other IX locks on remote nodes. Thus, Tx 4   740  may apply an IX lock on node 2  and will wait for Tx 2   720 . 
         [0060]    Next, Tx 5   750  may perform IX try_lock on node 3 . This will succeed and Tx 5   750  will obtain an IX lock on node 3 . 
         [0061]    Next, Tx 2   720  may obtain an X lock on node 3 . However, Tx 2   720  will wait for Tx 5   750 . After Tx 5   750  commits, then Tx 2   720  will be able to apply the X lock to Tx 5   750  and then commit. 
         [0062]    Thus, according to example embodiments, there is limited network cost in most cases and there is reduced master node overhead. 
         [0063]    According to embodiments,  FIG. 8  illustrates a method  800  of avoiding deadlock using an optimistic intentional exclusive locking protocol according to example embodiments. A multi-core parallel processing in-memory database system may execute database transactions over multiple partitions and mitigate deadlock. 
         [0064]    As an example, two separate transactions may execute in parallel over database partitions/nodes. In step  810 , a first DML transaction may begin on a second node. 
         [0065]    In step  820 , the first transaction may successfully execute IX try_lock on the second node and acquire an IX lock on the second node. 
         [0066]    In step  830 , a second DDL transaction may begin and the second transaction may acquire an X lock on a first node and acquire an X lock on the second node. However, the second transaction will have to wait for the first transaction to complete on node 2 . 
         [0067]    In step  840 , by applying statement routing, the first DML transaction may move from node 2  to node 1 . 
         [0068]    In step  850 , the first transaction may execute IX try_lock on node 1 . However, IX try_lock will fail because of the X lock held on node 1  by the second transaction. 
         [0069]    In step  860 , the first transaction may check if there is an IX lock on a remote node. 
         [0070]    In step  870 , the first transaction may determine that it has an IX lock on the second node, e.g. a remote node. 
         [0071]    In step  880 , the first transaction may skip locking on the first node and commit. Thus, according to example embodiments, deadlock does not occur. 
       Example Computer Implementation 
       [0072]    In an example embodiment, the systems, methods and computer products described herein are implemented using well known computers, such as computer  900  shown in  FIG. 9 . 
         [0073]    Computer  900  can be any commercially available and well known computer capable of performing the functions described herein, such as computers available from International Business Machines, Apple, Sun, HP, Dell, Compaq, Digital, Cray, etc. 
         [0074]    Computer  900  includes one or more processors (also called central processing units, or CPUs), such as a processor  906 . The processor  906  is connected to a communication bus  904 . Processors  906  may include any conventional or special purpose processor, including, but not limited to, digital signal processor (DSP), field programmable gate array (FPGA), and application specific integrated circuit (ASIC). 
         [0075]    Computer  900  includes one or more graphics processing units (also called GPUs), such as GPU  907 . GPU  907  is a specialized processor that executes instructions and programs selected for complex graphics and mathematical operations in parallel. 
         [0076]    Computer  900  also includes a main or primary memory  908 , such as random access memory (RAM). The primary memory  908  has stored therein control logic  928 A (computer software), and data. 
         [0077]    Computer  900  also includes one or more secondary storage devices  910 . The secondary storage devices  910  include, for example, a hard disk drive  912  and/or a removable storage device or drive  914 , as well as other types of storage devices, such as memory cards and memory sticks. The removable storage drive  914  represents a floppy disk drive, a magnetic tape drive, a compact disk drive, an optical storage device, tape backup, etc. 
         [0078]    The removable storage drive  914  interacts with a removable storage unit  916 . The removable storage unit  916  includes a computer useable or readable storage medium  924 A having stored therein computer software  928 B (control logic) and/or data. Removable storage unit  916  represents a floppy disk, magnetic tape, compact disk, DVD, optical storage disk, or any other computer data storage device. The removable storage drive  914  reads from and/or writes to the removable storage unit  916  in a well-known manner. 
         [0079]    Computer  900  also includes input/output/display devices  922 , such as monitors, keyboards, pointing devices, touch-screen displays, etc. 
         [0080]    Computer  900  further includes a communication or network interface  918 . The network interface  918  enables the computer  900  to communicate with remote devices. For example, the network interface  918  allows computer  900  to communicate over communication networks or mediums  924 B (representing a form of a computer useable or readable medium), such as LANs, WANs, the Internet, etc. The network interface  918  may interface with remote sites or networks via wired or wireless connections. 
         [0081]    Control logic  928 C may be transmitted to and from computer  900  via the communication medium  924 B. More particularly, the computer  900  may receive and transmit carrier waves (electromagnetic signals) modulated with control logic  930  via the communication medium  924 B. 
         [0082]    Any apparatus or manufacture comprising a computer useable or readable medium having control logic (software) stored therein is referred to herein as a computer program product or program storage device. This includes, but is not limited to, the computer  900 , the main memory  908 , the secondary storage devices  910 , the removable storage unit  916  and the carrier waves modulated with control logic  930 . Such computer program products, having control logic stored therein that, when executed by one or more data processing devices, cause such data processing devices to operate as described herein, represent embodiments of the disclosure. 
         [0083]    The disclosure can work with software, hardware, and/or operating system implementations other than those described herein. Any software, hardware, and operating system implementations suitable for performing the functions described herein can be used. 
         [0084]    Conclusion 
         [0085]    It is to be appreciated that the Detailed Description section, and not the Summary and Abstract sections, is intended to be used to interpret the claims. The Summary and Abstract sections may set forth one or more, but not all, exemplary embodiments as contemplated by the inventors, and thus, are not intended to limit the disclosure and the appended claims in any way. 
         [0086]    Embodiments have been described above with the aid of functional building blocks illustrating the implementation of specified functions and relationships thereof. The boundaries of these functional building blocks have been arbitrarily defined herein for the convenience of the description. Alternate boundaries can be defined so long as the specified functions and relationships thereof are appropriately performed. 
         [0087]    The foregoing description of the specific embodiments will so fully reveal the general nature of the disclosure that others can, by applying knowledge within the skill of the art, readily modify and/or adapt for various applications such specific embodiments, without undue experimentation, without departing from the general concept of the disclosure. Therefore, such adaptations and modifications are intended to be within the meaning and range of equivalents of the disclosed embodiments, based on the teaching and guidance presented herein. It is to be understood that the phraseology or terminology herein is for the purpose of description and not of limitation, such that the terminology or phraseology of the specification is to be interpreted by the skilled artisan in light of the teachings and guidance. 
         [0088]    The breadth and scope of the disclosure should not be limited by any of the above-described exemplary embodiments, but should be defined only in accordance with the following claims and their equivalents.