schema fingerprint verifier

config.go

@@ -764,6 +764,10 @@ type Config struct {
 	// The ghostferry target user should have SUPER permissions to actually write to the target DB,
 	// if ghostferry is ran with AllowSuperUserOnReadOnly = true and the target DB is set to read_only.
 	AllowSuperUserOnReadOnly bool
+
+	// The interval at which the periodic schema fingerprint verification occurs, in the
+	// format of time.ParseDuration. Default: 60s.
+	PeriodicallyVerifySchemaFingerPrintInterval string
 }
 
 func (c *Config) ValidateConfig() error {
@@ -833,5 +837,14 @@ func (c *Config) ValidateConfig() error {
 		c.CutoverRetryWaitSeconds = 1
 	}
 
+	if len(c.PeriodicallyVerifySchemaFingerPrintInterval) == 0 {
+		c.PeriodicallyVerifySchemaFingerPrintInterval = "60s"
+	} else {
+		_, err := time.ParseDuration(c.PeriodicallyVerifySchemaFingerPrintInterval)
+		if err != nil {
+			return fmt.Errorf("PeriodicallyVerifySchemaFingerPrintInterval invalid")
+		}
+	}
+
 	return nil
 }

docs/source/technicaloverview.rst

@@ -102,6 +102,10 @@ Limitations
   - For tables with foreign key constraints, the constraints should be removed
     before performing the data migration.
 
+- Ghostferry does not support schema changes during the migration or when the migration is
+   interrupted. Currently, Ghostferry checks in the background the schema of the
+   databases being migrated every 1 minute.
+
 Algorithm Correctness
 ---------------------
 

ferry.go

@@ -63,6 +63,8 @@ type Ferry struct {
 	DataIterator *DataIterator
 	BatchWriter  *BatchWriter
 
+	SchemaFingerPrintVerifier *SchemaFingerPrintVerifier
+
 	StateTracker                       *StateTracker
 	ErrorHandler                       ErrorHandler
 	Throttler                          Throttler
@@ -245,6 +247,29 @@ func (f *Ferry) NewInlineVerifier() *InlineVerifier {
 	}
 }
 
+func (f *Ferry) NewSchemaFingerPrintVerifier() (*SchemaFingerPrintVerifier, error) {
+	fingerPrint := map[string]string{}
+	if f.StateToResumeFrom != nil && f.StateToResumeFrom.SchemaFingerPrint != nil {
+		fingerPrint = f.StateToResumeFrom.SchemaFingerPrint
+	}
+	periodicallyVerifyInterval, err := time.ParseDuration(f.Config.PeriodicallyVerifySchemaFingerPrintInterval)
+	if err != nil {
+		return nil, fmt.Errorf("invalid PeriodicallyVerifySchemaFingerPrintInterval: %v. this error should have been caught via .Validate()", err)
+	}
+
+	return &SchemaFingerPrintVerifier{
+		SourceDB:                   f.SourceDB,
+		TableRewrites:              f.Config.TableRewrites,
+		TableSchemaCache:           f.Tables,
+		ErrorHandler:               f.ErrorHandler,
+		PeriodicallyVerifyInterval: periodicallyVerifyInterval,
+
+		FingerPrints: fingerPrint,
+
+		logger: logrus.WithField("tag", "schema_fingerprint_verifier"),
+	}, nil
+}
+
 func (f *Ferry) NewInlineVerifierWithoutStateTracker() *InlineVerifier {
 	v := f.NewInlineVerifier()
 	v.StateTracker = nil
@@ -489,6 +514,11 @@ func (f *Ferry) Initialize() (err error) {
 		}
 	}
 
+	f.SchemaFingerPrintVerifier, err = f.NewSchemaFingerPrintVerifier()
+	if err != nil {
+		return err
+	}
+
 	// The iterative verifier needs the binlog streamer so this has to be first.
 	// Eventually this can be moved below the verifier initialization.
 	f.BinlogStreamer = f.NewSourceBinlogStreamer()
@@ -686,6 +716,13 @@ func (f *Ferry) Run() {
 		}()
 	}
 
+	schemaFingerVerifierPrintWg := &sync.WaitGroup{}
+	schemaFingerVerifierPrintWg.Add(1)
+	go func() {
+		defer schemaFingerVerifierPrintWg.Done()
+		f.SchemaFingerPrintVerifier.PeriodicallyVerifySchemaFingerprints(ctx)
+	}()
+
 	inlineVerifierWg := &sync.WaitGroup{}
 	inlineVerifierContext, stopInlineVerifier := context.WithCancel(ctx)
 	if f.inlineVerifier != nil {
@@ -787,6 +824,8 @@ func (f *Ferry) Run() {
 	f.DoneTime = time.Now()
 
 	shutdown()
+
+	schemaFingerVerifierPrintWg.Wait()
 	supportingServicesWg.Wait()
 
 	if f.Config.ProgressCallback.URI != "" {
@@ -908,12 +947,14 @@ func (f *Ferry) SerializeStateToJSON() (string, error) {
 		err := errors.New("no valid StateTracker")
 		return "", err
 	}
-	var binlogVerifyStore *BinlogVerifyStore = nil
+	var (
+		binlogVerifyStore *BinlogVerifyStore = nil
+	)
 	if f.inlineVerifier != nil {
 		binlogVerifyStore = f.inlineVerifier.reverifyStore
 	}
 
-	serializedState := f.StateTracker.Serialize(f.Tables, binlogVerifyStore)
+	serializedState := f.StateTracker.Serialize(f.Tables, binlogVerifyStore, f.SchemaFingerPrintVerifier.FingerPrints)
 
 	if f.Config.DoNotIncludeSchemaCacheInStateDump {
 		serializedState.LastKnownTableSchemaCache = nil
@@ -945,7 +986,7 @@ func (f *Ferry) Progress() *Progress {
 	}
 
 	// Table Progress
-	serializedState := f.StateTracker.Serialize(nil, nil)
+	serializedState := f.StateTracker.Serialize(nil, nil, nil)
 	// Note the below will not necessarily be synchronized with serializedState.
 	// This is fine as we don't need to be super precise with performance data.
 	rowStatsWrittenPerTable := f.StateTracker.RowStatsWrittenPerTable()