scan

type BackingPool ¶

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

BackingPool is one scan source's free list of decoded row-group backings.

Ownership rule ¶

A backing may be handed to a later decode only when BOTH hold:

1. RELEASED — the consumer side said it is finished. That is the morsel dispenser's retire callback for the parent batch (fired once every zero-copy view minted over it has retired, which is after the whole op chain AND the sink consume), or the serial fragment path's return from driver.push. This is what a claim check alone cannot supply: the decode-ahead ring (ADR-0015) decodes group N+1 while group N is being consumed, k morsel consumers read one parent concurrently, and HashJoinProbe.emitViewOutput's view columns read our columns through Vector.Base after Execute returned — none of those readers claim.

2. UNCLAIMED — nobody kept it: neither RecordBatch.Detach on the batch we emitted nor Vector.Claim on any column, including a claim that arrived transitively through a derived batch (ColumnPrune, set-op emit, selView) or through Vector.Base from a view minted downstream over one of our columns. This is what a release signal alone cannot supply: Sort, Window, the hash-join build, SortMergeJoin and the spillable collector all retain past retire, and all of them Detach (ADR-0016).

The release is the liveness signal; the claim is the retention veto. The backing is surrendered whole or not at all — one claimed column surrenders the batch, permanently, because every column is reachable from the batch a consumer kept.

Identity, without a reference ¶

The pool only ever takes back a batch it minted, at the generation it minted it: a WSHF shuffle chunk, a row-based fallback batch, a batch from another pool or a batch already recycled is ignored, so no second owner can be created for storage someone else recycles.

That identity is a batch.MintStamp the pool writes ON the batch, never a registry of outstanding pointers. A registry is a strong reference: the batches this pool mints are whole decoded row groups (~280 MB each at SF100 lineitem widths), and most of the sources that own a pool have consumers with NO release edge at all (the shuffle task's plain Next loop, the hash-join build source, planner.StreamingSources) — every one of those would pin its entire live set for the source's lifetime, invisibly to the memory ledger, and any cap on the registry's size would silently turn reuse off rather than bound the damage. The stamp inverts the direction: the pool holds only its own idle list, a dropped batch (a bloom-filtered-to-nothing row group, a ring discard at a cross-file boundary) is plain garbage the moment the pipeline lets go of it, and a source whose consumer never releases costs exactly one stamp per decode.

A pool is created only where a RELEASE EDGE exists — see the caller side in internal/worker (batchRecyclerOf arms it). A pool without one could never take a backing back, so it would be pure overhead.

See docs/design/scan-output-backing-reuse.md for the full statement and the preconditions it rests on.

func NewBackingPool(opts BackingPoolOpts) *BackingPool

func (p *BackingPool) Drop()

func (p *BackingPool) Recycle(b *batch.RecordBatch, mint batch.MintStamp)

func (p *BackingPool) Stats() BackingPoolStats

type BackingPoolOpts ¶

type BackingPoolOpts struct {
	// MaxIdle bounds the idle free list by count.
	MaxIdle int
	// MaxIdleBytes bounds the idle free list by MemBytes sum. One backing is
	// always keepable regardless of size, mirroring the decode ring's
	// "the delivery-cursor group is always admitted" escape: without it a
	// single row group larger than the cap disables the mechanism on exactly
	// the table that motivates it (SF100 lineitem, ~280 MB per group).
	MaxIdleBytes int64
}

BackingPoolOpts sizes a BackingPool. Zero fields take the defaults.

type BackingPoolStats ¶

type BackingPoolStats struct {
	Hits      int64 // decodes that reused a backing
	Misses    int64 // decodes that had to mint one
	Claimed   int64 // releases refused because a consumer had claimed the batch
	IdleBytes int64
	// Held is the number of batches this pool holds a Go reference to. It is
	// exactly the idle free list: the pool never references a backing it has
	// handed out, so a source whose consumer has no release edge holds nothing
	// here however many row groups it decodes.
	Held int
}

BackingPoolStats is one pool's engagement counters.

type DecodeAdmission ¶

type DecodeAdmission interface {
	DecodeStallBegin()
	DecodeStallEnd()
}

DecodeAdmission is the optional half of TokenPool: a pool that implements it is told when a decode worker parks on a failed acquisition and when it stops waiting, so the pool can hold a floor of tokens back from its own consumer queue instead of granting every release around a decoder that is the only thing able to refill that queue. Without it decode is a non-blocking second-class caller whose demand is invisible — the closed loop the worker pool's admission policy documents (internal/worker/cpu_tokens.go). Begin/End are always paired and must be safe for concurrent use; they are called with the window lock HELD, so an implementation must never call back into the iterator.

type DecodeAheadIter ¶

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

DecodeAheadIter is the parallel sibling of RowGroupIter (docs/design/scan-decode-pipelining.md): k decode workers pull row-group indices in file order, each runs ReadRowGroupNative for its group, and results deliver to the consumer strictly in source order. The consumer-facing contract is identical to RowGroupIter — same batches in the same order, same error surfaced at the same position, same prune behavior — only the decode of group N+1..N+w overlaps the consumption of group N instead of waiting for it.

Memory is bounded by WindowBytes of decoded-but-undelivered batches, estimated per group from the projected columns' TotalUncompressedSize metadata (estimation error is bounded by one group per worker). The group at the delivery cursor is always admitted regardless of the window so a single oversized row group cannot deadlock the pipeline.

Concurrency safety rests on ReadRowGroupNative's documented contract: FileReader is read-only after construction and every ColumnPages call allocates a fresh ColumnPageReader, so concurrent decodes of distinct row groups never share mutable state (columnar_native.go:124-126).

Lifecycle: workers start on the FIRST Next() call, not at Open — filters attach after Open on the worker scan path (and may keep arriving mid-scan; assignments re-read them per group). Close() stops assignment and JOINS in-flight decodes before returning: the caller munmaps the file bytes right after Close, so no decode may touch the underlying slice once Close returns.

func OpenDecodeAheadIter(reader *pqt.Reader, schema []pqt.Column, selectedCols []string, shardIdx, shardCount int, opts DecodeAheadOpts) (*DecodeAheadIter, error)

func (it *DecodeAheadIter) AssignmentDrained() bool

func (it *DecodeAheadIter) Close() error

func (it *DecodeAheadIter) DecodeSpans() (ns, bytes int64)

func (it *DecodeAheadIter) Next() (*batch.RecordBatch, error)

func (it *DecodeAheadIter) PruneStats() (bloom, rangeP, read int)

func (it *DecodeAheadIter) RowOffset() int64

func (it *DecodeAheadIter) SetDynamicFilters(ranges []exec.DynamicRange, blooms []*exec.BloomScanFilter)

func (it *DecodeAheadIter) StallDurations() (windowFullNs, pressureNs, tokenNs, ledgerNs int64)

func (it *DecodeAheadIter) Start()

func (it *DecodeAheadIter) Stats() (groupsRead, windowFullStalls, pressureStalls, tokenStalls, ledgerStalls int64)

type DecodeAheadOpts ¶

type DecodeAheadOpts struct {
	// Workers is the decode worker count. <= 0 selects
	// DefaultDecodeAheadWorkers capped at GOMAXPROCS.
	Workers int
	// WindowBytes bounds decoded-but-undelivered batch bytes. <= 0
	// selects DefaultDecodeAheadWindowBytes. Ignored when Window is set.
	WindowBytes int64
	// Window shares an existing byte window (and its lock) with another
	// iterator — the cross-file continuation shape: the tail of file F
	// and the head of file F+1 draw from one budget.
	Window *DecodeWindow
	// Pressure, when set, is consulted before admitting any group beyond
	// the delivery cursor — the memory-pressure collapse hook. Must be
	// safe to call from multiple goroutines.
	//
	// While it reports true, admission is OCCUPANCY-FLOORED by default:
	// one non-cursor group may be in flight or parked (a 2-deep
	// pipeline), further admission waits. The 2026-07-18 SF100 sensor
	// A/B (memo §9.5) split the pressure regime in two: when decode
	// outruns the consumer the window fills and its held bytes displace
	// page cache (collapse is right — Q06 +46 % without it), but on
	// producer-bound repartition stages the window sits EMPTY and
	// cursor-only collapse is pure serialization with nothing to shed
	// (Q05 −12.7 % when spared). One group ahead holds ~one group-est of
	// bytes — nothing to displace with — while keeping the producer
	// pipelined.
	Pressure func() bool
	// PressureStrict restores cursor-only collapse under Pressure (no
	// group ahead at all). Set on edge-class envelopes (< 2 GiB
	// GOMEMLIMIT), where the capped repro measured even one extra
	// in-flight group as harmful (32 MiB window arm +37 % vs cursor-only
	// winning outright).
	PressureStrict bool
	// Tokens, when set, budgets decode CPU per ROW GROUP: a worker
	// acquires one token at admission and releases it when the decoded
	// group parks, so a worker stalled on the window (or between groups)
	// holds nothing. The delivery-cursor group is token-exempt — serial
	// progress is always allowed, mirroring the morsel "first consumer is
	// free" rule. The 2026-07-16 SF100 pair convicted the previous
	// source-lifetime acquisition: decode workers sat window-stalled
	// holding tokens (~35k stalls/40k groups), starving concurrent join
	// fragments' morsel width (Q20 +54%, Q05 +17%) while utilization
	// stayed flat.
	Tokens TokenPool
	// Advise, when set, receives the file-relative byte range of each
	// projected column chunk shortly BEFORE its row group is decoded —
	// the I/O-ahead seam (docs/design/rowgroup-readahead.md). The worker
	// wires an madvise(MADV_WILLNEED) closure over the scan mmap so a
	// steady-state re-read faults asynchronously via kernel readahead
	// instead of synchronously under a held CPU token. Ranges for group
	// N+workers are issued as group N is assigned, so the advice leads
	// decode by roughly one full assignment wave. Must be safe for
	// concurrent use; calls stop before Close returns (decode workers
	// are joined), so an mmap-backed closure never outlives its munmap.
	Advise func(off, n int64)
	// Cache, when set, consults/feeds the worker's decoded-chunk cache
	// inside ReadRowGroupNativeCached (docs/design/decoded-rowgroup-cache.md).
	// Inert unless the reader carries a CacheIdentity. nil = uncached.
	Cache *DecodedChunkCache
	// Backing, when set, is the SCAN SOURCE's row-group backing pool: a
	// decode writes into storage a previous group used once the consumer
	// released it and nobody claimed it (BackingPool's ownership rule,
	// docs/design/scan-output-backing-reuse.md). It belongs to the source,
	// not the iterator, so it survives the cross-file continuation — a
	// backing delivered from file F may be released after the source has
	// moved to F+1. nil = every group allocates fresh.
	Backing *BackingPool
}

DecodeAheadOpts sizes a DecodeAheadIter.

type DecodeWindow ¶

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

DecodeWindow is a byte budget shared by one or more DecodeAheadIters. Its mutex doubles as the owning iterators' state lock.

func NewDecodeWindow(bytes int64) *DecodeWindow

func NewDecodeWindowWithLedger(bytes int64, ledger WindowLedger) *DecodeWindow

type DecodedChunkCache ¶

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

DecodedChunkCache is a worker-lifetime cache of decoded parquet column chunks: one entry per (object identity, row group, leaf column, catalog type) holding a cache-owned *batch.Vector clone. It attacks the zstd decompress + decode-kernel bill (~24% + ~7% of SF100 worker CPU) for re-reads of the same immutable base-table bytes — cross-query within a run and across benchmark runs. See docs/design/decoded-rowgroup-cache.md.

Correctness model (v1, copy discipline): consumers NEVER share storage with the cache. A hit copies the cached vector into the caller's batch slot; an admit clones the freshly decoded vector into cache-owned storage. Entry vectors are immutable once inserted, so Get may return the entry pointer and the caller copies outside the cache lock (an eviction during the copy is safe — the GC keeps the clone alive).

Ledger model (ADR-0006): the cache OWNS its bytes once, surfaced through Size for a hard system reservoir (memory.NewReservoirFunc). Consumers' copies are ordinary scan output charged exactly as today. The cache also implements memory.AccountedOperator so RequestRelief can shed it — eviction is the cheapest relief in the process — before any operator pays a real spill.

Eviction is segmented LRU (probation/protected) with second-touch ghost admission: a key's first decode registers a ghost; the clone is stored on the second decode; hits promote probation entries to protected. Sequential scan floods (a cold first pass over a table) fill probation without displacing the protected hot set.

func NewDecodedChunkCache(capBytes int64) *DecodedChunkCache

func (c *DecodedChunkCache) CapBytes() int64

func (c *DecodedChunkCache) EstimateRelief(target int64) int64

func (c *DecodedChunkCache) Inspect() memory.OperatorFootprint

func (c *DecodedChunkCache) Offer(key decodedChunkKey, vec *batch.Vector, numRows int)

func (c *DecodedChunkCache) SetPressureFunc(f func() bool)

func (c *DecodedChunkCache) ShedUnderPressure(lowWater int64) int64

func (c *DecodedChunkCache) Size() int64

func (c *DecodedChunkCache) SpillSome(target int64) (int64, error)

func (c *DecodedChunkCache) Stats() DecodedChunkCacheStats

type DecodedChunkCacheStats ¶

type DecodedChunkCacheStats struct {
	Hits, Misses, HitBytes       int64
	Admitted, GhostRegistered    int64
	Evictions, ReliefBytes       int64
	RejectedTooLarge, CloneSkips int64
	FreqRejected, PressurePaused int64
	SizeBytes, CapBytes          int64
	Entries                      int
}

DecodedChunkCacheStats is the counter snapshot for the worker stats ticker.

type DeleteSet ¶ added in v0.18.2

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

Merge-on-read delete markers, and the one representation every read path shares.

A DELETE does not rewrite parquet. It records, per data file, the FILE-ABSOLUTE 0-based row indices of the rows it removed (catalog.DeleteMarker), and every scan of that file must skip them until compaction folds them in. "File-absolute" means counted over the file's row groups in order from row 0 — NOT relative to a row group, a shard, or whatever slice of the file a particular reader happens to be assigned. The single-process scanner tracks that with a per-row-group prefix sum (physical.rgUnit.rgRowOffset); the distributed scan source gets the same number from its row-group iterator (RowGroupIter.RowOffset).

DeleteSet is the runtime form: the sorted row indices coalesced into disjoint runs, which is both the compact wire encoding (EncodeDeleteRuns) and a fast membership test. A map[int64]bool costs ~50 B/row and is the wrong shape for a DELETE that removed a contiguous range — the common case, since a WHERE over a clustered column marks runs, not confetti.

func DecodeDeleteSet(b []byte) (*DeleteSet, error)

func NewDeleteSet(rows []int64) *DeleteSet

func (d *DeleteSet) Contains(row int64) bool

func (d *DeleteSet) Empty() bool

func (d *DeleteSet) Encode() []byte

func (d *DeleteSet) Overlaps(offset, count int64) bool

func (d *DeleteSet) Rows() int64

func (d *DeleteSet) Runs() int

type EqProbe ¶

type EqProbe struct {
	ColName string
	Value   any
}

EqProbe is one equality conjunct to test against a row group.

type FilterDecision ¶

type FilterDecision int

FilterDecision summarizes a row group evaluation.

const (
	FilterNone    FilterDecision = iota // no row matches: skip the row group
	FilterAll                           // every row matches: no selection needed
	FilterPartial                       // some rows match: apply Sel
)

func EvalRowGroupPreds(fr *pqt.FileReader, rgIdx int, preds []RowPred, numRows int) ([]uint32, FilterDecision, error)

type PartitionFilter ¶

type PartitionFilter map[string]string

PartitionFilter filters partitions based on partition key values.

type RowGroupIter ¶

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

RowGroupIter yields one RecordBatch per row group on demand, without pre-decoding the rest of the file. Use this in long-running scan pipelines (worker fragment runners) where holding every row group of a file in memory would blow the per-task working set.

At SF100 each lineitem file has ~10 row groups × ~28 MB decoded each, so eager decode (ReadFileBatchesShard) costs ~280 MB live per file, times 2–4 files prefetched, times 3–4 concurrent tasks = multi-GB transient that the GC can't reclaim until the consumer (HashAggregate) drains. Streaming one row group at a time bounds the scan-side live memory to one decoded RG per scan source plus whatever is in flight downstream — typically <300 MB instead of multi-GB.

Lifecycle:

it, err := OpenRowGroupIter(reader, schema, selectedCols, shardIdx, shardCount)
if err != nil { ... }
defer it.Close()
for {
    b, err := it.Next(ctx)
    if err != nil || b == nil { break }
    // consume b, then b.Release() when done
}

The iterator does NOT support schemas containing Array/Map types — the existing row-based fallback in readFileBatchesViaRows decodes the whole file in one shot and predates row-group sharding. Callers must check HasUnsupportedColumnarTypes(schema) and use ReadFileBatchesShard for those types. The slice path stays available; this iterator is a parallel fast lane for the common case.

func OpenRowGroupIter(reader *pqt.Reader, schema []pqt.Column, selectedCols []string, shardIdx, shardCount int) (*RowGroupIter, error)

func (it *RowGroupIter) Close() error

func (it *RowGroupIter) Next() (*batch.RecordBatch, error)

func (it *RowGroupIter) PruneStats() (bloom, rangeP, read int)

func (it *RowGroupIter) RowOffset() int64

func (it *RowGroupIter) SetBackingPool(p *BackingPool)

func (it *RowGroupIter) SetDecodedCache(c *DecodedChunkCache)

func (it *RowGroupIter) SetDynamicFilters(ranges []exec.DynamicRange, blooms []*exec.BloomScanFilter)

type RowPred ¶

type RowPred struct {
	Col   string
	Op    string // =, !=, <, <=, >, >=
	Value any    // int64, float64, or string (planner-normalized)
}

RowPred is one pushed conjunct.

type ScanStats ¶

type ScanStats struct {
	TotalPartitions  int
	PrunedPartitions int
	TotalFiles       int
	PrunedFiles      int
	TotalRowGroups   int
	PrunedRowGroups  int
	RowsScanned      int64
}

ScanStats tracks scan statistics.

type Scanner ¶

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

Scanner performs table scans with 3-level predicate pushdown.

func NewScanner(cat *catalog.Catalog, tableName string) *Scanner

func (s *Scanner) Close() error

func (s *Scanner) Init(ctx context.Context) error

func (s *Scanner) Next(ctx context.Context) (*batch.RecordBatch, error)

func (s *Scanner) Stats() ScanStats

func (s *Scanner) WithColumns(cols []string) *Scanner

func (s *Scanner) WithPartitionFilter(filter PartitionFilter) *Scanner

func (s *Scanner) WithRowFilter(pred exec.Predicate) *Scanner

func (s *Scanner) WithStatsPredicates(preds []StatsPredicate) *Scanner

type StatsPredicate ¶

type StatsPredicate struct {
	Column string
	Op     exec.CompareOp
	Value  any
}

StatsPredicate evaluates whether a row group can be skipped based on column stats.

Value must already be in the STATS domain — the representation parquet.RowGroupStats decodes the footer's bounds into, which for several types is not the representation a SQL literal arrives in. This layer compares two `any` values by their Go kind and cannot tell an unscaled DECIMAL bound from a scaled literal, or sixteen raw address bytes from an address in text; both pairs land in the same kind and get compared as if they agreed, which is #442. kernel.StatsDomainValue is the conversion, and the producer WITHHOLDS a predicate it cannot convert.

type TokenPool ¶

type TokenPool interface {
	TryAcquire(n int) int
	Release(n int)
}

TokenPool is the compute-budget seam shared with the caller's pool (the worker's cpuTokens). Both methods must be safe for concurrent use; TryAcquire must not block.

type WindowLedger ¶

type WindowLedger interface {
	Reserve(n int64) error
	ForceReserve(n int64)
	Release(n int64)
}

WindowLedger is the memory-budget seam shared with the caller's task ledger (the worker's shared pool tracker — *memory.Tracker satisfies it directly). Decoded-but-undelivered window bytes are charged here so they are visible to spill decisions and so admission collapses toward the cursor-only serial floor as budget headroom vanishes — the memo §9 fix: under memory pressure the window's held batches displace page cache and GC headroom that the Go-heap pressure hook cannot see, so the byte cost must ride the same ledger as every other operator.

Reserve returns a non-nil error to deny (nothing retained on denial); ForceReserve charges unconditionally — used for the delivery-cursor group, which is always admitted but whose bytes are still real. All methods must be safe for concurrent use and must not block.