Change data capture state tracking for multi-region multi-master noSQL database
A method for change data capture state tracking includes accessing a distributed database including a plurality of clusters, each cluster of the plurality of clusters including a respective plurality of partitions, each cluster of the plurality of clusters configured to receive read and write operation. The method includes receiving, at a second cluster, a plurality of changes for a second table and storing the plurality of changes at a replication log. The method also includes asynchronously replicating the plurality of changes from the second table to a first table and maintaining a respective change stream position tracking a respective position in the replication log indicating one or more changes of the plurality of changes that have been replicated. The method includes receiving a read request at the first cluster requesting one or more rows of the first table and returning the respective change stream position.
1 . A computer-implemented method executed by data processing hardware that causes the data processing hardware to perform operations comprising:
accessing a distributed database comprising a plurality of clusters, each cluster of the plurality of clusters comprising a respective plurality of partitions, each cluster of the plurality of clusters configured to receive read and write operations, wherein:
the respective plurality of partitions of a first cluster of the plurality of clusters stores a first table comprising a first respective plurality of rows, each respective partition of the respective plurality of partitions of the first cluster comprising a respective portion of the first respective plurality of rows; and
the respective plurality of partitions of a second cluster of the plurality of clusters stores a second table comprising a second respective plurality of rows, each respective partition of the respective plurality of partitions of the second cluster comprising a respective portion of the second respective plurality of rows;
receiving, at the second cluster, a plurality of changes for the second table;
storing the plurality of changes at a replication log;
asynchronously replicating the plurality of changes from the second table to the first table;
while asynchronously replicating the plurality of changes from the second table to the first table, for each row of the first respective plurality of rows of each respective partition of the respective plurality of partitions of the first cluster, maintaining a respective change stream position, the respective change stream position tracking a respective position in the replication log indicating one or more changes of the plurality of changes that have been replicated;
receiving a read request at the first cluster requesting one or more rows of the first respective plurality of rows, the one or more rows associated with a first subset of partitions of the respective plurality of partitions of the first cluster; and
in response to the read request, returning, for each respective partition of the first subset of partitions, the respective change stream position.
2 . The method of claim 1 , wherein returning the respective change stream position comprises returning an encrypted blob comprising the respective change stream position.
3 . The method of claim 1 , wherein the respective plurality of partitions of the first cluster is different from the respective plurality of partitions of the second cluster.
4 . The method of claim 3 , wherein each respective change of the plurality of changes for the second table is associated with a respective partition of the respective plurality of partitions of the second cluster.
5 . The method of claim 4 , wherein the operations further comprise, for each respective change of the plurality of changes for the second table:
determining a different partition of the respective plurality of partitions of the first cluster that aligns with the respective partition of the respective plurality of partitions of the second cluster associated with the respective change; and
adding the respective change to a respective partition replication log associated with the respective partition of the respective plurality of partitions of the first cluster.
6 . The method of claim 4 , wherein the operations further comprise, for each respective change of the plurality of changes for the second table, determining that the respective partition of the respective plurality of partitions of the second cluster associated with the respective change is active.
7 . The method of claim 4 , wherein the operations further comprise, for each respective change of the plurality of changes for the second table, determining that the respective partition of the respective plurality of partitions of the second cluster associated with the respective change is inactive.
8 . The method of claim 7 , wherein the operations further comprise, in response to determining that the respective partition of the respective plurality of partitions of the second cluster associated with the respective change is inactive, determining that a respective incarnation associated with the respective partition of the respective plurality of partitions of the second cluster associated with the respective change is closed.
9 . The method of claim 4 , wherein the operations further comprise, for a respective change of the plurality of changes for the second table:
determining that the respective partition of the respective plurality of partitions of the second cluster associated with the respective change is new; and
in response to determining that the respective partition of the respective plurality of partitions of the second cluster associated with the respective change is new, determining that the respective partition of the respective plurality of partitions of the second cluster associated with the respective change belongs to a new incarnation.
10 . The method of claim 1 , wherein:
the first cluster comprises a plurality of nodes; and
each partition of the respective plurality of partitions of the first cluster is stored on a respective node of the plurality of nodes.
11 . A system comprising:
data processing hardware; and
memory hardware in communication with the data processing hardware, the memory hardware storing instructions that when executed on the data processing hardware cause the data processing hardware to perform operations comprising:
accessing a distributed database comprising a plurality of clusters, each cluster of the plurality of clusters comprising a respective plurality of partitions, each cluster of the plurality of clusters configured to receive read and write operations, wherein:
the respective plurality of partitions of a first cluster of the plurality of clusters stores a first table comprising a first respective plurality of rows, each respective partition of the respective plurality of partitions of the first cluster comprising a respective portion of the first respective plurality of rows; and
the respective plurality of partitions of a second cluster of the plurality of clusters stores a second table comprising a second respective plurality of rows, each respective partition of the respective plurality of partitions of the second cluster comprising a respective portion of the second respective plurality of rows;
receiving, at the second cluster, a plurality of changes for the second table;
storing the plurality of changes at a replication log;
asynchronously replicating the plurality of changes from the second table to the first table;
while asynchronously replicating the plurality of changes from the second table to the first table, for each row of the first respective plurality of rows of each respective partition of the respective plurality of partitions of the first cluster, maintaining a respective change stream position, the respective change stream position tracking a respective position in the replication log indicating one or more changes of the plurality of changes that have been replicated;
receiving a read request at the first cluster requesting one or more rows of the first respective plurality of rows, the one or more rows associated with a first subset of partitions of the respective plurality of partitions of the first cluster; and
in response to the read request, returning, for each respective partition of the first subset of partitions, the respective change stream position.
12 . The system of claim 11 , wherein returning the respective change stream position comprises returning an encrypted blob comprising the respective change stream position.
13 . The system of claim 11 , wherein the respective plurality of partitions of the first cluster is different from the respective plurality of partitions of the second cluster.
14 . The system of claim 13 , wherein each respective change of the plurality of changes for the second table is associated with a respective partition of the respective plurality of partitions of the second cluster.
15 . The system of claim 14 , wherein the operations further comprise, for each respective change of the plurality of changes for the second table:
determining a different partition of the respective plurality of partitions of the first cluster that aligns with the respective partition of the respective plurality of partitions of the second cluster associated with the respective change; and
adding the respective change to a respective partition replication log associated with the respective partition of the respective plurality of partitions of the first cluster.
16 . The system of claim 14 , wherein the operations further comprise, for each respective change of the plurality of changes for the second table, determining that the respective partition of the respective plurality of partitions of the second cluster associated with the respective change is active.
17 . The system of claim 14 , wherein the operations further comprise, for each change of the plurality of changes for the second table, determining that the respective partition of the respective plurality of partitions of the second cluster associated with the respective change is inactive.
18 . The system of claim 17 , wherein the operations further comprise, in response to determining that the respective partition of the respective plurality of partitions of the second cluster associated with the respective change is inactive, determining that a respective incarnation associated with the respective partition of the respective plurality of partitions of the second cluster associated with the respective change is closed.
19 . The system of claim 14 , wherein the operations further comprise, for a respective change of the plurality of changes for the second table:
determining that the respective partition of the respective plurality of partitions of the second cluster associated with the respective change is new; and
in response to determining that the respective partition of the respective plurality of partitions of the second cluster associated with the respective change is new, determining that the respective partition of the respective plurality of partitions of the second cluster associated with the respective change belongs to a new incarnation.
20 . The system of claim 11 , wherein:
the first cluster comprises a plurality of nodes; and
each partition of the respective plurality of partitions of the first cluster is stored on a respective node of the plurality of nodes.