Documentation
¶
Overview ¶
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.
Index ¶
- Constants
- func BuildElementsFromBatch(batch *vectorized.RecordBatch, projectionTags []model.TagProjection) ([]*streamv1.Element, error)
- func BuildElementsFromBatches(batches []*vectorized.RecordBatch, projectionTags []model.TagProjection) ([]*streamv1.Element, error)
- func BuildStreamBatchSchema(tagProjection []model.TagProjection, orderTagFamily, orderTagName string) *vectorized.BatchSchema
- func BuildStreamMergePipeline(source vectorized.PullOperator, schema *vectorized.BatchSchema, desc bool, ...) (*vectorized.Pipeline, error)
- func ColumnToElementID(v int64) uint64
- func ColumnToSeriesID(v int64) uint64
- func ElementIDToColumn(id uint64) int64
- func IncrQueryCount()
- func QueryCount() int64
- func SeriesIDToColumn(id uint64) int64
- type Distinct
- type Limit
- type SortedMerge
- func (s *SortedMerge) Close() error
- func (s *SortedMerge) Consume(_ context.Context, batch *vectorized.RecordBatch) error
- func (s *SortedMerge) Finalize(context.Context) error
- func (s *SortedMerge) Init(context.Context) error
- func (s *SortedMerge) NextBatch(ctx context.Context) (*vectorized.RecordBatch, error)
- func (s *SortedMerge) OutputSchema() *vectorized.BatchSchema
- type VectorizedConfig
Constants ¶
const ( StreamColumnNameTimestamp = "timestamp" StreamColumnNameElementID = "elementID" StreamColumnNameSeriesID = "seriesID" StreamColumnNameOrderKey = "orderKey" )
Metadata column names for the stream vectorized batch schema.
Variables ¶
This section is empty.
Functions ¶
func BuildElementsFromBatch ¶
func BuildElementsFromBatch(batch *vectorized.RecordBatch, projectionTags []model.TagProjection, ) ([]*streamv1.Element, error)
BuildElementsFromBatch materializes []*streamv1.Element from a single columnar batch, honoring the batch Selection (post-pipeline batches carry one). The Element shape is byte-identical to the row path's BuildElementsFromStreamResult: hex-encoded elementID, timestamppb timestamp, and tag families/tags in projection order with NullTagValue for a missing projected tag.
This is the single shared columnar→proto egress used by both the data-node standalone path (banyand/stream) and the liaison distributed merge (pkg/query/logical/stream), so the two cannot diverge and neither triggers an import cycle with the other.
func BuildElementsFromBatches ¶
func BuildElementsFromBatches(batches []*vectorized.RecordBatch, projectionTags []model.TagProjection, ) ([]*streamv1.Element, error)
BuildElementsFromBatches concatenates BuildElementsFromBatch over a slice of batches, preserving batch order and each batch's Selection.
func BuildStreamBatchSchema ¶
func BuildStreamBatchSchema(tagProjection []model.TagProjection, orderTagFamily, orderTagName string) *vectorized.BatchSchema
BuildStreamBatchSchema builds the columnar batch schema for a stream query. Columns are laid out as Timestamp, ElementID, SeriesID, one passthrough tag column per projected tag (grouped by family), and — when an ordered tag is supplied — a trailing comparable-bytes order-key column.
func BuildStreamMergePipeline ¶
func BuildStreamMergePipeline( source vectorized.PullOperator, schema *vectorized.BatchSchema, desc bool, offset, limit uint32, batchSize, maxRows int, ) (*vectorized.Pipeline, error)
BuildStreamMergePipeline composes the liaison-side merge → distinct → limit pipeline over a source of stream RecordBatches. SortedMerge is the breaker (global ordering), Distinct and Limit are fusibles applied on the ordered output, in that strict order. There is no pre-merge short-circuit.
maxRows bounds the merge to the in-order top-N (0 = unbounded). This is the per-node scan cap (maxElementSize = limit+offset), applied AFTER the merge sorts — the correct top-N in sort order — matching the row path, which caps after its in-order heap merge (blockHeap.merge / MergeStreamResults). It is distinct from the client offset/limit slice the trailing Limit applies.
func ColumnToElementID ¶
ColumnToElementID reinterprets an int64 ElementID column value back to the original uint64 element id. This is an exact bit reinterpretation.
func ColumnToSeriesID ¶
ColumnToSeriesID reinterprets an int64 SeriesID column value back to the original uint64 series id. This is an exact bit reinterpretation.
func ElementIDToColumn ¶
ElementIDToColumn reinterprets a uint64 element id as the int64 stored in the ElementID column. This is an exact bit reinterpretation, not a value cast.
func IncrQueryCount ¶
func IncrQueryCount()
IncrQueryCount increments the process-wide vectorized stream query counter. Called when the vectorized execution path is taken for a stream query.
func QueryCount ¶
func QueryCount() int64
QueryCount returns the cumulative number of vectorized stream queries executed by this process.
func SeriesIDToColumn ¶
SeriesIDToColumn reinterprets a uint64 series id as the int64 stored in the SeriesID column. This is an exact bit reinterpretation, not a value cast.
Types ¶
type Distinct ¶
type Distinct struct {
// contains filtered or unexported fields
}
Distinct deduplicates stream rows by ElementID, keeping the first occurrence in the (already merged) input order. It is stateful across batches via a seen set keyed by the raw uint64 element id, mirroring the row path's first-seen dedup (stream_plan_distributed.go seen map keyed by ElementId). It MUST run downstream of SortedMerge so "first occurrence" reflects the global order.
func NewDistinct ¶
func NewDistinct(schema *vectorized.BatchSchema) *Distinct
NewDistinct constructs a stream first-seen dedup fusible.
func (*Distinct) OutputSchema ¶
func (d *Distinct) OutputSchema() *vectorized.BatchSchema
OutputSchema returns the unchanged input schema.
func (*Distinct) Process ¶
func (d *Distinct) Process(_ context.Context, batch *vectorized.RecordBatch) error
Process rewrites the selection to keep only rows whose ElementID is seen for the first time. Later duplicates are dropped by omission from the selection.
type Limit ¶
type Limit struct {
// contains filtered or unexported fields
}
Limit applies the client-facing global offset+limit slice to the merged, deduplicated stream. It skips the first offset active rows, emits up to limit rows, then signals EOF via ErrLimitExhausted.
These are the CLIENT offset/limit values, applied liaison-side as the final slice (mirrors distributedLimit.Execute in stream_plan_distributed.go). The per-node scan cap of limit+offset is applied UPSTREAM (M6), not here.
func NewLimit ¶
func NewLimit(schema *vectorized.BatchSchema, offset, limit uint32) *Limit
NewLimit constructs a stream global offset+limit fusible.
func (*Limit) OutputSchema ¶
func (l *Limit) OutputSchema() *vectorized.BatchSchema
OutputSchema returns the unchanged input schema.
func (*Limit) Process ¶
func (l *Limit) Process(_ context.Context, batch *vectorized.RecordBatch) error
Process rewrites the selection to the client offset+limit window. A zero limit emits nothing and signals EOF immediately.
type SortedMerge ¶
type SortedMerge struct {
// contains filtered or unexported fields
}
SortedMerge is a globally-ordering breaker over stream RecordBatches.
It Consumes every input batch (buffering row references), Finalize sorts them stably by the schema order key, and NextBatch emits key-ordered output batches sized to batchSize. Time-order keys on the Timestamp int64 column (stream timestamps are non-negative UnixNano, so a direct int64 compare matches the row path's convert.Uint64ToBytes big-endian byte order); index-order keys on the OrderKey comparable-bytes column (byte-lexicographic, matching the row path's MarshalTagValue bytes). asc/desc is honored; ties preserve input arrival order so downstream first-seen dedup stays deterministic.
func NewSortedMerge ¶
func NewSortedMerge(schema *vectorized.BatchSchema, desc bool, batchSize int) *SortedMerge
NewSortedMerge constructs a stream global-merge breaker over the given schema.
func NewSortedMergeWithCap ¶
func NewSortedMergeWithCap(schema *vectorized.BatchSchema, desc bool, batchSize, maxRows int) *SortedMerge
NewSortedMergeWithCap constructs a stream global-merge breaker that keeps only the first maxRows rows in sort order after the stable sort — the correct in-order top-N. A maxRows of 0 means unbounded (equivalent to NewSortedMerge). The cap is applied AFTER the merge sorts, so it matches the row path's cap-after-in-order-merge semantics (blockHeap.merge / MergeStreamResults).
func (*SortedMerge) Close ¶
func (s *SortedMerge) Close() error
Close is idempotent and releases buffered references.
func (*SortedMerge) Consume ¶
func (s *SortedMerge) Consume(_ context.Context, batch *vectorized.RecordBatch) error
Consume buffers references to every active row in the batch. The batch is retained (not copied); callers must not recycle a consumed batch until this operator is closed.
func (*SortedMerge) Finalize ¶
func (s *SortedMerge) Finalize(context.Context) error
Finalize stably sorts the buffered rows by the schema order key, then applies the in-order top-N cap.
The cap keeps the first maxRows rows counted by UNIQUE ElementID, in sort order. This matches the row path's oracle exactly: MergeStreamResults / blockHeap.merge dedup by ElementID DURING their in-order merge and stop once the DEDUPED count reaches the limit (mergedResult.Len() < topN). Capping on the raw pre-dedup row count would drop unique rows the oracle keeps whenever a duplicate ElementID falls inside the retained window (an ElementID can span parts). Rows are truncated at the sort position where the maxRows-th distinct ElementID first appears, so the retained prefix carries every row (including still-undeduped duplicates) up to that boundary; the downstream Distinct then removes the duplicates from that already-correct top-N prefix.
func (*SortedMerge) Init ¶
func (s *SortedMerge) Init(context.Context) error
Init sizes the batch size and prepares the output pool.
func (*SortedMerge) NextBatch ¶
func (s *SortedMerge) NextBatch(ctx context.Context) (*vectorized.RecordBatch, error)
NextBatch emits the next key-ordered output batch. Empty input returns EOF immediately.
func (*SortedMerge) OutputSchema ¶
func (s *SortedMerge) OutputSchema() *vectorized.BatchSchema
OutputSchema returns the shared input/output schema.
type VectorizedConfig ¶
type VectorizedConfig struct {
BatchSize int
// QueryMemoryMiB is a soft element-loading threshold: it caps the cumulative
// uncompressed element bytes fetched from disk per query. Tags, record-batch
// overhead, and other per-query allocations are not counted. The first block
// always loads regardless of the budget (first-block exception), so a single
// oversized block may exceed this value.
QueryMemoryMiB int
Enabled bool
}
VectorizedConfig controls the v1 vectorized Stream query path.
func DefaultConfig ¶
func DefaultConfig() VectorizedConfig
DefaultConfig returns the default stream vectorized configuration — enabled, with the shared default batch size and a 256 MiB per-query memory budget.
The query layer dedups by element_id and holds the seen-set for the whole query, because an element_id identifies an element GLOBALLY: two rows carrying one element_id are one element, whatever their timestamps or which parts they were read from. test/cases/stream's "deduplication test" pins that — 50 records over 27 distinct ids at 50 different timestamps must come back as 27 rows.
The row path uses the same key but allocates its seen-set per merge round (blockCursorHeap.merge / model.MergeStreamResults, one per runTabScanner call), so it only collapses duplicates that land in the same round. That is a weaker guarantee than this path gives, not a different semantic; a fixture that reuses an element_id across two writes is malformed either way.
Enabled also selects the liaison<->data wire format: a flag-on distributed data node emits the native columnar frame instead of protobuf. A liaison decodes both (it dispatches on the frame magic byte per message), but an older liaison has no frame decoder at all, so a cluster must upgrade liaison nodes BEFORE data nodes. See docs/operation/upgrade.md.
To roll back the vec path entirely, pass --stream-vectorized-enabled=false on the standalone or data-node command line and restart; the row path resumes immediately.
func (VectorizedConfig) Validate ¶
func (c VectorizedConfig) Validate() error
Validate rejects invalid stream vectorized configurations.
Source Files
Directories