mysql

package
v0.0.0-rc3 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 25, 2024 License: Apache-2.0 Imports: 62 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	MysqlFlavorTypeMysql   = "mysql"
	MysqlFlavorTypeMariaDB = "mariadb"
)
View Source 
const (
	TableTransferProgress = "__table_transfer_progress"
	TableTmGtidKeeper     = "__tm_gtid_keeper"
	TableTmKeeper         = "__tm_keeper"
)
View Source 
const (
	DisableFKQuery = "SET FOREIGN_KEY_CHECKS=0;\n"
	EnableFKQuery  = "SET FOREIGN_KEY_CHECKS=1;\n"
)
View Source 
const (
	UpdateAction = "update"
	InsertAction = "insert"
	DeleteAction = "delete"
)

The action name for sync.

View Source 
const (
	ErrCodeDuplicateKey = 1062
	ErrCodeSyntax       = 1064
	ErrCodeLockTimeout  = 1205
	ErrCodeDeadlock     = 1213
)
View Source 
const (
	SnapshotWait = Status("SnapshotWait")
	SyncWait     = Status("SyncWait")
	InSync       = Status("InSync")
)
View Source 
const ProviderType = abstract.ProviderType("mysql")

Variables

View Source 
var (
	UnknownTableRetryPeriod = time.Second * time.Duration(10)
	ErrExcludedTable        = xerrors.New("table is excluded")
)

canal will retry fetching unknown table's meta after UnknownTableRetryPeriod

View Source 
var (
	CodeSyntax   = coded.Register("mysql", "incorrect_syntax")
	CodeDeadlock = coded.Register("mysql", "deadlock")
)
View Source 
var AlreadyExistsCodes = map[int]bool{
	1050: true,
	1304: true,
	1359: true,
	1826: true,
}
View Source 
var NotEnoughProcPermissions = "" /* 186-byte string literal not displayed */
View Source 
var TableOrViewNotExistsCode = uint16(1146)

Functions

func CalculateLSN 

func CalculateLSN(file string, pos uint32) uint64

func CastRowsToDT 

func CastRowsToDT(event *RowsEvent, location *time.Location) error

func CastToMySQL 

func CastToMySQL(val interface{}, typ abstract.ColSchema) string

func CheckMySQLBinlogRowImageFormat 

func CheckMySQLBinlogRowImageFormat(source *MysqlSource) error

func CheckMySQLVersion 

func CheckMySQLVersion(storage *Storage) string

func ClearOriginalType 

func ClearOriginalType(colSchema abstract.ColSchema) string

func Connect 

func Connect(params *ConnectionParams, configAction func(config *mysql.Config) error) (*sql.DB, error)

func CopySchema 

func CopySchema(source *MysqlSource, steps *MysqlDumpSteps, pusher abstract.Pusher) error

func CreateCertPool 

func CreateCertPool(certPEMFile string, rootCAFiles []string) (*x509.CertPool, error)

func GetLogFilePosition 

func GetLogFilePosition(storage *Storage) (string, uint32, string, error)

func GetTableDDLs 

func GetTableDDLs(source *MysqlSource, connection *sqlx.DB, databases []string) ([]ddlValue, []ddlValue, error)

func GetTriggerDDLs 

func GetTriggerDDLs(source *MysqlSource, connection *sqlx.DB, databases []string) ([]ddlValue, error)

func GetViewDDLs 

func GetViewDDLs(source *MysqlSource, connection *sqlx.DB, databases []string) ([]ddlValue, error)

func IsErrorCode 

func IsErrorCode(err error, errNumber uint16) bool

func IsErrorCodes 

func IsErrorCodes(err error, codes map[int]bool) bool

func IsGtidModeEnabled 

func IsGtidModeEnabled(storage *Storage, flavor string) (bool, error)

func LoadMysqlSchema 

func LoadMysqlSchema(transfer *server.Transfer, registry metrics.Registry, isAfter bool) error

func LoadSchema 

func LoadSchema(tx queryExecutor, useFakePrimaryKey bool, includeViews bool) (abstract.DBSchema, error)

func LoadTableConstraints 

func LoadTableConstraints(tx queryExecutor, table abstract.TableID) (map[string][]string, error)

func MakeArrBacktickedColumnNames 

func MakeArrBacktickedColumnNames(tableSchema *[]abstract.ColSchema) []string

func New 

func New(lgr log.Logger, registry metrics.Registry, cp coordinator.Coordinator, transfer *server.Transfer) providers.Provider

func NewSinker 

func NewSinker(lgr log.Logger, cfg *MysqlDestination, mtrcs metrics.Registry) (abstract.Sinker, error)

func NewSource 

func NewSource(src *MysqlSource, transferID string, objects *server.DataObjects, logger log.Logger, registry metrics.Registry, cp coordinator.Coordinator, failOnDecimal bool) (abstract.Source, error)

func OrderByPrimaryKeys 

func OrderByPrimaryKeys(tableSchema []abstract.ColSchema, direction string) (string, error)

func RemoveTracker 

func RemoveTracker(src *MysqlSource, id string, cp coordinator.Coordinator) error

func RestoreStringAndBytes 

func RestoreStringAndBytes(event *RowsEvent) error

func SyncBinlogPosition 

func SyncBinlogPosition(src *MysqlSource, id string, cp coordinator.Coordinator) error

func TypeToMySQL 

func TypeToMySQL(column abstract.ColSchema) string

func TypeToYt 

func TypeToYt(rawColumnType string) schema.Type

func Validate 

func Validate(event *RowsEvent) error

Types

type Canal 

type Canal struct {
	// contains filtered or unexported fields
}

Canal can sync your MySQL data into everywhere, like Elasticsearch, Redis, etc... MySQL must open row format for binlog

func NewCanal 

func NewCanal(cfg *Config) (*Canal, error)

func (*Canal) CatchMasterPos 

func (c *Canal) CatchMasterPos(timeout time.Duration) error

func (*Canal) CheckBinlogRowImage 

func (c *Canal) CheckBinlogRowImage(image string) error

CheckBinlogRowImage checks MySQL binlog row image, must be in FULL, MINIMAL, NOBLOB

func (*Canal) ClearTableCache 

func (c *Canal) ClearTableCache(db []byte, table []byte)

ClearTableCache clear table cache

func (*Canal) Close 

func (c *Canal) Close()

func (*Canal) Ctx 

func (c *Canal) Ctx() context.Context

func (*Canal) Execute 

func (c *Canal) Execute(cmd string, args ...interface{}) (*mysql.Result, error)

Execute a SQL

func (*Canal) FlushBinlog 

func (c *Canal) FlushBinlog() error

func (*Canal) GetDelay 

func (c *Canal) GetDelay() uint32

func (*Canal) GetMasterGTIDSet 

func (c *Canal) GetMasterGTIDSet() (mysql.GTIDSet, error)

func (*Canal) GetMasterPos 

func (c *Canal) GetMasterPos() (mysql.Position, error)

func (*Canal) GetTable 

func (c *Canal) GetTable(db string, table string) (*schema.Table, error)

func (*Canal) Run 

func (c *Canal) Run() error

Run sync from the binlog position in the data. It will run forever until meeting an error or Canal closed.

func (*Canal) RunFrom 

func (c *Canal) RunFrom(pos mysql.Position) error

RunFrom will sync from the binlog position directly, ignore mysqldump.

func (*Canal) SetEventHandler 

func (c *Canal) SetEventHandler(h EventHandler)

`SetEventHandler` registers the sync handler, you must register your own handler before starting Canal.

func (*Canal) StartFromGTID 

func (c *Canal) StartFromGTID(set mysql.GTIDSet) error

func (*Canal) SyncedGTIDSet 

func (c *Canal) SyncedGTIDSet() mysql.GTIDSet

func (*Canal) SyncedPosition 

func (c *Canal) SyncedPosition() mysql.Position

func (*Canal) SyncedTimestamp 

func (c *Canal) SyncedTimestamp() uint32

func (*Canal) WaitUntilPos 

func (c *Canal) WaitUntilPos(pos mysql.Position, timeout time.Duration) error

type Config 

type Config struct {
	Addr     string
	User     string
	Password string

	Charset         string
	ServerID        uint32
	Flavor          string
	HeartbeatPeriod time.Duration
	ReadTimeout     time.Duration

	// discard row event without table meta
	DiscardNoMetaRowEvent bool

	UseDecimal    bool
	FailOnDecimal bool
	ParseTime     bool

	TimestampStringLocation *time.Location

	// SemiSyncEnabled enables semi-sync or not.
	SemiSyncEnabled bool

	// Set to change the maximum number of attempts to re-establish a broken
	// connection
	MaxReconnectAttempts int
	TLSConfig            *tls.Config
	Include              func(db string, table string) bool
}

TODO: remove

type ConnectionParams 

type ConnectionParams struct {
	Host        string
	Port        int
	User        string
	Password    string
	Database    string
	TLS         bool
	CertPEMFile string
	Location    *time.Location
	RootCAFiles []string
}

func NewConnectionParams 

func NewConnectionParams(config *MysqlStorageParams) (*ConnectionParams, error)

func (*ConnectionParams) ResolveHost 

func (params *ConnectionParams) ResolveHost(clusterID string) error

func (*ConnectionParams) ResolveLocation 

func (params *ConnectionParams) ResolveLocation(locationStr string) error

type CreateFunctionRow 

type CreateFunctionRow struct {
	Name                string `db:"Function"`
	Mode                string `db:"sql_mode"`
	DDL                 string `db:"Create Function"`
	CharacterSetClient  string `db:"character_set_client"`
	ConnectionCollation string `db:"collation_connection"`
	DBCollation         string `db:"Database Collation"`
}

type CreateProcedureRow 

type CreateProcedureRow struct {
	Name                string `db:"Procedure"`
	Mode                string `db:"sql_mode"`
	DDL                 string `db:"Create Procedure"`
	CharacterSetClient  string `db:"character_set_client"`
	ConnectionCollation string `db:"collation_connection"`
	DBCollation         string `db:"Database Collation"`
}

type CreateTableRow 

type CreateTableRow struct {
	Name string `db:"Table"`
	DDL  string `db:"Create Table"`
}

type CreateTriggerRow 

type CreateTriggerRow struct {
	Name                string `db:"Trigger"`
	Mode                string `db:"sql_mode"`
	DDL                 string `db:"SQL Original Statement"`
	CharacterSetClient  string `db:"character_set_client"`
	ConnectionCollation string `db:"collation_connection"`
	DBCollation         string `db:"Database Collation"`
	Created             string `db:"Created"`
}

type CreateViewRow 

type CreateViewRow struct {
	Name               string `db:"View"`
	DDL                string `db:"Create View"`
	CharacterSetClient string `db:"character_set_client"`
	Collation          string `db:"collation_connection"`
}

type DummyEventHandler 

type DummyEventHandler struct {
}

func (*DummyEventHandler) OnDDL 

func (h *DummyEventHandler) OnDDL(nextPos mysql.Position, queryEvent *replication.QueryEvent) error

func (*DummyEventHandler) OnGTID 

func (h *DummyEventHandler) OnGTID(mysql.GTIDSet) error

func (*DummyEventHandler) OnPosSynced 

func (h *DummyEventHandler) OnPosSynced(mysql.Position, mysql.GTIDSet, bool) error

func (*DummyEventHandler) OnRotate 

func (h *DummyEventHandler) OnRotate(*replication.RotateEvent) error

func (*DummyEventHandler) OnRow 

func (h *DummyEventHandler) OnRow(*RowsEvent) error

func (*DummyEventHandler) OnTableChanged 

func (h *DummyEventHandler) OnTableChanged(schema string, table string) error

func (*DummyEventHandler) OnXID 

func (h *DummyEventHandler) OnXID(mysql.Position) error

func (*DummyEventHandler) String 

func (h *DummyEventHandler) String() string

type EmptyProgress 

type EmptyProgress struct{}

func (*EmptyProgress) BuildLSNQuery 

func (t *EmptyProgress) BuildLSNQuery(name string, lsn uint64, status Status) string

func (*EmptyProgress) GetCurrentState 

func (t *EmptyProgress) GetCurrentState() (map[string]TableStatus, error)

type EventHandler 

type EventHandler interface {
	OnRotate(roateEvent *replication.RotateEvent) error
	// OnTableChanged is called when the table is created, altered, renamed or dropped.
	// You need to clear the associated data like cache with the table.
	// It will be called before OnDDL.
	OnTableChanged(schema string, table string) error
	OnDDL(nextPos mysql.Position, queryEvent *replication.QueryEvent) error
	OnRow(event *RowsEvent) error
	OnXID(nextPos mysql.Position) error
	OnGTID(gtid mysql.GTIDSet) error
	// OnPosSynced Use your own way to sync position. When force is true, sync position immediately.
	OnPosSynced(pos mysql.Position, set mysql.GTIDSet, force bool) error
	String() string
}

type LsnProgress 

type LsnProgress interface {
	GetCurrentState() (map[string]TableStatus, error)
	BuildLSNQuery(name string, lsn uint64, status Status) string
}

func NewEmptyProgress 

func NewEmptyProgress() (LsnProgress, error)

func NewTableProgressTracker 

func NewTableProgressTracker(db *sql.DB, dbName string) (LsnProgress, error)

type MysqlDestination 

type MysqlDestination struct {
	AllowReplace         bool
	Cleanup              server.CleanupType
	ClusterID            string
	Database             string
	DisableParallelWrite map[TableName]bool
	Host                 string
	IsPublic             bool
	MaintainTables       bool
	MaxParralelWriters   int64
	Password             server.SecretString
	PerTransactionPush   bool
	Port                 int
	ProgressTrackerDB    string
	SecurityGroupIDs     []string
	SkipKeyChecks        bool
	SQLMode              string
	SubNetworkID         string
	Timezone             string
	TLSFile              string
	TransformerConfig    map[string]string
	User                 string

	BufferTriggingSize     uint64
	BufferTriggingInterval time.Duration

	RootCAFiles []string
	// contains filtered or unexported fields
}

func (*MysqlDestination) BuffererConfig 

func (d *MysqlDestination) BuffererConfig() bufferer.BuffererConfig

func (*MysqlDestination) CleanupMode 

func (d *MysqlDestination) CleanupMode() server.CleanupType

func (*MysqlDestination) GetProviderType 

func (d *MysqlDestination) GetProviderType() abstract.ProviderType

func (*MysqlDestination) HasTLS 

func (d *MysqlDestination) HasTLS() bool

func (MysqlDestination) IsDestination 

func (MysqlDestination) IsDestination()

func (*MysqlDestination) MDBClusterID 

func (d *MysqlDestination) MDBClusterID() string

func (*MysqlDestination) PostSnapshotHacks 

func (d *MysqlDestination) PostSnapshotHacks()

func (*MysqlDestination) PreSnapshotHacks 

func (d *MysqlDestination) PreSnapshotHacks()

func (*MysqlDestination) ReliesOnSystemTablesTransferring 

func (d *MysqlDestination) ReliesOnSystemTablesTransferring() bool

func (*MysqlDestination) ToStorageParams 

func (d *MysqlDestination) ToStorageParams() *MysqlStorageParams

func (*MysqlDestination) Transformer 

func (d *MysqlDestination) Transformer() map[string]string

func (*MysqlDestination) Validate 

func (d *MysqlDestination) Validate() error

func (*MysqlDestination) WithDefaults 

func (d *MysqlDestination) WithDefaults()

type MysqlDumpSteps 

type MysqlDumpSteps struct {
	View    bool
	Routine bool
	Trigger bool
	Tables  bool
}

func DefaultMysqlDumpPostSteps 

func DefaultMysqlDumpPostSteps() *MysqlDumpSteps

func DefaultMysqlDumpPreSteps 

func DefaultMysqlDumpPreSteps() *MysqlDumpSteps

type MysqlFlavorType 

type MysqlFlavorType string

type MysqlSource 

type MysqlSource struct {
	Host              string
	User              string
	Password          server.SecretString
	ClusterID         string
	ServerID          uint32
	IncludeTableRegex []string
	ExcludeTableRegex []string
	IsPublic          bool
	Database          string
	TLSFile           string
	SubNetworkID      string
	SecurityGroupIDs  []string
	Port              int
	Timezone          string
	BufferLimit       uint32
	UseFakePrimaryKey bool
	IsHomo            bool `json:"-"`
	PreSteps          *MysqlDumpSteps
	PostSteps         *MysqlDumpSteps
	TrackerDatabase   string

	ConsistentSnapshot          bool
	SnapshotDegreeOfParallelism int
	AllowDecimalAsFloat         bool

	NoTracking  bool // deprecated: use Tracker
	YtTracking  bool // deprecated: use Tracker
	YdbTracking bool // deprecated: use Tracker

	Tracker     MysqlTrackerStorage // deprecated: we only have one tracker now
	PlzNoHomo   bool                // forcefully disable homo features, mostly for tests
	RootCAFiles []string
}

func (*MysqlSource) AllIncludes 

func (s *MysqlSource) AllIncludes() []string

func (*MysqlSource) FulfilledIncludes 

func (s *MysqlSource) FulfilledIncludes(tID abstract.TableID) (result []string)

func (*MysqlSource) GetProviderType 

func (s *MysqlSource) GetProviderType() abstract.ProviderType

func (*MysqlSource) HasTLS 

func (s *MysqlSource) HasTLS() bool

func (*MysqlSource) Include 

func (s *MysqlSource) Include(tID abstract.TableID) bool

func (*MysqlSource) InitServerID 

func (s *MysqlSource) InitServerID(transferID string)

func (MysqlSource) IsSource 

func (MysqlSource) IsSource()

func (MysqlSource) IsStrictSource 

func (MysqlSource) IsStrictSource()

func (*MysqlSource) MDBClusterID 

func (s *MysqlSource) MDBClusterID() string

func (*MysqlSource) ToStorageParams 

func (s *MysqlSource) ToStorageParams() *MysqlStorageParams

func (*MysqlSource) Validate 

func (s *MysqlSource) Validate() error

func (*MysqlSource) WithDefaults 

func (s *MysqlSource) WithDefaults()

type MysqlStorageParams 

type MysqlStorageParams struct {
	ClusterID   string
	Host        string
	Port        int
	User        string
	Password    string
	Database    string
	TLS         bool
	CertPEMFile string

	UseFakePrimaryKey   bool
	DegreeOfParallelism int
	Timezone            string

	TableFilter        abstract.Includeable
	PreSteps           *MysqlDumpSteps
	ConsistentSnapshot bool
	RootCAFiles        []string
}

type MysqlTrackerStorage 

type MysqlTrackerStorage string

type NotMasterError 

type NotMasterError struct {
	// contains filtered or unexported fields
}

func (NotMasterError) Error 

func (e NotMasterError) Error() string

func (NotMasterError) Is 

func (e NotMasterError) Is(err error) bool

type Provider 

type Provider struct {
	// contains filtered or unexported fields
}

func (*Provider) Activate 

func (p *Provider) Activate(ctx context.Context, task *server.TransferOperation, tables abstract.TableMap, callbacks providers.ActivateCallbacks) error

func (*Provider) Cleanup 

func (p *Provider) Cleanup(ctx context.Context, task *server.TransferOperation) error

func (*Provider) Deactivate 

func (p *Provider) Deactivate(ctx context.Context, task *server.TransferOperation) error

func (*Provider) DestinationSampleableStorage 

func (p *Provider) DestinationSampleableStorage() (abstract.SampleableStorage, error)

func (*Provider) Sink 

func (p *Provider) Sink(middlewares.Config) (abstract.Sinker, error)

func (*Provider) Source 

func (p *Provider) Source() (abstract.Source, error)

func (*Provider) SourceSampleableStorage 

func (p *Provider) SourceSampleableStorage() (abstract.SampleableStorage, []abstract.TableDescription, error)

func (*Provider) Storage 

func (p *Provider) Storage() (abstract.Storage, error)

func (*Provider) Type 

func (p *Provider) Type() abstract.ProviderType

func (*Provider) Update 

func (p *Provider) Update(ctx context.Context, addedTables []abstract.TableDescription) error

type Queryable 

type Queryable interface {
	ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
	QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error)
	QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row
}

type RoutineRow 

type RoutineRow struct {
	Name   string
	Schema string
	Typ    string
}

type RowsEvent 

type RowsEvent struct {
	Table  *schema.Table
	Action string
	Header *replication.EventHeader
	// binlog has three update event version, v0, v1 and v2.
	// for v1 and v2, the rows number must be even.
	// Two rows for one event, format is [before update row, after update row]
	// for update v0, only one row for a event, and we don't support this version.
	Data *replication.RowsEvent

	// rows query is defined if option binlog_rows_query_log_events is enabled
	Query string
}

RowsEvent is the event for row replication.

func (*RowsEvent) GetColumnCollation 

func (r *RowsEvent) GetColumnCollation(i int) string

func (*RowsEvent) GetColumnEnumValue 

func (r *RowsEvent) GetColumnEnumValue(columnIndex int, enumValueIndex int64) string

func (*RowsEvent) GetColumnName 

func (r *RowsEvent) GetColumnName(i int) string

func (*RowsEvent) GetColumnRawType 

func (r *RowsEvent) GetColumnRawType(i int) string

func (*RowsEvent) GetColumnSetValue 

func (r *RowsEvent) GetColumnSetValue(columnIndex int, setValueIndex int) string

func (*RowsEvent) IsAllColumnsPresent1 

func (r *RowsEvent) IsAllColumnsPresent1() bool

func (*RowsEvent) IsAllColumnsPresent2 

func (r *RowsEvent) IsAllColumnsPresent2() bool

func (*RowsEvent) IsColumnPresent1 

func (r *RowsEvent) IsColumnPresent1(i uint64) bool

func (*RowsEvent) IsColumnPresent2 

func (r *RowsEvent) IsColumnPresent2(i uint64) bool

func (*RowsEvent) Nullable 

func (r *RowsEvent) Nullable(i uint64) bool

func (*RowsEvent) String 

func (r *RowsEvent) String() string

String implements fmt.Stringer interface.

type StatementID 

type StatementID struct {
	Namespace string `db:"Namespace"`
	Name      string `db:"Name"`
}

type Status 

type Status string

type Storage 

type Storage struct {
	ConnectionParams *ConnectionParams

	IsHomo bool
	DB     *sql.DB
	// contains filtered or unexported fields
}

func NewStorage 

func NewStorage(config *MysqlStorageParams) (*Storage, error)

func (*Storage) BeginSnapshot 

func (s *Storage) BeginSnapshot(ctx context.Context) error

func (*Storage) Close 

func (s *Storage) Close()

func (*Storage) DatabaseSchema 

func (s *Storage) DatabaseSchema() string

func (*Storage) EndSnapshot 

func (s *Storage) EndSnapshot(context.Context) error

func (*Storage) EstimateTableRowsCount 

func (s *Storage) EstimateTableRowsCount(table abstract.TableID) (uint64, error)

func (*Storage) ExactTableRowsCount 

func (s *Storage) ExactTableRowsCount(table abstract.TableID) (uint64, error)

func (*Storage) ListViews 

func (s *Storage) ListViews() ([]abstract.TableID, error)

func (*Storage) LoadRandomSample 

func (s *Storage) LoadRandomSample(table abstract.TableDescription, pusher abstract.Pusher) error

func (*Storage) LoadSampleBySet 

func (s *Storage) LoadSampleBySet(table abstract.TableDescription, keySet []map[string]interface{}, pusher abstract.Pusher) error

func (*Storage) LoadSchema 

func (s *Storage) LoadSchema() (schema abstract.DBSchema, err error)

func (*Storage) LoadTable 

func (s *Storage) LoadTable(ctx context.Context, table abstract.TableDescription, pusher abstract.Pusher) error

func (*Storage) LoadTopBottomSample 

func (s *Storage) LoadTopBottomSample(table abstract.TableDescription, pusher abstract.Pusher) error

func (*Storage) Ping 

func (s *Storage) Ping() error

func (*Storage) Position 

func (s *Storage) Position(ctx context.Context) (*abstract.LogPosition, error)

func (*Storage) TableAccessible 

func (s *Storage) TableAccessible(table abstract.TableDescription) bool

func (*Storage) TableExists 

func (s *Storage) TableExists(table abstract.TableID) (bool, error)

func (*Storage) TableList 

func (s *Storage) TableList(includeTableFilter abstract.IncludeTableList) (abstract.TableMap, error)

func (*Storage) TableSchema 

func (s *Storage) TableSchema(ctx context.Context, table abstract.TableID) (*abstract.TableSchema, error)

func (*Storage) TableSizeInBytes 

func (s *Storage) TableSizeInBytes(table abstract.TableID) (uint64, error)

type TableName 

type TableName = string

type TableProgress 

type TableProgress struct {
	// contains filtered or unexported fields
}

func (*TableProgress) BuildLSNQuery 

func (t *TableProgress) BuildLSNQuery(name string, lsn uint64, status Status) string

func (*TableProgress) GetCurrentState 

func (t *TableProgress) GetCurrentState() (map[string]TableStatus, error)

type TableStatus 

type TableStatus struct {
	LSN    uint64
	Status Status
}

type TemplateCol 

type TemplateCol struct {
	Name, Typ, Comma string
}

type TemplateModel 

type TemplateModel struct {
	Cols  []TemplateCol
	Keys  []TemplateCol
	Table string
}

type Tracker 

type Tracker struct {
	// contains filtered or unexported fields
}

func NewTracker 

func NewTracker(src *MysqlSource, transferID string, cp coordinator.Coordinator) (result *Tracker, err error)

func (*Tracker) Close 

func (n *Tracker) Close() error

func (*Tracker) Get 

func (n *Tracker) Get() (file string, pos uint32, err error)

func (*Tracker) GetGtidset 

func (n *Tracker) GetGtidset() (mysql.GTIDSet, error)

func (*Tracker) Remove 

func (n *Tracker) Remove() error

func (*Tracker) RemoveGtidset 

func (n *Tracker) RemoveGtidset() error

func (*Tracker) Store 

func (n *Tracker) Store(file string, pos uint32) error

func (*Tracker) StoreGtidset 

func (n *Tracker) StoreGtidset(gtidset mysql.GTIDSet) error

type TriggerRow 

type TriggerRow struct {
	Namespace string
	Name      string
}

Source Files

View all Source files

Directories

Path Synopsis
unmarshaller
replication
snapshot
types

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.