logrepl

Documentation

Index

• Constants
• type CaptureMode
• type LogicalReplicator
  • func (r *LogicalReplicator) Run()
• type Node
• type Privacy
• type Publisher
• type QueryBuilder
  • func NewQueryBuilder(pgConnectionString string, schema *Schema) *QueryBuilder
  • func (q *QueryBuilder) GetRows(context context.Context, namespace string, table string, columns ...string) ([]*pgcdcmodels.Row, error)
  • func (q *QueryBuilder) JoinChildren(hasQi bool, namespace string, table string, node Node) string
  • func (q *QueryBuilder) ListChildColumns(namespace string, table string, node Node) []string
  • func (q *QueryBuilder) ResolveRelationships(context context.Context, data pgcdcmodels.Row) error
  • func (q *QueryBuilder) Select(table string, columns ...string) (string, bool)
  • func (q *QueryBuilder) SelectRowIncludeReferences(table string, id int64, columns ...string) string
  • func (q *QueryBuilder) SelectWithRelationships(namespace string, table string, columns ...string) string
• type Relationship
• type ReplicationMode
  • func (m ReplicationMode) String() string
• type Schema
  • func NewSchema(filePath string) *Schema
  • func (q *Schema) GetPrimaryKey(data pgcdcmodels.Row) pgcdcmodels.Field
• type Transaction
  • func NewTransaction(xid uint32) *Transaction
  • func (t *Transaction) DmlCommandQueue() *list.List
• type Transformer
  • func NewTransformer(schema *Schema) *Transformer
  • func (t *Transformer) Transform(table string, node Node, data pgcdcmodels.Row) error

Constants

const (
	TRANSFORM_RENAME string = "rename"
	TRANSFORM_CONCAT string = "concat"
)

Types

type CaptureMode

type CaptureMode string

const (
	CAPTURE_NONE CaptureMode = "none"
	CAPTURE_ALL  CaptureMode = "all"
)

type LogicalReplicator

type LogicalReplicator struct {
	Publishers            []Publisher
	OutputPlugin          string
	ConnectionString      string
	PublicationName       string
	SlotName              string
	StandbyMessageTimeout time.Duration
	Schema                *Schema

	Mode ReplicationMode
	// contains filtered or unexported fields
}

func (*LogicalReplicator) Run

func (r *LogicalReplicator) Run()

type Node

type Node struct {
	Relationship Relationship
	Namespace    string
	PrimaryKey   string `yaml:"pk"`
	Columns      []string
	Privacy      Privacy
	Transform    map[string]interface{}
	Children     map[string]Node
	Parent       *Node
	Capture      CaptureMode
}

type Privacy

type Privacy struct {
	Enabled          bool
	Namespace        string
	QuasiIdentifiers map[string]struct{} `yaml:"qi"`
}

This feature makes a LOT of assumptions on the database, certain rules needs to be followed

type Publisher

type Publisher interface {
	String() string
	Init(schema *Schema) error
	FullyReplicateTable(rows []*pgcdcmodels.Row, totalTables int) error
	OnBegin(xid uint32) error
	OnInsert(row pgcdcmodels.Row) error
	OnUpdate(row pgcdcmodels.Row) error
	OnDelete(row pgcdcmodels.Row) error
	OnCommit() error
}

type QueryBuilder

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

func NewQueryBuilder

func NewQueryBuilder(pgConnectionString string, schema *Schema) *QueryBuilder

func (*QueryBuilder) GetRows

func (q *QueryBuilder) GetRows(
	context context.Context,
	namespace string,
	table string,
	columns ...string,
) ([]*pgcdcmodels.Row, error)

func (*QueryBuilder) JoinChildren

func (q *QueryBuilder) JoinChildren(hasQi bool, namespace string, table string, node Node) string

func (*QueryBuilder) ListChildColumns

func (q *QueryBuilder) ListChildColumns(namespace string, table string, node Node) []string

func (*QueryBuilder) ResolveRelationships

func (q *QueryBuilder) ResolveRelationships(
	context context.Context,
	data pgcdcmodels.Row,
) error

func (*QueryBuilder) Select

func (q *QueryBuilder) Select(table string, columns ...string) (string, bool)

func (*QueryBuilder) SelectRowIncludeReferences

func (q *QueryBuilder) SelectRowIncludeReferences(table string, id int64, columns ...string) string

func (*QueryBuilder) SelectWithRelationships

func (q *QueryBuilder) SelectWithRelationships(namespace string, table string, columns ...string) string

type Relationship

type Relationship struct {
	Type       string
	ForeignKey struct {
		Child  string
		Parent string
	} `yaml:"fk"`
}

type ReplicationMode

type ReplicationMode uint8

const (
	STREAM_MODE ReplicationMode = iota
	POPULATE_MODE
)

func (ReplicationMode) String

func (m ReplicationMode) String() string

type Schema

type Schema struct {
	Nodes map[string]Node
}

func NewSchema

func NewSchema(filePath string) *Schema

func (*Schema) GetPrimaryKey

func (q *Schema) GetPrimaryKey(data pgcdcmodels.Row) pgcdcmodels.Field

type Transaction

type Transaction struct {
	Xid uint32
	// contains filtered or unexported fields
}

func NewTransaction

func NewTransaction(xid uint32) *Transaction

func (*Transaction) DmlCommandQueue

func (t *Transaction) DmlCommandQueue() *list.List

type Transformer

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

func NewTransformer

func NewTransformer(schema *Schema) *Transformer

func (*Transformer) Transform

func (t *Transformer) Transform(table string, node Node, data pgcdcmodels.Row) error