sinks

Sep 10, 2026

Changes in this version

+ func Kinds() []string

type ClickhouseSink

+ func (s *ClickhouseSink) BufferedRows() int

type ConsoleSink

+ func (s *ConsoleSink) BufferedRows() int

type IcebergSink

+ func (s *IcebergSink) BufferedRows() int

type KafkaSink

+ func (s *KafkaSink) BufferedRows() int

+ func (s *KafkaSink) Probe(ctx context.Context) error

type Option

+ func WithSinkRole(role string) Option

type SQLCommandSink

+ func (s *SQLCommandSink) BufferedRows() int

Sep 5, 2026

Changes in this version

+ const DefaultRetryDeadline

+ const DefaultRetryInitialBackoff

+ const DefaultRetryMaxAttempts

+ const DefaultRetryMaxBackoff

+ func NewWithContext(ctx context.Context, sink config.Sink, conn adbc.Connection, opts ...Option) (core.Sink, error)

type ClickhouseSink

+ func (s *ClickhouseSink) Probe(ctx context.Context) error

+ type Option func(*options)

+ func WithMeterProvider(mp metric.MeterProvider) Option

+ type Prober interface

+ Probe func(ctx context.Context) error

+ type RetryPolicy struct

+ Deadline time.Duration

+ InitialBackoff time.Duration

+ MaxAttempts int

+ MaxBackoff time.Duration

+ func RetryPolicyFrom(c *config.SinkRetry) RetryPolicy

+ func (p RetryPolicy) Enabled() bool

Sep 1, 2026

Changes in this version

+ func New(sink config.Sink, conn adbc.Connection) (core.Sink, error)

+ type ClickhouseSink struct

+ func NewClickhouseSink(conf config.ClickhouseSink) (*ClickhouseSink, error)

+ func (s *ClickhouseSink) Batch() (arrow.Table, error)

+ func (s *ClickhouseSink) Close() error

+ func (s *ClickhouseSink) Flush() error

+ func (s *ClickhouseSink) WriteTable(batch arrow.Table) error

+ type ConsoleSink struct

+ func NewConsoleSink() *ConsoleSink

+ func NewConsoleSinkTo(w io.Writer) *ConsoleSink

+ func (s *ConsoleSink) Batch() (arrow.Table, error)

+ func (s *ConsoleSink) Flush() error

+ func (s *ConsoleSink) WriteTable(batch arrow.Table) error

+ type IcebergSink struct

+ func NewIcebergSink(ctx context.Context, catalogName, tableName string) (*IcebergSink, error)

+ func (s *IcebergSink) Batch() (arrow.Table, error)

+ func (s *IcebergSink) Flush() error

+ func (s *IcebergSink) WriteTable(batch arrow.Table) error

+ type KafkaSink struct

+ func NewKafkaSink(conf config.KafkaSink) (*KafkaSink, error)

+ func (s *KafkaSink) Batch() (arrow.Table, error)

+ func (s *KafkaSink) Close() error

+ func (s *KafkaSink) Flush() error

+ func (s *KafkaSink) WriteTable(batch arrow.Table) error

+ type NoopSink struct

+ func (n *NoopSink) Batch() (arrow.Table, error)

+ func (n *NoopSink) Flush() error

+ func (n *NoopSink) WriteTable(batch arrow.Table) error

+ type SQLCommandSink struct

+ func NewSQLCommandSink(conn adbc.Connection, sql string, ...) (*SQLCommandSink, error)

+ func (s *SQLCommandSink) Batch() (arrow.Table, error)

+ func (s *SQLCommandSink) Flush() error

+ func (s *SQLCommandSink) WriteTable(batch arrow.Table) error