From d2bff7ae9bed3b9d041dc6874ca00ba4e8ea2f3a Mon Sep 17 00:00:00 2001 From: Noah Treuhaft Date: Thu, 8 Oct 2026 13:18:00 -0400 Subject: [PATCH] runtime: move exec.Environment.OpenPool to op.Robot.openPool This allows (and actually requires) movement of the contents of op/merge and op/poolscanner into op. --- compiler/compiler.go | 4 ++-- compiler/rungen/vop.go | 10 ++++----- runtime/exec/environment.go | 21 ------------------ runtime/op/{poolscanner => }/deleter.go | 2 +- runtime/op/{merge => }/merge.go | 7 +----- runtime/op/{poolscanner => }/poolscanner.go | 17 ++------------- runtime/op/robot.go | 24 ++++++++++++++++++++- 7 files changed, 33 insertions(+), 52 deletions(-) rename runtime/op/{poolscanner => }/deleter.go (99%) rename runtime/op/{merge => }/merge.go (98%) rename runtime/op/{poolscanner => }/poolscanner.go (91%) diff --git a/compiler/compiler.go b/compiler/compiler.go index 659251da2..02373aee5 100644 --- a/compiler/compiler.go +++ b/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 ksuid.KSUID, o } 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{} diff --git a/compiler/rungen/vop.go b/compiler/rungen/vop.go index 601f10a6e..5fa4cd5d0 100644 --- a/compiler/rungen/vop.go +++ b/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: diff --git a/runtime/exec/environment.go b/runtime/exec/environment.go index 977e24002..50c289921 100644 --- a/runtime/exec/environment.go +++ b/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 ksuid.KSUID, 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 { diff --git a/runtime/op/poolscanner/deleter.go b/runtime/op/deleter.go similarity index 99% rename from runtime/op/poolscanner/deleter.go rename to runtime/op/deleter.go index f8c52fe1b..d199828ec 100644 --- a/runtime/op/poolscanner/deleter.go +++ b/runtime/op/deleter.go @@ -1,4 +1,4 @@ -package poolscanner +package op import ( "sync" diff --git a/runtime/op/merge/merge.go b/runtime/op/merge.go similarity index 98% rename from runtime/op/merge/merge.go rename to runtime/op/merge.go index aca0b992e..b98d864ac 100644 --- a/runtime/op/merge/merge.go +++ b/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 diff --git a/runtime/op/poolscanner/poolscanner.go b/runtime/op/poolscanner.go similarity index 91% rename from runtime/op/poolscanner/poolscanner.go rename to runtime/op/poolscanner.go index e67f786d4..f071996d8 100644 --- a/runtime/op/poolscanner/poolscanner.go +++ b/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 -} diff --git a/runtime/op/robot.go b/runtime/op/robot.go index efc9679e8..35ba10976 100644 --- a/runtime/op/robot.go +++ b/runtime/op/robot.go @@ -1,9 +1,11 @@ package op import ( + "github.com/segmentio/ksuid" "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 +138,28 @@ 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 ksuid.KSUID) (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 + +}