diff --git a/cursor.go b/cursor.go index 3c83a1ea..7f3eef07 100644 --- a/cursor.go +++ b/cursor.go @@ -36,6 +36,8 @@ type CursorConfig struct { DB *sql.DB Throttler Throttler + SourceRuntime *SourceRuntime + ColumnsToSelect []string BuildSelect func([]string, *TableSchema, PaginationKey, uint64) (squirrel.SelectBuilder, error) // BatchSize is a pointer to the BatchSize in Config.UpdatableConfig which can be independently updated from this code. @@ -45,10 +47,22 @@ type CursorConfig struct { ReadRetries int } +func (c *CursorConfig) resolvedDB() *sql.DB { + if c.SourceRuntime != nil { + if db := c.SourceRuntime.DB(); db != nil { + return db + } + } + return c.DB +} + // returns a new Cursor with an embedded copy of itself func (c *CursorConfig) NewCursor(table *TableSchema, startPaginationKey, maxPaginationKey PaginationKey) *Cursor { + cfg := *c + // Bind the cursor to the current source connection for the whole scan. + cfg.DB = c.resolvedDB() return &Cursor{ - CursorConfig: *c, + CursorConfig: cfg, Table: table, MaxPaginationKey: maxPaginationKey, RowLock: true, diff --git a/data_iterator.go b/data_iterator.go index 0586fd8a..80e8fa1c 100644 --- a/data_iterator.go +++ b/data_iterator.go @@ -9,6 +9,7 @@ import ( type DataIterator struct { DB *sql.DB + SourceRuntime *SourceRuntime Concurrency int SelectFingerprint bool @@ -28,6 +29,16 @@ type TableMaxPaginationKey struct { MaxPaginationKey PaginationKey } +// currentSourceDB resolves the source per call, not at startup, so a runtime swap is observed between reads. +func (d *DataIterator) currentSourceDB() *sql.DB { + if d.SourceRuntime != nil { + if db := d.SourceRuntime.DB(); db != nil { + return db + } + } + return d.DB +} + func (d *DataIterator) Run(tables []*TableSchema) { d.logger = LogWithField("tag", "data_iterator") @@ -38,8 +49,10 @@ func (d *DataIterator) Run(tables []*TableSchema) { d.StateTracker = NewStateTracker(0) } + db := d.currentSourceDB() + d.logger.WithField("tablesCount", len(tables)).Info("starting data iterator run") - tablesWithData, emptyTables, err := MaxPaginationKeys(d.DB, tables, d.logger) + tablesWithData, emptyTables, err := MaxPaginationKeys(db, tables, d.logger) if err != nil { d.ErrorHandler.Fatal("data_iterator", err) } diff --git a/data_iterator_sorter.go b/data_iterator_sorter.go index 1a293106..b5f36400 100644 --- a/data_iterator_sorter.go +++ b/data_iterator_sorter.go @@ -60,7 +60,7 @@ func (s *MaxTableSizeSorter) Sort(unorderedTables map[*TableSchema]PaginationKey strings.Join(databaseSchemas, `", "`), strings.Join(tableNames, `", "`), ) - rows, err := s.DataIterator.DB.Query(query) + rows, err := s.DataIterator.currentSourceDB().Query(query) if err != nil { return orderedTables, err diff --git a/ferry.go b/ferry.go index 4d822ebd..8de2b23f 100644 --- a/ferry.go +++ b/ferry.go @@ -102,13 +102,15 @@ func (f *Ferry) NewDataIterator() *DataIterator { dataIterator := &DataIterator{ DB: f.SourceDB, + SourceRuntime: f.sourceRuntime, Concurrency: f.Config.DataIterationConcurrency, SelectFingerprint: f.Config.VerifierType == VerifierTypeInline, ErrorHandler: f.ErrorHandler, CursorConfig: &CursorConfig{ - DB: f.SourceDB, - Throttler: f.Throttler, + DB: f.SourceDB, + SourceRuntime: f.sourceRuntime, + Throttler: f.Throttler, BatchSize: &f.Config.UpdatableConfig.DataIterationBatchSize, BatchSizePerTableOverride: f.Config.DataIterationBatchSizePerTableOverride, @@ -220,6 +222,7 @@ func (f *Ferry) NewChecksumTableVerifier() *ChecksumTableVerifier { return &ChecksumTableVerifier{ SourceDB: f.SourceDB, + SourceRuntime: f.sourceRuntime, TargetDB: f.TargetDB, DatabaseRewrites: f.Config.DatabaseRewrites, TableRewrites: f.Config.TableRewrites, @@ -239,6 +242,7 @@ func (f *Ferry) NewInlineVerifier() *InlineVerifier { return &InlineVerifier{ SourceDB: f.SourceDB, + SourceRuntime: f.sourceRuntime, TargetDB: f.TargetDB, DatabaseRewrites: f.Config.DatabaseRewrites, TableRewrites: f.Config.TableRewrites, @@ -312,6 +316,7 @@ func (f *Ferry) NewIterativeVerifier() (*IterativeVerifier, error) { v := &IterativeVerifier{ CursorConfig: &CursorConfig{ DB: f.SourceDB, + SourceRuntime: f.sourceRuntime, BatchSize: &f.Config.UpdatableConfig.DataIterationBatchSize, BatchSizePerTableOverride: f.Config.DataIterationBatchSizePerTableOverride, ReadRetries: f.Config.DBReadRetries, @@ -319,6 +324,7 @@ func (f *Ferry) NewIterativeVerifier() (*IterativeVerifier, error) { BinlogStreamer: f.BinlogStreamer, SourceDB: f.SourceDB, + SourceRuntime: f.sourceRuntime, TargetDB: f.TargetDB, CompressionVerifier: compressionVerifier, diff --git a/inline_verifier.go b/inline_verifier.go index 909b3f90..4251a222 100644 --- a/inline_verifier.go +++ b/inline_verifier.go @@ -258,6 +258,7 @@ type InlineVerifierMismatches struct { type InlineVerifier struct { SourceDB *sql.DB + SourceRuntime *SourceRuntime TargetDB *sql.DB DatabaseRewrites map[string]string TableRewrites map[string]string @@ -274,7 +275,9 @@ type InlineVerifier struct { sourceStmtCache *StmtCache targetStmtCache *StmtCache - logger Logger + // sourceDBMu guards SourceDB and sourceStmtCache. + sourceDBMu sync.Mutex + logger Logger // Used only for the ControlServer initiated VerifyDuringCutover backgroundVerificationResultAndStatus VerificationResultAndStatus @@ -524,8 +527,23 @@ func (v *InlineVerifier) VerifyDuringCutover() (VerificationResult, error) { }, nil } +// currentSourceDB returns the source DB and its matching statement cache; on a handle swap the cache is replaced (statements are bound to the old DB) but not closed, as in-flight workers may still hold it. +func (v *InlineVerifier) currentSourceDB() (*sql.DB, *StmtCache) { + v.sourceDBMu.Lock() + defer v.sourceDBMu.Unlock() + + if v.SourceRuntime != nil { + if current := v.SourceRuntime.DB(); current != nil && current != v.SourceDB { + v.SourceDB = current + v.sourceStmtCache = NewStmtCache() + } + } + return v.SourceDB, v.sourceStmtCache +} + func (v *InlineVerifier) getFingerprintDataFromSourceDb(schemaName, tableName string, tx *sql.Tx, table *TableSchema, paginationKeys []interface{}) (map[string][]byte, map[string]map[string][]byte, error) { - return v.getFingerprintDataFromDb(v.SourceDB, v.sourceStmtCache, schemaName, tableName, tx, table, paginationKeys) + db, stmtCache := v.currentSourceDB() + return v.getFingerprintDataFromDb(db, stmtCache, schemaName, tableName, tx, table, paginationKeys) } func (v *InlineVerifier) getFingerprintDataFromTargetDb(schemaName, tableName string, tx *sql.Tx, table *TableSchema, paginationKeys []interface{}) (map[string][]byte, map[string]map[string][]byte, error) { diff --git a/iterative_verifier.go b/iterative_verifier.go index 0ba1b07c..b5dd5e45 100644 --- a/iterative_verifier.go +++ b/iterative_verifier.go @@ -120,6 +120,7 @@ type IterativeVerifier struct { BinlogStreamer *BinlogStreamer TableSchemaCache TableSchemaCache SourceDB *sql.DB + SourceRuntime *SourceRuntime TargetDB *sql.DB Tables []*TableSchema @@ -153,7 +154,7 @@ func (v *IterativeVerifier) SanityCheckParameters() error { return errors.New("BinlogStreamer must not be nil") } - if v.SourceDB == nil { + if v.currentSourceDB() == nil { return errors.New("SourceDB must not be nil") } @@ -289,6 +290,15 @@ func (v *IterativeVerifier) Result() (VerificationResultAndStatus, error) { return v.verificationResultAndStatus, v.verificationErr } +func (v *IterativeVerifier) currentSourceDB() *sql.DB { + if v.SourceRuntime != nil { + if db := v.SourceRuntime.DB(); db != nil { + return db + } + } + return v.SourceDB +} + func (v *IterativeVerifier) GetHashes(db *sql.DB, schemaName, tableName, paginationKeyColumn string, columns []schema.TableColumn, paginationKeys []interface{}) (map[string][]byte, error) { sql, args, err := GetMd5HashesSql(schemaName, tableName, paginationKeyColumn, columns, paginationKeys) if err != nil { @@ -595,7 +605,7 @@ func (v *IterativeVerifier) compareFingerprints(paginationKeys []interface{}, ta go func() { defer wg.Done() sourceErr = WithRetries(5, 0, v.logger, "get fingerprints from source db", func() (err error) { - sourceHashes, err = v.GetHashes(v.SourceDB, table.Schema, table.Name, table.GetPaginationColumn().Name, v.columnsToVerify(table), paginationKeys) + sourceHashes, err = v.GetHashes(v.currentSourceDB(), table.Schema, table.Name, table.GetPaginationColumn().Name, v.columnsToVerify(table), paginationKeys) return }) }() @@ -627,7 +637,7 @@ func (v *IterativeVerifier) compareFingerprints(paginationKeys []interface{}, ta } func (v *IterativeVerifier) compareCompressedHashes(targetDb, targetTable string, table *TableSchema, paginationKeys []interface{}) ([]string, error) { - sourceHashes, err := v.CompressionVerifier.GetCompressedHashes(v.SourceDB, table.Schema, table.Name, table.GetPaginationColumn().Name, v.columnsToVerify(table), paginationKeys) + sourceHashes, err := v.CompressionVerifier.GetCompressedHashes(v.currentSourceDB(), table.Schema, table.Name, table.GetPaginationColumn().Name, v.columnsToVerify(table), paginationKeys) if err != nil { return nil, err } diff --git a/source_runtime_migration_test.go b/source_runtime_migration_test.go new file mode 100644 index 00000000..30a75c66 --- /dev/null +++ b/source_runtime_migration_test.go @@ -0,0 +1,448 @@ +package ghostferry + +import ( + "context" + sqlorig "database/sql" + "database/sql/driver" + "errors" + "io" + "testing" + + sql "github.com/Shopify/ghostferry/sqlwrapper" + "github.com/go-mysql-org/go-mysql/schema" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// These tests need no MySQL: fixture DBs answer every query with canned rows, +// and a swapped-in handle is closed up front so consumers routed to it fail fast. + +var errUnsupportedSourceRuntimeTestOp = errors.New("unsupported source runtime test operation") + +// sourceRuntimeQueryConnector returns the configured rows for every query, prepared or not. +type sourceRuntimeQueryConnector struct { + columns []string + data [][]driver.Value +} + +func (c *sourceRuntimeQueryConnector) Connect(ctx context.Context) (driver.Conn, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + return &sourceRuntimeQueryConn{connector: c}, nil +} + +func (c *sourceRuntimeQueryConnector) Driver() driver.Driver { return sourceRuntimeQueryDriver{} } + +func (c *sourceRuntimeQueryConnector) rows() *sourceRuntimeQueryRows { + return &sourceRuntimeQueryRows{columns: c.columns, data: c.data} +} + +type sourceRuntimeQueryDriver struct{} + +func (sourceRuntimeQueryDriver) Open(string) (driver.Conn, error) { + return nil, errUnsupportedSourceRuntimeTestOp +} + +type sourceRuntimeQueryConn struct { + connector *sourceRuntimeQueryConnector +} + +func (c *sourceRuntimeQueryConn) Prepare(string) (driver.Stmt, error) { + return &sourceRuntimeQueryStmt{connector: c.connector}, nil +} + +func (c *sourceRuntimeQueryConn) Close() error { return nil } + +func (c *sourceRuntimeQueryConn) Begin() (driver.Tx, error) { + return nil, errUnsupportedSourceRuntimeTestOp +} + +func (c *sourceRuntimeQueryConn) QueryContext(ctx context.Context, _ string, _ []driver.NamedValue) (driver.Rows, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + return c.connector.rows(), nil +} + +type sourceRuntimeQueryStmt struct { + connector *sourceRuntimeQueryConnector +} + +func (s *sourceRuntimeQueryStmt) Close() error { return nil } +func (s *sourceRuntimeQueryStmt) NumInput() int { return -1 } + +func (s *sourceRuntimeQueryStmt) Exec([]driver.Value) (driver.Result, error) { + return nil, errUnsupportedSourceRuntimeTestOp +} + +func (s *sourceRuntimeQueryStmt) Query([]driver.Value) (driver.Rows, error) { + return s.connector.rows(), nil +} + +type sourceRuntimeQueryRows struct { + columns []string + data [][]driver.Value + next int +} + +func (r *sourceRuntimeQueryRows) Columns() []string { return r.columns } +func (r *sourceRuntimeQueryRows) Close() error { return nil } + +func (r *sourceRuntimeQueryRows) Next(dest []driver.Value) error { + if r.next >= len(r.data) { + return io.EOF + } + copy(dest, r.data[r.next]) + r.next++ + return nil +} + +func newSourceRuntimeQueryDB(t *testing.T, columns []string, data [][]driver.Value) *sql.DB { + t.Helper() + db := &sql.DB{DB: sqlorig.OpenDB(&sourceRuntimeQueryConnector{columns: columns, data: data})} + t.Cleanup(func() { _ = db.Close() }) + return db +} + +// replaceWithClosedHandle swaps in a fresh handle and closes it, so consumers resolving the source after the swap fail deterministically. +func replaceWithClosedHandle(t *testing.T, rt *SourceRuntime, host string) *sql.DB { + t.Helper() + previous := rt.DB() + replacement, err := rt.Replace(runtimeTestConfig(host), nil) + require.NoError(t, err) + require.NotNil(t, replacement) + require.NotSame(t, previous, replacement, "Replace must install a new handle") + require.Same(t, replacement, rt.DB()) + t.Cleanup(rt.CloseRetired) + t.Cleanup(func() { _ = replacement.Close() }) + require.NoError(t, replacement.Close()) + return replacement +} + +func numericReviewTable() *TableSchema { + columns := []schema.TableColumn{{Name: "id", Type: schema.TYPE_NUMBER}} + return &TableSchema{ + Table: &schema.Table{Schema: "review", Name: "t", Columns: columns}, + PaginationKeyColumn: &columns[0], + PaginationKeyIndex: 0, + } +} + +func newTestCursorConfig(db *sql.DB, rt *SourceRuntime) *CursorConfig { + batchSize := uint64(1) + return &CursorConfig{DB: db, SourceRuntime: rt, BatchSize: &batchSize, ReadRetries: 1} +} + +func newSingleRowDB(t *testing.T) *sql.DB { + t.Helper() + return newSourceRuntimeQueryDB(t, []string{"id"}, [][]driver.Value{{int64(1)}}) +} + +func scanCursor(cursor *Cursor) ([]RowData, error) { + var rows []RowData + err := cursor.Each(func(batch *RowBatch) error { + rows = append(rows, batch.Values()...) + return nil + }) + return rows, err +} + +var singleRow = []RowData{{int64(1)}} + +// --------------------------------------------------------------------------- +// MaxTableSizeSorter + +var sorterColumns = []string{"TABLE_NAME", "TABLE_SCHEMA"} + +func sorterInput() (map[*TableSchema]PaginationKey, *TableSchema, *TableSchema) { + small := &TableSchema{Table: &schema.Table{Schema: "review", Name: "small"}} + large := &TableSchema{Table: &schema.Table{Schema: "review", Name: "large"}} + return map[*TableSchema]PaginationKey{ + small: NewUint64Key(10), + large: NewUint64Key(20), + }, small, large +} + +func assertSortedTables(t *testing.T, sorted []TableMaxPaginationKey, expected ...*TableSchema) { + t.Helper() + require.Len(t, sorted, len(expected)) + for i, table := range expected { + assert.Same(t, table, sorted[i].Table, "position %d", i) + } + for _, entry := range sorted { + switch entry.Table.Name { + case "small": + assert.Equal(t, NewUint64Key(10), entry.MaxPaginationKey) + case "large": + assert.Equal(t, NewUint64Key(20), entry.MaxPaginationKey) + default: + t.Fatalf("unexpected table %s", entry.Table.Name) + } + } +} + +func TestMaxTableSizeSorterUsesSourceRuntime(t *testing.T) { + input, small, large := sorterInput() + + staticDB := newSourceRuntimeQueryDB(t, sorterColumns, [][]driver.Value{ + {"small", "review"}, + {"large", "review"}, + }) + runtimeDB := newSourceRuntimeQueryDB(t, sorterColumns, [][]driver.Value{ + {"large", "review"}, + {"small", "review"}, + }) + rt := NewSourceRuntime(runtimeDB, runtimeTestConfig("runtime")) + + iterator := &DataIterator{DB: staticDB, SourceRuntime: rt} + sorter := MaxTableSizeSorter{DataIterator: iterator} + + sorted, err := sorter.Sort(input) + require.NoError(t, err) + assertSortedTables(t, sorted, large, small) + + // The static DB is still open, so this failing proves the sorter resolves the source per call. + replaceWithClosedHandle(t, rt, "replacement") + _, err = sorter.Sort(input) + require.Error(t, err) +} + +func TestMaxTableSizeSorterStaticFallback(t *testing.T) { + input, small, large := sorterInput() + + cases := map[string]*SourceRuntime{ + "no runtime": nil, + "runtime with nil DB": NewSourceRuntime(nil, nil), + } + for name, rt := range cases { + t.Run(name, func(t *testing.T) { + staticDB := newSourceRuntimeQueryDB(t, sorterColumns, [][]driver.Value{ + {"large", "review"}, + {"small", "review"}, + }) + sorter := MaxTableSizeSorter{DataIterator: &DataIterator{DB: staticDB, SourceRuntime: rt}} + + sorted, err := sorter.Sort(input) + require.NoError(t, err) + assertSortedTables(t, sorted, large, small) + }) + } +} + +// --------------------------------------------------------------------------- +// Cursor + +func TestNewCursorBindsToResolvedDB(t *testing.T) { + table := numericReviewTable() + initialDB := newSingleRowDB(t) + rt := NewSourceRuntime(initialDB, runtimeTestConfig("old")) + cfg := newTestCursorConfig(initialDB, rt) + + before := cfg.NewCursorWithoutRowLock(table, NewUint64Key(0), NewUint64Key(1)) + + replaceWithClosedHandle(t, rt, "new") + + rows, err := scanCursor(before) + require.NoError(t, err) + assert.Equal(t, singleRow, rows, "existing cursor must keep its handle") + + after := cfg.NewCursorWithoutRowLock(table, NewUint64Key(0), NewUint64Key(1)) + rows, err = scanCursor(after) + require.Error(t, err) + assert.Empty(t, rows) +} + +func TestCursorRuntimeFallback(t *testing.T) { + table := numericReviewTable() + + cases := map[string]*SourceRuntime{ + "no runtime": nil, + "runtime with nil DB": NewSourceRuntime(nil, nil), + } + for name, rt := range cases { + t.Run(name, func(t *testing.T) { + cfg := newTestCursorConfig(newSingleRowDB(t), rt) + cursor := cfg.NewCursorWithoutRowLock(table, NewUint64Key(0), NewUint64Key(1)) + + rows, err := scanCursor(cursor) + require.NoError(t, err) + assert.Equal(t, singleRow, rows) + }) + } +} + +// --------------------------------------------------------------------------- +// InlineVerifier + +var ( + fingerprintColumns = []string{"id", "__ghostferry_row_md5"} + oldFingerprint = []byte("old-fingerprint") +) + +func newFingerprintDB(t *testing.T) *sql.DB { + t.Helper() + return newSourceRuntimeQueryDB(t, fingerprintColumns, [][]driver.Value{{int64(1), oldFingerprint}}) +} + +func TestInlineVerifierSourceRuntimeCacheLifetime(t *testing.T) { + table := numericReviewTable() + key := NewUint64Key(1).String() + paginationKeys := []interface{}{uint64(1)} + + oldDB := newFingerprintDB(t) + rt := NewSourceRuntime(oldDB, runtimeTestConfig("old")) + v := &InlineVerifier{SourceDB: oldDB, SourceRuntime: rt, sourceStmtCache: NewStmtCache()} + + // Prime the source statement cache against the old DB. + fingerprints, _, err := v.getFingerprintDataFromSourceDb("review", "t", nil, table, paginationKeys) + require.NoError(t, err) + assert.Equal(t, oldFingerprint, fingerprints[key]) + + inFlightDB, inFlightCache := v.currentSourceDB() + + replaceWithClosedHandle(t, rt, "new") + + // A stale cache would reuse the old DB's statement and incorrectly succeed. + _, _, err = v.getFingerprintDataFromSourceDb("review", "t", nil, table, paginationKeys) + require.Error(t, err) + + fingerprints, _, err = v.getFingerprintDataFromDb(inFlightDB, inFlightCache, "review", "t", nil, table, paginationKeys) + require.NoError(t, err) + assert.Equal(t, oldFingerprint, fingerprints[key]) + + rt.CloseRetired() + _, _, err = v.getFingerprintDataFromDb(inFlightDB, inFlightCache, "review", "t", nil, table, paginationKeys) + require.Error(t, err) +} + +func TestInlineVerifierSourceFallback(t *testing.T) { + table := numericReviewTable() + key := NewUint64Key(1).String() + + cases := map[string]*SourceRuntime{ + "no runtime": nil, + "runtime with nil DB": NewSourceRuntime(nil, nil), + } + for name, rt := range cases { + t.Run(name, func(t *testing.T) { + v := &InlineVerifier{SourceDB: newFingerprintDB(t), SourceRuntime: rt, sourceStmtCache: NewStmtCache()} + + fingerprints, _, err := v.getFingerprintDataFromSourceDb("review", "t", nil, table, []interface{}{uint64(1)}) + require.NoError(t, err) + assert.Equal(t, oldFingerprint, fingerprints[key]) + }) + } +} + +// --------------------------------------------------------------------------- +// Verifier validation + +type sourceSelection struct { + name string + static bool // provide a working static SourceDB + runtime bool // provide a SourceRuntime + runtimeDB bool // ... whose handle is a working DB (false: nil handle) + target bool // provide a working TargetDB + ok bool +} + +var sourceSelections = []sourceSelection{ + {name: "runtime only", runtime: true, runtimeDB: true, target: true, ok: true}, + {name: "static only", static: true, target: true, ok: true}, + {name: "nil-handle runtime with static fallback", static: true, runtime: true, target: true, ok: true}, + {name: "no source", target: true}, + {name: "nil-handle runtime without fallback", runtime: true, target: true}, + {name: "no target", static: true, runtime: true, runtimeDB: true}, +} + +func (s sourceSelection) build(t *testing.T, newDB func(*testing.T) *sql.DB) (static *sql.DB, rt *SourceRuntime, target *sql.DB) { + t.Helper() + if s.static { + static = newDB(t) + } + if s.runtime { + var db *sql.DB + if s.runtimeDB { + db = newDB(t) + } + rt = NewSourceRuntime(db, runtimeTestConfig("runtime")) + } + if s.target { + target = newDB(t) + } + return +} + +func TestIterativeVerifierSourceRuntimeValidation(t *testing.T) { + for _, tc := range sourceSelections { + t.Run(tc.name, func(t *testing.T) { + static, rt, target := tc.build(t, newSingleRowDB) + v := &IterativeVerifier{ + SourceDB: static, + SourceRuntime: rt, + TargetDB: target, + CursorConfig: newTestCursorConfig(static, rt), + BinlogStreamer: &BinlogStreamer{}, + Concurrency: 1, + } + + err := v.Initialize() + if tc.ok { + require.NoError(t, err) + } else { + require.Error(t, err) + } + }) + } +} + +var checksumColumns = []string{"Table", "Checksum"} + +func newChecksumDB(checksum int64) func(*testing.T) *sql.DB { + return func(t *testing.T) *sql.DB { + t.Helper() + return newSourceRuntimeQueryDB(t, checksumColumns, [][]driver.Value{{"review.t", checksum}}) + } +} + +func runChecksumVerifier(t *testing.T, static *sql.DB, rt *SourceRuntime, target *sql.DB) error { + t.Helper() + v := &ChecksumTableVerifier{ + Tables: []*TableSchema{numericReviewTable()}, + SourceDB: static, + SourceRuntime: rt, + TargetDB: target, + } + + if err := v.StartInBackground(); err != nil { + return err + } + v.Wait() + + result, err := v.Result() + require.NoError(t, err) + assert.True(t, result.DataCorrect) + return nil +} + +func TestChecksumVerifierSourceRuntimeValidation(t *testing.T) { + for _, tc := range sourceSelections { + t.Run(tc.name, func(t *testing.T) { + static, rt, target := tc.build(t, newChecksumDB(11)) + err := runChecksumVerifier(t, static, rt, target) + if tc.ok { + require.NoError(t, err) + } else { + require.Error(t, err) + } + }) + } + + t.Run("runtime over static", func(t *testing.T) { + static := newChecksumDB(22)(t) + rt := NewSourceRuntime(newChecksumDB(11)(t), runtimeTestConfig("runtime")) + target := newChecksumDB(11)(t) + require.NoError(t, runChecksumVerifier(t, static, rt, target)) + }) +} diff --git a/verifier.go b/verifier.go index 2db8d392..4047e65c 100644 --- a/verifier.go +++ b/verifier.go @@ -97,6 +97,7 @@ type ChecksumTableVerifier struct { DatabaseRewrites map[string]string TableRewrites map[string]string SourceDB *sql.DB + SourceRuntime *SourceRuntime TargetDB *sql.DB started *AtomicBoolean @@ -108,6 +109,15 @@ type ChecksumTableVerifier struct { wg *sync.WaitGroup } +func (v *ChecksumTableVerifier) currentSourceDB() *sql.DB { + if v.SourceRuntime != nil { + if db := v.SourceRuntime.DB(); db != nil { + return db + } + } + return v.SourceDB +} + func (v *ChecksumTableVerifier) VerifyBeforeCutover() error { // All verification occurs in cutover for this verifier. return nil @@ -151,7 +161,7 @@ func (v *ChecksumTableVerifier) VerifyDuringCutover() (VerificationResult, error go func() { defer wg.Done() query := fmt.Sprintf("CHECKSUM TABLE %s EXTENDED", sourceTable) - sourceRow := v.SourceDB.QueryRow(query) + sourceRow := v.currentSourceDB().QueryRow(query) sourceChecksum, sourceErr = v.fetchChecksumValueFromRow(sourceRow) }() @@ -210,7 +220,7 @@ func (v *ChecksumTableVerifier) fetchChecksumValueFromRow(row *sqlorig.Row) (int } func (v *ChecksumTableVerifier) StartInBackground() error { - if v.SourceDB == nil || v.TargetDB == nil { + if v.currentSourceDB() == nil || v.TargetDB == nil { return errors.New("must specify source and target db") }