optimizer: propagate sort keys through ValuesOp

compiler/optimizer/op.go

@@ -2,6 +2,7 @@ package optimizer
 
 import (
 	"errors"
+	"slices"
 	"strings"
 
 	"github.com/superdb/super/compiler/dag"
@@ -43,7 +44,20 @@ func (o *Optimizer) analyzeSortKeys(op dag.Op, in order.SortKeys) (order.SortKey
 	case *dag.FilterOp, *dag.HeadOp, *dag.InferOp, *dag.PassOp, *dag.UniqOp, *dag.TailOp, *dag.FuseOp, *dag.OutputOp:
 		return in, nil
 	case *dag.CutOp:
-		return analyzeCuts(op.Args, in), nil
+		return analyzeAssignments(op.Args, in), nil
+	case *dag.ValuesOp:
+		if len(op.Exprs) != 1 {
+			return nil, nil
+		}
+		rec, ok := op.Exprs[0].(*dag.RecordExpr)
+		if !ok {
+			return nil, nil
+		}
+		assignments, ok := recordAssignments(rec, nil, nil)
+		if !ok {
+			return nil, nil
+		}
+		return analyzeAssignments(assignments, in), nil
 	case *dag.DropOp:
 		for _, f := range op.Args {
 			if fieldOf(f).Equal(key.Path) {
@@ -133,7 +147,7 @@ func orderPreservingCall(e dag.Expr, key field.Path) bool {
 	return false
 }
 
-func analyzeCuts(assignments []dag.Assignment, sortKeys order.SortKeys) order.SortKeys {
+func analyzeAssignments(assignments []dag.Assignment, sortKeys order.SortKeys) order.SortKeys {
 	if sortKeys.IsNil() {
 		return nil
 	}
@@ -206,6 +220,30 @@ func analyzeCuts(assignments []dag.Assignment, sortKeys order.SortKeys) order.So
 	panic("unreachable")
 }
 
+// recordAssignments converts a record into an equivalent list of assignments.
+// It returns false if rec contains a spread.
+func recordAssignments(rec *dag.RecordExpr, path field.Path, out []dag.Assignment) ([]dag.Assignment, bool) {
+	for _, elem := range rec.Elems {
+		f, ok := elem.(*dag.Field)
+		if !ok {
+			return nil, false
+		}
+		p := append(slices.Clone(path), f.Name)
+		if child, ok := f.Value.(*dag.RecordExpr); ok {
+			if out, ok = recordAssignments(child, p, out); !ok {
+				return nil, false
+			}
+			continue
+		}
+		out = append(out, dag.Assignment{
+			Kind: "Assignment",
+			LHS:  dag.NewThis(p.Chain()),
+			RHS:  f.Value,
+		})
+	}
+	return out, true
+}
+
 func fieldKey(f field.Path) string {
 	return strings.Join(f, "\x00")
 }

compiler/optimizer/parallelize.go

@@ -217,7 +217,7 @@ func (o *Optimizer) liftIntoParPaths(seq dag.Seq) {
 		for k := range paths {
 			paths[k].Append(dag.CopyOp(op))
 		}
-	case *dag.CutOp, *dag.DropOp, *dag.FilterOp, *dag.PutOp, *dag.RenameOp:
+	case *dag.CutOp, *dag.DropOp, *dag.FilterOp, *dag.PutOp, *dag.RenameOp, *dag.ValuesOp:
 		if merge != nil {
 			// See if this op would disrupt the merge-on key
 			mergeKey, err := o.propagateSortKeyOp(merge, []order.SortKeys{nil})

compiler/ztests/par-values.yaml

@@ -0,0 +1,104 @@
+# This test checks that a values operator with a single record expression
+# preserves a known sort order through the fields it passes through unchanged
+# so that it can be lifted into parallel paths ahead of the merge.
+script: |
+  super db init -q test
+  super db attach -q test
+  super db create -q -orderby ts test
+  super db compile -C -P 2 "from test | values {ts:ts,x:y}" | sed -e 's/table .*/.../'
+  echo // ===
+  super db compile -C -P 2 "from test | values {ts:ts,x:y} | sort ts" | sed -e 's/table .*/.../'
+  echo // ===
+  super db compile -C -P 2 "select ts, x from test" | sed -e 's/table .*/.../'
+  echo // ===
+  super db compile -C -P 2 "from test | values {y:ts} | sort y" | sed -e 's/table .*/.../'
+  echo // ===
+  super db compile -C -P 2 "from test | values {...this,x:1}" | sed -e 's/table .*/.../'
+  echo // ===
+  super db compile -C -P 2 "from test | values {ts:ts}, {ts:ts}" | sed -e 's/table .*/.../'
+
+outputs:
+  - name: stdout
+    data: |
+      lister ...
+      | slicer
+      | scatter
+        (
+          seqscan ...
+          | values {ts:ts,x:y}
+        )
+        (
+          seqscan ...
+          | values {ts:ts,x:y}
+        )
+      | merge ts asc nulls last
+      | output main
+      // ===
+      lister ...
+      | scatter
+        (
+          seqscan ...
+          | values {ts:ts,x:y}
+          | sort ts asc nulls last
+        )
+        (
+          seqscan ...
+          | values {ts:ts,x:y}
+          | sort ts asc nulls last
+        )
+      | merge ts asc nulls last
+      | output main
+      // ===
+      lister ...
+      | slicer
+      | scatter
+        (
+          seqscan ...
+          | values {ts:this!.ts,x:this!.x}
+        )
+        (
+          seqscan ...
+          | values {ts:this!.ts,x:this!.x}
+        )
+      | merge ts asc nulls last
+      | output main
+      // ===
+      lister ...
+      | slicer
+      | scatter
+        (
+          seqscan ...
+        )
+        (
+          seqscan ...
+        )
+      | merge ts asc nulls last
+      | values {y:ts}
+      | sort y asc nulls last
+      | output main
+      // ===
+      lister ...
+      | slicer
+      | scatter
+        (
+          seqscan ...
+        )
+        (
+          seqscan ...
+        )
+      | merge ts asc nulls last
+      | values {...this,x:1}
+      | output main
+      // ===
+      lister ...
+      | slicer
+      | scatter
+        (
+          seqscan ...
+        )
+        (
+          seqscan ...
+        )
+      | merge ts asc nulls last
+      | values {ts:ts}, {ts:ts}
+      | output main