Documentation
¶
Overview ¶
Package vectorized provides RecordBatch-based columnar execution primitives: schema, columns, operator interfaces, pipeline composition, and memory tracking.
Index ¶
- Constants
- Variables
- func AppendActiveRows(dst, src *RecordBatch) error
- func AppendColumnRange(dst, src Column, srcPos, n int) error
- func ReleaseColumn(c Column)
- type BatchOperator
- type BatchPool
- type BatchSchema
- func (s *BatchSchema) ElementIDIndex() int
- func (s *BatchSchema) FieldIndex(name string) (int, bool)
- func (s *BatchSchema) OrderKeyIndex() int
- func (s *BatchSchema) SeriesIDIndex() int
- func (s *BatchSchema) ShardIDIndex() int
- func (s *BatchSchema) TagIndex(family, name string) (int, bool)
- func (s *BatchSchema) TimestampIndex() int
- func (s *BatchSchema) VersionIndex() int
- type BreakerOperator
- type Column
- type ColumnDef
- type ColumnRole
- type ColumnType
- type FusibleOperator
- type MemoryTracker
- type Pipeline
- type PipelineBuilder
- func (b *PipelineBuilder) Apply(f FusibleOperator) *PipelineBuilder
- func (b *PipelineBuilder) Break(br BreakerOperator) *PipelineBuilder
- func (b *PipelineBuilder) Build() (*Pipeline, error)
- func (b *PipelineBuilder) From(p PullOperator) *PipelineBuilder
- func (b *PipelineBuilder) WithMemoryTracker(t *MemoryTracker) *PipelineBuilder
- type PullOperator
- type RecordBatch
- type TagFamilyGroup
- type TypedColumn
- func NewBytesColumn(capacity int) *TypedColumn[[]byte]
- func NewFieldValueColumn(capacity int) *TypedColumn[*modelv1.FieldValue]
- func NewFloat64Column(capacity int) *TypedColumn[float64]
- func NewInt64ArrayColumn(capacity int) *TypedColumn[[]int64]
- func NewInt64Column(capacity int) *TypedColumn[int64]
- func NewStrArrayColumn(capacity int) *TypedColumn[[]string]
- func NewStringColumn(capacity int) *TypedColumn[string]
- func NewTagValueColumn(capacity int) *TypedColumn[*modelv1.TagValue]
- func (c *TypedColumn[T]) Append(v T)
- func (c *TypedColumn[T]) AppendNull()
- func (c *TypedColumn[T]) Data() []T
- func (c *TypedColumn[T]) IsNull(i int) bool
- func (c *TypedColumn[T]) Len() int
- func (c *TypedColumn[T]) MarkNullAt(i int)
- func (c *TypedColumn[T]) Reset()
- func (c *TypedColumn[T]) SetAt(i int, v T)
- func (c *TypedColumn[T]) Type() ColumnType
Constants ¶
const DefaultBatchSize = 1024
DefaultBatchSize is the canonical batch row capacity. uint16 selection vectors cap batch size at 65536.
Variables ¶
var ErrLimitExhausted = errors.New("vectorized: limit exhausted")
ErrLimitExhausted is returned by FusibleOperators that finished their work and want the fused stage to translate the signal into "emit current batch + EOF on next call".
Functions ¶
func AppendActiveRows ¶
func AppendActiveRows(dst, src *RecordBatch) error
AppendActiveRows appends every active row of src (its Selection when present, else all [0,Len) rows) onto dst, column by column, advancing dst.Len. dst and src must share the same schema/column layout. It is the columnar analog of a row-wise copy used to flatten a Selection-narrowed batch into a dense one (frame emit) or to concatenate per-node batches (liaison merge).
func AppendColumnRange ¶
AppendColumnRange copies n rows starting at srcPos from src into dst. Both columns must share the same TypedColumn[T] type. Validity bits are propagated cell-by-cell via dst.MarkNullAt when src.IsNull reports null at the corresponding row.
Slice-typed cell values ([]byte / []int64 / []string) are not deep-copied here. Storage decoders already produce owned cell slices, and avoiding a second copy keeps egress paths allocation-stable.
func ReleaseColumn ¶
func ReleaseColumn(c Column)
ReleaseColumn returns a column to its per-type pool. The column is Reset (length cleared, validity cleared) before being put back. A nil column is a no-op so callers can release defensively. Unknown column types are also no-ops — the column is simply dropped on the floor for the GC.
Types ¶
type BatchOperator ¶
type BatchOperator interface {
Init(ctx context.Context) error
OutputSchema() *BatchSchema
Close() error
}
BatchOperator is the lifecycle base. All operators implement it.
Close is idempotent and safe to call at any phase — after Init, mid-Consume, after Finalize, or after any error. It must release every MemoryTracker.Reserve the operator made during its lifetime.
type BatchPool ¶
type BatchPool struct {
// contains filtered or unexported fields
}
BatchPool reuses RecordBatches across pipeline iterations. All batches in a pool share the same schema and capacity.
Caller contract: Put a batch only when it is consistent. Error paths must discard rather than Put — see fusedStage.NextBatch.
func NewBatchPool ¶
func NewBatchPool(schema *BatchSchema, capacity int) *BatchPool
NewBatchPool returns a pool whose Get yields freshly Reset batches.
func (*BatchPool) Get ¶
func (p *BatchPool) Get() *RecordBatch
Get returns a Reset batch. Caller may write rows up to capacity.
func (*BatchPool) Put ¶
func (p *BatchPool) Put(b *RecordBatch)
Put returns a batch to the pool. Nil batches and batches with a foreign schema are silently dropped.
type BatchSchema ¶
type BatchSchema struct {
Columns []ColumnDef
TagFamilyGroups []TagFamilyGroup // ordered tag-column groups by family
FieldColumns []int // ordered field column indices
// contains filtered or unexported fields
}
BatchSchema is the immutable column layout shared by every RecordBatch in a pipeline.
func NewBatchSchema ¶
func NewBatchSchema(cols []ColumnDef) *BatchSchema
NewBatchSchema builds a BatchSchema and precomputes lookup indices.
func (*BatchSchema) ElementIDIndex ¶
func (s *BatchSchema) ElementIDIndex() int
ElementIDIndex returns the element-id column index, or -1 if absent.
func (*BatchSchema) FieldIndex ¶
func (s *BatchSchema) FieldIndex(name string) (int, bool)
FieldIndex returns the column index for a field name.
func (*BatchSchema) OrderKeyIndex ¶
func (s *BatchSchema) OrderKeyIndex() int
OrderKeyIndex returns the order-key column index, or -1 if absent.
func (*BatchSchema) SeriesIDIndex ¶
func (s *BatchSchema) SeriesIDIndex() int
SeriesIDIndex returns the series-id column index, or -1 if absent.
func (*BatchSchema) ShardIDIndex ¶
func (s *BatchSchema) ShardIDIndex() int
ShardIDIndex returns the shard-id column index, or -1 if absent.
func (*BatchSchema) TagIndex ¶
func (s *BatchSchema) TagIndex(family, name string) (int, bool)
TagIndex returns the column index for a (family, name) tag. Lookup uses a struct key, so it does not allocate.
func (*BatchSchema) TimestampIndex ¶
func (s *BatchSchema) TimestampIndex() int
TimestampIndex returns the timestamp column index, or -1 if absent.
func (*BatchSchema) VersionIndex ¶
func (s *BatchSchema) VersionIndex() int
VersionIndex returns the version column index, or -1 if absent.
type BreakerOperator ¶
type BreakerOperator interface {
BatchOperator
Consume(ctx context.Context, b *RecordBatch) error
Finalize(ctx context.Context) error
NextBatch(ctx context.Context) (*RecordBatch, error)
}
BreakerOperator buffers all input via Consume, then produces output via NextBatch after Finalize is called.
type Column ¶
type Column interface {
Type() ColumnType
Len() int
IsNull(i int) bool
Reset()
AppendNull()
MarkNullAt(i int)
}
Column is the storage-agnostic view of one column in a RecordBatch.
func AcquireColumn ¶
func AcquireColumn(t ColumnType, capacity int) Column
AcquireColumn returns a recycled Column of the requested type. The returned column is Reset (length 0, validity cleared) and has a backing capacity of at least the requested size. Callers must pair every AcquireColumn with a matching ReleaseColumn once they are done with the column.
Capacity behavior: pooled columns retain whichever capacity they had at the last Release. If a pooled column is too small for the new request the backing slice is reallocated; otherwise the existing backing slice is reused. Over time the pool converges to the largest capacity seen.
func NewColumnForType ¶
func NewColumnForType(t ColumnType, capacity int) Column
NewColumnForType constructs a fresh Column of the given type with the requested backing capacity. Callers that want pool recycling should use AcquireColumn / ReleaseColumn directly — those are wired into the MeasureBatch pool's Acquire path. NewColumnForType deliberately stays non-pooled so callers (operator-internal slot buffers, the egress coalesce buffer, etc.) that never Release their columns do not skew the pool refcount used by the test-teardown HaveZeroRef invariant.
Panics on unknown type — programmer error, not data error.
type ColumnDef ¶
type ColumnDef struct {
Name string
TagFamily string
Role ColumnRole
Type ColumnType
}
ColumnDef describes one column in a BatchSchema.
type ColumnRole ¶
type ColumnRole int
ColumnRole identifies the semantic role of a column within a RecordBatch.
const ( RoleTimestamp ColumnRole = iota RoleVersion RoleSeriesID RoleShardID RoleTag RoleField RoleElementID RoleOrderKey )
Column roles. Each batch schema may include at most one column per metadata role.
type ColumnType ¶
type ColumnType int
ColumnType is the runtime type tag for a Column.
const ( ColumnTypeInt64 ColumnType = iota ColumnTypeFloat64 ColumnTypeString ColumnTypeBytes ColumnTypeInt64Array ColumnTypeStrArray ColumnTypeTagValue ColumnTypeFieldValue )
ColumnType variants. Each value corresponds to a TypedColumn[T] specialization.
The TagValue / FieldValue variants are passthrough columns: they hold the original *modelv1.TagValue / *modelv1.FieldValue pointers from the scan source unchanged, eliminating the decode/re-encode round trip when no operator consumes the typed value. They are only useful when the scan output is destined for the egress serializer; an operator that needs typed primitives should pick a typed column type instead.
func (ColumnType) String ¶
func (c ColumnType) String() string
String returns a human label used in diagnostics and error messages.
type FusibleOperator ¶
type FusibleOperator interface {
BatchOperator
Process(ctx context.Context, b *RecordBatch) error
}
FusibleOperator transforms a batch in place. No state across batches.
Process must not retain references to the batch beyond the call. Returning ErrLimitExhausted signals "emit this batch then EOF on the next pull".
type MemoryTracker ¶
type MemoryTracker struct {
// contains filtered or unexported fields
}
MemoryTracker is a lock-free per-query memory budget.
Reserve grows used iff used+bytes <= limit; otherwise it returns an error and leaves used unchanged. Release shrinks used. Used reads the current value.
func NewMemoryTracker ¶
func NewMemoryTracker(limit int64) *MemoryTracker
NewMemoryTracker returns a tracker with the given byte limit.
func (*MemoryTracker) Release ¶
func (m *MemoryTracker) Release(bytes int64)
Release returns bytes to the budget. A negative bytes value indicates a programmer error and panics.
func (*MemoryTracker) Reserve ¶
func (m *MemoryTracker) Reserve(bytes int64) error
Reserve attempts to allocate bytes from the budget. On success used grows by bytes; on failure used is unchanged.
A negative bytes value indicates a programmer error (Reserve must not be used to decrement) and panics. Zero bytes is a valid no-op.
func (*MemoryTracker) Used ¶
func (m *MemoryTracker) Used() int64
Used returns the current outstanding reservation.
type Pipeline ¶
type Pipeline struct {
// contains filtered or unexported fields
}
Pipeline is the composed sequence of stages from source to final breaker. It exposes a single PullOperator-shaped Next method to the driver.
func (*Pipeline) Init ¶
Init cascades initialization down through every stage to the source. Must be called once before the first Next, after Build. Re-calling is safe but pointless — each stage's Init is idempotent only if its underlying operator's Init is.
func (*Pipeline) Next ¶
func (p *Pipeline) Next(ctx context.Context) (*RecordBatch, error)
Next returns the next batch from the head stage.
func (*Pipeline) Tracker ¶
func (p *Pipeline) Tracker() *MemoryTracker
Tracker returns the shared per-pipeline MemoryTracker, or nil if the builder did not set one. Operators that bookkeep memory should be constructed with this tracker so they all draw from a single budget.
type PipelineBuilder ¶
type PipelineBuilder struct {
// contains filtered or unexported fields
}
PipelineBuilder fluently composes a Pipeline.
Apply/Break ordering: every Apply call queues a fusible into the currently-open fused segment. Break closes that segment, wraps it with the supplied breaker, and starts a new (empty) segment on top. So the observable stage order matches the call order: each fusible runs before any breaker that was added later, and after any breaker that was added earlier. Plan-tree builders rely on this — e.g. a Limit Apply'd after a GroupByAgg Break must execute on the aggregated output, not on the raw source rows.
func NewPipelineBuilder ¶
func NewPipelineBuilder() *PipelineBuilder
NewPipelineBuilder starts a builder.
func (*PipelineBuilder) Apply ¶
func (b *PipelineBuilder) Apply(f FusibleOperator) *PipelineBuilder
Apply queues a FusibleOperator into the current open fused segment.
func (*PipelineBuilder) Break ¶
func (b *PipelineBuilder) Break(br BreakerOperator) *PipelineBuilder
Break closes the current fused segment and wraps it with br, starting a new empty segment for subsequent Apply calls to attach above the breaker.
func (*PipelineBuilder) Build ¶
func (b *PipelineBuilder) Build() (*Pipeline, error)
Build validates and constructs the Pipeline. Each closed segment becomes fusedStage(prev, preFused) → breakerStage(_, breaker); any fusibles still queued after the last Break form a final fused stage on top of the chain.
func (*PipelineBuilder) From ¶
func (b *PipelineBuilder) From(p PullOperator) *PipelineBuilder
From sets the leaf source.
func (*PipelineBuilder) WithMemoryTracker ¶
func (b *PipelineBuilder) WithMemoryTracker(t *MemoryTracker) *PipelineBuilder
WithMemoryTracker attaches a shared MemoryTracker to the pipeline. Operators that bookkeep memory (BatchGroupBy, BatchAggregation) should be constructed with this same tracker so reservations stack against a single budget.
type PullOperator ¶
type PullOperator interface {
BatchOperator
NextBatch(ctx context.Context) (*RecordBatch, error)
}
PullOperator produces batches. It is the source of a pipeline.
NextBatch contract:
- (non-nil, nil) → valid batch with Len > 0
- (nil, nil) → EOF; caller must not call NextBatch again
- (nil, non-nil) → error; pipeline stops
NextBatch may block — channel-backed implementations are explicitly supported for future distributed remote-scan operators.
type RecordBatch ¶
type RecordBatch struct {
Schema *BatchSchema
Columns []Column
Selection []uint16
Len int
}
RecordBatch is the unit of work flowing through the pipeline. All columns share Len logical rows. Selection optionally narrows that to a subset.
Selection contract:
- nil → all rows in [0, Len) are active
- [] → zero rows active (post-fusible-filter empty case)
- non-empty → only the listed row indices are active
Operators must use ActiveLen() to compute the effective row count.
func NewRecordBatch ¶
func NewRecordBatch(schema *BatchSchema, capacity int) *RecordBatch
NewRecordBatch allocates a batch with column capacities sized to capacity rows.
func (*RecordBatch) ActiveLen ¶
func (b *RecordBatch) ActiveLen() int
ActiveLen returns the number of rows operators should process.
nil Selection → Len. Non-nil Selection → len(Selection), even when zero.
func (*RecordBatch) Reset ¶
func (b *RecordBatch) Reset()
Reset clears every column and the selection vector. Schema is preserved.
type TagFamilyGroup ¶
TagFamilyGroup pre-computes the (family, [column indices]) layout used by the row-by-row serializer. Storing it on the schema lets the hot path stamp out one TagFamily per family per row without re-grouping or allocating a `map[string]*modelv1.TagFamily` on every row.
type TypedColumn ¶
type TypedColumn[T any] struct { // contains filtered or unexported fields }
TypedColumn is a generic Column with element type T. One instance per supported T — use the typed constructors below.
func NewBytesColumn ¶
func NewBytesColumn(capacity int) *TypedColumn[[]byte]
NewBytesColumn constructs a TypedColumn[[]byte] with the given capacity.
func NewFieldValueColumn ¶
func NewFieldValueColumn(capacity int) *TypedColumn[*modelv1.FieldValue]
NewFieldValueColumn is the field-side counterpart of NewTagValueColumn.
func NewFloat64Column ¶
func NewFloat64Column(capacity int) *TypedColumn[float64]
NewFloat64Column constructs a TypedColumn[float64] with the given capacity.
func NewInt64ArrayColumn ¶
func NewInt64ArrayColumn(capacity int) *TypedColumn[[]int64]
NewInt64ArrayColumn constructs a TypedColumn[[]int64] with the given capacity.
func NewInt64Column ¶
func NewInt64Column(capacity int) *TypedColumn[int64]
NewInt64Column constructs a TypedColumn[int64] with the given capacity.
func NewStrArrayColumn ¶
func NewStrArrayColumn(capacity int) *TypedColumn[[]string]
NewStrArrayColumn constructs a TypedColumn[[]string] with the given capacity.
func NewStringColumn ¶
func NewStringColumn(capacity int) *TypedColumn[string]
NewStringColumn constructs a TypedColumn[string] with the given capacity.
func NewTagValueColumn ¶
func NewTagValueColumn(capacity int) *TypedColumn[*modelv1.TagValue]
NewTagValueColumn constructs a TypedColumn[*modelv1.TagValue] passthrough column. Cells hold the original *modelv1.TagValue pointers from the scan source; the egress serializer returns those pointers directly, avoiding the decode-into-typed / re-encode-into-protobuf round trip that otherwise dominates allocation cost when no operator consumes the column.
func (*TypedColumn[T]) Append ¶
func (c *TypedColumn[T]) Append(v T)
Append adds a value, marking it valid.
func (*TypedColumn[T]) AppendNull ¶
func (c *TypedColumn[T]) AppendNull()
AppendNull adds a zero-value placeholder and marks it null.
func (*TypedColumn[T]) Data ¶
func (c *TypedColumn[T]) Data() []T
Data returns the backing slice for bulk access.
func (*TypedColumn[T]) IsNull ¶
func (c *TypedColumn[T]) IsNull(i int) bool
IsNull reports whether row i is null.
func (*TypedColumn[T]) MarkNullAt ¶
func (c *TypedColumn[T]) MarkNullAt(i int)
MarkNullAt marks an existing row at index i as null. Length is unchanged.
func (*TypedColumn[T]) Reset ¶
func (c *TypedColumn[T]) Reset()
Reset clears length and validity. Capacity is retained.
func (*TypedColumn[T]) SetAt ¶
func (c *TypedColumn[T]) SetAt(i int, v T)
SetAt overwrites the value at index i without changing length or validity. The validity bit at i is cleared (row becomes valid). Panics if i is out of range — matches the same contract as a direct slice write.
func (*TypedColumn[T]) Type ¶
func (c *TypedColumn[T]) Type() ColumnType
Type returns the static ColumnType this column was built with.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package frame defines the shared, engine-agnostic non-proto columnar binary frame carried as the SendResponse.body for a vec-native query topic.
|
Package frame defines the shared, engine-agnostic non-proto columnar binary frame carried as the SendResponse.body for a vec-native query topic. |
|
Package measure implements measure-specific vectorized operators (scan, cursor, extract, limit, group-by, aggregation, top, output serialization).
|
Package measure implements measure-specific vectorized operators (scan, cursor, extract, limit, group-by, aggregation, top, output serialization). |
|
frame
Package frame binds the shared vec columnar frame codec (pkg/query/vectorized/frame) to the measure engine.
|
Package frame binds the shared vec columnar frame codec (pkg/query/vectorized/frame) to the measure engine. |
|
plan
Package plan is the vectorized measure-query plan tree (G8).
|
Package plan is the vectorized measure-query plan tree (G8). |
|
Package stream implements stream-specific vectorized query operators: the pull pipeline that scans elements into columnar batches and merges them in index or timestamp order at egress.
|
Package stream implements stream-specific vectorized query operators: the pull pipeline that scans elements into columnar batches and merges them in index or timestamp order at egress. |
|
frame
Package frame binds the shared vec columnar frame codec (pkg/query/vectorized/frame) to the stream engine.
|
Package frame binds the shared vec columnar frame codec (pkg/query/vectorized/frame) to the stream engine. |
|
Package trace implements trace-specific vectorized query operators: the two-phase pull pipeline that resolves trace IDs (Phase 1) and materializes spans grouped per trace (Phase 2).
|
Package trace implements trace-specific vectorized query operators: the two-phase pull pipeline that resolves trace IDs (Phase 1) and materializes spans grouped per trace (Phase 2). |