Parallel batch metadata transfer update process within sharded columnar database system

A database management system includes a cluster of nodes over which a columnar database has been sharded into shards to which the nodes are subscribed. Source nodes of the cluster store up-to-date metadata objects for the shards, and target nodes of the cluster storing out-of-date metadata objects for the shards. A coordinator node initiates a parallel batch metadata transfer update process in which the out-of-date metadata objects at the target nodes are updated in parallel from the up-to-date metadata objects at the source nodes. The coordinator node can determine pairs of source nodes and target nodes; the source node and the target node of each pair respectively store an up-to-date metadata object and an out-of-date metadata object for a same shard.

BACKGROUND

Data is the lifeblood of many entities like business and governmental organizations, as well as individual users. Large-scale storage of data in an organized manner is commonly achieved using databases. Databases are collections of information that are organized for easy access, management, and updating. Data may be stored in tables over rows (i.e., records or tuples) and columns (i.e., fields or attributes). In a relational database, the tables have logical connections, or relationships, with one another, via keys, which facilitates searching, organization, and reporting of the data stored within the tables.

DETAILED DESCRIPTION

As noted in the background, databases store data in tables over rows and columns, where the tables can be interrelated with one another in relational databases. In more traditional row-oriented databases, tables store data by rows. By comparison, in column-oriented databases, which are also referred to as columnar databases, tables store data by columns.

Different columns, or different sets of columns, can be physically stored in different files, or file objects. Each file object has an associated metadata object. The metadata object stores metadata regarding a corresponding file object, such as the name of the file, how and whether the file is encrypted, and so on. The metadata object stores metadata regarding the data of the file object, too, such as the maximum and minimum values of the columns stored in the file object, and other information that may assist query processing.

Sharding a column-oriented database refers to the process of partitioning the rows of the database into different shards having corresponding file objects and associated metadata objects. For example, the rows may be sharded into multiple shards that each correspond to a different range of the values of one or more columns. Each shard includes a group of rows, and more specifically the columns of each of these rows. Sharding can be implemented using hashing, in which the values of one or more columns of a row are hashed to identify a particular shard to which the row is to be sharded.

In a shared-storage database, the file objects and their metadata objects of the shards are globally stored, and accessible by multiple nodes that can process queries in relation the data stored in the file objects. The nodes may be separate servers or other types of processing nodes that include processors, memory, and storage space. In a share-nothing database, each shard's file and metadata objects are accessible by one corresponding node of a cluster of nodes. Each node may locally store the objects for the shard(s) for which the node is responsible.

A modification of the shared-storage and share-nothing databases is to permit more than one node of the cluster to each locally store the metadata object of a shard. The file object for a shard may also be stored locally at each node, or may be stored on a shared storage. A node that stores the file and metadata objects for a shard is said to be subscribed to the shard. Each shard can have more than one subscribing node, and each node can subscribe to more than one shard. However, not all nodes are usually subscribed to all shards, as is effectively the case in a shared-storage database.

The number of nodes subscribed to a shard can vary over time. A subscribing node may fail and then recover. During lower-usage times of the database, subscribing nodes may go offline, and then brought back online during higher-usage times of the database. New nodes may be added over time as subscribers to the shards, to accommodate overall increased usage of the database.

When a new node is added to a database as a subscriber of a shard, or a prior node subscribing to the shard fails and recovers or goes offline and is brought back online, the node's metadata object for the shard will be out-of-date. That is, while a node is offline, it will have missed any updates to the metadata object for the shard. The metadata object has to be updated from the metadata of an up-to-date object of another node that also subscribes to the shard. The updating process can include transferring checkpoint and/or transaction logs and performing the operations identified in the logs until the out-of-date metadata object is in the same state as the up-to-date metadata object, and thus also up-to-date.

Each time a node has to have a metadata object brought up-to-date, a lock is placed on the cluster of nodes as a whole until the metadata object is updated. That is, the out-of-date metadata object of a node for a shard to which the node subscribes is brought up-to-date in a separate transaction. If multiple nodes have out-of-date metadata objects, they are serially updated over separate transactions. In each transaction, a lock is placed on the cluster of nodes.

The number of transactions is therefore equal to the number of nodes that have out-of-date metadata objects. If a large number of nodes have out-of-date metadata objects, a large number of transactions are serially processed, during which time the database cannot be updated due to the locks. Serially processing the transactions to update the out-of-data metadata objects of multiple nodes can take considerable time, during which database performance decreases.

Techniques described herein improve database performance by decreasing the length of time to update nodes having out-of-data objects for shards to which the nodes subscribe. A coordinator node of the cluster identifies each node storing an out-of-date metadata object for a shard. Such nodes are referred to as target nodes. For each out-of-date metadata object of each target node, the coordinate node identifies a node storing an up-to-date metadata object for the same shard. Such nodes are referred to as source nodes.

The coordinator node initiates a parallel batch metadata transfer update process in which the target nodes are updated in parallel from their respectively identified source nodes. Updating occurs more quickly, because the target nodes are updated in parallel, within one transaction. Multiple transactions are not needed. A lock is placed on the cluster of nodes prior to initiation of the update process, and released once the update process has finished. Because updating occurs within one transaction, multiple successive locks on the cluster are not needed.

FIG. 1shows an example system100including a columnar database102, which may be the Vertica database management system available from Vertica Systems, of Cambridge, Mass., which is a subsidiary of Micro Focus International plc, of Newbury, U.K. The database102has been sharded into shards104. Each shard104encompasses different values of one or more columns of the database102and stores the rows of the database102having these column values. As noted above, the database102can be sharded into shards104by hashing the values of one or more columns of each row to identify the shard104encompassing these columnar values and thus the shard to which each row belongs.

Nodes106are subscribed to each shard104. Each node106locally stores a metadata object110for a shard104to which it subscribes (i.e., including metadata regarding the file object108for the shard104). In the implementation ofFIG. 1, each node106also locally stores a file object108storing the data of a shard104to which it subscribes (i.e., including the values of the columns of the database102to which the shard104corresponds). In another implementation, the file objects108for the shards104can be stored on shared storage, and not locally at the nodes106.

As depicted inFIG. 1, more than one node106is subscribed to each shard104. More generally, there is at least one node106subscribed to each shard104; there can be shards104that each have just one subscribing node106. As depicted inFIG. 1, each node106subscribes to one shard104. More generally, each node106subscribes to one or more shards104; the nodes106can each subscribe to multiple shards104. For every shard104to which a node106subscribes, the node106stores a different metadata object110, and in the implementation ofFIG. 1, a different file object108. Every node106is not subscribed to every shard104. The nodes106are together referred to as a cluster of nodes106, over which the database102has been sharded into the shards104.

FIG. 2shows an example node cluster200. The cluster200includes the nodes106over which the columnar database102ofFIG. 1has been sharded into shards104, as well as a coordinator node202. The coordinator node202may be one of the nodes106over which the database102has been sharded, or may be a different node as is specifically shown inFIG. 2. The nodes106and202are communicatively connected to one another over a network204, such as an Ethernet network, a storage area network (SAN), or another type of network.

Each node106and202is a computing node, such as a server. Each node106and202may thus include hardware, such as one or more processors, memory, and physical storage. Each node106can process queries against the database102with respect to the columnar values of each shard104to which it subscribes. The physical storage of each node106stores the file object108and the metadata object110for each shard104to which it subscribes.

FIG. 3Ashows an example method300for updating the nodes106storing out-of-date metadata objects110for their subscribed shards104. The coordinator node202can perform the method300. The method300may be implemented as program code stored on a non-transitory computer-readable data storage medium and executable by a processor of the coordinator node202. The method300may be periodically performed, on a schedule or on-demand. When the method300is performed, a number of nodes106may be storing out-of-date metadata objects110for the shards104to which they are subscribed.

As noted above, an example of a node106storing an out-of-date metadata object110for a shard104is a node that failed and which has been brought back online. Another example of such a node106is a node that went offline and subsequently came back online according to a schedule. A third example of such a node106is a node that has been added to the cluster200of and subscribed to one or more shards104to accommodate increased usage of the columnar database102.

The coordinator node202determines the nodes106that have out-of-date metadata objects110(302). The nodes106identified in part302are referred to as target nodes. Each target node has at least one out-of-date metadata object110. Each out-of-date metadata object110of a target node corresponds to a different shard104to which the target node subscribes.

For each out-of-date metadata object110of each target node, the coordinator node202determines a node106having an up-to-date metadata object110for the same shard104(304). The nodes106identified in part304are referred to as source nodes. Identification of the source nodes thus defines source node-target node pairs. Each pair includes a target node and a source node subscribed to the same shard104. The target node of a pair stores an out-of-date metadata object110for the shard104, and the source node of the pair stores an up-to-date metadata object110for the shard104.

A target node can be part of more than one source node-target node pair. For example, a target node may be subscribed to more than one shard104and have an out-of-date metadata object110for each shard104. The target node will therefore be part of multiple pairs equal in number to the out-of-date metadata objects110that the target node stores.

A source node can similarly be part of more than one source node-target node pair. For example, for a shard104for which multiple target nodes each store an out-of-date metadata object110, a source node may have an up-to-date metadata object110. The source node can therefore be part of each pair including a target node that stores an out-of-date metadata object110for this shard104. The source node may not be part of every such pair, however, if a different source node also subscribes to the shard104.

As another example, a source node may be subscribed to more than one shard104and have an up-to-date metadata object110for each shard104. For each shard104to which the source node is subscribed, the same or different target node may store an out-of-date metadata object110. The source node can therefore be part of each pair including a target node that stores an out-of-date metadata object110for one of the shards104to which the source node subscribes. The source node may not be part of every such pair, however, if a different source node also subscribes to any of these shards104.

In determining the source nodes to pair with the target nodes, the coordinator node202may employ a variety of different approaches. For example, the coordinator node202may attempt to minimize the overall number of source nodes that are included in the source node-target node pairs. Therefore, a source node that stores up-to-date metadata objects110for multiple shards104for which target nodes store out-of-date metadata objects110may be selected over multiple source nodes that each store an up-to-date metadata object110for just one of these shards104.

As another example, the coordinator node202may attempt to minimize the number of source nodes that are included in the source node-target node pairs for any given target node. For example, a source node that stores up-to-date metadata objects for multiple shards104for which a given target node stores out-of-date metadata objects may be selected over multiple source nodes that each store an up-to-date metadata object110for just one of these shards. The coordinator node202may also take into account inter-node distance, inter-node network bandwidth, node processing capacity, and other factors in determining the source node-target node pairs.

The coordinator node202may acquire a lock on the cluster200of nodes106and202(306). The lock can be referred to as a commit lock. While the coordinator node202holds the lock over the cluster200as a whole, no file object108and thus no metadata object110of any node106can be changed. The columnar database102therefore cannot be updated while the coordinator node202holds the lock.

The coordinator node202initiates a parallel batch metadata transfer update process (308). In the process, the target nodes of the source node-target node pairs have their out-of-date metadata objects110respectively updated in parallel from the up-to-date metadata objects110of the pairs' source nodes. The process is a parallel process because the target nodes are updated in parallel. The process is a batch process because the target nodes are updated in one transaction. The process is a metadata process because metadata at the target nodes is updated. The process is a transfer process because metadata is transferred from the source nodes to the target nodes. The process is an update process because the target nodes are updated.

Once the parallel batch metadata transfer update process has been completed, the coordinator node202releases the lock that it is holding on the cluster200of nodes106and202(310). Updating of the columnar database102can thus resume. Per the method300, updating of the metadata objects110at multiple target nodes can occur within one transaction, and not over successive transactions, in which a total of one commit lock is held, as opposed to multiple locks.

FIG. 3Bshows an example method320of the parallel batch metadata transfer update process of part308as the process pertains to one source node-target node pair. The method320is performed for each pair defined in part304ofFIG. 3A. The coordinator node instructs the source node and target node of a pair that the target node is to update its out-of-date metadata object110for a shard104from the up-to-date metadata object110that the source node stores for the shard104(322). Once the coordinator node instructs the nodes of a source node-target node pair, the coordinator node can immediately proceed with instructing the nodes of another pair (i.e., in another iteration of the method320), until the nodes of every pair have been instructed.

The source node transmits metadata of its metadata object110for the shard104that is sufficient to bring the metadata object110for the shard104at the target node up-to-date (324). As noted above, this update process can include transferring checkpoint and/or transaction logs from the source node to the target node. The target node may indicate its latest state to the source node, for instance, so that the source node can determine which logs to send to the target node.

The target node updates its metadata object110for the shard104from the received metadata (326), to bring the metadata object110up-to-date. As noted above, this update process can include performing the operations identified in each log received from the source node. Once the target node has performed the operations, its metadata object110for the shard104is therefore up-to-date.

The target node notifies the coordinator node that the metadata object110for this shard104has been updated (328). When the coordinator node receives this notification from the target node of every source node-target node pair (i.e., for each iteration of the method320), the parallel batch metadata transfer update process is complete. The coordinator node may then release the commit lock from the cluster200, in part310ofFIG. 3A.

FIG. 4illustratively depicts example source node-target node pairs402A,402B, . . . ,404N, collectively referred to as the pairs402. The coordinator node202may determine the pairs402by performing parts302and304of the method300. The source node-target node pairs402respectively pertain to shards412A,412B, . . . ,412N, collectively referred to as the shards412. Each pair402pertains to one shard412. However, more than one source node-target node pair402may pertain to the same shard412. For example, the shard412A may be the same shard as the shard412B.

The source node-target node pairs402respectively included source nodes404A,404B, . . . ,404N, collectively referred to as the source nodes404, and target nodes406A,4066, . . . ,406N, collectively referred to as the target nodes406. Each pair402includes one source node404and one target node406. A source node404may be part of more than one pair402. For example, the source nodes404A and404B may be the same node. A target node406may likewise be part of more than one pair402. For example, the target nodes406A and406B may be the same node.

The source nodes404respectively store up-to-date metadata objects408A,4086, . . . ,408N, collectively referred to as the metadata objects408, for the shards412of their pairs402. The target nodes406similarly respectively store out-of-date metadata objects410A,410B, . . . ,410N, collectively referred to as the metadata objects410, for the shards412of their pairs402. For example, the pair402A includes the source node404A and the target node406A that respectively store metadata objects408A and410A for the same shard412A.

While more than one pair402may pertain to the same shard412, and each source node404and each target node406may be part of more than one pair402, each pair402identifies a unique tuple of a particular shard412, a particular source node404, and a particular target node406. For example, while the source node404A of the pair402A may be a part of one or more other pairs402, the pair402A is the only pair402that includes this source node404A and the target node406A, and that pertains to the shard412A. As another example, while the nodes404B and406B of the pair402B may both be part of one or more other pairs402, the pair402B is the only pair402including these two nodes404B and406B and that pertains to the shard412B.

For each pair402, the metadata object410pertaining to the shard412at the target node406is updated from metadata of the metadata object408pertaining to the same shard412at the source node404. The target node406of each pair402is updated in parallel with the target node406of every other pair402. Such a parallel, or concurrent, batch metadata transfer update process occurs more quickly than a serial update process in which the target nodes406are updated sequentially. Database performance improves, because normal database usage can resume sooner.

FIGS. 5A and 5Bshow different examples of how multiple source node-target node pairs can include the same node. The coordinator node202can determine the pairs ofFIGS. 5A and 5Bin parts302and304of the method300. InFIG. 5A, there are two source node-target node pairs502A and502B, which are collectively referred to as the pairs502. InFIG. 5B, there are two source node-target node pairs522A and522B, which are collectively referred to as the pairs502.

InFIG. 5A, both source node-target node pairs502include the same target node506. The pair502A also includes the source node504A and the pair502B also includes the source node504B. The source nodes504A and504B are collectively referred to as the source nodes504, and are different from one another. The source node504A and the target node506pertain to the shard508A, and the source node504B and the target node506pertain to the shard508B. The shards508A and508B are collectively referred to as the shards508, and are different from one another.

The source node504A stores an up-to-date metadata object510A for the shard508A, whereas the source node504B stores an up-to-date metadata object512B for the shard508B. The target node506stores an out-of-date metadata object510B for the shard508A, and a different out-of-date metadata object512A for the shard508B. The target node506is updated from both the source nodes504during the parallel batch metadata transfer update process. Specifically, the target node506's metadata object510B is updated in parallel from the source node504A's metadata object510A, and the target node506's metadata object512A is updated from the source node504B's metadata object512B.

InFIG. 5B, both source node-target node pairs522include the same source node524. The pair522A also includes the target node526A and the pair522B also includes the target node526B. The target nodes526A and526B are collectively referred to as the target nodes526, and are different from one another. The target nodes526both pertain to the shard528, as does the source node524.

The target node526A stores an out-of-date metadata object530A for the shard528, and the target node526B stores an out-of-date metadata object530B for the shard528. The source node524stores an up-to-date metadata object530C for the shard528. Both target nodes526are updated in parallel from the source node524during the parallel batch metadata transfer update process. Specifically, the target node526A's metadata object530A and the target node526B's metadata object530B are each updated from the source node524's metadata object530C.

FIG. 6Ashows how an example of how multiple source nodes and multiple target nodes can pertain to the same multiple shards. Source nodes602A and602B, collectively referred as the source nodes602, each pertain to shards606A and606B, collectively referred to as the shards606. Similarly, target nodes604A and604B, collectively referred to as the target nodes604, each pertain to the shards606.

The source nodes602respectively store up-to-date metadata objects608A and608B for the shard606A, which are collectively referred to as the metadata objects608. The source nodes602likewise respectively store up-to-date metadata objects612A and612B for the shard606B, which are collectively referred to as the metadata objects612. The target nodes604respectively store out-of-date metadata objects610A and610B for the shard606A, which are collectively referred to as the metadata objects610. The target nodes604likewise respectively store out-of-date metadata objects614A and614B for the shard606B, which are collectively referred to as the metadata objects614.

FIGS. 6B and 6Cshow different examples of how source node-target node pairs can be defined with respect to the source nodes602, the target nodes604, and the shards606. The coordinator node202can determine the pairs ofFIGS. 6B and 6Cin parts302and304of the method300. The coordinator node202may select the pairs ofFIG. 6Bas opposed to the pairs ofFIG. 6C(or vice-versa) depending on the approach that the coordinator node202uses to determine source node-target node pairs, examples of which have been noted above.

InFIG. 6B, four source node-target node pairs620have been identified. The first two pairs620both pertain to the source node602A and the target node604A, but to different shards606. The target node604A has both its metadata object610A for the shard606A and its metadata object614A for the shard606B updated from the same source node602A. Specifically, the target node604A's metadata object610A is updated from the source node602A's metadata object608A for the shard606A, and the target node604A's metadata object614A is updated from the source node602A's metadata object612A for the shard606B.

The second two pairs620inFIG. 6Bboth pertain to the source node602B and the target node604B, but to different shards606. The target node604B has both its metadata object610B for the shard606A and its metadata object614B for the shard606B updated from the same source node602B. Specifically, the target node604B's metadata object610B is updated from the source node602B's metadata object608B for the shard606A, and the target node604B's metadata object614B is updated from the source node602B's metadata object612B for the shard606B.

InFIG. 6C, four different source node-target node pairs630have been identified. The first two pairs630both pertain to the source node602A and the shard606A, but to different target nodes604. Both target nodes604have their respective metadata objects610for the shard606A updated from the same source node602A. Specifically, the metadata objects610of both target nodes604are updated from the source node602A's metadata object608A for the shard606A.

The second two pairs630inFIG. 6Cboth pertain to the source node602B and the shard606B, but to different target nodes604. Both target nodes604have their respective metadata objects612for the shard606B updated from the same source node602B. Specifically, the metadata objects612of both target nodes604are updated from the source node602B's metadata object612B for the shard606B.

FIG. 7shows an example method700. A coordinator node of a cluster of nodes over which a columnar database has been sharded into shards can perform the method700. The coordinator node determines that a first target node of the cluster is subscribed to a first shard of the plurality of shards and stores a metadata object for the first shard that is out-of-date (702). The coordinator node determines a first source node of the cluster that is subscribed to the first shard and that stores a metadata object for the first shard that is up-to-date (704). The coordinator node initiates a parallel batch metadata transfer update process comprising updating the metadata object for the first shard at the first target node from the metadata object at the first source node (706).

FIG. 8shows an example non-transitory computer-readable data storage medium storing program code (800). The program code is executable by a coordinator node of a cluster of nodes, over which a columnar database has been sharded into shards to which the nodes are subscribed, to perform processing (802). The processing includes determining pairs of source nodes and target nodes of the cluster of nodes (804). The source node and the target node of each pair respectively storing an up-to-date metadata object and an out-of-date metadata object for a same shard. The processing includes initiating a parallel batch metadata transfer update process in which the target nodes of the pairs are updated in parallel (806). For each pair, the out-of-date metadata object at the target node of the pair is updated from the up-to-date metadata object at the source node of the pair.

FIG. 9shows an example database management system (900). The system900includes a cluster of nodes902over which a columnar database has been sharded into shards to which the nodes are subscribed. The cluster of nodes902includes source nodes904storing up-to-date metadata objects for the shards, and target nodes906storing out-of-date metadata objects for the shards. The cluster of nodes902includes a coordinator node to initiate a parallel batch metadata transfer update process in which the out-of-date metadata objects at the target nodes are updated in parallel from the up-to-date metadata objects at the source nodes.

Techniques have been described to update nodes storing out-of-date metadata objects for shards of a columnar database from nodes storing up-to-date metadata objects for the shards. Rather than sequentially update the nodes storing out-of-date metadata objects, source node-target node pairs are identified so that the target nodes can be updated in parallel from the source nodes. Database performance improves, by minimizing the amount of time the database is unavailable when updating the out-of-date metadata objects.