postgres

Documentation

Index

• Variables
• type Column
  • func (col *Column) MarshalJSON() ([]byte, error)
  • func (col *Column) Omit() bool
• type Database
  • func New(stream *StreamPipe, logger *zap.Logger) *Database
  • func (db *Database) Close(ctx context.Context) error
  • func (db *Database) Commit(ctx context.Context) error
  • func (db *Database) Connect(ctx context.Context) error
  • func (db *Database) CreateReplicationSlot(ctx context.Context)
  • func (db *Database) Discover(ctx context.Context) error
  • func (db *Database) DropReplicationSlot(ctx context.Context)
  • func (db *Database) HandleLogical(ctx context.Context, lsn pglogrepl.LSN, msg pglogrepl.Message) error
  • func (db *Database) PrintSatus()
  • func (db *Database) RegisterSlotLagMetric(ctx context.Context)
  • func (db *Database) Reindex(ctx context.Context) error
  • func (db *Database) StartReplication(ctx context.Context, at pglogrepl.LSN) error
  • func (db *Database) Tx(ctx context.Context) error
• type DecoderValue
• type Doc
• type Document
  • func (d Document) NDJSON() [][]byte
• type ESAction
• type Inline
• type Position
  • func (p Position) LSN() pglogrepl.LSN
  • func (d Position) NDJSON() [][]byte
• type Schema
• type StreamPipe
  • func NewStreamPipe(ctx context.Context) *StreamPipe
  • func (p *StreamPipe) CommitPosition(pos pglogrepl.LSN)
  • func (p *StreamPipe) Next(ctx context.Context) (Doc, error)
  • func (p *StreamPipe) Position() pglogrepl.LSN
• type Table
  • func (t *Table) Column(name string) (col *Column)
  • func (t *Table) CopyAll(ctx context.Context, conn *pgconn.PgConn) error
  • func (t *Table) EncodeUpdateRowJSON() ([]byte, error)
  • func (t *Table) MarshalJSON() ([]byte, error)
  • func (t *Table) Schema() *Schema
  • func (t *Table) SetRelationID(id uint32)

Variables

var (
	// ErrZeroType Postgres type OID can not be zero
	ErrZeroType = errors.New("postgres type OID can't be zero")
	// ErrTypeNotFound can not find type OID among discovered types
	ErrTypeNotFound = errors.New("can not find discovered type")
)

var (
	ErrTimeout  = errors.New("timeout")
	ErrChClosed = errors.New("document channel is closed")
)

var (
	// ErrColumnOutOfRange means that received result tuple is smaller than expected column position
	ErrColumnOutOfRange = errors.New("column out of range")
)

var (
	// ErrUnknownType type discovery failed
	ErrUnknownType = errors.New("unknown type")
)

Types

type Column

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

func (*Column) MarshalJSON

func (col *Column) MarshalJSON() ([]byte, error)

func (*Column) Omit

func (col *Column) Omit() bool

type Database

type Database struct {
	SlotName string

	Publication    string
	StandbyTimeout time.Duration
	// contains filtered or unexported fields
}

func New

func New(stream *StreamPipe, logger *zap.Logger) *Database

func (*Database) Close

func (db *Database) Close(ctx context.Context) error

func (*Database) Commit

func (db *Database) Commit(ctx context.Context) error

func (*Database) Connect

func (db *Database) Connect(ctx context.Context) error

Connect makes two connections to DB for discovery and replication

func (*Database) CreateReplicationSlot

func (db *Database) CreateReplicationSlot(ctx context.Context)

CreateReplicationSlot creates a replication slot at current position and uses newly created snapshot in current transaction. For the sake of consistency it's important to use this method and initial data copying within transaction db.Tx(ctx) db.CreateReplicationSlot(ctx) ... copy data db.Commit(ctx)

func (*Database) Discover

func (db *Database) Discover(ctx context.Context) error

Discover uses Postgres publication and table or column comments to generate replication config Only tables explosed via Publication will be considered for exporting to ES.

func (*Database) DropReplicationSlot

func (db *Database) DropReplicationSlot(ctx context.Context)

func (*Database) HandleLogical

func (db *Database) HandleLogical(ctx context.Context, lsn pglogrepl.LSN, msg pglogrepl.Message) error

func (*Database) PrintSatus

func (db *Database) PrintSatus()

PrintStatus prints some debug information TODO: remove

func (*Database) RegisterSlotLagMetric

func (db *Database) RegisterSlotLagMetric(ctx context.Context)

func (*Database) Reindex

func (db *Database) Reindex(ctx context.Context) error

Select everything and push (streaming) it into elasticsearch.

func (*Database) StartReplication

func (db *Database) StartReplication(ctx context.Context, at pglogrepl.LSN) error

StartReplication switches replConn into `CopyBoth` mode and starts streaming. decoded logical messages are passed for further processing, while status updates happens here. during the streaming, replConn is locked and can not be used for anything else.

XXX: PrimaryKeepaliveMessage.ServerWALEnd is actually the location up to which the WAL is sent See: https://stackoverflow.com/questions/71016200/proper-standby-status-update-in-streaming-replication-protocol

func (*Database) Tx

func (db *Database) Tx(ctx context.Context) error

type DecoderValue

type DecoderValue interface {
	pgtype.TextDecoder
	pgtype.BinaryDecoder // Not all types support BinaryDecoder.
	pgtype.Value
}

DecoderValue can decode value and return pgtype object which implies Value interface.

type Doc

type Doc interface {
	NDJSON() [][]byte
	LSN() pglogrepl.LSN
}

type Document

type Document struct {
	Position
	Meta []byte // Op type, index and document id
	Data []byte // document content or script
}

Document represents one single operation in bulk request.

func (Document) NDJSON

func (d Document) NDJSON() [][]byte

type ESAction

type ESAction string

const (
	ESInsert ESAction = "insert"
	ESUpdate ESAction = "update"
	ESDelete ESAction = "delete"
	ESIndex  ESAction = "index" // Upsert
)

type Inline

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

Inline defines table like abstraction for inligning into ES doc

type Position

type Position pglogrepl.LSN

func (Position) LSN

func (p Position) LSN() pglogrepl.LSN

func (Position) NDJSON

func (d Position) NDJSON() [][]byte

type Schema

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

Schema describes Postgres schema/namespace

type StreamPipe

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

StreamPipe connects decoded postgres stream with elasticsearch client. There should not be any other interaction.

func NewStreamPipe

func NewStreamPipe(ctx context.Context) *StreamPipe

func (*StreamPipe) CommitPosition

func (p *StreamPipe) CommitPosition(pos pglogrepl.LSN)

func (*StreamPipe) Next

func (p *StreamPipe) Next(ctx context.Context) (Doc, error)

func (*StreamPipe) Position

func (p *StreamPipe) Position() pglogrepl.LSN

type Table

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

func (*Table) Column

func (t *Table) Column(name string) (col *Column)

Column gets (existing or default) column config.

func (*Table) CopyAll

func (t *Table) CopyAll(ctx context.Context, conn *pgconn.PgConn) error

Selects all rown from table, and populates results into Database.results chanel. Copy existing data snapshoted by slot creation, using simple protocol.

func (*Table) EncodeUpdateRowJSON

func (t *Table) EncodeUpdateRowJSON() ([]byte, error)

EncodeUpdateRowJSON wraps row into `{"doc": ... }` object, required by ElasticSearch bulk request syntax for update queries

func (*Table) MarshalJSON

func (t *Table) MarshalJSON() ([]byte, error)

func (*Table) Schema

func (t *Table) Schema() *Schema

Table returns column's owner table

func (*Table) SetRelationID

func (t *Table) SetRelationID(id uint32)

TODO: take RelID set cache out of this tree config