Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

ghostferry

package module

Go to main page

Versions in this module

v1

v1.3.0

Apr 8, 2026

Changes in this version

+ func SetLogBackend(backend LogBackendType)

+ func SetLogJSONFormatter()

+ func SetLogLevel(level LogLevel)

+ func SetLogOutput(w io.Writer)

type BinaryKey

+ Column string

+ Value []byte

+ func NewBinaryKeyWithColumn(column string, value []byte) BinaryKey

+ func (k BinaryKey) ColumnName() string

type Config

+ LogBackend string

+ LogLevel string

+ type Fields map[string]any

+ type LogBackendType string

+ const LogBackendLogrus

+ const LogBackendZerolog

+ func GetLogBackend() LogBackendType

+ type LogLevel int

+ const LogLevelDebug

+ const LogLevelError

+ const LogLevelInfo

+ const LogLevelWarn

+ func ParseLogLevel(s string) (LogLevel, bool)

+ type Logger interface

+ Debug func(args ...any)

+ Debugf func(format string, args ...any)

+ Error func(args ...any)

+ Errorf func(format string, args ...any)

+ Info func(args ...any)

+ Infof func(format string, args ...any)

+ Panicf func(format string, args ...any)

+ Warn func(args ...any)

+ Warnf func(format string, args ...any)

+ WithError func(err error) Logger

+ WithField func(key string, value any) Logger

+ WithFields func(fields Fields) Logger

+ func LogWithError(err error) Logger

+ func LogWithField(key string, value any) Logger

+ func LogWithFields(fields Fields) Logger

+ func NewDefaultLogger() Logger

type PaginationKey

+ ColumnName func() string

type Uint64Key

+ Column string

+ Value uint64

+ func NewUint64KeyWithColumn(column string, value uint64) Uint64Key

+ func (k Uint64Key) ColumnName() string

v1.2.0

Feb 6, 2026

Changes in this version

+ const CompressionSnappy

+ const DefaultMarginalia

+ const DefaultNet

+ const MismatchColumnMissingOnSource

+ const MismatchColumnMissingOnTarget

+ const MismatchColumnValueDifference

+ const MismatchRowChecksumDifference

+ const MismatchRowMissingOnSource

+ const MismatchRowMissingOnTarget

+ const MySQLNumParamsLimit

+ const StateCopying

+ const StateCutover

+ const StateDone

+ const StateStarting

+ const StateVerifyBeforeCutover

+ const StateWaitingForCutover

+ const TableActionCompleted

+ const TableActionCopying

+ const TableActionWaiting

+ const VerifierTypeChecksumTable

+ const VerifierTypeInline

+ const VerifierTypeIterative

+ const VerifierTypeNoVerification

+ var VersionString string = "?.?.?+??????????????+???????"

+ var WebUiBasedir string = ""

+ func CheckDbIsAReplica(db *sql.DB) (bool, error)

+ func ConvertTableColumnsToStrings(columns []schema.TableColumn) []string

+ func DefaultBuildSelect(columns []string, table *TableSchema, lastPaginationKey PaginationKey, ...) squirrel.SelectBuilder

+ func GetMd5HashesSql(schema, table, paginationKeyColumn string, columns []schema.TableColumn, ...) (string, []interface{}, error)

+ func Int64Value(value interface{}) (int64, bool)

+ func MarshalPaginationKey(k PaginationKey) ([]byte, error)

+ func MaskedDSN(c *mysql.Config) string

+ func NewMysqlPosition(file string, position uint32, err error, db *sql.DB) (mysql.Position, error)

+ func NonBinaryCollationError(schema, table, paginationKey, collation string) error

+ func NonExistingPaginationKeyColumnError(schema, table, paginationKey string) error

+ func NonExistingPaginationKeyError(schema, table string) error

+ func NonNumericPaginationKeyError(schema, table, paginationKey string) error

+ func QuoteField(field string) string

+ func QuoteFields(fields []string) (out []string)

+ func QuotedTableName(table *TableSchema) string

+ func QuotedTableNameFromString(database, table string) string

+ func ScanByteRow(rows *sqlorig.Rows, columnCount int) ([][]byte, error)

+ func ShowMasterStatusBinlogPosition(db *sql.DB) (mysql.Position, error)

+ func TargetToSourceRewrites(databaseRewrites map[string]string) (map[string]string, error)

+ func Uint64Value(value interface{}) (uint64, bool)

+ func WaitForThrottle(t Throttler)

+ func WithRetries(maxRetries int, sleep time.Duration, logger *logrus.Entry, verb string, ...) (err error)

+ func WithRetriesContext(ctx context.Context, maxRetries int, sleep time.Duration, logger *logrus.Entry, ...) (err error)

+ type AtomicBoolean int32

+ func (a *AtomicBoolean) Get() bool

+ func (a *AtomicBoolean) Set(b bool)

+ type BatchWriter struct

+ DB *sql.DB

+ DatabaseRewrites map[string]string

+ EnforceInlineVerification bool

+ InlineVerifier *InlineVerifier

+ StateTracker *StateTracker

+ TableRewrites map[string]string

+ WriteRetries int

+ func (w *BatchWriter) Initialize()

+ func (w *BatchWriter) WriteRowBatch(batch *RowBatch) error

+ type BatchWriterVerificationFailed struct

+ func (e BatchWriterVerificationFailed) Error() string

+ type BinaryKey []byte

+ func NewBinaryKey(value []byte) BinaryKey

+ func (k BinaryKey) Compare(other PaginationKey) int

+ func (k BinaryKey) IsMax() bool

+ func (k BinaryKey) MarshalJSON() ([]byte, error)

+ func (k BinaryKey) NumericPosition() float64

+ func (k BinaryKey) SQLValue() interface{}

+ func (k BinaryKey) String() string

+ type BinlogDeleteEvent struct

+ func (e *BinlogDeleteEvent) AsSQLString(schemaName, tableName string) (string, error)

+ func (e *BinlogDeleteEvent) NewValues() RowData

+ func (e *BinlogDeleteEvent) OldValues() RowData

+ func (e *BinlogDeleteEvent) PaginationKey() (string, error)

+ type BinlogEventState struct

+ type BinlogInsertEvent struct

+ func (e *BinlogInsertEvent) AsSQLString(schemaName, tableName string) (string, error)

+ func (e *BinlogInsertEvent) NewValues() RowData

+ func (e *BinlogInsertEvent) OldValues() RowData

+ func (e *BinlogInsertEvent) PaginationKey() (string, error)

+ type BinlogStreamer struct

+ DB *sql.DB

+ DBConfig *DatabaseConfig

+ DatabaseRewrites map[string]string

+ ErrorHandler ErrorHandler

+ Filter CopyFilter

+ LogTag string

+ MyServerId uint32

+ TableRewrites map[string]string

+ TableSchema TableSchemaCache

+ func (s *BinlogStreamer) AddBinlogEventHandler(evType replication.EventType, ...) error

+ func (s *BinlogStreamer) AddEventListener(listener func([]DMLEvent) error)

+ func (s *BinlogStreamer) ConnectBinlogStreamerToMysql() (mysql.Position, error)

+ func (s *BinlogStreamer) ConnectBinlogStreamerToMysqlFrom(startFromBinlogPosition mysql.Position) (mysql.Position, error)

+ func (s *BinlogStreamer) FlushAndStop()

+ func (s *BinlogStreamer) GetLastStreamedBinlogPosition() mysql.Position

+ func (s *BinlogStreamer) IsAlmostCaughtUp() bool

+ func (s *BinlogStreamer) Run()

+ type BinlogUpdateEvent struct

+ func (e *BinlogUpdateEvent) AsSQLString(schemaName, tableName string) (string, error)

+ func (e *BinlogUpdateEvent) NewValues() RowData

+ func (e *BinlogUpdateEvent) OldValues() RowData

+ func (e *BinlogUpdateEvent) PaginationKey() (string, error)

+ type BinlogVerifyBatch struct

+ PaginationKeys []interface{}

+ SchemaName string

+ TableName string

+ type BinlogVerifySerializedStore map[string]map[string]map[string]int

+ func (s BinlogVerifySerializedStore) Copy() BinlogVerifySerializedStore

+ func (s BinlogVerifySerializedStore) EntriesCount() uint64

+ func (s BinlogVerifySerializedStore) RowCount() uint64

+ type BinlogVerifyStore struct

+ EmitLogPerRowsAdded uint64

+ func NewBinlogVerifyStore() *BinlogVerifyStore

+ func NewBinlogVerifyStoreFromSerialized(serialized BinlogVerifySerializedStore) *BinlogVerifyStore

+ func (s *BinlogVerifyStore) Add(table *TableSchema, paginationKey string)

+ func (s *BinlogVerifyStore) Batches(batchsize int) []BinlogVerifyBatch

+ func (s *BinlogVerifyStore) CurrentEntriesCount() uint64

+ func (s *BinlogVerifyStore) CurrentRowCount() uint64

+ func (s *BinlogVerifyStore) RemoveVerifiedBatch(batch BinlogVerifyBatch)

+ func (s *BinlogVerifyStore) Serialize() BinlogVerifySerializedStore

+ type BinlogWriter struct

+ BatchSize int

+ DB *sql.DB

+ DatabaseRewrites map[string]string

+ ErrorHandler ErrorHandler

+ StateTracker *StateTracker

+ TableRewrites map[string]string

+ Throttler Throttler

+ WriteRetries int

+ func (b *BinlogWriter) BufferBinlogEvents(events []DMLEvent) error

+ func (b *BinlogWriter) Run()

+ func (b *BinlogWriter) Stop()

+ type CascadingPaginationColumnConfig struct

+ FallbackColumn string

+ PerTable map[string]map[string]string

+ func (c *CascadingPaginationColumnConfig) FallbackPaginationColumnName() (string, bool)

+ func (c *CascadingPaginationColumnConfig) PaginationColumnFor(schemaName, tableName string) (string, bool)

+ type ChecksumTableVerifier struct

+ DatabaseRewrites map[string]string

+ SourceDB *sql.DB

+ TableRewrites map[string]string

+ Tables []*TableSchema

+ TargetDB *sql.DB

+ func (v *ChecksumTableVerifier) Message() string

+ func (v *ChecksumTableVerifier) Result() (VerificationResultAndStatus, error)

+ func (v *ChecksumTableVerifier) StartInBackground() error

+ func (v *ChecksumTableVerifier) VerifyBeforeCutover() error

+ func (v *ChecksumTableVerifier) VerifyDuringCutover() (VerificationResult, error)

+ func (v *ChecksumTableVerifier) Wait()

+ type ColumnCompressionConfig map[string]map[string]map[string]string

+ func (c ColumnCompressionConfig) CompressedColumnsFor(schemaName, tableName string) map[string]string

+ type ColumnIgnoreConfig map[string]map[string]map[string]struct

+ func (c ColumnIgnoreConfig) IgnoredColumnsFor(schemaName, tableName string) map[string]struct{}

+ type CompressionVerifier struct

+ TableSchemaCache TableSchemaCache

+ func NewCompressionVerifier(tableColumnCompressions TableColumnCompressionConfig, ...) (*CompressionVerifier, error)

+ func (c *CompressionVerifier) Decompress(table, column, algorithm string, compressed []byte) ([]byte, error)

+ func (c *CompressionVerifier) GetCompressedHashes(db *sql.DB, schemaName, tableName, paginationKeyColumn string, ...) (map[string][]byte, error)

+ func (c *CompressionVerifier) HashRow(decompressedRowData [][]byte) ([]byte, error)

+ func (c *CompressionVerifier) IsCompressedTable(table string) bool

+ type Config struct

+ AllowSuperUserOnReadOnly bool

+ AutomaticCutover bool

+ BinlogEventBatchSize int

+ CascadingPaginationColumnConfig *CascadingPaginationColumnConfig

+ CompressedColumnsForVerification ColumnCompressionConfig

+ ControlServerConfig *ControlServerConfig

+ ControlServerCustomScripts map[string][]string

+ CopyFilter CopyFilter

+ CutoverLock HTTPCallback

+ CutoverRetryWaitSeconds int

+ CutoverUnlock HTTPCallback

+ DBReadRetries int

+ DBWriteRetries int

+ DataIterationBatchSize uint64

+ DataIterationBatchSizePerTableOverride *DataIterationBatchSizePerTableOverride

+ DataIterationConcurrency int

+ DatabaseRewrites map[string]string

+ DoNotIncludeSchemaCacheInStateDump bool

+ DumpStateOnSignal bool

+ DumpStateToStdoutOnError bool

+ EnablePProf bool

+ EnableRowBatchSize bool

+ ErrorCallback HTTPCallback

+ ForceIndexForVerification ForceIndexConfig

+ IgnoredColumnsForVerification ColumnIgnoreConfig

+ InlineVerifierConfig InlineVerifierConfig

+ IterativeVerifierConfig IterativeVerifierConfig

+ MaxCutoverRetries int

+ MyServerId uint32

+ ProgressCallback HTTPCallback

+ ProgressReportFrequency int

+ ServerBindAddr string

+ SkipBinlogRowImageCheck bool

+ SkipForeignKeyConstraintsCheck bool

+ SkipTargetVerification bool

+ Source *DatabaseConfig

+ StateCallback HTTPCallback

+ StateReportFrequency int

+ StateToResumeFrom *SerializableState

+ TableFilter TableFilter

+ TableRewrites map[string]string

+ Target *DatabaseConfig

+ UpdatableConfig UpdatableConfig

+ VerifierType string

+ WebBasedir string

+ func (c *Config) Update(updatedConfig UpdatableConfig)

+ func (c *Config) ValidateConfig() error

+ type ControlServer struct

+ Config *ControlServerConfig

+ F *Ferry

+ Verifier Verifier

+ func (this *ControlServer) HandleConfigGet(w http.ResponseWriter, r *http.Request)

+ func (this *ControlServer) HandleConfigPost(w http.ResponseWriter, r *http.Request)

+ func (this *ControlServer) HandleCutover(w http.ResponseWriter, r *http.Request)

+ func (this *ControlServer) HandleIndex(w http.ResponseWriter, r *http.Request)

+ func (this *ControlServer) HandlePause(w http.ResponseWriter, r *http.Request)

+ func (this *ControlServer) HandleScript(w http.ResponseWriter, r *http.Request)

+ func (this *ControlServer) HandleStatus(w http.ResponseWriter, r *http.Request)

+ func (this *ControlServer) HandleStop(w http.ResponseWriter, r *http.Request)

+ func (this *ControlServer) HandleUnpause(w http.ResponseWriter, r *http.Request)

+ func (this *ControlServer) HandleVerify(w http.ResponseWriter, r *http.Request)

+ func (this *ControlServer) Initialize() (err error)

+ func (this *ControlServer) Run()

+ func (this *ControlServer) ServeHTTP(w http.ResponseWriter, r *http.Request)

+ func (this *ControlServer) Wait()

+ type ControlServerConfig struct

+ CustomScripts map[string][]string

+ Enabled bool

+ ServerBindAddr string

+ WebBasedir string

+ func (c *ControlServerConfig) Validate() error

+ type ControlServerStatus struct

+ AllDatabaseNames []string

+ AllTableNames []string

+ AutomaticCutover bool

+ BinlogStreamerLag time.Duration

+ BinlogStreamerStopRequested bool

+ CompletedTableCount int

+ CurrentTime time.Time

+ CustomScriptStatuses map[string]CustomScriptStatus

+ ETA time.Duration

+ GhostferryVersion string

+ OverallState string

+ SourceHostPort string

+ StartTime time.Time

+ TableStatuses []*ControlServerTableStatus

+ TargetHostPort string

+ TimeTaken time.Duration

+ TotalTableCount int

+ VerificationDone bool

+ VerificationErr error

+ VerificationResult VerificationResult

+ VerificationStarted bool

+ VerifierAvailable bool

+ VerifierSupport bool

+ type ControlServerTableStatus struct

+ BatchSize uint64

+ LastSuccessfulPaginationKey uint64

+ PaginationKeyName string

+ Status string

+ TableName string

+ TargetPaginationKey uint64

+ type CopyFilter interface

+ ApplicableEvent func(DMLEvent) (bool, error)

+ BuildSelect func([]string, *TableSchema, PaginationKey, uint64) (sq.SelectBuilder, error)

+ type CountMetric struct

+ Value int64

+ type Cursor struct

+ MaxPaginationKey PaginationKey

+ RowLock bool

+ Table *TableSchema

+ func (c *Cursor) Each(f func(*RowBatch) error) error

+ func (c *Cursor) Fetch(db SqlPreparer) (batch *RowBatch, paginationKeypos PaginationKey, err error)

+ type CursorConfig struct

+ BatchSize *uint64

+ BatchSizePerTableOverride *DataIterationBatchSizePerTableOverride

+ BuildSelect func([]string, *TableSchema, PaginationKey, uint64) (squirrel.SelectBuilder, error)

+ ColumnsToSelect []string

+ DB *sql.DB

+ ReadRetries int

+ Throttler Throttler

+ func (c *CursorConfig) NewCursor(table *TableSchema, startPaginationKey, maxPaginationKey PaginationKey) *Cursor

+ func (c *CursorConfig) NewCursorWithoutRowLock(table *TableSchema, startPaginationKey, maxPaginationKey PaginationKey) *Cursor

+ func (c CursorConfig) GetBatchSize(schemaName string, tableName string) uint64

+ type CustomScriptStatus struct

+ ExitCode int

+ Logs string

+ Name string

+ Running bool

+ Status string

+ type DMLEvent interface

+ Annotation func() (string, error)

+ AsSQLString func(string, string) (string, error)

+ BinlogPosition func() mysql.Position

+ Database func() string

+ NewValues func() RowData

+ OldValues func() RowData

+ PaginationKey func() (string, error)

+ ResumableBinlogPosition func() mysql.Position

+ Table func() string

+ TableSchema func() *TableSchema

+ Timestamp func() time.Time

+ func NewBinlogDMLEvents(table *TableSchema, ev *replication.BinlogEvent, ...) ([]DMLEvent, error)

+ func NewBinlogDeleteEvents(eventBase *DMLEventBase, rowsEvent *replication.RowsEvent) ([]DMLEvent, error)

+ func NewBinlogInsertEvents(eventBase *DMLEventBase, rowsEvent *replication.RowsEvent) ([]DMLEvent, error)

+ func NewBinlogUpdateEvents(eventBase *DMLEventBase, rowsEvent *replication.RowsEvent) ([]DMLEvent, error)

+ type DMLEventBase struct

+ func NewDMLEventBase(table *TableSchema, pos, resumablePos mysql.Position, query []byte, ...) *DMLEventBase

+ func (e *DMLEventBase) Annotation() (string, error)

+ func (e *DMLEventBase) BinlogPosition() mysql.Position

+ func (e *DMLEventBase) Database() string

+ func (e *DMLEventBase) ResumableBinlogPosition() mysql.Position

+ func (e *DMLEventBase) Table() string

+ func (e *DMLEventBase) TableSchema() *TableSchema

+ func (e *DMLEventBase) Timestamp() time.Time

+ type DataIterationBatchSizePerTableOverride struct

+ ControlPoints map[int]uint64

+ MaxRowSize int

+ MinRowSize int

+ TableOverride map[string]map[string]uint64

+ func (d *DataIterationBatchSizePerTableOverride) CalculateBatchSize(rowSize int) int

+ func (d *DataIterationBatchSizePerTableOverride) UpdateBatchSizes(db *sql.DB, tables TableSchemaCache) error

+ func (d *DataIterationBatchSizePerTableOverride) Validate() error

+ type DataIterator struct

+ Concurrency int

+ CursorConfig *CursorConfig

+ DB *sql.DB

+ ErrorHandler ErrorHandler

+ SelectFingerprint bool

+ StateTracker *StateTracker

+ TableSorter DataIteratorSorter

+ TargetPaginationKeys *sync.Map

+ func (d *DataIterator) AddBatchListener(listener func(*RowBatch) error)

+ func (d *DataIterator) AddDoneListener(listener func() error)

+ func (d *DataIterator) Run(tables []*TableSchema)

+ type DataIteratorSorter interface

+ Sort func(unorderedTables map[*TableSchema]PaginationKey) ([]TableMaxPaginationKey, error)

+ type DatabaseConfig struct

+ Collation string

+ Host string

+ Marginalia string

+ Net string

+ Params map[string]string

+ Pass string

+ Port uint16

+ ReadTimeout uint64

+ TLS *TLSConfig

+ User string

+ WriteTimeout uint64

+ func (c *DatabaseConfig) MySQLConfig() (*mysql.Config, error)

+ func (c *DatabaseConfig) SqlDB(logger *logrus.Entry) (*sql.DB, error)

+ func (c *DatabaseConfig) Validate() error

+ type ErrorHandler interface

+ Fatal func(from string, err error)

+ ReportError func(from string, err error)

+ type Ferry struct

+ BatchWriter *BatchWriter

+ BinlogStreamer *BinlogStreamer

+ BinlogWriter *BinlogWriter

+ ControlServer *ControlServer

+ DataIterator *DataIterator

+ DoneTime time.Time

+ ErrorHandler ErrorHandler

+ OverallState atomic.Value

+ SourceDB *sql.DB

+ StartTime time.Time

+ StateTracker *StateTracker

+ Tables TableSchemaCache

+ TargetDB *sql.DB

+ TargetVerifier *TargetVerifier

+ Throttler Throttler

+ Verifier Verifier

+ WaitUntilReplicaIsCaughtUpToMaster *WaitUntilReplicaIsCaughtUpToMaster

+ func (f *Ferry) EndCutover(cutoverStart time.Time)

+ func (f *Ferry) FlushBinlogAndStopStreaming()

+ func (f *Ferry) Initialize() (err error)

+ func (f *Ferry) NewBatchWriter() *BatchWriter

+ func (f *Ferry) NewBatchWriterWithoutStateTracker() *BatchWriter

+ func (f *Ferry) NewBinlogWriter() *BinlogWriter

+ func (f *Ferry) NewBinlogWriterWithoutStateTracker() *BinlogWriter

+ func (f *Ferry) NewChecksumTableVerifier() *ChecksumTableVerifier

+ func (f *Ferry) NewControlServer() (*ControlServer, error)

+ func (f *Ferry) NewDataIterator() *DataIterator

+ func (f *Ferry) NewDataIteratorWithoutStateTracker() *DataIterator

+ func (f *Ferry) NewInlineVerifier() *InlineVerifier

+ func (f *Ferry) NewInlineVerifierWithoutStateTracker() *InlineVerifier

+ func (f *Ferry) NewIterativeVerifier() (*IterativeVerifier, error)

+ func (f *Ferry) NewSourceBinlogStreamer() *BinlogStreamer

+ func (f *Ferry) NewTargetBinlogStreamer() (*BinlogStreamer, error)

+ func (f *Ferry) Progress() *Progress

+ func (f *Ferry) ReportProgress()

+ func (f *Ferry) ReportState()

+ func (f *Ferry) Run()

+ func (f *Ferry) RunStandaloneDataCopy(tables []*TableSchema) error

+ func (f *Ferry) SerializeStateToJSON() (string, error)

+ func (f *Ferry) Start() error

+ func (f *Ferry) StartCutover() time.Time

+ func (f *Ferry) StopTargetVerifier()

+ func (f *Ferry) WaitUntilBinlogStreamerCatchesUp()

+ func (f *Ferry) WaitUntilRowCopyIsComplete()

+ type ForceIndexConfig map[string]map[string]string

+ func (c ForceIndexConfig) IndexFor(schemaName, tableName string) string

+ type GaugeMetric struct

+ Value float64

+ type HTTPCallback struct

+ Payload string

+ URI string

+ func (h HTTPCallback) Post(client *http.Client) error

+ type IncompleteVerificationError struct

+ func (e IncompleteVerificationError) Error() string

+ type InlineVerifier struct

+ BatchSize int

+ DatabaseRewrites map[string]string

+ ErrorHandler ErrorHandler

+ MaxExpectedDowntime time.Duration

+ SourceDB *sql.DB

+ StateTracker *StateTracker

+ TableRewrites map[string]string

+ TableSchemaCache TableSchemaCache

+ TargetDB *sql.DB

+ VerifyBinlogEventsInterval time.Duration

+ func (v *InlineVerifier) CheckFingerprintInline(tx *sql.Tx, targetSchema, targetTable string, sourceBatch *RowBatch, ...) ([]InlineVerifierMismatches, error)

+ func (v *InlineVerifier) Message() string

+ func (v *InlineVerifier) PeriodicallyVerifyBinlogEvents(ctx context.Context)

+ func (v *InlineVerifier) Result() (VerificationResultAndStatus, error)

+ func (v *InlineVerifier) StartInBackground() error

+ func (v *InlineVerifier) VerifyBeforeCutover() error

+ func (v *InlineVerifier) VerifyDuringCutover() (VerificationResult, error)

+ func (v *InlineVerifier) Wait()

+ type InlineVerifierConfig struct

+ MaxExpectedDowntime string

+ VerifyBinlogEventsInterval string

+ func (c *InlineVerifierConfig) Validate() error

+ type InlineVerifierMismatches struct

+ MismatchColumn string

+ MismatchType mismatchType

+ Pk string

+ SourceChecksum string

+ TargetChecksum string

+ type IterativeVerifier struct

+ BinlogStreamer *BinlogStreamer

+ CompressionVerifier *CompressionVerifier

+ Concurrency int

+ CursorConfig *CursorConfig

+ DatabaseRewrites map[string]string

+ IgnoredColumns map[string]map[string]struct{}

+ IgnoredTables []string

+ MaxExpectedDowntime time.Duration

+ SourceDB *sql.DB

+ TableRewrites map[string]string

+ TableSchemaCache TableSchemaCache

+ Tables []*TableSchema

+ TargetDB *sql.DB

+ func (v *IterativeVerifier) GetHashes(db *sql.DB, schemaName, tableName, paginationKeyColumn string, ...) (map[string][]byte, error)

+ func (v *IterativeVerifier) Initialize() error

+ func (v *IterativeVerifier) Message() string

+ func (v *IterativeVerifier) Result() (VerificationResultAndStatus, error)

+ func (v *IterativeVerifier) SanityCheckParameters() error

+ func (v *IterativeVerifier) StartInBackground() error

+ func (v *IterativeVerifier) VerifyBeforeCutover() error

+ func (v *IterativeVerifier) VerifyDuringCutover() (VerificationResult, error)

+ func (v *IterativeVerifier) VerifyOnce() (VerificationResult, error)

+ func (v *IterativeVerifier) Wait()

+ type IterativeVerifierConfig struct

+ Concurrency int

+ IgnoredColumns map[string][]string

+ IgnoredTables []string

+ MaxExpectedDowntime string

+ TableColumnCompression TableColumnCompressionConfig

+ func (c *IterativeVerifierConfig) Validate() error

+ type LagThrottler struct

+ DB *sql.DB

+ func NewLagThrottler(config *LagThrottlerConfig) (*LagThrottler, error)

+ func (t *LagThrottler) Run(ctx context.Context) error

+ func (t *LagThrottler) Throttled() bool

+ type LagThrottlerConfig struct

+ Connection *DatabaseConfig

+ MaxLag int

+ Query string

+ UpdateInterval string

+ type MaxPaginationKeySorter struct

+ func (s *MaxPaginationKeySorter) Sort(unorderedTables map[*TableSchema]PaginationKey) ([]TableMaxPaginationKey, error)

+ type MaxTableSizeSorter struct

+ DataIterator *DataIterator

+ func (s *MaxTableSizeSorter) Sort(unorderedTables map[*TableSchema]PaginationKey) ([]TableMaxPaginationKey, error)

+ type MetricBase struct

+ Key string

+ SampleRate float64

+ Tags []MetricTag

+ type MetricTag struct

+ Name string

+ Value string

+ type Metrics struct

+ DefaultTags []MetricTag

+ Prefix string

+ Sink chan interface{}

+ func SetGlobalMetrics(prefix string, sink chan interface{}) *Metrics

+ func (m *Metrics) AddConsumer()

+ func (m *Metrics) Count(key string, value int64, tags []MetricTag, sampleRate float64)

+ func (m *Metrics) DoneConsumer()

+ func (m *Metrics) Gauge(key string, value float64, tags []MetricTag, sampleRate float64)

+ func (m *Metrics) Measure(key string, tags []MetricTag, sampleRate float64, f func())

+ func (m *Metrics) StopAndFlush()

+ func (m *Metrics) Timer(key string, duration time.Duration, tags []MetricTag, sampleRate float64)

+ type PaginationKey interface

+ Compare func(other PaginationKey) int

+ IsMax func() bool

+ MarshalJSON func() ([]byte, error)

+ NumericPosition func() float64

+ SQLValue func() interface{}

+ String func() string

+ func MaxPaginationKey(column *schema.TableColumn) PaginationKey

+ func MinPaginationKey(column *schema.TableColumn) PaginationKey

+ func NewPaginationKeyFromRow(rowData RowData, index int, column *schema.TableColumn) (PaginationKey, error)

+ func UnmarshalPaginationKey(data []byte) (PaginationKey, error)

+ type PaginationKeyPositionLog struct

+ At time.Time

+ Position float64

+ type PanicErrorHandler struct

+ DumpStateToStdoutOnError bool

+ ErrorCallback HTTPCallback

+ Ferry *Ferry

+ func (this *PanicErrorHandler) Fatal(from string, err error)

+ func (this *PanicErrorHandler) ReportError(from string, err error)

+ type PauserThrottler struct

+ func (t *PauserThrottler) Run(ctx context.Context) error

+ func (t *PauserThrottler) SetPaused(paused bool)

+ func (t *PauserThrottler) Throttled() bool

+ type Progress struct

+ ActiveDataIterators int

+ BinlogStreamerLag float64

+ BinlogWriterLag float64

+ CurrentState string

+ CustomPayload string

+ ETA float64

+ FinalBinlogPos mysql.Position

+ LastSuccessfulBinlogPos mysql.Position

+ PaginationKeysPerSecond uint64

+ Tables map[string]TableProgress

+ TargetBinlogStreamerLag float64

+ Throttled bool

+ TimeTaken float64

+ VerifierMessage string

+ VerifierType string

+ type ReplicatedMasterPositionFetcher interface

+ Current func(*sql.DB) (mysql.Position, error)

+ type ReplicatedMasterPositionViaCustomQuery struct

+ Query string

+ func (r ReplicatedMasterPositionViaCustomQuery) Current(replicaDB *sql.DB) (mysql.Position, error)

+ type ReverifyBatch struct

+ PaginationKeys []interface{}

+ Table TableIdentifier

+ type ReverifyEntry struct

+ PaginationKey string

+ Table *TableSchema

+ type ReverifyStore struct

+ BatchStore []ReverifyBatch

+ EmitLogPerRowCount uint64

+ MapStore map[TableIdentifier]map[string]struct{}

+ RowCount uint64

+ func NewReverifyStore() *ReverifyStore

+ func (r *ReverifyStore) Add(entry ReverifyEntry)

+ func (r *ReverifyStore) FlushAndBatchByTable(batchsize int) []ReverifyBatch

+ type RowBatch struct

+ func NewRowBatch(table *TableSchema, values []RowData, paginationKeyIndex int) *RowBatch

+ func (e *RowBatch) AsSQLQuery(schemaName, tableName string) (string, []interface{}, error)

+ func (e *RowBatch) EstimateByteSize() uint64

+ func (e *RowBatch) Fingerprints() map[string][]byte

+ func (e *RowBatch) PaginationKeyIndex() int

+ func (e *RowBatch) Size() int

+ func (e *RowBatch) TableSchema() *TableSchema

+ func (e *RowBatch) Values() []RowData

+ func (e *RowBatch) ValuesContainPaginationKey() bool

+ type RowData []interface

+ func (r RowData) GetUint64(colIdx int) (uint64, error)

+ func ScanGenericRow(rows *sqlorig.Rows, columnCount int) (RowData, error)

+ type RowStats struct

+ NumBytes uint64

+ NumRows uint64

+ type SerializableState struct

+ BinlogVerifyStore BinlogVerifySerializedStore

+ CompletedTables map[string]bool

+ GhostferryVersion string

+ LastKnownTableSchemaCache TableSchemaCache

+ LastStoredBinlogPositionForInlineVerifier mysql.Position

+ LastStoredBinlogPositionForTargetVerifier mysql.Position

+ LastSuccessfulPaginationKeys map[string]PaginationKey

+ LastWrittenBinlogPosition mysql.Position

+ func (s *SerializableState) MarshalJSON() ([]byte, error)

+ func (s *SerializableState) MinSourceBinlogPosition() mysql.Position

+ func (s *SerializableState) UnmarshalJSON(data []byte) error

+ type SqlDBWithFakeRollback struct

+ func (d *SqlDBWithFakeRollback) Rollback() error

+ type SqlPreparer interface

+ Prepare func(string) (*sqlorig.Stmt, error)

+ type SqlPreparerAndRollbacker interface

+ Rollback func() error

+ type StateTracker struct

+ BinlogRWMutex *sync.RWMutex

+ CopyRWMutex *sync.RWMutex

+ func NewStateTracker(speedLogCount int) *StateTracker

+ func NewStateTrackerFromSerializedState(speedLogCount int, serializedState *SerializableState) *StateTracker

+ func (s *StateTracker) EstimatedPaginationKeysPerSecond() float64

+ func (s *StateTracker) IsTableComplete(table string) bool

+ func (s *StateTracker) LastSuccessfulPaginationKey(table string, tableSchema *TableSchema) PaginationKey

+ func (s *StateTracker) MarkTableAsCompleted(table string)

+ func (s *StateTracker) RowStatsWrittenPerTable() map[string]RowStats

+ func (s *StateTracker) Serialize(lastKnownTableSchemaCache TableSchemaCache, ...) *SerializableState

+ func (s *StateTracker) UpdateLastResumableBinlogPositionForTargetVerifier(pos mysql.Position)

+ func (s *StateTracker) UpdateLastResumableSourceBinlogPosition(pos mysql.Position)

+ func (s *StateTracker) UpdateLastResumableSourceBinlogPositionForInlineVerifier(pos mysql.Position)

+ func (s *StateTracker) UpdateLastSuccessfulPaginationKey(table string, paginationKey PaginationKey, rowStats RowStats)

+ type StmtCache struct

+ func NewStmtCache() *StmtCache

+ func (c *StmtCache) StmtFor(p SqlPreparer, query string) (*sqlorig.Stmt, error)

+ type TLSConfig struct

+ CertPath string

+ ServerName string

+ func (this *TLSConfig) BuildConfig() (*tls.Config, error)

+ type TableColumnCompressionConfig map[string]map[string]string

+ type TableFilter interface

+ ApplicableDatabases func([]string) ([]string, error)

+ ApplicableTables func([]*TableSchema) ([]*TableSchema, error)

+ type TableIdentifier struct

+ SchemaName string

+ TableName string

+ func NewTableIdentifierFromSchemaTable(table *TableSchema) TableIdentifier

+ type TableMaxPaginationKey struct

+ MaxPaginationKey PaginationKey

+ Table *TableSchema

+ type TableProgress struct

+ BatchSize uint64

+ BytesWritten uint64

+ CurrentAction string

+ LastSuccessfulPaginationKey uint64

+ RowsWritten uint64

+ TargetPaginationKey uint64

+ type TableSchema struct

+ CompressedColumnsForVerification map[string]string

+ ForcedIndexForVerification string

+ IgnoredColumnsForVerification map[string]struct{}

+ PaginationKeyColumn *schema.TableColumn

+ PaginationKeyIndex int

+ func MaxPaginationKeys(db *sql.DB, tables []*TableSchema, logger *logrus.Entry) (map[*TableSchema]PaginationKey, []*TableSchema, error)

+ func (t *TableSchema) FingerprintQuery(schemaName, tableName string, numRows int) string

+ func (t *TableSchema) GetPaginationColumn() *schema.TableColumn

+ func (t *TableSchema) GetPaginationKeyIndex() int

+ func (t *TableSchema) RowMd5Query() string

+ type TableSchemaCache map[string]*TableSchema

+ func LoadTables(db *sql.DB, tableFilter TableFilter, ...) (TableSchemaCache, error)

+ func (c TableSchemaCache) AllTableNames() (tableNames []string)

+ func (c TableSchemaCache) AsSlice() (tables []*TableSchema)

+ func (c TableSchemaCache) Get(database, table string) *TableSchema

+ func (c TableSchemaCache) GetTableListWithPriority(priorityList []string) (prioritizedTableNames []string)

+ type TargetVerifier struct

+ BinlogStreamer *BinlogStreamer

+ DB *sql.DB

+ StateTracker *StateTracker

+ func NewTargetVerifier(targetDB *sql.DB, stateTracker *StateTracker, binlogStreamer *BinlogStreamer) (*TargetVerifier, error)

+ func (t *TargetVerifier) BinlogEventListener(evs []DMLEvent) error

+ type Throttler interface

+ Disabled func() bool

+ Run func(context.Context) error

+ SetDisabled func(bool)

+ SetPaused func(bool)

+ Throttled func() bool

+ type ThrottlerBase struct

+ func (t *ThrottlerBase) Disabled() bool

+ func (t *ThrottlerBase) SetDisabled(disabled bool)

+ type TimerMetric struct

+ Value time.Duration

+ type Uint64Key uint64

+ func NewUint64Key(value uint64) Uint64Key

+ func (k Uint64Key) Compare(other PaginationKey) int

+ func (k Uint64Key) IsMax() bool

+ func (k Uint64Key) MarshalJSON() ([]byte, error)

+ func (k Uint64Key) NumericPosition() float64

+ func (k Uint64Key) SQLValue() interface{}

+ func (k Uint64Key) String() string

+ type UnsupportedCompressionError struct

+ func (e UnsupportedCompressionError) Error() string

+ type UpdatableConfig struct

+ DataIterationBatchSize uint64

+ func (c *UpdatableConfig) Validate() error

+ type VerificationResult struct

+ DataCorrect bool

+ IncorrectTables []string

+ Message string

+ func NewCorrectVerificationResult() VerificationResult

+ func (e VerificationResult) Error() string

+ type VerificationResultAndStatus struct

+ DoneTime time.Time

+ StartTime time.Time

+ func (r VerificationResultAndStatus) IsDone() bool

+ func (r VerificationResultAndStatus) IsStarted() bool

+ type Verifier interface

+ Message func() string

+ Result func() (VerificationResultAndStatus, error)

+ StartInBackground func() error

+ VerifyBeforeCutover func() error

+ VerifyDuringCutover func() (VerificationResult, error)

+ Wait func()

+ type WaitUntilReplicaIsCaughtUpToMaster struct

+ MasterDB *sql.DB

+ ReplicaDB *sql.DB

+ ReplicatedMasterPositionFetcher ReplicatedMasterPositionFetcher

+ Timeout time.Duration

+ func (w *WaitUntilReplicaIsCaughtUpToMaster) IsCaughtUp(targetMasterPos mysql.Position, retries int) (bool, error)

+ func (w *WaitUntilReplicaIsCaughtUpToMaster) Wait() error

+ type WorkerPool struct

+ Concurrency int

+ Process func(int) (interface{}, error)

+ func (p *WorkerPool) Run(n int) ([]interface{}, error)