data

Documentation

Index

• Constants
• func ConvertSQLServerUUID(sqlBytes []byte) []byte
• func GetMVRVars(prefix string) map[string]any
• type Batch
• type Column
  • func OverrideColumns(original []Column, overrides []Column) []Column
• type DBReaderConn
• type DataStream
  • func (ds *DataStream) BatchesToWriter(wg *sync.WaitGroup, writer DataWriter)
• type DataWriter
• type StreamConfig
  • func NewStreamConfigFromYaml(data []byte) (*StreamConfig, error)

Constants

const (
	RFC3339NanoNoTZ = "2006-01-02T15:04:05.999999999"
)

Functions

func ConvertSQLServerUUID

func ConvertSQLServerUUID(sqlBytes []byte) []byte

func GetMVRVars

func GetMVRVars(prefix string) map[string]any

Types

type Batch

type Batch struct {
	Rows [][]any
}

type Column

type Column struct {
	Name         string `json:"name" yaml:"name"`
	DatabaseType string `json:"database_type" yaml:"database_type"`
	Length       int64  `json:"length,omitempty" yaml:"length,omitempty"`
	Nullable     bool   `json:"nullable" yaml:"nullable"`
	ScanType     string `json:"scan_type" yaml:"scan_type"`
	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 DBReaderConn

type DBReaderConn interface {
	CreateDataStream(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
	IsSqlServer bool
}

func (*DataStream) BatchesToWriter

func (ds *DataStream) BatchesToWriter(wg *sync.WaitGroup, writer DataWriter)

type DataWriter

type DataWriter interface {
	WriteRow(row []any) error
	Flush() error
	Close() error
}

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]interface{} `json:"params,omitempty" yaml:"params,omitempty"`
}

func NewStreamConfigFromYaml

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