diff --git a/book/src/SUMMARY.md b/book/src/SUMMARY.md index 1d47e9ce0..0026c71eb 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 000000000..bb8618344 --- /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/compiler.go b/compiler/compiler.go index c109eeaa4..e821f0524 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 5e0b69654..41dc946d7 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 1a35c0c8f..629c0c60c 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 b17c9ceaf..9dde89c77 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 ff561d8ee..899a594e5 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/compiler/semantic/evaluator.go b/compiler/semantic/evaluator.go index 1906b7baf..905ebd47d 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/debug.go b/runtime/op/debug.go index c159b7e4a..3e07c9cea 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 393bf0afd..7db73981e 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,217 +30,150 @@ 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) - } - } - 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.pumped { + return &vector.Labeled{Any: res.vec, Label: res.label}, nil } - 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 - } - vec, err := s.Puller.Pull(false) - if vec == nil { - s.eos = true - } - return &vector.Labeled{Any: vec, Label: s.label}, err -} - -func Unlabel(vec vector.Any) (vector.Any, string) { - if vec, ok := vec.(*vector.Labeled); ok { - return vec.Any, vec.Label - } - 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() { +// 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 { - vec := <-d.vecCh - d.resultCh <- muxresult{vec, "debug", nil} - if vec == nil { - return + var vec vector.Any + vec, err = p.Pull(false) + if vec == nil || err != nil { + break } } + select { + case m.outputs.ch <- muxresult{err: err, pumped: true}: + case <-ctx.Done(): + } } diff --git a/runtime/op/mux_test.go b/runtime/op/mux_test.go new file mode 100644 index 000000000..ecacd85a1 --- /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 647c95269..b930965ab 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 000000000..ff2f94178 --- /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 000000000..98edc2fa5 --- /dev/null +++ b/runtime/ztests/op/output-subquery.yaml @@ -0,0 +1,101 @@ +# 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 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) + ~~~~~~~~~~~ + output operator cannot be used in a constant expression at line 1, column 23: + const x = (values 1 | output main) values x + ~~~~~~~~~~~