measure

func AppendColumnRange ¶

func AppendColumnRange(dst, src vectorized.Column, srcPos, n int) error

AppendColumnRange delegates to the shared vectorized helper.

func ApplyTopToReduce ¶

func ApplyTopToReduce(reduced []*vectorized.RecordBatch, spec ReduceTopSpec, batchSize int) ([]*vectorized.RecordBatch, error)

ApplyTopToReduce composes the Reduce output through a BatchTop operator, returning the top-N rows by FieldName. This is the liaison-side completion of the distributed Top-over-Agg pattern: data nodes emit AggModeMap partials (optionally pre-topped via the per-node BatchTop), the liaison Reduces them, then a second BatchTop selects the global top-N. Mirrors the row path's two-pass Top approach — sorting by partial value on each node, then re-sorting the merged reductions on the coordinator.

fieldName MUST match a RoleField column in the reduced output schema (typically the AggReduceSpec.OutputName). N <= 0 returns reduced input unchanged.

func BuildBatchSchema ¶

func BuildBatchSchema(measureSchema *databasev1.Measure, opts model.MeasureQueryOptions) (*vectorized.BatchSchema, error)

BuildBatchSchema derives a BatchSchema from a Measure schema and a query's projection list. The output column order is fixed:

timestamp, version, sid, shardID, then projected tags (in TagProjection
order, family-by-family, name-by-name), then projected fields (in
FieldProjection order).

Tag families and tag names that are not present in the Measure schema are dropped — the row path silently skips unknown tags as well, so this matches existing semantics. Fields not present in the schema yield a Null-typed column so projection still produces a slot in the output.

Column types — passthrough vs native (G8d.2):

Tag and field projections default to passthrough columns: the column cell type is *modelv1.TagValue / *modelv1.FieldValue, holding the original protobuf pointer from the scan source unchanged. The egress serializer returns those pointers directly, matching the row path's zero-alloc per-cell behavior. With the gRPC wire format frozen (`*measurev1.InternalDataPoint` is row-shaped), passthrough wins for plain scans because native columns would force egress to reconstruct the protobuf wrapper (3 allocs/cell), regressing the G5a bench gates.

When opts.GroupBy or opts.Agg name a projected tag or field, that specific column is emitted as a NATIVE typed column instead. The downstream BatchAggregation operator reads typed primitives directly from those columns (computeKey / fold) and produces aggregated rows whose count is bounded by group cardinality, amortizing the eventual wrapper reconstruction at egress. Columns not referenced by GroupBy / Agg remain passthrough.

func BuildMeasureBatchFromResult ¶

func BuildMeasureBatchFromResult(r *model.MeasureResult, schema *vectorized.BatchSchema) (*model.MeasureBatch, error)

BuildMeasureBatchFromResult converts a *model.MeasureResult produced by a row-shaped MeasureQueryResult.Pull() call into a *model.MeasureBatch whose columns match the supplied BatchSchema.

This is the G5b "dual-emit" wrapper helper: the storage layer can implement MeasureBatchResult.PullBatch by calling its existing Pull() and passing the result through this converter. It preserves the existing row-path decode pipeline; the architectural decode-elimination is left to G5c/G5d (block_cursor native column emit).

Schema-declared tag/field columns missing from the result are null-filled using pbv1.Null{Tag,Field}Value singletons — matching the multi-group projection behavior in fillTags / fillFields when one group's schema lacks a tag the other has.

Returns (nil, nil) when r is nil. Returns (nil, err) when a length invariant is violated (a value slice is shorter than the timestamp count).

func BuildOperators ¶

func BuildOperators(
	opts model.MeasureQueryOptions, schema *vectorized.BatchSchema,
	tracker *vectorized.MemoryTracker, batchSize int, mode AggMode,
) ([]vectorized.BreakerOperator, error)

BuildOperators translates the GroupBy + Agg slice of MeasureQueryOptions into a list of BreakerOperators that, chained after the scan source, form the vectorized aggregation pipeline.

Routing rules:

• GroupBy + Agg both set → emit a single BatchAggregation. The operator's own keyIndex map produces per-group buckets and folds the agg slot; a separate BatchGroupBy would double the row materialization.
• Agg set without GroupBy → scalar reduce: a BatchAggregation with no key columns. Every row maps to one group, so a single output row is emitted carrying the first-seen projected tags plus the agg result, matching the row path's aggAllIterator.
• GroupBy set without Agg → raw GroupBy: a first-seen-row-per-group BatchGroupBy. The output preserves the input schema; one row per group is emitted in group-insertion order, matching the row path's groupIterator + processor.go's current[0] read.
• Neither set → empty operator list; caller emits raw rows.

tracker is the per-pipeline MemoryTracker (G7a); it must be non-nil when any operator is emitted so per-group reservations route to the shared budget. batchSize controls the operator's output pagination.

mode selects the BatchAggregation strategy when an agg operator is built (Agg set): AggModeAll for single-node final reduce; AggModeMap for the distributed Map phase (G9f.2) that emits typed-column partials. mode is ignored for the raw-GroupBy branch (BatchGroupBy doesn't carry partial state — first-seen-row per group is already the final shape). AggModeReduce is rejected here loudly; its operator is built by the liaison-side reduce plan (G9f.3), not via BuildOperators.

func DecodeFramesPerSource ¶

func DecodeFramesPerSource(frames [][]byte) ([][]*measurev1.InternalDataPoint, error)

DecodeFramesPerSource decodes a sequence of vec raw frame bodies and returns one []*measurev1.InternalDataPoint slice per non-empty frame — preserving the per-data-node grouping the row path's sortedMIterator + sort.NewItemIter merger needs to dedup replicas. Each data node's scan output is internally sort-field ordered; sort.NewItemIter merges the per-source streams and loadOneGroup's hashDataPoint map removes (sid, timestamp) duplicates within each equal-sortField group. Concatenating into a single slice (as DecodeFramesToInternalDataPoints does) defeats the dedup because duplicate rows from different sources end up in different sortField groups across the flat sequence.

nil / empty frame bodies produce no slice in the output — the codec carve-out for empty bodies is honored here too.

func DecodeFramesToInternalDataPoints ¶

func DecodeFramesToInternalDataPoints(frames [][]byte) ([]*measurev1.InternalDataPoint, error)

DecodeFramesToInternalDataPoints decodes a sequence of vec raw frame bodies and concatenates their active rows into a single []*measurev1.InternalDataPoint, ready to feed the row-side pushedDownAggregatedIterator (agg path's flatten step).

For the NON-agg distributed merge path use DecodeFramesPerSource instead — concatenation here destroys the per-source ordering the sortedMIterator's cross-iterator merge + (sid, timestamp) dedup depends on, which is what caused replica duplicates to slip past the dedup map under flag-on.

nil/empty bodies are skipped — the codec layer's RawFrameCodec carve-out returns nil for an empty distributed result and the decoder is below that carve-out.

func DrainPipelineToFrame ¶

func DrainPipelineToFrame(ctx context.Context, p *vectorized.Pipeline, schema *vectorized.BatchSchema) ([]byte, error)

DrainPipelineToFrame consumes a vec Pipeline end-to-end, concatenates every emitted batch into a single RecordBatch, and frame.Encode-s the result. It is the data-node side of the G9f throughout-vec wire: under flag-on, TopicInternalMeasureQuery responses are exactly this single raw frame body (one row per group for AggModeMap, one row per source row for non-agg queries).

schema is the propagated terminal-operator output schema (typically the same as the egress pool's). It MUST match each emitted batch — a mismatch is a planner bug, not a data error, so this function will fail loud on a per-batch shape check instead of silently coercing.

Multi-batch coalesce: pipeline.Next is called until it returns nil; every non-nil batch's active rows are appended to a single output batch via copyOneValue. This is intentionally NOT a streaming-write of multiple frames — the codec contract carries one body per response (api/data/codec.go), and frame.Encode expects one batch. For typical distributed agg responses (one row per group, group count ≤ batchSize) the pipeline emits a single batch and coalesce is a no-op fast path.

Returns the encoded frame body and the close error from pipeline.Close joined into a single error (mirrors vectorizedMIterator.Close).

func FormatReduceOutput ¶

func FormatReduceOutput(b *vectorized.RecordBatch) string

FormatReduceOutput is a debug helper for the topology matrix harness: it renders the reduced batch as "key1=v1,key2=v2|<aggname>=<val>" lines for stable, sorted comparison against the AggModeAll oracle.

func ReduceFramesToInternalDataPoints ¶

func ReduceFramesToInternalDataPoints(
	frames [][]byte,
	keyTagNames []string,
	aggSpecs []AggReduceSpec,
	topSpec *ReduceTopSpec,
	batchSize int,
	tracker *vectorized.MemoryTracker,
) ([]*measurev1.InternalDataPoint, error)

ReduceFramesToInternalDataPoints is the liaison-side composition for distributed agg queries under flag-on: it decodes the per-data-node partial frames, runs the (shard, group)-deduped Reduce, optionally applies a final Top-N (when topSpec.N > 0), and serializes the resulting batches back to []*measurev1.InternalDataPoint so the row-side pushedDownAggregatedIterator + downstream MIterator surface stay unchanged.

The conversion back to proto is the price of bridging vec output into the row-path liaison's existing iterator. Once the row-path distributedPlan is replaced by a fully vec-distributed plan (out of scope for G9f.5), this round trip drops out.

Empty input (no non-empty frames) returns an empty slice with no error — matches the row path's behavior when every data node returned an empty distributed result.

func SerializeDataPointsToFrame ¶

func SerializeDataPointsToFrame(idps []*measurev1.InternalDataPoint) ([]byte, error)

SerializeDataPointsToFrame is the fallback wire-emit path for iterator wrappers (hiddenTagsMIterator, sortedMIterator) whose internal sort / strip / cross-group merge logic operates on []*InternalDataPoint rather than on a vec Pipeline. The wrapper drains itself via the row-side Next / Current API (so its existing strip / merge / dedup logic still runs); the resulting rows are reverse-serialized into a passthrough RecordBatch, convertPassthroughForFrame decodes the passthrough columns to typed wire columns, and frame.Encode produces the body.

This path is less efficient than the vec native pipeline drain — it allocates a *modelv1.TagValue / *FieldValue per cell during reverse- serialize — but it keeps the wrapper's egress semantics intact end- to-end on the wire (hidden tags stripped, cross-group order honored, version dedup applied) without re-implementing each one in columnar form.

idps in the empty/zero case yields a nil body, matching the codec layer's RawFrameCodec carve-out for empty distributed results.