Add native vector builder (#6812)

This commit adds a native vector builder to the vector package that
concats vectors of the same type together. The previous Builder is
renamed to ValueBuilder. The interface for vector aggregates is also
changed to return vectors instead of scalar values and the vector
runtime aggregate uses the new Builder to concat these results
together.

Author: mattnibs

csup/csup_test.go

@@ -66,7 +66,7 @@ func TestCSUPBatchBug(t *testing.T) {
 }
 
 func valToVec(sctx *super.Context, val super.Value) vector.Any {
-	b := vector.NewDynamicBuilder()
+	b := vector.NewDynamicValueBuilder()
 	b.Write(val)
 	return b.Build(sctx)
 }

csup/object_test.go

@@ -22,7 +22,7 @@ func TestObjectProjectMetadata(t *testing.T) {
 		"{a:2,b:{c:5,d:0.8}}",
 		"{a:3,b:{c:6,d:0.9}}",
 	}
-	builder := vector.NewDynamicBuilder()
+	builder := vector.NewDynamicValueBuilder()
 	for _, s := range supValues {
 		builder.Write(sup.MustParseValue(sctx, s))
 	}

csup/writer.go

@@ -103,7 +103,7 @@ func (w *Serializer) finalizeObject() error {
 type ValWriter struct {
 	sctx       *super.Context
 	serializer *Serializer
-	builder    *vector.DynamicBuilder
+	builder    *vector.DynamicValueBuilder
 }
 
 var _ sio.Writer = (*ValWriter)(nil)
@@ -113,7 +113,7 @@ func NewValWriter(w io.WriteCloser) *ValWriter {
 	return &ValWriter{
 		sctx:       sctx,
 		serializer: NewSerializer(w),
-		builder:    vector.NewDynamicBuilder(),
+		builder:    vector.NewDynamicValueBuilder(),
 	}
 }
 

db/writer.go

@@ -217,7 +217,7 @@ again:
 		// XXX TBD: this is slow and creates a vector per value when writing vectors
 		// to a database.  This will change when we convert the database from
 		// mixed BSUP/CSUP to CSUP only.
-		builder := vector.NewBuilder(val.Type())
+		builder := vector.NewValueBuilder(val.Type())
 		builder.Write(val.Bytes())
 		if err := w.vectorWriter.Push(builder.Build(w.sctx)); err != nil {
 			w.Abort()

runtime/sam/op/debug/op.go

@@ -65,7 +65,7 @@ func (o *Op) Pull(done bool) (sbuf.Batch, error) {
 }
 
 func (o *Op) evalBatch(in sbuf.Batch) vector.Any {
-	builder := vector.NewDynamicBuilder()
+	builder := vector.NewDynamicValueBuilder()
 	for _, x := range in.Values() {
 		if o.filter == nil || o.where(x) {
 			builder.Write(o.expr.Eval(x))

runtime/vam/expr/agg/agg.go

@@ -10,8 +10,8 @@ import (
 type Func interface {
 	Consume(vector.Any)
 	ConsumeAsPartial(vector.Any)
-	Result(*super.Context) super.Value
-	ResultAsPartial(*super.Context) super.Value
+	Result(*super.Context) vector.Any
+	ResultAsPartial(*super.Context) vector.Any
 }
 
 type Pattern func() Func

runtime/vam/expr/agg/any.go

@@ -2,35 +2,35 @@ package agg
 
 import (
 	"github.com/brimdata/super"
-	"github.com/brimdata/super/scode"
 	"github.com/brimdata/super/vector"
 )
 
 type Any struct {
-	val super.Value
+	result vector.Any
 }
 
 func NewAny() *Any {
-	return &Any{val: super.Null}
+	return &Any{}
 }
 
 func (a *Any) Consume(vec vector.Any) {
-	if !a.val.IsNull() || vec.Kind() == vector.KindNull {
+	if a.result != nil || vec.Kind() == vector.KindNull {
 		return
 	}
-	var b scode.Builder
-	vec.Serialize(&b, 0)
-	a.val = super.NewValue(vec.Type(), b.Bytes().Body())
+	a.result = vector.Pick(vec, []uint32{0})
 }
 
 func (a *Any) ConsumeAsPartial(vec vector.Any) {
 	a.Consume(vec)
 }
 
-func (a *Any) Result(*super.Context) super.Value {
-	return a.val
+func (a *Any) Result(sctx *super.Context) vector.Any {
+	if a.result == nil {
+		return vector.NewNull(1)
+	}
+	return a.result
 }
 
-func (a *Any) ResultAsPartial(*super.Context) super.Value {
+func (a *Any) ResultAsPartial(*super.Context) vector.Any {
 	return a.Result(nil)
 }

runtime/vam/expr/agg/avg.go

@@ -2,7 +2,6 @@ package agg
 
 import (
 	"github.com/brimdata/super"
-	"github.com/brimdata/super/scode"
 	"github.com/brimdata/super/vector"
 )
 
@@ -22,11 +21,12 @@ func (a *avg) Consume(vec vector.Any) {
 	a.sum = sum(a.sum, vec)
 }
 
-func (a *avg) Result(*super.Context) super.Value {
+func (a *avg) Result(*super.Context) vector.Any {
 	if a.count > 0 {
-		return super.NewFloat64(a.sum / float64(a.count))
+		f := a.sum / float64(a.count)
+		return vector.NewFloat(super.TypeFloat64, []float64{f})
 	}
-	return super.Null
+	return vector.NewNull(1)
 }
 
 const (
@@ -62,13 +62,12 @@ func (a *avg) ConsumeAsPartial(partial vector.Any) {
 	a.count += vector.UintValue(countVal, idx)
 }
 
-func (a *avg) ResultAsPartial(sctx *super.Context) super.Value {
-	var b scode.Builder
-	b.Append(super.EncodeFloat64(a.sum))
-	b.Append(super.EncodeUint(a.count))
+func (a *avg) ResultAsPartial(sctx *super.Context) vector.Any {
+	sum := vector.NewFloat(super.TypeFloat64, []float64{a.sum})
+	count := vector.NewUint(super.TypeUint64, []uint64{a.count})
 	typ := sctx.MustLookupTypeRecord([]super.Field{
 		super.NewField(sumName, super.TypeFloat64),
 		super.NewField(countName, super.TypeUint64),
 	})
-	return super.NewValue(typ, b.Bytes())
+	return vector.NewRecord(typ, []vector.Any{sum, count}, 1)
 }

runtime/vam/expr/agg/collect.go

@@ -3,6 +3,7 @@ package agg
 import (
 	"github.com/brimdata/super"
 	samagg "github.com/brimdata/super/runtime/sam/expr/agg"
+	"github.com/brimdata/super/sbuf"
 	"github.com/brimdata/super/scode"
 	"github.com/brimdata/super/vector"
 )
@@ -21,8 +22,9 @@ func (c *collect) Consume(vec vector.Any) {
 	}
 }
 
-func (c *collect) Result(sctx *super.Context) super.Value {
-	return c.samcollect.Result(sctx)
+func (c *collect) Result(sctx *super.Context) vector.Any {
+	val := c.samcollect.Result(sctx)
+	return sbuf.Dematerialize(sctx, sbuf.NewArray([]super.Value{val}))
 }
 
 func (c *collect) ConsumeAsPartial(partial vector.Any) {
@@ -53,6 +55,7 @@ func (c *collect) ConsumeAsPartial(partial vector.Any) {
 	}
 }
 
-func (c *collect) ResultAsPartial(sctx *super.Context) super.Value {
-	return c.samcollect.ResultAsPartial(sctx)
+func (c *collect) ResultAsPartial(sctx *super.Context) vector.Any {
+	val := c.samcollect.ResultAsPartial(sctx)
+	return sbuf.Dematerialize(sctx, sbuf.NewArray([]super.Value{val}))
 }

runtime/vam/expr/agg/collectmap.go

@@ -3,6 +3,7 @@ package agg
 import (
 	"github.com/brimdata/super"
 	samagg "github.com/brimdata/super/runtime/sam/expr/agg"
+	"github.com/brimdata/super/sbuf"
 	"github.com/brimdata/super/scode"
 	"github.com/brimdata/super/vector"
 )
@@ -28,14 +29,16 @@ func (c *collectMap) Consume(vec vector.Any) {
 	}
 }
 
-func (c *collectMap) Result(sctx *super.Context) super.Value {
-	return c.samCollectMap.Result(sctx)
+func (c *collectMap) Result(sctx *super.Context) vector.Any {
+	val := c.samCollectMap.Result(sctx)
+	return sbuf.Dematerialize(sctx, sbuf.NewArray([]super.Value{val}))
 }
 
 func (c *collectMap) ConsumeAsPartial(partial vector.Any) {
 	c.Consume(partial)
 }
 
-func (c *collectMap) ResultAsPartial(sctx *super.Context) super.Value {
-	return c.samCollectMap.ResultAsPartial(sctx)
+func (c *collectMap) ResultAsPartial(sctx *super.Context) vector.Any {
+	val := c.samCollectMap.ResultAsPartial(sctx)
+	return sbuf.Dematerialize(sctx, sbuf.NewArray([]super.Value{val}))
 }