physical

type AggSpec struct {
	Func      string
	InputCol  string
	OutputCol string
	// InputExpr is the SQL text of a derived input expression, e.g.
	// "l_extendedprice * (1 - l_discount)". Empty when InputCol is a
	// bare column reference. Distributed workers compile this into a
	// Project operator before the aggregate so HashAggregate sees a
	// column whose name matches InputCol.
	InputExpr string
}

type AggregateShuffleCandidate struct {
	JoinStageID      string   // outer join whose build side is a derived aggregate
	AggregateStageID string   // aggregate stage directly feeding the join (through any shuffles)
	InputScanID      string   // base-table scan feeding the aggregate's input
	InputScanAlias   string   // scan alias (e.g. "lineitem:1" for Q17's inner scan)
	InputScanBytes   int64    // EstimatedBytes of the aggregate's input scan
	GroupByKeys      []string // the aggregate's GROUP BY columns (= partition keys)
	JoinBuildKeys    []string // the outer join's keys on this side (must be ⊆ GroupByKeys)
	JoinProbeKeys    []string // the outer join's keys on the probe side
}

type AggregateShuffleDiag struct {
	Candidate AggregateShuffleCandidate
	Reason    AggregateShuffleRejectReason
	// ObservedScanBytes holds the input scan size of the closest-matching
	// rejected candidate (populated when we got past followToScan). Useful
	// for tuning the threshold on real data.
	ObservedScanBytes int64
	// JoinStageID / InputScanAlias of the closest-matching rejected join, when
	// available. Empty for NoJoin / BuildNotAggregate paths.
	JoinStageID    string
	InputScanAlias string
}

type AggregateShuffleRejectReason int

type ChainedJoinSpec struct {
	JoinType        string
	JoinLeftKeys    []string
	JoinRightKeys   []string
	BuildDepStage   string // stage providing build-side data
	BuildTableAlias string
	BuildColOrigins map[string]string
	JoinFilter      string
	// FilterExprs are the absorbed stage's residual post-join filters —
	// emitted as an OpFilter immediately after this chained probe.
	FilterExprs []string
	// BuildFilterExprs filter the build input rows before hash-table
	// construction (exchange-subsume flag filters on the absorbed stage).
	BuildFilterExprs    []string
	QualifyAllBuildCols bool
	// Columns is the absorbed stage's output projection; applied as the
	// chained probe's OutputFilter so the fused stage emits exactly what
	// the absorbed stage emitted.
	Columns []string
	// Partitioned marks a hash-partitioned 1:1 build input (the absorbed
	// stage was a hash_join): task i reads build partition i. False means
	// a replicated broadcast build read whole by every task.
	Partitioned bool
}

type ComputedCol struct {
	Name string
	Expr string
}

type DistKind int

type Distribution struct {
	Kind  DistKind
	Keys  []string // for DistHashPartitioned
	Count int      // for DistHashPartitioned
}

type DynamicFilterConsume struct {
	FilterID      string
	SourceStageID string
	TargetColumn  string
	KeyType       string
	// AttachOnArrival marks this consume as non-blocking: the stat-dep edge
	// is removed, the consumer dispatches immediately, and its tasks install
	// the bloom mid-scan when the emitter's merged artifact lands at the
	// deterministic staged key. Set only by applyAttachOnArrival under its
	// structural rules; drop-only bloom semantics keep results identical.
	AttachOnArrival bool
}

type DynamicFilterEmit struct {
	FilterID  string
	KeyColumn string
	KeyType   string // "int32" | "int64" | "date"
	BloomBits int    // total bloom-bitset size; identical across all tasks so union = bitwise OR
	// AtOutput accumulates over the stage's OUTPUT stream (pre-sink) rather
	// than the scan source — required when the emitting stage is a join or
	// filtered scan whose output, not input, defines the key set
	// (markSemiAntiBuildFilters).
	AtOutput bool
	// LateAttach forces the coordinator to stage the merged filter to its
	// deterministic S3 key regardless of inline size — an attach-on-arrival
	// consumer polls that key, so it must exist even for tiny blooms
	// (applyAttachOnArrival; docs/design/attach-on-arrival-dynamic-filters.md).
	LateAttach bool
	// GuardConsumes lists the FilterIDs of this stage's OWN attach-mode
	// consumes that must retro-filter this emit's buffered head rows at
	// finalize (guarded re-emit — applyAttachOnArrival rule-1 relaxation).
	// The worker's emit op buffers (emit-key, guard-column) pairs for rows
	// scanned before those blooms install and drops non-matching pairs
	// before the partial uploads, keeping the emitted bloom exactly as
	// tight as under the start barrier.
	GuardConsumes []string
	// InFlow marks an emitter whose tasks must ride NORMAL scheduling
	// instead of the priority lane. The lane's contract is planner-bounded
	// tiny tasks (extra slots above MaxConcurrent are memory-safe only
	// because dimension scans are tiny) and its purpose is overtaking bulk
	// work that is ALREADY consuming the filter attach-mode. An in-flow
	// cascade mid (e.g. 15M-row customer) violates the first and doesn't
	// need the second: its consumer is WAIT-blocked on a stat-dep, so
	// ordinary slots serve it correctly (docs/design/dimension-cascade.md
	// §In-flow mid emitters).
	InFlow bool
}

type ExchangeStage struct {
	Keys       []string      // Repartition only
	Count      int           // Repartition only
	Ordering   []SortKeySpec // Gather only (optional sort-merge gather)
	BuildAlias string        // Repartition, Replicate
	ProbeAlias string        // Repartition, Replicate
	BuildBytes int64         // Repartition (for logging / threshold checks)
	// ComputedCols are expression columns APPENDED to the shuffle payload
	// (after the projected scan columns). Set by dedupeSubsumedScanExchanges
	// so one raw exchange can serve a dropped filtered sibling: the
	// sibling's scan filter ships as a cheap computed flag (1 byte/row)
	// instead of a second full scan+shuffle of the table. Workers evaluate
	// Expr per batch and append the result under Name.
	ComputedCols []ComputedCol
	// ExtraReadCols are columns the shuffle's source scan must READ so the
	// ComputedCols expressions can evaluate, but which are NOT part of the
	// shipped payload — the worker drops them after computing the flags.
	ExtraReadCols []string
	// PartialAggGroupBy/PartialAggSpecs mark this Repartition for
	// sender-side partial aggregation (markExchangePartialAgg): the
	// shuffle task pre-combines rows on PartialAggGroupBy, shipping
	// name-preserving SUM/MIN/MAX partials (OutputCol == InputCol)
	// instead of raw rows. Set only when every consumer is proven
	// merge-compatible; empty means ship raw.
	PartialAggGroupBy []string
	PartialAggSpecs   []AggSpec
}

type FusedJoinSpec struct {
	JoinType        string
	JoinLeftKeys    []string
	JoinRightKeys   []string
	BuildDepStage   string // stage providing build-side data
	BuildTableAlias string
	BuildColOrigins map[string]string // bare build col → owning scan alias (multi-table builds only)
	JoinFilter      string
	FilterExprs     []string
}

type OutputRename struct {
	From string
	To   string
	Expr plansql.Node
}

type PhysicalPlan struct {
	Pipeline *exec.Pipeline
	Stages   []Stage // for distributed execution
	Cleanup  func()  // optional: called after pipeline finishes to clean up spill files
}

type Planner struct {
	MemoryBudget int64  // per-query memory budget in bytes (0 = unlimited)
	SpillDir     string // directory for spill files (empty = os temp dir)

	// SharedTracker / SharedSpillMgr (if set) are used in place of per-query
	// Tracker+SpillManager creation. Workers set these to point at the
	// executor-level pool so concurrent tasks on the same worker compete for
	// ONE budget and spill cooperatively under pool pressure, matching the
	// Trino/Spark unified memory manager model. When nil, getSpillManager()
	// falls back to creating a per-query pool as before.
	SharedTracker  *memory.Tracker
	SharedSpillMgr *memory.SpillManager
	QueryLimits    *config.QueryLimits // cost-based query guard (nil = no limits)

	WorkerCount int // number of distributed workers (for shuffle partitioning)

	// BroadcastBytesThreshold is the maximum estimated build-side size for
	// a join to be planned as broadcast_join. Builds above this become
	// hash_join (hash-shuffle), eliminating the N× build-cache duplication
	// every worker pays under broadcast.
	//
	// Zero = use absolute default (100 MB), preserving legacy behavior for
	// embedded callers that don't set this. Distributed callers should set
	// this from per-worker pool budget — e.g., budget * 0.3 capped at
	// 200 MB — so the broadcast/shuffle decision adapts to cluster memory
	// instead of trusting an absolute constant. Negative = never broadcast
	// (force every join to hash_join).
	//
	// Architectural rule the threshold encodes: broadcast is sound when
	// every worker can comfortably hold the full build state in memory
	// alongside concurrent hash tables. As cluster width or per-worker
	// budget shrinks, the threshold shrinks too — without changing the
	// planner's join-type logic.
	BroadcastBytesThreshold int64

	// SortMergeJoinBytes gates the sort-merge join path for big-vs-big inner
	// equi-joins (docs/design/sort-merge-join.md). A join takes SMJ only when
	// BOTH sides' estimated post-selectivity bytes reach this threshold —
	// small builds keep the strictly-better hash/broadcast paths. Zero (the
	// shipped default) disables SMJ entirely; the planner behaves exactly as
	// before. Distributed callers should derive it from per-worker pool
	// budget (the broadcast-threshold pattern) so "too big to sit resident"
	// tracks cluster memory.
	SortMergeJoinBytes int64

	// LateMaterialization emits inner/left hash-join output as view
	// (dictionary) columns over the probe input and build batches; the
	// gather is deferred to the first consumer needing owned storage
	// (docs/design/late-materialization.md). Off by default.
	LateMaterialization bool

	// DynamicFiltersEnabled gates the Trino-style dynamic-filter planner
	// pass. When true, applyDynamicFilters annotates eligible hash_join
	// build/probe leaf scans with Emit/Consume specs and adds the stat-dep
	// edge from build-scan to probe-scan. Off by default for v1 rollout;
	// distributed callers flip this on after the local-harness gate passes.
	DynamicFiltersEnabled bool

	// MaterializedInputs holds pre-scanned data for scan-split pipeline mode.
	// When populated, buildScan uses these batches instead of reading from the
	// object store, allowing parallel scan I/O with single-worker compute.
	// Keyed by scan alias: "table" or "table:N" for self-joins.
	MaterializedInputs map[string][]*batch.RecordBatch

	// StreamingSources holds lazy sources for scan-split pipeline mode.
	// Unlike MaterializedInputs, these yield batches on demand without
	// materializing all data upfront. Checked before MaterializedInputs.
	// Keyed by scan alias: "table" or "table:N" for self-joins.
	StreamingSources map[string]exec.Source

	// ScanFileFilter restricts which files each scan alias reads. Used in
	// probe-split pipeline mode where the probe table is partitioned across
	// workers while build tables read all files. Keyed by scan alias.
	ScanFileFilter map[string][]string
	// contains filtered or unexported fields
}

type PreComputedAggregateMeta struct {
	InputTable  string
	GroupByCols []string
	AggSpecs    []AggSpec
	CacheFiles  []string
}

type ProjectExprSpec struct {
	Expr string
	Name string
	Type parquet.TypeID
}

type QueryCost struct {
	TotalBytes int64
	TotalRows  int64
	TotalFiles int
	HasFilter  bool
	HasLimit   bool
}

type RecordBatch = batch.RecordBatch

type RequiredDistribution struct {
	Kind  RequiredKind
	Keys  []string
	Count int
}

type RequiredKind int

type ShuffleCandidate struct {
	JoinStageID string   // the join stage to be served by shuffled inputs
	BuildAlias  string   // which scan stage produces the build side
	ProbeAlias  string   // which scan stage produces the probe side (the largest scan)
	BuildKeys   []string // build-side join keys (the join's JoinRightKeys)
	ProbeKeys   []string // probe-side join keys (the join's JoinLeftKeys)
	JoinKeys    []string // canonical (build-side) join key names for partitioning
	BuildBytes  int64    // EstimatedBytes of the build scan (for logging)
}

type SortKeySpec struct {
	Column    string
	Desc      bool
	NullsLast bool
}

type Stage struct {
	ID           string
	Type         string // see exchange.go constants (scan, aggregate, sort, hash_join, broadcast_join, window, pipeline, exchange-repartition, exchange-replicate, exchange-gather)
	ClusterID    string // target cluster for routing ("" = local/coordinator's cluster)
	Dependencies []string
	Tasks        int

	// Scan metadata
	TableName string
	ScanAlias string // unique scan identity: "table" or "table:N" for Nth duplicate
	Columns   []string
	// OutputColumns, when non-empty, narrows the stage's EMITTED columns
	// to this set (worker inserts a zero-copy ColumnPrune before the
	// sink). Columns stays the READ set — a scan must read its pushed
	// filter columns but must not ship them: Q13's orders scan read
	// o_comment for the NOT LIKE and then materialized+shuffled it,
	// 68.8 B/row where 16 were consumed (~16 GB excess at SF100).
	// Set by pruneScanOutputColumns from consumer declarations.
	OutputColumns   []string
	PartitionFilter map[string]string
	ScanFiles       []string // files to distribute across scan tasks
	// ScanFileSizes aligns 1:1 with ScanFiles (catalog SizeBytes). Feeds
	// byte-balanced affinity fan-outs (coordinator scan_affinity.go);
	// empty/misaligned degrades to count-based splitting.
	ScanFileSizes []int64
	FilterExprs   []string // SQL filter expressions pushed down to scan

	// Aggregate metadata
	GroupByCols []string
	AggSpecs    []AggSpec
	// GroupByAll marks a keys-only hash aggregate over EVERY input column —
	// the DISTINCT shape. The key set is resolved at runtime from the input
	// schema (no plan-time column list), matching exec.HashAggregate.GroupByAll
	// and the single-process buildDistinct path.
	GroupByAll bool

	// Sort metadata
	SortKeys []SortKeySpec
	Limit    int
	// SortShardLocal marks a grouped final_aggregate whose SortKeys/Limit
	// are SHARD-LOCAL: the stage fans out across disjoint group-key shards
	// (each computes exact aggregates for its groups, then sorts and
	// applies Limit locally) and a surviving downstream Singleton sort
	// stage merges the N sorted ≤Limit-row outputs. Distribution rules
	// treat such a stage like a sort-free grouped final (input clustered
	// on GroupByCols, output mirrors the input partitioning) — without
	// this flag SortKeys/Limit force the Singleton collapse. Set only by
	// fuseSortIntoPredecessor's shard-local fold.
	SortShardLocal bool

	// Join metadata
	JoinType        string // inner, left, right, full, cross
	JoinLeftKeys    []string
	JoinRightKeys   []string
	LeftDepStage    string // stage providing probe (left) side
	RightDepStage   string // stage providing build (right) side
	BuildTableAlias string // build-side table alias for column disambiguation in self-joins
	// BuildColOrigins maps each bare build-output column (lowercased) to the
	// scan alias that owns it. Only set when the build subtree spans multiple
	// tables (bushy shapes) — nil for single-scan builds, where
	// BuildTableAlias is already exact. The join executor qualifies duplicate
	// build columns with the OWNING alias instead of BuildTableAlias.
	BuildColOrigins map[string]string
	JoinFilter      string // semi/anti join inequality filter (e.g., "l2.l_suppkey != l1.l_suppkey")
	// BuildFilterExprs are row predicates applied to the BUILD input before
	// hash-table insertion. Set by dedupeSubsumedScanExchanges when this
	// join's build was rewired from a filtered exchange to a subsuming raw
	// exchange: the dropped exchange's scan filter (or its computed
	// __subsume flag) must now run at build-read time. Semantically
	// identical to filtering at the dropped scan.
	BuildFilterExprs []string

	// Fused broadcast joins absorbed into this stage (avoids separate
	// shuffle+join stages for small dimension tables like nation, region).
	FusedJoins []FusedJoinSpec

	// ChainedJoins are 1:1 downstream joins absorbed into this stage by
	// fuseStageChains (docs/design/stage-chain-fusion.md). Unlike
	// FusedJoins (broadcast probes applied BEFORE the primary join), these
	// run AFTER it, in order — the fragment pipes the primary's output
	// through each chained probe in-process, eliding the per-link
	// materialization the separate stages paid.
	ChainedJoins []ChainedJoinSpec

	// ChainedAgg* describe a downstream PARTIAL aggregate absorbed as the
	// chain's terminal step (fuseStageChains step 2): the fragment runs
	// OpHashAggregate (raw mode) after the chained joins, so the join
	// output collapses to partials in-process instead of materializing
	// for a separate round-robin aggregate stage. Partial aggregation is
	// partition-agnostic, so the fused stage keeps ITS OWN distribution
	// and task count — it just emits N partial outputs instead of the
	// dropped stage's fan-out count; finals merge either way.
	ChainedAggGroupBy []string
	ChainedAggSpecs   []AggSpec

	// Window metadata
	WindowCols []WindowColSpec

	// JoinPartitionCount is the number of partitions for a hash-join stage
	// that was preceded by repartition exchanges. Zero means the join is
	// not partitioned (broadcast or single-partition). Exchange stages carry
	// their partition count on Exchange.Count instead.
	JoinPartitionCount int

	// Fused scan-aggregate: partial aggregation is performed at the scan
	// level, eliminating the scan→aggregate S3 round-trip. Workers produce
	// partial aggregate results instead of raw rows.
	FusedAggGroupBy []string
	FusedAggSpecs   []AggSpec

	// RawInputAggregate marks a final_aggregate whose input is RAW rows
	// from an exchange hash-partitioned on the group keys, not partial
	// aggregates — set by rewireAggOverRawExchange when it rewires the
	// final from a duplicate fused scan-agg leg onto a sibling raw
	// exchange. Partition-disjoint keys make per-partition raw aggregation
	// exact, so the dispatcher builds the fragment with MergeMode=false
	// (no InputCol→OutputCol remap, no COUNT→SUM rewrite). AggSpecs carry
	// the raw form (the dropped scan's FusedAggSpecs).
	RawInputAggregate bool

	// Probe-split pipeline: partition the probe table's files across workers.
	// Each worker scans build tables in full and probes its file partition.
	ProbeSplitAlias string   // scan alias to partition (e.g., "lineitem")
	ProbeSplitFiles []string // full file list to split across tasks

	// BuildCachePreScans holds pre-scanned result file paths for large build
	// tables. When populated by the coordinator (after pre-scanning them once),
	// workers load these cached files via PreScannedInputs instead of scanning
	// the large source table N times — eliminating the N× build-side duplication
	// that causes OOM on Q09 at SF100. Keyed by scan alias (e.g., "orders").
	BuildCachePreScans map[string][]string

	// PreComputedAggregates holds signatures + cache paths for derived-
	// aggregate builds that were computed once by the coordinator before
	// dispatch. Each probe-split task carries the same list; the worker's
	// plan-rewrite pass matches logical Aggregate subtrees against the
	// signatures and replaces them with synthetic scans of the cache files.
	// Spec: 2026-04-18-shuffle-distributed-aggregate.md.
	PreComputedAggregates []PreComputedAggregateMeta

	// Multi-level merge: partitions upstream results among parallel merge groups.
	// When MergeGroupCount > 0, this stage processes only the MergeGroup-th
	// fraction of its dependency results. Independent merge groups run on
	// different workers for parallel merging.
	MergeGroup      int // 0-based index of this merge group
	MergeGroupCount int // total groups (0 = not grouped, process all results)

	// Cost estimation (populated at plan time from manifest metadata)
	EstimatedBytes int64
	EstimatedRows  int64

	// Distribution describes how this stage's output is partitioned.
	// Default zero value is {Kind: DistSingleton} which is correct for
	// most existing stages (single-worker output). Shuffle stages set this
	// to DistHashPartitioned with Keys and Count populated. Broadcast pre-scans
	// (build cache) set Kind: DistBroadcast.
	Distribution Distribution

	// Exchange carries per-variant metadata for StageExchange* stages.
	// nil for non-Exchange stages.
	Exchange *ExchangeStage

	// ScalarDependencies maps placeholder names (e.g. ":scalar_1") to
	// producer stage IDs that emit a single-row, single-column output. The
	// native-DAG coordinator awaits each producer, extracts the scalar from
	// its stage output, and string-substitutes the placeholder in this
	// stage's FilterExprs / AggSpecs.InputExpr before dispatching tasks.
	// This lets CTE-referencing scalar subqueries share the distributed
	// float-accumulation path with the filter-carrying stage's upstream,
	// eliminating the single-process vs distributed bit-pattern divergence
	// that caused Q15 to return 0 rows at SF0.1.
	ScalarDependencies map[string]string

	// OutputRenames is the SELECT-list alias map applied by the coordinator
	// to the Gather stage's result schema. walkStages currently passes
	// NodeProject through without applying its projections, so without this
	// the final result schema carries raw worker column names ("n1.n_name",
	// "substr(l_shipdate, 1, 4)") instead of the user's aliases
	// ("supp_nation", "l_year"). Only populated on the Gather stage.
	OutputRenames []OutputRename

	// ProjectExprs, set on a leaf scan stage whose output feeds the gather
	// directly, makes the scan fragment compute the SELECT list (worker-side
	// exec.Project after scan+filter). Without it a bare expression SELECT
	// over a scan reaches the gather as raw scan columns — the gather's
	// applyOutputRenames can rename/drop but not evaluate (#169). Expression
	// entries are named by the lowercased expression text, matching the
	// convention extractOutputRenames already expects for worker-computed
	// expressions; bare columns are passthrough entries so the fragment
	// output is exactly the SELECT-list inputs.
	ProjectExprs []ProjectExprSpec

	// SecurityProjectExprs is the ABAC security barrier absorbed from a
	// SecurityBarrier logical Project wrapping this scan
	// (absorbSecurityBarrier): visible columns pass through, masked columns
	// are literal expressions, denied columns are absent. Applied as the
	// FIRST projection in the scan fragment — before ProjectExprs and
	// before any aggregate — so restricted values never leave the worker.
	SecurityProjectExprs []ProjectExprSpec

	// Dynamic filter (Trino-style semi-join pushdown) annotations.
	// EmitDynamicFilters is set on a build-side leaf scan stage; each task
	// computes a partial KeyRange+Bloom and uploads as a sideband artifact.
	// ConsumeDynamicFilters is set on a probe-side leaf scan stage; the
	// coordinator unions the upstream partials and injects the result into
	// each scan task's OpSpec.DynamicFilters.
	//
	// The stat-dep edge (emit stage ID appended to the consume stage's
	// Dependencies) makes execute_stage_dag.go serialize them via the
	// existing dependency mechanism — no new edge type or async broker.
	EmitDynamicFilters    []DynamicFilterEmit
	ConsumeDynamicFilters []DynamicFilterConsume

	// QualifyAllBuildCols, when true, instructs the join executor to always
	// emit build-side columns under their qualified name
	// ("BuildTableAlias.col_name") instead of the default behavior of
	// qualifying only on probe-collision. Set by the planner when the same
	// source table is scanned more than once and the scans co-path into the
	// same join chain (Q07's "nation n1" + "nation n2"). Without this flag
	// the FIRST self-join leaves its column unqualified, the SECOND qualifies
	// only its own copy, and references to the FIRST alias resolve to NULL
	// downstream.
	QualifyAllBuildCols bool
}

type WindowColSpec struct {
	Func        string
	InputCol    string
	OutputCol   string
	PartitionBy []string
	OrderBy     []SortKeySpec
	Frame       *logical.WindowFrameSpec
}