Google Cloud - Community
Published in

Google Cloud - Community

Christoph Bussler

Jun 23, 2020

47 min read

Online Database Migration by Dual-Write: This is not for Everyone

(to be more precise: for almost no-one)

tl;dr

Online database migration

Key requirements and expectations

  • Migration is complete: all data in the source database is completely migrated (or a defined subset of the source database if not all data is to be migrated)
  • Migration is consistent: the target database is fully consistent (transactional consistency), no data loss, no data duplicates, and no transaction reordering
  • Migration is repeatable: target databases can be recreated and migration started from the beginning (for example, to mitigate unforeseen errors)
  • Migration does not impact client operations: clients operate without any impact during ongoing migration (for example, no downtime)
  • Migration does not impact client code: client code does not have to be changed to support migration to avoid any implementation effort (aka, design, testing, etc.)
  • Migration can be tested: testing of migration must be possible, especially of migrations where the source schema is different from the target schema, without impacting the production clients accessing the source databases

Brief terminology

  • Source, target database. The source database has the current data set and the target database is the database the data from the source database is migrated to. The target database might or might not have the same schema as the source database. If the schemas are different in the source and target database, data transformation is required.
  • Primary database. At any point in time, either the source database or the target database is the source of truth, aka, the primary database accessed by clients. In a given database migration there will be the point in time where the target database will be made the primary and becomes the authoritative source of truth for clients.

Baseline: CDC online database migration process

CDC online database migration architecture
  • Create a target database. This might be the same or a different database engine compared to the source database, and it might have the same or a different (modified schema) compared to the source database.
  • Create the migration logic using migration technology. This can be built-in functionality a database provides (like logical replication), or it can be a separate migration system like Striim or Fivetran (to name two examples from a large variety of available migration technologies).
  • Perform an initial load. This is the transfer (including possibly data transformation) of a transactionally consistent source database snapshot to the target database.
  • Perform continuous migration using CDC. Change data capture (CDC) is a mechanism that reads the database’s transaction log and applies committed source database transactions in commit order exactly once to the target database (including possibly data transformation). This process starts from the database time the initial snapshot was taken to ensure all concurrently ongoing transactions during the initial load are captured and migrated. At the end of the blog an alternative to transaction log access is discussed in case the source database does not expose those.
  • Complete draining. When the day and time of the cut-over of clients from the source database (former primary) to the target database (current primary) nears, write access to the source database is stopped. The remaining transactions from the transaction log that have not been migrated yet to the target database have to be migrated (“drained”) to the target database. If configured carefully, the draining takes place in a matter of seconds.
  • Cut-over clients. At this point the target database has all data from the source database, and their state is equivalent wrt. transactional consistency, clients connect to the target database, access it and continue the production workload.

Dual-write online database migration process

Dual-write online database migration architecture
  • Step 1: Build and deploy the “new” database schema onto production.
  • Step 2: Add a new DAO to your app that writes to the “new” database.
  • Step 3: Start writing to the “new” database but use the “old” one as primary.
  • Step 4: Enable the read path. Change the feature toggle to enable reading from both databases.
  • Step 5: Making the “new” database the primary one. Change the feature toggle to first write to the new database (you still read from both but now prefer the new DB).
  • Step 6: Stop writing to the “old” database (read from both).
  • Step 7: Eagerly migrate data from the “old” database to the “new” one.
  • Step 8: Delete the “old” DAO.

Preliminaries — system properties

  • Distributed transactions. Cross-database transactions are only available if you use a distributed transaction manager that can coordinate between the two servers chosen and the servers can participate in the distributed transaction protocol. In all other cases, transactions do not span the two databases and any access of both databases is not serialized (and therefore not automatically ensuring data consistency). Neither sequential nor concurrent access of both databases is serialized by the two databases involved).
  • Consistent backup. Since the two databases are independent of each other, it is not possible to have a consistent backup of both databases unless you stop write access on both when creating backups. Two backups, one for each database, are only consistent when there is no change during the backup in any of the databases.
  • High availability and disaster recovery. The databases are independent of each other wrt. high-availability or disaster recovery. If one database (source or target database) fails over then the fail over process is independent of the other database. There is in general a possibility of downtime as well as a set of lost transactions during high-availability or disaster recovery that might violate data consistency (in the absence of distributed transactions). In this case both databases are out of sync and have to be brought in sync first before continuing migration. For example, the fail over of the target database loses transactions that are present in the source database. The lost transactions have to be determined and migrated again.
  • Networking and latency. The two databases in general might run in different environments with different network properties (like throughput or routing structure). This difference might impact latency between clients and the database access interface. In the absence of distributed transactions this might lead to unintentional reordering of transactions between the source and target database from a client perspective leading to data inconsistency.

Preliminaries — sample use case: user and addresses

Sample use case: user and its addresses

Discussion of dual-write database migration process

Client code modification

Multiple concurrent clients

Data access path management

  • Insert: data has to be inserted into both databases
  • Update: data is to be updated in the target database (in addition to the source database) only if the data was migrated before
  • Delete: data is to be deleted in the target database (in addition to the source database) only if the data was migrated before

Process

Details of migration logic functionality

  • Primary key row insert (without relationship). A single row insert by primary key can take place in each, the source and target database, in the same way. In the use case, a new user can be added as a single insert.
  • Primary key row insert (with explicit relationship). A single row insert by primary key that has a foreign key relationship can be different for the source database and the target database. In the use case, a new address for a given user is a single insert into the source database. In the target database, however, the code has to first check if the user is present since the address has a foreign key relationship to the user. If not, the user has to be migrated from the source before the address can be inserted.
  • Primary key row insert (with implicit relationship). A single row insert by primary key that has an implicit relationship can be different for the source database and the target database. An implicit relationship is not defined by the schema, but managed by the client by means of code. If an insert requires related data to be present (one or more relationships), the code for the target database has to check for the data presence and has to ensure its migration if it is not in the target database. Implicit relationships are difficult as those cannot be found by examining the database schema, but only by understanding the business logic in the client code.
  • Primary key row batch insert. Batch inserts on the target only work for those insert use cases that do not involve a relationship. As soon as a relationship is involved checks have to be performed (see discussion in bullet points above).
  • Primary key row update (all columns). If all columns are updated, in the target database a check has to ensure that the row is present first. If not, it has to be migrated first, and then updated. Alternatively the update can be converted into an insert. It depends on additional context (like database triggers or predicates in the update statement or relevance for transaction log content) which way is chosen.
  • Primary key row update (subset of columns). If a subset of columns are to be updated, the row has to be migrated first, before the update takes place.
  • Primary key row update with relationships. A row update with explicit or implicit relationships (like a foreign key) has to ensure that the mandatory related data (one or many) is present first in order to not violate the referential integrity. In addition, the row to be updated has to be present or migrated first.
  • Primary key row batch updates. Batch updates on the target database only work if all rows that are to be updated are present. Any update requiring additional logic cannot be part of a batch update.
  • Primary key row delete (without relationship). A delete can take place on the target database if the row is present or not. If it is not present, the delete will succeed. However, context matters: if transactions logs on the target have to be consistent with those on the source as well, then a delete has to be preceded with migrating the data first so that the delete actually takes place on the target database. Other examples are triggers that might require the delete to actually take place.
  • Primary key row delete (with implicit or explicit relationship). A delete or a row might have to trigger the delete of additional rows. For example, if a user is deleted, all addresses have to be deleted. In case of relationships specified in a schema the database supervises those; in case of implicit relationships the code has to ensure proper traversal and deletion itself.
  • Primary key row batch deletes. As before, a batch delete only works if no additional logic has to be executed when deleting.
  • Read from source as the primary database. If the source database is the primary database all read operations are executed on the source database as it has the source of truth and represents the consistent database state.
  • Read from target as the primary database. The target database can only be the primary database if it has the complete set of data that is consistent as well. Until the source is switched off, this implies that both, the source and the target are 100% consistent (aka, a perfect copy of each other or are perfectly in sync wrt. data consistency in case of different schemas). The above process taken from [2] does not accomplish this: it makes the target database the primary (step 5) without having migrated all data (step 7). I think the term “primary” in the reference refers to which database is written to first, and has no meaning wrt. data consistency — which is the customary use of the term “primary” in database architectures.
  • Read from both databases? Reference [2] mentions obtaining read results by accessing both databases and combining (“merging”) the results based on timestamps in the schema (step 4). This has several implications: first, a timestamp has to be added to every table and the timestamp has to be taken from the same clock across two databases. This is not accurately possible unless the databases are within the same database instance. Second, this means that neither the source nor the target database have a consistent (and complete) data sets. It is not clear to me how it can come to this situation and how to manage consistency (I suspect the discussion in step 3 means that write failures to the target database can be ignored). However, since the references [2] mentions it, it is important to mention it here because of the data consistency implications.
  • Aggregation. Any form of aggregation depends on a complete data set (or a complete subset that is in scope of the aggregation queries). If a query counts the number of unique addresses, all users and all addresses have to be accessible. In a dual-write situation this is only supported and possible if at any point in time the source or the target is the primary database that is consistent and complete. If the dual-write variant cannot ensure a single consistent database at any point, aggregation queries will return inaccurate results during the data migration. Alternatively, aggregation queries have to be rewritten to derive their result by accessing both databases.
  • Join. Like in case of aggregation operations, a join only provides correct data on a consistent and complete data set.
  • Stored procedures, triggers, etc. In some cases database schemas have stored procedures, triggers, or additional operations that are database local. The individual implementation has to determine if those can operate concurrently on the source database and the target database. If they cannot work concurrently on both (meaning, on incomplete data on the target database), those will have to be switched off on the target database (non-primary database) until it is data complete and consistent. However, if switched off, any data migrated, updated, etc. will not cause any side effect that is required for consistency (e.g., a stored procedure updating the database itself). This will make the dual-write migration an impossible variant to use.
  • External database access. Some database schemas use triggers to communicate with downstream systems, or use external tables to access upstream systems. As before, the specific implementation has to determine if those can be enabled on the target database during migration and if consistent behavior is taking place.
  • Transaction logs. Some use cases rely on transactions logs. Changes made in the database are observed and downstream functionality is triggered. In order for these use cases to work it is necessary that operations in the source as well as the target database are executed in the same way so that the transaction logs are equivalent as well. For example, as discussed above, an update on the target must not be converted into an insert if the data was not migrated before.
  • One-level relationships: As discussed above, one-level relationships — relationships between two tables — that are implicit or explicit have to be addressed by the code accessing the target database.
  • Multi-level relationships: Like one-level relationships multi-level relationships have to be taken care of by the client code accessing the target database. In a schema with many multi-level relationships complex traversals might be required to ensure that the required data is migrated during operations that require the data to be available. In the worst case, if the data set represents a tightly connected graph, the first operation requiring migration might have to migrate over all data before it can continue (e.g., insert or update).

Data models and transformation

Tables without primary keys

Consistency

  • Partial transaction abort (transaction coordination). As discussed in the preliminaries above, transactions cannot span the source and the target database (unless migration between tables within the same database is implemented). This means, for example, that an update of a primary key row in the source and the target database are two different transactions: one on the source and one on the target database ([2] points this out explicitly). In order for the databases to be consistent, both transactions have to succeed or fail. While distributed transactions would guarantee this behavior, in their absence the client code has to ensure this behavior. This means that a client has to be able to recover from a source transaction failure or a target transaction failure if the transaction of the other database succeeds. In addition, this recovery has to work in presence of concurrent transactions on the same object (primary key) while transaction failures are taking place (this is core database logic of concurrency and serializability). If implemented correctly, it replicates the two-phase distributed transaction protocol in the client code.
  • Sequential transaction reordering. If a client issues two updates right after each other on the same user (same primary key), four transactions will be executed. However, in the absence of distributed transactions, the order of the two transactions on the target database can be the reverse from those on the source database, leading to data inconsistencies, as the user’s state would different in the source and the target database. This can only be avoided if the client code explicitly ensures that sequential transactions on the same primary key are executed in that order on both, the source and the target database.
  • Concurrent transaction reordering. If two different clients issue for example an update on the same user (same primary key) concurrently, four transactions are executed. Since the transactions are independent of each other, they can be executed on the source and the target database in a different order. Unless client code implements cross-client coordination, data inconsistency can occur as the state of the source and target database can be different.
  • Concurrent transactions requiring data migration. One of the use cases is that two concurrent updates are initiated by clients on a user (same primary key) whereby the user has not yet been migrated to the target database. Both updates will determine that the user has to be migrated first (see above update discussion). While the first client migrates the user first before executing the update, the second client might try to do the same (as it is executing concurrently and unaware that the migration took place in the meanwhile), experiencing a failing insert (migration) because the data has been migrated already by the first client. This scenario has to be explicitly covered by the client code. One possible approach is re-execution of the second client’s update transaction.
  • Concurrent transactions with relationships. Similar to the discussion in the previous bullet, concurrent transactions issued by clients might require migration of related data. One use case is when migrating a user, its addresses have to be migrated as well. Concurrent transactions might try to do this twice, as discussed before. A different use case is n:1 relationships where the “1” side has to be migrated only once, while each of the “n” has to be migrated n times (each row once). In this case n-1 transactions have to be prepared for the case that the “1” side was already migrated.
  • The most helpful approach is to have distributed transactions as many possible violations would be caught automatically by it.
  • Another approach is to have each transaction check if it encounters a consistent data set before it applies changes, and that it leaves a consistent data set behind after making changes. The client would abort the transaction if it encounters or would leave behind an inconsistent state (however, that would run into the larger transaction coordination problem as discussed above).
  • Yet another approach is comparing source and target databases on a regular schedule. However, the question arises: what to do if inconsistencies are detected? How would those be mitigated and fixed?
  • Various references (e.g. [3]) suggest constant monitoring or continued statistical analysis. While this might detect inconsistencies (or might not as it is not 100% guaranteed to find all inconsistencies), the question arises how to react to found inconsistencies and how to compensate or fix them. And, what to do if inconsistent data was read by users or downstream systems that in turn modified data based on the read data?

Back fill

  • Concurrent client data migration. While back fill is ongoing, and while it determines that it needs to migrate a data item, that data item might have already been migrated concurrently or is in the process of being migrated concurrently in a concurrent client transaction. This concurrent processing must guarantee the data consistency and back fill for this data item must fail in this case to avoid interference.
  • Data relationships. Client code by its nature understands and is aware of data relationships (e.g., that a user has addresses in a foreign key relationship). Back fill as separate code needs to ensure that it implements the exact same data semantics in its logic. This applies not only for explicit relationships, but also for implicit relationships. Depending on how the migration code is implemented and structured, back fill might reuse the same code. If it implements its own code, migration functionality is duplicated and needs to be kept in sync with the corresponding client code.
  • Client code logic. Some client code logic might not only migrate the data, but might update additional tables, like for example history tables to keep track of data changes. Back fill has to implement the exact same logic.

Fallback option

Schema updates

System failures

  • Client(s)
  • Source database
  • Target database

Completion

Migration between tables in the same database

Testing

  • Stop the migration, fix the migration code, test the fixes (and update the test suite), and reset as well as restart the migration from the beginning
  • Stop the migration, fix the code, fix the data, and continue the migration

Observation: data consistency uncertainty

  • [1] expresses uncertainty about data consistency between source and target database: “We need to be sure that it’s safe to read from the new Subscriptions table: our subscription data needs to be consistent. We’ll use GitHub’s Scientist to help us verify our read paths.” First of all, this means that there is the expectation that inconsistencies can occur, meaning, it is “clear” that the client code performing the migration most likely will not create a consistent target database. A separate system (Scientist) is put in place comparing the source and the target database alerting any difference when reading the “same” data item. However, it is not explained how differences are dealt with, and how the code that created the inconsistency is found and fixed. All this has to take place during a running migration as the goal is to perform an online migration (aka, zero downtime). So while the inconsistencies are addressed, additional inconsistencies might be created. In addition, in order to find all inconsistencies, the system detecting inconsistencies must have a complete coverage of all possible inconsistencies.
  • [1] also expresses that there is concern after the target database is made the primary database: “For each code path, we’ll need to use a holistic approach to ensure our changes are safe. We can’t just substitute new records with old records: every piece of logic needs to be considered carefully. If we miss any cases, we might end up with data inconsistency. Thankfully, we can run more Scientist experiments to alert us to any potential inconsistencies along the way.” This means that there is a constant and continued concern about data consistency and that the client code might not have addressed all access correctly.
  • [2] realizes that code might fail and states “Notice that in this step the “old” database is in a consistent state while the “new” database can potentially be inconsistent since the writes to it can fail while the “old” database write succeeded. It is important to let this step run for a while (several days or even weeks) before moving to the next step. This will give you the confidence that the write path of your new code works as expected and that the “new” database is configured correctly with all the replications in place.” This means that production is monitored to detect any issues. Since there is no guarantee when errors show up, a longer time frame of observation is suggested expecting that errors show up if there are any. Like in the other references it is not described how any missing or inconsistent data is corrected.
  • [3] states “Over the course of several days, we keep an eye on our bug reporting service (Bugsnag) for incorrect single reads in production and fix them as they come in.” This (like the other references) indicates the uncertainty if data is consistent and correct, that the bug system is able to record errors and that the bugs are fixed in production. As discussed, the bug fix probably refers to the code, and it is not clear if the bug caused downstream issues if inconsistent data was used.
  • [3] employed additional monitoring to check for correctness: “We also add a regular audit that verifies that our waivers table has 0 rows and our subscriptions table only contains processed subscriptions!” After the data was migrated from the waivers table (a source table), it was empty, and had to stay empty if all the code locations work properly. Since there was uncertainty if all code locations were refactored, monitoring the waiver table was necessary to see if any incorrect write takes place.

When is dual-write online database migration an option to consider?

  • A data set might represent the data as closed and complete data structure. For example, all data about a user are in a single document and there are no references to and from that document to any other. All access is by primary key only and a client reads and writes the complete document every time. No aggregation or join queries are in the system. Another example is a Key/Value store where the data item is self-contained in one K/V (thanks Antonio for mentioning this specific case to me). In such a case, dual-write has less aspects to consider. What needs to be addressed are for example concurrent access supervision, system failure handling, back fill, aka, all aspects that are not related to individual query processing.
  • Another use case might be an extremely simple data model and schema that allows understanding the data set completely independent of any client code accessing it. In this case it might be possible to implement verification after every database access that migrates or modifies the data set. I am not sure how many production systems fall into this category. One interesting note here is that the use case of [1] has a simple schema, yet its migration has to address a lot of the aspects discussed here.
  • A very different use case is the situation where the client code can be written in such a way that it guarantees data consistency by its construction. This means that the code addresses all the above aspects completely and does not leave the possibility of data inconsistency (neither by itself, or any external event like a system failure). The hard part is to ensure that this is actually the case and to prove it.

Dual-write process variation

Why is CDC database migration more reliable?

What if the source database does not have a transaction log for CDC?

Conclusion

Additional discussion (added 8/27/2020)

ATM operations: BASE

  • In the general case, any writer of the source database would have to keep a ledger of all operations so that it is known what happened in the source database while the target was offline (huge engineering effort).
  • That ledger would have to be written in such a way that any concurrent access to the same primary key data set is serializable later (so that changes are applied in the correct order during reconciliation). This would require a global time service.
  • Depending on the implementation of dual-write, the source might or might not have the latest consistent state when the disconnect happens.
  • If it does not, consistency is compromised as the application will work on outdated state of the source database. Any ledger reconciliation would reset the target database and lose all newer state.
  • If it does, then the application would see a consistent state when accessing the source database.
  • Assuming that writing the ledger would be transactional consistent (e.g., it is located in the same source database — as distributed transactions are not available in general), and that it can be written in such a way that reconciliation is possible by applying the same operations to the target database then we have a good system, if, and only if the changes are reconciled on the target once the target is back up and running again before (!) any application modification access of the target system.
  • This is exactly how CDC works (with the variation of reconciliation). So this would implement a poor man’s CDC system that the application has to implement.

Data Lake

  • The data is written into the large store and will never be changed anymore. In this case “replaying” means creating a copy and possibly transforming the data to meet the target database’s schema.
  • If the data written will be changed (or possibly be changed), then the question arises of how replay would work to pick up the individual changes over time. Either the whole complete data set is being copied regularly to a target, or an incremental approach is applied. Complete copying on a regular basis has many aspects that might not make it practical for all cases (including delay). The only incremental approach available is CDC.

Active-active replication

  • The Redbook is here: Readings in Database Systems — Fifth Edition, the Stonebraker article starts on page 12. Stonebraker discusses online fail over pointing out that it is superior to have replicas be synchronized with the master database in active-active mode rather than active-passive.
  • While this is true in the context of replication, this also requires transactional coordination between master and replica. If the replica becomes unavailable, the transactions on the master will fail, and this will stop application processing.
  • The point of zero downtime migration is exactly that: the source can continue processing with the target being consistently advanced, however, without a transactional coordination. This is where CDC comes in: CDC supports the replay of source changes to the target database to ensure it is consistent in transaction commit order — an option that Stonebraker does not discuss.

References

Disclaimer

Database Migration
Change Data Capture
Database
Google Cloud Platform

--

--

More from Google Cloud - Community

A collection of technical articles and blogs published or curated by Google Cloud Developer Advocates. The views expressed are those of the authors and don't necessarily reflect those of Google.

Read more from Google Cloud - Community

AboutHelpTermsPrivacy

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store
Christoph Bussler

Christoph Bussler

107 Followers

www.real-programmer.com

Help

Status

Writers

Blog

Careers

Privacy

Terms

About

Text to speech