row

Documentation

Overview

Package row copies rows from one table to another. it makes use of tableinfo.Chunker, and does the parallelism and retries here. It fails on the first error.

Index

• type Copier
  • func NewCopier(db *sql.DB, tbl, newTable *table.TableInfo, config *CopierConfig) (*Copier, error)
  • func NewCopierFromCheckpoint(db *sql.DB, tbl, newTable *table.TableInfo, config *CopierConfig, ...) (*Copier, error)
  • func (c *Copier) CopyChunk(ctx context.Context, chunk *table.Chunk) error
  • func (c *Copier) GetETA() string
  • func (c *Copier) GetLowWatermark() (string, error)
  • func (c *Copier) GetProgress() string
  • func (c *Copier) KeyAboveHighWatermark(key interface{}) bool
  • func (c *Copier) Next4Test() (*table.Chunk, error)
  • func (c *Copier) Open4Test() error
  • func (c *Copier) Run(ctx context.Context) error
  • func (c *Copier) SetThrottler(throttler throttler.Throttler)
  • func (c *Copier) StartTime() time.Time
• type CopierConfig
  • func NewCopierDefaultConfig() *CopierConfig

Types

type Copier

type Copier struct {
	sync.Mutex

	CopyRowsStartTime    time.Time
	CopyRowsExecTime     time.Duration
	CopyRowsCount        uint64 // used for estimates: the exact number of rows copied
	CopyRowsLogicalCount uint64 // used for estimates on auto-inc PKs: rows copied including any gaps
	CopyChunksCount      uint64

	ExecTime  time.Duration
	Throttler throttler.Throttler
	// contains filtered or unexported fields
}

func NewCopier

func NewCopier(db *sql.DB, tbl, newTable *table.TableInfo, config *CopierConfig) (*Copier, error)

NewCopier creates a new copier object.

func NewCopierFromCheckpoint

func NewCopierFromCheckpoint(db *sql.DB, tbl, newTable *table.TableInfo, config *CopierConfig, lowWatermark string, rowsCopied uint64, rowsCopiedLogical uint64) (*Copier, error)

NewCopierFromCheckpoint creates a new copier object, from a checkpoint (copyRowsAt, copyRows)

func (*Copier) CopyChunk

func (c *Copier) CopyChunk(ctx context.Context, chunk *table.Chunk) error

CopyChunk copies a chunk from the table to the newTable. it is public so it can be used in tests incrementally.

func (*Copier) GetETA

func (c *Copier) GetETA() string

func (*Copier) GetLowWatermark

func (c *Copier) GetLowWatermark() (string, error)

GetLowWatermark returns the low watermark of the chunker, i.e. the lowest key that has been guaranteed to be written to the new table.

func (*Copier) GetProgress

func (c *Copier) GetProgress() string

GetProgress returns the progress of the copier

func (*Copier) KeyAboveHighWatermark

func (c *Copier) KeyAboveHighWatermark(key interface{}) bool

KeyAboveHighWatermark returns true if the key is above where the chunker is currently at.

func (*Copier) Next4Test

func (c *Copier) Next4Test() (*table.Chunk, error)

Next4Test is typically only used in integration tests that don't want to actually migrate data, but need to advance the chunker.

func (*Copier) Open4Test

func (c *Copier) Open4Test() error

Open4Test is typically only used in integration tests that don't want to actually migrate data, but need to open the chunker.

func (*Copier) Run

func (c *Copier) Run(ctx context.Context) error

func (*Copier) SetThrottler

func (c *Copier) SetThrottler(throttler throttler.Throttler)

func (*Copier) StartTime

func (c *Copier) StartTime() time.Time

type CopierConfig

type CopierConfig struct {
	Concurrency     int
	TargetChunkTime time.Duration
	FinalChecksum   bool
	Throttler       throttler.Throttler
	Logger          loggers.Advanced
	MetricsSink     metrics.Sink
	DBConfig        *dbconn.DBConfig
}

func NewCopierDefaultConfig

func NewCopierDefaultConfig() *CopierConfig

NewCopierDefaultConfig returns a default config for the copier.