sharding

Documentation

Index

• func InitializeMetrics(prefix string, config *Config) error
• func SetGlobalMetrics(prefix string, metricsChan chan interface{})
• func StopAndFlushMetrics()
• type Config
  • func (c *Config) ValidateConfig() error
• type JoinTable
• type ShardedCopyFilter
  • func (f *ShardedCopyFilter) ApplicableDMLEvent(event ghostferry.DMLEvent) (bool, error)
  • func (f *ShardedCopyFilter) BuildSelect(columns []string, table *ghostferry.TableSchema, ...) (sq.SelectBuilder, error)
• type ShardedTableFilter
  • func (s *ShardedTableFilter) ApplicableDatabases(dbs []string) ([]string, error)
  • func (s *ShardedTableFilter) ApplicableTables(tables []*ghostferry.TableSchema) (applicable []*ghostferry.TableSchema, err error)
• type ShardingFerry
  • func NewFerry(config *Config) (*ShardingFerry, error)
  • func (r *ShardingFerry) AbortIfTargetDbNoLongerWriteable()
  • func (r *ShardingFerry) Initialize() error
  • func (r *ShardingFerry) Run()
  • func (r *ShardingFerry) Start() error

Functions

func InitializeMetrics

func InitializeMetrics(prefix string, config *Config) error

func SetGlobalMetrics

func SetGlobalMetrics(prefix string, metricsChan chan interface{})

func StopAndFlushMetrics

func StopAndFlushMetrics()

Types

type Config

type Config struct {
	*ghostferry.Config

	ShardingKey   string
	ShardingValue int64
	SourceDB      string
	TargetDB      string

	SourceReplicationMaster       *ghostferry.DatabaseConfig
	ReplicatedMasterPositionQuery string
	RunFerryFromReplica           bool

	StatsDAddress string
	CutoverLock   ghostferry.HTTPCallback
	CutoverUnlock ghostferry.HTTPCallback
	ErrorCallback ghostferry.HTTPCallback

	JoinedTables     map[string][]JoinTable
	IgnoredTables    []string
	PrimaryKeyTables []string

	Throttle *ghostferry.LagThrottlerConfig

	// These two values configure the amount of times Ferry should attempt to
	// retry acquiring the cutover lock, and for how long the Ferry should wait
	// before attempting another lock acquisition
	MaxCutoverRetries       int
	CutoverRetryWaitSeconds int
}

func (*Config) ValidateConfig

func (c *Config) ValidateConfig() error

type JoinTable

type JoinTable struct {
	TableName, JoinColumn string
}

type ShardedCopyFilter

type ShardedCopyFilter struct {
	ShardingKey      string
	ShardingValue    interface{}
	JoinedTables     map[string][]JoinTable
	PrimaryKeyTables map[string]struct{}
	// contains filtered or unexported fields
}

func (*ShardedCopyFilter) ApplicableDMLEvent

func (f *ShardedCopyFilter) ApplicableDMLEvent(event ghostferry.DMLEvent) (bool, error)

func (*ShardedCopyFilter) BuildSelect

func (f *ShardedCopyFilter) BuildSelect(columns []string, table *ghostferry.TableSchema, lastPaginationKey *ghostferry.PaginationKeyData, batchSize uint64, sortDescending bool) (sq.SelectBuilder, error)

type ShardedTableFilter

type ShardedTableFilter struct {
	SourceShard      string
	ShardingKey      string
	JoinedTables     map[string][]JoinTable
	IgnoredTables    []*regexp.Regexp
	PrimaryKeyTables map[string]struct{}
}

func (*ShardedTableFilter) ApplicableDatabases

func (s *ShardedTableFilter) ApplicableDatabases(dbs []string) ([]string, error)

func (*ShardedTableFilter) ApplicableTables

func (s *ShardedTableFilter) ApplicableTables(tables []*ghostferry.TableSchema) (applicable []*ghostferry.TableSchema, err error)

type ShardingFerry

type ShardingFerry struct {
	Ferry *ghostferry.Ferry
	// contains filtered or unexported fields
}

func NewFerry

func NewFerry(config *Config) (*ShardingFerry, error)

func (*ShardingFerry) AbortIfTargetDbNoLongerWriteable

func (r *ShardingFerry) AbortIfTargetDbNoLongerWriteable()

func (*ShardingFerry) Initialize

func (r *ShardingFerry) Initialize() error

func (*ShardingFerry) Run

func (r *ShardingFerry) Run()

func (*ShardingFerry) Start

func (r *ShardingFerry) Start() error