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 ¶
- func EncodeChanges(changes []rowchange.Change, table string, schema core.Schema) (arrow.RecordBatch, error)
- 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) SetOnSchemas(fn func(map[string]core.Schema))
- func (p *Puller) SetResume(pos string)
- 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 ¶
func EncodeChanges ¶ added in v0.5.0
func EncodeChanges(changes []rowchange.Change, table string, schema core.Schema) (arrow.RecordBatch, error)
EncodeChanges encodes one table's changes into a wire record against its canonical schema, applying the SOURCE-boundary drift gate and the delete-with-no-PK guard exactly as Puller.makeBatch does. It is the shared encode stage: the puller passes its cached column index, a caller that holds ready changes (a columnar source re-encoding a drift record) passes nil and gets one built. A zero schema is schema-less: the shape is inferred/merged from the rows, as the puller's inference fallback does.
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.
A puller is EITHER row-oriented (ch) or column-oriented (batchCh). A column-oriented source decodes straight into Arrow (transport.RowEncoder) and emits ready wire batches; the puller then forwards them unchanged, so no map[string]any ever exists on the live path (#455). The reader owns transaction bounding, the safe-position split and drift, because it sees the native row shape and the transaction boundary.
func NewColumnar ¶ added in v0.5.0
NewColumnar builds a puller over a column-oriented source's batch channel. The source emits wire-schema *dataplane.Batch values (one table each, transaction-bounded and position-stamped); the puller forwards them.
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 the puller batches whole transactions and never splits one at a batch target it did not choose: a commit that recorded a transaction's position with part of its rows would let a resume skip the rest. A transaction larger than maxBatchRows is split anyway, and only its last piece then carries the transaction's position.
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) SetOnSchemas ¶ added in v0.5.0
SetOnSchemas installs a callback that receives the resolved canonical schemas SetSourceSchemas assigns. A columnar source uses it to rebuild its encoders against the cast-resolved column types before decoding.
func (*Puller) SetResume ¶ added in v0.5.0
SetResume records the position the stream resumes from, the safe position for a partial piece of the first transaction. Call before Next.
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