exec

package
v0.18.0 Latest Latest
Warning

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

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

Documentation

Overview

Package exec provides the push-based pipeline execution framework.

Index

Constants

View Source 
const (
	CovarKindCorr      = "corr"
	CovarKindCovarSamp = "covar_samp"
	CovarKindCovarPop  = "covar_pop"
)

Covariance-family kinds, carried in the synthetic column's name exactly as the variance kinds are: one state serves all three, and which function finishes it is decided once, after the last merge.

View Source 
const (
	VarKindStddevSamp = "stddev_samp"
	VarKindVarSamp    = "var_samp"
	VarKindStddevPop  = "stddev_pop"
	VarKindVarPop     = "var_pop"
)

Variance-family kinds. A decomposed partial carries the same (count, mean, M2) triple whatever the caller asked for, so the kind travels beside it (in the synthetic column's name) and is applied once, by FinalizeVarianceState, after the last merge.

STDDEV and VARIANCE without a suffix are the SAMPLE forms, matching DuckDB and PostgreSQL.

View Source 
const MaxProbeOutputRows = joinOutputPoolSize

MaxProbeOutputRows bounds how many match pairs one probe call materialises. Past it the fan-out is suspended at (probe row, hash-chain position) and resumed on the next call, making probe output O(batch) instead of O(batch x fan-out) — see the resume protocol on HashJoinProbe.

The value is joinOutputPoolSize, which is also pairsBuf's pre-allocated capacity and the large output pool's size. Bounding at exactly that point means every join shape that fits today keeps its current batch shape, pool, and allocation profile; only the shapes that used to grow pairsBuf without limit change behaviour.

Exported because the drivers that hold the probe to it live outside this package (internal/worker, internal/planner/physical), as do the tests that assert they do.

View Source 
const NoLimit = int64(-1)

NoLimit is the Max of a Limit that only skips rows: `OFFSET n` with no LIMIT. Any negative Max means unbounded.

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 (
	TwoLevelConversions  atomic.Int64
	TwoLevelDirectBuilds atomic.Int64
	TwoLevelBornFlat     atomic.Int64
)

TwoLevelConversions counts flat→bucketed conversions, TwoLevelDirectBuilds counts indexes built bucketed from an NDV hint, and TwoLevelBornFlat counts sinks whose layout was decided FLAT at construction because their epoch cap bounds them below twoLevelBoundedMinGroups (observability + test assertions; exported into the worker's task logs).

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 HashOnceRoutedRows atomic.Int64

HashOnceRoutedRows counts rows whose sink hash table consumed the router's hash instead of computing its own (observability + test assertions).

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 ParallelEmitRuns atomic.Int64

ParallelEmitRuns counts emissions that took the parallel drain (observability + test assertions).

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 EnableBoundedOutput 

func EnableBoundedOutput(ops []UnaryOperator)

EnableBoundedOutput opts every operator in a chain that supports the bounded-output protocol into it. Only call it from a driver that drains NextOutput after each Execute — ChainDriver, or a driver that resumes pending output the same way (the worker's fragment executors, the planner's pipelineSource).

func FinalizeCovarianceState 

func FinalizeCovarianceState(encoded, kind string) (float64, bool)

FinalizeCovarianceState decodes a merged partial state and finishes it as the named kind. ok=false means SQL NULL, on the same thresholds the single-process finalization applies: fewer than two rows for CORR and COVAR_SAMP, no rows at all for COVAR_POP.

Used by the distributed final-aggregate fold (worker/var_fold.go).

func FinalizeVarianceState 

func FinalizeVarianceState(encoded, kind string) (float64, bool)

FinalizeVarianceState decodes a merged partial state and finishes it as the named kind. ok=false means the result is SQL NULL: an unparseable or absent state, fewer than two rows for a sample form, or no rows at all for a population form — the same thresholds the single-process finalization applies.

Used by the distributed final-aggregate fold (worker/var_fold.go), which is the only consumer outside this package.

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 PauseOrDrainOnHeapBackpressure 

func PauseOrDrainOnHeapBackpressure(ctx context.Context, sink Sink) error

PauseOrDrainOnHeapBackpressure is the sink-aware valve response for a consume loop feeding a pipeline breaker: when backpressure fires, a spill-capable breaker holding the dominant tracked share drains instead of sleeping; every other sink keeps the 50 ms GC catch-up pause.

func RecoverFatalEval 

func RecoverFatalEval(r any) error

RecoverFatalEval is recoverFatalEval for the drivers that live outside this package: the embedded API's Query/Execute and the coordinator's ExecuteSQL, which reach batch-writing code (DML row building, result merging) without ever entering Pipeline.Run. batch.TypeMismatchError (#361's silent-write guard) rides this contract too, so those entries must convert it the same way. Call only with a non-nil recover() result; anything that is not a FatalEvalPanic is re-raised untouched.

func TryPressureDrain 

func TryPressureDrain(ctx context.Context, sink Sink) (bool, error)

TryPressureDrain gives sink the chance to answer heap pressure by draining its own state. It does NOT check the valve — callers do that first. handled=true means the caller must skip its sleep.

func WindowMinMaxType 

func WindowMinMaxType(in parquet.TypeID) (parquet.TypeID, bool)

WindowMinMaxType is the output type MIN/MAX over a window declare for an input column of type in, and whether they may re-declare at all.

The vetted set is exactly the types whose Vector.GetValue representation compareAny orders correctly — the ints and floats, strings (byte order), TIMESTAMP/DURATION (bare int64), DATE (its ISO string form is byte-ordered), PORT/PROTOCOL (int32). The output type is the input type: MIN/MAX return one of their input's values untouched.

ok=false declines: BYTES compares as 0-everywhere under compareAny (no []byte arm), the network types and UUID surface display strings whose byte order is not their value order ("9.x" > "10.x"), and DECIMAL surfaces a formatted string with the same problem. Those keep the planner's float64 declaration, where Vector.SetValue's type guard now reports the write it cannot perform instead of dropping it — an error, not a silently wrong ordering.

Exported because the physical planner declares from the catalog with this same function; two vetting lists would drift.

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
	// AggVarState and AggVarStateMerge carry the variance family across a
	// partial/final aggregate split. A finished STDDEV/VARIANCE cannot be
	// re-aggregated — combining partials needs the (count, mean, M2) triple,
	// not the scalar each partial reports — so a distributed plan replaces
	// STDDEV(c) AS x with VAR_STATE(c) AS __var_state#<kind>#x on the partial
	// stage (emits the encoded triple, see varianceState.encode) and
	// VAR_STATE_MERGE on every merge stage above it (pairwise-combines the
	// encoded triples and re-emits one). The final stage's fold
	// (worker/var_fold.go) decodes the triple into the value <kind> asks for.
	AggVarState
	AggVarStateMerge
	// AggCovarState and AggCovarStateMerge do the same for CORR, COVAR_SAMP
	// and COVAR_POP, whose state is the (count, meanX, meanY, C, M2x, M2y)
	// sextuple and which combine by the same pairwise rule
	// (covarianceState.merge). Before #353 the DAG had no case for these
	// function names at all, so they fell to the worker's `default: AggSum`
	// and CORR(o_totalprice, o_custkey) answered 2.127e9 — the sum of its
	// first argument.
	AggCovarState
	AggCovarStateMerge
)

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 BoundedOutputOperator 

type BoundedOutputOperator interface {
	EnableBoundedOutput()
	HasPendingOutput() bool
	NextOutput(ctx context.Context) (*batch.RecordBatch, error)
}

BoundedOutputOperator is implemented by operators whose output for a single input batch can be far larger than that batch — a hash-join probe fans one probe row out to every build row sharing its key — and which can therefore suspend mid-input and emit the remainder across later calls.

The protocol is opt-in, and the opt-in is a promise by the driver: EnableBoundedOutput says "after every Execute I will drain NextOutput until HasPendingOutput reports false, before handing you another input batch, and I will keep the input batch alive until then". Until a driver opts in the operator emits an input's whole output from Execute in one batch, so chain drivers that do not implement the protocol keep working unchanged.

Bounding the producer is what makes back-pressure possible at all: an operator that materialises O(batch x fan-out) rows in one live allocation gives the memory tracker nothing to reclaim and blows straight through GOMEMLIMIT, because the memory is live rather than garbage (#317).

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 ChainDriver 

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

ChainDriver pushes a batch through an operator chain and hands every resulting batch to deliver. It exists because an operator's output for one input batch is not always one batch: a hash-join probe bounds its fan-out per call and suspends the rest (see BoundedOutputOperator), so the chain below it has to be re-driven for each resumed slice before the operator is given its next input.

Push keeps its input alive across the whole resume loop, because a suspended fan-out still reads the probe-side columns of the batch it was handed. That is the driver half of the BoundedOutputOperator contract, so a chain driven through a ChainDriver may be opted in with EnableBoundedOutput — and one that is not stays unbounded.

It is exported because the drivers that matter live outside this package: the worker's fragment executors run the same operator chains on the distributed path, and a probe that only suspends under exec.Pipeline is a probe that still OOMs a worker (#317).

func NewChainDriver 

func NewChainDriver(ops []UnaryOperator, deliver func(context.Context, *batch.RecordBatch) error) *ChainDriver

NewChainDriver returns a driver for ops that hands every output batch to deliver. The driver does not touch batch ownership: callers that pool their batches opt in with ReleaseInputs.

func (*ChainDriver) Inspect 

func (d *ChainDriver) Inspect(fn func(opIdx int, op UnaryOperator, out *batch.RecordBatch)) *ChainDriver

Inspect installs a callback run on every operator's output, including each resumed slice. Its one caller is the worker's #277 forensics, which needs the producing operator's identity — something deliver, one step past the end of the chain, cannot see.

func (*ChainDriver) Push 

func (d *ChainDriver) Push(ctx context.Context, b *batch.RecordBatch) (exhausted bool, err error)

Push runs b through the whole chain. The bool result is the pipeline's `exhausted` signal: a DoneSignaler operator (LIMIT) reported satisfaction.

A FatalEvalPanic raised by an expression inside the chain is converted to an error here, the same contract Pipeline.Run provides. ChainDriver's callers (the worker's fragment drivers) run on errgroup goroutines with no recover of their own, so without this a runtime query error — 22012 division by zero, 22P02 invalid cast — would take the worker process down instead of failing the query.

func (*ChainDriver) ReleaseInputs 

func (d *ChainDriver) ReleaseInputs() *ChainDriver

ReleaseInputs makes the driver release every operator's input batch once that operator is finished reading it — which is after the last resumption, not after Execute. Only for callers whose deliver also releases: the batches travel to a pool, and a caller that releases what it does not own hands the same batch out twice.

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
	// RowFallback, when non-nil, evaluates the original comparison row-at-
	// a-time. Used when the two columns resolve to DIFFERENT types: the
	// typed kernel reads both sides through the left column's storage slice,
	// so a FLOAT64-vs-INT32 comparison would index the right vector's empty
	// Float64Data and panic (issue #375). The compiled expression coerces
	// numeric types per SQL semantics, so it is the correct evaluator for
	// the mixed-type case.
	RowFallback Predicate
	// 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(ctx 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.

The cache is a lazyColIdx and not a captured int because Project.Clone copies the ProjectColumn structs but SHARES this closure with every parallel worker — see lazyColIdx (filter.go) for the race that is.

func Literal 

func Literal(val any) Expression

Literal creates an expression that returns a constant.

type FatalEvalPanic 

type FatalEvalPanic interface {
	error
	FatalEvalError() error
}

FatalEvalPanic marks a panic value that carries a query ERROR rather than a bug: an expression that cannot produce a correct answer and must not fall back to NULL. Expression evaluation has no error return (Expr.Eval yields a value and nothing else), so the one such condition — a correlated subquery whose outer column is not in the batch, issue #347 — raises a panic carrying a value that implements this, and the pipeline drivers convert it back into an error. Panics that do not implement it are re-raised untouched.

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
	// GroupNDVHint is the planner's HLL-based estimate of GROUP-KEY
	// cardinality (catalog merged sketches; ~2% error) — the quantity the
	// hash table actually holds, unlike InputRowHint's input-row proxy.
	// 0 = unknown. cloneNDVDivisor spreads the hint across partitioned
	// clones (each owns a disjoint 1/k of the key space).
	GroupNDVHint int64
	// 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) ConsumeHashed 

func (h *HashAggregate) ConsumeHashed(ctx context.Context, b *batch.RecordBatch, hashes []uint64, plan *routePlan) error

ConsumeHashed is Consume with the group-key hash of every active row already computed by the partition router (hash once — see the bit-budget note in partitioned_agg.go). hashes[i] belongs to b's i'th active row.

The hashes are advisory: each consume path checks that the plan names the hash ITS table uses before consuming them, and recomputes otherwise. A sink that migrated to the generic path mid-query (a NULL key arrived) therefore keeps working with a stale plan on the wire.

func (*HashAggregate) DrainOnHeapPressure 

func (h *HashAggregate) DrainOnHeapPressure(ctx context.Context) (bool, error)

DrainOnHeapPressure implements PressureDrainer for HashAggregate. The drain routes through drainAndAccount, so the #325 productivity gate and non-convergence detection apply exactly as they do to a self-triggered spill; when the gate refuses, the aggregate reports handled (its state is live — sleeping cannot reclaim it) without writing a run.

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) GroupCeiling 

func (h *HashAggregate) GroupCeiling() int64

GroupCeiling reports the maximum group count this sink's epoch cap allows (0 when the sink is unbounded).

func (*HashAggregate) IndexBornFlat 

func (h *HashAggregate) IndexBornFlat() bool

IndexBornFlat reports whether the bounded-sink rule pinned this aggregate's group index flat — see SetEpochByteCap.

func (*HashAggregate) IndexConversions 

func (h *HashAggregate) IndexConversions() int

IndexConversions reports how many flat→bucketed group-index conversions this aggregate paid; IndexBornFlat reports whether its layout was pinned flat at construction, and GroupCeiling the Gmax that pinned it (0 = unbounded sink). Read by the worker for its per-task log lines.

func (*HashAggregate) IndexFlatReason 

func (h *HashAggregate) IndexFlatReason() string

IndexFlatReason names the bound that pinned this aggregate's group index flat at construction: "epoch-cap", "row-bound", or "" when nothing did.

func (*HashAggregate) Init 

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

func (*HashAggregate) InputRowBound 

func (h *HashAggregate) InputRowBound() int64

InputRowBound reports the declared bound (0 = none).

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.

With adopted disjoint partitions (partitioned parallel aggregation) the emission is fanned across one goroutine per partition when eligible — see aggregate_parallel_emit.go. The serial fallback below streams this aggregate's own state and then each adopted partition in turn.

func (*HashAggregate) PartitionSelectors 

func (h *HashAggregate) PartitionSelectors(b *batch.RecordBatch, parts int, sc *partitionScratch) *routedBatch

PartitionSelectors splits b's active rows into parts selection lists by group-key hash. The result carries the per-partition row ids, the matching per-row key hashes (nil runs when the plan threads nothing), the plan, and the pooled buffer the caller must hand to the sharedBatch that owns the views. Returns nil when a group column is missing or of an unsupported type — callers fall back to unpartitioned consumption.

Shape: a counting sort, not per-partition appends. One pass hashes every active row and builds the histogram; the prefix sums cut one output array into per-partition runs; the second pass scatters row ids (and hashes) into place. Row order within a partition is unchanged (the scatter is stable), and the per-batch allocation goes from ~parts append chains to a single pooled buffer the caller hands to the shared batch.

func (*HashAggregate) SetEpochByteCap 

func (h *HashAggregate) SetEpochByteCap(cap int64)

SetEpochByteCap declares that this aggregate's owner will finalize it and build a fresh one whenever StateBytes() crosses cap — a BOUNDED sink. Call it BEFORE Init: the layout of the group index is decided from this bound (two_level_hash.go, twoLevelBoundedMinGroups) and a sink that learns its cap after the first batch has already chosen.

func (*HashAggregate) SetInputRowBound 

func (h *HashAggregate) SetInputRowBound(rows int64)

SetInputRowBound declares an EXACT upper bound on the number of rows this aggregate will consume. Call it BEFORE Init, for the same reason SetEpochByteCap must be: the group-index layout is decided from it once, and a sink that learns its bound after the first batch has already chosen.

"Exact" is load-bearing. The bound may over-state (a clone reads a subset of its parent's rows and inherits the parent's bound), because over-stating only keeps the adaptive path. It must never under-state: a bound below the truth pins a genuinely high-cardinality index flat. That is why estimates — InputRowHint, GroupNDVHint — are not routed here.

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 (packed, 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

	// Residual is the ON-clause residual predicate of a LEFT, RIGHT or FULL
	// OUTER join (#358): every ON conjunct that is not an equi-join key,
	// evaluated on the COMBINED row (probe row + candidate build row) before a
	// key match is accepted. An outer join's ON runs BEFORE the NULL-padding,
	// so this cannot be a filter above the join: a probe row whose candidates
	// all fail the residual is UNMATCHED — a LEFT/FULL join still emits it
	// NULL-padded rather than dropping it — and a build row counts as matched
	// only when some probe row passed BOTH key and residual, which is what
	// FlushUnmatched consults for RIGHT/FULL. A residual returning false OR
	// NULL rejects the candidate (the compiled evaluator folds UNKNOWN to
	// false, which is the SQL ON semantics).
	//
	// With no join keys at all (`LEFT JOIN r ON n.x = r.y + 3` — no conjunct
	// is a bare-column equality) the build degenerates to a single empty-key
	// chain holding every build row, so each probe row's candidate set is the
	// whole build side and the residual does all of the work.
	Residual 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

	// BuildSchemaHint / ProbeSchemaHint declare each side's output columns at
	// PLAN time, for the case where that side delivers no batch at all and the
	// runtime therefore never learns its schema.
	//
	// An outer join still owes rows when one side is empty — a LEFT JOIN emits
	// every probe row with the build columns NULL, a RIGHT/FULL JOIN emits
	// every build row with the probe columns NULL — and it cannot name those
	// columns without a schema. buildSchema left nil produced a join output
	// carrying ONLY the preserved side: the values still read as NULL through
	// the projection's missing-column fallback, but the column was ABSENT
	// rather than NULL, so `COUNT(o.o_orderstatus)` counted 1500 of them and
	// `WHERE r.r_regionkey IS NULL` matched none (#348).
	//
	// The hints are only consulted when the side produced nothing; a real
	// batch's schema always wins, so an imprecise hint cannot corrupt a
	// non-empty join.
	BuildSchemaHint []parquet.Column
	ProbeSchemaHint []parquet.Column
	// 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) and its own fan-out cursor — parallel pipeline workers each suspend and resume independently.

func (*HashJoinProbe) Close 

func (p *HashJoinProbe) Close() error

Close drops the fan-out cursor's reference to the last input batch. A probe closed mid-suspension (a cancelled query) would otherwise keep that batch reachable for as long as the operator is.

func (*HashJoinProbe) EnableBoundedOutput 

func (p *HashJoinProbe) EnableBoundedOutput()

EnableBoundedOutput opts this probe into the BoundedOutputOperator protocol: Execute emits at most MaxProbeOutputRows joined rows and suspends the rest of the input batch's fan-out for NextOutput to resume. Only a driver that drains NextOutput after every Execute may call it.

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.

leftSchema is a fallback for the probe-side schema: the probe's own cached mapping is preferred (it is what its output batches were built from), then the schema recorded on the first Execute, then the plan-declared ProbeSchemaHint. A caller with nothing better may pass nil.

func (*HashJoinProbe) FlushUnmatchedRows 

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

FlushUnmatchedRows emits a RIGHT/FULL join's unmatched build rows exactly once per join, whichever probe clone reaches it first, and names the probe half of each row from the schema the probe itself observed.

It is the single entry point for the two drivers that own the end of a probe pipeline: physical.joinFlushSource (single process) and the worker's FlushableOperator drain (stage DAG). Before it existed only the first of those flushed at all, so every unmatched row of a distributed RIGHT or FULL join was dropped (#352) — and the caller passed the join's OUTPUT schema where the PROBE schema was wanted, so on the shapes where it did run the preserved side came back NULL.

func (*HashJoinProbe) HasPendingFlush 

func (p *HashJoinProbe) HasPendingFlush() bool

HasPendingFlush returns true if there are spilled partitions still left to process, or if a RIGHT/FULL join still owes its unmatched build rows.

func (*HashJoinProbe) HasPendingOutput 

func (p *HashJoinProbe) HasPendingOutput() bool

HasPendingOutput reports whether the last input batch still has fan-out left to emit.

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.

func (*HashJoinProbe) NextOutput 

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

NextOutput resumes a suspended fan-out and returns the next bounded slice of the current input batch's join output.

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 Max rows, optionally skipping the first Offset rows. A negative Max is unbounded, which is how OFFSET without LIMIT is expressed.

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) (err 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 PressureDrainer 

type PressureDrainer interface {
	// DrainOnHeapPressure is called between batches when the heap
	// backpressure valve fires. It returns handled=true when the operator
	// either drained state or holds the dominant live share (in which case
	// sleeping reclaims nothing and the caller should skip its pause);
	// handled=false means the operator holds too little for draining to
	// relieve the pressure, and the caller should sleep as before.
	DrainOnHeapPressure(ctx context.Context) (handled bool, err error)
}

PressureDrainer is implemented by pipeline breakers that can answer heap backpressure by draining their own state to disk instead of having their feed loop sleep.

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) EnableBoundedOutput 

func (p *ProfiledOperator) EnableBoundedOutput()

EnableBoundedOutput forwards the bounded-output opt-in to the wrapped operator. Without the forward, wrapping a hash-join probe in a profiler would silently hide its resume protocol from the driver.

func (*ProfiledOperator) Execute 

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

func (*ProfiledOperator) HasPendingOutput 

func (p *ProfiledOperator) HasPendingOutput() bool

HasPendingOutput reports whether the wrapped operator suspended part of the current input batch's output.

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).

func (*ProfiledOperator) NextOutput 

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

NextOutput resumes the wrapped operator's suspended output.

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
	// Computed marks an output whose value comes from Expr rather than from
	// an input column of the same name. Such an output must NOT be typed by
	// looking its own name up in the input: when the alias shadows an input
	// column, that lookup types the vector from a column the value paths
	// never read. Only the planner can tell the two apart — Expression is
	// an opaque func here.
	Computed bool
}

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 SetOpEmit 

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

SetOpEmit turns grouped per-arm counts into an INTERSECT/EXCEPT answer (#346). Its input is the drain of a counting hash aggregate: one row per DISTINCT result row, carrying the result columns plus two count columns — the row's multiplicity in arm A (leftCol) and in arm B (rightCol). The operator emits k copies of each row per the operation's rule and drops the count columns: 

INTERSECT      k = 1 if countA > 0 && countB > 0, else 0
INTERSECT ALL  k = min(countA, countB)
EXCEPT         k = 1 if countA > 0 && countB == 0, else 0
EXCEPT ALL     k = max(0, countA − countB)

The distinct forms never copy a row: k ∈ {0,1} means the output is a selection over the input, so the operator sets a selection vector and re-slices the batch's columns (the count columns drop zero-copy, like ColumnPrune). The ALL forms materialize, because k > 1 has no selection representation; output size is the operation's true answer size for the input batch, the same expansion contract a hash-join probe has.

NULL handling is inherited, which is the point: the upstream GROUP BY already treats NULLs as equal (SQL's set-operation membership rule), so by the time a row reaches this operator its counts are settled and its values are opaque. The count columns themselves are SUMs of literal 0/1 tags over ≥1 row per group and therefore never NULL; a NULL count is read as 0 defensively.

func NewSetOpEmit 

func NewSetOpEmit(op string, all bool, leftCol, rightCol string) (*SetOpEmit, error)

NewSetOpEmit validates the spec and constructs the operator.

func (*SetOpEmit) Clone 

func (s *SetOpEmit) Clone() UnaryOperator

Clone gives parallel drivers a scratch-independent copy.

func (*SetOpEmit) Close 

func (s *SetOpEmit) Close() error

func (*SetOpEmit) Execute 

func (s *SetOpEmit) Execute(_ context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error)

func (*SetOpEmit) Init 

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

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) DrainOnHeapPressure 

func (s *Sort) DrainOnHeapPressure(ctx context.Context) (bool, error)

DrainOnHeapPressure implements PressureDrainer for Sort. Draining flushes the buffered batches to a sorted run — the same operation Sort's own ShouldSpillFor trigger performs — bounded below by minSortRunBytes so run files stay merge-worthy.

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) DrainOnHeapPressure 

func (w *Window) DrainOnHeapPressure(ctx context.Context) (bool, error)

DrainOnHeapPressure implements PressureDrainer for Window; same shape as Sort (columnar runs, minSortRunBytes floor).

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
)

func ParseWindowFunc 

func ParseWindowFunc(s string) (WindowFunc, bool)

ParseWindowFunc maps a SQL window function name (case-insensitive) onto its WindowFunc constant. ok is false for a name this operator does not implement, and the returned WindowFunc is then WinRowNumber — the zero value, which the single-process planner has always fallen back to. A caller that ships the spec somewhere else (the distributed fragment builder) should refuse on !ok instead: computing ROW_NUMBER for a function nobody recognized is a wrong answer with no error attached.

It lives here rather than in the planner because both the planner and the worker turn a name into this package's constant, and two switch statements are two chances to disagree.

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.