snapshot

Documentation

Index

• Variables
• type Chunk
• type ChunkStatus
• type ConnectionPool
  • func NewConnectionPool(ctx context.Context, dsn string, size int) (*ConnectionPool, error)
  • func (p *ConnectionPool) Close(ctx context.Context)
  • func (p *ConnectionPool) Get(ctx context.Context) (pg.Connection, error)
  • func (p *ConnectionPool) Put(conn pg.Connection)
• type DecoderCache
  • func NewDecoderCache() *DecoderCache
  • func (c *DecoderCache) Get(oid uint32) *TypeDecoder
  • func (c *DecoderCache) Size() int
• type Handler
• type Job
• type Snapshotter
  • func New(ctx context.Context, snapshotConfig config.SnapshotConfig, ...) (*Snapshotter, error)
  • func (s *Snapshotter) Close(ctx context.Context)
  • func (s *Snapshotter) Execute(ctx context.Context, handler Handler, slotName string) error
  • func (s *Snapshotter) LoadJob(ctx context.Context, slotName string) (*Job, error)
  • func (s *Snapshotter) Prepare(ctx context.Context, slotName string) error
• type TypeDecoder
  • func (d *TypeDecoder) Decode(typeMap *pgtype.Map, data []byte) (any, error)

Variables

var ErrSnapshotInvalidated = fmt.Errorf("snapshot invalidated by coordinator restart")

Types

type Chunk

type Chunk struct {
	ClaimedAt      *time.Time
	HeartbeatAt    *time.Time
	CompletedAt    *time.Time
	RangeEnd       *int64
	RangeStart     *int64
	RangeEndText   *string
	RangeStartText *string
	Status         ChunkStatus
	TableName      string
	ClaimedBy      string
	TableSchema    string
	SlotName       string
	PKColumn       string
	ID             int64
	ChunkIndex     int
	ChunkStart     int64
	ChunkSize      int64
	UseCtid        bool
}

type ChunkStatus

type ChunkStatus string

const (
	ChunkStatusPending    ChunkStatus = "pending"
	ChunkStatusInProgress ChunkStatus = "in_progress"
	ChunkStatusCompleted  ChunkStatus = "completed"
)

type ConnectionPool

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

func NewConnectionPool

func NewConnectionPool(ctx context.Context, dsn string, size int) (*ConnectionPool, error)

func (*ConnectionPool) Close

func (p *ConnectionPool) Close(ctx context.Context)

func (*ConnectionPool) Get

func (p *ConnectionPool) Get(ctx context.Context) (pg.Connection, error)

func (*ConnectionPool) Put

func (p *ConnectionPool) Put(conn pg.Connection)

type DecoderCache

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

func NewDecoderCache

func NewDecoderCache() *DecoderCache

func (*DecoderCache) Get

func (c *DecoderCache) Get(oid uint32) *TypeDecoder

func (*DecoderCache) Size

func (c *DecoderCache) Size() int

type Handler

type Handler func(event *format.Snapshot) error

type Job

type Job struct {
	StartedAt       time.Time
	SlotName        string
	SnapshotID      string
	SnapshotLSN     pg.LSN
	TotalChunks     int
	CompletedChunks int
	Completed       bool
}

type Snapshotter

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

func New

func New(ctx context.Context, snapshotConfig config.SnapshotConfig, tables publication.Tables, dsn string) (*Snapshotter, error)

func (*Snapshotter) Close

func (s *Snapshotter) Close(ctx context.Context)

func (*Snapshotter) Execute

func (s *Snapshotter) Execute(ctx context.Context, handler Handler, slotName string) error

func (*Snapshotter) LoadJob

func (s *Snapshotter) LoadJob(ctx context.Context, slotName string) (*Job, error)

func (*Snapshotter) Prepare

func (s *Snapshotter) Prepare(ctx context.Context, slotName string) error

type TypeDecoder

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

func (*TypeDecoder) Decode

func (d *TypeDecoder) Decode(typeMap *pgtype.Map, data []byte) (any, error)