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(ctx 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 TagFilter
- 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, preMerge ...vectorized.FusibleOperator, ) (*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.
preMerge fusibles run on the RAW source batches, before the merge consumes them. A row-level filter belongs here rather than at the egress: once the merge sees only surviving rows, maxRows bounds the top-N of the FILTERED set, which is what makes a cap sound for a criteria query at all. A pre-merge fusible must be row-level and must never signal ErrLimitExhausted, which would truncate the scan.
Filtering first also settles duplicate ElementIDs. The criteria is evaluated before Distinct picks a winner, so an element is represented by its first MATCHING row in the requested sort order, and one matching version makes the element eligible even when another version fails. Two rows of one element commonly carry the same ordered-tag value and therefore the same sort key; among equal keys the first row the scan yields wins, in both directions.
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 own in-order merge: blockHeap.merge / MergeStreamResults for timestamp order, idxResult.loadSortingData + mergeByTagValue for index order. 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, and for a criteria query already FILTERED) 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 — the correct in-order top-N. A maxRows of 0 means unbounded (equivalent to NewSortedMerge).
The cap is enforced INCREMENTALLY: Consume prunes the buffer back to the in-order top-N whenever it grows past 2*maxRows, so merge state stays O(maxRows) rows — peaking just under 2*maxRows plus the batch being consumed — and sort work is O(N log maxRows) rather than O(N log N).
This yields the same QUERY result as capping after one full sort (the emitted rows are the deduped top-N rather than a duplicate-carrying prefix of it, which the downstream Distinct makes indistinguishable). A row dropped by a prune already sits behind maxRows distinct ElementIDs. Rows consumed later may sort to either side of it, but neither side can rescue it: rows sorting ahead only add to the distinct ElementIDs preceding it, and rows sorting behind cannot remove any. Its distinct rank therefore never improves, so a row a prune drops is one the full sort would have dropped too. Equal sort keys cannot reorder across a prune either, because ties break on the globally monotonic arrive sequence. The result still 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(ctx context.Context, batch *vectorized.RecordBatch) error
Consume buffers references to every active row in the batch, then prunes the buffer back to the in-order top-N once it outgrows the prune threshold. 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). What survives is the FIRST row of each of the maxRows lowest-sorting distinct ElementIDs — an element's first row in sort order is the only one the downstream Distinct can ever emit, so later duplicates are dropped here rather than carried.
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 TagFilter ¶ added in v0.11.1
type TagFilter struct {
// contains filtered or unexported fields
}
TagFilter narrows a batch's selection to the rows a criteria tag filter keeps. Applied BEFORE SortedMerge, it is what makes a merge cap sound for a criteria query: the merge then bounds the top-N of the FILTERED set instead of filtering a set that was already truncated.
It evaluates the same logical.TagFilter the egress evaluates, over the same cells. Hidden-tag stripping is NOT done here — it operates on materialized Elements and stays at the egress.
func NewTagFilter ¶ added in v0.11.1
func NewTagFilter(schema *vectorized.BatchSchema, projection []model.TagProjection, filter logical.TagFilter, registry logical.TagSpecRegistry, ) *TagFilter
NewTagFilter builds the pre-merge criteria filter. projection MUST be the same []model.TagProjection the egress materializes from and registry MUST be the projected schema the *tagFilterPlan carries: together they define the (tagFamilyIdx, tagIdx) space the filter indexes with.
Columns are resolved by NAME, exactly as BuildElementsFromBatch resolves them. Positional resolution would be wrong wherever the executed schema carries a column the client projection does not (an ordered tag added for the OrderKey).
func (*TagFilter) OutputSchema ¶ added in v0.11.1
func (t *TagFilter) OutputSchema() *vectorized.BatchSchema
OutputSchema returns the unchanged input schema.
func (*TagFilter) Process ¶ added in v0.11.1
func (t *TagFilter) Process(_ context.Context, batch *vectorized.RecordBatch) error
Process rewrites the selection to the matching rows. A Match error aborts the query, matching the egress filter.
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 reused element_id is supported rather than malformed: for a filtered index-order query the criteria is evaluated first and the element is represented by its first MATCHING row in the requested sort order — see BuildStreamMergePipeline.
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