Count operator

This commit adds the count operator an operator that outputs the current
count of values into a record expression.

Author: mattnibs

compiler/ast/op.go

@@ -60,6 +60,11 @@ type (
 		Args []Expr `json:"args"`
 		Loc  `json:"loc"`
 	}
+	CountOp struct {
+		Kind string `json:"kind" unpack:""`
+		Expr Expr   `json:"expr"`
+		Loc  `json:"loc"`
+	}
 	CutOp struct {
 		Kind string      `json:"kind" unpack:""`
 		Args Assignments `json:"args"`
@@ -348,6 +353,7 @@ func (*AggregateOp) opNode()  {}
 func (*AssertOp) opNode()     {}
 func (*AssignmentOp) opNode() {}
 func (*CallOp) opNode()       {}
+func (*CountOp) opNode()      {}
 func (*CutOp) opNode()        {}
 func (*DebugOp) opNode()      {}
 func (*Delete) opNode()       {}

compiler/dag/op.go

@@ -63,6 +63,11 @@ type (
 	CombineOp struct {
 		Kind string `json:"kind" unpack:""`
 	}
+	CountOp struct {
+		Kind  string `json:"kind" unpack:""`
+		Alias string `json:"alias"`
+		Expr  Expr   `json:"expr"`
+	}
 	CutOp struct {
 		Kind string       `json:"kind" unpack:""`
 		Args []Assignment `json:"args"`
@@ -205,6 +210,7 @@ type (
 
 func (*AggregateOp) opNode() {}
 func (*CombineOp) opNode()   {}
+func (*CountOp) opNode()     {}
 func (*CutOp) opNode()       {}
 func (*DistinctOp) opNode()  {}
 func (*DropOp) opNode()      {}

compiler/optimizer/demand.go

@@ -72,6 +72,8 @@ func demandForSimpleOp(op dag.Op, downstream demand.Demand) demand.Demand {
 		return d
 	case *dag.CombineOp:
 		return downstream
+	case *dag.CountOp:
+		return demand.Union(downstream, demandForExpr(op.Expr))
 	case *dag.CutOp:
 		return demandForAssignments(op.Args, demand.None())
 	case *dag.DistinctOp:

compiler/optimizer/parallelize.go

@@ -277,6 +277,8 @@ func (o *Optimizer) concurrentPath(seq dag.Seq, sortKeys order.SortKeys) (length
 				return k, sortExprsForSortKeys(sortKeys), true, nil
 			}
 			return k, nil, false, nil
+		case *dag.CountOp:
+			return k, nil, true, nil
 		case *dag.SortOp:
 			if len(op.Exprs) == 0 {
 				// No analysis for sort without expression since we can't