Multi-master with ownership transfer

A method, a system and a computer program product for executing management of ownership of data. An index in a plurality of indexes is selected. The index corresponds to a plurality of ranges of data values stored in a plurality of database slices of a database. The index further corresponds to a partitioning structure that includes a plurality of hierarchically arranged nodes. Each node corresponds to a range of data values stored in at least one database slice. The structure is replicated across a plurality of computing systems. A computing system executes an update to one or more ranges of data values. The system replicates at least one of a database slice including the updated ranges of data values and a node that includes the updated ranges of data values to another computing system for storage of a replicate of the updated ranges of data values.

TECHNICAL FIELD

This disclosure relates generally to data processing and, in particular, to multi-master data management, including ownership transfer of one or more data partitions.

BACKGROUND

Database management systems have become an integral part of many computer systems. For example, some systems handle hundreds if not thousands of transactions per second. On the other hand, some systems perform very complex multidimensional analysis on data. In both cases, the underlying database may need to handle responses to queries very quickly in order to satisfy systems requirements with respect to transaction time. Data stored by such systems may be stored in one or more partitions. Given the complexity of queries, volume of data stored, and/or their volume, the underlying databases face challenges in order to optimize performance.

SUMMARY

In some implementations, the current subject matter relates to a computer implemented method for executing management of ownership of data. The method may include selecting an index in a plurality of indexes. The index may correspond to a plurality of ranges of data values stored in a plurality of database slices of a database. The index further may correspond to a partitioning structure including a plurality of hierarchically arranged nodes. Each node may correspond to a range of data values in the plurality of ranges of data values stored in at least one database slice. The partitioning structure may be replicated across a plurality of computing systems. The method may further include executing, by a computing system in the plurality of computing systems, an update to one or more ranges of data values, and replicating, by the computing system, at least one of: a database slice including the updated one or more ranges of data values and a node including the updated one or more ranges of data values, to another computing system in the plurality of computing systems for storage of a replicate of the updated one or more ranges of data values.

In some implementations, the current subject matter can include one or more of the following optional features. In some implementations, the replication may include replicating the node including the updated one or more ranges of data values to the other computing system based on a number of updates to the one or more ranges of data values being greater than a predetermined threshold number of updates.

In some implementations, the update execution may include generating another database slice configured to store the updated one or more ranges of data values. The method may further include replicating, by the computing system, the generated another database slice to the other database system. The method may also include storing the generated other database slice by another computing, executing, by another computing system, an update to one or more ranges of data values in another database slice, and replicating, by another computing system, another database slice to the computing system and storing a replica of the updated one or more ranges of data values in another database slice by the computing system.

In some implementations, one or more slices in the plurality of slices may be configured to be owned by one or more computing systems in the plurality of computing systems independently of one or more nodes in the plurality of hierarchically arranged nodes. Ownership of one or more slices may be configured to be transferred independently of ownership of one or more nodes by one or more computing systems.

In some implementations, the replication may include an asynchronous replication.

In some implementations, execution of an update may include at least one of the following: an insertion of the update to one or more ranges of data values into one or more database slices, modification of data stored in one or more database slices using the update to one or more ranges of data values, deletion of data one or more database slices, and any combination thereof.

DETAILED DESCRIPTION

To address these and potentially other deficiencies of currently available solutions, one or more implementations of the current subject matter relate to methods, systems, articles of manufacture, and the like that can, among other possible advantages, provide multi-master data management, including ownership transfer of one or more data partitions.

Database management systems and operations performed on the data managed by a database management system have become increasingly complex. For example, a database management systems (or database for short) may support relatively complex online analytical processing (OLAP, which may perform multi-dimensional analysis) to more straightforward transaction based online transaction processing (OLTP). Moreover, the database may be configured as a row store database or column store database, each of which may have certain aspects with respect to queries and other operations at the database. For example, the database may encode data using dictionaries, while some databases may not. In addition to these various databases layer differences, the queries performed at a database may include a complex sequence of operations in order to generate corresponding responses. To implement the complex sequence, a query execution plan (or query plan for short) may be implemented. The query plan may represent a sequence of operations, such as instructions, commands, and/or the like, to access data in the database. The database may also include a query plan optimizer to determine an efficient way to execute the query plan.

From an application or client perspective, it may be extremely cumbersome to access databases. For example, an application may need to query different types of databases using complex queries. As a consequence, the application layer may need to be configured to handle the various types of databases and various query types. Additionally or alternatively, each database may need to process queries from the application into a format and structure that can be handled by the given database. Pushing complex operations and support for a variety of different database types to the application layer may contravene the need to have relatively lighter weight and/or readily deployable applications. On the other hand, pushing complex operations to the database layer where data is stored may draw processing and/or memory resources at the database and may thus reduce the performance and response times for queries on that database layer.

Database systems may store data using one or more partitioning configurations. A partition in a database may refer to a division of a logical database or its elements into separate independent parts. Partitioning allows improved manageability, performance, load balancing, etc. In some cases, partitions may be distributed over multiple nodes, where each node may allow users to perform various operations (e.g., execution of transactions, etc.) on a partition. Such distribution may increase performance for nodes that may be subject to frequent transactions that may involve retrieval, insertion, modification, generation of views of data, etc. while at the same time maintaining availability and security of data. Data partitioning may be performed by building separate smaller databases, splitting selected elements, etc. Data may be partitioned using horizontal or vertical partitioning methodologies. A horizontal partitioning may place different rows into different tables (e.g., splitting users of different age groups). A vertical partitioning may create new tables having fewer columns and may use additional tables to store any remaining columns.

In some implementations, the current subject matter may be configured to generate a partitioning specification for data that may be stored in a database system. The partitioning specification may be defined “on the fly” using slices of data that are included in the database and/or its partitions (it may be assumed that the data stored in the database is implicitly partitioned). The current subject matter may generate a partition specification using a tree structure, where nodes in a tree may correspond to specific data slices in the database. Data slices may be relatively small. This way, if a partitioning scheme is not ideal, there is no major drawback in terms of performance as the processing of small slices may be relatively quick. One of the advantages of this approach is that no costly re-organizations of data may be required.

In some implementations, one or more partitions of a database(s) may be located or stored on one or more servers that may be disposed at different locations (e.g., in different geographical regions). Servers may “own” a particular partition of data, such as, for example, by virtue of the partition being created and stored on that server, more frequently accessed on that server (e.g., users that are more frequently accessing a particular partition are located geographically proximate to the server). However, in some cases, ownership of partitions may need to be transferred (e.g., temporarily, permanently, etc.) from one server to another server. This may be done for the purposes of providing updates, performing transactions local to the other server, executing writes that may be local to that server, etc. In some exemplary implementations, once the other server completes the tasks, it may transfer the ownership of the partition back to the original server. In some implementations, ownership of an entire partition may be transferred. Alternatively, only a portion of a partition (e.g., a “branch” of a tree, as will be discussed below) may be transferred.

FIG. 1illustrates an exemplary system100for multi-master data management, according to some implementations of the current subject matter. The system100may include one or more users (user1, user2, . . . user n)102, an execution engine104, and a database system106, which may store data in one or more slices108. The users102, the execution engine104, and the database system106may be communicatively coupled with one another using any type of network, including but not limited to, wired, wireless, and/or a combination of both. The users102may include at least one of the following: computer processors, computing networks, software applications, servers, user interfaces, and/or any combination of hardware and/or software components. Database system106may include at least one of the following: databases, storage locations, memory locations, column stores, document stores, and/or any combination of hardware and/or software components. In some implementations, the database system106may be a High Performance Analytic Appliance (“HANA”) system as developed by SAP SE, Walldorf, Germany, as will be described below.

The execution engine104may include any combination of software and/or hardware components and may be configured to receive and execute a query from one or more users102to obtain data from one or more slices108in the database system106, insert data into from one or more slices108the database system106, modify data stored from one or more slices108in the database system106, delete data stored from one or more slices108in the database system106, generate one or more new slices108(e.g., for insertion of new data), etc., and any combination thereof. In some implementations, the execution engine106may be included in the database system106.

Execution of a query may typically require generation of a query plan or query execution plan, which may be an ordered set of operations that may be used to access stored data (e.g., access data in a SQL relational database management system). Upon submission of a query to the database system106, requested data may be retrieved based on parameters of the query. The retrieved data may be aggregated/joined with any other data that may be sought by the user. Insertion, modification, deletion, etc. of data in the database system106may be performed using various SQL or other statements.

FIG. 2illustrates an exemplary multi-master data management system200, according to some implementations of the current subject matter. The system200may include a first computing system202and a second computing system204communicatively coupled via a networking connection206. The systems202,204may include at least one of the following: one or more servers, one or more computing users, one or more databases, one or more memory locations, computer processors, computing networks, software applications, servers, user interfaces, and/or any combination of hardware and/or software components, and/or any other computing components. The communication link206may include any type of network, including but not limited to, wired, wireless, and/or a combination of both. The databases that may be disposed at the systems202,204may include at least one of the following: any storage locations, any memory locations, any column stores, any document stores, and/or any combination of hardware and/or software components.

The systems202,204may be configured to perform local transactions on data that may be stored in either of the systems. This may include any updates, deletions, modifications, writes, etc. of such data. Each system may own (either temporarily, permanently, etc.) any data that that is being accessed by the users of the system. In some implementations, ownership of data may be transferred from one system to another. Ownership of data may be transferred for a specific partition (e.g., a slice) of data, a portion of the specific partition (e.g., a branch of a tree in a partition specification). In some implementations, once the ownership of the data is transferred, the receiving system may become the new owner of that data until it is requested or may be determined that the ownership may need to be transferred back to the system that initially owned that data. Alternatively, the system that may provide update may never own the data from another system but instead, simply provide any requisite updates to the data.

FIG. 3illustrates another exemplary multi-master data management system300, according to some implementations of the current subject matter. The system300may be similar to the system200shown inFIG. 2. The system300may include a first system302(e.g., similar to system202(shown inFIG. 2) and located in Germany) communicatively coupled to a second system304(e.g., similar to system204(shown inFIG. 2) and located in the United States). Each system may store one or more slices (or partitions)306of data. Each location may also “own” a particular slice of data (and/or its version). As shown inFIG. 3, slices of data that are not shaded may be owned by the system302and slices that are shaded may be owned by the system304.

Each slice of data may also have an appropriate version of that slice associated with it. For example, slice1, owned by the location302may have a “version 40” associated with it. If an update is provided by the system302to slice1's version 40, such update may be communicated (as indicated by an arrow) to the corresponding slice1's version stored at the system304(as shown by updates to slices12-14to version 40 stored at system302that are communicated to update corresponding slice versions 39 stored at system304). Similarly, any updates to slices owned by and executed at system304may be communicated to the corresponding slices at system302(as shown by updates to slice4,7-10and15). As stated above, updates may be performed to specific slices and/or portions of slices. Such updates may allow users to operate (e.g., update, write, etc.) on different portions of the same slice or table (which may be replicated across both systems), whereby two separate systems302,304may logically appear as a single system.

FIG. 4illustrates exemplary partitioning specifications412and414(e.g., in the form of tree structures) that have been generated together with slices1-5402-410. As shown inFIG. 4, slices402-405may be tables that may include an index. For example, slice1402may include an identifier column along with minimum (“min”) and maximum (“max”) value columns. The identifier column may include an “id”, “city” and “order_date”. Each of these include a range of values, for example, the “id” identifier may include a range of “100-8100”, the “city” identifier may include a range of “Heidelberg-New York”, and the “order_date” identifier may include a range of “2018-04-01-2018-04-02”.

Similarly, slice2404may include different ranges for its identifiers. For example, the “id” identifier range may be “110-200”; the “city” identifier may include a range of “Aachen-Wuerzburg”, and the “order_date” identifier may include a range of “2018-04-01-2018-05-01”. In slice3406, the “id” identifier may include a range of “8100-8200”, the “city” identifier may include a range of “Chicago-Washington”, and the “order_date” identifier may include a range of “2018-04-02-2018-05-01”. In slice4408, the “id” identifier may include a range of “180-250”, the “city” identifier may include a range of “Bonn-Wolfsburg”, and the “order_date” identifier may include a range of “2018-04-15-2018-05-15”. Lastly, in slice5410, the “id” identifier may include a range of “8150-8250”, the “city” identifier may include a range of “Denver-Washington”, and the “order_date” identifier may include a range of “2018-04-16-2018-05-16”. In some implementations, as a consequence of the implicit partitioning, data may be arranged in a way that slices may contain data that may highly correlate respectively and, hence, implicitly organized by subject matter (e.g., slices2and4may include data related to orders in Germany; slices3and5may include data related to orders in the United States, and slice1may include all data relating to orders in the United States and Germany).

In some implementations, the slices may be used to generate a partition specification (e.g., trees412and414) where data may be organized based on ranges identified in the slices in accordance with the identifiers. The partition specifications may be used for insertion, retrieval, modification, deletion, etc. of data stored in the slices. Additionally, these specifications may be used for creation of new slices of data, if necessary or desired. In some implementations, the slices and the partition specification may grow simultaneously and may influence one another based on actual data value ranges that may be inserted. The latter may cause generation of slices, whereby, within each slice, correlation between one or more identifiers/fields (e.g., “id”) may be high.

As shown inFIG. 4, the tree412may be hierarchically organized based on the identifier “id” in the slices1-5, with the identifier “id” being a parent node413and may be linked to one or more child nodes. For example, the “id” parent node413may be linked to a node415corresponding to “id” being in a range of less than 200 (“<200”) and a node417corresponding to “id” being in a range of greater than or equal to 200 (“≥200”). As can be understood, any other number or numbers may be used for generation of a partitioning specification based on this tree. Further, more than one node may be linked to the parent node413.

The node415may be further linked to child nodes419and421, where node419may include data corresponding to “id” identifier being less than 160 (“<160”) and node421may include data corresponding to “id” identifier being greater than or equal to 160 (“≥160”). Further, node417may be linked to child nodes423and441, where the child node423may include data corresponding to “id” identifier being less than 8100 (“≤8100”) and node441may include data corresponding to “id” identifier being greater than or equal to 8100 (“≥8100”). Further, the node441may include its own child nodes443(values less than 8130) and445(values greater than or equal to 8130).

Based on this partitioning, node419may correspond to some or all of the data in slices1,2; node421may correspond to some or all of the data in slices1,2,4; node423may correspond to some or all of the data in slices1and4; node443may correspond to some or all of the data in slices1and3; and node445may include one or more data values in slices3and5.

The data values (or ranges of data values) corresponding to nodes in the tree412may be owned by different systems (e.g., systems302(Germany),304(United States) shown inFIG. 3). As shown inFIG. 4, data values corresponding to nodes413,415,417,419,421and423may be owned by the Germany system302, whereas data values corresponding to nodes441,443, and445may be owned by the United States system304.

Similarly, the tree414may be hierarchically organized based on the identifier “order_date” in the slices1-5, with the identifier “order_date” being a parent node427. The “order_date” parent node427may be linked to nodes429and431, where node429including data corresponding to “order_date” being in a range of less than 2018-04-10 (“<2018-04-10”) and node431corresponding to “order_date” being in a range of greater than or equal to 2018-04-10 (“≥2018-04-10”). As can be understood, any other order date or dates may be used for generation of a partitioning specification based on this tree and/or more than one node may be linked to the parent node427.

The node429may be further linked to child nodes433and435, where node433may include data corresponding to “order_date” identifier being less than 2018-04-03 (“<2018-04-03”) and node435may include data corresponding to “order_date” identifier being greater than or equal to 2018-04-03 (“≥2018-04-03”). Further, node431may be linked to child nodes437and439, where the child node437may include data corresponding to “order_date” identifier being less than 2018-05-01 (“<2018-05-01”) and node439may include data corresponding to “order_date” identifier being greater than or equal to 2018-05-01 (“≥2018-05-01”).

Based on this partitioning, node433may correspond to some or all of the data in slices1,2and3; node435may correspond to some or all of the data in slices2and3; node437may correspond to some or all of the data in slices2,3,4and5; and node439may correspond to some or all of the data in slices4and5. Similar to the partitioning tree412, the data values (or ranges of data values) corresponding to nodes in the tree414, may be owned by different systems shown inFIG. 3. For example, data values corresponding to nodes427,431,437, and439may be owned by the Germany system302and data values corresponding to nodes429,433, and435may be owned by the United States system304.

In some implementations, slices1-5may receive various updates, e.g., data inserts, data deletions, data modifications, etc. For example, a typical insertion of data may be executed using an INSERT statement (e.g., INSERT {“id”: 100, “city”: “Heidelberg”, “order_date”: “2018-04-01”}). Such inserts may be performed based on a location corresponding to the location that owns the data (e.g., an insert into slice2owned by the Germany system302), by a location that does not own the data (e.g., an update may be performed on a replica of a slice or a portion of a slice), and/or by both locations.

FIG. 5illustrates another exemplary multi-master data management system500, according to some implementations of the current subject matter. The system500may be similar to the systems200and300shown inFIGS. 2, 3, respectively. The system500may include a first system502(e.g., similar to system302(shown inFIG. 3) and located in Germany) communicatively coupled to a second system504(e.g., similar to system304(shown inFIG. 3) and located in the United States). Likewise, each system may own one or more slices (or partitions) of data. As shown inFIG. 5, slices of data that are not shaded may be owned by the system502and slices that are shaded may be owned by the system504.

As shown inFIG. 5, an update may be requested by a user connected to the Germany system502to a slice9of the data. It may be determined that slice9is owned by the United States system504(as shown by the shaded box). The update request may be sent to the system504for actual execution. An update may include generation of a new version of the slice (e.g., “version 41”). The system500may further determine that the system502may include an older version of the slice9(e.g., “version 39”) and hence, the system504may replicate the updated version (i.e., “version 41”) to the system502for storage. This way a remote system may perform updates to data that it does not own—by letting the owner perform the actual operation. It is the responsibility of the owner to perform data checks, handle transactional snapshots etc.

In some implementations, the system500may use updating frequency by a system (e.g., how often a particular slice or its portion are being updated) to determine whether ownership of a particular version of a slice may need to be changed or retained with the original system. A threshold updating frequency may be used for such determination. By way of a non-limiting example, if an updating frequency of a slice is more than 500 updates per minute by a system, ownership of that slice (assuming it is not owned by the updating system) may be transferred to that system. Referring toFIG. 5, if slice9is being updated by the system502less than 500 times per minute, the ownership of slice9may be retained with system504. Otherwise, it may be switched to system502.

In some implementations, the system500may determine that ownership of slices is being switched too frequently (e.g., exceeding another threshold). In that case, the system500may determine that instead of switching slice ownership, an ownership of a particular branch of a tree partition (e.g., trees412,414shown inFIG. 4) corresponding to a node in that branch that is being frequently updated may be switched (e.g., permanently, temporarily, etc.). Switching of branch ownership is further discussed in connection withFIGS. 6a-cbelow.

FIGS. 6a-cillustrate an exemplary multi-master data management system600, according to some implementations of the current subject matter. In particular,FIGS. 6a-cillustrate “id” partition specifications similar to the “id” partition specification shown inFIG. 4and that are based on slices1-5shown inFIG. 4. As shown in6a, the system600may include a first system602(i.e., “Germany view”) and a second system604(i.e., “US view”), which are similar to the corresponding systems shown inFIGS. 2-5.FIGS. 6a-cshow respective system states at a particular point in time. Due to the asynchronous nature of the replication between the systems, they may have a different state at such points in time. These different states are illustrated as respective “views”.

Each system602,604may include a copy of the “id” partition specification or tree612, i.e., system602may include tree612aand system604may include tree612b. As shown inFIG. 6a, the partition specification612may be owned by the system602(i.e., as shown by the unshaded circles). Each partition specification612may include a parent node613lined to child nodes615and617, which, in turn, may be linked to further child nodes.

FIG. 6aillustrates an exemplary update (e.g., insertion) of data that may be performed by the system604. In particular, as shown inFIG. 6a, the node617aof the system602may be linked to a branch that includes a node641athat is linked to two child nodes643aand645a. The branch headed by the node641amay be owned by the system604(i.e., US system). The node643amay include a range of data values less than 8130, corresponding to slices1and3, and node645amay include a range of data values greater than or equal to 8130, corresponding to slices3and5. The slices3and5, at the system602, may correspond to version 40 of data values contained in these slices.

The system604may execute an update process by updating data values (e.g., inserting) in the branch headed by the node645b. Specifically, two nodes647band649bmay be linked to the node645b. The node647bmay correspond to data values being less than 8500 and node649bmay correspond to data values being greater than or equal to 8500. Node647bmay correspond to new ranges of data values that may be used to update existing ranges of data values with a new version (e.g., “version 41” in slices3and5), as shown inFIG. 6a. Node649bmay correspond to new ranges of data values for which no previous data ranges or corresponding slices exist. Thus, new slices—slices6and7—may need to be generated. The slices6and7may correspond to the new version of ranges of data values (i.e., “version 41”).

As shown inFIG. 6a, to execute an update to the system602using data in system604, an asynchronous branch and/or slice replication process may be executed. This may allow both systems to operate while the update process is undergoing. During the update to system602, new branch data values (corresponding to node647b) may be replicated to the system602from system604(as shown by the arrows inFIG. 6a). New slices6and7corresponding to node649bmay be asynchronously replicated to the system602.

FIG. 6billustrates an exemplary process for execution of a data update process by the system602, which may occur after completion of the update process shown inFIG. 6a. As shown inFIG. 6, after the update process inFIG. 6ais completed, the system602may include nodes645a,647a, and649athat may correspond to nodes (and hence range of data values)645b,647b, and649bof system604, respectively.

In some implementations, the system602may determine that an updated range of data values has been provided with respect to slices5and6(e.g., “version 43”). The updated range of data values may be asynchronously replicated to the system604(by reversing direction of the replication). Moreover, the update may cause the system602to assume ownership of slices5and6, as shown by the unshaded slices. While the ownership of the slices has changed, it does not necessarily mean that the ownership of branches in the partitioning specification changes. As shown inFIG. 6b, the branches641-649remain owned by system604. This means that system602may use values in the given ranges (even though they are owned by another system), but it may not extend ranges, add sub-nodes etc.

FIG. 6cillustrates an exemplary process of changing ownership of branches upon a determination that one system has updated more slices. This may occur when updates by one system (e.g., system602) to a branch (or a node) (e.g., node649) owned by another system (e.g., system604) may exceed a predetermined threshold, where the threshold corresponds to a number of slices being updated (e.g., more than 50%) in a particular branch.

By a way of a non-limiting example, system602may execute updates to slices6and7with updated versions of ranges of data values (e.g., “version 43” for slice6and “version 44” for slice7). As a result of this update, slices4-7are now owned by system602, where the updates are asynchronously replicated to system604. Because the number of slices that are now owned by system602, the ownership of node649bmay be changed to system602from system604. This may happen as a result of more frequent updates by one system (e.g., system602) to data stored by another (e.g., system604).

In some implementations, the current subject matter can be implemented in various in-memory database systems, such as a High Performance Analytic Appliance (“HANA”) system as developed by SAP SE, Walldorf, Germany. Various systems, such as, enterprise resource planning (“ERP”) system, supply chain management system (“SCM”) system, supplier relationship management (“SRM”) system, customer relationship management (“CRM”) system, and/or others, can interact with the in-memory system for the purposes of accessing data, for example. Other systems and/or combinations of systems can be used for implementations of the current subject matter. The following is a discussion of an exemplary in-memory system.

FIG. 7illustrates an exemplary system700in which a computing system702, which can include one or more programmable processors that can be collocated, linked over one or more networks, etc., executes one or more modules, software components, or the like of a data storage application704, according to some implementations of the current subject matter. The data storage application704can include one or more of a database, an enterprise resource program, a distributed storage system (e.g. NetApp Filer available from NetApp of Sunnyvale, Calif.), or the like.

The one or more modules, software components, or the like can be accessible to local users of the computing system702as well as to remote users accessing the computing system702from one or more client machines706over a network connection710. One or more user interface screens produced by the one or more first modules can be displayed to a user, either via a local display or via a display associated with one of the client machines706. Data units of the data storage application704can be transiently stored in a persistence layer712(e.g., a page buffer or other type of temporary persistency layer), which can write the data, in the form of storage pages, to one or more storages714, for example via an input/output component716. The one or more storages714can include one or more physical storage media or devices (e.g. hard disk drives, persistent flash memory, random access memory, optical media, magnetic media, and the like) configured for writing data for longer term storage. It should be noted that the storage714and the input/output component716can be included in the computing system702despite their being shown as external to the computing system702inFIG. 7.

Data retained at the longer term storage714can be organized in pages, each of which has allocated to it a defined amount of storage space. In some implementations, the amount of storage space allocated to each page can be constant and fixed. However, other implementations in which the amount of storage space allocated to each page can vary are also within the scope of the current subject matter.

FIG. 8illustrates exemplary software architecture800, according to some implementations of the current subject matter. A data storage application704, which can be implemented in one or more of hardware and software, can include one or more of a database application, a network-attached storage system, or the like. According to at least some implementations of the current subject matter, such a data storage application704can include or otherwise interface with a persistence layer712or other type of memory buffer, for example via a persistence interface802. A page buffer804within the persistence layer712can store one or more logical pages806, and optionally can include shadow pages, active pages, and the like. The logical pages806retained in the persistence layer712can be written to a storage (e.g. a longer term storage, etc.)714via an input/output component716, which can be a software module, a sub-system implemented in one or more of software and hardware, or the like. The storage714can include one or more data volumes810where stored pages812are allocated at physical memory blocks.

In some implementations, the data storage application704can include or be otherwise in communication with a page manager814and/or a savepoint manager816. The page manager814can communicate with a page management module820at the persistence layer712that can include a free block manager822that monitors page status information824, for example the status of physical pages within the storage714and logical pages in the persistence layer712(and optionally in the page buffer804). The savepoint manager816can communicate with a savepoint coordinator826at the persistence layer712to handle savepoints, which are used to create a consistent persistent state of the database for restart after a possible crash.

In some implementations of a data storage application704, the page management module of the persistence layer712can implement a shadow paging. The free block manager822within the page management module820can maintain the status of physical pages. The page buffer804can include a fixed page status buffer that operates as discussed herein. A converter component840, which can be part of or in communication with the page management module820, can be responsible for mapping between logical and physical pages written to the storage714. The converter840can maintain the current mapping of logical pages to the corresponding physical pages in a converter table842. The converter840can maintain a current mapping of logical pages806to the corresponding physical pages in one or more converter tables842. When a logical page806is read from storage714, the storage page to be loaded can be looked up from the one or more converter tables842using the converter840. When a logical page is written to storage714the first time after a savepoint, a new free physical page is assigned to the logical page. The free block manager822marks the new physical page as “used” and the new mapping is stored in the one or more converter tables842.

The persistence layer712can ensure that changes made in the data storage application704are durable and that the data storage application704can be restored to a most recent committed state after a restart. Writing data to the storage714need not be synchronized with the end of the writing transaction. As such, uncommitted changes can be written to disk and committed changes may not yet be written to disk when a writing transaction is finished. After a system crash, changes made by transactions that were not finished can be rolled back. Changes occurring by already committed transactions should not be lost in this process. A logger component844can also be included to store the changes made to the data of the data storage application in a linear log. The logger component844can be used during recovery to replay operations since a last savepoint to ensure that all operations are applied to the data and that transactions with a logged “commit” record are committed before rolling back still-open transactions at the end of a recovery process.

With some data storage applications, writing data to a disk is not necessarily synchronized with the end of the writing transaction. Situations can occur in which uncommitted changes are written to disk and while, at the same time, committed changes are not yet written to disk when the writing transaction is finished. After a system crash, changes made by transactions that were not finished must be rolled back and changes by committed transaction must not be lost.

To ensure that committed changes are not lost, redo log information can be written by the logger component844whenever a change is made. This information can be written to disk at latest when the transaction ends. The log entries can be persisted in separate log volumes while normal data is written to data volumes. With a redo log, committed changes can be restored even if the corresponding data pages were not written to disk. For undoing uncommitted changes, the persistence layer712can use a combination of undo log entries (from one or more logs) and shadow paging.

The persistence interface802can handle read and write requests of stores (e.g., in-memory stores, etc.). The persistence interface802can also provide write methods for writing data both with logging and without logging. If the logged write operations are used, the persistence interface802invokes the logger844. In addition, the logger844provides an interface that allows stores (e.g., in-memory stores, etc.) to directly add log entries into a log queue. The logger interface also provides methods to request that log entries in the in-memory log queue are flushed to disk.

Log entries contain a log sequence number, the type of the log entry and the identifier of the transaction. Depending on the operation type additional information is logged by the logger844. For an entry of type “update”, for example, this would be the identification of the affected record and the after image of the modified data.

When the data application704is restarted, the log entries need to be processed. To speed up this process the redo log is not always processed from the beginning. Instead, as stated above, savepoints can be periodically performed that write all changes to disk that were made (e.g., in memory, etc.) since the last savepoint. When starting up the system, only the logs created after the last savepoint need to be processed. After the next backup operation the old log entries before the savepoint position can be removed.

When the logger844is invoked for writing log entries, it does not immediately write to disk. Instead it can put the log entries into a log queue in memory. The entries in the log queue can be written to disk at the latest when the corresponding transaction is finished (committed or aborted). To guarantee that the committed changes are not lost, the commit operation is not successfully finished before the corresponding log entries are flushed to disk. Writing log queue entries to disk can also be triggered by other events, for example when log queue pages are full or when a savepoint is performed.

With the current subject matter, the logger844can write a database log (or simply referred to herein as a “log”) sequentially into a memory buffer in natural order (e.g., sequential order, etc.). If several physical hard disks/storage devices are used to store log data, several log partitions can be defined. Thereafter, the logger844(which as stated above acts to generate and organize log data) can load-balance writing to log buffers over all available log partitions. In some cases, the load-balancing is according to a round-robin distributions scheme in which various writing operations are directed to log buffers in a sequential and continuous manner. With this arrangement, log buffers written to a single log segment of a particular partition of a multi-partition log are not consecutive. However, the log buffers can be reordered from log segments of all partitions during recovery to the proper order.

As stated above, the data storage application704can use shadow paging so that the savepoint manager816can write a transactionally-consistent savepoint. With such an arrangement, a data backup comprises a copy of all data pages contained in a particular savepoint, which was done as the first step of the data backup process. The current subject matter can be also applied to other types of data page storage.

In some implementations, the current subject matter can be configured to be implemented in a system900, as shown inFIG. 9. The system900can include a processor910, a memory920, a storage device930, and an input/output device940. Each of the components910,920,930and940can be interconnected using a system bus950. The processor910can be configured to process instructions for execution within the system900. In some implementations, the processor910can be a single-threaded processor. In alternate implementations, the processor910can be a multi-threaded processor. The processor910can be further configured to process instructions stored in the memory920or on the storage device930, including receiving or sending information through the input/output device940. The memory920can store information within the system900. In some implementations, the memory920can be a computer-readable medium. In alternate implementations, the memory920can be a volatile memory unit. In yet some implementations, the memory920can be a non-volatile memory unit. The storage device930can be capable of providing mass storage for the system900. In some implementations, the storage device930can be a computer-readable medium. In alternate implementations, the storage device930can be a floppy disk device, a hard disk device, an optical disk device, a tape device, non-volatile solid state memory, or any other type of storage device. The input/output device940can be configured to provide input/output operations for the system900. In some implementations, the input/output device940can include a keyboard and/or pointing device. In alternate implementations, the input/output device940can include a display unit for displaying graphical user interfaces.

FIG. 10illustrates an exemplary method1000for executing management of data ownership, according to some implementations of the current subject matter. At1002, an index (e.g., index “id” at system602or604shown inFIGS. 6a-c)) in a plurality of indexes may be selected. The index may correspond to a plurality of ranges of data values stored in a plurality of database slices (e.g., slices1-5shown inFIG. 4) of a database. The index corresponding to a partitioning structure (e.g., structure412shown inFIG. 4) may include a plurality of hierarchically arranged nodes (e.g., nodes413,415, etc.). Each node may correspond to a range of data values in the plurality of ranges of data values stored in at least one database slice. The partitioning structure may be replicated across a plurality of computing systems (e.g., systems602,604, as shown inFIGS. 6a-c). At1004, a computing system (e.g., system604) may execute an update to one or more ranges of data values (e.g., an update to a range of data values in nodes647b, new branch or node649b, new slices6,7, etc.). At1006, the computing system may replicate at least one of: a database slice including the updated one or more ranges of data values and a node including the updated one or more ranges of data values, to another computing system (e.g., system602) in the plurality of computing systems for storage of a replicate of the updated one or more ranges of data values.

In some implementations, the current subject matter can include one or more of the following optional features. In some implementations, the replication may include replicating the node including the updated one or more ranges of data values to the other computing system based on a number of updates to the one or more ranges of data values being greater than a predetermined threshold number of updates.

In some implementations, the update execution may include generating another database slice configured to store the updated one or more ranges of data values. The method may further include replicating, by the computing system, the generated another database slice to the other database system. The method may also include storing the generated other database slice by another computing, executing, by another computing system, an update to one or more ranges of data values in another database slice, and replicating, by another computing system, another database slice to the computing system and storing a replica of the updated one or more ranges of data values in another database slice by the computing system.

In some implementations, one or more slices in the plurality of slices may be configured to be owned by one or more computing systems in the plurality of computing systems independently of one or more nodes in the plurality of hierarchically arranged nodes. Ownership of one or more slices may be configured to be transferred independently of ownership of one or more nodes by one or more computing systems.

In some implementations, the replication may include an asynchronous replication.

In some implementations, execution of an update may include at least one of the following: an insertion of the update to one or more ranges of data values into one or more database slices, modification of data stored in one or more database slices using the update to one or more ranges of data values, deletion of data one or more database slices, and any combination thereof.

As used herein, the term “user” can refer to any entity including a person or a computer.