clone

type Batch struct {
	Type  MutationType
	Table *Table
	Rows  []*Row
}

type Checksum struct {
	ReaderConfig
}

type Chunk struct {
	Table *Table

	// Seq is the sequence number of chunks for this table
	Seq int64

	// Start is the first id of the chunk inclusive. If nil, chunk starts at -inf.
	Start []interface{}
	// End is the first id of the next chunk (i.e. the last id of this chunk exclusively). If nil, chunk ends at +inf.
	End []interface{} // exclusive

	// First chunk of a table
	First bool
	// Last chunk of a table
	Last bool

	// Size is the expected number of rows in the chunk
	Size int
}

type ChunkSnapshot struct {
	InsideWatermarks bool
	Rows             []*Row
	Chunk            Chunk
}

type Clone struct {
	WriterConfig
}

type Config struct {
	Tables map[string]TableConfig `toml:"table"`
}

type DBConfig struct {
	Type             DataSourceType `help:"Datasource name" enum:"mysql,vitess" optional:"" default:"mysql"`
	Host             string         `help:"Hostname" optional:""`
	EgressSocket     string         `help:"Use an egress socket when connecting to Vitess, for example '@egress.sock'" optional:""`
	Username         string         `help:"User" optional:""`
	Password         string         `help:"Password" optional:""`
	Database         string         `help:"Database or Vitess shard with format <keyspace>/<shard>" optional:""`
	MiskDatasource   string         `help:"Misk formatted config yaml file" optional:"" path:""`
	MiskReader       bool           `help:"Use the reader endpoint from Misk (defaults to writer)" optional:"" default:"false"`
	GrpcCustomHeader []string       `help:"Custom GRPC headers separated by ="`
	CA               string         `help:"CA root file, if this is specified then TLS will be enabled (PEM encoded)"`
	Cert             string         `help:"Certificate file for client side authentication (PEM encoded)"`
	Key              string         `help:"Key file for client side authentication (PEM encoded)"`
}

type DBReader interface {
	QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error)
}

type DBWriter interface {
	DBReader
	ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
}

type DataSourceType string

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

type Diff struct {
	Type MutationType
	// Row is the row to update to or insert or delete
	Row *Row

	// Target is in case of the Update MutationType also set so that it can be compared
	Target *Row
}

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

type IdStreamer interface {
	Next(context.Context) ([]interface{}, error)
}

type Mutation struct {
	Type  MutationType
	Table *Table
	Rows  [][]interface{}

	// Chunk is only sent with a Repair mutation type
	Chunk Chunk
}

type MutationType byte

type PeekingIdStreamer interface {
	// Next returns next id and a boolean indicating if there is a next after this one
	Next(context.Context) ([]interface{}, bool, error)
	// Peek returns the id ahead of the current, Next above has to be called first
	Peek() []interface{}
}

type Ping struct {
	SourceTargetConfig

	Table string `help:"If set select a row from this table" optional:""`
}

type Position struct {
	File     string
	Position uint32
	// Gset can be nil in case database does not support GTID
	Gset mysql.GTIDSet
}

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

type ReaderConfig struct {
	SourceTargetConfig

	ChunkSize     int  `help:"Default size of the chunks to diff (can also be overridden per table)" default:"5000"`
	ShuffleChunks bool `` /* 161-byte string literal not displayed */

	TableParallelism  int64         `help:"Number of tables to process concurrently" default:"10"`
	ReaderCount       int           `help:"Number of reader connections" default:"20"`
	ReaderParallelism int64         `help:"Number of reader goroutines" default:"200"`
	ReadTimeout       time.Duration `help:"Timeout for faster reads like diffing a single chunk" default:"30s"`
	ReadRetries       uint64        `help:"How many times to retry reading a single chunk (with backoff)" default:"10"`

	UseCRC32Checksum bool `` /* 126-byte string literal not displayed */

	UseConcurrencyLimits bool `help:"Use concurrency limits to automatically find the throughput of the underlying databases" default:"false"`

	ConfigFile string `help:"TOML formatted config file" short:"f" optional:"" type:"path"`

	// WriteBatchSize doesn't belong to ReaderConfig but we put that in the TableConfig when we load the table which is
	// code reused by both checksum and clone so it's easier to put this here for now
	WriteBatchSize int `help:"Default size of the write batch per transaction (can also be overridden per table)" default:"100"`

	FailedChunkRetryCount int `help:"Retry a chunk if it fails the checksum, this can be used to checksum across a replica with a master" default:"0"`

	Config Config `kong:"-"`
}

type Replicate struct {
	WriterConfig

	TaskName string `` /* 163-byte string literal not displayed */
	ServerID uint32 `help:"Unique identifier of this server, defaults to a hash of the TaskName" optional:""`

	// TODO should this just be ReadParallelism
	ChunkParallelism int `help:"Number of chunks to snapshot concurrently" default:"10"`

	CheckpointTable    string        `` /* 142-byte string literal not displayed */
	WatermarkTable     string        `` /* 127-byte string literal not displayed */
	HeartbeatTable     string        `` /* 163-byte string literal not displayed */
	HeartbeatFrequency time.Duration `` /* 173-byte string literal not displayed */
	CreateTables       bool          `help:"Create the required tables if they do not exist" default:"true"`
	ChunkBufferSize    int           `help:"Size of internal queues" default:"100"`
	ReconnectTimeout   time.Duration `help:"How long to try to reconnect after a replication failure (set to 0 to retry forever)" default:"5m"`

	DoSnapshot      bool          `` /* 127-byte string literal not displayed */
	DoSnapshotDelay time.Duration `help:"How long to wait until running a snapshot" default:"60s"`

	ReplicationParallelism          int64         `help:"Many transactions to apply in parallel during replication" default:"1"`
	ParallelTransactionBatchMaxSize int           `help:"How large batch of transactions to parallelize" default:"100"`
	ParallelTransactionBatchTimeout time.Duration `help:"How long to wait for a batch of transactions to fill up before executing them anyway" default:"5s"`
}

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

type RetryOptions struct {
	Limiter       core.Limiter
	AcquireMetric prometheus.Observer
	MaxRetries    uint64
	Timeout       time.Duration
}

type Row struct {
	Table *Table
	Data  []interface{}
}

type RowStream interface {
	// Next returns the next row or nil if we're done
	Next() (*Row, error)
	// Close releases any potential underlying resources
	Close() error
}

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

type SourceTargetConfig struct {
	Source DBConfig `help:"Database config of source to be copied from" prefix:"source-" embed:""`
	Target DBConfig `help:"Database config of source to be copied from" prefix:"target-" embed:""`
}

type Table struct {
	Name string

	// KeyColumns is the columns the table is chunked by, by default the primary key columns
	KeyColumns []string
	// KeyColumnList is KeyColumns quoted and comma separated
	KeyColumnList string
	// KeyColumnIndexes is KeyColumns quoted and comma separated
	KeyColumnIndexes []int

	Config TableConfig

	Columns       []string
	ColumnsQuoted []string
	CRC32Columns  []string
	// ColumnList is a comma separated list of quoted strings
	ColumnList    string
	EstimatedRows int64

	// MysqlTable is the go-mysql schema, we should probably start using this one as much as possible
	MysqlTable *mysqlschema.Table
}

type TableConfig struct {
	IgnoreColumns  []string `toml:"ignore_columns" help:"Ignore columns in table"`
	TargetWhere    string   `toml:"target_where" help:"Extra where clause that is added on the target"`
	TargetHint     string   `toml:"target_hint" help:"Hint placed after the SELECT on target reads"`
	SourceWhere    string   `toml:"source_where" help:"Extra where clause that is added on the source"`
	SourceHint     string   `toml:"source_hint" help:"Hint placed after the SELECT on target reads"`
	ChunkSize      int      `toml:"chunk_size" help:"Global chunk size if chunk size not specified on the table"`
	WriteBatchSize int      `toml:"write_batch_size" help:"Global chunk size if chunk size not specified on the table"`
	WriteTimout    duration `toml:"write_timeout" help:"Global chunk size if chunk size not specified on the table"`
	KeyColumns     []string `toml:"keys" help:"Use these columns as a unique key for this table, defaults to primary key columns"`
}

type Transaction struct {
	Mutations     []Mutation
	FinalPosition Position
}

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

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

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

type WriterConfig struct {
	ReaderConfig

	WriteBatchStatementSize int           `help:"Size of the write batch per statement" default:"100"`
	WriterParallelism       int64         `help:"Number of writer goroutines" default:"200"`
	WriterCount             int           `help:"Number of writer connections" default:"10"`
	WriteRetries            uint64        `help:"Number of retries" default:"5"`
	WriteTimeout            time.Duration `help:"Timeout for each write" default:"30s"`

	NoDiff bool `help:"Clone without diffing using INSERT IGNORE can be faster as a first pass" default:"false"`
}