runtime: move exec.Environment.OpenPool to op.Robot.openPool

compiler/compiler.go

@@ -13,7 +13,7 @@ import (
 	"github.com/superdb/super/pkg/storage"
 	"github.com/superdb/super/runtime"
 	"github.com/superdb/super/runtime/exec"
-	"github.com/superdb/super/runtime/op/poolscanner"
+	"github.com/superdb/super/runtime/op"
 	"github.com/superdb/super/runtime/sam/op/meta"
 	"github.com/superdb/super/vector/vio"
 )
@@ -78,7 +78,7 @@ func (l *compiler) NewObjectScanner(rctx *runtime.Context, poolID uuid.UUID, obj
 	}
 	lister := meta.NewSortedListerFromObjects(rctx, rctx.Sctx, pool, objects, nil)
 	slicer := meta.NewSlicer(lister, rctx.Sctx)
-	return poolscanner.NewPoolScanner(rctx, slicer, pool, nil, nil, nil), nil
+	return op.NewPoolScanner(rctx, slicer, pool, nil, nil, nil), nil
 }
 
 type InvalidDeleteWhereQuery struct{}

compiler/rungen/vop.go

@@ -14,8 +14,6 @@ import (
 	"github.com/superdb/super/runtime/expr/agg"
 	"github.com/superdb/super/runtime/op"
 	"github.com/superdb/super/runtime/op/aggregate"
-	"github.com/superdb/super/runtime/op/merge"
-	"github.com/superdb/super/runtime/op/poolscanner"
 	samexpr "github.com/superdb/super/runtime/sam/expr"
 	"github.com/superdb/super/runtime/sam/op/meta"
 	"github.com/superdb/super/sbuf"
@@ -65,7 +63,7 @@ func (b *Builder) compileVam(o dag.Op, parents []vio.Puller) ([]vio.Puller, erro
 			return nil, err
 		}
 		cmp := samexpr.NewComparator(exprs...)
-		return []vio.Puller{merge.NewMerge(b.rctx, parents, cmp.Compare)}, nil
+		return []vio.Puller{op.NewMerge(b.rctx, parents, cmp.Compare)}, nil
 	case *dag.ScatterOp:
 		return b.compileVamScatter(o, parents)
 	case *dag.SwitchOp:
@@ -244,7 +242,7 @@ func (b *Builder) compileVamLeaf(o dag.Op, parent vio.Puller) (vio.Puller, error
 				return nil, err
 			}
 		}
-		return poolscanner.NewDeleter(b.rctx, sbuf.NewMaterializer(parent), pool, whereNot, pruner, b.progress, b.deletes), nil
+		return op.NewDeleter(b.rctx, sbuf.NewMaterializer(parent), pool, whereNot, pruner, b.progress, b.deletes), nil
 	case *dag.DistinctOp:
 		e, err := b.compileVamExpr(o.Expr)
 		if err != nil {
@@ -333,7 +331,7 @@ func (b *Builder) compileVamLeaf(o dag.Op, parent vio.Puller) (vio.Puller, error
 			return nil, err
 		}
 		slicer := meta.NewSlicer(l, b.mctx)
-		return poolscanner.NewPoolScanner(b.rctx, slicer, pool, nil, nil, b.progress), nil
+		return op.NewPoolScanner(b.rctx, slicer, pool, nil, nil, b.progress), nil
 	case *dag.SeqScan:
 		pool, err := b.lookupPool(o.Pool)
 		if err != nil {
@@ -353,7 +351,7 @@ func (b *Builder) compileVamLeaf(o dag.Op, parent vio.Puller) (vio.Puller, error
 				return nil, err
 			}
 		}
-		return poolscanner.NewPoolScanner(b.rctx, sbuf.NewMaterializer(parent), pool, filter, pruner, b.progress), nil
+		return op.NewPoolScanner(b.rctx, sbuf.NewMaterializer(parent), pool, filter, pruner, b.progress), nil
 	case *dag.SkipOp:
 		return op.NewSkip(parent, o.Count), nil
 	case *dag.SortOp:

runtime/exec/environment.go

@@ -15,9 +15,6 @@ import (
 	"github.com/superdb/super/dbid"
 	"github.com/superdb/super/order"
 	"github.com/superdb/super/pkg/storage"
-	"github.com/superdb/super/runtime"
-	"github.com/superdb/super/runtime/op/poolscanner"
-	"github.com/superdb/super/runtime/sam/op/meta"
 	"github.com/superdb/super/sbuf"
 	"github.com/superdb/super/sio/anyio"
 	"github.com/superdb/super/vector"
@@ -126,24 +123,6 @@ func (e *Environment) OpenHTTP(ctx context.Context, sctx *super.Context, url, fo
 	return file, nil
 }
 
-func (e *Environment) OpenPool(ctx context.Context, sctx *super.Context, id uuid.UUID, pushdown sbuf.Pushdown) (vio.Puller, error) {
-	pool, err := e.db.OpenPool(ctx, id)
-	if err != nil {
-		return nil, err
-	}
-	branch, err := e.CommitObject(ctx, id, "main")
-	if err != nil {
-		return nil, err
-	}
-	l, err := meta.NewSortedLister(ctx, sctx, pool, branch, nil)
-	if err != nil {
-		return nil, err
-	}
-	// XXX We're passing nil for pushdown here.  In a subsequent PR, we will
-	// work out vector pushdown and put this back.
-	return poolscanner.NewPoolScanner(runtime.NewContext(ctx, sctx), l, pool, nil, nil, nil), nil
-}
-
 func newConcurrentPuller(path string, puller vio.Puller) ConcurrentPuller {
 	cp, ok := puller.(ConcurrentPuller)
 	if !ok {

runtime/op/poolscanner/deleter.go → runtime/op/deleter.go

@@ -1,4 +1,4 @@
-package poolscanner
+package op
 
 import (
 	"sync"

runtime/op/merge/merge.go → runtime/op/merge.go

@@ -1,4 +1,4 @@
-package merge
+package op
 
 import (
 	"container/heap"
@@ -147,11 +147,6 @@ func (m *Merge) Pop() any {
 	return x
 }
 
-type result struct {
-	vector vector.Any
-	err    error
-}
-
 type mergeParent struct {
 	ctx      context.Context
 	parent   vio.Puller

runtime/op/poolscanner/sequence.go → runtime/op/poolscanner.go

@@ -1,4 +1,4 @@
-package poolscanner
+package op
 
 import (
 	"context"
@@ -9,7 +9,6 @@ import (
 	"github.com/superdb/super/db/data"
 	"github.com/superdb/super/runtime"
 	"github.com/superdb/super/runtime/expr"
-	"github.com/superdb/super/runtime/op/merge"
 	samexpr "github.com/superdb/super/runtime/sam/expr"
 	"github.com/superdb/super/runtime/sam/op/meta"
 	"github.com/superdb/super/sbuf"
@@ -134,7 +133,7 @@ func newObjectsScanner(ctx context.Context, sctx *super.Context, pool *db.Pool,
 	if len(pullers) == 1 {
 		return pullers[0], nil
 	}
-	return merge.NewMerge(ctx, pullers, db.ImportComparator(sctx, pool).Compare), nil
+	return NewMerge(ctx, pullers, db.ImportComparator(sctx, pool).Compare), nil
 }
 
 func newObjectScanner(ctx context.Context, sctx *super.Context, pool *db.Pool, object *data.Object, filter expr.Evaluator, progress *vio.Progress) (vio.Puller, error) {
@@ -186,15 +185,3 @@ func (s *statScanner) Pull(done bool) (vector.Any, error) {
 func (s *statScanner) Close() error {
 	return nil
 }
-
-func applyMask(vec, mask vector.Any) (vector.Any, bool) {
-	// errors are ignored for filters
-	b, _ := expr.BoolMask(mask)
-	if b.IsEmpty() {
-		return nil, false
-	}
-	if b.GetCardinality() == uint64(mask.Len()) {
-		return vec, true
-	}
-	return vector.Pick(vec, b.ToArray()), true
-}

runtime/op/robot.go

@@ -1,9 +1,12 @@
 package op
 
 import (
+	"uuid"
+
 	"github.com/superdb/super/runtime"
 	"github.com/superdb/super/runtime/exec"
 	"github.com/superdb/super/runtime/expr"
+	"github.com/superdb/super/runtime/sam/op/meta"
 	"github.com/superdb/super/sbuf"
 	"github.com/superdb/super/scode"
 	"github.com/superdb/super/vector"
@@ -136,8 +139,27 @@ func (o *Robot) open(path string) (vio.Puller, error) {
 		// XXX we should support committish values
 		id, err := o.env.PoolID(o.rctx, path)
 		if err == nil {
-			return o.env.OpenPool(o.rctx, o.rctx.Sctx, id, o.pushdown)
+			return o.openPool(id)
 		}
 	}
 	return o.env.Open(o.rctx.Context, o.rctx.Sctx, path, o.format, o.pushdown, 1)
 }
+
+func (r *Robot) openPool(id uuid.UUID) (vio.Puller, error) {
+	ctx := r.rctx.Context
+	pool, err := r.env.DB().OpenPool(ctx, id)
+	if err != nil {
+		return nil, err
+	}
+	branch, err := r.env.DB().CommitObject(ctx, id, "main")
+	if err != nil {
+		return nil, err
+	}
+	l, err := meta.NewSortedLister(ctx, r.rctx.Sctx, pool, branch, nil)
+	if err != nil {
+		return nil, err
+	}
+	// XXX We're passing nil for pushdown here.  In a subsequent PR, we will
+	// work out vector pushdown and put this back.
+	return NewPoolScanner(r.rctx, l, pool, nil, nil, nil), nil
+}

runtime/op/wrapper.go

@@ -0,0 +1,20 @@
+package op
+
+import (
+	"fmt"
+
+	"github.com/superdb/super/vector"
+	"github.com/superdb/super/vector/vio"
+)
+
+type Wrapper struct {
+	Parent vio.Puller
+	Msg    string
+}
+
+func (w *Wrapper) Pull(done bool) (vector.Any, error) {
+	fmt.Printf("%p %T %s %v\n", w.Parent, w.Parent, w.Msg, done)
+	vec, err := w.Parent.Pull(done)
+	fmt.Printf("%p %T %s %v %v %v\n", w.Parent, w.Parent, w.Msg, done, vec, err)
+	return vec, err
+}