aggregate

type AccumOp ¶

type AccumOp uint8

AccumOp identifies a $group accumulator operator.

const (
	AccumSum AccumOp = iota
	AccumAvg
	AccumMin
	AccumMax
	AccumCount
	AccumFirst
	AccumLast
	AccumPush
	AccumAddToSet
)

func (op AccumOp) String() string

type AccumSpec ¶

type AccumSpec struct {
	Name string
	Op   AccumOp
	Arg  Expr
}

AccumSpec is one accumulator of a $group stage. Arg is nil for $count.

type Accumulator ¶

type Accumulator interface {
	Step(val *anyenc.Value) error
	Finalize(out *anyenc.Arena) (*anyenc.Value, error)
}

Accumulator folds the per-row values of one $group output field.

Step receives the already-evaluated argument expression (nil = missing); the value is transient — valid only during the call — so implementations retain data by marshaling into self-owned buffers. Finalize is called once, after the upstream drain; results may be allocated on out or on an accumulator-owned parser (the stage consumes the result before the next emit either way).

type AddFieldsSpec ¶

type AddFieldsSpec struct{ Fields []ProjectField }

func (s AddFieldsSpec) String() string

type AddFieldsStage ¶

type AddFieldsStage struct {
	Src    Stage
	Fields []ProjectField
	// contains filtered or unexported fields
}

AddFieldsStage overlays computed fields onto each upstream document by mutating it in place (the same pattern FtsIter uses to inject _score): zero-copy, no output object rebuild. Mutation is safe because rows are transient — the underlying parse buffers are rewritten on the next row.

When an upstream $unwind may emit the same underlying document for several consecutive rows (undoOverlay, set by Build), the original value of every overlaid field is captured and rolled back before the next row is pulled: otherwise a later row of the same document would evaluate its expressions against the previous row's overlays instead of the stored fields.

func (s *AddFieldsStage) Close()

func (s *AddFieldsStage) Next(ctx *Ctx) (*anyenc.Value, error)

func (s *AddFieldsStage) String() string

type ArrayExpr ¶

type ArrayExpr struct {
	Exprs []Expr
}

ArrayExpr is an array expression: every element is an expression. Missing elements evaluate to null so the array length is preserved.

func (e *ArrayExpr) Eval(a *anyenc.Arena, doc *anyenc.Value) (*anyenc.Value, error)

func (e *ArrayExpr) String() string

type CountSpec ¶

type CountSpec struct{ Field string }

func (s CountSpec) String() string

type CountStage ¶

type CountStage struct {
	Src   Stage
	Field string
	// contains filtered or unexported fields
}

CountStage drains the upstream stream and emits a single {"<field>": N} document.

func (s *CountStage) Close()

func (s *CountStage) Next(ctx *Ctx) (*anyenc.Value, error)

func (s *CountStage) String() string

type Ctx ¶

type Ctx struct {
	Context  context.Context
	RowArena *anyenc.Arena
	Buf      *syncpool.DocBuffer
	Mem      *MemAccount
}

Ctx carries per-pipeline shared state through the stage chain.

RowArena ownership: the arena is reset only by the current "row owner" — the source stage (once per source row), a blocking stage on its output side (once per emitted row), and $project/$addFields when Build proves nothing below them keeps arena-allocated data alive across pulls. Non-owning stages allocate on RowArena freely and never reset it, so synthesized values and zero-copy aliases of the source document stay valid for the whole life of the row.

type Expr ¶

type Expr interface {
	// Eval returns the expression value for doc; nil means "missing".
	// Results may be allocated on a; field refs alias doc (zero-copy) and are
	// valid only while doc is.
	Eval(a *anyenc.Arena, doc *anyenc.Value) (*anyenc.Value, error)
	fmt.Stringer
}

Expr is an aggregation expression evaluated against one document.

v1 supports field references ("$a.b.c"), literals ({"$literal": x} or any plain value), and document/array expressions (objects and arrays whose members are themselves expressions, Mongo semantics). Compute operators ($add, $cond, ...) are additive future work: they become composite Expr implementations dispatched in ParseExpr.

func ParseExpr(v *anyenc.Value) (Expr, error)

type FieldRefExpr ¶

type FieldRefExpr struct {
	Field string // original spelling without "$", for String()
	Path  []string
}

FieldRefExpr resolves a pre-split document path ("$a.b.c").

func (e *FieldRefExpr) Eval(_ *anyenc.Arena, doc *anyenc.Value) (*anyenc.Value, error)

func (e *FieldRefExpr) String() string

type GroupSpec ¶

type GroupSpec struct {
	Key    Expr
	Accums []AccumSpec
}

func (s GroupSpec) String() string

type GroupStage ¶

type GroupStage struct {
	Src    Stage
	Spec   GroupSpec
	Limits Limits
	// contains filtered or unexported fields
}

GroupStage implements $group: a blocking hash aggregation.

Group keys are the marshaled anyenc bytes of the key expression (missing → null, Mongo semantics); the byte encoding is canonical, so byte equality is value equality. Lookups are alloc-free (map[string]X with a string(bytes) conversion the compiler elides); new keys are interned into a grow-only buffer with unsafe.String map keys (the query.In pattern). Group states live in one slab slice in first-seen order, which is also the emit order (unspecified for callers).

func (s *GroupStage) Close()

func (s *GroupStage) Next(ctx *Ctx) (*anyenc.Value, error)

func (s *GroupStage) String() string

type LimitSpec ¶

type LimitSpec struct{ N int }

func (s LimitSpec) String() string

type LimitStage ¶

type LimitStage struct {
	Src Stage
	N   int
	// contains filtered or unexported fields
}

LimitStage emits at most N rows and stops pulling upstream afterwards.

func (s *LimitStage) Close()

func (s *LimitStage) Next(ctx *Ctx) (*anyenc.Value, error)

func (s *LimitStage) String() string

type Limits ¶

type Limits struct {
	// MaxGroups caps the number of unique $group keys.
	MaxGroups int
	// MaxAccumArrayLen caps $push / $addToSet result arrays.
	MaxAccumArrayLen int
	// MaxMemoryBytes caps retained bytes across all blocking stages.
	MaxMemoryBytes int
}

Limits bounds the memory-unbounded parts of a pipeline. Zero values mean "use default"; negative values mean "unlimited".

func (l Limits) WithDefaults() Limits

type LiteralExpr ¶

type LiteralExpr struct {
	// contains filtered or unexported fields
}

LiteralExpr yields the same constant value for every document. The value is detached from the input at parse time (owned bytes + dedicated parser), so it stays valid for the lifetime of the pipeline regardless of caller arenas.

func NewLiteralExpr(v *anyenc.Value) (*LiteralExpr, error)

func (e *LiteralExpr) Eval(_ *anyenc.Arena, _ *anyenc.Value) (*anyenc.Value, error)

func (e *LiteralExpr) String() string

type MatchSpec ¶

type MatchSpec struct{ Filter query.Filter }

func (s MatchSpec) String() string

type MatchStage ¶

type MatchStage struct {
	Src    Stage
	Filter query.Filter
}

MatchStage filters the upstream stream with a query filter. Matched rows are passed through untouched (1:1, no arena allocations); discarded rows are freed by the source's per-row arena reset on the following pull.

func (s *MatchStage) Close()

func (s *MatchStage) Next(ctx *Ctx) (*anyenc.Value, error)

func (s *MatchStage) String() string

type MemAccount ¶

type MemAccount struct {
	// contains filtered or unexported fields
}

MemAccount is a coarse retained-bytes budget shared by the blocking stages of one pipeline ($group state, $sort arena). Streaming stages retain nothing and don't report.

func NewMemAccount(limit int) *MemAccount

func (m *MemAccount) Add(n int) error

func (m *MemAccount) Used() int

type ObjectExpr ¶

type ObjectExpr struct {
	Names []string
	Exprs []Expr
}

ObjectExpr is a document expression: an object whose values are expressions (Mongo semantics for non-operator objects in expression context, e.g. compound $group keys {"a":"$x","b":"$y"}). Missing member values are omitted from the result, like Mongo.

func (e *ObjectExpr) Eval(a *anyenc.Arena, doc *anyenc.Value) (*anyenc.Value, error)

func (e *ObjectExpr) String() string

type Pipeline ¶

type Pipeline []StageSpec

Pipeline is a parsed aggregation pipeline: an ordered list of stage specs.

func MustParsePipeline(pipeline any) Pipeline

func ParsePipeline(pipeline any) (Pipeline, error)

type Prefix ¶

type Prefix struct {
	Filter query.Filter // folded leading $match chain; nil if none
	Sort   query.Sort   // nil if no $sort was pushed
	Skip   int
	Limit  int // 0 = none
}

Prefix is the leading pipeline part that can be pushed down into the access plan (the regular Find machinery: index/CBO selection, FTS/vector sources, index-order sorts, cursor-level offset skips).

type ProjectField ¶

type ProjectField struct {
	Name string
	Expr Expr
}

ProjectField is one output field of $project / $addFields.

type ProjectSpec ¶

type ProjectSpec struct{ Fields []ProjectField }

func (s ProjectSpec) String() string

type ProjectStage ¶

type ProjectStage struct {
	Src    Stage
	Fields []ProjectField
	// contains filtered or unexported fields
}

ProjectStage replaces each upstream document with a new object containing only the specified fields. The output is built on RowArena; field values alias subtrees of the input document (zero-copy), which is safe because the input stays valid for the whole row lifetime.

func (s *ProjectStage) Close()

func (s *ProjectStage) Next(ctx *Ctx) (*anyenc.Value, error)

func (s *ProjectStage) String() string

type SkipSpec ¶

type SkipSpec struct{ N int }

func (s SkipSpec) String() string

type SkipStage ¶

type SkipStage struct {
	Src Stage
	N   int
	// contains filtered or unexported fields
}

SkipStage drops the first N upstream rows.

func (s *SkipStage) Close()

func (s *SkipStage) Next(ctx *Ctx) (*anyenc.Value, error)

func (s *SkipStage) String() string

type SortSpec ¶

type SortSpec struct {
	Sort   query.Sorts
	Fields []query.SortField // parsed order, for String()/pushdown checks
}

func (s SortSpec) String() string

type SortStage ¶

type SortStage struct {
	Src  Stage
	Spec SortSpec
	TopK int
	// contains filtered or unexported fields
}

SortStage is the blocking in-pipeline $sort over synthesized documents (stored-document sorts are pushed down to the access plan instead).

It mirrors qplanner.SortIter's packed-arena design, but each entry carries the full marshaled document after its sort-key tuple — rows here no longer correspond to stored docIds. With TopK > 0 (a following $skip/$limit was folded in by Build) a max-heap keeps only the smallest K entries: a losing row is built in the arena's spare tail and truncated away — zero retained bytes — and a fragmentation guard re-packs the arena when dead bytes from evictions exceed live data, keeping the high-water mark O(K).

Stability: the full sort uses slices.SortStableFunc over insertion-ordered entries; the top-K path appends a row sequence number to each key, which both totals the order (deterministic heap eviction: of equal keys the earliest rows survive) and makes the final sort stable by construction.

func (s *SortStage) Close()

func (s *SortStage) Next(ctx *Ctx) (*anyenc.Value, error)

func (s *SortStage) String() string

type Stage ¶

type Stage interface {
	// Next returns the next document, or (nil, nil) at the end of the stream.
	// The returned value is valid only until the next Next call on this stage.
	Next(ctx *Ctx) (*anyenc.Value, error)

	// Close releases stage resources. It must be safe to call multiple times
	// and after a mid-stream abort.
	Close()

	// String describes the stage for Explain output. Never called on the hot
	// path.
	fmt.Stringer
}

Stage is a pull-driven aggregation operator over anyenc documents.

The chain is strictly synchronous: a stage produces a row only when its consumer asks for one, so row lifetimes are phase-locked across the chain.

func Build(source Stage, specs Pipeline, limits Limits) (Stage, error)

type StageSpec ¶

type StageSpec interface {
	fmt.Stringer
	// contains filtered or unexported methods
}

StageSpec is a parsed pipeline stage description (not yet an executable Stage; see Build).

type UnwindSpec ¶

type UnwindSpec struct {
	Field                      string // original spelling without "$"
	Path                       []string
	PreserveNullAndEmptyArrays bool
}

func (s UnwindSpec) String() string

type UnwindStage ¶

type UnwindStage struct {
	Src                        Stage
	Field                      string
	Path                       []string
	PreserveNullAndEmptyArrays bool
	// contains filtered or unexported fields
}

UnwindStage emits one row per element of an array field, replacing the array with the current element by mutating the document in place: the element values stay alive because they belong to the original (detached) array value, while the document's field slot is repointed per emission. No arena allocations, no document rebuild.

Before pulling the next source row the original array is restored into the field slot. An upstream stage may hold and re-emit the same document (another $unwind multiplying rows does), so leaving the field pointing at the last element would corrupt every later row of that document.

Mongo semantics: missing / null / empty array drop the document (unless PreserveNullAndEmptyArrays, which emits it as-is — with the field removed for an empty array); a non-array value passes through unchanged.

func (s *UnwindStage) Close()

func (s *UnwindStage) Next(ctx *Ctx) (*anyenc.Value, error)

func (s *UnwindStage) String() string