destination

Documentation

Index

• func PluginTestSuiteRunner(t *testing.T, newPlugin NewPluginFunc, destSpec specs.Destination, ...)
• func RecordDiff(l arrow.Record, r arrow.Record) string
• type Client
• type ClientResource
• type ManagedWriter
• type Metrics
• type MigrateStrategy
• type NewClientFunc
• type NewPluginFunc
• type Option
  • func WithBatchTimeout(seconds int) Option
  • func WithDefaultBatchSize(defaultBatchSize int) Option
  • func WithDefaultBatchSizeBytes(defaultBatchSizeBytes int) Option
  • func WithManagedWriter() Option
• type Plugin
  • func NewPlugin(name string, version string, newClientFunc NewClientFunc, opts ...Option) *Plugin
  • func (p *Plugin) Close(ctx context.Context) error
  • func (p *Plugin) DeleteStale(ctx context.Context, tables schema.Schemas, sourceName string, ...) error
  • func (p *Plugin) Init(ctx context.Context, logger zerolog.Logger, spec specs.Destination) error
  • func (p *Plugin) Metrics() Metrics
  • func (p *Plugin) Migrate(ctx context.Context, tables schema.Schemas) error
  • func (p *Plugin) Name() string
  • func (p *Plugin) Read(ctx context.Context, table *arrow.Schema, sourceName string, ...) error
  • func (p *Plugin) Version() string
  • func (p *Plugin) Write(ctx context.Context, sourceSpec specs.Source, tables schema.Schemas, ...) error
• type PluginTestSuite
• type PluginTestSuiteTests
• type UnimplementedManagedWriter
  • func (*UnimplementedManagedWriter) WriteTableBatch(context.Context, *arrow.Schema, []arrow.Record) error
• type UnimplementedUnmanagedWriter
  • func (*UnimplementedUnmanagedWriter) Metrics() Metrics
  • func (*UnimplementedUnmanagedWriter) Write(context.Context, schema.Schemas, <-chan arrow.Record) error
• type UnmanagedWriter

Functions

func PluginTestSuiteRunner

func PluginTestSuiteRunner(t *testing.T, newPlugin NewPluginFunc, destSpec specs.Destination, tests PluginTestSuiteTests)

func RecordDiff

func RecordDiff(l arrow.Record, r arrow.Record) string

Types

type Client

type Client interface {
	Migrate(ctx context.Context, tables schema.Schemas) error
	Read(ctx context.Context, table *arrow.Schema, sourceName string, res chan<- arrow.Record) error
	ManagedWriter
	UnmanagedWriter
	DeleteStale(ctx context.Context, tables schema.Schemas, sourceName string, syncTime time.Time) error
	Close(ctx context.Context) error
}

type ClientResource

type ClientResource struct {
	TableName string
	Data      []any
}

type ManagedWriter

type ManagedWriter interface {
	WriteTableBatch(ctx context.Context, table *arrow.Schema, data []arrow.Record) error
}

type Metrics

type Metrics struct {
	// Errors number of errors / failed writes
	Errors uint64
	// Writes number of successful writes
	Writes uint64
}

type MigrateStrategy

type MigrateStrategy struct {
	AddColumn           specs.MigrateMode
	AddColumnNotNull    specs.MigrateMode
	RemoveColumn        specs.MigrateMode
	RemoveColumnNotNull specs.MigrateMode
	ChangeColumn        specs.MigrateMode
}

MigrateStrategy defines which tests we should include

type NewClientFunc

type NewClientFunc func(context.Context, zerolog.Logger, specs.Destination) (Client, error)

type NewPluginFunc

type NewPluginFunc func() *Plugin

type Option

type Option func(*Plugin)

func WithBatchTimeout

func WithBatchTimeout(seconds int) Option

func WithDefaultBatchSize

func WithDefaultBatchSize(defaultBatchSize int) Option

func WithDefaultBatchSizeBytes

func WithDefaultBatchSizeBytes(defaultBatchSizeBytes int) Option

func WithManagedWriter

func WithManagedWriter() Option

type Plugin

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

func NewPlugin

func NewPlugin(name string, version string, newClientFunc NewClientFunc, opts ...Option) *Plugin

NewPlugin creates a new destination plugin

func (*Plugin) Close

func (p *Plugin) Close(ctx context.Context) error

func (*Plugin) DeleteStale

func (p *Plugin) DeleteStale(ctx context.Context, tables schema.Schemas, sourceName string, syncTime time.Time) error

func (*Plugin) Init

func (p *Plugin) Init(ctx context.Context, logger zerolog.Logger, spec specs.Destination) error

we need lazy loading because we want to be able to initialize after

func (*Plugin) Metrics

func (p *Plugin) Metrics() Metrics

func (*Plugin) Migrate

func (p *Plugin) Migrate(ctx context.Context, tables schema.Schemas) error

we implement all DestinationClient functions so we can hook into pre-post behavior

func (*Plugin) Name

func (p *Plugin) Name() string

func (*Plugin) Read

func (p *Plugin) Read(ctx context.Context, table *arrow.Schema, sourceName string, res chan<- arrow.Record) error

func (*Plugin) Version

func (p *Plugin) Version() string

func (*Plugin) Write

func (p *Plugin) Write(ctx context.Context, sourceSpec specs.Source, tables schema.Schemas, syncTime time.Time, res <-chan arrow.Record) error

type PluginTestSuite

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

type PluginTestSuiteTests

type PluginTestSuiteTests struct {
	// SkipOverwrite skips testing for "overwrite" mode. Use if the destination
	// plugin doesn't support this feature.
	SkipOverwrite bool

	// SkipDeleteStale skips testing "delete-stale" mode. Use if the destination
	// plugin doesn't support this feature.
	SkipDeleteStale bool

	// SkipAppend skips testing for "append" mode. Use if the destination
	// plugin doesn't support this feature.
	SkipAppend bool

	// SkipSecondAppend skips the second append step in the test.
	// This is useful in cases like cloud storage where you can't append to an
	// existing object after the file has been closed.
	SkipSecondAppend bool

	// SkipMigrateAppend skips a test for the migrate function where a column is added,
	// data is appended, then the column is removed and more data appended, checking that the migrations handle
	// this correctly.
	SkipMigrateAppend bool
	// SkipMigrateAppendForce skips a test for the migrate function where a column is changed in force mode
	SkipMigrateAppendForce bool

	// SkipMigrateOverwrite skips a test for the migrate function where a column is added,
	// data is appended, then the column is removed and more data overwritten, checking that the migrations handle
	// this correctly.
	SkipMigrateOverwrite bool
	// SkipMigrateOverwriteForce skips a test for the migrate function where a column is changed in force mode
	SkipMigrateOverwriteForce bool

	MigrateStrategyOverwrite MigrateStrategy
	MigrateStrategyAppend    MigrateStrategy
}

type UnimplementedManagedWriter

type UnimplementedManagedWriter struct{}

func (*UnimplementedManagedWriter) WriteTableBatch

func (*UnimplementedManagedWriter) WriteTableBatch(context.Context, *arrow.Schema, []arrow.Record) error

type UnimplementedUnmanagedWriter

type UnimplementedUnmanagedWriter struct{}

func (*UnimplementedUnmanagedWriter) Metrics

func (*UnimplementedUnmanagedWriter) Metrics() Metrics

func (*UnimplementedUnmanagedWriter) Write

func (*UnimplementedUnmanagedWriter) Write(context.Context, schema.Schemas, <-chan arrow.Record) error

type UnmanagedWriter

type UnmanagedWriter interface {
	Write(ctx context.Context, tables schema.Schemas, res <-chan arrow.Record) error
	Metrics() Metrics
}