wal

Documentation

Index

• type FileWAL
  • func Open(logger log.Logger, path string, opts ...Option) (*FileWAL, error)
  • func (w *FileWAL) Close() error
  • func (w *FileWAL) FirstIndex() (uint64, error)
  • func (w *FileWAL) LastIndex() (uint64, error)
  • func (w *FileWAL) Log(tx uint64, record *walpb.Record) error
  • func (w *FileWAL) LogRecord(tx uint64, table string, record arrow.Record) error
  • func (w *FileWAL) Replay(tx uint64, handler ReplayHandlerFunc) (err error)
  • func (w *FileWAL) Reset(nextTx uint64) error
  • func (w *FileWAL) RunAsync()
  • func (w *FileWAL) Truncate(tx uint64) error
• type Metrics
• type NopWAL
  • func (w *NopWAL) Close() error
  • func (w *NopWAL) FirstIndex() (uint64, error)
  • func (w *NopWAL) LastIndex() (uint64, error)
  • func (w *NopWAL) Log(_ uint64, _ *walpb.Record) error
  • func (w *NopWAL) LogRecord(_ uint64, _ string, _ arrow.Record) error
  • func (w *NopWAL) Replay(_ uint64, _ ReplayHandlerFunc) error
  • func (w *NopWAL) Reset(_ uint64) error
  • func (w *NopWAL) Truncate(_ uint64) error
• type Option
  • func WithMetrics(m *Metrics) Option
  • func WithStoreMetrics(m *wal.Metrics) Option
  • func WithTestingCallbackWithDroppedLogsOnClose(cb func([]types.LogEntry)) Option
  • func WithTestingLogStoreWrapper(newLogStoreWrapper func(wal.LogStore) wal.LogStore) Option
  • func WithTestingLoopTicker(t Ticker) Option
• type ReplayHandlerFunc
• type Ticker

Types

type FileWAL

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

func Open

func Open(
	logger log.Logger,
	path string,
	opts ...Option,
) (*FileWAL, error)

func (*FileWAL) Close

func (w *FileWAL) Close() error

func (*FileWAL) FirstIndex

func (w *FileWAL) FirstIndex() (uint64, error)

func (*FileWAL) LastIndex

func (w *FileWAL) LastIndex() (uint64, error)

func (*FileWAL) Log

func (w *FileWAL) Log(tx uint64, record *walpb.Record) error

func (*FileWAL) LogRecord

func (w *FileWAL) LogRecord(tx uint64, table string, record arrow.Record) error

func (*FileWAL) Replay

func (w *FileWAL) Replay(tx uint64, handler ReplayHandlerFunc) (err error)

func (*FileWAL) Reset

func (w *FileWAL) Reset(nextTx uint64) error

func (*FileWAL) RunAsync

func (w *FileWAL) RunAsync()

func (*FileWAL) Truncate

func (w *FileWAL) Truncate(tx uint64) error

Truncate queues a truncation of the WAL at the given tx. Note that the truncation will be performed asynchronously. A nil error does not indicate a successful truncation.

type Metrics

type Metrics struct {
	FailedLogs            prometheus.Counter
	LastTruncationAt      prometheus.Gauge
	WalRepairs            prometheus.Counter
	WalRepairsLostRecords prometheus.Counter
	WalCloseTimeouts      prometheus.Counter
	WalQueueSize          prometheus.Gauge
}

type NopWAL

type NopWAL struct{}

func (*NopWAL) Close

func (w *NopWAL) Close() error

func (*NopWAL) FirstIndex

func (w *NopWAL) FirstIndex() (uint64, error)

func (*NopWAL) LastIndex

func (w *NopWAL) LastIndex() (uint64, error)

func (*NopWAL) Log

func (w *NopWAL) Log(_ uint64, _ *walpb.Record) error

func (*NopWAL) LogRecord

func (w *NopWAL) LogRecord(_ uint64, _ string, _ arrow.Record) error

func (*NopWAL) Replay

func (w *NopWAL) Replay(_ uint64, _ ReplayHandlerFunc) error

func (*NopWAL) Reset

func (w *NopWAL) Reset(_ uint64) error

func (*NopWAL) Truncate

func (w *NopWAL) Truncate(_ uint64) error

type Option

type Option func(*FileWAL)

func WithMetrics

func WithMetrics(m *Metrics) Option

func WithStoreMetrics

func WithStoreMetrics(m *wal.Metrics) Option

func WithTestingCallbackWithDroppedLogsOnClose

func WithTestingCallbackWithDroppedLogsOnClose(cb func([]types.LogEntry)) Option

WithTestingCallbackWithDroppedLogsOnClose is called when the WAL times out on close with all the entries that could not be written.

func WithTestingLogStoreWrapper

func WithTestingLogStoreWrapper(newLogStoreWrapper func(wal.LogStore) wal.LogStore) Option

func WithTestingLoopTicker

func WithTestingLoopTicker(t Ticker) Option

WithTestingLoopTicker allows the caller to force processing of pending WAL entries by providing a custom ticker implementation.

type ReplayHandlerFunc

type ReplayHandlerFunc func(tx uint64, record *walpb.Record) error

type Ticker

type Ticker interface {
	C() <-chan time.Time
	Stop()
}