From e7c74de79d944a04c712c72ee80dab3963231d2d Mon Sep 17 00:00:00 2001 From: Steven McCanne Date: Tue, 10 Nov 2026 13:08:41 -0800 Subject: [PATCH 1/2] rework runtime output model This commit reworks the output model so all outputs are managed how debug used to work. Debug now shares this refactored approach. This means output can be invoked from a subquery. It also performs better than the old approach and fixes a prior deadlock in the debug design. This will be utilized by the forthcoming catch PR, which will provide a way to handle dynamic errors and use this new model to optionally route values to outputs. --- compiler/compiler.go | 4 +- compiler/package.go | 32 ++- compiler/rungen/op.go | 39 ++-- compiler/rungen/vop.go | 10 +- compiler/semantic/dagen.go | 22 +- runtime/op/debug.go | 56 ++--- runtime/op/mux.go | 275 ++++++++++--------------- runtime/op/mux_test.go | 107 ++++++++++ runtime/op/output.go | 36 +++- runtime/ztests/op/debug-subquery.yaml | 14 ++ runtime/ztests/op/output-subquery.yaml | 99 +++++++++ 11 files changed, 425 insertions(+), 269 deletions(-) create mode 100644 runtime/op/mux_test.go create mode 100644 runtime/ztests/op/debug-subquery.yaml create mode 100644 runtime/ztests/op/output-subquery.yaml diff --git a/compiler/compiler.go b/compiler/compiler.go index c109eeaa4d..e821f05241 100644 --- a/compiler/compiler.go +++ b/compiler/compiler.go @@ -66,11 +66,11 @@ func (l *compiler) NewDeleteQuery(rctx *runtime.Context, ast *parser.AST, head * if err = optimizer.New(rctx, l.env).OptimizeDeleter(main, Parallelism); err != nil { return nil, err } - outputs, debugs, b, err := BuildWithBuilder(rctx, main, l.env) + leaves, outputs, b, err := BuildWithBuilder(rctx, main, l.env) if err != nil { return nil, err } - return exec.NewDeleteQuery(rctx, bundleOutputs(rctx, outputs, debugs), b.Deletes()), nil + return exec.NewDeleteQuery(rctx, bundleOutputs(rctx, leaves, outputs), b.Deletes()), nil } func (l *compiler) NewObjectScanner(rctx *runtime.Context, tableID uuid.UUID, objects []*data.Object) (vio.Puller, error) { diff --git a/compiler/package.go b/compiler/package.go index 5e0b696544..41dc946d77 100644 --- a/compiler/package.go +++ b/compiler/package.go @@ -34,22 +34,22 @@ func Optimize(ctx context.Context, main *dag.Main, env *exec.Environment, parall return nil } -func Build(rctx *runtime.Context, main *dag.Main, env *exec.Environment) (map[string]vio.Puller, *op.DebugChans, vio.Meter, error) { +func Build(rctx *runtime.Context, main *dag.Main, env *exec.Environment) ([]vio.Puller, *op.Outputs, vio.Meter, error) { b := rungen.NewBuilder(rctx, env) - outputs, debugs, err := b.Build(main) + leaves, outputs, err := b.Build(main) if err != nil { return nil, nil, nil, err } - return outputs, debugs, b.Meter(), nil + return leaves, outputs, b.Meter(), nil } -func BuildWithBuilder(rctx *runtime.Context, main *dag.Main, env *exec.Environment) (map[string]vio.Puller, *op.DebugChans, *rungen.Builder, error) { +func BuildWithBuilder(rctx *runtime.Context, main *dag.Main, env *exec.Environment) ([]vio.Puller, *op.Outputs, *rungen.Builder, error) { b := rungen.NewBuilder(rctx, env) - outputs, debugs, err := b.Build(main) + leaves, outputs, err := b.Build(main) if err != nil { return nil, nil, nil, err } - return outputs, debugs, b, nil + return leaves, outputs, b, nil } func CompileWithAST(rctx *runtime.Context, ast *parser.AST, env *exec.Environment, optimize bool, parallel int) (*exec.Query, error) { @@ -63,11 +63,11 @@ func CompileWithAST(rctx *runtime.Context, ast *parser.AST, env *exec.Environmen return nil, err } } - outputs, debugs, meter, err := Build(rctx, main, env) + leaves, outputs, meter, err := Build(rctx, main, env) if err != nil { return nil, err } - return exec.NewQuery(rctx, bundleOutputs(rctx, outputs, debugs), meter), nil + return exec.NewQuery(rctx, bundleOutputs(rctx, leaves, outputs), meter), nil } func Compile(rctx *runtime.Context, env *exec.Environment, optimize bool, parallel int, inputs []srcfiles.Input) (*exec.Query, error) { @@ -78,17 +78,11 @@ func Compile(rctx *runtime.Context, env *exec.Environment, optimize bool, parall return CompileWithAST(rctx, ast, env, optimize, parallel) } -func bundleOutputs(rctx *runtime.Context, outputs map[string]vio.Puller, chans *op.DebugChans) vio.Puller { - switch len(outputs) + len(chans.Debug) { - case 0: +// bundleOutputs returns the mux for the leaves and outputs of a flowgraph or +// nil if there are no leaves. +func bundleOutputs(rctx *runtime.Context, leaves []vio.Puller, outputs *op.Outputs) vio.Puller { + if len(leaves) == 0 { return nil - case 1: - var puller vio.Puller - for k, p := range outputs { - puller = op.NewCatcher(op.NewSingle(k, p)) - } - return puller - default: - return op.NewMux(rctx, outputs, chans) } + return op.NewMux(rctx, outputs, leaves) } diff --git a/compiler/rungen/op.go b/compiler/rungen/op.go index 1a35c0c8f5..629c0c60ce 100644 --- a/compiler/rungen/op.go +++ b/compiler/rungen/op.go @@ -24,13 +24,15 @@ import ( var ErrJoinParents = errors.New("join requires two upstream parallel query paths") type Builder struct { - rctx *runtime.Context - mctx *super.Context - mapper *super.TypeDefsMapper - env *exec.Environment - progress *vio.Progress - debugs *op.DebugChans - channels map[string][]vio.Puller + rctx *runtime.Context + mctx *super.Context + mapper *super.TypeDefsMapper + env *exec.Environment + progress *vio.Progress + // outputs is the channel to the mux for the output and debug operators. + // It's nil when there is no mux (e.g., when an expression is evaluated + // at compile time). + outputs *op.Outputs deletes *sync.Map funcs map[string]*dag.FuncDef compiledVamUDFs map[string]*expr.UDF @@ -43,15 +45,16 @@ func NewBuilder(rctx *runtime.Context, env *exec.Environment) *Builder { mctx: super.NewContext(), env: env, progress: &vio.Progress{}, - debugs: op.NewDebugChans(), - channels: make(map[string][]vio.Puller), + outputs: op.NewOutputs(), funcs: make(map[string]*dag.FuncDef), compiledVamUDFs: make(map[string]*expr.UDF), } } -// Build builds a flowgraph for main. -func (b *Builder) Build(main *dag.Main) (map[string]vio.Puller, *op.DebugChans, error) { +// Build builds a flowgraph for main and returns its leaves, each of which +// ends in an output operator, and the channel on which its output and debug +// operators send (see op.Mux). +func (b *Builder) Build(main *dag.Main) ([]vio.Puller, *op.Outputs, error) { if !isEntry(main.Body, true) { return nil, nil, errors.New("internal error: DAG entry point is not a data source") } @@ -62,14 +65,13 @@ func (b *Builder) Build(main *dag.Main) (map[string]vio.Puller, *op.DebugChans, } b.mapper = super.NewTypeDefsMapper(b.rctx.Sctx, defs) } - if _, err := b.compileVamMain(main, nil); err != nil { + leaves, err := b.compileVamMain(main, nil) + if err != nil { return nil, nil, err } - channels := make(map[string]vio.Puller) - for key, pullers := range b.channels { - channels[key] = b.combineVam(pullers) - } - return channels, b.debugs, nil + // A leaf is nil when an output operator has no parent. + leaves = slices.DeleteFunc(leaves, func(p vio.Puller) bool { return p == nil }) + return leaves, b.outputs, nil } func (b *Builder) sctx() *super.Context { @@ -178,6 +180,9 @@ func EvalAtCompileTime(sctx *super.Context, main *dag.MainExpr) (val super.Value // the query of a type declaration. func EvalAtCompileTimeWithEnv(ctx context.Context, sctx *super.Context, env *exec.Environment, main *dag.MainExpr) (val super.Value, err error) { b := NewBuilder(runtime.NewContext(ctx, sctx), env) + // There's no mux so the values of output and debug operators are + // discarded. + b.outputs = nil for _, f := range main.Funcs { b.funcs[f.Tag] = f } diff --git a/compiler/rungen/vop.go b/compiler/rungen/vop.go index b17c9ceaf4..9dde89c770 100644 --- a/compiler/rungen/vop.go +++ b/compiler/rungen/vop.go @@ -218,8 +218,7 @@ func (b *Builder) compileVamLeaf(o dag.Op, parent vio.Puller) (vio.Puller, error if err != nil { return nil, err } - d := op.NewDebug(b.rctx, e, filter, b.debugs, parent) - return d, nil + return op.NewDebug(b.rctx, e, filter, b.outputs, parent), nil case *dag.DeleterScan: table, err := b.lookupTable(o.Table) if err != nil { @@ -316,11 +315,10 @@ func (b *Builder) compileVamLeaf(o dag.Op, parent vio.Puller) (vio.Puller, error case *dag.NullScan: return vio.NewPuller(vector.NewNull(1)), nil case *dag.OutputOp: - if parent != nil { - parent = op.NewOutput(parent) + if parent == nil { + return nil, nil } - b.channels[o.Name] = append(b.channels[o.Name], parent) - return parent, nil + return op.NewOutput(b.rctx, parent, b.outputs, o.Name), nil case *dag.PassOp: return parent, nil case *dag.PutOp: diff --git a/compiler/semantic/dagen.go b/compiler/semantic/dagen.go index ff561d8eeb..899a594e57 100644 --- a/compiler/semantic/dagen.go +++ b/compiler/semantic/dagen.go @@ -28,7 +28,7 @@ func newDagen(r reporter) *dagen { func (d *dagen) assemble(seq sem.Seq, types []byte, funcs map[string]*funcDef) *dag.Main { dagSeq := d.seq(seq) - dagSeq = d.checkOutputs(true, dagSeq) + dagSeq = d.checkOutputs(true, true, dagSeq) dagFuncs := make([]*dag.FuncDef, 0, len(d.funcs)) for _, f := range funcs { dagFuncs = append(dagFuncs, d.fn(f)) @@ -307,7 +307,7 @@ func (d *dagen) unnest(op *sem.UnnestOp) dag.Op { if op.Body == nil { return u } - body := d.seq(op.Body) + body := d.checkOutputs(true, false, d.seq(op.Body)) body.Prepend(u) body = collectThis(body) s := &dag.SubqueryExpr{ @@ -591,7 +591,7 @@ func (d *dagen) subquery(e *sem.SubqueryExpr) *dag.SubqueryExpr { Kind: "SubqueryExpr", Correlated: e.Correlated, SQL: e.SQL, - Body: d.seq(e.Body), + Body: d.checkOutputs(true, false, d.seq(e.Body)), } if e.Array { subquery.Body = collectThis(subquery.Body) @@ -635,12 +635,14 @@ func (d *dagen) fn(f *funcDef) *dag.FuncDef { } } -func (d *dagen) checkOutputs(isLeaf bool, seq dag.Seq) dag.Seq { +// checkOutputs reports an error for any output operator of seq that isn't +// located in a leaf and, if addMain is true, adds an output operator for +// the main output to any leaf without one. (The leaves of a subquery's +// flowgraph produce its values so outputs aren't added there.) +func (d *dagen) checkOutputs(isLeaf, addMain bool, seq dag.Seq) dag.Seq { if len(seq) == 0 { return seq } - // - Report an error in any outputs are not located in the leaves. - // - Add output operators to any leaves where they do not exist. lastN := len(seq) - 1 for i, o := range seq { isLast := lastN == i @@ -651,22 +653,22 @@ func (d *dagen) checkOutputs(isLeaf bool, seq dag.Seq) dag.Seq { } case *dag.ScatterOp: for k := range o.Paths { - o.Paths[k] = d.checkOutputs(isLast && isLeaf, o.Paths[k]) + o.Paths[k] = d.checkOutputs(isLast && isLeaf, addMain, o.Paths[k]) } case *dag.ForkOp: for k := range o.Paths { - o.Paths[k] = d.checkOutputs(isLast && isLeaf, o.Paths[k]) + o.Paths[k] = d.checkOutputs(isLast && isLeaf, addMain, o.Paths[k]) } case *dag.SwitchOp: for k := range o.Cases { - o.Cases[k].Path = d.checkOutputs(isLast && isLeaf, o.Cases[k].Path) + o.Cases[k].Path = d.checkOutputs(isLast && isLeaf, addMain, o.Cases[k].Path) } } } switch seq[lastN].(type) { case *dag.ForkOp, *dag.OutputOp, *dag.ScatterOp, *dag.SwitchOp: default: - if isLeaf { + if isLeaf && addMain { return append(seq, &dag.OutputOp{Kind: "OutputOp", Name: "main"}) } } diff --git a/runtime/op/debug.go b/runtime/op/debug.go index c159b7e4a9..3e07c9cea8 100644 --- a/runtime/op/debug.go +++ b/runtime/op/debug.go @@ -2,55 +2,37 @@ package op import ( "context" - "sync" "github.com/superdb/super/runtime/expr" "github.com/superdb/super/vector" "github.com/superdb/super/vector/vio" ) +// Debug passes its input through and sends the values of expr to the query's +// outputs labeled "debug". If outputs is nil (e.g., when an expression is +// evaluated at compile time), the values are discarded. type Debug struct { - parent vio.Puller - ctx context.Context - expr expr.Evaluator - filter expr.Evaluator - ch chan vector.Any - eos <-chan struct{} - once sync.Once + parent vio.Puller + ctx context.Context + expr expr.Evaluator + filter expr.Evaluator + outputs *Outputs } -func NewDebug(ctx context.Context, expr expr.Evaluator, filter expr.Evaluator, chans *DebugChans, parent vio.Puller) *Debug { +func NewDebug(ctx context.Context, expr expr.Evaluator, filter expr.Evaluator, outputs *Outputs, parent vio.Puller) *Debug { return &Debug{ - parent: parent, - ctx: ctx, - expr: expr, - filter: filter, - ch: chans.Next(), - eos: chans.EOS, + parent: parent, + ctx: ctx, + expr: expr, + filter: filter, + outputs: outputs, } } func (d *Debug) Pull(done bool) (vector.Any, error) { - local := make(chan vector.Any) - d.once.Do(func() { - go func() { - for { - select { - case vec := <-local: - d.ch <- vec - case <-d.eos: - // send eos ack - d.ch <- nil - return - case <-d.ctx.Done(): - return - } - } - }() - }) val, err := d.parent.Pull(done) - if val == nil { - return nil, err + if val == nil || d.outputs == nil { + return val, err } filtered := val if d.filter != nil { @@ -58,10 +40,8 @@ func (d *Debug) Pull(done bool) (vector.Any, error) { } // A plain none is dropped, as for the values operator. if e := dropPlainNones(d.expr.Eval(filtered)); e != nil && e.Len() != 0 { - select { - case local <- e: - case <-d.ctx.Done(): - return nil, d.ctx.Err() + if err := d.outputs.send(d.ctx, "debug", e); err != nil { + return nil, err } } return val, err diff --git a/runtime/op/mux.go b/runtime/op/mux.go index 393bf0afd3..cda05de811 100644 --- a/runtime/op/mux.go +++ b/runtime/op/mux.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "runtime/debug" + "slices" "sync" "github.com/superdb/super/runtime" @@ -29,152 +30,152 @@ func (c *Catcher) Pull(done bool) (vec vector.Any, err error) { return c.parent.Pull(done) } -// Mux implements the muxing of a set of parallel paths at the output of -// a flowgraph. It also implements the double-EOS algorithm with proc.Latch -// to detect the end of each parallel stream. Its output protocol is a single EOS -// when all of the upstream legs are done at which time it cancels the flowgraoh. -// Each batch returned by the mux is wrapped in a Batch, which can be unwrappd -// with Unwrap to extract the integer index of the output (in left-to-right -// DFS traversal order of the flowgraph). This proc requires more than one -// parent; use proc.Latcher for a single-output flowgraph. -type Mux struct { - rctx *runtime.Context - once sync.Once - ch <-chan muxresult - parents []*puller - nparents int - debugger *debugger - eosCh chan<- struct{} +// Outputs is the channel by which the output and debug operators of a query, +// whether in the query's flowgraph or in a subquery, send labeled vectors to +// the mux (see Mux). A nil *Outputs discards the vectors sent to it. +type Outputs struct { + ch chan muxresult + mu sync.Mutex + // labels holds the labels of the output operators. + labels map[string]struct{} } -type DebugChans struct { - Debug []chan vector.Any - EOS chan struct{} +func NewOutputs() *Outputs { + return &Outputs{ + ch: make(chan muxresult), + labels: make(map[string]struct{}), + } } -func NewDebugChans() *DebugChans { - return &DebugChans{EOS: make(chan struct{})} +func (o *Outputs) addLabel(label string) { + if o == nil { + return + } + // Output operators are created at runtime for the copies of a + // correlated subquery (see Subquery), so this may race with others. + o.mu.Lock() + o.labels[label] = struct{}{} + o.mu.Unlock() +} + +func (o *Outputs) sortedLabels() []string { + o.mu.Lock() + defer o.mu.Unlock() + var labels []string + for label := range o.labels { + labels = append(labels, label) + } + slices.Sort(labels) + return labels } -func (d *DebugChans) Next() chan vector.Any { - ch := make(chan vector.Any) - d.Debug = append(d.Debug, ch) - return ch +// send sends vec labeled label to the mux. +func (o *Outputs) send(ctx context.Context, label string, vec vector.Any) error { + if o == nil { + return nil + } + select { + case o.ch <- muxresult{vec: vec, label: label}: + return nil + case <-ctx.Done(): + return ctx.Err() + } } type muxresult struct { vec vector.Any label string err error + // pumped is true when a pump is done (with err if it failed). + pumped bool +} + +// Mux collects the outputs of a query. Its output and debug operators send +// vectors on the query's Outputs, and the mux returns each vector labeled with +// the name of its output (see vector.Labeled). The mux drives the leaves of +// the query's flowgraph, each of which ends in an output operator, with a pump +// for each leaf: a goroutine that pulls the leaf until EOS. (An output +// operator in a subquery is driven by the subquery's evaluation.) +// +// Since the mux only reads from the Outputs channel and never pulls from the +// flowgraph, an operator's send always makes progress as long as the mux is +// pulled. +// +// When all of the pumps are done, no operator can send, so the mux returns an +// end of channel (a Labeled with a nil vector) for each output's label and +// then EOS, and it cancels the query. There's no end of channel for debug. +type Mux struct { + rctx *runtime.Context + once sync.Once + outputs *Outputs + pumps []vio.Puller + running int + ends []string } -func (m muxresult) vector() *vector.Labeled { - return &vector.Labeled{Any: m.vec, Label: m.label} -} - -type puller struct { - vio.Puller - ch chan<- muxresult - label string -} - -func (p *puller) run(ctx context.Context) { - for { - vec, err := p.Pull(false) - select { - case p.ch <- muxresult{vec, p.label, err}: - if vec == nil || err != nil { - return - } - case <-ctx.Done(): - return - } - } -} - -func NewMux(rctx *runtime.Context, parents map[string]vio.Puller, chans *DebugChans) *Mux { - if len(parents)+len(chans.Debug) <= 1 { - panic("NewMux must be called with two or more parents") - } - ch := make(chan muxresult) - pullers := make([]*puller, 0, len(parents)) - for label, parent := range parents { - pullers = append(pullers, &puller{NewCatcher(parent), ch, label}) - } - var debugger *debugger - if len(chans.Debug) != 0 { - debugger = newDebugger(chans) - } +// NewMux returns a Mux for outputs and the leaves of a query's flowgraph, +// pumps. +func NewMux(rctx *runtime.Context, outputs *Outputs, pumps []vio.Puller) *Mux { return &Mux{ - rctx: rctx, - ch: ch, - parents: pullers, - nparents: len(parents), - debugger: debugger, - eosCh: chans.EOS, + rctx: rctx, + outputs: outputs, + pumps: pumps, + running: len(pumps), } } -// Pull implements the merge logic for returning data from the upstreams. +// Pull returns the next vector from any output. The done argument is ignored +// since the mux doesn't pull from the flowgraph: to stop a query early, cancel +// its context (as exec.Query.Pull does). func (m *Mux) Pull(bool) (vector.Any, error) { m.once.Do(func() { - for _, puller := range m.parents { - go puller.run(m.rctx.Context) - } - if m.debugger != nil { - m.debugger.run() + for _, p := range m.pumps { + go m.pump(m.rctx.Context, NewCatcher(p)) } }) - for m.nparents != 0 || m.debugger.active() { + for m.running != 0 { select { - case res := <-m.ch: + case res := <-m.outputs.ch: if res.err != nil { m.rctx.Cancel() return nil, res.err } - if res.vec == nil { - m.nparents-- - if m.nparents == 0 { - close(m.eosCh) - } + if !res.pumped { + return &vector.Labeled{Any: res.vec, Label: res.label}, nil } - return res.vector(), nil - case res := <-m.debugger.channel(): - m.debugger.check(res) - if res.err != nil { - m.rctx.Cancel() - return nil, res.err - } - if res.vec != nil { - return res.vector(), nil + m.running-- + if m.running == 0 { + m.ends = m.outputs.sortedLabels() } case <-m.rctx.Context.Done(): return nil, m.rctx.Context.Err() } } + if len(m.ends) != 0 { + label := m.ends[0] + m.ends = m.ends[1:] + return &vector.Labeled{Label: label}, nil + } m.rctx.Cancel() return nil, nil } -type Single struct { - vio.Puller - label string - eos bool -} - -func NewSingle(label string, parent vio.Puller) *Single { - return &Single{Puller: parent, label: label} -} - -func (s *Single) Pull(bool) (vector.Any, error) { - if s.eos { - return nil, nil +// pump pulls p until EOS or an error and then tells the mux that it's done. +// Since an output operator's send is synchronous, all of p's vectors have +// been received by the mux when p returns EOS. +func (m *Mux) pump(ctx context.Context, p vio.Puller) { + var err error + for { + var vec vector.Any + vec, err = p.Pull(false) + if vec == nil || err != nil { + break + } } - vec, err := s.Puller.Pull(false) - if vec == nil { - s.eos = true + select { + case m.outputs.ch <- muxresult{err: err, pumped: true}: + case <-ctx.Done(): } - return &vector.Labeled{Any: vec, Label: s.label}, err } func Unlabel(vec vector.Any) (vector.Any, string) { @@ -183,63 +184,3 @@ func Unlabel(vec vector.Any) (vector.Any, string) { } return vec, "" } - -type debugger struct { - threads []dthread - resultCh chan muxresult - nrun int -} - -func newDebugger(chans *DebugChans) *debugger { - var threads []dthread - resultCh := make(chan muxresult) - for _, ch := range chans.Debug { - threads = append(threads, dthread{ - vecCh: ch, - resultCh: resultCh, - }) - } - return &debugger{ - threads: threads, - resultCh: resultCh, - nrun: len(threads), - } -} - -func (d *debugger) active() bool { - return d != nil && d.nrun != 0 -} - -func (d *debugger) run() { - for _, t := range d.threads { - go t.run() - } -} - -func (d *debugger) check(r muxresult) { - if r.vec == nil { - d.nrun-- - } -} - -func (d *debugger) channel() <-chan muxresult { - if d == nil { - return nil - } - return d.resultCh -} - -type dthread struct { - vecCh <-chan vector.Any - resultCh chan<- muxresult -} - -func (d *dthread) run() { - for { - vec := <-d.vecCh - d.resultCh <- muxresult{vec, "debug", nil} - if vec == nil { - return - } - } -} diff --git a/runtime/op/mux_test.go b/runtime/op/mux_test.go new file mode 100644 index 0000000000..ecacd85a1d --- /dev/null +++ b/runtime/op/mux_test.go @@ -0,0 +1,107 @@ +package op + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/require" + "github.com/superdb/super" + "github.com/superdb/super/runtime" + "github.com/superdb/super/vector" + "github.com/superdb/super/vector/vio" +) + +// pullAll pulls mux until EOS and returns the number of values with each +// label and the labels of the channel ends. It fails if a value follows the +// end of its channel. +func pullAll(t *testing.T, mux *Mux) (map[string]int, []string) { + counts := map[string]int{} + var ends []string + for { + vec, err := mux.Pull(false) + require.NoError(t, err) + if vec == nil { + return counts, ends + } + labeled := vec.(*vector.Labeled) + if labeled.Any == nil { + ends = append(ends, labeled.Label) + continue + } + require.NotContains(t, ends, labeled.Label) + counts[labeled.Label]++ + } +} + +// sender sends a vector with each of labels from inside Pull, as an output +// operator in a subquery or a debug operator does, and then returns vec once. +type sender struct { + ctx context.Context + outputs *Outputs + labels []string + vec vector.Any + done bool +} + +func (s *sender) Pull(bool) (vector.Any, error) { + if s.done { + return nil, nil + } + s.done = true + for _, label := range s.labels { + if err := s.outputs.send(s.ctx, label, vector.NewNull(1)); err != nil { + return nil, err + } + } + return s.vec, nil +} + +func TestMux(t *testing.T) { + rctx := runtime.NewContext(context.Background(), super.NewContext()) + outputs := NewOutputs() + null := vector.NewNull(1) + // Two leaves output to main and one to foo, and a subquery in the + // second leaf outputs to main and bar and debugs. + subquery := &sender{ + ctx: rctx, + outputs: outputs, + labels: []string{"main", "bar", "debug", "main"}, + vec: null, + } + NewOutput(rctx, nil, outputs, "bar") + leaves := []vio.Puller{ + NewOutput(rctx, vio.NewPuller(null, null), outputs, "main"), + NewOutput(rctx, subquery, outputs, "main"), + NewOutput(rctx, vio.NewPuller(null), outputs, "foo"), + } + counts, ends := pullAll(t, NewMux(rctx, outputs, leaves)) + require.Equal(t, map[string]int{"main": 5, "foo": 1, "bar": 1, "debug": 1}, counts) + // The channels end in order and there's no end of channel for debug. + require.Equal(t, []string{"bar", "foo", "main"}, ends) + require.Error(t, rctx.Err()) +} + +type errPuller struct{} + +func (errPuller) Pull(bool) (vector.Any, error) { + return nil, errors.New("leaf failed") +} + +func TestMuxError(t *testing.T) { + rctx := runtime.NewContext(context.Background(), super.NewContext()) + outputs := NewOutputs() + leaves := []vio.Puller{NewOutput(rctx, errPuller{}, outputs, "main")} + _, err := NewMux(rctx, outputs, leaves).Pull(false) + require.EqualError(t, err, "leaf failed") + require.Error(t, rctx.Err()) +} + +func TestOutputsNil(t *testing.T) { + // A nil Outputs discards its values, as at compile time. + rctx := runtime.NewContext(context.Background(), super.NewContext()) + out := NewOutput(rctx, vio.NewPuller(vector.NewNull(1)), nil, "main") + vec, err := out.Pull(false) + require.NoError(t, err) + require.Nil(t, vec) +} diff --git a/runtime/op/output.go b/runtime/op/output.go index 647c95269c..b930965abc 100644 --- a/runtime/op/output.go +++ b/runtime/op/output.go @@ -1,30 +1,46 @@ package op import ( + "context" + "github.com/superdb/super/vector" "github.com/superdb/super/vector/vio" ) -// Output is the puller for an output of a query. It drops any plain none -// values, which come from an expression like ok() or ?. that handles a none -// (see dropPlainNones). A none of a named type is a value and is output -// as is. +// Output implements the output operator, which may appear in a leaf of a +// query's flowgraph or of a subquery's. It sends each vector from its parent +// to the query's outputs labeled with the output's name and, as a leaf, +// returns no values. In a leaf of a query, it's pulled by the mux (see Mux); +// in a subquery, it's pulled by the subquery's evaluation. It drops any plain +// none values, which come from an expression like ok() or ?. that handles a +// none (see dropPlainNones). A none of a named type is a value and is output +// as is. If outputs is nil (e.g., when an expression is evaluated at compile +// time), the values are discarded. type Output struct { - parent vio.Puller + ctx context.Context + parent vio.Puller + outputs *Outputs + label string } -func NewOutput(parent vio.Puller) *Output { - return &Output{parent: parent} +func NewOutput(ctx context.Context, parent vio.Puller, outputs *Outputs, label string) *Output { + outputs.addLabel(label) + return &Output{ctx: ctx, parent: parent, outputs: outputs, label: label} } func (o *Output) Pull(done bool) (vector.Any, error) { + if done { + return o.parent.Pull(true) + } for { - vec, err := o.parent.Pull(done) + vec, err := o.parent.Pull(false) if vec == nil || err != nil { - return vec, err + return nil, err } if vec = dropPlainNones(vec); vec != nil { - return vec, nil + if err := o.outputs.send(o.ctx, o.label, vec); err != nil { + return nil, err + } } } } diff --git a/runtime/ztests/op/debug-subquery.yaml b/runtime/ztests/op/debug-subquery.yaml new file mode 100644 index 0000000000..ff2f94178b --- /dev/null +++ b/runtime/ztests/op/debug-subquery.yaml @@ -0,0 +1,14 @@ +# A debug operator in a correlated subquery is pulled once per value. +script: | + super -s -c 'values 1,2,3 | values (values this+10 | debug f"debug {this}")' 2>&1 | + super -s -c 'sort this' - + +outputs: + - name: stdout + data: | + 11 + 12 + 13 + "debug 11" + "debug 12" + "debug 13" diff --git a/runtime/ztests/op/output-subquery.yaml b/runtime/ztests/op/output-subquery.yaml new file mode 100644 index 0000000000..265f8101c9 --- /dev/null +++ b/runtime/ztests/op/output-subquery.yaml @@ -0,0 +1,99 @@ +# An output operator in a subquery sends its values to the query's output. +# The values are sorted since the order of values from a subquery's output +# and from the query's other outputs is undefined. +script: | + for f in *.spq; do + echo "// $f" + super -s -I $f | super -s -c "sort this" - + done + echo "// leaf error" + ! super -s -c 'values (values 1 | output main | values 2)' + echo "// constant" + super -s -c 'const x = (values 1 | output main) values typeof(x)' + +inputs: + - name: correlated.spq + data: | + values 1,2,3 | values (values this+10 | output main) + - name: fork.spq + data: | + values 1,2 + | values ( + fork + (values f"sent {this}" | output main) + (values this*10) + ) + - name: nested.spq + data: | + values 1,2 | values (values (values this+100 | output main) | output main) + - name: unnest-into.spq + data: | + values [1,2],[3] | unnest this into (output main) + - name: array.spq + data: | + values [1,2] | values [unnest this | output main], "after" + - name: recursive.spq + data: | + fn f(n): ( + case when n<=0 then 0 + else (fork (values n | output main) (values f(n-1))) + end + ) + values f(3) + - name: per-row.spq + data: | + values 1,2 | values {x:this,y:(values 7 | output main)} + # A subquery that reads data is uncorrelated and evaluated once. + - name: uncorrelated.spq + data: | + values 1,2 | values {x:this,y:(from seven.sup | fork (output main) (values a+1))} + - name: seven.sup + data: | + {a:7} + +outputs: + - name: stdout + data: | + // array.spq + 1 + 2 + "after" + [] + // correlated.spq + 11 + 12 + 13 + // fork.spq + 10 + 20 + "sent 1" + "sent 2" + // nested.spq + 101 + 102 + // per-row.spq + 7 + 7 + {x:1} + {x:2} + // recursive.spq + 0 + 1 + 2 + 3 + // uncorrelated.spq + {a:7} + {x:1,y:8} + {x:2,y:8} + // unnest-into.spq + 1 + 2 + 3 + // leaf error + // constant + + - name: stderr + data: | + output operator must be a flowgraph leaf at line 1, column 20: + values (values 1 | output main | values 2) + ~~~~~~~~~~~ From 770a17533154092c622ceaee24bd032b5a5de05c Mon Sep 17 00:00:00 2001 From: Steven McCanne Date: Tue, 10 Nov 2026 17:59:29 -0800 Subject: [PATCH 2/2] deep review --- book/src/SUMMARY.md | 1 + book/src/super-sql/operators/output.md | 90 ++++++++++++++++++++++++++ compiler/semantic/evaluator.go | 4 +- runtime/op/mux.go | 7 -- runtime/ztests/op/output-subquery.yaml | 6 +- 5 files changed, 98 insertions(+), 10 deletions(-) create mode 100644 book/src/super-sql/operators/output.md diff --git a/book/src/SUMMARY.md b/book/src/SUMMARY.md index 1d47e9ce03..0026c71eba 100644 --- a/book/src/SUMMARY.md +++ b/book/src/SUMMARY.md @@ -75,6 +75,7 @@ - [infer](super-sql/operators/infer.md) - [join](super-sql/operators/join.md) - [load](super-sql/operators/load.md) + - [output](super-sql/operators/output.md) - [pass](super-sql/operators/pass.md) - [put](super-sql/operators/put.md) - [rename](super-sql/operators/rename.md) diff --git a/book/src/super-sql/operators/output.md b/book/src/super-sql/operators/output.md new file mode 100644 index 0000000000..bb86183447 --- /dev/null +++ b/book/src/super-sql/operators/output.md @@ -0,0 +1,90 @@ +# output + +[✅](../intro.md#data-order)  send values to a named output + +## Synopsis + +``` +output +``` + +## Description + +The `output` operator sends its input values to the query's output named +``, where `` is an [identifier](../queries.md#identifiers). + +A query's results are delivered on one or more named outputs. +Each leaf of a query's flowgraph that doesn't end in an `output` operator +implicitly ends in `output main`, so the results of most queries are delivered +on the output named `main`. An `output` operator may only appear at a leaf +of a flowgraph, e.g., at the end of a query or at the end of a path of a +[fork](fork.md) or [switch](switch.md). + +An `output` operator may also appear at a leaf of a +[subquery](../expressions/subqueries.md), including the body of +[unnest](unnest.md) `into`. In this case, its values are sent to the +query's output, not to the subquery's result, so the leaf contributes no +values to the subquery. A subquery whose leaves all end in `output` produces +no values. Since a constant expression is evaluated before the query runs, +an `output` operator can't appear in one (e.g., in a +[const declaration](../declarations/constants.md)). + +As for the [values](values.md) operator, a plain `none` (e.g., from +[ok](../functions/generics/ok.md) or `?.`) is dropped. + +The values of a single `output` operator are output in order, but the order of +values from different `output` operators, whether in the query's flowgraph +or in a subquery, is undefined. + +When running a query with the [super](../../command/super.md) command, +the output named `main` is written to the command's output +and the output named `debug` is printed to standard error (as for the +[debug](debug.md) operator). Other outputs are discarded. + +When running a query in a database service, each output is transmitted to the +client on the output channel of the same name. + +## Examples + +--- + +_Send values to the main output from a subquery_ +```mdtest-spq +# spq +values (values this*10 | output main) +# input +1 +2 +# expected output +10 +20 +``` + +--- + +_Output the sum of each array_ +```mdtest-spq +# spq +unnest this into (sum(this) | output main) +# input +[1,2] +[3,4] +# expected output +3 +7 +``` + +--- + +_The super command shows only the main output_ +```mdtest-spq +# spq +switch + case this > 2 ( output main ) + case true ( output small ) +# input +1 +3 +# expected output +3 +``` diff --git a/compiler/semantic/evaluator.go b/compiler/semantic/evaluator.go index 1906b7bafb..905ebd47d5 100644 --- a/compiler/semantic/evaluator.go +++ b/compiler/semantic/evaluator.go @@ -143,7 +143,9 @@ func (e *evaluator) op(op sem.Op) bool { // Like join, if any of the parent legs is non-const, the constThis if false here return e.sortExprs(op.Exprs) && e.constThis case *sem.OutputOp: - return true + // There's no mux at compile time to receive the values. + e.errs.error(op, errors.New("output operator cannot be used in a constant expression")) + return false case *sem.PutOp: return e.assignments(op.Args) && e.constThis case *sem.RenameOp: diff --git a/runtime/op/mux.go b/runtime/op/mux.go index cda05de811..7db73981e0 100644 --- a/runtime/op/mux.go +++ b/runtime/op/mux.go @@ -177,10 +177,3 @@ func (m *Mux) pump(ctx context.Context, p vio.Puller) { case <-ctx.Done(): } } - -func Unlabel(vec vector.Any) (vector.Any, string) { - if vec, ok := vec.(*vector.Labeled); ok { - return vec.Any, vec.Label - } - return vec, "" -} diff --git a/runtime/ztests/op/output-subquery.yaml b/runtime/ztests/op/output-subquery.yaml index 265f8101c9..98edc2fa5b 100644 --- a/runtime/ztests/op/output-subquery.yaml +++ b/runtime/ztests/op/output-subquery.yaml @@ -9,7 +9,7 @@ script: | echo "// leaf error" ! super -s -c 'values (values 1 | output main | values 2)' echo "// constant" - super -s -c 'const x = (values 1 | output main) values typeof(x)' + ! super -s -c 'const x = (values 1 | output main) values x' inputs: - name: correlated.spq @@ -91,9 +91,11 @@ outputs: 3 // leaf error // constant - - name: stderr data: | output operator must be a flowgraph leaf at line 1, column 20: values (values 1 | output main | values 2) ~~~~~~~~~~~ + output operator cannot be used in a constant expression at line 1, column 23: + const x = (values 1 | output main) values x + ~~~~~~~~~~~