Documentation
¶
Overview ¶
Package physical turns a logical plan into runnable operators, and expressions into columns.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Eval ¶
Eval computes an expression over a batch, producing one column.
The contract ¶
For every resolved node n and every batch b:
Eval(ctx, n, b).DType() == n.Field(b.Schema()).Type
That equality is what makes lazy schema resolution HONEST. CollectSchema promises a schema before any data is read; if the evaluator can produce a column of a different type, the promise is a lie and the failure surfaces far downstream as corrupted output. TestEvaluatorContract asserts it across the whole op × dtype matrix.
Keeping the promise has one rule: the evaluator must not re-derive types. It asks expr.ResolveBinary for the operand and result types and obeys the answer. A second, independent copy of the promotion rules would drift, and the drift would be exactly this bug.
func Pull ¶ added in v0.3.0
Pull is op.Next, with a panic in it returned as an error.
Every driver pulls through this rather than calling Next: the serial loops in internal/exec, and the dispatchers of parallelOp, parProbeOp and parallelSink. A panic anywhere below — in a kernel, a source, a decoder — then reaches the caller as ErrInternal instead of killing its goroutine, and with it, for a worker's, the process. Recovering around the CALL rather than around a goroutine is what keeps each driver's own bookkeeping — closing its lanes, delivering the error in sequence — running as it does for any other error.
Types ¶
type BatchOp ¶
type BatchOp interface {
Schema() *dtype.Schema
Apply(ctx context.Context, in *data.Batch) (*data.Batch, error)
Close() error
}
BatchOp is a stateless, order-independent transform of one batch into one batch.
Every operator that CAN be a BatchOp MUST be one. That is the rule that keeps the pull-to-push migration mechanical.
type KernelFolder ¶
type KernelFolder struct{}
KernelFolder evaluates constant expressions for the optimizer's simplification rule.
Why it lives here and not in internal/plan ¶
Folding must not compute the answer a second way. `Lit(0.1).Add(Lit(0.2))` has to fold to exactly what the executor would produce, bit for bit, or a query mixing folded and unfolded values disagrees with itself — and the only way to guarantee that is to run the SAME evaluator. So this is Eval over a one-row batch with no columns, not a reimplementation of arithmetic.
But Eval lives at L50 with Arrow, the allocator and GOEXPERIMENT=simd behind it, and internal/plan at L40 has none of those: design/logical.md §13's claim is that a plan can be built, resolved, optimized and explained "without an executor, without Arrow, and without a Parquet file". Importing kernel from there to fold two literals would retire that for the whole layer.
So the dependency runs the way the levels already allow — physical knows about plan — and plan takes the folder as an interface naming no Arrow type. lazy.go injects it. Nothing else changes.
func (KernelFolder) Fold ¶
Fold evaluates n and returns it as a literal.
The (nil, false, nil) return means "not folded" and is the normal outcome for anything this cannot round-trip exactly. An error return means the same thing to the caller — see simplify.foldExpr for why a kernel error must never become a query failure — but is passed back so a genuinely internal fault is not disguised as a declined optimization.
type Operator ¶
type Operator interface {
Schema() *dtype.Schema
// Next returns the next non-empty batch, or (nil, io.EOF) at end of stream.
// Implementations must check ctx.Err() before doing work.
Next(ctx context.Context) (*data.Batch, error)
Close() error
}
Operator is the pull-based execution interface.
Pull now, push later, without rewriting the operators ¶
v0.1 executes a single-threaded volcano: the consumer calls Next and the tree pulls. Pull was chosen because it is debuggable (one stack trace shows the whole pipeline), it starts no goroutines so there is nothing to leak on cancellation, and it is the natural shape for CollectBatches where the caller drives.
The morsel-driven push engine planned for v0.2 wants the opposite polarity. The move that makes that migration cheap is below: Filter and Project are NOT Operators. They are BatchOps — stateless per-batch transforms — and a 20-line adapter runs a BatchOp inside the pull chain. In v0.2 the same BatchOp values are handed to the scheduler unchanged. Zero kernels, zero evaluator and zero BatchOps have to change; only the driver does.
A driver MUST deliver batches in input order ¶
This is an obligation, not an accident, and it went unwritten for four steps while the only driver was a serial loop whose `append` happened to satisfy it. Five things depend on it:
sortSink stability — ties keep their input order positionAcc First and Last are POSITIONAL distinctOp keeps the FIRST row per key limitOp which rows survive, not just how many hashAggSink group ids are assigned in first-appearance order
None of the five would error if it were violated; they would return different rows. So a scheduler may run operators concurrently — parallelOp does — but it may not reorder what they produce. See internal/physical/parallel.go.
func Plan ¶
Plan lowers a resolved, optimized logical plan into an operator tree.
It may assume the plan is resolved: every Scan has its source schema, every Project holds single-output expressions, and every Filter predicate is Boolean. Resolve guarantees all three, so violations here are ursus bugs, not user errors.
func PlanRoot ¶
PlanRoot lowers a plan and parallelises the top of it if it is a pipeline.
Plan itself parallelises each child that a NON-pipeline operator consumes; this covers the remaining case, a query that is nothing but a pipeline. The two together mean every maximal pipeline in the tree is wrapped exactly once.
type Options ¶
type Options struct {
BatchSize int
// Threads is the number of workers over the pipeline portion of the plan.
// 0 or 1 means serial, and reproduces the pre-step-5 operator tree exactly.
Threads int
// Budget is the query's memory ceiling and spill location, shared by every
// buffering operator in the tree. A nil Budget means unlimited and untracked,
// so an Options built by hand still works.
Budget *execopt.Budget
}
Options tunes physical planning.
func DefaultOptions ¶
func DefaultOptions() Options
DefaultOptions returns sensible planning defaults.
8192 rows is the usual columnar sweet spot: large enough that per-batch overhead and SIMD setup amortise away, small enough that a batch's working set stays in L2 and that a cancelled query stops promptly.
type ProbeBuilder ¶
type ProbeBuilder interface {
Sink
// Probe returns the operator that streams the probe side and emits the join.
// Called ONCE, after the last Consume, INSTEAD of Finish.
//
// OWNERSHIP: the returned Operator does NOT own probe. joinBreaker owns both
// children and closes both; the returned operator's Close releases only its
// own state. An operator that also closed probe would double-close it.
Probe(ctx context.Context, probe Operator) (Operator, error)
}
ProbeBuilder is the BUILD half of a two-input pipeline breaker.
It is a Sink — the whole Consume/Merge/Finish contract, unchanged — plus one method. Sink is deliberately not widened: a one-input breaker genuinely has one child, and making every sink carry a second would be the cost without the benefit.
Finish and Probe are two views of one result:
Finish(ctx) == Probe(ctx, an operator that yields no rows)
That is not a stub. It is the join of the build side against an empty probe side: zero rows for Inner/Left/Semi/Anti/Cross, and every build row null-padded for Right/Full. Keeping Finish total rather than panic("use Probe") keeps Sink's contract honest and gives the suite a free equivalence — see TestJoinFinishEqualsProbeOfEmpty.
type Sink ¶
type Sink interface {
Schema() *dtype.Schema
// Consume accumulates one batch.
//
// It MAY retain in without copying, and four of the five implementations do —
// a Batch is a schema pointer plus immutable Columns, so retaining one retains
// immutable values. This comment used to say the opposite, which contradicted
// three sinks that state their retention in a comment and the whole of step
// 10's accounting, whose central fact is that "every pipeline breaker retains
// the source's own batches rather than copies".
//
// A sink that retains must account for what it holds through its execopt
// Account, or a memory limit measures something other than what is held.
Consume(ctx context.Context, in *data.Batch) error
// Merge folds another Sink's partial state into this one. `other` must have
// consumed a LATER portion of the input than this one, which is what makes
// order-sensitive aggregates (First, Last) well-defined under merging.
Merge(other Sink) error
// Finish produces the result as an Operator. Called once, after the last
// Consume.
Finish(ctx context.Context) (Operator, error)
Close() error
}
Sink is a PIPELINE BREAKER: an operator that must consume all of its input before it can emit anything.
Why this is a third shape rather than just an Operator ¶
Sort and Aggregate could each be written as a plain Operator that drains its child on the first Next call. That works today and forfeits the property step 1 spent its architecture budget buying: that v0.2's morsel scheduler consumes the same operator values unchanged.
Consume/Merge/Finish is the shape a parallel scheduler needs. N workers each get their own Sink, consume morsels independently, and the partial results are merged. Written as a monolithic Operator, both would have to be rewritten the day the scheduler lands — which is exactly the rewrite step 1 was designed to avoid.
Merge is NOT the spilling seam, whatever this comment used to say ¶
Three operators spill now — sort (step 10), group_by (step 12) and join (step 13) — and not one of them goes through Merge. Each spills inside Consume and rejoins inside Finish or Probe, and all three REFUSE to Merge once they have spilled, because a fold that dropped the other sink's files would be silent.
It said: "a Sink that runs out of memory writes its partial state out, starts fresh, and merges at the end." Read against the interface, every verb in that sentence except "merges" is missing — there is no spill, no reset, and Merge takes a live Sink rather than anything on disk.
It is also wrong in the specific case it was written for. Merge's contract is that `other` consumed a LATER portion of the input, and sortSink.Merge honours it by concatenating in INPUT order: neither sink has sorted anything at that point, so the sort's Merge does the opposite of a merge. External sort therefore spills inside Consume and merges inside Finish (see extsort.go), and Merge keeps the meaning it actually has.
Note also that the SEVEN implementations give the "LATER portion" clause four different readings — append (sort AND reverse), remap-and-append (join), refuse-unless-identical (group_by), refuse-always (window, temporal group, as-of build). That is a per-operator rule rather than a universal one, and it is incompatible with a radix-partitioned aggregation, whose partials are disjoint by hash rather than ordered by input position.
This paragraph used to claim five implementations and a fifth reading — "PREPEND (reverse, because reversal turns later into earlier)". Both were wrong. reverseSink buffers untouched and reverses the whole concatenation in Finish, so it appends for the same reason sortSink does; prepending merely put the concatenation in the wrong order, and the count was never five.
Nothing calls Merge concurrently yet, and it is implemented and tested anyway, because a Merge written later against forgotten invariants is a Merge that is wrong. reverseSink is the proof: it went from step 16 to step 50 with no test at all, and it was not merely unverified but WRONG.
type SinkFactory ¶
SinkFactory builds one worker's Sink. Every sink a factory returns must be independently usable and mutually mergeable — same schema, same aggregates, same numbering rules.
Source Files