scan

package
v0.16.0-correctness Latest Latest
Warning

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

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

Documentation

Overview

Package scan provides table scanning with 3-level predicate pushdown.

Index

Constants

View Source 
const (
	// DefaultDecodeAheadWorkers is deliberately modest: decode workers
	// multiply with the per-column errgroup inside ReadRowGroupNative
	// (min(#cols, GOMAXPROCS) each), and with concurrent fragments per
	// worker process. cpuToken integration (memo S3) replaces this with
	// budget-aware width.
	DefaultDecodeAheadWorkers = 4

	// DefaultDecodeAheadWindowBytes matches the scanPrefetch byte-window
	// order of magnitude; the morsel dispenser's budget bounds the next
	// stage downstream.
	DefaultDecodeAheadWindowBytes int64 = 256 << 20
)
View Source 
const (
	OpLike    = "like"
	OpNotLike = "not_like"
)

RowPred.Op values for pattern predicates.

Variables

View Source 
var DictPrune = optswitch.Register("dict-prune", "WADJET_DICT_PRUNE",
	"dictionary-probe row-group pruning for equality predicates on pure-dictionary chunks")

DictPrune gates dictionary-probe row-group pruning. The planner checks it when collecting equality conjuncts into EqProbes (plan.go), so with the switch off no probe is ever built. Kill switch: WADJET_DICT_PRUNE=0.

View Source 
var LengthsOnlyColumnDecodes atomic.Int64

LengthsOnlyColumnDecodes counts column chunks decoded as lengths. Tests assert engagement with it — a suite that never takes the path proves nothing about it.

View Source 
var RLERunPreds = optswitch.Register("rle-run-preds", "WADJET_RLE_RUN_PREDS",
	"run-granularity predicate evaluation over RLE dictionary-index pages in the scan filter")

RLERunPreds gates run-granularity predicate evaluation over dictionary-index pages. With it off, dictionary pages expand their index stream and the mask is applied per row, exactly as before. Kill switch: WADJET_RLE_RUN_PREDS=0.

View Source 
var StatsPrune = optswitch.Register("stats-prune", "WADJET_STATS_PRUNE",
	"min/max zonemap row-group pruning from static scan predicates")

StatsPrune gates static-predicate min/max (zonemap) row-group pruning at every consumption site: the planner's rgUnit build, readBatchDirect, and the Scanner decode path. Dynamic-filter pruning (join-build ranges, blooms) is separately gated. Kill switch: WADJET_STATS_PRUNE=0.

Functions

func CanBloomPruneRowGroup 

func CanBloomPruneRowGroup(bf *exec.BloomScanFilter, stats pqt.RowGroupStats) bool

CanBloomPruneRowGroup returns true when every integer value in the row group's min..max range is absent from the bloom — i.e., the row group cannot contain any rows matching the build side. Only applicable for single-column integer keys with a small (≤1024) value range; larger ranges return false (no pruning) to keep the check O(small).

func CanDictPruneRowGroup 

func CanDictPruneRowGroup(fr *pqt.FileReader, rgIdx int, probes []EqProbe) bool

CanDictPruneRowGroup reports whether ANY of the equality conjuncts is provably unsatisfiable in this row group via a pure-dictionary probe. A true return means the row group cannot produce a matching row.

func CanPruneRowGroup 

func CanPruneRowGroup(pred StatsPredicate, stats pqt.RowGroupStats) bool

CanPruneRowGroup returns true if the row group can be skipped based on min/max stats.

func CanRangePruneRowGroup 

func CanRangePruneRowGroup(ranges []exec.DynamicRange, stats pqt.RowGroupStats) bool

CanRangePruneRowGroup returns true when a row group can be skipped based on dynamic min/max ranges supplied by an upstream hash-join build (or a distributed dynamic filter). A row group prunes when its column range has no overlap with ANY of the supplied range filters.

Behavior matches the in-process planner's canRangePruneRowGroup (planner/physical/util.go) — moved here so the worker fragment-runner scan path can apply the same logic without a circular import.

func CompareValues 

func CompareValues(a, b any) int

CompareValues compares two typed values (int32, int64, float32, float64, string). Returns -1, 0, or 1. Exported for use by dynamic filter row-group pruning.

func DictPruneStatsSnapshot 

func DictPruneStatsSnapshot() (int64, int64)

DictPruneStatsSnapshot returns (pruned row groups, non-pruning probes).

func HasUnsupportedColumnarTypes 

func HasUnsupportedColumnarTypes(schema []pqt.Column) bool

HasUnsupportedColumnarTypes returns true if any column uses a type the native columnar reader cannot handle (Array, Map). TypeRow is supported: child fields are read as separate leaf column chunks. TypeDecimal is supported by the native reader.

func LengthsOnlyDecodeOn 

func LengthsOnlyDecodeOn() bool

LengthsOnlyDecodeOn reports whether lengths-only column decode is enabled.

func ReadFileBatches 

func ReadFileBatches(reader *pqt.Reader, schema []pqt.Column, selectedCols []string) ([]*batch.RecordBatch, error)

ReadFileBatches reads all row groups from a Parquet file into separate RecordBatches (one per row group). Supports column projection via selectedCols. Falls back to row-based reading for schemas containing Array or Map types.

func ReadFileBatchesNative 

func ReadFileBatchesNative(fr *pqt.FileReader, schema []pqt.Column, selectedCols []string) ([]*batch.RecordBatch, error)

ReadFileBatchesNative reads all row groups from a Parquet file using our custom FileReader (no parquet-go dependency). Returns one RecordBatch per row group for schemas without unsupported types.

func ReadFileBatchesNativeShard 

func ReadFileBatchesNativeShard(fr *pqt.FileReader, schema []pqt.Column, selectedCols []string, shardIdx, shardCount int) ([]*batch.RecordBatch, error)

ReadFileBatchesNativeShard reads only the row-group slice assigned to one shard. With shardCount=1 the behavior matches ReadFileBatchesNative.

Row-group ownership: shardIdx i reads row groups in [i*total/count, (i+1)*total/count). When total < count, the early shards each read one row group and later shards read nothing — a degenerate but correct split.

func ReadFileBatchesShard 

func ReadFileBatchesShard(reader *pqt.Reader, schema []pqt.Column, selectedCols []string, shardIdx, shardCount int) ([]*batch.RecordBatch, error)

ReadFileBatchesShard is like ReadFileBatches but reads only the row-group slice assigned to one shard of a multi-task scan. With shardCount=1 the behavior is identical to ReadFileBatches (whole file). With shardCount>1 the file's row groups are split evenly into shardCount disjoint ranges and shardIdx selects which range to read; the union over all shards equals the whole file.

This is the primitive that lets a single compacted parquet file (e.g. SF10 partsupp = 691 MB single file) fan out across N tasks without requiring the file to be physically chunked. The downstream broadcast-join chain then inherits parallelism through probe-split (which checks `len(probeFiles) >= 2`) because each shard task emits its own output file.

func ReadFileColumnar 

func ReadFileColumnar(reader *pqt.Reader, schema []pqt.Column) (*batch.RecordBatch, error)

ReadFileColumnar reads all row groups from a Parquet reader into a single RecordBatch. Used by the DML executor to read entire files for DELETE/UPDATE operations.

func ReadRowGroupNative 

func ReadRowGroupNative(fr *pqt.FileReader, rgIdx int, schema []pqt.Column, pool *batch.BatchPool) (*batch.RecordBatch, error)

ReadRowGroupNative reads a row group using our custom page reader, bypassing parquet-go entirely for the data path.

func ReadRowGroupNativeCached 

func ReadRowGroupNativeCached(fr *pqt.FileReader, rgIdx int, schema []pqt.Column, pool *batch.BatchPool, cache *DecodedChunkCache) (*batch.RecordBatch, error)

ReadRowGroupNativeCached is ReadRowGroupNative with an optional decoded chunk cache: plain leaf columns whose (identity, row group, column, type) is cached skip decompress+decode and copy from the cache; fresh decodes are offered back for admission. A nil cache (or a reader without a CacheIdentity) is byte-identical to ReadRowGroupNative.

func ReadRowGroupNativeSel 

func ReadRowGroupNativeSel(fr *pqt.FileReader, rgIdx int, schema []pqt.Column, pool *batch.BatchPool, sel []uint32) (*batch.RecordBatch, error)

ReadRowGroupNativeSel is ReadRowGroupNative under a partial scan-filter selection: eligible byte-array columns materialize only the rows in sel (ascending row indices; see sel_decode.go). A nil sel — or the sel-decode kill switch off — is identical to ReadRowGroupNative.

Selectivity gate (metal-validated 2026-08-17): the sel path copies per selected value, the full path bulk-copies the page. At sparse selections the skipped values dominate (ClickBench Q22 −30% hot at ~0.1%); past ~25% selected the per-value loop loses to the single memcpy (Q28 +8s, Q29 +1.9s on `Referer <> ”`, which selects most rows) — those decode in full.

func ReadRowGroupNativeShaped 

func ReadRowGroupNativeShaped(fr *pqt.FileReader, rgIdx int, schema []pqt.Column, pool *batch.BatchPool, sel []uint32, shapeOnly map[string]bool) (*batch.RecordBatch, error)

ReadRowGroupNativeShaped is ReadRowGroupNativeSel plus the set of columns (lowercased names) the planner proved are consumed for their SHAPE only. Those decode to per-row lengths with no value bytes materialized at all (lengths_decode.go). A nil/empty set — or the lengths-only kill switch off — is identical to ReadRowGroupNativeSel.

func RunPathStatsSnapshot 

func RunPathStatsSnapshot() (int64, int64)

RunPathStatsSnapshot returns (pages, rows) evaluated run-wise.

func ScanFilterStatsSnapshot 

func ScanFilterStatsSnapshot() (int64, int64)

ScanFilterStatsSnapshot returns (evaluated, skipped) row-group counts.

func SelDecodeOn 

func SelDecodeOn() bool

SelDecodeOn reports whether sel-aware materialization is enabled; the physical planner keys the LIKE selected-column pushdown gate off it (pattern columns in SELECT only pay a mask-eval + selected-only copy when this path is live).

func SetLengthsOnlyDecodeForTest 

func SetLengthsOnlyDecodeForTest(on bool) bool

SetLengthsOnlyDecodeForTest flips the kill switch and returns its previous value. Test-only: production reads the env var once at Register time.

func StorageClass 

func StorageClass(t pqt.TypeID) pqt.TypeID

StorageClass exposes storageClass to other packages. It is the file-vs- catalog compatibility relation the columnar decoder uses: when the file's type and the catalog's type share a storage class the page values are copied into the vector VERBATIM, and only a mismatch routes through copyNativeCoerced* (which converts values). Callers that want to reason about a parquet value without reading it — the planner's footer-statistics MIN/MAX path — need exactly this test.

Types

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 

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

OpenDecodeAheadIter constructs a decode-ahead iterator over the row-group range assigned to (shardIdx, shardCount), mirroring OpenRowGroupIter's contract (empty-shard sentinel, Array/Map rejection, projection via selectedCols).

func (*DecodeAheadIter) AssignmentDrained 

func (it *DecodeAheadIter) AssignmentDrained() bool

AssignmentDrained reports whether every row group has been assigned to a decode worker (not necessarily delivered) — the cross-file continuation's pre-open trigger: once true, idle workers exist or soon will, and the next file's head is the only work left to feed them.

func (*DecodeAheadIter) Close 

func (it *DecodeAheadIter) Close() error

Close stops assignment, wakes everything, and joins in-flight decodes. The caller may munmap the file bytes as soon as Close returns — workers never touch the FileReader after the join. Undelivered decoded batches are dropped (GC-eligible). Idempotent.

func (*DecodeAheadIter) DecodeSpans 

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

DecodeSpans returns the total wall time (ns) spent inside ReadRowGroupNative and the projected compressed bytes those decodes covered. Unlike StallDurations — which only measures parked waiters — this captures inline mmap fault time hidden inside token-holding decode spans: ns/byte materially above a page-cache-hot run's ratio means decode workers are faulting synchronously despite I/O-ahead.

func (*DecodeAheadIter) Next 

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

Next returns the next decoded row group in source order, or (nil, nil) when exhausted. Contract identical to RowGroupIter.Next.

func (*DecodeAheadIter) PruneStats 

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

PruneStats mirrors RowGroupIter.PruneStats.

func (*DecodeAheadIter) SetDynamicFilters 

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

SetDynamicFilters attaches dynamic-filter pushdowns. Safe mid-scan: decode workers re-read the filter set under the window lock at every group assignment, so a set that lands while the scan is running prunes every group not yet assigned (attach-on-arrival delivery). Groups already assigned or advised before the call decode/advise unfiltered — drop-only semantics, results identical. Multiple calls overwrite; callers pass the accumulated union. Mirrors RowGroupIter.SetDynamicFilters.

func (*DecodeAheadIter) StallDurations 

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

StallDurations returns the total blocked time (ns) behind each Stats counter, in the same order (window-full, pressure, token, ledger). Counts say how often a gate closed; these say how much wall it cost.

func (*DecodeAheadIter) Start 

func (it *DecodeAheadIter) Start()

Start spawns the decode workers immediately instead of on the first Next — the cross-file continuation pre-opens the next file's iterator and wants its head decoding while the current file's tail delivers. Dispatch-time filters should already be attached; late (attach-on- arrival) filters may still land afterward via SetDynamicFilters. Idempotent.

func (*DecodeAheadIter) Stats 

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

Stats returns the decode-ahead engagement counters (memo §5/§9 markers): row groups decoded, worker stalls on a full window, admissions refused under memory pressure, admissions deferred for lack of a cpu token, and admissions denied by the memory ledger (the group still decodes later — serially at worst, in every case).

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
}

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 

func NewDecodeWindow(bytes int64) *DecodeWindow

NewDecodeWindow returns a window bounding decoded-but-undelivered bytes across every iterator opened with it. bytes <= 0 selects DefaultDecodeAheadWindowBytes.

func NewDecodeWindowWithLedger 

func NewDecodeWindowWithLedger(bytes int64, ledger WindowLedger) *DecodeWindow

NewDecodeWindowWithLedger is NewDecodeWindow with a memory ledger attached: beyond the fixed byte ceiling, non-cursor admission must also clear ledger.Reserve, and every inflight byte is charged to the ledger for its parked lifetime. A nil ledger yields the fixed-window behavior. Callers passing a concrete pointer type must nil-check it themselves — a nil *T wrapped in the interface reads as non-nil here.

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 

func NewDecodedChunkCache(capBytes int64) *DecodedChunkCache

NewDecodedChunkCache returns a cache bounded to capBytes. capBytes <= 0 returns nil — a nil *DecodedChunkCache is valid and inert on every method.

func (*DecodedChunkCache) CapBytes 

func (c *DecodedChunkCache) CapBytes() int64

CapBytes returns the configured budget. Nil-safe.

func (*DecodedChunkCache) EstimateRelief 

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

EstimateRelief implements memory.AccountedOperator (pure read).

func (*DecodedChunkCache) Inspect 

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

Inspect implements memory.AccountedOperator.

func (*DecodedChunkCache) Offer 

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

Offer presents a freshly decoded chunk vector for admission. First touch registers a ghost; later touches admit only when there is free budget or the candidate's frequency STRICTLY beats the eviction victim's — the churn gate from the 2026-08-12 SF100 pair (doc §9.2): under a uniform flood ties go to the incumbent, so the resident set stabilizes and the wasted-clone admission storms cannot form. The clone runs outside the cache lock, and only after the admission decision — a rejected offer costs a map touch, not a memmove.

func (*DecodedChunkCache) SetPressureFunc 

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

SetPressureFunc wires the admission-pause pressure signal (worker: the heap-backpressure gauge OR the page-cache refault sensor). Call before the cache is shared with readers; nil leaves admission ungated. Nil-safe.

func (*DecodedChunkCache) ShedUnderPressure 

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

ShedUnderPressure evicts down to lowWater bytes and returns bytes freed. The pressure-yield valve (doc §9.3): the worker stats loop calls this while the heap-backpressure gauge or the page-cache refault sensor is active, because resident cache heap is exactly what those channels see as displacement — a cache must be the first thing to yield, before decode-ahead collapses and producers pause on its behalf. Evicted entries re-ghost with their frequency, so the hot set re-admits through the normal gate once pressure clears. Nil-safe.

func (*DecodedChunkCache) Size 

func (c *DecodedChunkCache) Size() int64

Size returns the cached bytes (the reservoir's live accessor). Nil-safe.

func (*DecodedChunkCache) SpillSome 

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

SpillSome implements memory.AccountedOperator: "spilling" a cache is eviction — entries re-decode from local compressed bytes on next miss.

func (*DecodedChunkCache) Stats 

func (c *DecodedChunkCache) Stats() DecodedChunkCacheStats

Stats returns a point-in-time counter snapshot. Nil-safe.

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

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

EvalRowGroupPreds evaluates the AND of preds over one row group and returns the matching selection. sel is only meaningful for FilterPartial and holds row indices in ascending order.

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 

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

OpenRowGroupIter constructs a streaming iterator over the row-group range assigned to (shardIdx, shardCount) of the given reader. With shardCount=1 the iterator covers the whole file. Returns ErrUnsupportedColumnar for schemas with Array/Map types (callers must use ReadFileBatchesShard).

func (*RowGroupIter) Close 

func (it *RowGroupIter) Close() error

Close marks the iterator as exhausted. Idempotent. No file handles are owned by the iterator (the caller's *pqt.Reader owns them), so Close is mostly a cancellation signal — subsequent Next calls return (nil, nil).

func (*RowGroupIter) Next 

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

Next returns the next decoded row group as a RecordBatch, or (nil, nil) when exhausted. The returned batch's lifetime is the caller's; Release() to a pool when done. Subsequent calls after exhaustion or Close return (nil, nil).

func (*RowGroupIter) PruneStats 

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

PruneStats returns counters for diagnostic logging: row groups skipped via bloom, via range, and actually read. Snapshot at any point.

func (*RowGroupIter) SetDecodedCache 

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

SetDecodedCache attaches the worker's decoded-chunk cache. Call before the first Next (the field is read without synchronization). nil = uncached.

func (*RowGroupIter) SetDynamicFilters 

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

SetDynamicFilters attaches dynamic-filter pushdowns to the iterator. Safe mid-scan from the goroutine calling Next: filters are consulted per group, so a set attached after some groups were read prunes every remaining group (attach-on-arrival delivery; drop-only semantics). Empty slices clear any existing filters. Multiple calls overwrite prior state — callers pass the accumulated union.

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 

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

NewScanner creates a new scanner for the given table.

func (*Scanner) Close 

func (s *Scanner) Close() error

func (*Scanner) Init 

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

func (*Scanner) Next 

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

func (*Scanner) Stats 

func (s *Scanner) Stats() ScanStats

Stats returns scan statistics after scanning.

func (*Scanner) WithColumns 

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

WithColumns sets the columns to select (projection pushdown).

func (*Scanner) WithPartitionFilter 

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

WithPartitionFilter sets the partition filter (Level 1: partition pruning).

func (*Scanner) WithRowFilter 

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

WithRowFilter sets the row-level filter predicate (Level 3: row-level evaluation).

func (*Scanner) WithStatsPredicates 

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

WithStatsPredicates sets predicates for Level 2 row-group pruning.

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.

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.

Source Files

View all Source files

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.