stage

Documentation

Index

• type Stager
  • func NewStager(sconfig *StagerConfig, chk *audit.Checkpoint) (*Stager, error)
  • func (s *Stager) GetETA() string
  • func (s *Stager) GetLowWatermark() (string, error)
  • func (s *Stager) GetProgress() string
  • func (s *Stager) HasClearEstimates() bool
  • func (s *Stager) RetryableStageChunk(ctx context.Context, chunk *table.Chunk) error
  • func (s *Stager) RowsPerSecond() uint64
  • func (s *Stager) SetThrottler(throttler throttler.Throttler)
  • func (s *Stager) Stage(ctx context.Context) error
  • func (s *Stager) StartTime() time.Time
• type StagerConfig

Types

type Stager

type Stager struct {
	sync.Mutex

	StagedRowsCount   uint64 // used for estimates: the exact number of rows staged
	StagedChunksCount uint64

	ChunkerOpen atomic.Bool

	Throttler throttler.Throttler
	// contains filtered or unexported fields
}

func NewStager

func NewStager(sconfig *StagerConfig, chk *audit.Checkpoint) (*Stager, error)

NewStager creates a new stager, from a checkpoint (copyRowsAt, copyRows).

func (*Stager) GetETA

func (s *Stager) GetETA() string

func (*Stager) GetLowWatermark

func (s *Stager) 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 staging table.

func (*Stager) GetProgress

func (s *Stager) GetProgress() string

GetProgress returns the progress of the stager.

func (*Stager) HasClearEstimates

func (s *Stager) HasClearEstimates() bool

func (*Stager) RetryableStageChunk

func (s *Stager) RetryableStageChunk(ctx context.Context, chunk *table.Chunk) error

RetryableStageChunk copies the chunk to staging table, compares checksum between staging and source table and deletes chunk from source table, retrying the process until it succeeds or runs out of max configured retries.

func (*Stager) RowsPerSecond

func (s *Stager) RowsPerSecond() uint64

func (*Stager) SetThrottler

func (s *Stager) SetThrottler(throttler throttler.Throttler)

func (*Stager) Stage

func (s *Stager) Stage(ctx context.Context) error

Stage stages the rows matching in src table to the staging table. Staging means copying rows from source table to staging table and deleting them from source table.

func (*Stager) StartTime

func (s *Stager) StartTime() time.Time

type StagerConfig

type StagerConfig struct {
	DB            *sql.DB
	DBConfig      *dbconn.DBConfig
	DryRun        bool
	Key           string
	Where         string
	TryNum        int
	Logger        loggers.Advanced
	SrcTbl        *table.TableInfo
	StageTbl      *table.TableInfo
	ChunkDuration time.Duration
	Threads       int
}