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) ApplicableEvent(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 ShardedTableFilterType
• 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

	JoinedTables map[string][]JoinTable

	// IgnoredTables and IncludedTables are mutually exclusive. Specifying both is an error.
	IgnoredTables  []string
	IncludedTables []string

	PrimaryKeyTables []string

	Throttle *ghostferry.LagThrottlerConfig
}

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) ApplicableEvent

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

func (*ShardedCopyFilter) BuildSelect

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

type ShardedTableFilter

type ShardedTableFilter struct {
	SourceShard      string
	ShardingKey      string
	JoinedTables     map[string][]JoinTable
	Type             ShardedTableFilterType
	Tables           []*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 ShardedTableFilterType

type ShardedTableFilterType int64

const (
	IgnoredTablesFilter ShardedTableFilterType = iota
	IncludedTablesFilter
)

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