buffer

Documentation

Overview

Package buffer provides an in-memory event buffer for BYOS mode. It stores recent binlog events so the agent can answer resolve_pk and recover commands without hitting S3 for recent data. The buffer is safe for concurrent use: one goroutine (stream) writes via Insert, while another (agent handler) reads via Fetch and ResolvePK.

Index

• func WriteParquet(rows []query.ResultRow, outputPath, compression string) (int64, error)
• type Buffer
  • func New(cfg Config) *Buffer
  • func (b *Buffer) ApproxBytes() int64
  • func (b *Buffer) Evict() int
  • func (b *Buffer) Fetch(_ context.Context, opts query.Options) []query.ResultRow
  • func (b *Buffer) Insert(events []event.Event)
  • func (b *Buffer) Len() int
  • func (b *Buffer) ResolvePK(hash, schema, table string) (string, bool)
  • func (b *Buffer) SizeEvictions() int64
  • func (b *Buffer) Snapshot() []query.ResultRow
• type Config

Functions

func WriteParquet

func WriteParquet(rows []query.ResultRow, outputPath, compression string) (int64, error)

WriteParquet writes result rows to a Parquet file at outputPath using the same 15-column schema as archive.ArchivePartition. The output files are compatible with parquetquery.Fetch for querying.

Types

type Buffer

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

Buffer is a thread-safe, in-memory event store for recent binlog events.

func New

func New(cfg Config) *Buffer

New creates a Buffer with the given configuration. Zero values for MaxEvents and MaxBytes mean unlimited (no cap).

func (*Buffer) ApproxBytes

func (b *Buffer) ApproxBytes() int64

ApproxBytes returns the approximate in-memory byte usage of all buffered events.

func (*Buffer) Evict

func (b *Buffer) Evict() int

Evict removes events older than maxAge and returns the count removed.

func (*Buffer) Fetch

func (b *Buffer) Fetch(_ context.Context, opts query.Options) []query.ResultRow

Fetch returns events matching the query options. The returned rows are compatible with query.MergeResults for merging with MySQL/Parquet results.

func (*Buffer) Insert

func (b *Buffer) Insert(events []event.Event)

Insert converts parser events to result rows and appends them to the buffer.

func (*Buffer) Len

func (b *Buffer) Len() int

Len returns the number of events in the buffer.

func (*Buffer) ResolvePK

func (b *Buffer) ResolvePK(hash, schema, table string) (string, bool)

ResolvePK looks up pk_values for the given pk_hash, schema, and table. Returns the pk_values string and true if found.

Deliberately NOT wired with the #1137 CanonicalPKValues compat fallback: the buffer is in-memory and process-lifetime, fed only by the live (post-#1132) parser via Insert, and is never rehydrated from Parquet — so no entry can carry the pre-fix raw spelling of a binary PK. The durable pre-fix spellings live in payload/archive Parquet, which the agent handler's archive path covers.

func (*Buffer) SizeEvictions

func (b *Buffer) SizeEvictions() int64

SizeEvictions returns the cumulative count of events evicted due to size caps (maxEvents or maxBytes) since the buffer was created.

func (*Buffer) Snapshot

func (b *Buffer) Snapshot() []query.ResultRow

Snapshot returns a copy of all current events for archival purposes (e.g. writing to Parquet). The caller gets its own slice and can process it without holding any lock.

type Config

type Config struct {
	MaxAge    time.Duration // age-based eviction threshold
	MaxEvents int           // max event count; 0 = unlimited
	MaxBytes  int64         // max approximate byte usage; 0 = unlimited
	Logger    *slog.Logger
}

Config controls buffer capacity and eviction behavior.