exec

package
v0.13.0-late-materiali... Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 16, 2026 License: AGPL-3.0 Imports: 25 Imported by: 0

Documentation

Overview

Package exec provides the push-based pipeline execution framework.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// LateMatBatchesEmitted counts join-probe output batches emitted in view
	// (reference) form instead of eagerly gathered.
	LateMatBatchesEmitted atomic.Int64
	// LateMatViewColumns counts view columns across those batches.
	LateMatViewColumns atomic.Int64
	// LateMatFlattens counts FlattenViews calls made on behalf of consumers
	// that don't accept views — each one is a deferred gather finally paid.
	LateMatFlattens atomic.Int64
	// LateMatViewColumnsSerialized counts view columns serialized straight
	// through their indirection by the shuffle writers (no flatten copy at
	// all) — the phase-4 engagement marker.
	LateMatViewColumnsSerialized atomic.Int64
)

Late-materialization runtime counters. Process-wide, monotonic — they exist so every A/B arm can PROVE the treatment engaged (the dynamic-filter lesson: a silent feature that never fires reads as a wash). Zero while the flag is off is the dormancy assertion; >0 under the flag is the engagement marker.

View Source 
var ErrCollectBudget = errors.New("collect sink result budget exceeded")

ErrCollectBudget is returned by CollectSink.Consume when the accumulated result exceeds CollectSink.MaxBytes. Callers match with errors.Is to distinguish "result too big for this execution path" from real failures.

View Source 
var KeyAssignmentRepairs atomic.Int64

KeyAssignmentRepairs counts runtime probe/build key swaps performed by FixKeyAssignment after a build completed — cases where PLAN-TIME side assignment (assignJoinKeySides) got a pair wrong and the runtime safety net rescued it. Tripwire observability: the TPC-H suites assert this stays 0; a nonzero count means a plan shape leaked through the planner's ownership resolution (rebuild cost + a hazard for partitioned builds).

View Source 
var PartitionedAggRuns atomic.Int64

PartitionedAggRuns counts pipelines that ran in partitioned mode (observability + test assertions).

View Source 
var SMJCounterpartAdoptions atomic.Int64

SortMergeJoin joins two large inputs by sorting both sides on the join keys and streaming a two-cursor merge. Unlike HashJoin, neither side is held resident: each side buffers under the shared tracker and self-spills sorted columnar runs (Sort's external-merge machinery), so peak memory is O(run buffer + one batch per merge cursor) regardless of input size.

The build side (right, as in HashJoin) arrives via Build; the probe side (left) arrives via Consume. Both are pipeline breakers here — inherent to sort-based joins over unsorted input. After Finalize, Next streams joined batches: probe columns first, then build columns, with the same duplicate-name qualification and OutputFilter semantics as HashJoinProbe.

v1 scope (docs/design/sort-merge-join.md): INNER equi-joins only, no JoinFilter. Rows with a NULL in any join key are excluded at buffer time (SQL equi-join semantics: NULL matches nothing — mirrors the hash paths, where null keys produce no index entry). Not Cloneable: the breaker path runs it single-consumer. SMJCounterpartAdoptions counts key resolutions that only succeeded by adopting the OTHER side's key name (swapped pair) — the sort-merge analog of KeyAssignmentRepairs. Tripwire: should stay 0 on planner-produced plans.

Functions

func BloomAddBatch 

func BloomAddBatch(bloom []uint64, bloomMask uint64, b *batch.RecordBatch, keyCol string)

BloomAddBatch hashes one batch's key column into the bloom. Incremental counterpart of BuildBloomFromBatches, used when the source batches stream from a spill-backed collector instead of sitting in memory all at once.

func BloomContains 

func BloomContains(bloom []uint64, mask, hash uint64) bool

BloomContains checks if a hash may be in the bloom filter. Exported for use by the scan layer for row-group-level pruning.

func BloomHashInt 

func BloomHashInt(key int64) uint64

BloomHashInt computes the bloom filter hash for an integer key. Exported for use by the scan layer for row-group-level pruning.

func BuildBloomFromBatches 

func BuildBloomFromBatches(batches []*batch.RecordBatch, keyCol string) (bloom []uint64, bloomMask uint64)

BuildBloomFromBatches constructs a bloom filter from a column across multiple batches. Returns nil bloom if no rows. Used for reverse bloom pushdown: the probe side's join key values filter the build side's scan.

func ColumnIndexFallback 

func ColumnIndexFallback(b *batch.RecordBatch, name string) int

ColumnIndexFallback is the exported alias for columnIndexFallback so other packages (worker shuffle sinks) can resolve column names with the same bidirectional table-qualifier fallback.

func FlattenForConsumer 

func FlattenForConsumer(b *batch.RecordBatch, consumer any)

FlattenForConsumer materializes any view columns before handing the batch to a consumer that hasn't declared view-awareness. This is the structural correctness guarantee: no operator or sink ever sees a view unless it opted in, so a missed call site degrades to an eager copy, never to reading nil typed slices. Exported because manual operator-chain loops exist outside this package (physical.pipelineSource, the worker fragment runner) and must apply the same guard the Pipeline loops do.

func FormatAnalyzeStats 

func FormatAnalyzeStats(stats []*ProfileStats) []string

FormatAnalyzeStats formats profiling stats as annotation lines. Each entry becomes: " OperatorName (actual rows=N, time=Xms, calls=N)"

func GatherColumn 

func GatherColumn(dst, src *batch.Vector, sel []uint32)

GatherColumn copies selected rows from src to contiguous dst positions. Exported for use by aggPreProject materialize.

func NewBloomSized 

func NewBloomSized(totalRows int) (bloom []uint64, bloomMask uint64)

NewBloomSized allocates a bloom filter for totalRows keys (~10 bits per key for ~1% FPR). Returns nil bloom for zero rows.

func WithProgressReporter 

func WithProgressReporter(ctx context.Context, p ProgressReporter) context.Context

WithProgressReporter attaches a reporter to ctx. The worker's task dispatch path calls this; everyone else reads via ProgressReporterFromContext.

Types

type AggColumn 

type AggColumn struct {
	Func       AggFunc
	InputCol   string // input column name (empty for COUNT(*))
	OutputCol  string // output column name
	OutputType parquet.TypeID
	Separator  string  // separator for STRING_AGG (default ',')
	InputCol2  string  // second input column (corr, covar, min_by, max_by)
	Percentile float64 // percentile value for percentile_cont/percentile_disc
}

AggColumn defines an aggregation to perform.

type AggFunc 

type AggFunc int

AggFunc identifies an aggregate function. 

const (
	AggSum AggFunc = iota
	AggCount
	AggMin
	AggMax
	AggAvg
	AggCountDistinct
	AggStringAgg
	AggBoolAnd
	AggBoolOr
	AggStddev
	AggVariance
	AggStddevPop
	AggVarPop
	AggApproxDistinct
	AggCorr
	AggCovarSamp
	AggCovarPop
	AggPercentileCont
	AggPercentileDisc
	AggMode
	AggMinBy
	AggMaxBy
	AggMedian
)

type BatchSink 

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

CollectSink collects all consumed batches. Data is stored columnar internally. Rows are converted lazily on first access to ToRows(), not during Finalize. Use Batches() for zero-copy columnar access. Thread-safe: Consume() is protected by a mutex for parallel pipeline workers. BatchSink is a Sink that only stores RecordBatches, never converting to rows. Use it from internal pipelines (e.g. reverseBloomBridge) that consume the batches directly and never need the row representation. CollectSink's Finalize unconditionally calls ToRows() for backward compatibility, which is unnecessary work and — more importantly — panics if any batch has a row count beyond the underlying bitmap capacity (as we hit when the reverseBloomBridge collected probe-side batches at SF100).

func (*BatchSink) Batches 

func (s *BatchSink) Batches() []*batch.RecordBatch

Batches returns the collected RecordBatches. Safe to call after Finalize.

func (*BatchSink) Close 

func (s *BatchSink) Close() error

func (*BatchSink) Consume 

func (s *BatchSink) Consume(_ context.Context, b *batch.RecordBatch) error

func (*BatchSink) Finalize 

func (s *BatchSink) Finalize(_ context.Context) error

func (*BatchSink) Init 

func (s *BatchSink) Init(_ context.Context) error

type BatchSource 

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

BatchSource is a source that yields pre-loaded record batches.

func NewBatchSource 

func NewBatchSource(batches []*batch.RecordBatch) *BatchSource

NewBatchSource creates a source from pre-loaded record batches.

func (*BatchSource) Close 

func (s *BatchSource) Close() error

func (*BatchSource) Init 

func (s *BatchSource) Init(_ context.Context) error

func (*BatchSource) Next 

func (s *BatchSource) Next(_ context.Context) (*batch.RecordBatch, error)

type BloomFilterOp 

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

BloomFilterOp is a UnaryOperator that pre-filters probe batches using the build-side bloom filter. Rows whose join key hash is not in the bloom are eliminated via selection vector before reaching the probe operator.

This is read-only on the bloom data, so multiple clones safely share it.

func NewBloomFilterOp 

func NewBloomFilterOp(bloom []uint64, bloomMask uint64, keys []string, useIntKey bool) *BloomFilterOp

NewBloomFilterOp creates a BloomFilterOp from pre-built bloom data. Used for reverse bloom pushdown where the probe side's key set filters the build side's scan.

func (*BloomFilterOp) BloomScanFilter 

func (op *BloomFilterOp) BloomScanFilter() *BloomScanFilter

BloomScanFilter returns a BloomScanFilter for scan-level pushdown. Only applicable for single-column integer join keys. Returns nil if not applicable.

func (*BloomFilterOp) Clone 

func (op *BloomFilterOp) Clone() UnaryOperator

Clone returns a new BloomFilterOp sharing the same bloom data.

func (*BloomFilterOp) Close 

func (op *BloomFilterOp) Close() error

func (*BloomFilterOp) Execute 

func (op *BloomFilterOp) Execute(_ context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error)

func (*BloomFilterOp) Init 

func (op *BloomFilterOp) Init(_ context.Context) error

func (*BloomFilterOp) KeyColumns 

func (op *BloomFilterOp) KeyColumns() []string

KeyColumns returns the probe-side key column names this bloom filter checks.

type BloomScanFilter 

type BloomScanFilter struct {
	Bloom     []uint64 // shared, read-only
	BloomMask uint64
	Column    string // probe-side join key column name
	UseIntKey bool   // true for single integer column join key
}

BloomScanFilter holds bloom filter data for row-group-level scan pushdown.

type BufferReusingOperator 

type BufferReusingOperator interface {
	ReusesOutputBuffers() bool
}

BufferReusingOperator marks operators whose Execute output aliases buffers reused on the NEXT call (e.g. the aggregate pre-projection's computed vectors). Their batches cannot be shared across partition owners that consume asynchronously, so partitioned aggregation is disabled when one sits in the chain.

type ChainFilter 

type ChainFilter struct {
	Ops []UnaryOperator
}

ChainFilter applies a sequence of unary filter operators in order. Each operator narrows the selection vector before passing to the next.

func NewChainFilter 

func NewChainFilter(ops []UnaryOperator) *ChainFilter

func (*ChainFilter) Clone 

func (f *ChainFilter) Clone() UnaryOperator

Clone returns a new ChainFilter with cloned sub-operators. Sub-operators that implement Cloneable are cloned; others are shared.

func (*ChainFilter) Close 

func (f *ChainFilter) Close() error

func (*ChainFilter) Execute 

func (f *ChainFilter) Execute(ctx context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error)

func (*ChainFilter) Init 

func (f *ChainFilter) Init(ctx context.Context) error

type Cloneable 

type Cloneable interface {
	Clone() UnaryOperator
}

Cloneable is implemented by operators that can be cloned for parallel pipeline execution. Each clone gets its own scratch buffers so that multiple goroutines can call Execute concurrently without data races.

type ColColFilter 

type ColColFilter struct {
	LeftCol  string
	RightCol string
	Op       CompareOp
	// contains filtered or unexported fields
}

ColColFilter compares two columns element-wise using a vectorized kernel. Resolves column indices and kernel on first Execute; inner loop has no type switches.

func NewColColFilter 

func NewColColFilter(leftCol, rightCol string, op CompareOp) *ColColFilter

func (*ColColFilter) Clone 

func (f *ColColFilter) Clone() UnaryOperator

Clone returns a new ColColFilter with the same parameters but fresh resolution state and scratch buffers for concurrent Execute calls.

func (*ColColFilter) Close 

func (f *ColColFilter) Close() error

func (*ColColFilter) Execute 

func (f *ColColFilter) Execute(_ context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error)

func (*ColColFilter) Init 

func (f *ColColFilter) Init(_ context.Context) error

type CollectSink 

type CollectSink struct {
	Rows []map[string]any // populated lazily on first access

	// MaxBytes, when >0, bounds the collected result: Consume returns
	// ErrCollectBudget once accumulated batch bytes exceed it. Callers that
	// have a cheaper place to put oversized results (the coordinator's
	// local fast path re-dispatches to the DAG, whose gather spills to
	// scratch) set this to bail out instead of growing the heap unboundedly.
	MaxBytes int64

	// SkipFinalizeToRows, when true, makes Finalize a no-op instead of
	// eagerly materializing ToRows. Callers that consume Batches() directly
	// (native-DAG worker stage path) should set this — otherwise Finalize
	// allocates a map[string]any per row of every collected batch.
	// At SF10 Q18, this single allocation pattern held 21 GB of live heap
	// (heap-1347079-130.pprof: CollectSink.ToRows = 68% inuse_space —
	// project_q18_sf10_native_dag_oom_2026-04-24).
	SkipFinalizeToRows bool
	// contains filtered or unexported fields
}

func (*CollectSink) Batches 

func (s *CollectSink) Batches() []*batch.RecordBatch

Batches returns the raw columnar batches (zero-copy, no conversion). Returns nil once ToRows has converted the result.

func (*CollectSink) Close 

func (s *CollectSink) Close() error

func (*CollectSink) Consume 

func (s *CollectSink) Consume(_ context.Context, b *batch.RecordBatch) error

func (*CollectSink) Finalize 

func (s *CollectSink) Finalize(_ context.Context) error

func (*CollectSink) Init 

func (s *CollectSink) Init(_ context.Context) error

func (*CollectSink) Schema 

func (s *CollectSink) Schema() []parquet.Column

Schema returns the schema of the first consumed batch (nil if no batch was consumed). Unlike Batches()[0].Schema, it remains available after ToRows releases the batches.

func (*CollectSink) ToRows 

func (s *CollectSink) ToRows() []map[string]any

ToRows returns all results as rows, converting from batches on first call. Each batch reference is dropped as it is boxed: holding both forms alive for the sink's lifetime doubled the result's residency (columnar + boxed) on every row-consuming path. Dropping is safe — Consume Detach()ed the batches from their pools, so the arenas that boxed TypeBytes values alias are never recycled out from under them. Batches() returns nil after this; use Schema() for post-conversion schema access.

type CollectorReplaySource 

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

CollectorReplaySource is a non-consuming exec.Source over a populated SpillableBatchCollector: spilled runs first (decoded fresh per replay), then the in-memory tail as Sel-isolated shallow clones. Next is mutex-guarded so a parallel pipeline can share one instance.

func (*CollectorReplaySource) Close 

func (s *CollectorReplaySource) Close() error

func (*CollectorReplaySource) Init 

func (s *CollectorReplaySource) Init(_ context.Context) error

func (*CollectorReplaySource) Next 

func (s *CollectorReplaySource) Next(_ context.Context) (*batch.RecordBatch, error)

type Column 

type Column = parquet.Column

Column is imported from parquet for convenience.

type ColumnPrune 

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

ColumnPrune is a lightweight UnaryOperator that drops unneeded columns from batches flowing through a pipeline. Unlike Project, it operates at the column/vector level (zero-copy) — O(keepCols) per batch, not O(rows).

func NewColumnPrune 

func NewColumnPrune(keep []string) *ColumnPrune

NewColumnPrune creates a column prune operator that keeps only the named columns.

func (*ColumnPrune) Clone 

func (c *ColumnPrune) Clone() UnaryOperator

func (*ColumnPrune) Close 

func (c *ColumnPrune) Close() error

func (*ColumnPrune) Execute 

func (c *ColumnPrune) Execute(_ context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error)

func (*ColumnPrune) Init 

func (c *ColumnPrune) Init(_ context.Context) error

type CompareOp 

type CompareOp int

CompareOp represents a comparison operation. 

const (
	OpEq CompareOp = iota
	OpNe
	OpLt
	OpLe
	OpGt
	OpGe
	OpIsNull
	OpIsNotNull
)

type DoneSignaler 

type DoneSignaler interface {
	Done() bool
}

DoneSignaler is implemented by operators (like Limit) that can signal early pipeline termination. When Done() returns true, the pipeline stops pulling from the source, enabling LIMIT pushdown without scanning the full table.

type DualSource 

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

DualSource emits a single batch with 1 row and 0 columns, then EOF. Used for table-less SELECT (e.g., SELECT CURRENT_DATE, SELECT 1+1).

func (*DualSource) Close 

func (d *DualSource) Close() error

func (*DualSource) Init 

func (d *DualSource) Init(_ context.Context) error

func (*DualSource) Next 

func (d *DualSource) Next(_ context.Context) (*batch.RecordBatch, error)

type DynamicFilterEmitOp 

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

DynamicFilterEmitOp is a pass-through UnaryOperator that accumulates a streaming partial bloom + min/max over a single integer key column. Inserted by the fragment runner at the head of a build-scan task's op chain when the OpScan spec carries a DynamicFilterEmit. Partial stats are drained on close via Snapshot and uploaded by the fragment runner.

All tasks for one emit use the same bloomBits so the coordinator unions partials with a trivial bitwise OR (no rehash).

func NewDynamicFilterEmitOp 

func NewDynamicFilterEmitOp(filterID, keyColumn, keyType string, bloomBits int) *DynamicFilterEmitOp

NewDynamicFilterEmitOp constructs an emit op. bloomBits must be a power of two; values <64 are clamped to 64. The bitset is allocated lazily on the first batch to keep the cost out of constructor paths in the planner.

func (*DynamicFilterEmitOp) AddGuard 

func (op *DynamicFilterEmitOp) AddGuard(filterID, column string, probe GuardProbe)

AddGuard attaches an upstream pending bloom to this emit. Must be called before the first Execute.

func (*DynamicFilterEmitOp) Close 

func (op *DynamicFilterEmitOp) Close() error

func (*DynamicFilterEmitOp) Execute 

func (op *DynamicFilterEmitOp) Execute(_ context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error)

func (*DynamicFilterEmitOp) FilterID 

func (op *DynamicFilterEmitOp) FilterID() string

FilterID lets the runner correlate the op back to its OpSpec entry.

func (*DynamicFilterEmitOp) FinalizeGuards 

func (op *DynamicFilterEmitOp) FinalizeGuards(ctx context.Context)

FinalizeGuards settles outstanding guards (waiting up to the ctx deadline) and flushes the buffer. Must be called before Snapshot on guarded ops; a nil-guard op returns immediately. The caller bounds ctx — waiting preserves emitted-bloom quality when the scan outran the upstream bloom, at the cost of delaying this stage's partial upload.

func (*DynamicFilterEmitOp) GuardStats 

func (op *DynamicFilterEmitOp) GuardStats() (guards int, buffered, dropped int64, overflowed, unresolvable bool)

GuardStats reports the guarded-emit telemetry for the runner's log line.

func (*DynamicFilterEmitOp) Init 

func (op *DynamicFilterEmitOp) Init(_ context.Context) error

func (*DynamicFilterEmitOp) Snapshot 

func (op *DynamicFilterEmitOp) Snapshot() DynamicFilterPartial

Snapshot returns the accumulated partial. Safe to call once after the pipeline has finished — caller takes ownership of the bloom slice.

type DynamicFilterPartial 

type DynamicFilterPartial struct {
	FilterID  string
	KeyType   string
	KeyColumn string
	BloomBits int
	Bloom     []uint64
	BloomMask uint64
	HasRange  bool
	Min, Max  int64
	RowCount  int64
	// Unresolved is set when the op observed input rows but could not
	// resolve KeyColumn in the batch schema — the partial is missing keys
	// that exist in the stream and MUST NOT be uploaded (a bloom missing
	// live keys falsely rejects rows at the consume side). A task that saw
	// zero rows is NOT unresolved: its key set is legitimately empty.
	Unresolved bool
}

DynamicFilterPartial is the in-memory representation of one task's emitted partial stats, ready for serialization by the fragment runner.

type DynamicRange 

type DynamicRange struct {
	Column   string // probe-side join key column name
	MinValue any
	MaxValue any
}

DynamicRange holds min/max values for a probe-side join key column, collected during build. Used for row-group-level scan pruning.

type Expression 

type Expression func(b *batch.RecordBatch, row int) any

Expression computes a value for a row in a batch.

func ArithExpr 

func ArithExpr(left, right Expression, op string) Expression

ArithExpr creates an arithmetic expression between two expressions.

func ColumnRef 

func ColumnRef(name string) Expression

ColumnRef creates an expression that reads a column value. The column index is resolved on first call and cached for subsequent rows.

func Literal 

func Literal(val any) Expression

Literal creates an expression that returns a constant.

type Filter 

type Filter struct {
	Pred Predicate
	// contains filtered or unexported fields
}

Filter is a UnaryOperator that filters rows using a selection vector.

func NewFilter 

func NewFilter(pred Predicate) *Filter

func (*Filter) Clone 

func (f *Filter) Clone() UnaryOperator

Clone returns a new Filter that shares the same predicate closure but has its own scratch buffer, allowing concurrent Execute calls.

func (*Filter) Close 

func (f *Filter) Close() error

func (*Filter) Execute 

func (f *Filter) Execute(_ context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error)

func (*Filter) Init 

func (f *Filter) Init(_ context.Context) error

type Float64Expression 

type Float64Expression func(b *batch.RecordBatch, row int) (float64, bool)

Float64Expression evaluates to float64 without boxing.

type FlushableOperator 

type FlushableOperator interface {
	HasPendingFlush() bool
	NextFlush(ctx context.Context) (*batch.RecordBatch, error)
}

FlushableOperator is implemented by operators that may need to emit additional batches after the main pipeline loop completes. Used by the Grace Hash Join to process spilled partitions after the streaming probe.

type GuardProbe 

type GuardProbe interface {
	// TryResolve returns the bloom if it is available now.
	TryResolve() (bloom []uint64, mask uint64, ok bool)
	// Done reports whether the poll has terminated (bloom available or
	// permanently given up).
	Done() bool
	// Wait blocks until the poll terminates or ctx ends.
	Wait(ctx context.Context) (bloom []uint64, mask uint64, ok bool)
}

GuardProbe is the emit op's view of one pending upstream bloom. The worker adapts its attach-on-arrival poll slot to this interface.

type HashAggregate 

type HashAggregate struct {
	GroupByCols []string
	// GroupByAll makes the aggregate group by every input column, resolved
	// from the first batch's schema (GroupByCols must be empty). This is how
	// DISTINCT is planned: a keys-only hash aggregate inherits the spill
	// machinery, where the dedicated Distinct operator's seen-set grew
	// without bound or tracking. With no Aggs the output is exactly the
	// distinct key tuples, in input column order.
	GroupByAll bool
	Aggs       []AggColumn
	Spill      *memory.SpillManager // optional: enables spill-to-disk
	// PartialDrainBytes bounds this aggregate's in-memory state when it is a
	// morsel-parallel CLONE partial: past the threshold, Consume drains the
	// whole state to canonical partial-state run files (drainedRuns) that
	// MergeSink hands to the primary for Finalize's k-way merge. Clones run
	// on a tracking-only SpillManager view whose ShouldSpillFor is
	// unconditionally false, so without this bound a high-cardinality GROUP
	// BY multiplies serial state by k with no pressure valve — the SF100 Q17
	// worker deaths (morsel-agg-partials-v2.md §3.A). 0 = disabled (primary
	// aggregates keep their ShouldSpillFor-driven spill machinery).
	PartialDrainBytes int64
	// PartitionedDisjoint marks a sink participating in partitioned
	// parallel aggregation (partitioned_agg.go): every group key lives in
	// exactly one sink. MergeSink then ADOPTS clone partitions instead of
	// re-inserting their groups, and Next() streams each adopted
	// partition's state after its own.
	PartitionedDisjoint bool

	NullGroupCols []string // GROUPING SETS: columns to output as NULL (legacy per-node)
	GroupingSets  [][]int  // single-pass grouping sets: column indices within GroupByCols per set
	InputRowHint  int64    // estimated input rows for pre-sizing hash table
	// contains filtered or unexported fields
}

HashAggregate is a Sink that performs grouped aggregation with a hash map. Uses kernel-resolved typed updaters and cached column indices. When a SpillManager is set, input batches are spilled to disk under memory pressure and re-processed during Finalize.

func NewHashAggregate 

func NewHashAggregate(groupByCols []string, aggs []AggColumn) *HashAggregate

func (*HashAggregate) CloneSink 

func (h *HashAggregate) CloneSink() SinkSource

CloneSink returns a new HashAggregate with the same configuration but fresh state. Used by parallel pipeline execution: each worker gets its own cloned sink.

func (*HashAggregate) Close 

func (h *HashAggregate) Close() error

Close releases any tracker reservation HashAggregate still holds for group-state memory and buffered-but-unspilled rows. Without this, a non-spilling HashAggregate accumulates a phantom reservation in the shared tracker for the lifetime of the process; see HashJoin.Close for the full background.

func (*HashAggregate) Consume 

func (h *HashAggregate) Consume(_ context.Context, b *batch.RecordBatch) error

func (*HashAggregate) EstimateRelief 

func (h *HashAggregate) EstimateRelief(target int64) int64

EstimateRelief implements memory.AccountedOperator: a pure read of the rebuild-safe spillable bytes, capped at target.

func (*HashAggregate) Finalize 

func (h *HashAggregate) Finalize(_ context.Context) error

func (*HashAggregate) Init 

func (h *HashAggregate) Init(_ context.Context) error

func (*HashAggregate) Inspect 

func (h *HashAggregate) Inspect() memory.OperatorFootprint

Inspect implements memory.AccountedOperator. Wait-free w.r.t. the registry but takes h.mu to read group-state fields consistently.

func (*HashAggregate) MergeSink 

func (h *HashAggregate) MergeSink(other SinkSource)

MergeSink merges another HashAggregate's partial state into this one. Called after all parallel workers finish to combine partial aggregates.

After the state merge, h's group footprint has grown by the clone's state; reconcileGroupMemory recharges h so the shared-tracker reservation follows the state (morsel-parallel clones charge a tracking-only SpillManager view; their own charge is released at clone Close, AFTER this recharge, so the tracker never under-reports in between). No-op when h.Spill is nil — the single-process planner path.

func (*HashAggregate) Next 

func (h *HashAggregate) Next(ctx context.Context) (*batch.RecordBatch, error)

Next returns the aggregated results in batches of DefaultBatchSize rows.

func (*HashAggregate) PartitionSelectors 

func (h *HashAggregate) PartitionSelectors(b *batch.RecordBatch, parts int, scratch [][]uint32) [][]uint32

PartitionSelectors splits b's active rows into parts selection lists by group-key hash. Returns nil when a group column is missing or of an unsupported type — callers fall back to unpartitioned consumption. scratch is reused across calls; the returned slices alias it.

func (*HashAggregate) SpillSome 

func (h *HashAggregate) SpillSome(target int64) (int64, error)

SpillSome drains a portion of the SoA hash state to a partial-state spill file and releases the freed bytes back to the tracker, returning the number of bytes released. Called by SpillManager.RequestRelief on behalf of a peer operator under memory pressure.

On the int-keyed path, spillPartialState drains a hash-partition slice sized roughly to `target` bytes, leaving surviving groups in place. This breaks the drain-rebuild loop that whole-table draining created at SF100 scale (PR #88 → drain-rebuild loop → heartbeat starvation): future Consume rows whose keys hash to a surviving partition continue to hit existing in-memory groups, paying no rebuild cost.

On other paths (dual-int, compact, string, generic) and when target covers the full footprint, falls through to the whole-drain path — semantically identical to the pre-partial-drain behavior.

Implements memory.Spillable and memory.AccountedOperator. The OpSpilling state is published for the duration so a concurrent RequestRelief snapshot skips this instance rather than double-dispatching.

func (*HashAggregate) StateBytes 

func (h *HashAggregate) StateBytes() int64

StateBytes reports the current in-memory group-state size. Exposed for callers that bound memory without a SpillManager (the shuffle sender's capped partial aggregate flushes an epoch when this crosses its cap); Inspect() reports zero footprint when no Spill is attached, so it can't serve that role.

type HashJoin 

type HashJoin struct {
	JoinType  JoinType
	LeftKeys  []string // join key columns from left (probe) side
	RightKeys []string // join key columns from right (build) side

	// Memory tracking (optional). When set, Reserve() is called for each
	// build-side batch. If the budget is exceeded, Build returns ErrMemoryExceeded.
	MemTracker *memory.Tracker

	// Spill-to-disk (optional). When set, build-side batches are spilled to disk
	// when memory pressure exceeds 80% of budget using Grace Hash Join partitioning.
	Spill *memory.SpillManager

	// SemiAntiFilter is an optional predicate applied during semi/anti join probe.
	// When set, each candidate build row is checked in addition to hash key equality.
	// This enables non-equality join conditions (e.g., "!=") from decorrelated EXISTS.
	SemiAntiFilter func(probe *batch.RecordBatch, probeRow int, build *batch.RecordBatch, buildRow int) bool

	// BuildTableAlias is the table alias of the build side. When set, duplicate
	// column names in the output schema are qualified as "alias.column" to avoid
	// ambiguity (e.g., self-joins like nation n1 JOIN nation n2).
	BuildTableAlias string

	// QualifyAllBuildCols forces every build-side column into the output
	// schema under its qualified name ("BuildTableAlias.col"), not just the
	// columns that would collide with the probe schema. The planner sets
	// this on co-pathing self-join scans so the FIRST self-join's column
	// is reachable downstream by its qualified name (avoiding the NULL-via-
	// columnIndexFallback path that breaks Q07).
	QualifyAllBuildCols bool

	// BuildColOrigins maps each bare build-column name (lowercased) to the
	// scan alias that owns it. Set by the planner only when the build side
	// spans multiple tables (bushy join subtrees); nil for single-scan
	// builds. Duplicate qualification then uses the owning alias instead of
	// the single BuildTableAlias, which is ambiguous for multi-table builds.
	BuildColOrigins map[string]string

	// SemiAntiKeyOnly enables a lightweight build for semi/anti joins that have
	// no SemiAntiFilter. Only the key index and bloom filter are built — batch
	// storage and arena refs are skipped. Reduces memory and build time by ~2-4x
	// for large build sides (e.g., 6M-row lineitem scan for EXISTS subqueries).
	SemiAntiKeyOnly bool

	// SemiAntiNEProbeCol/SemiAntiNEBuildCol carry the planner-recognized
	// single-condition `probe.col <> build.col` join filter (the
	// decorrelated-EXISTS self-inequality class). When both are set on a
	// semi/anti join, Build collapses to a distinct-pair table — see
	// join_semianti_ne.go. SemiAntiFilter stays wired as the fallback for
	// shapes the runtime can't activate (non-int value column).
	SemiAntiNEProbeCol string
	SemiAntiNEBuildCol string

	// BuildStoreCols, when non-empty, narrows every stored build batch to the
	// named columns (join keys + SemiAntiFilter-referenced columns) at arrival
	// time. Filtered semi/anti joins never emit build rows, but the probe must
	// evaluate SemiAntiFilter against the filter's build-side columns — storing
	// only keys + those columns keeps partitioned builds, their per-partition
	// accumulators, and their spill files narrow from the first batch. The
	// post-build PruneBuildColumns cannot achieve this: it skips partition-on-
	// arrival builds entirely (evicted entries are nil'd and spilled files
	// carry the storage schema), which is every spill-eligible build. Names
	// are resolved once against the first arrival batch; if any name fails to
	// resolve, projection is disabled and full batches are stored.
	BuildStoreCols []string

	// BuildRowHint is an optional hint for the expected number of build-side rows.
	// When set, the arena and hash table are pre-allocated to avoid repeated growth.
	BuildRowHint int64
	// contains filtered or unexported fields
}

HashJoin implements a hash join with build and probe phases. Build side is stored in columnar RecordBatches, indexed by a hash map of join keys to batch/row references. This avoids the ~10x memory overhead of storing build-side rows as map[string]any.

func NewHashJoin 

func NewHashJoin(joinType JoinType, leftKeys, rightKeys []string) *HashJoin

NewHashJoin creates a new hash join operator.

func (*HashJoin) BloomPushdownOp 

func (h *HashJoin) BloomPushdownOp() *BloomFilterOp

BloomPushdownOp returns a UnaryOperator that pre-filters probe batches using the build-side bloom filter. Must be called after Build() completes. Returns nil if bloom filter pushdown is not applicable (empty build, wrong join type). Safe for InnerJoin, SemiJoin, and RightJoin only.

func (*HashJoin) Build 

func (h *HashJoin) Build(ctx context.Context, source Source) error

Build consumes all rows from the build (right) side into the columnar hash table. Uses parallel workers when the build side is large enough to benefit from concurrent hash table construction with per-worker local tables.

func (*HashJoin) BuildFromRows 

func (h *HashJoin) BuildFromRows(schema []parquet.Column, rows []map[string]any)

BuildFromRows loads the build side directly from rows (used by tests and worker).

func (*HashJoin) BuildKeyRange 

func (h *HashJoin) BuildKeyRange() []DynamicRange

BuildKeyRange returns the min/max range of each build-side join key column. Column names are mapped to probe-side names (LeftKeys) since the scan knows its own columns. Must be called after Build() completes. Returns nil if no rows were built or if the join type doesn't support range pushdown.

func (*HashJoin) BuildRows 

func (h *HashJoin) BuildRows() int64

BuildRows returns the number of rows in the build side.

func (*HashJoin) Close 

func (h *HashJoin) Close() error

Close releases any memory still reserved with the shared MemTracker and drops references to the build-side state so Go's GC can reclaim it promptly. Must be called by the owner of the HashJoin (the worker executor) after the probe pipeline has fully drained — including any spilled-partition flushes. Calling Close more than once is safe; the release amount goes to zero on the first call.

Without this, an operator that builds-probes-completes WITHOUT spilling never returns its reservation to the shared tracker. The hash table is GC-eligible but the tracker thinks it's still in use, so the worker reports inflated PoolPressure to the coordinator and worker-side spill thresholds fire prematurely. With many concurrent broadcast joins (e.g., TPC-H Q02), phantom reservations accumulate query-over-query.

func (*HashJoin) EstimateRelief 

func (h *HashJoin) EstimateRelief(target int64) int64

EstimateRelief implements memory.AccountedOperator: the reclaimable partition bytes, capped at target.

func (*HashJoin) FixKeyAssignment 

func (h *HashJoin) FixKeyAssignment() bool

FixKeyAssignment corrects misassigned join keys after the build phase. SQL may place the build-side column on the left of "=" (e.g., JOIN t ON t.id = src.id), causing parseJoinKeys to assign it as a left/probe key. This detects and swaps misassigned pairs by checking which keys exist in the build schema. It returns true when any pair was swapped, so callers can surface the repair (it should never fire on planner-produced plans).

func (*HashJoin) Inspect 

func (h *HashJoin) Inspect() memory.OperatorFootprint

Inspect implements memory.AccountedOperator. OwnedBytes includes the hash arena/index overhead (trackedMem) plus keyBuf scratch; SpillableBytes is the reclaimable in-memory partition bytes only (the arena can't be freed without rebuilding). RetainedBytes is the build-side column data.

func (*HashJoin) NEActive 

func (h *HashJoin) NEActive() bool

NEActive reports whether the distinct-pair build engaged (post-Build). Callers use it for engagement logging — the A/B marker for this path.

func (*HashJoin) Probe 

func (h *HashJoin) Probe() *HashJoinProbe

Probe is a UnaryOperator that probes the hash table for each input batch.

func (*HashJoin) PruneBuildColumns 

func (h *HashJoin) PruneBuildColumns(keepCols []string)

PruneBuildColumns removes non-essential columns from the build-side batches. For SEMI/ANTI joins, the build side never appears in the output, so after the hash index is built we only need columns referenced by SemiAntiFilter. If keepCols is empty and no SemiAntiFilter is set, buildBatches are cleared.

func (*HashJoin) SpillSome 

func (h *HashJoin) SpillSome(target int64) (int64, error)

SpillSome attempts to free at least target bytes from this join's in-memory partitions. Picks the largest partition repeatedly until target is met or no partitions remain. Implements memory.Spillable.

func (*HashJoin) SpillState 

func (h *HashJoin) SpillState() *spillState

SpillState returns the spill state (nil if no spill has occurred). Test-only.

func (*HashJoin) TrackedMem 

func (h *HashJoin) TrackedMem() int64

TrackedMem returns how much memory this join has reserved from the shared tracker.

type HashJoinProbe 

type HashJoinProbe struct {

	// OutputFilter restricts which columns the probe materializes.
	// When set, only columns in this map appear in the output batch.
	// This avoids allocating and gathering unneeded intermediate columns
	// in multi-way join pipelines.
	OutputFilter map[string]bool

	// LateMaterialize emits inner/left join output as view (dictionary)
	// columns over the probe input and build batches instead of gathering
	// copies — the deferred gather happens at the first consumer that needs
	// owned storage (see flattenForConsumer). The probe input is Detach()ed
	// when views reference it so pool recycling can't mutate the shared
	// vectors; GC reclaims it once the views die. Off by default.
	LateMaterialize bool
	// contains filtered or unexported fields
}

HashJoinProbe is a UnaryOperator that probes the build-side hash table.

func (*HashJoinProbe) AcceptsViews 

func (p *HashJoinProbe) AcceptsViews() bool

AcceptsViews marks the probe view-aware: Execute self-manages view input via prepareViewInput — key columns (the only probe-side columns the probe reads positionally) are flattened individually, pass-through columns stay lazy and compose in emitViewOutput, and the paths that read or persist arbitrary columns flatten everything. This is what makes a fused join chain one-copy-per-column: join N's output views compose over join N-1's bases instead of materializing between probes.

func (*HashJoinProbe) Clone 

func (p *HashJoinProbe) Clone() UnaryOperator

Clone returns a new HashJoinProbe that shares the same build-side hash table but has its own scratch buffers (pairsBuf, semiSelBuf, lookupBuf, indexBuf).

func (*HashJoinProbe) Close 

func (p *HashJoinProbe) Close() error

func (*HashJoinProbe) Execute 

func (p *HashJoinProbe) Execute(ctx context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error)

func (*HashJoinProbe) FlushAntiMatched 

func (p *HashJoinProbe) FlushAntiMatched() *batch.RecordBatch

FlushAntiMatched returns build-side rows that were NOT matched. For RightAntiJoin.

func (*HashJoinProbe) FlushMatched 

func (p *HashJoinProbe) FlushMatched() *batch.RecordBatch

FlushMatched returns a RecordBatch containing build-side rows that WERE matched during probing. For RightSemiJoin only. Returns build-side columns only.

func (*HashJoinProbe) FlushUnmatched 

func (p *HashJoinProbe) FlushUnmatched(leftSchema []parquet.Column) *batch.RecordBatch

FlushUnmatched returns a RecordBatch containing build-side rows that were never matched during probing. For RightJoin and FullOuterJoin only.

func (*HashJoinProbe) HasPendingFlush 

func (p *HashJoinProbe) HasPendingFlush() bool

HasPendingFlush returns true if there are spilled partitions still left to process.

func (*HashJoinProbe) Init 

func (p *HashJoinProbe) Init(_ context.Context) error

func (*HashJoinProbe) NextFlush 

func (p *HashJoinProbe) NextFlush(ctx context.Context) (*batch.RecordBatch, error)

NextFlush returns the next result batch from spilled-partition processing, fully streaming. One partition is held in memory at a time; within that partition probe batches are read from disk one at a time and the resulting joined batch is yielded immediately. The previous implementation accumulated every joined output from every spilled partition into a single in-memory slice before yielding any — for SF100 Q05 lineitem⋈orders that pinned tens of GB of joined output simultaneously.

type HeldStateSource 

type HeldStateSource interface {
	ServesHeldState() bool
}

HeldStateSource marks sources that serve a pipeline-breaker's HELD state (aggregate/sort/window output phases). Heap-backpressure pauses must not throttle pipelines draining such sources: the held state IS the memory pressure, and draining it is the only way the pressure clears — pausing the drain turned ClickBench Q33's output phase into 100+ seconds of sleeping on a single goroutine while 15GB of finished aggregate state waited to be streamed out.

type InFilter 

type InFilter struct {
	ColName string
	Values  []any
	Negate  bool
	// contains filtered or unexported fields
}

InFilter uses a vectorized kernel for set membership testing (IN / NOT IN).

func NewInFilter 

func NewInFilter(colName string, values []any, negate bool) *InFilter

func (*InFilter) Clone 

func (f *InFilter) Clone() UnaryOperator

func (*InFilter) Close 

func (f *InFilter) Close() error

func (*InFilter) Execute 

func (f *InFilter) Execute(_ context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error)

func (*InFilter) Init 

func (f *InFilter) Init(_ context.Context) error

type Int64Expression 

type Int64Expression func(b *batch.RecordBatch, row int) (int64, bool)

Int64Expression evaluates to int64 without boxing.

type JoinType 

type JoinType int

JoinType identifies the kind of join. 

const (
	InnerJoin JoinType = iota
	LeftJoin
	RightJoin
	FullOuterJoin
	CrossJoin
	SemiJoin      // returns left row if match found, no duplicates
	AntiJoin      // returns left row only if NO match found
	RightSemiJoin // builds LEFT (small), probes RIGHT (large), returns matched build rows
	RightAntiJoin // builds LEFT (small), probes RIGHT (large), returns unmatched build rows
)

type KernelFilter 

type KernelFilter struct {
	ColName string
	Op      CompareOp
	Value   any
	// RowFallback, when non-nil, evaluates the original comparison row-at-
	// a-time. Used when ColName is a ROW-field access ("attrs.score") that
	// the typed kernel cannot evaluate — the planner attaches the compiled
	// expression's predicate, and resolution delegates to it instead of
	// silently matching nothing (issue #147).
	RowFallback Predicate
	// contains filtered or unexported fields
}

KernelFilter is a UnaryOperator that uses a pre-resolved typed filter kernel. The type dispatch happens once on first Execute; the inner loop has no type switches.

func NewKernelFilter 

func NewKernelFilter(colName string, op CompareOp, value any) *KernelFilter

NewKernelFilter creates a filter that uses typed kernels for comparison.

func (*KernelFilter) Clone 

func (f *KernelFilter) Clone() UnaryOperator

Clone returns a new KernelFilter with the same parameters but fresh resolution state and scratch buffers for concurrent Execute calls.

func (*KernelFilter) Close 

func (f *KernelFilter) Close() error

func (*KernelFilter) Execute 

func (f *KernelFilter) Execute(ctx context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error)

func (*KernelFilter) Init 

func (f *KernelFilter) Init(_ context.Context) error

type LikeFilter 

type LikeFilter struct {
	ColName string
	Pattern string
	Negate  bool
	// contains filtered or unexported fields
}

LikeFilter uses a vectorized kernel for SQL LIKE pattern matching.

func NewLikeFilter 

func NewLikeFilter(colName, pattern string, negate bool) *LikeFilter

func (*LikeFilter) Clone 

func (f *LikeFilter) Clone() UnaryOperator

func (*LikeFilter) Close 

func (f *LikeFilter) Close() error

func (*LikeFilter) Execute 

func (f *LikeFilter) Execute(_ context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error)

func (*LikeFilter) Init 

func (f *LikeFilter) Init(_ context.Context) error

type Limit 

type Limit struct {
	Max    int64
	Offset int64
	// contains filtered or unexported fields
}

Limit is a UnaryOperator that passes through at most N rows, optionally skipping the first Offset rows.

func NewLimit 

func NewLimit(n, offset int64) *Limit

func (*Limit) AcceptsViews 

func (l *Limit) AcceptsViews() bool

AcceptsViews: Limit manipulates only the selection vector and row counts — it never reads column storage, so view columns pass through untouched.

func (*Limit) Clone 

func (l *Limit) Clone() UnaryOperator

Clone returns the same Limit instance. Limit uses atomic counters for seen/passed tracking, making it safe for concurrent Execute calls from multiple pipeline workers.

func (*Limit) Close 

func (l *Limit) Close() error

func (*Limit) Done 

func (l *Limit) Done() bool

Done returns true when the limit has been satisfied, enabling pipeline early termination (LIMIT pushdown).

func (*Limit) Execute 

func (l *Limit) Execute(_ context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error)

func (*Limit) Init 

func (l *Limit) Init(_ context.Context) error

type MergeableSink 

type MergeableSink interface {
	SinkSource
	CloneSink() SinkSource
	MergeSink(other SinkSource)
}

MergeableSink is a SinkSource that supports per-worker partial aggregation. When the pipeline has multiple workers, each worker gets its own cloned sink. After all workers finish, partial sinks are merged into the primary sink.

type NullCheckFilter 

type NullCheckFilter struct {
	ColName   string
	CheckNull bool // true = IS NULL, false = IS NOT NULL
	// contains filtered or unexported fields
}

NullCheckFilter is a vectorized filter for IS NULL / IS NOT NULL predicates. Scans the null bitmap directly — no per-row type switch or function call overhead.

func NewNullCheckFilter 

func NewNullCheckFilter(colName string, checkNull bool) *NullCheckFilter

func (*NullCheckFilter) Clone 

func (f *NullCheckFilter) Clone() UnaryOperator

func (*NullCheckFilter) Close 

func (f *NullCheckFilter) Close() error

func (*NullCheckFilter) Execute 

func (f *NullCheckFilter) Execute(_ context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error)

func (*NullCheckFilter) Init 

func (f *NullCheckFilter) Init(_ context.Context) error

type OrFilter 

type OrFilter struct {
	Left, Right UnaryOperator
	// contains filtered or unexported fields
}

OrFilter evaluates two filter branches and unions their selection vectors. Both branches run on the same input batch; results are merged with dedup.

func NewOrFilter 

func NewOrFilter(left, right UnaryOperator) *OrFilter

func (*OrFilter) Clone 

func (f *OrFilter) Clone() UnaryOperator

func (*OrFilter) Close 

func (f *OrFilter) Close() error

func (*OrFilter) Execute 

func (f *OrFilter) Execute(ctx context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error)

func (*OrFilter) Init 

func (f *OrFilter) Init(ctx context.Context) error

type OutputSharingAware 

type OutputSharingAware interface {
	EnableSharedOutputs()
}

OutputSharingAware operators can switch to per-call output allocation so their batches become safe to share across asynchronous partition owners (at the cost of losing buffer reuse). The pipeline enables it instead of disabling partitioned aggregation.

type Pipeline 

type Pipeline struct {
	Source  Source
	Ops     []UnaryOperator
	Sink    Sink
	Workers int // number of parallel workers (0 or 1 = serial)
}

Pipeline represents Source → [UnaryOps...] → Sink.

func (*Pipeline) Close 

func (p *Pipeline) Close() error

Close releases all resources in the pipeline.

func (*Pipeline) Run 

func (p *Pipeline) Run(ctx context.Context) error

Run executes the pipeline by pulling from source, transforming through operators, and pushing to sink. When Workers > 1 and all operators implement Cloneable, batches are processed by multiple goroutines concurrently. Otherwise falls back to serial execution.

type Predicate 

type Predicate func(b *batch.RecordBatch, row int) bool

Predicate evaluates a row and returns true if it passes the filter.

func And 

func And(preds ...Predicate) Predicate

And combines predicates with logical AND.

func ColumnCompare 

func ColumnCompare(colName string, op CompareOp, value any) Predicate

ColumnCompare creates a predicate that compares a column against a constant value.

func ColumnLike 

func ColumnLike(colName, pattern string, not bool) Predicate

ColumnLike creates a predicate that evaluates col LIKE pattern using SQL LIKE semantics. The pattern uses % for any sequence of characters and _ for any single character.

func Or 

func Or(preds ...Predicate) Predicate

Or combines predicates with logical OR.

type ProfileCollector 

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

ProfileCollector aggregates stats from all operators in a pipeline.

func NewProfileCollector 

func NewProfileCollector() *ProfileCollector

NewProfileCollector creates a new collector.

func WrapPipeline 

func WrapPipeline(pipeline *Pipeline) *ProfileCollector

WrapPipeline wraps all operators in a pipeline with profiling decorators. Returns the collector for reading stats after execution.

func (*ProfileCollector) Add 

func (c *ProfileCollector) Add(name string) *ProfileStats

Add registers a stats entry and returns it. The caller (wrapper) writes to it.

func (*ProfileCollector) Stats 

func (c *ProfileCollector) Stats() []*ProfileStats

Stats returns all collected stats in pipeline order.

type ProfileStats 

type ProfileStats struct {
	Name     string        // operator name (e.g., "Filter", "HashAggregate")
	WallTime time.Duration // total wall time spent in this operator
	RowsIn   int64         // total input rows
	RowsOut  int64         // total output rows
	Calls    int64         // number of calls (Next/Execute/Consume)
}

ProfileStats holds execution statistics for a single pipeline operator.

type ProfiledOperator 

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

ProfiledOperator wraps a UnaryOperator to collect timing and row count stats.

func WrapOperator 

func WrapOperator(op UnaryOperator, collector *ProfileCollector, name string) *ProfiledOperator

WrapOperator decorates a UnaryOperator with profiling.

func (*ProfiledOperator) Clone 

func (p *ProfiledOperator) Clone() UnaryOperator

Clone delegates to the inner operator if it implements Cloneable. Returns a new ProfiledOperator wrapping the cloned inner, with fresh stats.

func (*ProfiledOperator) Close 

func (p *ProfiledOperator) Close() error

func (*ProfiledOperator) Done 

func (p *ProfiledOperator) Done() bool

Done delegates to the inner operator if it implements DoneSignaler.

func (*ProfiledOperator) Execute 

func (p *ProfiledOperator) Execute(ctx context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error)

func (*ProfiledOperator) Init 

func (p *ProfiledOperator) Init(ctx context.Context) error

func (*ProfiledOperator) Inner 

func (p *ProfiledOperator) Inner() UnaryOperator

Inner returns the wrapped operator (for type assertions like DoneSignaler).

type ProfiledSink 

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

ProfiledSink wraps a Sink to collect timing and row count stats.

func WrapSink 

func WrapSink(sink Sink, collector *ProfileCollector, name string) *ProfiledSink

WrapSink decorates a Sink with profiling.

func (*ProfiledSink) Close 

func (p *ProfiledSink) Close() error

func (*ProfiledSink) Consume 

func (p *ProfiledSink) Consume(ctx context.Context, b *batch.RecordBatch) error

func (*ProfiledSink) Finalize 

func (p *ProfiledSink) Finalize(ctx context.Context) error

func (*ProfiledSink) Init 

func (p *ProfiledSink) Init(ctx context.Context) error

func (*ProfiledSink) Inner 

func (p *ProfiledSink) Inner() Sink

Inner returns the wrapped sink.

type ProfiledSource 

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

ProfiledSource wraps a Source to collect timing and row count stats.

func WrapSource 

func WrapSource(src Source, collector *ProfileCollector, name string) *ProfiledSource

WrapSource decorates a Source with profiling. Only call when EXPLAIN ANALYZE is active.

func (*ProfiledSource) Close 

func (p *ProfiledSource) Close() error

func (*ProfiledSource) Init 

func (p *ProfiledSource) Init(ctx context.Context) error

func (*ProfiledSource) Inner 

func (p *ProfiledSource) Inner() Source

Inner returns the wrapped source (for type assertions like ScanStatsProvider).

func (*ProfiledSource) Next 

func (p *ProfiledSource) Next(ctx context.Context) (*batch.RecordBatch, error)

type ProgressReporter 

type ProgressReporter interface {
	AddRows(int64)
	AddBytes(int64)
}

ProgressReporter is a tiny interface the pipeline uses to report per-batch forward progress to the surrounding task. Implemented by the worker's per-task TaskProgress; the engine doesn't depend on any worker types because of import-cycle constraints.

Callers MUST tolerate a nil receiver — it's used through ProgressReporterFromContext which returns nil for callers outside of a worker task (standalone queries, tests).

func ProgressReporterFromContext 

func ProgressReporterFromContext(ctx context.Context) ProgressReporter

ProgressReporterFromContext returns the active progress reporter or nil if none. AddRows / AddBytes on the returned value are nil-safe only when the type embeds the nil-tolerance behaviour itself — callers should ALWAYS guard with `if p != nil` before invoking.

type Project 

type Project struct {
	Projections []ProjectColumn
	// contains filtered or unexported fields
}

Project is a UnaryOperator that selects and computes columns.

func NewProject 

func NewProject(projections []ProjectColumn) *Project

func (*Project) Clone 

func (p *Project) Clone() UnaryOperator

Clone returns a new Project that shares the same (immutable) projections. Each clone gets its own pool (created lazily on first Execute).

func (*Project) Close 

func (p *Project) Close() error

func (*Project) Execute 

func (p *Project) Execute(_ context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error)

func (*Project) Init 

func (p *Project) Init(_ context.Context) error

type ProjectColumn 

type ProjectColumn struct {
	Name            string
	Type            parquet.TypeID
	Expr            Expression
	Float64Eval     Float64Expression           // optional typed path (avoids interface{} boxing)
	Int64Eval       Int64Expression             // optional typed path
	VecFloat64Eval  VecFloat64Expression        // optional vectorized path (entire column at once)
	VecFloat64Clone func() VecFloat64Expression // creates a clone with independent scratch buffers
	VecEval         VecExpression               // optional vectorized evaluation for any output type
	SourceCol       string                      // source column name for type resolution on renames
	DirectCopy      string                      // if set, bulk copy this input column (no per-row eval)
	Dimension       int                         // VECTOR output dimensionality (e.g. embed()); 0 = not a vector
}

ProjectColumn defines an output column of a projection.

type ScanStatsProvider 

type ScanStatsProvider interface {
	RowsScanned() int64
}

ScanStatsProvider is implemented by sources that can report scan statistics.

type Sink 

type Sink interface {
	Init(ctx context.Context) error
	Consume(ctx context.Context, b *batch.RecordBatch) error
	Finalize(ctx context.Context) error
	Close() error
}

Sink consumes all input before results can be read (pipeline breaker). Must handle concurrent Consume() calls from multiple goroutines.

type SinkSource 

type SinkSource interface {
	Sink
	Source
}

SinkSource is a Sink that can also act as a Source after Finalize (e.g., hash aggregate).

type SliceSource 

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

SliceSource is a simple Source that yields batches from a slice of rows.

func NewSliceSource 

func NewSliceSource(schema []Column, rows []map[string]any) *SliceSource

NewSliceSource creates a source from in-memory rows.

func (*SliceSource) Close 

func (s *SliceSource) Close() error

func (*SliceSource) Init 

func (s *SliceSource) Init(_ context.Context) error

func (*SliceSource) Next 

func (s *SliceSource) Next(_ context.Context) (*batch.RecordBatch, error)

type Sort 

type Sort struct {
	Keys  []SortKey
	Limit int // 0 = no limit, >0 = only materialize top N rows

	Spill *memory.SpillManager // optional: enables spill-to-disk
	// contains filtered or unexported fields
}

Sort is a Sink that accumulates all batches columnar and sorts them using typed comparisons on an index array (no row-oriented conversion). When a SpillManager is set, Sort will spill to disk when memory pressure is high. When Limit > 0, only the top Limit rows are materialized (Top-K optimization).

func NewSort 

func NewSort(keys []SortKey) *Sort

func (*Sort) CloneSink 

func (s *Sort) CloneSink() SinkSource

CloneSink returns a new Sort with the same configuration but fresh state. Used by parallel pipeline execution: each worker gets its own cloned sink, eliminating mutex contention during the parallel Consume phase.

func (*Sort) Close 

func (s *Sort) Close() error

Close releases any tracker reservation Sort still holds for buffered rows that never crossed the spill threshold, and drops references so the GC can reclaim immediately. Without this, a non-spilling Sort accumulates a phantom reservation in the shared tracker for the lifetime of the process; see HashJoin.Close for the full background.

func (*Sort) Consume 

func (s *Sort) Consume(_ context.Context, b *batch.RecordBatch) error

func (*Sort) EstimateRelief 

func (s *Sort) EstimateRelief(target int64) int64

EstimateRelief implements memory.AccountedOperator. Sort frees all-or-nothing, so it reports the full spillable footprint whenever target > 0.

func (*Sort) Finalize 

func (s *Sort) Finalize(_ context.Context) error

func (*Sort) Init 

func (s *Sort) Init(_ context.Context) error

func (*Sort) Inspect 

func (s *Sort) Inspect() memory.OperatorFootprint

Inspect implements memory.AccountedOperator.

func (*Sort) MergeSink 

func (s *Sort) MergeSink(other SinkSource)

MergeSink merges another Sort's accumulated batches into this one. Called after all parallel workers finish to combine partial batch lists before the single-threaded Finalize sort.

Memory accounting transfers with the batches: when the clone tracked its buffered rows (morsel-parallel clones charge a tracking-only SpillManager view against the shared pool), the reservation must follow the state or the merged bytes become invisible to the primary's spill trigger. Charge the primary FIRST, then release the clone, so the shared tracker never under-reports in between. No-op when neither side tracks (the single-process planner path).

func (*Sort) Next 

func (s *Sort) Next(_ context.Context) (*batch.RecordBatch, error)

Next returns sorted results in batches. On the external-merge path it streams from the k-way run merger; otherwise it drains the materialized s.sorted list.

func (*Sort) SpillSome 

func (s *Sort) SpillSome(_ int64) (int64, error)

SpillSome drains accumulated input batches to disk and releases the freed bytes — a sorted columnar run on the external-merge path, a raw-row file on the nested-type fallback. All-or-nothing either way: input batches are pre-sort, so there's no partial state to keep.

Implements memory.Spillable and memory.AccountedOperator.

func (*Sort) Truncate 

func (s *Sort) Truncate(n int)

Truncate keeps only the first n rows of sorted output (Top-K).

type SortKey 

type SortKey struct {
	Column    string
	Order     SortOrder
	NullsLast bool
}

SortKey defines a column and direction for sorting.

type SortMergeJoin 

type SortMergeJoin struct {
	JoinType  JoinType // v1: InnerJoin only; validated in Init/Finalize
	LeftKeys  []string // join key columns from left (probe) side
	RightKeys []string // join key columns from right (build) side

	// BuildTableAlias / BuildColOrigins / QualifyAllBuildCols / OutputFilter
	// carry HashJoinProbe's output-schema semantics — see
	// joinOutputSchemaWithMapping.
	BuildTableAlias     string
	BuildColOrigins     map[string]string
	QualifyAllBuildCols bool
	OutputFilter        map[string]bool

	// Spill enables tracker accounting and spill-to-disk. When nil the
	// operator buffers unbounded in memory (embedded/test paths), like Sort.
	Spill *memory.SpillManager
	// contains filtered or unexported fields
}

func NewSortMergeJoin 

func NewSortMergeJoin(leftKeys, rightKeys []string) *SortMergeJoin

NewSortMergeJoin creates an inner sort-merge join. leftKeys are the probe (left) side's join columns, rightKeys the build (right) side's, positionally paired.

func (*SortMergeJoin) Build 

func (j *SortMergeJoin) Build(ctx context.Context, source Source) error

Build drains the build-side (right) source, mirroring HashJoin.Build's contract: inits and closes the source, reports row progress.

func (*SortMergeJoin) Close 

func (j *SortMergeJoin) Close() error

Close releases buffered state, merge state, run scratch, and any tracker reservation still held — see Sort.Close for why this matters (phantom reservations outlive the query otherwise).

func (*SortMergeJoin) Consume 

func (j *SortMergeJoin) Consume(_ context.Context, b *batch.RecordBatch) error

Consume buffers a probe-side (left) batch.

func (*SortMergeJoin) EstimateRelief 

func (j *SortMergeJoin) EstimateRelief(target int64) int64

EstimateRelief implements memory.AccountedOperator: what SpillSome(target) would free — the larger buffered side, plus the smaller one when the larger alone doesn't cover target.

func (*SortMergeJoin) Finalize 

func (j *SortMergeJoin) Finalize(_ context.Context) error

Finalize collapses each side into one sorted stream (spilled runs + the in-memory remainder under a bounded-fan-in merger) and resolves the cross-side comparison kernels. Nothing is materialized — Next streams the merge one output batch at a time.

func (*SortMergeJoin) Init 

func (j *SortMergeJoin) Init(_ context.Context) error

Init prepares the probe/merge state. It deliberately does NOT reset the build side: Build runs before the probe pipeline starts (HashJoin's contract), and pipeline Init would otherwise wipe it.

func (*SortMergeJoin) Inspect 

func (j *SortMergeJoin) Inspect() memory.OperatorFootprint

Inspect implements memory.AccountedOperator.

func (*SortMergeJoin) Next 

func (j *SortMergeJoin) Next(_ context.Context) (*batch.RecordBatch, error)

Next streams joined output batches (up to DefaultBatchSize rows each) from the two-cursor merge, or (nil, nil) once both sides are exhausted. Merge memory: one batch per live run cursor per side, the pinned duplicate-group batches, and the pending flush window.

func (*SortMergeJoin) SpillSome 

func (j *SortMergeJoin) SpillSome(target int64) (int64, error)

SpillSome drains buffered batches to sorted runs, larger side first, stopping once target is met. Implements memory.Spillable and memory.AccountedOperator.

type SortOrder 

type SortOrder int

SortOrder specifies sort direction. 

const (
	Ascending SortOrder = iota
	Descending
)

type Source 

type Source interface {
	Init(ctx context.Context) error
	Next(ctx context.Context) (*batch.RecordBatch, error) // nil batch = end of data
	Close() error
}

Source produces batches (table scan, hash table probe side).

type SpillableBatchCollector 

type SpillableBatchCollector struct {
	Spill *memory.SpillManager // optional; nil = plain in-memory buffering
	// contains filtered or unexported fields
}

SpillableBatchCollector is a Sink that buffers a child pipeline's entire output for later replay, with tracker accounting and spill-to-disk past memory pressure. It replaces the raw BatchSink in the join bridges (deferredJoinBridge, reverseBloomBridge), which pinned the full collected probe side in untracked heap while the downstream join simultaneously held its build side — double residency invisible to SpillManager victim selection (sweep finding #12; observed collecting probe batches at SF100).

Lifecycle: use as a Pipeline Sink, then read back with Iterate (a non-consuming pre-scan, e.g. a bloom build) and/or NextReplay (a consuming stream that drops references and deletes spill scratch as it goes). Release frees everything still held; it is idempotent and must be called from the owner's Close and error paths.

func (*SpillableBatchCollector) Close 

func (c *SpillableBatchCollector) Close() error

func (*SpillableBatchCollector) Consume 

func (c *SpillableBatchCollector) Consume(_ context.Context, b *batch.RecordBatch) error

func (*SpillableBatchCollector) Finalize 

func (c *SpillableBatchCollector) Finalize(_ context.Context) error

func (*SpillableBatchCollector) Init 

func (c *SpillableBatchCollector) Init(_ context.Context) error

func (*SpillableBatchCollector) Iterate 

func (c *SpillableBatchCollector) Iterate(fn func(*batch.RecordBatch) error) error

Iterate streams every collected batch (spilled runs first, then the in-memory tail) through fn without consuming the collector. Use for pre-replay scans like bloom builds. Must not be called once NextReplay has started consuming.

func (*SpillableBatchCollector) NewReplaySource 

func (c *SpillableBatchCollector) NewReplaySource() *CollectorReplaySource

NewReplaySource returns an exec.Source that streams the collected batches WITHOUT consuming the collector: spilled runs are re-read from disk (and not deleted), in-memory batches are handed out as Sel-isolated shallow clones so downstream operators cannot corrupt the cached data by setting Sel in place (same pattern as catalogScanSource cache replay). Multiple replay sources may be created and run concurrently — each re-reads the spill scratch independently.

Contract: the collector must be fully populated before the first replay source is created, and must not be Consumed into or Released while any replay is in flight. Spill scratch lives until Release.

func (*SpillableBatchCollector) NextReplay 

func (c *SpillableBatchCollector) NextReplay(_ context.Context) (*batch.RecordBatch, error)

NextReplay streams the collected batches in arrival order. Spilled runs are read back one batch at a time and each consumed file is deleted as soon as it is exhausted; in-memory batches have their references dropped and tracker charge returned as they are handed out, so peak residency falls monotonically during replay. Returns (nil, nil) when exhausted.

func (*SpillableBatchCollector) Release 

func (c *SpillableBatchCollector) Release()

Release frees everything still held: the open replay reader, remaining spill scratch, buffered batch references, and the tracker reservation. Idempotent — call from the owner's Close and every error path.

func (*SpillableBatchCollector) Rows 

func (c *SpillableBatchCollector) Rows() int

Rows returns the total active rows collected.

func (*SpillableBatchCollector) Schema 

func (c *SpillableBatchCollector) Schema() []parquet.Column

Schema returns the schema of the first consumed batch (nil if none).

type TopN 

type TopN struct {
	Keys []SortKey
	N    int
	// contains filtered or unexported fields
}

TopN is a Sink that combines sort + limit efficiently by keeping only the top N rows in a heap. For small N this is much more efficient than full sort.

func NewTopN 

func NewTopN(keys []SortKey, n int) *TopN

func (*TopN) Close 

func (t *TopN) Close() error

func (*TopN) Consume 

func (t *TopN) Consume(ctx context.Context, b *batch.RecordBatch) error

func (*TopN) Finalize 

func (t *TopN) Finalize(ctx context.Context) error

func (*TopN) Init 

func (t *TopN) Init(ctx context.Context) error

func (*TopN) InnerSort 

func (t *TopN) InnerSort() *Sort

InnerSort returns the underlying Sort for use with sortSourceAdapter.

func (*TopN) Next 

func (t *TopN) Next(ctx context.Context) (*batch.RecordBatch, error)

type UnaryOperator 

type UnaryOperator interface {
	Init(ctx context.Context) error
	Execute(ctx context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error) // nil = fully filtered
	Close() error
}

UnaryOperator transforms batches in-place (filter, project) — non-blocking.

type VecExpression 

type VecExpression func(b *batch.RecordBatch, out *batch.Vector, n int)

VecExpression evaluates an entire column at once, writing to the output vector. More general than VecFloat64Expression — handles any output type (string, int, etc.).

type VecFloat64Expression 

type VecFloat64Expression func(b *batch.RecordBatch, dst []float64, n int) bool

VecFloat64Expression evaluates float64 for all rows at once.

type ViewAware 

type ViewAware interface {
	AcceptsViews() bool
}

ViewAware marks an operator or sink that can consume batches containing view (dictionary) columns directly — composing or reading through the indirection — so the pipeline skips the defensive flatten in front of it.

type Window 

type Window struct {
	Columns []WindowColumn
	Spill   *memory.SpillManager // optional: enables spill-to-disk
	// contains filtered or unexported fields
}

Window is a SinkSource that collects all rows, partitions and sorts them, computes window function values, and emits the original rows with computed window columns appended. Operates directly on column vectors to avoid map[string]any materialization overhead. When a SpillManager is set, Window will spill input batches to disk under memory pressure and read them back during Finalize.

func NewWindow 

func NewWindow(cols []WindowColumn) *Window

NewWindow creates a new window operator.

func (*Window) Close 

func (w *Window) Close() error

Close releases any tracker reservation Window still holds for buffered rows that never crossed the spill threshold, and unregisters from the relief registry. The previous no-op leaked a phantom reservation for the lifetime of the process when a Window accumulated but never spilled and Finalize was not reached (early cancel).

func (*Window) Consume 

func (w *Window) Consume(_ context.Context, b *batch.RecordBatch) error

func (*Window) EstimateRelief 

func (w *Window) EstimateRelief(target int64) int64

EstimateRelief implements memory.AccountedOperator (all-or-nothing).

func (*Window) Finalize 

func (w *Window) Finalize(_ context.Context) error

func (*Window) Init 

func (w *Window) Init(_ context.Context) error

func (*Window) Inspect 

func (w *Window) Inspect() memory.OperatorFootprint

Inspect implements memory.AccountedOperator.

func (*Window) Next 

func (w *Window) Next(_ context.Context) (*batch.RecordBatch, error)

Next returns windowed results in batches. On the external path it streams one partition at a time from the final pass's merge.

func (*Window) SpillSome 

func (w *Window) SpillSome(_ int64) (int64, error)

SpillSome implements memory.AccountedOperator: drains all buffered input batches to disk and releases the freed bytes — a sorted columnar run on the external path, a raw-row file on the nested-type fallback.

type WindowBound 

type WindowBound struct {
	Type   string // "unbounded_preceding", "preceding", "current_row", "following", "unbounded_following"
	Offset int
}

WindowBound describes one end of a window frame.

type WindowColumn 

type WindowColumn struct {
	Func           WindowFunc
	InputCol       string // for aggregate window functions (empty for ranking funcs)
	OutputCol      string
	OutputType     parquet.TypeID
	PartitionBy    []string
	OrderBy        []SortKey
	Frame          *WindowFrameSpec // optional frame specification
	LagLeadOffset  int              // offset for LAG/LEAD (default 1)
	LagLeadDefault any              // default value for LAG/LEAD (default NULL)
	NtileBuckets   int              // number of buckets for NTILE
	NthValueN      int              // N for NTH_VALUE (1-based)
}

WindowColumn defines a window function computation.

type WindowFrameSpec 

type WindowFrameSpec struct {
	Mode  string // "rows" or "range"
	Start WindowBound
	End   WindowBound
}

WindowFrameSpec describes a window frame specification for execution.

type WindowFunc 

type WindowFunc int

WindowFunc identifies a window function type. 

const (
	WinRowNumber WindowFunc = iota
	WinRank
	WinDenseRank
	WinSum
	WinCount
	WinAvg
	WinMin
	WinMax
	WinLag
	WinLead
	WinFirstValue
	WinLastValue
	WinNtile
	WinPercentRank
	WinCumeDist
	WinNthValue
)

Source Files

View all Source files

Directories

Path Synopsis
Package kernel provides type-specialized vectorized operations for the query engine.
 Package kernel provides type-specialized vectorized operations for the query engine.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.