runtime: convert sam/op/meta contents to vio.Puller

compiler/compiler.go

@@ -14,7 +14,6 @@ import (
 	"github.com/superdb/super/runtime"
 	"github.com/superdb/super/runtime/exec"
 	"github.com/superdb/super/runtime/op"
-	"github.com/superdb/super/runtime/sam/op/meta"
 	"github.com/superdb/super/vector/vio"
 )
 
@@ -76,8 +75,8 @@ func (l *compiler) NewObjectScanner(rctx *runtime.Context, poolID uuid.UUID, obj
 	if err != nil {
 		return nil, err
 	}
-	lister := meta.NewListerFromObjects(rctx, rctx.Sctx, pool, objects, nil)
-	slicer := meta.NewSlicer(lister, rctx.Sctx)
+	lister := op.NewListerFromObjects(rctx, rctx.Sctx, pool, objects, nil)
+	slicer := op.NewSlicer(rctx.Sctx, lister)
 	return op.NewPoolScanner(rctx, slicer, pool, nil, nil, nil), nil
 }
 

compiler/rungen/op.go

@@ -17,7 +17,6 @@ import (
 	"github.com/superdb/super/runtime/expr"
 	"github.com/superdb/super/runtime/op"
 	samexpr "github.com/superdb/super/runtime/sam/expr"
-	"github.com/superdb/super/runtime/sam/op/meta"
 	"github.com/superdb/super/runtime/sam/op/top"
 	"github.com/superdb/super/runtime/sam/op/uniq"
 	"github.com/superdb/super/sbuf"
@@ -109,46 +108,6 @@ func (b *Builder) lookupType(id int) (super.Type, error) {
 
 func (b *Builder) compileLeaf(o dag.Op, parent sbuf.Puller) (sbuf.Puller, error) {
 	switch v := o.(type) {
-	//
-	// Scanners in alphatbetical order.
-	//
-	case *dag.CommitMetaScan:
-		var pruner samexpr.Evaluator
-		if v.Tap && v.KeyPruner != nil {
-			var err error
-			pruner, err = compileExpr(v.KeyPruner)
-			if err != nil {
-				return nil, err
-			}
-		}
-		return meta.NewCommitMetaScanner(b.rctx.Context, b.sctx(), b.env.DB(), v.Pool, v.Commit, v.Meta, pruner)
-	case *dag.DBMetaScan:
-		return meta.NewDBMetaScanner(b.rctx.Context, b.sctx(), b.env.DB(), v.Meta)
-	case *dag.ListerScan:
-		if parent != nil {
-			return nil, errors.New("internal error: data source cannot have a parent operator")
-		}
-		pool, err := b.lookupPool(v.Pool)
-		if err != nil {
-			return nil, err
-		}
-		var pruner samexpr.Evaluator
-		if v.KeyPruner != nil {
-			pruner, err = compileExpr(v.KeyPruner)
-			if err != nil {
-				return nil, err
-			}
-		}
-		return meta.NewLister(b.rctx.Context, b.mctx, pool, v.Commit, pruner)
-	case *dag.NullScan:
-		return sbuf.NewPuller(sbuf.NewArray([]super.Value{super.Null})), nil
-	case *dag.PoolMetaScan:
-		return meta.NewPoolMetaScanner(b.rctx.Context, b.sctx(), b.env.DB(), v.ID, v.Meta)
-	case *dag.SlicerOp:
-		return meta.NewSlicer(parent, b.mctx), nil
-	//
-	// Non-scanner operators in alphabetical order.
-	//
 	case *dag.TopOp:
 		exprs, err := b.compileSortExprs(v.Exprs)
 		if err != nil {

compiler/rungen/vop.go

@@ -15,7 +15,6 @@ import (
 	"github.com/superdb/super/runtime/op"
 	"github.com/superdb/super/runtime/op/aggregate"
 	samexpr "github.com/superdb/super/runtime/sam/expr"
-	"github.com/superdb/super/runtime/sam/op/meta"
 	"github.com/superdb/super/sbuf"
 	"github.com/superdb/super/vector"
 	"github.com/superdb/super/vector/vio"
@@ -190,6 +189,16 @@ func (b *Builder) compileVamLeaf(o dag.Op, parent vio.Puller) (vio.Puller, error
 	switch o := o.(type) {
 	case *dag.AggregateOp:
 		return b.compileVamAggregate(o, parent)
+	case *dag.CommitMetaScan:
+		var pruner samexpr.Evaluator
+		if o.Tap && o.KeyPruner != nil {
+			var err error
+			pruner, err = compileExpr(o.KeyPruner)
+			if err != nil {
+				return nil, err
+			}
+		}
+		return op.NewCommitMetaScanner(b.rctx.Context, b.sctx(), b.env.DB(), o.Pool, o.Commit, o.Meta, pruner)
 	case *dag.CountOp:
 		var e expr.Evaluator
 		if o.Expr != nil {
@@ -209,6 +218,8 @@ func (b *Builder) compileVamLeaf(o dag.Op, parent vio.Puller) (vio.Puller, error
 			return nil, err
 		}
 		return op.NewValues(b.sctx(), parent, []expr.Evaluator{e}), nil
+	case *dag.DBMetaScan:
+		return op.NewDBMetaScanner(b.rctx.Context, b.sctx(), b.env.DB(), o.Meta)
 	case *dag.DebugOp:
 		e, err := b.compileVamExpr(o.Expr)
 		if err != nil {
@@ -283,13 +294,33 @@ func (b *Builder) compileVamLeaf(o dag.Op, parent vio.Puller) (vio.Puller, error
 		return op.NewHead(parent, o.Count), nil
 	case *dag.InferOp:
 		return op.NewInfer(b.rctx, parent, o.Limit), nil
+	case *dag.ListerScan:
+		if parent != nil {
+			return nil, errors.New("internal error: data source cannot have a parent operator")
+		}
+		pool, err := b.lookupPool(o.Pool)
+		if err != nil {
+			return nil, err
+		}
+		var pruner samexpr.Evaluator
+		if o.KeyPruner != nil {
+			pruner, err = compileExpr(o.KeyPruner)
+			if err != nil {
+				return nil, err
+			}
+		}
+		return op.NewLister(b.rctx.Context, b.mctx, pool, o.Commit, pruner)
 	case *dag.LoadOp:
 		return op.NewLoad(b.rctx, b.env.DB(), parent, o.Pool, o.Branch, o.Author, o.Message, o.Meta), nil
+	case *dag.NullScan:
+		return vio.NewPuller(vector.NewNull(1)), nil
 	case *dag.OutputOp:
 		b.channels[o.Name] = append(b.channels[o.Name], parent)
 		return parent, nil
 	case *dag.PassOp:
 		return parent, nil
+	case *dag.PoolMetaScan:
+		return op.NewPoolMetaScanner(b.rctx.Context, b.sctx(), b.env.DB(), o.ID, o.Meta)
 	case *dag.PutOp:
 		rec, err := newRecordExprFromAssignments(o.Args)
 		if err != nil {
@@ -326,11 +357,11 @@ func (b *Builder) compileVamLeaf(o dag.Op, parent vio.Puller) (vio.Puller, error
 		if err != nil {
 			return nil, err
 		}
-		l, err := meta.NewLister(b.rctx.Context, b.mctx, pool, o.Commit, nil)
+		l, err := op.NewLister(b.rctx.Context, b.mctx, pool, o.Commit, nil)
 		if err != nil {
 			return nil, err
 		}
-		slicer := meta.NewSlicer(l, b.mctx)
+		slicer := op.NewSlicer(b.mctx, l)
 		return op.NewPoolScanner(b.rctx, slicer, pool, nil, nil, b.progress), nil
 	case *dag.SeqScan:
 		pool, err := b.lookupPool(o.Pool)
@@ -351,9 +382,11 @@ func (b *Builder) compileVamLeaf(o dag.Op, parent vio.Puller) (vio.Puller, error
 				return nil, err
 			}
 		}
-		return op.NewPoolScanner(b.rctx, sbuf.NewMaterializer(parent), pool, filter, pruner, b.progress), nil
+		return op.NewPoolScanner(b.rctx, parent, pool, filter, pruner, b.progress), nil
 	case *dag.SkipOp:
 		return op.NewSkip(parent, o.Count), nil
+	case *dag.SlicerOp:
+		return op.NewSlicer(b.mctx, parent), nil
 	case *dag.SortOp:
 		exprs, err := b.compileSortExprs(o.Exprs)
 		if err != nil {

db/data/partition.go

@@ -0,0 +1,31 @@
+package data
+
+import (
+	"fmt"
+
+	"github.com/superdb/super"
+	"github.com/superdb/super/sup"
+)
+
+// A Partition is a logical view of the records within a pool-key span, stored
+// in one or more data objects.  This provides a way to return the list of
+// objects that should be scanned along with a span to limit the scan
+// to only the span involved.
+type Partition struct {
+	Min     super.Value `super:"min"`
+	Max     super.Value `super:"max"`
+	Objects []*Object   `super:"objects"`
+}
+
+func (p Partition) IsZero() bool {
+	return p.Objects == nil
+}
+
+func (p Partition) FormatRangeOf(index int) string {
+	o := p.Objects[index]
+	return fmt.Sprintf("[%s-%s,%s-%s]", sup.String(p.Min), sup.String(p.Max), sup.String(o.Min), sup.String(o.Max))
+}
+
+func (p Partition) FormatRange() string {
+	return fmt.Sprintf("[%s-%s]", sup.String(p.Min), sup.String(p.Max))
+}

runtime/sam/op/meta/lister.go → runtime/op/lister.go

@@ -1,4 +1,4 @@
-package meta
+package op
 
 import (
 	"bytes"
@@ -13,6 +13,7 @@ import (
 	"github.com/superdb/super/order"
 	"github.com/superdb/super/runtime/sam/expr"
 	"github.com/superdb/super/sbuf"
+	"github.com/superdb/super/vector"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -22,6 +23,7 @@ import (
 // sensitivity of the downstream flowgraph.
 type Lister struct {
 	ctx       context.Context
+	sctx      *super.Context
 	pool      *db.Pool
 	pruner    *pruner
 	group     *errgroup.Group
@@ -31,8 +33,6 @@ type Lister struct {
 	err       error
 }
 
-var _ sbuf.Puller = (*Lister)(nil)
-
 func NewLister(ctx context.Context, sctx *super.Context, pool *db.Pool, commit uuid.UUID, pruner expr.Evaluator) (*Lister, error) {
 	snap, err := pool.Snapshot(ctx, commit)
 	if err != nil {
@@ -58,6 +58,7 @@ func NewListerFromObjects(ctx context.Context, sctx *super.Context, pool *db.Poo
 	m.Decorate(super.StylePackage)
 	l := &Lister{
 		ctx:       ctx,
+		sctx:      sctx,
 		pool:      pool,
 		objects:   objects,
 		group:     &errgroup.Group{},
@@ -69,7 +70,7 @@ func NewListerFromObjects(ctx context.Context, sctx *super.Context, pool *db.Poo
 	return l
 }
 
-func (l *Lister) Pull(done bool) (sbuf.Batch, error) {
+func (l *Lister) Pull(done bool) (vector.Any, error) {
 	l.mu.Lock()
 	defer l.mu.Unlock()
 	if l.err != nil {
@@ -84,7 +85,7 @@ func (l *Lister) Pull(done bool) (sbuf.Batch, error) {
 			return nil, err
 		}
 		if !l.pruner.prune(val) {
-			return sbuf.NewArray([]super.Value{val}), nil
+			return sbuf.Dematerialize(l.sctx, val), nil
 		}
 	}
 	return nil, nil

runtime/sam/op/meta/scanner.go → runtime/op/metascanner.go

@@ -1,4 +1,4 @@
-package meta
+package op
 
 import (
 	"context"
@@ -9,12 +9,13 @@ import (
 	"github.com/superdb/super/db"
 	"github.com/superdb/super/db/commits"
 	"github.com/superdb/super/order"
-	"github.com/superdb/super/runtime/sam/expr"
+	samexpr "github.com/superdb/super/runtime/sam/expr"
 	"github.com/superdb/super/sbuf"
 	"github.com/superdb/super/sio"
+	"github.com/superdb/super/vector/vio"
 )
 
-func NewDBMetaScanner(ctx context.Context, sctx *super.Context, r *db.Root, meta string) (sbuf.Scanner, error) {
+func NewDBMetaScanner(ctx context.Context, sctx *super.Context, r *db.Root, meta string) (vio.Puller, error) {
 	var vals []super.Value
 	var err error
 	switch meta {
@@ -28,10 +29,10 @@ func NewDBMetaScanner(ctx context.Context, sctx *super.Context, r *db.Root, meta
 	if err != nil {
 		return nil, err
 	}
-	return sbuf.NewScanner(ctx, sbuf.NewArray(vals))
+	return vio.NewPuller(sbuf.Dematerialize(sctx, vals...)), nil
 }
 
-func NewPoolMetaScanner(ctx context.Context, sctx *super.Context, r *db.Root, poolID uuid.UUID, meta string) (sbuf.Scanner, error) {
+func NewPoolMetaScanner(ctx context.Context, sctx *super.Context, r *db.Root, poolID uuid.UUID, meta string) (vio.Puller, error) {
 	p, err := r.OpenPool(ctx, poolID)
 	if err != nil {
 		return nil, err
@@ -48,10 +49,10 @@ func NewPoolMetaScanner(ctx context.Context, sctx *super.Context, r *db.Root, po
 	default:
 		return nil, fmt.Errorf("unknown pool metadata type: %q", meta)
 	}
-	return sbuf.NewScanner(ctx, sbuf.NewArray(vals))
+	return vio.NewPuller(sbuf.Dematerialize(sctx, vals...)), nil
 }
 
-func NewCommitMetaScanner(ctx context.Context, sctx *super.Context, r *db.Root, poolID, commit uuid.UUID, meta string, pruner expr.Evaluator) (sbuf.Puller, error) {
+func NewCommitMetaScanner(ctx context.Context, sctx *super.Context, r *db.Root, poolID, commit uuid.UUID, meta string, pruner samexpr.Evaluator) (vio.Puller, error) {
 	p, err := r.OpenPool(ctx, poolID)
 	if err != nil {
 		return nil, err
@@ -62,38 +63,32 @@ func NewCommitMetaScanner(ctx context.Context, sctx *super.Context, r *db.Root,
 		if err != nil {
 			return nil, err
 		}
-		return sbuf.NewScanner(ctx, sbuf.PullerReader(lister))
+		return lister, nil
 	case "partitions":
 		lister, err := NewLister(ctx, sctx, p, commit, pruner)
 		if err != nil {
 			return nil, err
 		}
-		slicer, err := NewSlicer(lister, sctx), nil
+		slicer, err := NewSlicer(sctx, lister), nil
 		if err != nil {
 			return nil, err
 		}
-		return sbuf.NewScanner(ctx, sbuf.PullerReader(slicer))
+		return slicer, nil
 	case "log":
 		tips, err := p.BatchifyBranchTips(ctx, sctx, nil)
 		if err != nil {
 			return nil, err
 		}
-		tipsScanner, err := sbuf.NewScanner(ctx, sbuf.NewArray(tips))
-		if err != nil {
-			return nil, err
-		}
+		tipsPuller := vio.NewPuller(sbuf.Dematerialize(sctx, tips...))
 		log := p.OpenCommitLog(ctx, sctx, commit)
-		logScanner, err := sbuf.NewScanner(ctx, log)
-		if err != nil {
-			return nil, err
-		}
-		return sbuf.MultiScanner(tipsScanner, logScanner), nil
+		logPuller := sbuf.NewDematerializer(sctx, sbuf.NewPuller(log))
+		return vio.ConcatPuller(tipsPuller, logPuller), nil
 	case "rawlog":
 		reader, err := p.OpenCommitLogAsBSUP(ctx, sctx, commit)
 		if err != nil {
 			return nil, err
 		}
-		return sbuf.NewScanner(ctx, reader)
+		return sbuf.NewDematerializer(sctx, sbuf.NewPuller(reader)), nil
 	default:
 		return nil, fmt.Errorf("unknown commit metadata type: %q", meta)
 	}