pglogicalstream

Apr 26, 2024

Changes in this version

+ const DeleteMessageTupleTypeKey

+ const DeleteMessageTupleTypeOld

+ const PrimaryKeepaliveMessageByteID

+ const StandbyStatusUpdateByteID

+ const TruncateOptionCascade

+ const TruncateOptionRestartIdentity

+ const TupleDataTypeBinary

+ const TupleDataTypeNull

+ const TupleDataTypeText

+ const TupleDataTypeToast

+ const UpdateMessageTupleTypeKey

+ const UpdateMessageTupleTypeNew

+ const UpdateMessageTupleTypeNone

+ const UpdateMessageTupleTypeOld

+ const XLogDataByteID

+ func DropReplicationSlot(ctx context.Context, conn *pgconn.PgConn, slotName string, ...) error

+ func NextTableSpace(ctx context.Context, conn *pgconn.PgConn) (err error)

+ func SendStandbyStatusUpdate(_ context.Context, conn *pgconn.PgConn, ssu StandbyStatusUpdate) error

+ func StartReplication(ctx context.Context, conn *pgconn.PgConn, slotName string, startLSN LSN, ...) error

+ type BaseBackupOptions struct

+ Fast bool

+ Label string

+ MaxRate int32

+ NoVerifyChecksums bool

+ NoWait bool

+ Progress bool

+ TablespaceMap bool

+ WAL bool

+ type BaseBackupResult struct

+ LSN LSN

+ Tablespaces []BaseBackupTablespace

+ TimelineID int32

+ func FinishBaseBackup(ctx context.Context, conn *pgconn.PgConn) (result BaseBackupResult, err error)

+ func StartBaseBackup(ctx context.Context, conn *pgconn.PgConn, options BaseBackupOptions) (result BaseBackupResult, err error)

+ type BaseBackupTablespace struct

+ Location string

+ OID int32

+ Size int8

+ type BeginMessage struct

+ CommitTime time.Time

+ FinalLSN LSN

+ Xid uint32

+ func (m *BeginMessage) Decode(src []byte) error

+ func (m *BeginMessage) SetType(t MessageType)

+ func (m *BeginMessage) Type() MessageType

+ type ChangeFilter struct

+ func NewChangeFilter(tableSchemas []schemas.DataTableSchema, schema string) ChangeFilter

+ func (c ChangeFilter) FilterChange(lsn string, change []byte, OnFiltered Filtered)

+ type CommitMessage struct

+ CommitLSN LSN

+ CommitTime time.Time

+ Flags uint8

+ TransactionEndLSN LSN

+ func (m *CommitMessage) Decode(src []byte) error

+ func (m *CommitMessage) SetType(t MessageType)

+ func (m *CommitMessage) Type() MessageType

+ type CopyDoneResult struct

+ LSN LSN

+ Timeline int32

+ func SendStandbyCopyDone(_ context.Context, conn *pgconn.PgConn) (cdr *CopyDoneResult, err error)

+ type CreateReplicationSlotOptions struct

+ Mode ReplicationMode

+ SnapshotAction string

+ Temporary bool

+ type CreateReplicationSlotResult struct

+ ConsistentPoint string

+ OutputPlugin string

+ SlotName string

+ SnapshotName string

+ func CreateReplicationSlot(ctx context.Context, conn *pgconn.PgConn, slotName string, outputPlugin string, ...) (CreateReplicationSlotResult, error)

+ func ParseCreateReplicationSlot(mrr *pgconn.MultiResultReader) (CreateReplicationSlotResult, error)

+ type DeleteMessage struct

+ OldTuple *TupleData

+ OldTupleType uint8

+ RelationID uint32

+ func (m *DeleteMessage) Decode(src []byte) (err error)

+ func (m *DeleteMessage) SetType(t MessageType)

+ func (m *DeleteMessage) Type() MessageType

+ type DropReplicationSlotOptions struct

+ Wait bool

+ type Filtered func(change Wal2JsonChanges)

+ type IdentifySystemResult struct

+ DBName string

+ SystemID string

+ Timeline int32

+ XLogPos LSN

+ func IdentifySystem(ctx context.Context, conn *pgconn.PgConn) (IdentifySystemResult, error)

+ func ParseIdentifySystem(mrr *pgconn.MultiResultReader) (IdentifySystemResult, error)

+ type InsertMessage struct

+ RelationID uint32

+ Tuple *TupleData

+ func (m *InsertMessage) Decode(src []byte) error

+ func (m *InsertMessage) SetType(t MessageType)

+ func (m *InsertMessage) Type() MessageType

+ type LSN uint64

+ func ParseLSN(s string) (LSN, error)

+ func (lsn *LSN) Scan(src interface{}) error

+ func (lsn LSN) String() string

+ func (lsn LSN) Value() (driver.Value, error)

+ type LogicalDecodingMessage struct

+ Content []byte

+ LSN LSN

+ Prefix string

+ Transactional bool

+ func (m *LogicalDecodingMessage) Decode(src []byte) (err error)

+ func (m *LogicalDecodingMessage) SetType(t MessageType)

+ func (m *LogicalDecodingMessage) Type() MessageType

+ type Message interface

+ Type func() MessageType

+ func Parse(data []byte) (m Message, err error)

+ type MessageDecoder interface

+ Decode func([]byte) error

+ type MessageType uint8

+ const MessageTypeBegin

+ const MessageTypeCommit

+ const MessageTypeDelete

+ const MessageTypeInsert

+ const MessageTypeMessage

+ const MessageTypeOrigin

+ const MessageTypeRelation

+ const MessageTypeStreamAbort

+ const MessageTypeStreamCommit

+ const MessageTypeStreamStart

+ const MessageTypeStreamStop

+ const MessageTypeTruncate

+ const MessageTypeType

+ const MessageTypeUpdate

+ func (t MessageType) String() string

+ type OriginMessage struct

+ CommitLSN LSN

+ Name string

+ func (m *OriginMessage) Decode(src []byte) error

+ func (m *OriginMessage) SetType(t MessageType)

+ func (m *OriginMessage) Type() MessageType

+ type PrimaryKeepaliveMessage struct

+ ReplyRequested bool

+ ServerTime time.Time

+ ServerWALEnd LSN

+ func ParsePrimaryKeepaliveMessage(buf []byte) (PrimaryKeepaliveMessage, error)

+ type RelationMessage struct

+ ColumnNum uint16

+ Columns []*RelationMessageColumn

+ Namespace string

+ RelationID uint32

+ RelationName string

+ ReplicaIdentity uint8

+ func (m *RelationMessage) Decode(src []byte) error

+ func (m *RelationMessage) SetType(t MessageType)

+ func (m *RelationMessage) Type() MessageType

+ type RelationMessageColumn struct

+ DataType uint32

+ Flags uint8

+ Name string

+ TypeModifier int32

+ type ReplicationMode int

+ const LogicalReplication

+ const PhysicalReplication

+ func (mode ReplicationMode) String() string

+ type Snapshotter struct

+ func NewSnapshotter(dbConf pgconn.Config, snapshotName string) (*Snapshotter, error)

+ func (s *Snapshotter) CalculateBatchSize(safetyFactor float64, availableMemory uint64, estimatedRowSize uint64) int

+ func (s *Snapshotter) CloseConn() error

+ func (s *Snapshotter) FindAvgRowSize(table string) sql.NullInt64

+ func (s *Snapshotter) Prepare() error

+ func (s *Snapshotter) QuerySnapshotData(table string, columns []string, pk string, limit, offset int) (rows pgx.Rows, err error)

+ func (s *Snapshotter) ReleaseSnapshot() error

+ type StandbyStatusUpdate struct

+ ClientTime time.Time

+ ReplyRequested bool

+ WALApplyPosition LSN

+ WALFlushPosition LSN

+ WALWritePosition LSN

+ type StartReplicationOptions struct

+ Mode ReplicationMode

+ PluginArgs []string

+ Timeline int32

+ type TimelineHistoryResult struct

+ Content []byte

+ FileName string

+ func ParseTimelineHistory(mrr *pgconn.MultiResultReader) (TimelineHistoryResult, error)

+ func TimelineHistory(ctx context.Context, conn *pgconn.PgConn, timeline int32) (TimelineHistoryResult, error)

+ type TruncateMessage struct

+ Option uint8

+ RelationIDs []uint32

+ RelationNum uint32

+ func (m *TruncateMessage) Decode(src []byte) (err error)

+ func (m *TruncateMessage) SetType(t MessageType)

+ func (m *TruncateMessage) Type() MessageType

+ type TupleData struct

+ ColumnNum uint16

+ Columns []*TupleDataColumn

+ func (m *TupleData) Decode(src []byte) (int, error)

+ func (m *TupleData) SetType(t MessageType)

+ func (m *TupleData) Type() MessageType

+ type TupleDataColumn struct

+ Data []byte

+ DataType uint8

+ Length uint32

+ func (c *TupleDataColumn) Int64() (int64, error)

+ type TypeMessage struct

+ DataType uint32

+ Name string

+ Namespace string

+ func (m *TypeMessage) Decode(src []byte) error

+ func (m *TypeMessage) SetType(t MessageType)

+ func (m *TypeMessage) Type() MessageType

+ type UpdateMessage struct

+ NewTuple *TupleData

+ OldTuple *TupleData

+ OldTupleType uint8

+ RelationID uint32

+ func (m *UpdateMessage) Decode(src []byte) (err error)

+ func (m *UpdateMessage) SetType(t MessageType)

+ func (m *UpdateMessage) Type() MessageType

+ type WallMessage struct

+ Change []struct{ ... }

+ type XLogData struct

+ ServerTime time.Time

+ ServerWALEnd LSN

+ WALData []byte

+ WALStart LSN

+ func ParseXLogData(buf []byte) (XLogData, error)

Apr 26, 2024

Changes in this version

+ type Wal2JsonChange struct

+ Kind string

+ Row arrow.Record

+ Schema string

+ Table string

+ type Wal2JsonChanges struct

+ Changes []Wal2JsonChange

+ Lsn string

Nov 10, 2023

Changes in this version

type Stream

+ func (s *Stream) AckLSN(lsn string)

Oct 27, 2023

Changes in this version

+ type DbSchemaColumn struct

+ DatabrewType string

+ Name string

+ NativeConnectorType string

+ Nullable bool

+ Pk bool

Oct 27, 2023

Changes in this version

type Config

+ DbTablesSchema []DbTablesSchema

+ type DbTablesSchema struct

+ Columns []struct{ ... }

+ Table string

Oct 25, 2023

Changes in this version

type Config

+ BatchSize int

Oct 25, 2023

Changes in this version

type Config

+ SnapshotMemorySafetyFactor float64

Oct 24, 2023

Changes in this version

+ type CheckPointer interface

+ GetCheckPoint func(slot string) string

+ SetCheckPoint func(lsn, slot string) error

+ type Config struct

+ DbHost string

+ DbName string

+ DbPassword string

+ DbPort int

+ DbSchema string

+ DbTables []string

+ DbUser string

+ ReplicationSlotName string

+ SeparateChanges bool

+ StreamOldData bool

+ TlsVerify TlsVerify

+ type OnMessage = func(message []byte)

+ type Stream struct

+ func NewPgStream(config Config, checkpointer CheckPointer) (*Stream, error)

+ func (s *Stream) LrMessageC() chan []byte

+ func (s *Stream) OnMessage(callback OnMessage)

+ func (s *Stream) SnapshotMessageC() chan []byte

+ func (s *Stream) Stop() error

+ type TlsVerify string

+ const TlsNoVerify

+ const TlsRequireVerify