Documentation
¶
Overview ¶
Package sourcepull adapts a push-based change channel into the pull-based source.Reader.Next surface, encoding buffered changes into wire batches.
The built-in decoders emit rowchange.Change (binlog/JSON events are row-shaped — a private detail of the decoders); makeBatch buffers a few and encodes one wire RecordBatch via transport.RecordFromChanges. A source that can introspect its tables supplies the canonical schemas via SetSchemas so batches encode against a STABLE schema — never a per-batch inference that drifts when a drain happens to omit a sparse column. Without schemas, makeBatch falls back to inference for schema-less producers (their resolved schema is owned upstream).
Index ¶
- type Puller
- func (p *Puller) BoundTransactions()
- func (p *Puller) Drain(ctx context.Context, emit func(*dataplane.Batch) error) error
- func (p *Puller) Next(ctx context.Context) (*dataplane.Batch, error)
- func (p *Puller) SetErr(errCh <-chan error)
- func (p *Puller) SetSchemas(schemas map[string]core.Schema)
- func (p *Puller) SetSourceSchemas(resolved map[string]core.Schema)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Puller ¶
type Puller struct {
// contains filtered or unexported fields
}
Puller wraps a source's change channel and terminal-error channel into the pull-based Next surface. The concrete source calls Start to launch its decoder and wire the error channel.
func (*Puller) BoundTransactions ¶ added in v0.4.0
func (p *Puller) BoundTransactions()
BoundTransactions declares that the decoder ends every transaction with an OpTxnEnd change. Every row of a transaction carries its position, so a batch then holds only whole transactions: a commit that recorded a transaction's position with part of its rows would let a resume skip the rest (#456).
func (*Puller) Drain ¶
Drain emits every batch of pending changes (buffered or still on the channel) without blocking for new data. Used by the runner's relay to flush decoded events ahead of a Closes marker.
func (*Puller) Next ¶
Next reads accumulated changes and bridges them into one batch. Returns (nil, nil) at a clean stream end.
func (*Puller) SetSchemas ¶
SetSchemas installs the canonical schema per target table so makeBatch encodes against a stable shape instead of per-drain inference. Call before Next.
func (*Puller) SetSourceSchemas ¶ added in v0.4.0
SetSourceSchemas implements source.SchemaSetter for every stream that embeds a Puller: the engine hands over the resolved wire schema, and each column the source introspected as KindUnknown (an unsigned MySQL integer, say) takes its resolved kind, which only the declared cast knows. Without it such a column cannot be encoded ("unsupported canonical kind unknown"). Every other column keeps the source's own type, and no column is added: the wire carries the source type, and the sink applies the cast.
Source Files