Documentation
¶
Overview ¶
Package plugin implements adapters that wrap the Flight client into the standard source.Source and sink.Sink interfaces. This lets the runner drive external plugins identically to native drivers.
Index ¶
- type ClientFunc
- type ServerStopper
- type SinkAdapter
- func (a *SinkAdapter) Close() error
- func (a *SinkAdapter) EnsureTable(ctx context.Context, ref core.TableRef, schema core.Schema, ...) error
- func (a *SinkAdapter) Position(ctx context.Context, ref core.TableRef) (string, error)
- func (a *SinkAdapter) Properties(_ context.Context, _ core.TableRef) (map[string]string, error)
- func (a *SinkAdapter) SetProperties(_ context.Context, _ core.TableRef, _ map[string]string) error
- func (a *SinkAdapter) Writer(_ context.Context, ref core.TableRef, _ core.CastPolicy, ...) (sink.TableWriter, error)
- type SourceAdapter
- func (a *SourceAdapter) Close() error
- func (a *SourceAdapter) InitialPosition(_ context.Context) (position.Position, error)
- func (a *SourceAdapter) Introspect(ctx context.Context, t spec.Table) (core.TableRef, core.Schema, []core.Warning, error)
- func (a *SourceAdapter) Open(ctx context.Context, refs []core.TableRef) (source.Reader, error)
- func (a *SourceAdapter) ParsePosition(s string) (position.Position, error)
- type StringPosition
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type ClientFunc ¶ added in v0.5.0
ClientFunc resolves the plugin's CURRENT Flight client, or nil when it is not connected. A subprocess plugin can be restarted by its supervisor, so an adapter must resolve the client per use instead of pinning the one that was live when it was built; an in-process plugin (flightwrap) passes a constant (issue #570).
type ServerStopper ¶ added in v0.5.0
type ServerStopper interface{ Stop() }
ServerStopper is an in-process Flight server an adapter owns and must stop when it closes. flightwrap passes one; the subprocess path passes none — the stage supervisor owns that process (issue #496).
type SinkAdapter ¶
type SinkAdapter struct {
// contains filtered or unexported fields
}
SinkAdapter wraps a Flight client as a sink.Sink. It translates the internal rowchange.Batch into Arrow records and writes them via DoPut, then calls Flush to guarantee durability.
func NewSinkAdapter ¶
func NewSinkAdapter(clients ClientFunc, logger *slog.Logger, servers ...ServerStopper) *SinkAdapter
NewSinkAdapter creates a sink adapter over a connected Flight client. An optional ServerStopper is the in-process server flightwrap started, which Close stops (issue #496).
func (*SinkAdapter) Close ¶
func (a *SinkAdapter) Close() error
Close waits for in-flight operations, releases the Flight client, and stops the in-process server flightwrap started (issue #496).
func (*SinkAdapter) EnsureTable ¶
func (a *SinkAdapter) EnsureTable(ctx context.Context, ref core.TableRef, schema core.Schema, partitionBy []string, _ core.CastPolicy, mode dataplane.WriteMode) error
EnsureTable hands the plugin sink the table's canonical schema, primary key and write mode, so it stores the real type shape and dedup key instead of re-inferring them from stringified rows (issue #569). A plugin that does not implement the action (an older binary) is left to manage its own schema.
func (*SinkAdapter) Position ¶
Position reads the plugin sink's committed position for a table, so a restart resumes instead of re-reading from the beginning (issue #569). A plugin that does not implement the action, or one not connected yet, reads as empty.
func (*SinkAdapter) Properties ¶
Properties returns an empty map for external plugins.
func (*SinkAdapter) SetProperties ¶
SetProperties is a no-op for external plugins.
func (*SinkAdapter) Writer ¶
func (a *SinkAdapter) Writer(_ context.Context, ref core.TableRef, _ core.CastPolicy, _ []core.MetadataColumn) (sink.TableWriter, error)
Writer returns a table writer that streams records via DoPut.
type SourceAdapter ¶
type SourceAdapter struct {
// contains filtered or unexported fields
}
SourceAdapter wraps a Flight client as a source.Source. It speaks the external plugin contract (GetFlightInfo + DoGet) and projects the plugin's Arrow CDC record stream (contract §8.1) straight into the flat urutau wire schema.
func NewSourceAdapter ¶
func NewSourceAdapter(clients ClientFunc, src spec.Source, logger *slog.Logger, servers ...ServerStopper) *SourceAdapter
NewSourceAdapter creates a source adapter over a connected Flight client. An optional ServerStopper is the in-process server flightwrap started, which Close stops (issue #496).
func (*SourceAdapter) Close ¶ added in v0.5.0
func (a *SourceAdapter) Close() error
Close releases the adapter: it closes the Flight client and stops the in-process server flightwrap started. Idempotent, so a reader's Close and a direct caller may both call it (issue #496).
func (*SourceAdapter) InitialPosition ¶
InitialPosition returns the empty position (first boot). External plugins always start from the beginning unless a snapshot resume is in progress.
func (*SourceAdapter) Introspect ¶
func (a *SourceAdapter) Introspect(ctx context.Context, t spec.Table) (core.TableRef, core.Schema, []core.Warning, error)
Introspect resolves one spec table into its ref, schema, and warnings. For external plugins, schema comes from GetFlightInfo in snapshot mode. Validates that declared primary key columns exist in the returned schema.
func (*SourceAdapter) Open ¶
Open opens the replication reader. It calls GetFlightInfo in changes mode to get the schema and then starts streaming via DoGet.
func (*SourceAdapter) ParsePosition ¶
func (a *SourceAdapter) ParsePosition(s string) (position.Position, error)
ParsePosition decodes a stored position string.
type StringPosition ¶
type StringPosition struct {
Offset string
}
StringPosition is an opaque position backed by a base64 offset string. External plugins use opaque offsets; the runner never inspects them.
func (StringPosition) Compare ¶
func (p StringPosition) Compare(other position.Position) int
Compare is identity-only. An opaque offset is a plugin cookie (contract §8.2), not an ordinal — the base64 alphabet does not preserve byte order, so lexicographic comparison is meaningless and can silently reorder resume points (skipping uncommitted data). The only defined relation is identity; anything else is position.Incomparable and callers MUST handle it conservatively.
func (StringPosition) Contains ¶
func (p StringPosition) Contains(other position.Position) bool
Contains is identity-only for the same reason: an opaque offset does not contain another unless it is the same cookie.
func (StringPosition) String ¶
func (p StringPosition) String() string
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
Package flightserver serves the plugin Arrow Flight contract (internal/ plugin/contract) in-process, backed by a real source.Source or sink.Sink.
|
Package flightserver serves the plugin Arrow Flight contract (internal/ plugin/contract) in-process, backed by a real source.Source or sink.Sink. |
|
Package flightwrap implements driver.PluginWrap: it turns a .so's raw registered source/sink factories into factories that serve the real driver over an in-process Arrow Flight server (internal/plugin/ flightserver) and hand back a Flight client adapter (internal/plugin) — the SAME contract a subprocess plugin speaks.
|
Package flightwrap implements driver.PluginWrap: it turns a .so's raw registered source/sink factories into factories that serve the real driver over an in-process Arrow Flight server (internal/plugin/ flightserver) and hand back a Flight client adapter (internal/plugin) — the SAME contract a subprocess plugin speaks. |
|
Package proc manages external plugin processes: spawn, readiness, log pumping, and lifecycle.
|
Package proc manages external plugin processes: spawn, readiness, log pumping, and lifecycle. |