data

Documentation

Index

• Constants
• func ConvertSQLServerUUID(sqlBytes []byte) []byte
• func GeneratePKCS8StringSupress(key *rsa.PrivateKey) string
• func GetMVRVars(prefix string) map[string]string
• func ParseAndExecuteTemplate(data []byte, config *StreamConfig) ([]byte, error)
• func ParsePEMPrivateKey(pemKey string) (*rsa.PrivateKey, error)
• func SortKeys(params []string)
• func TypeAlias(columnType string) string
• type Batch
• type BatchWriter
• type Column
  • func OverrideColumns(original []Column, overrides []Column) []Column
• type DBExec
• type DBReaderConn
• type DataStream
  • func (ds *DataStream) BatchesToWriter(ctx context.Context, writer BatchWriter) error
• type DataWriter
• type MultiStreamConfig
• type Param
• type RawStreamConfig
• type StreamConfig
  • func BuildConfig(data []byte, cliArgs *StreamConfig) (*StreamConfig, error)
  • func NewStreamConfigFromYaml(data []byte) (*StreamConfig, error)
  • func (sc *StreamConfig) GetBatchCount() int
  • func (sc *StreamConfig) GetBatchSize() int
  • func (sc *StreamConfig) OverrideValues(cliArgs *StreamConfig)
  • func (sc *StreamConfig) Validate() error
• type TemplateString
  • func (t *TemplateString) UnmarshalYAML(unmarshal func(interface{}) error) error

Constants

const (
	RFC3339MicroNoTZ = "2006-01-02T15:04:05.999999"
	RFC3339MicroTZ   = "2006-01-02T15:04:05.999999Z07:00"
)

Functions

func ConvertSQLServerUUID

func ConvertSQLServerUUID(sqlBytes []byte) []byte

func GeneratePKCS8StringSupress

func GeneratePKCS8StringSupress(key *rsa.PrivateKey) string

func GetMVRVars

func GetMVRVars(prefix string) map[string]string

func ParseAndExecuteTemplate

func ParseAndExecuteTemplate(data []byte, config *StreamConfig) ([]byte, error)

func ParsePEMPrivateKey

func ParsePEMPrivateKey(pemKey string) (*rsa.PrivateKey, error)

func SortKeys

func SortKeys(params []string)

func TypeAlias

func TypeAlias(columnType string) string

TypeAlias corrects for the various ways that Postgres types are named the goal is to only have the name column from this table https://www.postgresql.org/docs/current/datatype.html#DATATYPE-TABLE

Types

type Batch

type Batch struct {
	Rows [][]any
}

type BatchWriter

type BatchWriter interface {
	WriteBatch(batch Batch) error
}

type Column

type Column struct {
	Name         string `json:"name" yaml:"name"`
	DatabaseType string `json:"database_type" yaml:"database_type"`
	Type         string `json:"type" yaml:"type"`
	Length       int64  `json:"length,omitempty" yaml:"length,omitempty"`
	Nullable     bool   `json:"nullable" yaml:"nullable"`
	Position     int    `json:"position" yaml:"position"`
	Scale        int64  `json:"scale,omitempty" yaml:"scale,omitempty"`
	Precision    int64  `json:"precision,omitempty" yaml:"precision,omitempty"`
}

func OverrideColumns

func OverrideColumns(original []Column, overrides []Column) []Column

type DBExec

type DBExec interface {
	ExecuteCommand(ctx context.Context, sql_command string) (string, error)
}

type DBReaderConn

type DBReaderConn interface {
	CreateDataStream(ctx context.Context, cs *url.URL, config *StreamConfig) (*DataStream, error)
	ExecuteDataStream(ctx context.Context, ds *DataStream, config *StreamConfig) error
	Close() error
}

type DataStream

type DataStream struct {
	TotalRows   int
	Mux         sync.Mutex
	BatchChan   chan Batch
	BatchSize   int
	Columns     []Column
	DestColumns []Column
}

func (*DataStream) BatchesToWriter

func (ds *DataStream) BatchesToWriter(ctx context.Context, writer BatchWriter) error

type DataWriter

type DataWriter interface {
	CreateBatchWriter() BatchWriter
	Flush() error
	Close() error
}

type MultiStreamConfig

type MultiStreamConfig struct {
	StreamConfig `json:",inline" yaml:",inline"`
	Tables       []StreamConfig `json:"tables" yaml:"tables"`
}

type Param

type Param struct {
	Value string `json:"value" yaml:"value"`
	Type  string `json:"type" yaml:"type"`
}

type RawStreamConfig

type RawStreamConfig struct {
	Filename  TemplateString `yaml:"filename"`
	BatchSize int            `yaml:"batch_size"`
}

type StreamConfig

type StreamConfig struct {
	StreamName  string           `json:"stream_name,omitempty" yaml:"stream_name,omitempty"`
	Filename    string           `json:"filename,omitempty" yaml:"filename,omitempty"`
	Format      string           `json:"format,omitempty" yaml:"format,omitempty"`
	SQL         string           `json:"sql,omitempty" yaml:"sql,omitempty"`
	Compression string           `json:"compression,omitempty" yaml:"compression,omitempty"`
	Columns     []Column         `json:"columns,omitempty" yaml:"columns,omitempty"`
	Params      map[string]Param `json:"params,omitempty" yaml:"params,omitempty"`
	ParamKeys   []string
	BatchSize   int `json:"batch_size,omitempty" yaml:"batch_size,omitempty"`
	BatchCount  int `json:"batch_count,omitempty" yaml:"batch_count,omitempty"`
}

func BuildConfig

func BuildConfig(data []byte, cliArgs *StreamConfig) (*StreamConfig, error)

func NewStreamConfigFromYaml

func NewStreamConfigFromYaml(data []byte) (*StreamConfig, error)

func (*StreamConfig) GetBatchCount

func (sc *StreamConfig) GetBatchCount() int

func (*StreamConfig) GetBatchSize

func (sc *StreamConfig) GetBatchSize() int

func (*StreamConfig) OverrideValues

func (sc *StreamConfig) OverrideValues(cliArgs *StreamConfig)

func (*StreamConfig) Validate

func (sc *StreamConfig) Validate() error

type TemplateString

type TemplateString struct {
	Raw string
}

func (*TemplateString) UnmarshalYAML

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