Documentation
¶
Index ¶
- Constants
- func New(sink config.Sink, conn adbc.Connection, opts ...Option) (core.Sink, error)
- func NewWithContext(ctx context.Context, sink config.Sink, conn adbc.Connection, opts ...Option) (core.Sink, error)
- type ClickhouseSink
- type ConsoleSink
- type IcebergSink
- type KafkaSink
- type NoopSink
- type Option
- type Prober
- type RetryPolicy
- type SQLCommandSink
Constants ¶
const ( DefaultRetryMaxAttempts = 4 DefaultRetryInitialBackoff = 100 * time.Millisecond DefaultRetryMaxBackoff = 2 * time.Second DefaultRetryDeadline = 10 * time.Second )
Retry defaults. Deliberately short: the point is to absorb a hiccup, not to wait out an outage. A destination that is still refusing after a few seconds is better reported, because the exit code says "retryable" and a supervisor restarts the pipeline anyway.
Retrying is on by default. Before this, one refused connection killed the process, which cost a cold start, a group rejoin and a rebalance to recover from a blip. That default was the defect, not a safe baseline.
Variables ¶
This section is empty.
Functions ¶
func NewWithContext ¶ added in v1.0.5
func NewWithContext(ctx context.Context, sink config.Sink, conn adbc.Connection, opts ...Option) (core.Sink, error)
NewWithContext builds a sink and wraps it in a retry ladder where one helps.
Not every sink is wrapped. The Kafka sink hands records to franz-go, which already retries a produce with its own backoff; a second ladder on top of that one delays the report without improving delivery. The console, noop and sqlcommand sinks reach nothing that can be temporarily unavailable -- the sqlcommand sink writes through the pipeline's own DuckDB connection, and a failure there is not a network blip.
That leaves the sinks that cross a network to somebody else's server.
Types ¶
type ClickhouseSink ¶
type ClickhouseSink struct {
// contains filtered or unexported fields
}
ClickhouseSink inserts result batches into a ClickHouse table.
The Python sink hands the Arrow table to clickhouse_connect's insert_arrow, which maps Arrow columns to table columns by name. clickhouse-go has no Arrow entry point, so the batch is unpacked into rows against an explicit column list taken from the Arrow schema, which preserves that name-based mapping.
func NewClickhouseSink ¶
func NewClickhouseSink(conf config.ClickhouseSink) (*ClickhouseSink, error)
func (*ClickhouseSink) Batch ¶
func (s *ClickhouseSink) Batch() (arrow.Table, error)
Batch returns nothing. The Python ClickhouseSink alone among the sinks reports no batch: rows go straight to ClickHouse and are not held for a downstream reader.
func (*ClickhouseSink) Close ¶
func (s *ClickhouseSink) Close() error
func (*ClickhouseSink) Probe ¶ added in v1.0.5
func (s *ClickhouseSink) Probe(ctx context.Context) error
Probe dials the server. clickhouse.Open only validates options, so this is the first time the pipeline learns whether the destination is there.
func (*ClickhouseSink) WriteTable ¶
type ConsoleSink ¶
type ConsoleSink struct {
// contains filtered or unexported fields
}
ConsoleSink writes each result row to stdout as a JSON object.
func NewConsoleSink ¶
func NewConsoleSink() *ConsoleSink
func NewConsoleSinkTo ¶
func NewConsoleSinkTo(w io.Writer) *ConsoleSink
func (*ConsoleSink) WriteTable ¶
type IcebergSink ¶
type IcebergSink struct {
// contains filtered or unexported fields
}
IcebergSink appends result batches to an Iceberg table.
func NewIcebergSink ¶
func NewIcebergSink(ctx context.Context, catalogName, tableName string) (*IcebergSink, error)
func (*IcebergSink) WriteTable ¶
type KafkaSink ¶
type KafkaSink struct {
// contains filtered or unexported fields
}
KafkaSink produces one message per result row, JSON encoded, matching the Python KafkaSink.
type Option ¶ added in v1.0.5
type Option func(*options)
Option configures how a sink is built.
func WithMeterProvider ¶ added in v1.0.5
func WithMeterProvider(mp metric.MeterProvider) Option
WithMeterProvider supplies the provider the retry counter records through. Without one the counter records nothing, which is what a pipeline started with no --metrics wants.
type Prober ¶ added in v1.0.5
Prober is implemented by a sink that can check its destination before the first batch arrives.
Without it a sink only discovers its destination when a batch reaches it. clickhouse.Open validates options and never dials, so a pipeline whose DSN names a host that does not resolve starts normally, logs "consumer loop starting", and runs. With a long flush interval the failure appears minutes later, and a supervisor calls the pipeline healthy for all of them.
type RetryPolicy ¶ added in v1.0.5
type RetryPolicy struct {
MaxAttempts int
InitialBackoff time.Duration
MaxBackoff time.Duration
Deadline time.Duration
}
RetryPolicy bounds how long a sink keeps trying a destination that is not answering.
Deadline bounds the whole ladder rather than one attempt, and it is the field that matters most. The retry runs inside the pipeline's open state transaction, and DuckDB's now() returns that transaction's start time, so a ladder that outlives the flush interval freezes the window clock -- the bug #158 fixed, reached by a second route.
func RetryPolicyFrom ¶ added in v1.0.5
func RetryPolicyFrom(c *config.SinkRetry) RetryPolicy
RetryPolicyFrom resolves a config block into a policy, filling in defaults for anything the user left out.
func (RetryPolicy) Enabled ¶ added in v1.0.5
func (p RetryPolicy) Enabled() bool
Enabled reports whether the policy will ever make a second attempt.
type SQLCommandSink ¶
type SQLCommandSink struct {
// contains filtered or unexported fields
}
SQLCommandSink materializes the result batch as a table and runs arbitrary DuckDB SQL against it. This is how the Python engine writes parquet, S3, Postgres, DuckLake and MotherDuck outputs.
func NewSQLCommandSink ¶
func NewSQLCommandSink(conn adbc.Connection, sql string, substitutions []config.SQLCommandSubstitution) (*SQLCommandSink, error)
Source Files