rework runtime output model

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)

book/src/super-sql/operators/output.md

@@ -0,0 +1,90 @@
+# output
+
+[✅](../intro.md#data-order)  send values to a named output
+
+## Synopsis
+
+```
+output <name>
+```
+
+## Description
+
+The `output` operator sends its input values to the query's output named
+`<name>`, where `<name>` 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
+```

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

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

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
 	}

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:

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"})
 		}
 	}

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: