exec

package
v0.18.5 Latest Latest
Warning

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

 Go to latest Published: Aug 30, 2026 License: AGPL-3.0 Imports: 30 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 KeyTypeUnresolved = batch.TypeID(-1)

A join key is built at the pair's COMMON type, not at each side's own storage type (#615, #650, #663).

ADR-0023's rule is that a key and the comparator name ONE relation: "these two rows compare equal" and "these two rows key alike" have to be the same statement, or the join answers something the WHERE spelling of it does not. A comparison already resolves its operand pair to PostgreSQL's common type before comparing; the key did not. It called appendColumnValue with `v.Type` on each side independently, so `a.i = b.d` (INT64 against DECIMAL) built eight little-endian bytes on one side and a canonical decimal key on the other, and matched only where those byte strings coincided by accident — one pair over a ten-row fixture, not zero, which is why the failure reads as a wrong answer rather than as an empty one.

The resolved type is decided at PLAN time (physical.resolveJoinKeyTypes, which is where the two sides' declared types are both in hand) and carried into the operator, on BOTH execution paths and into the shuffle's partition hash — a repartition that routes at the column's own width sends equal values to different partitions, which is the same defect one layer down.

KeyTypeUnresolved means "this pair needs no widening": either the planner could not type one of the sides, or the two sides already agree. It is the value every existing caller gets, and it takes exactly the code path the key had before, byte for byte.

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.

View Source 
const SQLStateInternalError = "XX000"

SQLStateInternalError is PostgreSQL's internal_error, what a client is told when the server hit something it has no better code for.

Variables

View Source 
var (
	// BloomSelfCheckFailures counts filters that could not match keys taken
	// from their own insert side. Every one is a wrong-answer bug.
	BloomSelfCheckFailures atomic.Int64
	// BloomKeyTypeMismatches counts filters disengaged because the column
	// they resolved encodes its keys differently from the column they were
	// built from.
	BloomKeyTypeMismatches atomic.Int64
	// BloomFullRejections counts filters observed rejecting every row they
	// have seen. Legal (disjoint key sets) but always worth a look.
	BloomFullRejections atomic.Int64
)

Counters for the three things a bloom filter can do that are not a normal filtering decision. Exported so tests and the harness can assert on them: #543 was invisible precisely because a filter rejecting 100% of its input produced no counter, no log line and no failure.

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

ContainerKeyDrainWrites counts container group-key values written into a partial-state run — one per group per drain, across spills AND the morsel-parallel clone handoff, which use the same run format.

It is exported because a gate at the SQL layer cannot otherwise tell a query that exercised this path from one that never left memory. The first version of the 1 KiB-budget gate compared a budgeted answer with an unbudgeted one for all four container types and was VACUOUS for three of them: only VECTOR ever reached a drain, so ARRAY/ROW/MAP compared two in-memory runs to each other and would have passed with this whole file deleted. A gate that cannot say whether it engaged is not a gate.

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

spillOneInMemoryPartition picks the largest in-memory partition, writes its batches to disk via the spillState writer, frees the corresponding entries in h.buildBatches (sets them nil so column data is GC-eligible) and releases that partition's bytes from h.MemTracker. Returns the number of bytes freed from the tracker, or 0 if no in-memory partition exists.

Caller must hold h.mu. The hash-table arena entries that point to the freed batch slots are NOT removed: they remain in the chain but are unreachable on the in-memory probe path because HashJoinProbe.Execute routes spilled- partition rows to disk before any hash lookup. Their ~12 bytes/row of arena + arenaNext overhead is a tracked-but-not-freed residual; the dominant memory cost (column data) is freed cleanly. JoinPartitionsEvicted counts grace-partition evictions (spillOneInMemoryPartition) across the process.

It exists so a gate can PROVE it reached the eviction path instead of skipping when it did not. #550's pin had to `t.Skip` when no partition was evicted, which on a machine where the fixture stayed resident made the pin silently vacuous; a counter turns "the shape stopped spilling" into a failure of the test that depends on it.

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.

"Rescued" is only sometimes the word. The repair's premise is that a left key present in the build schema must be misassigned, and that premise is FALSE whenever the bare name is on BOTH sides — every self-join. There the swap leaves the probe resolving a name only the build has, the join matches nothing, and the query answers zero rows with no error: it was the second half of #516 and it is the mechanism of #526. So a firing on a planner-produced plan is a defect signal, not a save.

Asserted to stay 0 by benchmarks/tpch.TestTPCHQueries over the whole 22-query corpus and by physical.TestBushyBuild_* over the bushy-build shapes. A nonzero count means a plan shape leaked through the planner's ownership resolution (rebuild cost, a hazard for partitioned builds, and possibly a wrong answer).

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, and carrying the same warning: on a self-join the swap can be wrong rather than corrective. Should stay 0 on planner-produced plans. NOTE: unlike KeyAssignmentRepairs, no suite asserts that today — this is observability, not a gate.

Functions

func AppendBoxedGroupKey added in v0.18.3 

func AppendBoxedGroupKey(dst []byte, v any, col *parquet.Column) []byte

AppendBoxedGroupKey is appendKeyValueWithMeta under an exported name, for the ONE consumer outside this package that builds the same key from the same boxed value: the coordinator's cross-worker GROUP BY re-aggregation (reAggregatePartials' keyEncoders).

That layer keys a container column by `fmt.Appendf("%v", ...)`, which is not injective for any of the four — ARRAY['a b'] and ARRAY['a','b'] both render `[a b]`, ROW{a:'b c:d'} and ROW{a:'b',c:'d'} both render `map[a:b c:d]` — so two distinct groups merged into one at the coordinator while every worker, and the single-process engine, kept them apart. It was unreachable while a container GROUP BY failed outright (#566/#576); making those queries answer is what exposes it, so the two fixes belong together.

Exporting THIS rather than a fresh encoder is the point: the bytes have to be the ones the engine's own boxed merge key produces, or the coordinator re-splits what a worker merged. That includes the float fold (a NaN payload and a -0.0 are not part of a value's identity, kernel/float_order.go) and the CIDR re-key (#520), both of which a value-preserving encoding — appendContainerKeyValue, which the drained partial's VALUE uses — must not apply and this one must.

func AppendWidenedKeyValue added in v0.18.5 

func AppendWidenedKeyValue(buf []byte, v *batch.Vector, row int, target batch.TypeID) []byte

AppendWidenedKeyValue encodes v's row as key bytes AT `target`, which is the resolved common type of the pair this column is one half of.

The `target == v.Type` case is the whole of the common case — every same-type join, which is every TPC-H join — and it is a single comparison ahead of the untouched appendColumnValue call. Nothing else here runs for it.

The caller has already established that the row is NOT null and written the flag byte, exactly as it does for appendColumnValue.

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.

Prefer BloomBuilder: it freezes the encoding and retains the sample that makes a broken filter loud. This wrapper keeps the free-function shape for callers that already own the bits, and shares the builder's one encoder so it cannot drift from the probe side again.

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 CatchQueryPanic added in v0.18.2 

func CatchQueryPanic(ctx context.Context, where string, report func(error))

CatchQueryPanic is the deferred form, for a goroutine that reports its error somewhere other than a return value: 

defer CatchQueryPanic(ctx, "hash join build", func(err error) {
    buildErr = err
    cancel()
})

report is called only when a panic was in flight, so the happy path costs one deferred call per goroutine and nothing per batch or per row.

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 IsQueryPanicMessage added in v0.18.2 

func IsQueryPanicMessage(msg string) bool

IsQueryPanicMessage reports whether a failure message came from a query boundary — including one that crossed the distributed wire as text and was then wrapped by the coordinator's stage/task framing, hence Contains rather than HasPrefix.

func KeyTypeAt added in v0.18.5 

func KeyTypeAt(types []batch.TypeID, i int, own batch.TypeID) batch.TypeID

KeyTypeAt is the type the i'th key column of `types` must be encoded at, or the column's own type when the pair is unresolved or already agreed.

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 QueryIDFromContext added in v0.18.2 

func QueryIDFromContext(ctx context.Context) string

QueryIDFromContext returns the id WithQueryID attached, or "".

func QueryPanicsRecovered added in v0.18.2 

func QueryPanicsRecovered() int64

QueryPanicsRecovered returns how many unexpected panics this process has converted at a query boundary. Deliberately raised FatalEvalPanics are query errors, not panics in this sense, and are not counted.

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 RecoverQueryPanic added in v0.18.2 

func RecoverQueryPanic(ctx context.Context, where string, r any) error

RecoverQueryPanic converts a non-nil recover() value into an error and never re-panics.

  • A FatalEvalPanic keeps its precise error and SQLSTATE — the designed class, unchanged.
  • Anything else becomes a *QueryPanic: SQLSTATE XX000, logged at error level with the query id and a truncated stack, and counted.

where names the boundary, for the log line and the message.

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 WindowDecimalAggMeta added in v0.18.5 

func WindowDecimalAggMeta(fn WindowFunc, inScale int) (precision, scale int)

WindowDecimalAggMeta is the (precision, scale) a windowed SUM or AVG declares over a DECIMAL input of scale inScale. It is the window's copy of aggSpecOutputDecimal's SUM/AVG arms — deliberately the same two lines, so a change to one is a visible divergence from the other.

Exported for the same reason WindowMinMaxType is: the physical planner declares this column from the catalog and the operator declares it from the vector it reads, and two implementations of one rule are two chances to disagree about the type of one answer.

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 output type IS the input type, for every type the engine has: MIN/MAX return one of their input's values untouched, so the only declaration that can hold the answer is the one the value came out of. That is minMaxOutputType's rule (aggregate.go) and MIN_BY's before it (#392), and the two must agree — `MIN(c) OVER (PARTITION BY g)` and `MIN(c) … GROUP BY g` are the same question asked twice, and a client that reads both in one result set gets two column types for one answer if they disagree.

This used to be an ALLOW-LIST of ten types, and everything else kept the planner's float64 declaration on the reasoning that the in-memory MIN/MAX deque chose its answer with compareAny over Vector.GetValue's box, which has no type tag to route a CIDR to kernel.CidrOrderKey. Both halves of that have since stopped being true: the deque compares COLUMNAR (kernel.CompareValuesAt, right here in computePartitionColumnar) and the spill and global-window paths resolve newBoxedCompare from the declaration (compare_boxed.go). What the declining left behind was not a safe fallback but a FAILED QUERY — Vector.SetValue's #361 guard reporting "cannot store string into FLOAT64 vector" for a shape BI tools generate routinely, over twelve of the twenty-two types (#569): the eight scalars CIDR/UUID/IPV6/IPV4/MAC/DECIMAL/BYTES/BOOL, and ARRAY/ROW/MAP/VECTOR, while the plain aggregate over the identical column answered correctly.

Two types' window output differs from the grouped aggregate's — INT32 and FLOAT32. The grouped MIN/MAX widens INT32 to INT64 and FLOAT32 to FLOAT64 because its accumulator is the wider type; the window copies an input value rather than accumulating one, so nothing forces the widening and it keeps INT32 and FLOAT32. Those narrower declarations are the PostgreSQL-correct ones: `min(int4)` is `int4` and `min(real)` is `real` there, both ways.

The bool result is kept, rather than returning a bare type, because the planner's caller has a second question the exec caller does not: whether to leave windowOutputType's fallback standing for an input type it could not resolve at all. Every type the engine has answers true.

Exported because the physical planner declares from the catalog with this same function; two 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.

func WithQueryID added in v0.18.2 

func WithQueryID(ctx context.Context, id string) context.Context

WithQueryID tags ctx with the id every panic logged under it will name.

Types

type AggColumn 

type AggColumn struct {
	Func       AggFunc
	InputCol   string // input column name (empty for COUNT(*))
	OutputCol  string // output column name
	OutputType parquet.TypeID
	// OutputPrecision/OutputScale carry a DECIMAL OutputType's (p,s) — the
	// piece a bare TypeID cannot hold, and the half of a DECIMAL's VALUE that
	// the .wshf header carries (ADR-0010). outputSchema fills them in from the
	// input VECTOR whenever one was observed; these are what it declares when
	// one never was, which is exactly the ungrouped identity row a partial
	// task emits after a selective filter matched none of its rows. Without
	// them that row shipped a file declaring DECIMAL(0,0), and the aggregate
	// merging it read a scaled Int128 as unscaled — 10^scale too large (#685).
	//
	// Zero means "the planner declared no (p,s)", which is every non-DECIMAL
	// aggregate and a DECIMAL one whose input is not a bare column reference.
	OutputPrecision int
	OutputScale     int
	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
	// InputColIdx pins the input to a physical column POSITION, bypassing
	// name resolution, when InputColIdxSet is true. A distributed merge over
	// two aggregates sharing one alias (#575) reads two partial columns of
	// the SAME name; the name path resolves both to the first, collapsing
	// them, so the worker addresses each by its ordinal partial slot instead.
	InputColIdx    int
	InputColIdxSet bool
}

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 BloomBuilder added in v0.18.3 

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

BloomBuilder accumulates one key column's values into a bloom and then hands back the BloomFilterOp that probes it.

It exists so the two sides cannot disagree about the key encoding: the encoding is decided once, here, from the inserted column's own type, frozen against a later batch that disagrees, and handed to the op the builder produces along with a sample of the keys that went in (#543).

func NewBloomBuilder added in v0.18.3 

func NewBloomBuilder(totalRows int) *BloomBuilder

NewBloomBuilder allocates a builder sized for totalRows keys. Returns nil for zero rows, matching NewBloomSized.

func (*BloomBuilder) Add added in v0.18.3 

func (bb *BloomBuilder) Add(b *batch.RecordBatch, keyCol string) error

Add hashes one batch's key column into the bloom. A batch whose key column has a different type from the first one's is refused rather than encoded two ways.

func (*BloomBuilder) Bloom added in v0.18.3 

func (bb *BloomBuilder) Bloom() ([]uint64, uint64)

Bloom returns the raw bits and mask, for callers that hand a bloom on rather than probing it here.

func (*BloomBuilder) FilterOp added in v0.18.3 

func (bb *BloomBuilder) FilterOp(column string) *BloomFilterOp

FilterOp returns the operator that probes this bloom over the named column, carrying the encoding the builder froze and the sample SelfCheck replays.

func (*BloomBuilder) Inserted added in v0.18.3 

func (bb *BloomBuilder) Inserted() int64

Inserted returns how many non-NULL keys went into the bloom, and Resolved whether the key column was ever found. A caller must not install a filter built from an unresolved column: it rejects everything, which for an anti-join means inventing unmatched probe rows.

func (*BloomBuilder) Resolved added in v0.18.3 

func (bb *BloomBuilder) Resolved() bool

Resolved reports whether the key column was found in at least one batch.

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.

func (*BloomFilterOp) SelfCheck added in v0.18.3 

func (op *BloomFilterOp) SelfCheck() error

SelfCheck probes the filter with a sample of the very values that were inserted into it. All of them must survive. A miss means the insert side and the probe side hashed different byte streams, which is not a lost optimization but a lost row on every rejection the filter makes (#543).

Returns nil when the op carries no sample — see BloomBuilder.FilterOp.

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 panic 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. Since #511 that holds for an UNEXPECTED panic too: re-raising it past this point reached an errgroup goroutine and ended the process.

The defer is per batch, not per row, and it already existed — the boundary added no new cost to this path.

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
	// SchemaHint is the PLAN-DERIVED output schema, set by the planner
	// before the pipeline runs. Schema() falls back to it when the sink
	// consumed nothing, which is the only case it can be needed and the only
	// case it is consulted.
	//
	// A query's output schema is a property of the PLAN, not of the data,
	// but this sink learned it from the first batch it consumed — so
	// `SELECT a, b FROM t WHERE false` had no schema, and every route that
	// reads names or types off this sink handed the client nothing:
	// pgwire's coord path then declared OID 25 (text) for every column, and
	// the coordinator's correlated-local route a RowDescription with ZERO
	// FIELDS — not an empty table with headers, no table at all (#416).
	//
	// It is a HINT, not an override: a consumed batch always wins, because
	// the runtime saw the vectors and the planner only predicted them. Init
	// deliberately does not clear it — it is configuration the planner
	// attaches once, not per-run state.
	SchemaHint []parquet.Column
	// SchemaHintWireUnconstrainedDecimal names the DECIMAL output columns
	// whose PostgreSQL wire typmod must say "unconstrained" (-1) — an
	// aggregate function call, never a bare column reference. Unlike
	// SchemaHint, this is consulted for EVERY result, zero-row or not:
	// which columns are aggregate output is a property of the PLAN, not of
	// whether Consume ever ran (FIX 2, #457/#458 fold-in).
	SchemaHintWireUnconstrainedDecimal map[string]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, or the planner's SchemaHint when no batch was consumed. Unlike Batches()[0].Schema, it remains available after ToRows releases the batches.

func (*CollectSink) ToRowValues added in v0.18.3 

func (s *CollectSink) ToRowValues() [][]any

ToRowValues returns the result POSITIONALLY — one []any per row, aligned with Schema() — or nil when the map form is already lossless.

A result may legally carry two output columns of the SAME NAME: PostgreSQL answers `SELECT abs(a), abs(b)` with two columns both called `abs`, and #513 made this engine agree. A map keyed by name cannot hold both, so the second overwrites the first and a consumer reads column 0's value under column 1's name — a wrong VALUE, which is strictly worse than a wrong name.

nil is the answer when the schema's names are unique, and it is not a hedge: it says the map IS the positional form, and a caller reading it by name gets the same cells. Materializing a []any per row unconditionally would add to a sink that has been measured at 68% of inuse_space on large results (SkipFinalizeToRows' comment), for a shape almost no query has, so the cost is paid exactly where the loss would otherwise happen.

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 DecimalCoerce added in v0.18.3 

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

DecimalCoerce puts named columns into ONE declared DECIMAL(p,s), rewriting the unscaled carrier as it goes.

It exists for a set operation's arms. A DECIMAL value is an UNSCALED integer plus the column's declared SCALE (ADR-0018 §4), and the two travel apart on the stage DAG: each arm's task writes its own .wshf file carrying its own scale in the header, and a downstream task that reads several such files writes ONE file under the schema of the first batch it saw. Two arms at different scales therefore hand the same unscaled integer to a reader that believes a different scale — 12.7501 from a DECIMAL(18,4) arm came back as 1275.01 under a DECIMAL(9,2) first arm, 100x too large, with nothing anywhere reporting a problem (#533).

The fix is to make the arms AGREE before they meet, which means moving the values: rescaling to the set operation's output scale is a multiplication by a power of ten, not a reinterpretation. INT32/INT64 arms are coerced the same way, because `numeric UNION ALL bigint` is `numeric` in PostgreSQL and an integer box is a value at scale 0.

Only UPWARD moves are accepted. The output scale is the max over the arms, so no arm is ever asked to drop digits; a request to scale DOWN is a planner defect and is refused rather than silently truncating.

A value with no Int128 at the output scale is an ERROR naming the column, not a wrapped one — the same rule, and the same reason, as SUM's overflow (ADR-0012 item 9): a wrapped number is a different number wearing the right type, and nobody downstream can tell.

func NewDecimalCoerce added in v0.18.3 

func NewDecimalCoerce(cols []DecimalCoerceColumn) *DecimalCoerce

NewDecimalCoerce returns an operator that coerces the named columns. An empty list is a pass-through.

func (*DecimalCoerce) Clone added in v0.18.3 

func (d *DecimalCoerce) Clone() UnaryOperator

Clone satisfies Cloneable so a fragment carrying this operator can still run its morsel workers in parallel. The clone re-resolves against its own first batch; nothing here is shared state.

func (*DecimalCoerce) Close added in v0.18.3 

func (d *DecimalCoerce) Close() error

func (*DecimalCoerce) Execute added in v0.18.3 

func (d *DecimalCoerce) Execute(_ context.Context, in *batch.RecordBatch) (*batch.RecordBatch, error)

func (*DecimalCoerce) Init added in v0.18.3 

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

type DecimalCoerceColumn added in v0.18.3 

type DecimalCoerceColumn struct {
	Name      string
	Precision int
	Scale     int
}

DecimalCoerceColumn names one column and the DECIMAL(Precision, Scale) it must carry from here on.

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 lazyFieldIdx 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
	// Check is the row path's half of the #147 guard, run ONCE on the first
	// batch: a predicate whose column references name nothing in the input is
	// a query error, never UNKNOWN on every row.
	//
	// KernelFilter has refused that since #147, because a filter that matches
	// nothing is indistinguishable from genuinely empty data. The row
	// evaluator had no equivalent — expr.ColRef.Eval simply answers nil — so
	// every defect that handed this operator the wrong NAME came back as a
	// silent zero-row answer (#653). The check lives here rather than inside
	// the predicate because a Predicate returns bool and has nowhere to put
	// an error; callers that know the predicate's references set it
	// (expr.CheckFilterColumns), and callers that do not leave it nil.
	//
	// WHO sets it is the whole of its safety, and the two paths differ. The
	// single-process planner sets it on every non-correlated row filter,
	// because in one process each operator DECLARES its output schema and an
	// empty join side still declares the columns it would have produced. The
	// DAG sets it only on a filter reading a base-table SCAN
	// (OpSpec.ScanSchemaFilter): a stage's input schema there is read back
	// from what an upstream task WROTE, and a hash-join partition whose build
	// side was empty writes only the join keys for the missing side — so a
	// build column that is legitimately NULL for every row of that partition
	// is absent from the schema, which is TPC-H Q20's
	// `ps_availqty > 0.5 * __scalar_0` and not a defect.
	Check func(*batch.RecordBatch) error
	// 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 FirstError added in v0.18.2 

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

FirstError records the first error any goroutine in a group reports.

sync/atomic.Value panics when two stores into the same Value carry different CONCRETE types, and "whichever worker failed first" is exactly the slot where that happens: one worker stores what the panic boundary produced, another stores a *fmt.wrapError from its own return path, and the loser of that race takes the whole process down — every connection, not just the query that raced (#512, seen 17 times in one SQLancer soak).

Boxing every error in this package's own struct makes the stored type uniform by construction, so no call site has to remember the rule and no future call site can break it by returning a differently-shaped error. Store the box, never the error.

func (*FirstError) Err added in v0.18.2 

func (f *FirstError) Err() error

Err returns the first recorded error, or nil when none was recorded.

func (*FirstError) Set added in v0.18.2 

func (f *FirstError) Set(err error)

Set records err if nothing has been recorded yet. A nil err is ignored, so callers can hand it an unconditional result.

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

	// KeyTypes[i] is the resolved COMMON type of the pair
	// (LeftKeys[i], RightKeys[i]) — PostgreSQL's operator resolution over
	// the two sides' declared types, computed at plan time by
	// physical.resolveJoinKeyTypes. Both sides' key bytes are built at it,
	// and the integer / bloom fast paths are gated on it rather than on
	// either column's storage. Nil, short, or KeyTypeUnresolved means "no
	// widening for this pair", which is every same-type join and the whole
	// of TPC-H: the encoder then takes exactly the path it took before.
	// See join_key_width.go.
	KeyTypes []batch.TypeID

	// 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

	// NullAwareAnti marks an AntiJoin that must answer `NOT IN (subquery)`
	// rather than the two-valued question an anti join asks on its own.
	//
	// An anti join emits a probe row when nothing in the build matched it.
	// `k NOT IN (SELECT v …)` is three-valued: TRUE only when k differs from
	// EVERY v, FALSE when it equals one, and UNKNOWN — so WHERE drops the
	// row — the moment k itself is NULL, or the subquery yielded a NULL that
	// k did not match on some other value. A plain anti join cannot tell "no
	// match because the row genuinely differs" from "no match because NULL
	// equals nothing", and emits both (#507).
	//
	// Two rules restore the difference, and both are decided by the BUILD:
	// a probe row whose own key is NULL never survives, and a NULL anywhere
	// in the build poisons every non-matching comparison, so the join's whole
	// output is empty. See buildHasNullKey.
	NullAwareAnti 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
	// ValueTexts holds each value's exact literal source text, parallel to
	// Values, for a DECIMAL column — see KernelFilter.LitText (#452). Nil, or
	// an empty entry, keeps the boxed value.
	ValueTexts []string
	// RowFallback mirrors KernelFilter.RowFallback: the row-at-a-time
	// predicate for a ROW field path the set kernel cannot address, which
	// otherwise reported the field path as a column that does not exist
	// (#568).
	RowFallback Predicate
	// 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 NewInFilterLit added in v0.18.1 

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

NewInFilterLit is NewInFilter for a list of numeric literals, carrying each literal's exact source text for a DECIMAL column (#452).

func (*InFilter) Clone 

func (f *InFilter) Clone() UnaryOperator

func (*InFilter) Close 

func (f *InFilter) Close() error

func (*InFilter) Execute 

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

func (*InFilter) Init 

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

func (*InFilter) SetSyntacticLen added in v0.18.4 

func (f *InFilter) SetSyntacticLen(n int)

SetSyntacticLen records the list's pre-NULL-strip element count, which the caller that does the stripping (planner inFilterForList) knows and the constructor cannot. It is the arity PostgreSQL decides a real IN list's width from (#549); leaving it at the constructor's len(Values) is correct only when no NULL was dropped.

type Int64Expression 

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

Int64Expression evaluates to int64 without boxing.

type Int128Sum added in v0.18.5 

type Int128Sum = batch.Int128

Int128Sum is batch.Int128 under a name that says what it holds here. The alias keeps the streaming state's declarations readable next to the float ones they parallel.

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
	// LitText is the constant's exact source text, carried alongside the box
	// for the one type whose values a float64 cannot hold: a DECIMAL column's
	// kernel takes the TEXT and converts it at the column's own scale, so a
	// literal with more significant digits than a double survives (#452).
	// Used only when the resolved column is a DECIMAL — every other kernel
	// reads Value exactly as before. Empty for a non-numeric constant.
	LitText string
	// 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 NewKernelFilterLit added in v0.18.1 

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

NewKernelFilterLit is NewKernelFilter for a constant that came from a numeric literal, carrying the literal's exact source text for a DECIMAL column (#452).

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
	// RowFallback mirrors KernelFilter.RowFallback: the row-at-a-time
	// predicate for a ROW field path the kernel cannot address (#568).
	RowFallback Predicate
	// 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(ctx 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 MatchNothingFilter added in v0.18.1 

type MatchNothingFilter struct{}

MatchNothingFilter admits no rows. It is the operator for a predicate that is UNKNOWN on every row whatever the data says — a comparison against a NULL literal, and its negation too, since NOT UNKNOWN is UNKNOWN. A WHERE admits only TRUE, so the answer is no rows.

Saying that in the PLAN is the point. Lowering `col = NULL` to a typed kernel handed it a nil constant, which every coercion in kernel.ResolveFilterKernel turns into the column type's ZERO: `WHERE c_i64 = NULL` answered the rows where the column is 0, `WHERE c_str = NULL` the rows where the string is empty (#450).

func NewMatchNothingFilter added in v0.18.1 

func NewMatchNothingFilter() *MatchNothingFilter

func (*MatchNothingFilter) Clone added in v0.18.1 

func (f *MatchNothingFilter) Clone() UnaryOperator

func (*MatchNothingFilter) Close added in v0.18.1 

func (f *MatchNothingFilter) Close() error

func (*MatchNothingFilter) Execute added in v0.18.1 

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

Execute returns the no-rows-survive signal every other filter uses when its selection comes back empty.

func (*MatchNothingFilter) Init added in v0.18.1 

func (f *MatchNothingFilter) 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
	// RowFallback, when non-nil, evaluates the original IS [NOT] NULL
	// row-at-a-time for the shape this bitmap scan cannot serve: a ROW FIELD
	// PATH, whose null is the FIELD's and not the container's. Without it
	// the name resolved to nothing and `WHERE rw.f IS NULL` answered NO ROWS
	// on every input — an empty result indistinguishable from real data,
	// the #147 failure mode one level down (#568).
	RowFallback Predicate
	// 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(ctx 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 ColumnCompareLit added in v0.18.1 

func ColumnCompareLit(colName string, op CompareOp, value any, litText string) Predicate

ColumnCompareLit is ColumnCompare for a constant that came from a numeric literal, carrying the literal's exact source text so a DECIMAL column is compared in its own domain rather than against a float64 (#452).

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)
	// SourceIdx names the input column by POSITION, for a projection whose
	// source cannot be identified by name: an output list may legally carry
	// two columns of the same name (`SELECT abs(a), abs(b)` — PostgreSQL
	// calls both `abs`), and a name-keyed copy then gives the second one the
	// first one's values. SourceIdxSet is required because 0 is a valid
	// index, the same reason ProjectExprSpec carries TypeKnown.
	SourceIdx    int
	SourceIdxSet bool
	// VecDecimalEval is the vectorized EXACT fixed-point path: it writes
	// unscaled Int128 carriers straight into a DECIMAL output vector, with no
	// box and no allocation (expr.BinOpNumeric.EvalDecimalVec, ADR-0024
	// item 3).
	//
	// It is a field of its own rather than VecEval because the DECIMAL arm of
	// Execute deliberately runs AHEAD of every vectorized path — the checked
	// per-row writer is the only route with an error channel, and no other vec
	// kernel writes DecimalData. Giving this its own field keeps that ordering
	// intact for everything else while letting the one kernel that DOES write
	// DecimalData skip the box.
	//
	// It returns whether it WROTE the batch. False means the exact mode does
	// not apply to this batch after all — the planner declared DECIMAL from
	// the AST and the runtime resolved the operands differently — and the
	// caller must run the boxed checked writer instead. Returning nothing and
	// writing nothing leaves the output vector's zeros standing, which reads
	// back as the value 0 on every row: a silent wrong answer of exactly the
	// class this work exists to close.
	VecDecimalEval VecDecimalExpression
	Dimension      int // VECTOR output dimensionality (e.g. embed()); 0 = not a vector
	// Precision and Scale declare a COMPUTED DECIMAL output, the same way
	// Dimension declares a computed VECTOR one: the output column does not
	// exist in the input, so there is no vector to read (p,s) off, and a
	// DECIMAL vector built without them comes out at SCALE 0 — every value
	// in it read back a hundred- or ten-thousand-fold out. Zero for every
	// other type, and for a passthrough, where the input column answers
	// (ADR-0024 item 2; #529, #555).
	Precision int
	Scale     int
	// 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 QueryPanic added in v0.18.2 

type QueryPanic struct {
	// Where names the boundary that caught it — "hash join build",
	// "pipeline worker", "coordinator query" — so a log line says which
	// goroutine died without needing the stack parsed.
	Where string
	// Value is the recovered panic value.
	Value any
	// Stack is the goroutine's stack at the panic, truncated.
	Stack string
}

QueryPanic is the error an unexpected panic becomes at a query boundary.

func (*QueryPanic) Error added in v0.18.2 

func (p *QueryPanic) Error() string

func (*QueryPanic) SQLState added in v0.18.2 

func (p *QueryPanic) SQLState() string

SQLState satisfies sqlerr.Coder, so pgwire reports XX000 rather than the blanket class it applies to an uncoded error.

func (*QueryPanic) Unwrap added in v0.18.2 

func (p *QueryPanic) Unwrap() error

Unwrap exposes a panicked error value (a runtime.Error, say) to errors.Is and errors.As.

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 is the Top-K row bound, or NoLimit (-1, the same sentinel
	// exec.Limit.Max and logical.NoLimit use) when the sort is unbounded.
	// Every reader below tests `>= 0` (never `> 0`) so a real zero is never
	// mistaken for "unbounded" — that collision was #481:
	// `ORDER BY ... LIMIT 0` returned every row because 0 doubled as the
	// "no limit" sentinel.
	Limit int

	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) — Limit is a real row count, so LIMIT 0 correctly materializes zero rows.

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 VecDecimalExpression added in v0.18.5 

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

VecDecimalExpression is VecExpression for an EXACT fixed-point result, reporting whether it wrote the batch. See ProjectColumn.VecDecimalEval for why the report is load-bearing.

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. NewWindow copies the spec slice. retypeValueColumns REWRITES OutputType in place, so sharing the caller's backing array lets one Window's correction land in another's specs — and the other then sees retypeValueColumns report "nothing changed", skips the w.groups rebuild, and keeps the spec COPIES groupWindowSpecs took at Init under the OLD type. The external path reads its types from those copies, so it would allocate a FLOAT64 output vector for an ARRAY value and raise the #361 guard on the write.

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.