Documentation
¶
Index ¶
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 New ¶
func New(ctx context.Context, sink config.Sink, conn adbc.Connection, opts ...Option) (core.Sink, error)
New 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.
The context is required rather than optional. It bounds the probe below, which dials, and a convenience overload that supplied context.Background() was how two of the three call sites came to start a pipeline that could hang forever against a host dropping packets rather than refusing them.
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) Flush ¶
func (s *ClickhouseSink) Flush(ctx context.Context) error
Flush delivers the buffered batches, and keeps them buffered if it cannot.
The retry ladder calls Flush again after a retryable failure. Discarding the buffer on the way in made that second call find nothing to send and return nil, so the ladder reported success for rows that never left the process and the pipeline committed their offsets. Batches are released only once ClickHouse has acknowledged them.
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) Flush ¶
func (s *IcebergSink) Flush(ctx context.Context) error
Flush appends the buffered batches, and keeps whatever it could not append.
The retry ladder calls Flush again after a retryable failure. Discarding the buffer on the way in made that second call find nothing to append and return nil, so the ladder reported success for rows that were never written and the pipeline committed their offsets.
Each batch is a separate append, so a failure part-way through has already committed the batches before it. Only the ones still undelivered are kept; requeueing all of them would append the earlier ones twice.
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.
func (*KafkaSink) Flush ¶
Flush blocks until every buffered record has been acknowledged, so a batch is durable before its source offsets are committed.
It waits on ctx, not on a background context. franz-go retries a produce indefinitely by default, so against a broker that stopped answering a background context has nothing to stop it: the flush interval elapsing, a cancelled run and a SIGTERM all wait for a broker that may never come back, and the supervisor kills the process instead. The pipeline's drain already hands the sink a context stripped of cancellation, so honouring the context here cannot cut the final write short.
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