Systems and methods for a distributed in-memory database

CROSS-REFERENCE TO RELATED APPLICATIONS

0001This Application is related to the following U.S. patent applications, which are filed concurrently with this Application, and which are incorporated herein by reference to the extent permitted by law:

0002Attorney Docket No. 30014200-1127, entitled “Systems and Methods for a Distributed Cache;”

0003Attorney Docket No. 30014200-1128, entitled “Systems and Methods for a Distributed In-Memory Database and Distributed Cache;” and

0004Attorney Docket No. 30014200-1129, entitled “Systems and Methods for Synchronizing Data in a Cache and Database.”

FIELD OF THE INVENTION

0005The present invention relates to computer-implemented databases, and in particular, to distributed in-memory databases and database caches.

BACKGROUND OF THE INVENTION

0006As memory becomes less expensive, an increasing number of databases may fit in a computer's main memory. These in-memory databases typically have been managed by relational database management systems (“RDBMS.”) However, RDBMS performance is limited even with large amounts of main memory. In particular, a program may need to translate an application's object representation of data (e.g., Java™ object representation) into and out of a relational representation for the relational database. In addition, RDBMSs generally cache databases on database servers, requiring several layers of software and network protocol to access from an application. Data is typically not cached in the same virtual memory as the application, except for small or short-lived caches in a database driver. Further, a RDBMS is typically optimized for data on disk, not in memory. For example, a typical RDBMS uses b-trees, database pages, locking, and logging. These are optimized to minimize disk access rather than central processing unit (“CPU”) time or memory space.

0007Another problem with traditional RDBMS caches is update speed. There is a bottleneck in the requirement to write through to the backing database when a transaction is committed to the backing database. The data must be transferred from the client that made the update all the way through to the server's disk before acknowledging the commit.

SUMMARY OF THE INVENTION

0008Methods, systems, and articles of manufacture consistent with the present invention provide a memory-based relational data store that can be a cache to a backend relational database or as a standalone in-memory database. The memory-based relational data store may be distributed, for example, over a plurality of data processing systems or processes. For purposes of this invention, a data store that is in-memory is located in directly-addressable memory and not on disk. The store can run in the same virtual memory as an application, or it can run as a separate process. The data store provides extremely fast reads, because it avoids the overhead of RDBMS layers. Further, the data store provides extremely fast updates, because updates need not be pushed to disk if the store is mirrored across two machines. A transaction commit can be performed by updating both the primary and standby stores.

0009When the data store acts as a cache for backend databases, high commit performance can be achieved with transactional integrity, compared to conventional single-system caches that require data to be transferred from the client that made the update all the way through to the server's disk before acknowledging a commit. Further, when the data store acts as a cache, it either writes the committed data through to a backing store, writes the data to a standby replica and thus avoids write through to a disk, or may delay writing to the backing store.

0010In accordance with methods consistent with the present invention, a method in a data processing system having a plurality of independent sub data processing systems is provided. The data processing system has a program for managing an in-memory database. The method comprises the steps of: providing an in-memory database; and distributing the in-memory database over at least two of the plurality of sub data processing systems in memory.

0011In accordance with articles of manufacture consistent with the present invention, a computer-readable medium containing instructions that cause a program to perform a method for managing an in-memory database is provided. The data processing system has a plurality of independent sub data processing systems. The method comprises the steps of: providing an in-memory database; and distributing the in-memory database over at least two of the plurality of sub data processing systems in memory.

0012In accordance with systems consistent with the present invention, a data processing system is provided. The data processing system comprises a plurality of independent sub data processing systems, each data processing system having a memory and a processing unit. A first of the sub data processing systems has a program that provides an in-memory database, and distributes the in-memory database over at least two of the plurality of sub data processing systems in memory. The processing unit of the first sub data processing runs the program.

0013Other systems, methods, features, and advantages of the invention will become apparent to one with skill in the art upon examination of the following figures and detailed description. It is intended that all such additional systems, methods, features, and advantages be included within this description, be within the scope of the invention, and be protected by the accompanying drawings.

BRIEF DESCRIPTION OF THE DRAWINGS

0014The accompanying drawings, which are incorporated in and constitute a part of this specification, illustrate an implementation of the invention and, together with the description, serve to explain the advantages and principles of the invention. In the drawings,

0015<figref idref="DRAWINGS">FIG. 1</figref> shows a block diagram illustrating a data processing system suitable for use with methods and systems consistent with the present invention;

0016<figref idref="DRAWINGS">FIG. 2</figref> is a block diagram illustrating a database host system suitable for use with methods and systems consistent with the present invention;

0017<figref idref="DRAWINGS">FIG. 3</figref> is a block diagram illustrating a remote system suitable for use with methods and systems consistent with the present invention;

0018<figref idref="DRAWINGS">FIG. 4</figref> is a block diagram showing illustrative persistence interfaces;

0019<figref idref="DRAWINGS">FIG. 5</figref> is a block diagram of an illustrative database table;

0020<figref idref="DRAWINGS">FIG. 6</figref> is a functional block diagram showing storing data in a record cache;

0021<figref idref="DRAWINGS">FIG. 7</figref> is a block diagram of a store and its components;

0022<figref idref="DRAWINGS">FIG. 8</figref> is a block diagram of a first illustrative configuration consistent with the present invention;

0023<figref idref="DRAWINGS">FIG. 9</figref> is a block diagram of a second illustrative configuration consistent with the present invention;

0024<figref idref="DRAWINGS">FIG. 10</figref> is a block diagram of a third illustrative configuration consistent with the present invention;

0025<figref idref="DRAWINGS">FIG. 11</figref> is a block diagram of a fourth illustrative configuration consistent with the present invention;

0026<figref idref="DRAWINGS">FIG. 12</figref> is a block diagram of a fifth illustrative configuration consistent with the present invention;

0027<figref idref="DRAWINGS">FIG. 13</figref> is a block diagram of a sixth illustrative configuration consistent with the present invention;

0028<figref idref="DRAWINGS">FIG. 14</figref> is a block diagram of a seventh illustrative configuration consistent with the present invention;

0029<figref idref="DRAWINGS">FIG. 15</figref> is a block diagram of an eighth illustrative configuration consistent with the present invention;

0030<figref idref="DRAWINGS">FIG. 16</figref> is a block diagram of illustrative components of the store;

0031<figref idref="DRAWINGS">FIG. 17</figref> is a sequence diagram showing illustrative steps for establishing a session in the first illustrative configuration;

0032<figref idref="DRAWINGS">FIG. 18</figref> is sequence diagram showing illustrative steps for beginning a session in the first illustrative configuration;

0033<figref idref="DRAWINGS">FIG. 19</figref> is a sequence diagram showing illustrative steps for inserting a row in the first illustrative configuration;

0034<figref idref="DRAWINGS">FIG. 20</figref> is a sequence diagram showing illustrative steps for updating a row in the first illustrative configuration;

0035<figref idref="DRAWINGS">FIG. 21</figref> is a sequence diagram showing illustrative steps for deleing a row in the first illustrative configuration;

0036<figref idref="DRAWINGS">FIGS. 22A and 22B</figref> are sequence diagrams showing illustrative steps for querying the store in the first illustrative configuration;

0037<figref idref="DRAWINGS">FIGS. 23A and 23B</figref> are sequence diagrams showing illustrative steps for committing data in the first illustrative configuration;

0038<figref idref="DRAWINGS">FIG. 24</figref> is a sequence diagram showing illustrative communication between a primary store and a hot standby store;

0039<figref idref="DRAWINGS">FIG. 25</figref> is sequence diagram showing additional steps for inserting a row when there is a replicated store;

0040<figref idref="DRAWINGS">FIG. 26</figref> is a sequence diagram showing illustrative steps for updating a field in a row in a cache when there is a replicated store;

0041<figref idref="DRAWINGS">FIG. 27</figref> is a sequence diagram showing illustrative steps for querying a store when there is a replicated store;

0042<figref idref="DRAWINGS">FIG. 28</figref> is a sequence diagram showing illustrative steps for a two-phase commit transaction when there is a replicated store;

0043<figref idref="DRAWINGS">FIG. 29</figref> is a sequence diagram showing illustrative steps for a one-phase commit transaction when there is a replicated store;

0044<figref idref="DRAWINGS">FIG. 30</figref> is sequence diagram showing illustrative steps for a standby cache taking over the role of primary;

0045<figref idref="DRAWINGS">FIG. 31</figref> is a sequence diagram showing illustrative steps for a repair operation;

0046<figref idref="DRAWINGS">FIG. 32</figref> is a sequence diagram that shows illustrative steps for a recovery operation;

0047<figref idref="DRAWINGS">FIGS. 33A and 33</figref><i>b </i>are sequence diagrams that depict illustrative steps for communication between a client and server for a partitioned store; and

0048<figref idref="DRAWINGS">FIG. 34</figref> is a sequence diagram showing illustrative steps for inserting a record for a partitioned store.

DETAILED DESCRIPTION OF THE INVENTION

0049Reference will now be made in detail to an implementation consistent with the present invention as illustrated in the accompanying drawings. Wherever possible, the same reference numbers will be used throughout the drawings and the following description to refer to the same or like parts.

0050Methods, systems, and articles of manufacture consistent with the present invention provide a memory-based relational data store that can act as a cache to a backend relational database or as a standalone in-memory database. The store can run in the same virtual memory as an application, or it can run as a separate process. <figref idref="DRAWINGS">FIG. 1</figref> depicts a block diagram of a data processing system <b>100</b> suitable for use with methods and systems consistent with the present invention. Data processing system <b>100</b> is referred to hereinafter as “the system.” The system includes one or more database host systems <b>102</b>, <b>104</b>, and <b>106</b>, such as servers. The database host computers can be accessed by one or more remote systems <b>108</b> and <b>110</b> (e.g., client workstations) via a network <b>112</b>. The network is a network suitable for use with methods and systems consistent with the present invention, such as a local area network or wide area network. In the illustrative embodiment, the network is a local area network.

0051<figref idref="DRAWINGS">FIGS. 2 and 3</figref> depict more detailed views of illustrative database host systems and remote systems, respectively. <figref idref="DRAWINGS">FIG. 2</figref> depicts a more detailed view of a database host system, such as database host system <b>102</b>. The database host system is, for example, a Sun SPARC® data processing system running the Solaris® operating system. One having skill in the art will appreciate that hardware and programs other than those described in the illustrative examples can be implemented. Sun, SPARC, Java, the Sun logo, Solaris, StarOffice, and Sun ONE are trademarks or registered trademarks of Sun Microsystems, Inc., Palo Alto, Calif., in the United States and other countries. Other names used herein are the property of their respective owners.

0052The illustrative database host system comprises a central processing unit (CPU) <b>202</b>, an input/output (I/O) unit <b>204</b>, a display device <b>206</b>, a secondary storage device <b>208</b>, and a memory <b>210</b>. The database host system may further comprise standard input devices such as a keyboard, a mouse or a speech processing means (each not illustrated). Memory <b>210</b> may comprise a database program <b>212</b> and a store <b>214</b> which may act as a cache for a backend relational database <b>216</b> or may act as an in-memory database. Store <b>214</b> may comprise one or more sub-stores, each of which can act as a cache or as an in-memory database. As will be described in more detail below, the cache <b>214</b> may be distributed across one or more database host systems or processes. In this case, there is a cache component in main memory on each relevant database host system. One of skill in the art will appreciate that each program and module described herein can be a stand-alone program and can reside in memory on a data processing system other than the described system. The program and modules may comprise or may be included in one or more code sections containing instructions for performing their respective operations. While the programs and modules are described as being implemented as software, the present implementation may be implemented as a combination of hardware and software or hardware alone. Also, one having skill in the art will appreciate that the programs and modules may comprise or may be included in a data processing device, which may be a client or a server, communicating with described system.

0053Although aspects of methods, systems, and articles of manufacture consistent with the present invention are depicted as being stored in memory, one having skill in the art will appreciate that these aspects may be stored on or read from other computer-readable media, such as secondary storage devices, like hard disks, floppy disks, and CD-ROM; a carrier wave received from a network such as the Internet; or other forms of ROM or RAM either currently known or later developed. Further, although specific components of system <b>100</b> have been described, one skilled in the art will appreciate that a data processing system suitable for use with methods, systems, and articles of manufacture consistent with the present invention may contain additional or different components.

0054One having skill in the art will appreciate that the database host systems and remote systems can themselves also be implemented as client-server data processing systems. In that case, a program or module can be stored on, for example, a database host system as a client, while some or all of the steps of the processing of the program or module described below can be carried out on a remote server, which is accessed by the database host system over the network. The remote server can comprise components similar to those described above with respect to the database host system, such as a CPU, an I/O, a memory, a secondary storage, and a display device.

0055<figref idref="DRAWINGS">FIG. 3</figref> depicts an illustrative remote system, such as remote system <b>108</b> or <b>110</b>. In the illustrated example, remote system <b>108</b> is represented. The remote system can be, for example, a Sun SPARC® data processing system running the Solaris® operating system. The partner system comprises a central processing unit (CPU) <b>302</b>, an input/output (I/O) unit <b>304</b>, a display device <b>306</b>, a secondary storage device <b>308</b>, and a memory <b>310</b>. The remote system may further comprise standard input devices such as a keyboard, a mouse or a speech processing means (each not illustrated). Memory <b>310</b> may comprise a database program <b>314</b> and data <b>316</b>. Memory <b>310</b> may comprise a database program <b>314</b> and a store <b>316</b> which may act as a cache for a backend relational database <b>316</b> or may act as an in-memory database. Store <b>316</b> may comprise one or more sub-stores, each of which can act as a cache or as an in-memory database. As will be described in more detail below, the cache <b>316</b> may be distributed across one or more database host systems or processes. In this case, there is a cache component in main memory on each relevant database host system.

0056As shown in <figref idref="DRAWINGS">FIG. 4</figref>, one or more persistence interfaces <b>402</b>, <b>404</b>, and <b>406</b> may be used to access store <b>408</b>. In the illustrative example, the persistence interfaces include Java Data Objects (“JDO”) <b>402</b>, EJB Container Managed Persistence (“CMP”) <b>404</b>, and Java Database Connectivity (“JDBC”) <b>406</b>. Other persistence interfaces may be used, such as the JAXB persistence interface. Persistence interfaces are known to one having skill in the art and will not be described in more detail herein. The persistence interface uses a cache programmatic interface (“CPI”) <b>410</b> to access store <b>408</b>. A user of the CPI, such as JDO, CMP, or JDBC, is referred to as a “CPI adaptor” herein.

0057The store's data model is an in-memory relational model. The value of this approach is that it is portable across virtual machines and can be mapped directly to a relational database schema when the store is used as a cache. Using this approach enables methods, systems, and articles of manufacture consistent with the present invention to take advantage of the strengths of the relational model in terms of flexibility and expressive power without having to pay the cost and complexity incurred by algorithms and data structures that traditional relational databases are forced to use to reduce the number of disk accesses.

0058In the illustrative example, the store is relational and does not perform object-relational mapping. Therefore, a CPI adaptor (e.g., a JDO or CMP CPI adaptor) that exposes an object interface to its users, performs object-relational mapping of data that is sent to the store or that has been retrieved from the store via the CPI adaptor. Object-relational mapping is known in the art and will not be described further herein. A description of object-relational mapping may be found in Christian Bauer, et al., Hibernate in Action, ISBN 193239415X, which is incorporated herein by reference. Further, data may be stored in standard relational form, using tables and rows.

0059<figref idref="DRAWINGS">FIG. 5</figref> depicts an illustrative table <b>502</b>. Rows <b>504</b>-<b>510</b> are each assigned a unique id. The unique id may be viewed as a primary key for that row. This unique id may be provided by the CPI adaptor, or it may be generated by the user, such as the programmer. The data in a row is comprised of a row id and a collection of fields. In the illustrative example, row <b>1</b> includes row id <b>512</b> and fields <b>514</b>-<b>518</b>; row <b>2</b> includes row id <b>520</b> and fields <b>522</b>-<b>526</b>; row <b>3</b> includes row id <b>528</b> and fields <b>530</b>-<b>534</b>; and row <b>4</b> includes row id <b>536</b> and field <b>538</b>-<b>542</b>. The fields can include a variety of data and datatypes. For example, the fields may support text, integers, and Java types, including primitive types and wrappers, arrays of primitive types and wrappers, Date (such as java.sql and java.util Date types), String, and Serializable, or more complex types that are stored as binary large objects or character large objects.

0060Records can be stored in a variety of manners. In a first illustrative approach, each record is stored as an object array, where each field of the record is an element in the array. In the illustrative example, an illustrative CPI adaptor uses Java types and thus the access of fields in a record can be very fast, and would not require conversion or data copy. In a second illustrative approach, each record is stored in a compact byte array. The array may have a header, which provides offsets for each field, and each field is stored as a series of bytes rather than as a Java object. For this approach, the CPI adaptor converts between a byte array representation and a Java object representation. This approach saves space compared to the first approach, and also provides for fast recovery and repair by allowing the creation of a map of the data image into a local shared memory segment. Further, by using remote direct memory access (“RDMA”), the data image can be sent to another process' memory space to enable fast cross-machine recovery. Faster recovery can improve the availability of the system. For example, communication between components of a cache or in-memory database may be implemented using RDMA.

0061A third illustrative approach is a hybrid of the first and second approaches. Referring to <figref idref="DRAWINGS">FIG. 6</figref> as an example, in the third approach, a record object <b>602</b> is used to read and write fields within a record. The record object provides an external interface to the underlying record storage. Access to record data is through the record object. The record object stores committed records in a byte array format in a record byte array <b>604</b>. The record object includes an object array, which is referred to herein as a record cache <b>606</b>. Each element of record cache <b>606</b> represents a field in record byte array <b>604</b>.

0062Within a transaction, when a field is first accessed, either for a read or a write, the record object converts the field into its Java object representation, and this is “cached” within the record cache. Subsequent accesses to that field are done to this cached version. This cached representation represents the “after image” of a record for a given transaction. Fields in the record that have not been accessed have null references in the record cache, as represented by the diagonal line through the array element. Fields that have been accessed have references to Java objects for that field. These objects are referred to herein as field objects. In the illustrative example, fields <b>1</b> and <b>3</b> have been accessed. Field <b>3</b> has been modified, and is marked as dirty (as represented by the solid and dashed lines). As shown, the record cache provides the after image for this record within the transaction.

0063The record object creates a separate record cache for each transaction. In an alternative embodiment, unmodified field objects may be shared across transactions.

0064<figref idref="DRAWINGS">FIG. 7</figref> is a block diagram that depicts a high-level overview of the store and its components. The lines represent associations and the numbers at the ends of the lines represent cardinality. In the illustrative example, store <b>214</b> contains one or more sub-stores <b>702</b>. Sub-store <b>702</b> has one or more store partitions <b>704</b>. A sub-store may act as a cache to a remote relational store <b>706</b>. Each sub-store can be configured independently. For example, a first sub-store may be configured as a single virtual memory cache to a backing database and another may be configured as a replicated store with no backing database. Illustrative configurations are described in more detail below.

0065Each sub-cache includes a set of one or more tables <b>708</b>. If the sub-store is acting as a cache, each table is mapped to a specific table in the remote store. Each table has zero or more rows <b>710</b>. If the sub-store is acting as a cache, then each row is mapped to a specific row in the remote store.

0066A sub-store is divided into one or more store partitions <b>704</b>. In a simple configuration there may be one partition, but in a distributed configuration there may be multiple partitions to achieve better scalability and availability. Each partition may replicate its data to a mirror partition <b>712</b>. This provides enhanced durability (when there is no backing remote store) and availability.

0067The store may be configured with a variety of configurations. For example: data can be cached for a remote database or stored in-memory; data may be replicated and/or partitioned; or the store may reside in the same virtual memory as an application, or in a separate virtual memory using a client/server model. There are a number of possible combinations of these features. Six illustrative configurations are described below. Each of these illustrative configurations can be applied independently to each sub-store. Data in a particular sub-store can be associated with a given sub-store based on the overall quality of service desired for that data, in terms of performance, availability, and durability. For example, a data may be set that is volatile and does not need to be stored in a database, but which is desired to be highly available. In that case, the data may be located in a replicated store.

0000Illustrative Configuration 1: Cache in a Single Virtual Memory

0068Referring to <figref idref="DRAWINGS">FIG. 8</figref> as an example, in this configuration, the sub-store <b>802</b> runs in the same virtual memory as the database application <b>804</b> and acts as a cache to a backend relational database <b>806</b>. This provides the database application with beneficially fast read performance. Smart synchronization strategies, which are described below, can provide improvements in write performance.

0000Illustrative Configuration 2: Replicated In-Memory Database

0069This illustrative configuration is depicted in <figref idref="DRAWINGS">FIG. 9</figref> and is preferably implemented in a fault tolerant environment with redundant hardware and power supplies. Under such conditions, methods, systems, and articles of manufacture consistent with the present invention can provide at least as good mean-time-between-failure for two in-memory copies of data as with a single disk drive. Accordingly, when the data is stored exclusively in memory, significant improvements in write performance are achieved compared to when data is stored on disk. Replicated sub-stores <b>902</b> and <b>904</b> act as in-memory databases. Each sub-store exists in the same virtual memory as a respective database application <b>906</b> and <b>908</b>.

0000Illustrative Configuration 3: Database Cache Plus In-Memory Database

0070In this configuration, some data may be mapped to a backend database, while other data may be stored in an in-memory database. In the illustrative example of <figref idref="DRAWINGS">FIG. 10</figref>, the in-memory database is replicated. Accordingly, the example of <figref idref="DRAWINGS">FIG. 10</figref> is a combination of the first and second illustrative configurations. Replicated sub-stores <b>1002</b> and <b>1004</b> act as in-memory databases. Each sub-store exists in the same virtual memory as a respective database application <b>1006</b> and <b>1008</b>. Cache stores <b>1010</b> and <b>1012</b>, which are located in the same virtual memory as their respective database application <b>1006</b> and <b>1008</b>, act as a cache to a backend relational database <b>1014</b>.

0000Illustrative Configuration 4: Client/Server In-Memory Database

0071When the sub-store is used as an in-memory database, it may be desirable to run the sub-store outside of the database application processes, so that one database application does not have to play the role of “server” to all the other database applications. As shown in <figref idref="DRAWINGS">FIG. 11</figref>, in this case, a client version of the CPI <b>1102</b>-<b>1106</b> is resident in each respective database application <b>1108</b>-<b>1112</b>, while the data exists in the replicated sub-stores <b>1114</b> and <b>1116</b>. The CPI client may use a load balancing technique to balance connections across the cache replicas.

0072In an embodiment, the remote sub-store may appear like a remote database store, such that the client-side code is essentially a local cache to the remote sub-store. In an alternative embodiment, data on the client may instead be cached for the length of the transaction.

0000Illustrative Configuration 5: Partitioned Cache

0073When the data in the sub-store is distributed across multiple processes, horizontal scalability can be achieved. In an illustrative example, each row belongs to a single partition, which is determined by a partitioning function on the row id. The sub-store can quickly determine where a row exists using this partitioning function. The partitioning function may be a suitable partitioning function, such as hashing on a primary key. In another illustrative example, partitions may overlap, for example, through a combination of partitioning and replication. The partitioning function then defines where the primary copy of each row is stored, while replicas of the row are stored in other partitions.

0074As shown in <figref idref="DRAWINGS">FIG. 12</figref>, in this configuration, the sub-stores <b>1202</b>-<b>1206</b> preferably run outside the virtual memory of the database application processes <b>1208</b>, in client/server mode. A CPI client <b>1210</b> resides in the same virtual memory as the database application process. In the illustrative example, the sub-stores transact with a single database. Transactions against a partitioned cache may involve multiple partitions. If a partitioned sub-store acts as a cache, then a transaction may span multiple database connections, perhaps across multiple machines. This results in a distributed transaction.

0000Illustrative Configuration 6: Replicated Cache

0075This configuration provides high availability and durability for a cache where the update strategy involves deferred writes to the backing database. Reads are as fast as in a non-replicated cache, while writes are replicated to another cache. In case of failure to the primary cache, deferred writes are performed by the replica.

0076This illustrative configuration is depicted in <figref idref="DRAWINGS">FIG. 13</figref>. Database application <b>1302</b> uses replicated caches <b>1306</b> and <b>1308</b> via CPI client <b>1304</b>. The replicated caches transact with database <b>1310</b>.

0000Illustrative Configuration 7: Database Cache with Multiple Backend Databases

0077As shown in <figref idref="DRAWINGS">FIG. 14</figref>, when the data store <b>1406</b> acts as a cache for backend databases <b>1408</b>, <b>1410</b>, <b>1412</b>, the backend databases may comprise one or more remote databases, at least one of which may be an in-memory database. To the user of the cache, the several databases appear as one database. In the illustrative example, database application <b>1402</b> uses cache <b>1406</b> via CPI client <b>1404</b>. Although <figref idref="DRAWINGS">FIG. 14</figref> shows a single cache, the cache may be distributed. That is, the cache may be a partitioned or replicated cache.

0078This configuration provides high performance and scalability. Databases can be split into multiple databases for growth, integration, or workload reasons, for example. The cache provides for access to data from applications without changing the applications to account for the multiplicity of databases.

0000Illustrative Configuration 8: Partitioned In-Memory Database

0079Partitioning provides horizontal scalability. Since the partitioned database is memory resident, it does not have to use disk-based structures and algorithms. <figref idref="DRAWINGS">FIG. 15</figref> shows a distributed, replicated configuration with the caches <b>1502</b>-<b>1516</b> in-process with the database applications <b>1518</b>-<b>1524</b> and no backend relational database. This configuration provides high read and write performance, scalability, and high availability. Illustrative Configuration 8 provides the benefits of scalability and availability, with the high performance of an in-memory database.

0080A partitioned sub-store requires a partitioning scheme that is used to determine which partition a record belongs to. In the illustrative example, the partitioning scheme is the high availability database (“HADB”) hash partition scheme. The HADB partitioning scheme is described, for example, in Svein Olaf Hvasshovd, et al., <i>The Clustra telecom database: high availability, high throughput, and real</i>-<i>time response</i>, Proceedings of the 21^stInternational Conference on Very Large Databases, Zurich, Switzerland, pp. 469-477, September 1995, which is incorporated herein by reference. Alternatively, different partitioning schemes may be used. The partitioned sub-store requires a distributed query mechanism. In the illustrative example, the query algorithms and techniques used in the above-described HADB partitioning scheme are used.

0081Each of the above-described illustrative configurations is described in more detail below. Initially, a description of synchronization strategies, isolation levels, and components of the store is described below. When a remote backing database is used, the user may configure different synchronization strategies between a sub-store and the remote backing database. This allows the user to control disparity between the sub-store and the remote backing database. For example, a synchronization strategy may effect a push to the backing database whenever there is a write. In another example, a trigger in the database may update the cache, so that the cache is updated at all times. These illustrative examples may be used in combination with other synchronization strategies.

0082For configurations that use a remote backing database, users may configure different synchronization strategies between a sub-store and the remote backing store. The backing store could be a backend database or a remote sub-store (for example, in client/server mode where the client is a cache to an in-memory database). These strategies may be applied on a per-request basis, where a request is a specific operation performed by the CPI adaptor on a sub-store within the context of a transaction. The synchronization strategy can be set for a transaction, a particular request, or on a specific table or set of tables.

0083Further, a store may use different synchronization strategies for different data. For example, a cache or a part of a cache may use the optimistic synchronization strategy for some data, pessimistic for other data, and lazy for other data. The data strategies may be distinguished by criteria, such as by data type or by run-time optimization by access patterns. Further, the synchronization strategies may be selected according to an algorithm. For example, the synchronization strategy may be selected based on field data. In another example, selection may be based on a predicate consisting of an expression containing values of fields. For example, for stored data that describes employees' state of residence, an optimistic synchronization strategy may be used for residents of California and a pessimistic synchronization strategy may be used for residents of other states.

0084The synchronization strategy may be chosen manually or the system may automatically choose the synchronization strategy. The strategy may be chosen manually or statically, for example, by a system administrator or on a per-table or per-request basis. For example, the system administrator may specify a strategy for particular fields, rows, tables, or parts of tables.

0085The system may automatically choose the synchronization strategy, for example, by analyzing usage statistics. For example, the system may analyze access patterns to identify one of the following illustrative situations: which tables, columns, or rows are heavily updated by transactions; which ones are updated through particular systems; or which ones are infrequently updated. Based on the access patterns, the system may implement particular strategies. For example, if data is rarely updated, the system may use a lazy, optimistic strategy. If updates are typically through a particular system, the system may assume that that system's cache includes the truth and overwrite the backing store from it. In another example, if data is heavily updated from many machines, the system may use either a pessimistic locking on that data, or use an incremental merge strategy when the updates are typically increments to numeric values.

0086Illustrative synchronization strategies are described below in order of least conservative to most conservative. The write strategies are strategies that may be applied at the time of transaction commit. The read strategies are strategies that may be applied whenever data is accessed from the backing store.

0087Illustrative write strategies are described below. Although the synchronization strategy is applied independently for each request, write strategies are evaluated at commit time. For requests, a row that is modified by a given request is “tagged” with the write strategy for the request.

0088Lazy: In this strategy, data is written to the backing store in a “lazy” mode—that is, outside the scope of the transaction in which the request was made. This can be used, for example, where someone wants a single-machine version of a store, where the backing database provides some level of durability and database application access is through the store. In other examples, data may be written in accordance with conflict resolution rules.

0089Overwrite: Data tagged with this strategy overwrites data in the remote store. When the overwrite strategy is used, it is preferable that the sub-store is the only item modifying the data for the affected rows.

0090Optimistic: Prior to committing the transaction, the store checks to see if the rows tagged with this strategy have been updated in the remote store since they were read into the cache. If they have been updated, the store rolls back the transaction. This provides enhanced concurrency if there are multiple applications accessing the same data.

0091Write-Locked: The first time a row is modified, a write lock may be obtained in the remote store.

0092Checked: Prior to committing the transaction, the store checks to make sure that rows in the cache touched in the transaction, even rows that were read and not modified, are consistent with the database, and rolls back the transaction if they are not. For example, this strategy is useful when the user makes changes to rows based on the values in other rows, even though these rows have not changed.

0093Further, updates may be applied to the cache and synchronization of the database may be deferred. For example, some or all of the cache may be updated, while database synchronization is deferred until a later time. When the database is synchronized, the synchronization may be performed using resolution rules. Illustrative resolution rules are described in more detail below.

0094In another example of deferred synchronization, the store may assume for at least some data that the truth is in the cache, and overwrite the data in the database later. Accordingly, the cache returns to the user quickly after writing the cache. Alternatively, the store may assume that the truth is in the database, and perform fast reads on the cache. In this case, the store may accept data that is possibly out of date.

0095Further, when the database is updated, but not via the cache, this may trigger a synchronous or asynchronous write-back to the cache to keep the cache up to date.

0096Illustrative read strategies are described below.

0097Optimistic: When reading an object, the store assumes the data in the cache is correct and does not check with the database to see if more up-to-date data is available.

0098Pessimistic: When reading a row, the store throws away the cache row and reads it from the remote store, but does not lock the row in the remote store.

0099Read-Locked: This strategy is similar to the pessimistic read strategy, but also a read lock is acquired on the row in the remote store. Accordingly, no synchronization check is required at commit time.

0100Exclusive-Locked: The store acquires an exclusive lock on the row (vs. a read lock) when a row is accessed.

0101Each request has a specific synchronization strategy. The user can set the synchronization strategy for a transaction, a particular request, or for a given table or set of tables. The synchronization strategy to be used for a given request may be determined by the following illustrative rules: <ul id="ul0001" list-style="none"><li id="ul0001-0001" num="0000"><ul id="ul0002" list-style="none"><li id="ul0002-0001" num="0102"> If a strategy is specified for a specific request, this overrides the strategy specified for the transaction. </li><li id="ul0002-0002" num="0103"> If a strategy is specified for a specific table, this overrides the strategy specified for request or the transaction. There is an exception to this rule. Because the Checked write strategy applies to rows affected by the transaction, then as soon as the Checked write strategy is encountered anywhere within the transaction, the entire transaction strategy runs at the Checked write strategy. </li></ul></li></ul>

0104When the remote store is a Structured Query Language (“SQL”) database, database locks may be accomplished through the standard American National Standards Institute (“ANSI”) isolation levels. A description of ANSI isolation levels may be found, for example, in ANSI X3.135-1992, “Database Language SQL,” which is incorporated herein by reference. In the illustrative embodiment, the synchronization strategies are implemented using isolation levels. For a given cache transaction, a single connection is established with the database, with a specific isolation level, for operations that involve obtaining database locks. This situation is referred to as a “transaction connection” below. Some requests within the cache transaction may have a read strategy that does not require a database lock, and so they may use an independent connection running outside the scope of the current cache transaction. Illustrative scenarios are described below for how the illustrative synchronization strategies are implemented using isolation levels.

0105Lazy Write: In one illustrative example, in a background thread, the store obtains a connection to the database and sets the isolation level to READ_COMMITTED. The store submits modified rows, checking to see if the data has been modified in the database since it was read from the cache by the application.

0106In another example of a lazy write, data is written when updates go through a particular cache, for example with read-mostly access from other caches. This allows transactions to proceed at cache speed as the backing database is updated in the background. In yet another example, data is written when all updates to a particular table or group of tables or groups of columns all go through a particular machine's cache, so that there is one writer for each group of data. This also allows transactions to proceed at cache speed. In a further example, data is written when it is not important that queries be based on the latest version of the data being streamed to the back end. For example, if 1% of the data is being changed each day, and the queries are decision support queries that can be based on approximate data. This example allows updates on the back end to be done in large batches and in the background when there is lesser load.

0107If there are conflicts, the store may use conflict resolution mechanisms to resolve conflicts automatically, or to allow the user to resolve them manually. For example, a database administrator could specify on a per-table or per-column basis which resolution rules should be applied. Resolution rules may be applied when it is discovered, in synchronization of a cache with a back-end database (which could also be a cache), that the same record has been modified both places. Illustrative conflict resolution rules are identified below: <ul id="ul0003" list-style="none"><li id="ul0003-0001" num="0000"><ul id="ul0004" list-style="none"><li id="ul0004-0001" num="0108"> Field merge: Allow updates to a record from two different transactions as long as they did not change the same fields. Updates from both transactions are applied to the different fields of the record. </li><li id="ul0004-0002" num="0109"> Field group merge: Allow updates to a record from different transactions as long as they did not change fields in the same field group. The field group may be defined, for example, by an administrator, an API, or metadata. For example, an update to the zip code in one transaction may conflict with an update to the city in another transaction, if the city and zip code were defined to be in the same field group. </li><li id="ul0004-0003" num="0110"> Priority overwrite: Establish a priority (cache overrides database, or vice versa) establishing which update to a record takes precedence. The other update may be thrown away. </li><li id="ul0004-0004" num="0111"> Application/user priority overwrite: Track which updates are made by which applications or users (e.g., with an additional field on records), and give precedence to updates with a higher specified priority. </li><li id="ul0004-0005" num="0112"> Most-recent priority overwrite: Accept the changes that were made most recently, discarding the older record updates. This may be performed using a time stamp or some other mechanism that identifies when record updates were performed. </li><li id="ul0004-0006" num="0113"> Undo/Redo merge: Specify a list of allowed transactions (e.g., Add Customer, Place Order, Delete Employee), and track which updates are made by which transactions. Keep enough information to “back out” of one of the transactions, and then re-apply it to the current database (using the field values from conflicting transactions) to get serialization. </li><li id="ul0004-0007" num="0114"> Additive merge: For fields that are always incremented, e.g. Total Sales for a customer, or Order Count, compute the total increment since last synchronization in both the cache and the backing data store, and add the sum to the field value. For example, if the Order Count has increased from 1000 to 1055 in the database, and 1000 to 1015 in the cache, set the Order Count to 1070. </li><li id="ul0004-0008" num="0115"> Application escalation: When a record has been updated in both the cache and the backing database, call a procedure supplied by the database administrator to resolve the conflict. This procedure would be implemented using application-specific knowledge about how to combine the updates. </li><li id="ul0004-0009" num="0116"> User-administrator escalation: Conflicting record updates are resolved by a human operator, who is provided with information regarding the conflict and asked which updates to keep. </li></ul></li></ul>

0117Overwrite Write synchronization strategy: When the transaction is committed in the cache, the store uses the transaction connection if it exists or obtains a new connection with the isolation level to READ_COMMITTED. The store submits changes to rows tagged with this strategy without checking to see if the data has been modified since it was read into the cache.

0118Optimistic Write: When the transaction is committed in the cache, the store submits the changes using the transaction connection if it exists or obtain a new connection with an isolation level of READ_COMMITTED. The store ensures that the rows tagged with this strategy have not been modified since they were read into the cache. The operation may be aborted if there is a conflict, such as when a tagged row has been modified. Alternatively, resolution rules, such as those described above for Lazy Write, may be applied to resolve the conflict.

0119Checked Write: When the transaction is committed in the cache, the store submits the changes using the transaction connection if it exists or obtain a new connection with an isolation level of READ_COMMITTED. The store validates that rows touched by the transaction have not been modified in the remote store. The operation is aborted if there is a conflict, such as when a tagged row has been modified.

0120Optimistic Read: If the data is in the cache, the store reads it from the cache. If it is not in the cache, the store obtains an independent connection with READ_COMMITTED isolation level, reads the data, and releases the connection.

0121Pessimistic Read: The store obtains an independent connection with READ_COMMITTED isolation level, reads the data, and releases the connection.

0122Read-Locked Read: The store uses the transaction connection if it exists, or obtains a new one with READ_COMMITTED isolation level. The store reads the data with this connection but do not release the connection, it is now the transaction connection.

0123Exclusive-Locked Read: If the transaction connection exists, the store checks to see if the isolation level is SERIALIZABLE. If it is not, then this request can not be correctly serviced within the context of this transaction; so the store issues an exception. The CPI adaptor can then choose to either not perform this request or roll back the transaction.

0124If the transaction connection does not exist, the store obtains a new connection with the isolation level set to SERIALIZABLE, and reads the data. The store does not release the connection, it is now the transaction connection. Subsequent requests that use the transaction connection will operate at the SERIALIZABLE isolation level.

0125<figref idref="DRAWINGS">FIG. 16</figref> shows an overview of illustrative components of the store. The dashed arrows represent dependencies. Descriptions of each component are presented below. One having skill in the art will appreciate that additional or alternative components may be implemented.

0126CPI adaptor <b>1602</b> is not a component of the store, but is an external user of the store. As described above, CPI adaptors may be provided to implement/plug in to particular persistence APIs, such as JDO, CMP, JDBC, and JAXB. A session store <b>1604</b> provides an entry point for the CPI adaptor into a store. It contains context for the current conversation with the store. In particular it keeps track of transactional context. A store manager <b>1606</b> is responsible for managing a particular store. It stores the rows and is responsible for maintaining transactional consistency within the store. It is also responsible for coordinating data with remote repositories, be they other stores or a database. When a store needs to participate in global transactions, the store manager can act as an XA resource adapter—it will provide an implementation of the XAResource interface. In a replicated or partitioned environment there will be multiple store managers. There is one store manager for each virtual memory participating in the distributed store. These individual virtual memories running in a distributed store are called nodes herein. A cluster manager <b>1608</b> is an external program or module that is responsible for coordinating the multiple nodes in a distributed store. In the preferred embodiment, the cluster manager is the Sun Cluster product manufactured by Sun Microsystems, Inc. Alternatively, a different cluster manager may be used. The cluster manager detects the health of a node, initiates takeover, recovery and repair, and implements policies for network partitioning.

0127A query processor <b>1610</b> takes query trees provided by the CPI adaptor, generates a query plan, and executes the query. The query processor works with the store manager to obtain metadata and to acquire cache objects as may be needed. A metadata module <b>1612</b> stores the metadata for the store. The stored metadata includes, for example, the store schema, the mapping to the remote store, and partition information. Additional or alternative metadata can also be stored. A remote store <b>1614</b> is an abstraction that allows a consistent approach when the data for a given cache table is owned remotely rather than locally. The remote store can be to another store (native remote store <b>1616</b>) or it can be to a backend database (SQL remote store <b>1618</b>.) When a backend database is used, a driver <b>1622</b> allows communication to the backend database. In the illustrative embodiment, the driver is a JDBC driver, however, other drivers may be used.

0128The node communication module <b>1620</b> is responsible for communication between nodes. This module replicates data from a primary partition to a standby partition, routes a request from one node to another (e.g., when a given node receives an update row for a row belonging to a partition on another node), and administers other inter-node communications.

0129A runtime module <b>1624</b> provides an underlying framework for the various components of the store. The runtime module provides services such as network communication, threading, logging, debug/trace, internationalization, and security. These services may be implemented using, e.g., Java 2 Runtime, which is manufactured by Sun Microsystems, Inc.

0130A transaction coordinator <b>1626</b> regulates transactions between the various modules to ensure that transaction steps occur in the proper sequence.

0131Eight illustrative configuration were introduced above. A more detailed description of the illustrative configurations is presented below, including a description of the interactions between the various modules of the store. The interactions are described with reference to sequence diagrams. One having skill in the art will appreciate that the sequences are illustrative, and that alternative sequences may be implemented.

0000Illustrative Configuration 1: Cache in a Single Virtual Memory

0132The following scenarios relate to a store running as a cache running in a single virtual memory (i.e., not partitioned or replicated), with a backing store (e.g., an SQL store.)

0133<figref idref="DRAWINGS">FIG. 17</figref> shows a sequence diagram for a scenario of establishing a session. To communicate with a particular sub-store, the CPI adaptor establishes a session with the sub-store. The CPI adaptor does so by asking the store to create a store session by specifying the sub-store name (step <b>1702</b>). The session factory finds that sub-store based on the name provided by the CPI adaptor (step <b>1704</b>). The session factory then requests the store session module to create a session to the sub-store (step <b>1706</b>) and returns the session to the CPI adaptor (step <b>1708</b>).

0134After a session has been established with the sub-store, the CPI adaptor can transact with the sub-store. In the preferred embodiment, the CPI adaptor explicitly indicates the beginning and end of transactions. Alternatively, transactions can be automatically committed. <figref idref="DRAWINGS">FIG. 18</figref> depicts a sequence diagram for beginning a transaction. At this point, the store session is running within a specific transactional context. A transaction has not been started with the remote store. A transaction is started with the remote store as necessary depending upon the rows being read or modified and the synchronization strategy associated with those rows.

0135The CPI adaptor first requests the store session to begin the transaction (step <b>1802</b>). The store session, in turn, requests the transaction coordinator for a transaction object (step <b>1804</b>). The transaction object provides a handle to represent a transaction. Then, the transaction coordinator instantiates the transaction object and returns it to the store session (step <b>1806</b>). After receiving the transaction object, the store session forwards the transaction object to the CPI adaptor (step <b>1808</b>).

0136The CPI adaptor can engage in various operations with the store, such as inserting rows, deleting rows, committing data, and other types of operations. <figref idref="DRAWINGS">FIG. 19</figref> shows a sequence diagram for the case of inserting a row in the cache. First, the CPI adaptor requests to insert a row in a particular table by identifying the table name (step <b>1902</b>). The store session receives the CPI adaptor's request and forwards the request to the store (step <b>1904</b>). Then, the store inserts a row into the table and locks the row (step <b>1906</b>). After inserting the row into the table, the store creates a row object (step <b>1908</b>). The a row object is used to represent the row to the CPI adaptor. It also tracks changes to the row and acts as an “after image” of the row. In the illustrative example, the underlying row data is not changed until the transaction is committed. This allows the store to roll back a transaction in case there is some type of failure. It also allows the store to send the changes made to a standby replica.

0137The store sends the row object to the store session (step <b>1910</b>), which forwards the row object to the CPI adaptor (step <b>1912</b>). The row object that is returned to the CPI adaptor is empty and uninitialized, except for the row id. The CPI adaptor may fill in the values in the object as part of this transaction. The store session locks the underlying row so another transaction cannot see the row in its semi-initialized state.

0138<figref idref="DRAWINGS">FIG. 20</figref> shows a sequence diagram for an illustrative transaction for updating a field in a row in the cache. In the illustrative example, when a field is updated, the modification is made in the row object but is not applied to the underlying row until the transaction commits. First, the CPI adaptor sends a request to update a field to the row object (step <b>2002</b>). The request identifies the field (e.g., by the field's number) and includes the value to be entered into the field. Then, the row object marks the row as dirty (step <b>2004</b>) and sets the value in the field (step <b>2006</b>). The row object then notifies the CPI adaptor that the row has been updated (step <b>2008</b>).

0139<figref idref="DRAWINGS">FIG. 21</figref> depicts a sequence diagram for a delete row transaction. In the illustrative example, the underlying row is not deleted—instead, the row object is marked as deleted. When the transaction is applied, the row is then deleted from the store. As shown in <figref idref="DRAWINGS">FIG. 21</figref>, the CPI adaptor sends a request to the row object to delete a row (step <b>2102</b>). The row object then marks the row as deleted (step <b>2104</b>). Then, the row object notifies the CPI adaptor that the row has been marked as deleted (step <b>2106</b>).

0140The CPI adaptor can also query the store for data. For example, the CPI adaptor can request to obtain one or more rows from the store. In the illustrative example, the CPI adaptor submits a query to the store using a data structure called a query tree. This tree is built by an interpreter within the CPI adapter, such as an SQL interpreter, a JDO query processor, an JAXB query interpreter, or a CMP execution engine. Query trees and interpreters that build query trees are known in the art and will not be described in more detail herein. By using a query tree, the CPI adapter's queries can be language independent and do not require implementation of a query language.

0141The query tree is processed by the query processor. For a store configured as a cache, if the data is in the cache and the affected tables have a synchronization strategy that does not require checking the remote store for updated data, then the query processor performs the query against the data in the cache. However, if a synchronization strategy requires it, then changes may be flushed to the remote store and the query submitted to the remote store for execution. Alternatively, data may be fetched from the backing store into the cache, and a query may be executed on the cache if at least some of the data is in the remote store. Further, the query may be executed on a combination of the cache and the database. In this later example, the query may be split into two parts and the results may be merged. A further description of this example may be found for example in U.S. Pat. No. 6,243,710, which is incorporated herein by reference.

0142The query processor runs against data that is consistent with the current transaction. This means it works with both the data stored in row objects (e.g., rows that have been modified as part of a transaction) and the data in the underlying rows themselves (e.g., rows that have not been modified as part of a transaction).

0143<figref idref="DRAWINGS">FIG. 22</figref> depicts a sequence diagram for an illustrative query. In the illustrative example, the CPI adaptor prepares the query, including a query tree, and sends the query to the store session (step <b>2202</b>). The store session forwards the query to the query processor (step <b>2204</b>). After receiving the query, the query processor requests metadata from the store to compile and optimize the query (step <b>2206</b>). The metadata describes, for example, the number of columns in tables, the types of columns, and which columns have indices. The store obtains the relevant metadata and returns the metadata to the query processor (step <b>2208</b>).

0144The query processor analyzes the metadata to determine whether to execute the query in the remote store (step <b>2210</b>). In the illustrative example, the query is executed in the remote store if the synchronization strategy requires it or if at least some of the data is in the remote store. When the query is to be executed in the remote store, the query processor flushes the affected tables to the remote store via the store (step <b>2212</b>). The store receives the query processor's instruction to flush the affected tables and forwards the instruction to the remote store (step <b>2214</b>). The query processor also submits the query to the remote store, via the store, for execution (step <b>2216</b>). The store receives the query and forwards the query to the remote store (step <b>2218</b>). After receiving the query, the remote store executes the query and returns the results to the store (step <b>2220</b>). The store updates the cache from the results (step <b>2222</b>) and returns the obtained row object to the query processor (step <b>2224</b>).

0145If the query processor determines that the query can be run locally in the store instead of in the remote store (step <b>2226</b>), then the query processor processes the query locally (step <b>2228</b>). The query processor requests the relevant rows from the store (step <b>2230</b>), which in turn returns one or more row objects with the data (step <b>2232</b>).

0146After receiving the row objects that include the data, either from the remote store in step <b>2224</b> or from the store in step <b>2232</b>, the query processor returns the row objects to the store session (step <b>2234</b>). The store session forwards the row objects to the CPI adaptor (step <b>2236</b>).

0147<figref idref="DRAWINGS">FIG. 23</figref> depicts a sequence diagram for an illustrative commit transaction. The store keeps track of operations that have been performed within a current transaction, and just prior to commit these operations are applied to the remote store. This is a more efficient solution than applying the changes piecemeal throughout the life of the transaction. As shown in <figref idref="DRAWINGS">FIG. 23</figref>, the illustrative transaction commit may occur in four phases: beforeCompletion, prepare, commit, and afterCompletion. These phases may be driven by an external transaction coordinator. For example, the CPI Adapter may be registered to receive synchronization events (beforeCompletion and afterCompletion) and the JDBC driver may be registered as an XAResource and will receive the prepare and commit events.

0148When the CPI adaptor receives a beforeCompletion event from the transaction coordinator (step <b>2302</b>), the CPI adaptor forwards a beforeCompletion call to the store session (step <b>2304</b>). The store session forwards the call to the store (step <b>2306</b>), which flushes the affected rows to the remote store (step <b>2308</b>). In turn, the remote store instructs the driver (e.g., the JDBC driver) to flush the affected rows (step <b>2310</b>). When the flush is completed, the driver signals the remote store that it is clear to return (step <b>2312</b>). The store passes the return OK to the store session (step <b>2314</b>), which in turn passes the return OK to the CPI adaptor (step <b>2316</b>). Then, the CPI adaptor informs the transaction coordinator that the affected rows have been flushed (step <b>2318</b>).

0149Then, the driver receives the commit message from the transaction coordinator (step <b>2320</b>), and commits the transaction (step <b>2322</b>). Although it is not shown in <figref idref="DRAWINGS">FIG. 23</figref>, a prepare event may also occur. For example, a transaction manager may prepare each resource if there are multiple resources. However, if there is one resource, there may be no need to prepare the resource. When the commit is completed, the driver signals a return OK to the transaction coordinator (step <b>2324</b>).

0150The store then receives the afterCompletion call from the CPI adaptor, and the changes are applied and locks held for this transaction within the store are released. That is, the transaction coordinator sends the afterCompletion call to the CPI adaptor (step <b>2326</b>). The CPI adaptor forwards the call to the store session (step <b>2328</b>), which in turn forwards the call to the store (step <b>2330</b>). The store applies the changes and releases the locks (step <b>2332</b>). Then, the store sends a return OK to the store session (step <b>2334</b>), which forwards the return OK to the CPI adaptor (step <b>2336</b>). The CPI adaptor then signals the return OK to the transaction coordinator to identify that the commit transaction has been completed (step <b>2338</b>).

0000Illustrative configurations 2 and 3: Replicated In-Memory Database and Database Cache Plus In-Memory Database:

0151In the illustrative configuration, a replicated store is a logical store with two physical elements: a primary and hot standby. Although the primary store and the hot standby store are each described as a store, they are perceived by the database application as a single logical store.

0152<figref idref="DRAWINGS">FIG. 24</figref> is a sequence diagram that shows illustrative communication between a primary store and a hot standby store. The illustrative node communication subsystem provides buffered, asynchronous communication. Messages sent to the node communication subsystem from the store do not need to be immediately sent across the network. The node communication subsystem may, for example, buffer messages and send the buffered messages to the replica when it is full. Further, messages sent to the node communication subsystem from the store and messages sent between replicas over the network are transmitted asynchronously and do not require immediate acknowledgment.

0153In <figref idref="DRAWINGS">FIG. 24</figref>, half-arrows indicate asynchronous messages. The illustrative scenario shows messages for a single transaction, however, the node communication subsystem may receive messages for multiple simultaneous transactions, potentially from multiple stores within the same node. The primary store sends a begin transaction message to the primary node communication instance (step <b>2402</b>). The illustrative begin transaction message identifies the transaction to the primary node communication instance. After sending the begin transaction message in step <b>2402</b>, the primary store sends a first update row message to the primary node communication instance (step <b>2404</b>). The first update row message identifies the transaction and the row to update in the backup store. In the example, the primary store effects two updates to the same row by sending a second update row message (step <b>2406</b>). As described below, the two update row messages result in a single update row message on the standby replica, because there are two updates to the same row. After sending the update row messages, the primary store sends a “done” message to the primary node communication instance to indicate that there are no additional update messages (step <b>2408</b>).

0154The primary node communication instance buffers the update messages in the illustrative example and then sends them to the standby node communication instance (step <b>2410</b>). The standby node communication instance initiates the update of the standby store by sending a begin transaction message (step <b>2412</b>). Then, the standby node communication instance also sends an update row message (step <b>2414</b>). This update row message includes both changes to the relevant row. The “done” message is incorporated into the last change record using an “islast” parameter. This provides a small reduction in the number of network messages, and may prevent an extra packet being sent over the network.

0155The standby store acknowledges the begin transaction message (step <b>2416</b>), updates the row as described in the update message, and acknowledges that the row is being updated (step <b>2418</b>). The standby node communication instance forwards the acknowledgement message to the primary node communication instance (<b>2420</b>), which in turn forwards the acknowledgement to the primary store (step <b>2422</b>). Then, the primary store issues a commit message to commit the update (step <b>2424</b>). Commit transactions are described below.

0156<figref idref="DRAWINGS">FIG. 25</figref> is a sequence diagram showing additional steps for inserting a row. In the illustrative example, the additional steps comprise posting the insert record to the node communication system for each inserted row. The store session sends an insert row message to the store (step <b>2502</b>). Then, the store forwards the message to the node communication subsystem (step <b>2504</b>).

0157When a row is updated, the store sends an update record message to the node communication subsystem. As described above, multiple updates to the same row are “combined” into a single update message to the standby replica. <figref idref="DRAWINGS">FIG. 26</figref> is a sequence diagram showing illustrative steps for updating a field in a row in a cache. In the illustrative example, when a field is updated, the modification is made in the row object but is not applied to the underlying row until the transaction commits. First, the CPI adaptor sends a request to update a field to the row object (step <b>2602</b>). The request identifies the field (e.g., by the field's number) and includes the value to be entered into the field. Then, the row object marks the row as dirty (step <b>2604</b>) and sets the value in the field (step <b>2606</b>). The row object then sends a row update message to the node communication subsystem (step <b>2608</b>) and notifies the CPI adaptor that the row has been updated (step <b>2610</b>).

0158When querying in this configuration, the system does not have to interact with a backing store. The cache keeps track of whether information is stored in the cache or in the backing store. If the information is in the cache, then the system does not have to interact with the backing store. <figref idref="DRAWINGS">FIG. 27</figref> is a sequence diagram that shows an illustrative query transaction. In the illustrative example, the CPI adaptor prepares the query, including a query tree, and sends the query to the store session (step <b>2702</b>). The store session forwards the query to the query processor (step <b>2704</b>). After receiving the query, the query processor requests metadata from the store to compile and optimize the query (step <b>2706</b>). The store obtains the relevant metadata and returns the metadata to the query processor (step <b>2708</b>).

0159The query processor determines whether there are additional rows to obtain (step <b>2710</b>). If there are additional rows to obtain, the query processor requests the relevant rows from the store (step <b>2712</b>), which in turn returns one or more row objects with the data (step <b>2714</b>). If there are no additional rows to obtain, the query processor processes the rows (step <b>2716</b>) and creates a result set (step <b>2718</b>). Process rows may include, for example, selecting a subset of data from the rows and determining which rows satisfy the query. The result set may be, for example, the selected and combined rows. After creating the result set, the query processor sends the result set to the store session (step <b>2720</b>), which forwards the result set to the CPI adaptor (step <b>2722</b>).

0160In the client/server configuration, transactions may be committed in either two phases or one phase. Similar to the commit transaction described with respect to <figref idref="DRAWINGS">FIG. 23</figref>, the illustrative transaction commit for the client/server configuration may occur in four phases: beforeCompletion, prepare, commit, and afterCompletion. These phases may be driven by an external transaction coordinator. For committing transactions in two phases, a store may be registered as an XA resource manager with the external transaction coordinator. The CPI adapter implements the synchronization interface and is registered with the external transaction coordinator so it can receive beforeCompletion and afterCompletion events. As shown in <figref idref="DRAWINGS">FIG. 28</figref>, the transaction coordinator sends a beforeCompletion event to the CPI adaptor (step <b>2802</b>). The CPI adaptor forwards the beforeCompletion event to the store session (step <b>2804</b>), which in turn forwards the event to the primary store (step <b>2808</b>). Then, the primary store sends a “done” message to the primary node communication instance (step <b>2810</b>). It does not need to wait for acknowledgment from the standby cache but can return immediately to the caller. After sending the done message, the primary store sends a return OK message to the store session (step <b>2812</b>). The store session forwards the return OK message to the CPI adaptor (step <b>2814</b>), which in turn forwards the return OK message to the transaction coordinator (step <b>2816</b>).

0161In the prepare phase, the transaction coordinator sends a prepare message to the primary store (step <b>2818</b>). When the primary store receives the prepare message from the transaction coordinator, it waits for a done acknowledgment from the standby replica (step <b>2820</b>). Once it has this acknowledgment, it returns a vote commit message to the transaction coordinator, as now it knows that the standby is prepared to commit the transaction.

0162The transaction coordinator initiates the commit phase by sending a commit message to the primary store (step <b>2826</b>). Then, the primary store sends an asynchronous commit message to the primary node communication instance (step <b>2828</b>). Since the primary store's commit message is asynchronous, it does not have to wait for a response from the standby store. When the standby store acknowledges the last change record for the transaction, it will commit the transaction. The primary store then marks the data as committed and releases locks on the data (step <b>2830</b>), and then sends a return OK message to the transaction coordinator to signal that the commit has been completed (step <b>2832</b>).

0163After the commit phase has ended, the transaction coordinator sends an afterCompletion message to the CPI adaptor to initiate the afterCompletion phase (step <b>2834</b>). The CPI adaptor forwards the afterCompletion message to the store session (step <b>2836</b>), which in turn forwards the message to the primary store (step <b>2838</b>). The primary store acknowledges the afterCompletion message by sending a return OK message back to the transaction coordinator (step <b>2840</b>) via the store session (step <b>2842</b>) and the CPI adaptor (step <b>2844</b>).

0164In another illustrative example, an external transaction coordinator is not involved and the CPI adaptor is responsible for committing the transaction. In this case, the CPI adaptor submits the commit request, and the session store initiates the four commit phases. As shown in <figref idref="DRAWINGS">FIG. 29</figref>, the CPI adaptor asks the store session to commit the data (step <b>2902</b>). The store session sends a beforeCompletion message to the store (step <b>2904</b>). After receiving a return OK from the store (step <b>2906</b>), the store session sends a commit message to the store (step <b>2908</b>). The store signals that the data has been committed by sending another return OK message to the store session (step <b>2910</b>). In the afterCompletion phase, the store session sends an afterCompletion message to the store (step <b>2912</b>) and receives back a return OK message from the store (step <b>2914</b>). In response, the store session forwards the return OK message to the CPI adaptor to signal that the afterCompletion phase is done (step <b>2916</b>).

0165When the primary replica becomes unavailable, the standby cache may take over the role of primary. How this is initiated depends on whether or not the store is running in its own separate process (e.g., as a server) or whether it is running embedded in an application process. If the store is running in-process with the application, then the application detects whether the primary has become unavailable and notifies the standby replica that it needs to take over as the primary. The reason for this is because the application failure detection and takeover is entwined with store failure detection and takeover in the illustrative example. In this case, the application provides an implementation of the node manager and communicates with the store through a node manager interface to notify the standby store that it needs to take over as the primary replica.

0166If the store is running in client/server mode, then a node manager that is separate from the application detects whether the primary is unavailable and notifies the standby that it needs to take over.

0167Once a standby has been notified it needs to take over as the primary, it rolls back incomplete transactions. It also keeps open transactions for which it had prepared to commit by sending the done message to the primary replica. For these transactions, it will commit or roll back once it receives instructions from the transaction coordinator.

0168<figref idref="DRAWINGS">FIG. 30</figref> is a sequence diagram that shows illustrative steps for a standby cache taking over the role of primary. This may occur, for example, on crash of the primary. First, the cluster manager notifies the standby store that it needs to take over as primary (step <b>3002</b>). The standby store rolls back uncompleted transactions (step <b>3004</b>), and sends a return OK message when completed (step <b>3006</b>).

0169Then, the cluster manager registers the standby store as an XA resource with the transaction coordinator (step <b>3008</b>). After registering the standby store, the transaction coordinator acknowledges that the standby store has been registered as the primary (step <b>3010</b>). The transaction coordinator then determines whether there are committed transactions for which the standby has prepared to commit (step <b>3012</b>). For each of these transactions, the transaction coordinator instructs the standby store to commit (step <b>3014</b>). In response, the standby store commits each completed transaction (step <b>3016</b>), and returns a return OK message to the transaction coordinator (step <b>3018</b>).

0170The transaction coordinator also determines whether there are uncommitted transactions for which the standby store has prepared to commit (step <b>3020</b>). For each of these transactions, the transaction coordinator instructs the standby store to rollback (step <b>3022</b>). In response, the standby store rolls back each completed transaction (step <b>3024</b>), and returns a return OK message to the transaction coordinator (step <b>3026</b>).

0171The node then takes on the role of primary and begins accepting requests from CPI Adaptors. In an illustrative example, it does not replicate transactions until it detects that a replica is available again. At that point, it participates in a recovery process and then begins replicating to the new standby replica.

0172Similarly, in the case where the secondary crashes, the primary does not replicate transactions until it detects that a replica is available again. At that point, it participates in a recovery process and then begins replicating to the new standby replica.

0173In the illustrative example, there are two ways for a node to recover. The first is called repair, where the standby node gets a full copy of the data from the primary replica. The second is called recovery, where the standby is able to get at least some of its data from a local copy stored in local shared memory. Recovery may be faster than repair, except perhaps in deployments where RDMA is available. Recovery therefore may be attempted first, before falling back to repair.

0174<figref idref="DRAWINGS">FIG. 31</figref> is a sequence diagram that shows illustrative steps for a repair operation. First, the node manager initializes the store by indicating it is the standby and letting it know the location of the primary (step <b>3102</b>). Then, the standby contacts the primary to obtain a snapshot of the current data in the store (step <b>3104</b>). In response, the primary sends the data to the standby (step <b>3106</b>). The primary then sends subsequent transactions that have occurred since the snapshot was taken (step <b>3108</b>). The standby signals to the node manager that the data has been received (step <b>3110</b>). The standby is then considered up-to-date.

0175In the recovery scenario, a local copy of the data is available in shared memory or some other local storage, and the node can recovery from the local copy. As shown in <figref idref="DRAWINGS">FIG. 32</figref>, prior to loading the data, the store first ensures that the local copy is consistent. If the local copy is not consistent, then the store falls back to repairing from the primary replica. The node manager initiates replication by placing the store in standby and letting it know the location of the primary (step <b>3202</b>). The standby then checks the consistency of the local copy of the data (step <b>3204</b>). If the local copy of the data is inconsistent, then the store repairs from the primary replica (step <b>3206</b>). Otherwise, the store recovers from the local copy (step <b>3208</b>). To recover from the local copy, the store loads the data from its local storage location, such as shared memory (step <b>3210</b>). Then, the store requests transactions that may have been missed from the primary (step <b>3212</b>). The request identifies the last transaction, so that the primary may identify missed transactions. The primary returns missed transactions (step <b>3214</b>) and subsequent transactions (step <b>3216</b>) to the store. The store notifies the node manager that recovery has been completed by sending a return OK message (step <b>3218</b>). Transfer time can be decreased during recovery, for example, by implementing recovery using RDMA.

0176In a situation where there is replication, there is the risk of network partitioning, where both replicas are up and running but they have lost communication with each other. Each replica may then assume the role of primary, and there is a risk of data conflicts if transactions are allowed to continue against both replicas. Methods, systems, and articles of manufacture consistent with the present invention avoid data conflicts by having the node manager implement one or more policies to handle network partitioning. In various embodiments, the illustrative policies include: the node with the most recent transaction wins, and the other node repairs from this node; the node with the most activity wins; an attempt to merge is made, and conflicts are logged or otherwise reported, so that the user can manually resolve these conflicts; the user manually decides which node wins; the user initiates a merge, and interactively resolves conflicts; and avoiding partitioning by using a quorum algorithm in which a replica will not become primary unless it knows that it can communicate with a quorum of other nodes.

0177Illustrative Configuration 4: Client/Server in Memory Database The client/server scenario relates to a single virtual memory cache (i.e., not partitioned or replicated), where the backing store is a sub-store rather than a database. The above-described semantics of a single virtual memory cache apply to a cache client. Having a sub-store client behave as a local cache to a sub-store server not only simplifies the internal architecture; it also enables the local client to have similar performance characteristics to a single virtual memory cache. This configuration provides significantly better performance than a cache to a traditional SQL database, because the local cache does not have to pay the cost of translating to SQL, and the sub-store server may store its data in memory.

0178When the remote store is a replicated store, the client is provided with information about both the primary and the replica, and is able to transparently reconnect to the replica if the primary fails. Failover and retry are discussed above.

0000Illustrative Configurations 5-8: Partitioned Cache, Replicated Cache. Cache Plus Multiple Databases, and Partitioned In-Memory Database

0179A distributed store may be a partitioned store or a distributed store, or both. The distributed store may be a cache or an in-memory database, or both. For example, a partitioned in-memory database may be a cache for another in-memory database or backing store. A partitioned store is one in which the data for a single logical store is partitioned across multiple physical partitions in order to obtain better scalability. When a partitioned store is used as a cache, this may provide high scalability. When a store is replicated, this may provide high availability.

0180When a client connects to a partitioned store to establish a session, the client connects to a particular node. In a partitioned store that is also replicated, a node may host the primary replica for a partition and the standby replica for another partition. When a client connects to a node, the server side of the native remote store subsystem establishes a store session with primary partition for that node. Requests for the client are then dispatched to that store session. The client may connect to any one of a number of nodes.

0181A partitioned store may act as a server, or as a remote store, for an embedded client cache store running in the same virtual memory as an application. In this scenario, the remote store receives changes just prior to the transaction being committed. The client-server communications between a client and a remote store may be optimized to take advantage of this functionality and provide an interface that allows for a batch of inserts and updates to be communicated with as few network round-trips as possible. Further, a distributed cache or in-memory database may be configured to be a backing store for another distributed cache.

0182<figref idref="DRAWINGS">FIG. 33</figref> is a sequence diagram that depicts illustrative steps for communication between a client and server. In the illustrative example, modified rows for a transaction are sent in a single network message. Further, the communications between the store and client remote store and between the client and server remote stores are asynchronous. In the illustrative example, the store on the client is registered as an XA resource and receives the prepare and commit commands from the transaction coordinator. Therefore, the partitioned store is treated as a single resource from the perspective of the client's transaction coordinator.

0183The store session initiates communication by sending a beforeCompletion message to the store (step <b>3302</b>). The store identifies to the client native remote store the affected rows that are to be flushed (step <b>3306</b>), and sends a return OK message to the store session (step <b>3304</b>). Then, the client native remote store notifies the server native remote store of the changes made to the rows (step <b>3308</b>). Changes may include, for example, inserts, updates, and deletions of rows. The server native remote store determines whether there are rows to insert (step <b>3310</b>). For each inserted row, the server native remote store instructs the remote store session of the inserted row (step <b>3312</b>). For each updated row (step <b>3318</b>), the server native remote store instructs the remote store session of the updated row (step <b>3320</b>). Further, for each deleted row (step <b>3322</b>), the server native remote store instructs the remote store session of the deleted row (step <b>3324</b>). Then, the server native remote store returns a done acknowledgement message to the client native remote store (step <b>3326</b>).

0184The transaction coordinator sends a prepare command to the store (step <b>3314</b>), which waits for a done acknowledgement message from the client native remote store (step <b>3316</b>). After receiving the done acknowledgement from the server native remote store in step <b>3326</b>, the client native remote store sends the done acknowledgement message to the store (step <b>3328</b>). Then, the store sends a return OK message to the transaction coordinator to signal that the prepare phase is completed (step <b>3330</b>).

0185During the commit phase, the transaction coordinator sends a commit message to the store (step <b>3332</b>). The store then forwards the commit message to the client native remote store (step <b>3334</b>), which also forwards the commit message to the server native remote store (step <b>3336</b>). In turn, the server native remote store instructs the remote store session to commit the data (step <b>3338</b>). After committing the data, the remote store session sends a return OK message to the server native remote store (step <b>3344</b>). Then, the server native remote session store sends a commit acknowledgement message to the client native remote session store (step <b>3346</b>), which in turn sends a commit acknowledgement message to the store (step <b>3348</b>).

0186If a partitioned store acts as a cache to a backend database, then the partitioned store would implement the commit message as a distributed transaction spanning, for example, JDBC connections across multiple machines, using a distributed transaction coordinator. However, from the client's perspective, and from the perspective of the transaction coordinator servicing the client, it would still appear as a single resource committing a single transaction. In this illustrative example, there are two levels of transaction coordinators participating in this transaction.

0187When the store on a given node receives a request to insert a record, it uses the partitioning algorithm to determine which partition owns the given record, and sends it to the store for that partition. For simplicity, <figref idref="DRAWINGS">FIG. 34</figref> shows this as a direct message, however, one having skill in the art will appreciate that the message is sent through the node communication subsystem, which buffers multiple messages before sending a buffer to the target node, which then forwards the messages to the store for the target partition. As shown in <figref idref="DRAWINGS">FIG. 34</figref>, the store session sends the insert record message to the store (step <b>3402</b>). The store then determines which partition owns the given record (step <b>3404</b>), and sends insert record message to the store for that partition (step <b>3406</b>).

0188Further, the distributed cache may provide faster writes by returning to the client after recording the updates in memory on two or more independent machines. The updates can then be pushed back to the database in the background.

0189The scenario for inserting a record is similar for updating and deleting records. That is, the request is forwarded through the node communication subsystem to the appropriate node for that record.

0190The query processor may handle queries in different manners in the partitioned store depending on the type of query. If the query is a primary key lookup, then the query processor finds the partition that owns the relevant record and forwards the query to the node owning that record. If the query selects an ordered list of records based on the primary key, then the query processor selects the appropriate records from each partition and then combines them into a final result set that it sends to the client. However, if the query is more complicated, involving joins, ordering and filtering that is not based on the primary key, then the query processor creates a temporary data space where it is able to gather data from multiple partitions and merge them together into a result set. The processes for handling these types of queries are known in the art, such as the processes used by HADB, and will not be described in more detail herein.

0191A transaction in a partitioned database is a distributed transaction, and a two-phase commit approach may be used to commit the transaction.

0192A partitioned store can be highly available if it is also replicated. In the illustrative example, the behavior and mechanisms for takeover, repair, and recovery are the same for a replicated store with multiple partitions as with a replicated store with a single partition. However, in a partitioned store, the store is running as a server, and therefore a node manager is provided.

0193As described above, in the illustrative example, the HADB partitioning scheme may be implemented to determine to which partition a record belongs. The HABD partitioning scheme locates items based on characteristics. For example, HADB partitioning may partition using a hash function that is based on a primary key, which is a unique identifier for rows in a database, comprising a column or a group of columns, that is used to determine the partition in which a row of data is stored. For example, the primary key may be a project name or employee number. In an illustrative example, when the primary key is an employee number, data for a first group of employees may be partitioned to a first cache, while data for a different group of employees may be partitioned to a second cache. Further, the store may keep track of the version of data in a cache or database. For example, the store may associate a version number or time stamp with data entries.

0194As discussed above, the CPI adaptor may expose an object interface to its clients, and may perform object-relational mapping of data that is sent to the store or that has been retrieved from the store via the CPI adaptor. Therefore, the CPI adaptor may provide an object view to the client while caching relational data. For example, the CPI adaptor may provide a JAXB view and perform JAXB-relational mapping. Further, the relational cache may be optimized for object-relational mapping, such as by pre-computing foreign key references or in an alternative manner.

0195A plurality of CPI adaptors may be implemented to provide access to a cache via a plurality of data models simultaneously. For example, different CPI adaptors may provide object access, SQL access, and JAXB access to the cache. One or more of the CPI adaptors may provide different mappings, for example, for different classes. For example, a CPI adaptor may provide different views of the same employee table for employee1 class and employee2 class. Further, the CPI adaptors may access an in-memory database via a plurality of data models or mappings. For example, when a distributed cache is used in connection with a distributed in-memory database, data may be partitioned based on data type or table. Different data models may be used within the same cache or in-memory database as well as between cache and in-memory database partitions.

0196Thus, methods, systems, and articles of manufacture consistent with the present invention provide a memory-based relational data store that can act as a cache to a backend relational database or as a standalone in-memory database. The store can run as a cache in the same virtual memory as an application, or it can run as a separate process. The data store provides extremely fast reads, because it avoids the overhead of RDBMS layers. Further, the data store provides extremely fast updates, because updates need not be pushed to disk if the store is mirrored across two machines. A transaction commit can be performed by updating both the primary and standby stores.

0197When the data store acts as a cache for one or more backend databases, a high commit performance can be achieved with transactional integrity, compared to conventional single-system caches that require data to be transferred from the client that made the update all the way through to the server's disk before acknowledging a commit. When the data store acts as a standalone in-memory database, it either writes the committed data through to a backing store or writes the data to a standby replica and thus avoids write through to a disk.

0198The foregoing description of an implementation of the invention has been presented for purposes of illustration and description. It is not exhaustive and does not limit the invention to the precise form disclosed. Modifications and variations are possible in light of the above teachings or may be acquired from practicing the invention. For example, the described implementation includes software but the present implementation may be implemented as a combination of hardware and software or hardware alone. The invention may be implemented with both object-oriented and non-object-oriented programming systems. The scope of the invention is defined by the claims and their equivalents.