config

Documentation

Index

• Constants
• Variables
• type CHConnConfig
• type ChColumn
• type ChTableName
  • func (ct ChTableName) IsEmpty() bool
  • func (ct ChTableName) String() string
  • func (ct *ChTableName) UnmarshalYAML(unmarshal func(interface{}) error) error
• type Column
• type ColumnProperty
• type Config
  • func New(filepath string) (*Config, error)
  • func (c Config) Print()
• type GzipComprLevel
  • func (gc GzipComprLevel) String() string
  • func (gc *GzipComprLevel) UnmarshalYAML(unmarshal func(interface{}) error) error
  • func (gc GzipComprLevel) UseCompression() bool
• type PgColumn
  • func (c PgColumn) IsIstore() bool
  • func (c PgColumn) IsTime() bool
• type PgTableName
  • func (tn PgTableName) KeyName() string
  • func (tn PgTableName) MarshalYAML() (interface{}, error)
  • func (tn PgTableName) NamespacedName() string
  • func (tn *PgTableName) Parse(val string) error
  • func (tn *PgTableName) ParseKey(key string) error
  • func (tn PgTableName) String() string
  • func (tn *PgTableName) UnmarshalYAML(unmarshal func(interface{}) error) error
• type Table
  • func (t *Table) UnmarshalYAML(unmarshal func(interface{}) error) error

Constants

const (
	ApplicationName = "pg2ch"

	DefaultRowIDColumn = "row_id"

	TableLSNKeyPrefix = "table_lsn_"
)

const (
	// CollapsingMergeTree represents CollapsingMergeTree table engine
	CollapsingMergeTree tableEngine = iota

	//ReplacingMergeTree represents ReplacingMergeTree table engine
	ReplacingMergeTree

	//MergeTree represents MergeTree table engine
	MergeTree
)

Variables

var (
	DefaultPostgresPort           uint16 = 5432
	DefaultInactivityMergeTimeout        = time.Minute
)

these variables could be changed in tests

Types

type CHConnConfig

type CHConnConfig struct {
	Host     string            `yaml:"host"`
	Port     uint32            `yaml:"port"`
	Database string            `yaml:"database"`
	User     string            `yaml:"username"`
	Password string            `yaml:"password"`
	Params   map[string]string `yaml:"params"`
}

type ChColumn

type ChColumn struct {
	Column
	Name string
}

ChColumn describes ClickHouse column

type ChTableName

type ChTableName struct {
	DatabaseName string
	TableName    string
}

func (ChTableName) IsEmpty

func (ct ChTableName) IsEmpty() bool

func (ChTableName) String

func (ct ChTableName) String() string

func (*ChTableName) UnmarshalYAML

func (ct *ChTableName) UnmarshalYAML(unmarshal func(interface{}) error) error

type Column

type Column struct {
	BaseType   string
	IsArray    bool
	IsNullable bool
	Ext        []int
}

type ColumnProperty

type ColumnProperty struct {
	IstoreKeysSuffix   string        `yaml:"istore_keys_suffix"`
	IstoreValuesSuffix string        `yaml:"istore_values_suffix"`
	Coalesce           coalesceValue `yaml:"coalesce"`
}

ColumnProperty describes column properties

type Config

type Config struct {
	ClickHouse             CHConnConfig           `yaml:"clickhouse"`
	Postgres               pgConnConfig           `yaml:"postgres"`
	Tables                 map[PgTableName]*Table `yaml:"tables"`
	InactivityFlushTimeout time.Duration          `yaml:"inactivity_flush_timeout"`
	PersStoragePath        string                 `yaml:"db_path"`
	PersStorageType        string                 `yaml:"db_type"`
	RedisBind              string                 `yaml:"redis_bind"`
	PprofBind              string                 `yaml:"pprof_bind"`
	SyncWorkers            int                    `yaml:"sync_workers"`
	LogLevel               zapcore.Level          `yaml:"loglevel"`
	GzipBufSize            int                    `yaml:"gzip_buffer_size"`
	GzipCompression        GzipComprLevel         `yaml:"gzip_compression"`
	PipeBufferSize         int64                  `yaml:"pipe_buffer_size"`
	CreateSlotMaxAttempts  int                    `yaml:"create_slot_max_attempts"`
}

Config contains config

func New

func New(filepath string) (*Config, error)

New instantiates config

func (Config) Print

func (c Config) Print()

type GzipComprLevel

type GzipComprLevel int

func (GzipComprLevel) String

func (gc GzipComprLevel) String() string

func (*GzipComprLevel) UnmarshalYAML

func (gc *GzipComprLevel) UnmarshalYAML(unmarshal func(interface{}) error) error

func (GzipComprLevel) UseCompression

func (gc GzipComprLevel) UseCompression() bool

type PgColumn

type PgColumn struct {
	Column
	PkCol int
}

func (PgColumn) IsIstore

func (c PgColumn) IsIstore() bool

func (PgColumn) IsTime

func (c PgColumn) IsTime() bool

type PgTableName

type PgTableName struct {
	SchemaName string
	TableName  string
}

PgTableName represents namespaced name

func (PgTableName) KeyName

func (tn PgTableName) KeyName() string

func (PgTableName) MarshalYAML

func (tn PgTableName) MarshalYAML() (interface{}, error)

func (PgTableName) NamespacedName

func (tn PgTableName) NamespacedName() string

func (*PgTableName) Parse

func (tn *PgTableName) Parse(val string) error

func (*PgTableName) ParseKey

func (tn *PgTableName) ParseKey(key string) error

func (PgTableName) String

func (tn PgTableName) String() string

func (*PgTableName) UnmarshalYAML

func (tn *PgTableName) UnmarshalYAML(unmarshal func(interface{}) error) error

type Table

type Table struct {
	ChMainTable          ChTableName                `yaml:"main_table"`
	ChSyncAuxTable       ChTableName                `yaml:"sync_aux_table"`
	IsDeletedColumn      string                     `yaml:"is_deleted_column"`
	SignColumn           string                     `yaml:"sign_column"`
	RowIDColumnName      string                     `yaml:"row_id_column"`
	TableNameColumnName  string                     `yaml:"table_name_column_name"`
	Engine               tableEngine                `yaml:"engine"`
	BufferSize           int                        `yaml:"max_buffer_length"`
	InitSyncSkip         bool                       `yaml:"init_sync_skip"`
	InitSyncSkipTruncate bool                       `yaml:"init_sync_skip_truncate"`
	Columns              map[string]string          `yaml:"columns"`
	ColumnProperties     map[string]*ColumnProperty `yaml:"column_properties"`
	LsnColumnName        string                     `yaml:"lsn_column_name"`

	PgOID         dbtypes.OID          `yaml:"-"`
	PgTableName   PgTableName          `yaml:"-"`
	TupleColumns  []message.Column     `yaml:"-"` // columns in the order they are in the table
	PgColumns     map[string]*PgColumn `yaml:"-"` // map of pg column structs
	ColumnMapping map[string]ChColumn  `yaml:"-"` // mapping pg column -> ch column
}

Table contains information about the table

func (*Table) UnmarshalYAML

func (t *Table) UnmarshalYAML(unmarshal func(interface{}) error) error

UnmarshalYAML ...