From c2351c5f11635e5dbcea8f6b6e4bb8586b74449e Mon Sep 17 00:00:00 2001 From: Noah Treuhaft Date: Fri, 9 Oct 2026 21:35:12 -0400 Subject: [PATCH] runtime: convert sam/op/meta contents to vio.Puller Convert the operators in runtime/sam/op/meta from sbuf.Puller and sbuf.Scanner to vio.Puller and move them to runtime/op. To avoid creating an import cycle in the process, move runtime/sam/op/meta.Partition to db/data. --- compiler/compiler.go | 5 +- compiler/rungen/op.go | 41 ------------ compiler/rungen/vop.go | 41 ++++++++++-- db/data/partition.go | 31 +++++++++ .../sam/op/meta => db}/ztests/nulls-max.yaml | 0 .../sam/op/meta => db}/ztests/partitions.yaml | 0 .../sam/op/meta => db}/ztests/pruner.yaml | 0 .../op/meta => db}/ztests/type-context.yaml | 0 runtime/{sam/op/meta => op}/lister.go | 11 ++-- .../op/meta/scanner.go => op/metascanner.go} | 35 +++++------ runtime/op/poolscanner.go | 18 +++--- runtime/{sam/op/meta => op}/pruner.go | 2 +- runtime/op/robot.go | 3 +- runtime/{sam/op/meta => op}/slicer.go | 63 +++++++------------ sio/dbio/unmarshal.go | 3 +- sio/dbio/writer.go | 5 +- 16 files changed, 125 insertions(+), 133 deletions(-) create mode 100644 db/data/partition.go rename {runtime/sam/op/meta => db}/ztests/nulls-max.yaml (100%) rename {runtime/sam/op/meta => db}/ztests/partitions.yaml (100%) rename {runtime/sam/op/meta => db}/ztests/pruner.yaml (100%) rename {runtime/sam/op/meta => db}/ztests/type-context.yaml (100%) rename runtime/{sam/op/meta => op}/lister.go (93%) rename runtime/{sam/op/meta/scanner.go => op/metascanner.go} (73%) rename runtime/{sam/op/meta => op}/pruner.go (96%) rename runtime/{sam/op/meta => op}/slicer.go (64%) diff --git a/compiler/compiler.go b/compiler/compiler.go index 3febdb9f0a..28d8ad51bb 100644 --- a/compiler/compiler.go +++ b/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 } diff --git a/compiler/rungen/op.go b/compiler/rungen/op.go index c8132bff41..88d36b6a58 100644 --- a/compiler/rungen/op.go +++ b/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 { diff --git a/compiler/rungen/vop.go b/compiler/rungen/vop.go index 5596cd4e0a..fd1054062d 100644 --- a/compiler/rungen/vop.go +++ b/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 { diff --git a/db/data/partition.go b/db/data/partition.go new file mode 100644 index 0000000000..4911096cd5 --- /dev/null +++ b/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)) +} diff --git a/runtime/sam/op/meta/ztests/nulls-max.yaml b/db/ztests/nulls-max.yaml similarity index 100% rename from runtime/sam/op/meta/ztests/nulls-max.yaml rename to db/ztests/nulls-max.yaml diff --git a/runtime/sam/op/meta/ztests/partitions.yaml b/db/ztests/partitions.yaml similarity index 100% rename from runtime/sam/op/meta/ztests/partitions.yaml rename to db/ztests/partitions.yaml diff --git a/runtime/sam/op/meta/ztests/pruner.yaml b/db/ztests/pruner.yaml similarity index 100% rename from runtime/sam/op/meta/ztests/pruner.yaml rename to db/ztests/pruner.yaml diff --git a/runtime/sam/op/meta/ztests/type-context.yaml b/db/ztests/type-context.yaml similarity index 100% rename from runtime/sam/op/meta/ztests/type-context.yaml rename to db/ztests/type-context.yaml diff --git a/runtime/sam/op/meta/lister.go b/runtime/op/lister.go similarity index 93% rename from runtime/sam/op/meta/lister.go rename to runtime/op/lister.go index b6bfb17841..c3170259c0 100644 --- a/runtime/sam/op/meta/lister.go +++ b/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 diff --git a/runtime/sam/op/meta/scanner.go b/runtime/op/metascanner.go similarity index 73% rename from runtime/sam/op/meta/scanner.go rename to runtime/op/metascanner.go index 6261b9c8dd..1448ca405f 100644 --- a/runtime/sam/op/meta/scanner.go +++ b/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) } diff --git a/runtime/op/poolscanner.go b/runtime/op/poolscanner.go index f071996d88..3d9af82149 100644 --- a/runtime/op/poolscanner.go +++ b/runtime/op/poolscanner.go @@ -10,8 +10,6 @@ import ( "github.com/superdb/super/runtime" "github.com/superdb/super/runtime/expr" 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" ) @@ -19,7 +17,7 @@ import ( // PoolScanner implements an op that pulls metadata partitions to scan // from its parent and for each partition, scans the object. type PoolScanner struct { - parent sbuf.Puller + parent vio.Puller scanner vio.Puller where expr.Evaluator pruner samexpr.Evaluator @@ -31,7 +29,7 @@ type PoolScanner struct { err error } -func NewPoolScanner(rctx *runtime.Context, parent sbuf.Puller, pool *db.Pool, where expr.Evaluator, pruner samexpr.Evaluator, progress *vio.Progress) *PoolScanner { +func NewPoolScanner(rctx *runtime.Context, parent vio.Puller, pool *db.Pool, where expr.Evaluator, pruner samexpr.Evaluator, progress *vio.Progress) *PoolScanner { return &PoolScanner{ rctx: rctx, parent: parent, @@ -57,19 +55,19 @@ func (p *PoolScanner) Pull(done bool) (vector.Any, error) { } for { if p.scanner == nil { - batch, err := p.parent.Pull(false) - if batch == nil || err != nil { + vec, err := p.parent.Pull(false) + if vec == nil || err != nil { p.close(err) return nil, err } - vals := batch.Values() - if len(vals) != 1 { + if vec.Len() != 1 { // We currently support only one partition per batch. err := errors.New("system error: SequenceScanner encountered multi-valued batch") p.close(err) return nil, err } - p.scanner, _, err = newScanner(p.rctx.Context, p.rctx.Sctx, p.pool, p.unmarshaler, p.pruner, p.where, p.progress, vals[0]) + val := vector.ValueAt(nil, vec, 0) + p.scanner, _, err = newScanner(p.rctx.Context, p.rctx.Sctx, p.pool, p.unmarshaler, p.pruner, p.where, p.progress, val) if err != nil { p.close(err) return nil, err @@ -105,7 +103,7 @@ func newScanner(ctx context.Context, sctx *super.Context, pool *db.Pool, u *supe } objects = []*data.Object{&object} } else { - var part meta.Partition + var part data.Partition if err := u.Unmarshal(val, &part); err != nil { return nil, nil, err } diff --git a/runtime/sam/op/meta/pruner.go b/runtime/op/pruner.go similarity index 96% rename from runtime/sam/op/meta/pruner.go rename to runtime/op/pruner.go index 8785213889..9a4af1612d 100644 --- a/runtime/sam/op/meta/pruner.go +++ b/runtime/op/pruner.go @@ -1,4 +1,4 @@ -package meta +package op import ( "github.com/superdb/super" diff --git a/runtime/op/robot.go b/runtime/op/robot.go index 589b28aef1..e9ee39b334 100644 --- a/runtime/op/robot.go +++ b/runtime/op/robot.go @@ -6,7 +6,6 @@ import ( "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/scode" "github.com/superdb/super/vector" "github.com/superdb/super/vector/vio" @@ -154,7 +153,7 @@ func (r *Robot) openPool(id uuid.UUID) (vio.Puller, error) { if err != nil { return nil, err } - l, err := meta.NewLister(ctx, r.rctx.Sctx, pool, branch, nil) + l, err := NewLister(ctx, r.rctx.Sctx, pool, branch, nil) if err != nil { return nil, err } diff --git a/runtime/sam/op/meta/slicer.go b/runtime/op/slicer.go similarity index 64% rename from runtime/sam/op/meta/slicer.go rename to runtime/op/slicer.go index d3c134d100..16ba21a2f8 100644 --- a/runtime/sam/op/meta/slicer.go +++ b/runtime/op/slicer.go @@ -1,8 +1,7 @@ -package meta +package op import ( "errors" - "fmt" "sync" "github.com/superdb/super" @@ -10,14 +9,16 @@ import ( "github.com/superdb/super/order" "github.com/superdb/super/runtime/sam/expr" "github.com/superdb/super/sbuf" - "github.com/superdb/super/sup" + "github.com/superdb/super/vector" + "github.com/superdb/super/vector/vio" ) // Slicer implements an op that pulls data objects and organizes // them into overlapping object Slices forming a sequence of // non-overlapping Partitions. type Slicer struct { - parent sbuf.Puller + sctx *super.Context + parent vio.Puller marshaler *super.Marshaler unmarshaler *super.Unmarshaler objects []*data.Object @@ -27,10 +28,11 @@ type Slicer struct { mu sync.Mutex } -func NewSlicer(parent sbuf.Puller, sctx *super.Context) *Slicer { +func NewSlicer(sctx *super.Context, parent vio.Puller) *Slicer { m := super.NewMarshaler(sctx) m.Decorate(super.StylePackage) return &Slicer{ + sctx: sctx, parent: parent, marshaler: m, unmarshaler: super.NewUnmarshaler(), @@ -39,7 +41,7 @@ func NewSlicer(parent sbuf.Puller, sctx *super.Context) *Slicer { } } -func (s *Slicer) Pull(done bool) (sbuf.Batch, error) { +func (s *Slicer) Pull(done bool) (vector.Any, error) { //XXX for now we use a mutex because multiple downstream trunks can call // Pull concurrently here. We should change this to use a fork. But for now, // this does not seem like a performance critical issue because the bottleneck @@ -47,20 +49,20 @@ func (s *Slicer) Pull(done bool) (sbuf.Batch, error) { s.mu.Lock() defer s.mu.Unlock() for { - batch, err := s.parent.Pull(done) + vec, err := s.parent.Pull(done) if err != nil { return nil, err } - if batch == nil { + if vec == nil { return s.nextPartition() } - vals := batch.Values() - if len(vals) != 1 { + if vec.Len() != 1 { // We currently support only one object per batch. return nil, errors.New("system error: Slicer encountered multi-valued batch") } + val := vector.ValueAt(nil, vec, 0) var object data.Object - if err := s.unmarshaler.Unmarshal(vals[0], &object); err != nil { + if err := s.unmarshaler.Unmarshal(val, &object); err != nil { return nil, err } if batch, err := s.stash(&object); batch != nil || err != nil { @@ -70,8 +72,8 @@ func (s *Slicer) Pull(done bool) (sbuf.Batch, error) { } // nextPartition takes collected up slices and forms a partition returning -// a batch containing a single value comprising the serialized partition. -func (s *Slicer) nextPartition() (sbuf.Batch, error) { +// a vector containing a single value comprising the serialized partition. +func (s *Slicer) nextPartition() (vector.Any, error) { if len(s.objects) == 0 { return nil, nil } @@ -86,7 +88,7 @@ func (s *Slicer) nextPartition() (sbuf.Batch, error) { max = o.Max } } - val, err := s.marshaler.Marshal(&Partition{ + val, err := s.marshaler.Marshal(&data.Partition{ Min: min, Max: max, Objects: s.objects, @@ -95,18 +97,18 @@ func (s *Slicer) nextPartition() (sbuf.Batch, error) { if err != nil { return nil, err } - return sbuf.NewArray([]super.Value{val}), nil + return sbuf.Dematerialize(s.sctx, val), nil } -func (s *Slicer) stash(o *data.Object) (sbuf.Batch, error) { - var batch sbuf.Batch +func (s *Slicer) stash(o *data.Object) (vector.Any, error) { + var vec vector.Any if len(s.objects) > 0 { // We collect all the subsequent objects that overlap with any object in the // accumulated set so far. Since first times are non-decreasing this is // guaranteed to generate partitions that are non-decreasing and non-overlapping. if s.cmp(o.Max, *s.min) < 0 || s.cmp(o.Min, *s.max) > 0 { var err error - batch, err = s.nextPartition() + vec, err = s.nextPartition() if err != nil { return nil, err } @@ -121,28 +123,5 @@ func (s *Slicer) stash(o *data.Object) (sbuf.Batch, error) { if s.max == nil || s.cmp(*s.max, o.Max) < 0 { s.max = o.Max.Copy().Ptr() } - return batch, nil -} - -// 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 []*data.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)) + return vec, nil } diff --git a/sio/dbio/unmarshal.go b/sio/dbio/unmarshal.go index 11a903fd32..1869010768 100644 --- a/sio/dbio/unmarshal.go +++ b/sio/dbio/unmarshal.go @@ -7,7 +7,6 @@ import ( "github.com/superdb/super/db/data" "github.com/superdb/super/db/pools" "github.com/superdb/super/pkg/field" - "github.com/superdb/super/runtime/sam/op/meta" ) var unmarshaler *super.Unmarshaler @@ -19,10 +18,10 @@ func init() { commits.Commit{}, commits.Delete{}, field.Path{}, - meta.Partition{}, pools.Config{}, db.BranchMeta{}, db.BranchTip{}, data.Object{}, + data.Partition{}, ) } diff --git a/sio/dbio/writer.go b/sio/dbio/writer.go index 3123ec3499..86a8999233 100644 --- a/sio/dbio/writer.go +++ b/sio/dbio/writer.go @@ -17,7 +17,6 @@ import ( "github.com/superdb/super/pkg/charm" "github.com/superdb/super/pkg/terminal/color" "github.com/superdb/super/pkg/units" - "github.com/superdb/super/runtime/sam/op/meta" "github.com/superdb/super/sbuf" "github.com/superdb/super/sup" "github.com/superdb/super/vector" @@ -94,7 +93,7 @@ func (w *Writer) formatValue(t table, b *bytes.Buffer, v any, width int, colors formatDataObject(b, &v, "", 0) case *data.Object: formatDataObject(b, v, "", 0) - case meta.Partition: + case data.Partition: formatPartition(b, v) case *commits.Commit: branches := w.branches[v.ID] @@ -176,7 +175,7 @@ func formatDataObject(b *bytes.Buffer, object *data.Object, prefix string, inden b.WriteByte('\n') } -func formatPartition(b *bytes.Buffer, p meta.Partition) { +func formatPartition(b *bytes.Buffer, p data.Partition) { b.WriteString("min ") b.WriteString(sup.String(p.Min)) b.WriteString(" max ")