README
¶
Logical Sync
Package swosync handles the logical replication from the source DB to the destination database during switchover.
Theory of operation
Triggers record all changes (INSERT, UPDATE, DELETE) in the change_log table as table, row_id pairs, only tracking a set of changed rows but not their point-in-time data. The changes are then read in and applied in batches by reading the CURRENT state of the row from the source database and writing it to the destination database at the time of sync.
Replicating point-in-time differences between "snapshots" avoids the need for a sequential solution for concurrent updates and intermediate row states by only syncing the final result. It also becomes more efficient because each row must be replicated at most once, even when multiple updates occur between sync points.
The process depends on having a consistent view of the source database, which a serializable transaction can obtain, or during a stop-the-world lock (during the final sync).
Basic strategy
- Read all changes (table and row ids)
- Fetch row data for each changed row
- Insert rows from old DB that are missing in new DB, in fkey-dependency order
- Update rows from old DB that exist in both, in fkey-dependency order
- Delete rows missing from the old DB that exist in the new DB, in reverse-fkey-dependency order
- Delete synced entries from the
change_logtable - Repeat until both DBs are close in sync
- Obtain a stop-the-world lock
- Perform final sync, and update the
use_next_dbpointer - Release the stop-the-world lock
- New DB is used for all future transactions
Further Notes
It is essential to keep the sync loop as tight as possible, particularly in "final sync" mode. The final sync will pause all transactions during its synchronization process; this is necessary to ensure that the database is in a consistent state with no leftover changes before setting the use_next_db pointer.
Round Trips
- 1 to start tx, read all change ids & sequences (also stop-the-world lock in final mode)
- 1 to fetch row data from each table (single batch, 1 query per table)
- 1 to apply all updates to the new DB
- 1 to commit src tx (also updates
use_next_dbin final mode) - 1 to delete all synced change rows from the DB
An extra round-trip for the last delete is a trade-off to favor a shorter stop-the-world time since deleting the previous change records isn't necessary after the switchover.
Documentation
¶
Index ¶
- Constants
- type LogicalReplicator
- func (l *LogicalReplicator) FinalSync(ctx context.Context) error
- func (l *LogicalReplicator) FullInitialSync(ctx context.Context) error
- func (l *LogicalReplicator) LogicalSync(ctx context.Context) error
- func (l *LogicalReplicator) ResetChangeTracking(ctx context.Context) error
- func (l *LogicalReplicator) SetDestinationDB(db *pgx.Conn)
- func (l *LogicalReplicator) SetProgressFunc(fn func(ctx context.Context, format string, args ...interface{}))
- func (l *LogicalReplicator) SetSourceDB(db *pgx.Conn)
- func (l *LogicalReplicator) StartTrackingChanges(ctx context.Context) error
- type SequenceSync
- type TableSync
- func (c *TableSync) AddBatchChangeRead(b *pgx.Batch)
- func (c *TableSync) AddBatchRowReads(b *pgx.Batch)
- func (c *TableSync) AddBatchWrites(b *pgx.Batch)
- func (c *TableSync) ExecDeleteChanges(ctx context.Context, srcConn *pgx.Conn) (int64, error)
- func (c *TableSync) HasChanges() bool
- func (c *TableSync) ScanBatchChangeRead(res pgx.BatchResults) error
- func (c *TableSync) ScanBatchRowReads(res pgx.BatchResults) error
Constants ¶
const ConnLockQuery = `` /* 375-byte string literal not displayed */
ConnLockQuery will result in a failed assertion if it is unable to get the exec lock or switchover state is use_next_db
const ConnWaitLockQuery = `` /* 336-byte string literal not displayed */
ConnLockQuery will result in a failed assertion if it is unable to get the exec lock or switchover state is use_next_db
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type LogicalReplicator ¶
type LogicalReplicator struct {
// contains filtered or unexported fields
}
LogicalReplicator manages synchronizing the source database to the destination database.
func NewLogicalReplicator ¶
func NewLogicalReplicator() *LogicalReplicator
NewLogicalReplicator creates a new LogicalReplicator.
func (*LogicalReplicator) FinalSync ¶
func (l *LogicalReplicator) FinalSync(ctx context.Context) error
FinalSync will sync the source database to the destination database, using the stop-the-world lock and updating switchover_state to use_next_db.
func (*LogicalReplicator) FullInitialSync ¶
func (l *LogicalReplicator) FullInitialSync(ctx context.Context) error
FullInitialSync will insert all rows from the source database into the destination database.
While doing so it will update the rowID maps to track the rows that have been inserted.
func (*LogicalReplicator) LogicalSync ¶
func (l *LogicalReplicator) LogicalSync(ctx context.Context) error
LogicalSync will sync the source database to the destination database as fast as possible.
func (*LogicalReplicator) ResetChangeTracking ¶
func (l *LogicalReplicator) ResetChangeTracking(ctx context.Context) error
ResetChangeTracking disables tracking changes and truncates the tables in the destination database.
func (*LogicalReplicator) SetDestinationDB ¶
func (l *LogicalReplicator) SetDestinationDB(db *pgx.Conn)
SetDestinationDB sets the destination database and must be called before Start.
func (*LogicalReplicator) SetProgressFunc ¶
func (l *LogicalReplicator) SetProgressFunc(fn func(ctx context.Context, format string, args ...interface{}))
SetProgressFunc sets the function to call when progress is made, such as the currently syncing table.
func (*LogicalReplicator) SetSourceDB ¶
func (l *LogicalReplicator) SetSourceDB(db *pgx.Conn)
SetSourceDB sets the source database and must be called before Start.
func (*LogicalReplicator) StartTrackingChanges ¶
func (l *LogicalReplicator) StartTrackingChanges(ctx context.Context) error
StartTrackingChanges instruments and begins tracking changes to the DB.
- Creates the change_log table - Gets the list of tables and sequences to track - Creates the change trigger for each table - Disables triggers in the new DB - Waits for any in-flight transactions to finish (since these may not have picked up the change trigger)
type SequenceSync ¶
type SequenceSync struct {
// contains filtered or unexported fields
}
SequenceSync is a helper for synchronizing sequences.
func NewSequenceSync ¶
func NewSequenceSync(names []string) *SequenceSync
NewSequenceSync creates a new SequenceSync for the given sequence names.
func (*SequenceSync) AddBatchReads ¶
func (s *SequenceSync) AddBatchReads(b *pgx.Batch)
AddBatchReads queues up select statements to retrieve the current values of the sequences.
func (*SequenceSync) AddBatchWrites ¶
func (s *SequenceSync) AddBatchWrites(b *pgx.Batch)
AddBatchWrites queues up update statements to set the current values of the sequences.
func (*SequenceSync) ScanBatchReads ¶
func (s *SequenceSync) ScanBatchReads(res pgx.BatchResults) error
ScanReads scans the results of the added batch reads.
type TableSync ¶
type TableSync struct {
// contains filtered or unexported fields
}
TableSync is a helper for syncing tables from the source database to the target database.
func NewTableSync ¶
NewTableSync creates a new TableSync for the given tables.
func (*TableSync) AddBatchChangeRead ¶
func (c *TableSync) AddBatchChangeRead(b *pgx.Batch)
AddBatchChangeRead adds a query to the batch to read the changes from the source database.
func (*TableSync) AddBatchRowReads ¶
func (c *TableSync) AddBatchRowReads(b *pgx.Batch)
AddBatchRowReads adds a query to the batch to read all changed rows from the source database.
func (*TableSync) AddBatchWrites ¶
func (c *TableSync) AddBatchWrites(b *pgx.Batch)
func (*TableSync) ExecDeleteChanges ¶
ExecDeleteChanges executes a query to deleted the change_log entries from the source database.
func (*TableSync) HasChanges ¶
HasChanges returns true after ScanBatchChangeRead has been called, if there are changes.
func (*TableSync) ScanBatchChangeRead ¶
ScanBatchChangeRead scans the results of the change read query.
func (*TableSync) ScanBatchRowReads ¶
ScanBatchRowReads scans the results of the row read queries.
Source Files