From fe8cce24a8d50312e774d8522991d665ce819758 Mon Sep 17 00:00:00 2001 From: Steven McCanne Date: Tue, 6 Oct 2026 15:20:52 -0700 Subject: [PATCH 1/3] replace database data storage with BSUP columns instead of old-bsup This commit replaces the storage format of the databsae with BSUP columns. We're currently suboptimally writing one big column frame per storage object but this will change in a subsequent PR when we add a row-group-like construct (pruner and vcache just need to understand this). We put merge and the pool scanners in subdirectories of runtime/op to break import loops. We need to revisit interfaces again to come up with a cleaner design but this will come later. The ingest path is now native columns though there is an scode boomerang to sort the input. We'll avoid this and add optional sort keys in a subsequent PR. --- runtime/op/poolscanner/sequence.go | 200 +++++++++++++++++++++++++++++ 1 file changed, 200 insertions(+) create mode 100644 runtime/op/poolscanner/sequence.go diff --git a/runtime/op/poolscanner/sequence.go b/runtime/op/poolscanner/sequence.go new file mode 100644 index 000000000..e67f786d4 --- /dev/null +++ b/runtime/op/poolscanner/sequence.go @@ -0,0 +1,200 @@ +package poolscanner + +import ( + "context" + "errors" + + "github.com/superdb/super" + "github.com/superdb/super/db" + "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" + "github.com/superdb/super/vector" + "github.com/superdb/super/vector/vio" +) + +// 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 + scanner vio.Puller + where expr.Evaluator + pruner samexpr.Evaluator + rctx *runtime.Context + pool *db.Pool + progress *vio.Progress + unmarshaler *super.Unmarshaler + done bool + err error +} + +func NewPoolScanner(rctx *runtime.Context, parent sbuf.Puller, pool *db.Pool, where expr.Evaluator, pruner samexpr.Evaluator, progress *vio.Progress) *PoolScanner { + return &PoolScanner{ + rctx: rctx, + parent: parent, + where: where, + pruner: pruner, + pool: pool, + progress: progress, + unmarshaler: super.NewUnmarshaler(), + } +} + +func (p *PoolScanner) Pull(done bool) (vector.Any, error) { + if p.done { + return nil, p.err + } + if done { + if p.scanner != nil { + _, err := p.scanner.Pull(true) + p.close(err) + p.scanner = nil + } + return nil, p.err + } + for { + if p.scanner == nil { + batch, err := p.parent.Pull(false) + if batch == nil || err != nil { + p.close(err) + return nil, err + } + vals := batch.Values() + if len(vals) != 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]) + if err != nil { + p.close(err) + return nil, err + } + } + vec, err := p.scanner.Pull(false) + if err != nil { + p.close(err) + return nil, err + } + if vec != nil { + return vec, nil + } + p.scanner = nil + } +} + +func (p *PoolScanner) close(err error) { + p.err = err + p.done = true +} + +func newScanner(ctx context.Context, sctx *super.Context, pool *db.Pool, u *super.Unmarshaler, pruner samexpr.Evaluator, where expr.Evaluator, progress *vio.Progress, val super.Value) (vio.Puller, *data.Object, error) { + named, ok := val.Type().(*super.TypeNamed) + if !ok { + return nil, nil, errors.New("system error: SequenceScanner encountered unnamed object") + } + var objects []*data.Object + if named.Name == "data.Object" { + var object data.Object + if err := u.Unmarshal(val, &object); err != nil { + return nil, nil, err + } + objects = []*data.Object{&object} + } else { + var part meta.Partition + if err := u.Unmarshal(val, &part); err != nil { + return nil, nil, err + } + objects = part.Objects + } + scanner, err := newObjectsScanner(ctx, sctx, pool, objects, pruner, where, progress) + return scanner, objects[0], err +} + +func newObjectsScanner(ctx context.Context, sctx *super.Context, pool *db.Pool, objects []*data.Object, pruner samexpr.Evaluator, filter expr.Evaluator, progress *vio.Progress) (vio.Puller, error) { + pullers := make([]vio.Puller, 0, len(objects)) + pullersDone := func() { + for _, puller := range pullers { + puller.Pull(true) + } + } + for _, object := range objects { + s, err := newObjectScanner(ctx, sctx, pool, object, filter, progress) + if err != nil { + pullersDone() + return nil, err + } + pullers = append(pullers, s) + } + if len(pullers) == 1 { + return pullers[0], nil + } + return merge.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) { + // XXX We will figure out how concurrency is determined and how + // the pushdown works for BSUP columns in a subsequent PR. + const concurrency = 1 + scanner, err := pool.NewReader(ctx, sctx, object, nil, concurrency) + if err != nil { + return nil, err + } + return &statScanner{ + scanner: scanner, + progress: progress, + filter: filter, + }, nil +} + +type statScanner struct { + scanner vio.ScanCloser + err error + progress *vio.Progress + filter expr.Evaluator +} + +func (s *statScanner) Pull(done bool) (vector.Any, error) { + for { + if s.scanner == nil { + return nil, s.err + } + vec, err := s.scanner.Pull(done) + if vec == nil || err != nil { + s.progress.Add(s.scanner.Progress()) + if err2 := s.scanner.Close(); err == nil { + err = err2 + } + s.err = err + s.scanner = nil + return vec, err + } + if s.filter == nil { + return vec, err + } + if masked, ok := applyMask(vec, s.filter.Eval(vec)); ok { + return masked, nil + } + } +} + +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 +} From 146bdfd167af7f0c044cce1f1ed71646bf0b9e63 Mon Sep 17 00:00:00 2001 From: Steven McCanne Date: Thu, 8 Oct 2026 08:34:37 -0700 Subject: [PATCH 2/3] add fast path to database ingest This commit adds a fast path for sorting ingest regarding the common case that a table sort key is a simple primary type. In this case, we do native sorts on the vector data to compute an index which is used to create a view. I believe the changes to the tests has to do with variation in stability of the sort between sam and this new path. Sam sort will go away soon so I'm not concerned about maintaining the invariant between the two. --- book/src/tutorials/join.md | 2 +- db/writer.go | 190 ++++++++++++++++++++++++++++-- db/ztests/seek-index-overlap.yaml | 4 +- 3 files changed, 186 insertions(+), 10 deletions(-) diff --git a/book/src/tutorials/join.md b/book/src/tutorials/join.md index 4642b81d0..c2f66acc4 100644 --- a/book/src/tutorials/join.md +++ b/book/src/tutorials/join.md @@ -204,8 +204,8 @@ produces {name:"banana",color:"yellow",flavor:"sweet",eater:"quinn"} {name:"dates",color:"brown",flavor:"sweet",note:"in season",eater:"quinn"} {name:"strawberry",color:"red",flavor:"sweet",eater:"quinn"} -{name:"apple",color:"red",flavor:"tart",eater:"chris"} {name:"apple",color:"red",flavor:"tart",eater:"morgan"} +{name:"apple",color:"red",flavor:"tart",eater:"chris"} ``` ### Alternate Syntax diff --git a/db/writer.go b/db/writer.go index b98998c97..2005ad5cb 100644 --- a/db/writer.go +++ b/db/writer.go @@ -1,15 +1,21 @@ package db import ( + "bytes" + "cmp" "context" "io" + "slices" + "strings" "sync/atomic" "github.com/superdb/super" "github.com/superdb/super/bsup" "github.com/superdb/super/db/data" "github.com/superdb/super/order" - "github.com/superdb/super/runtime/sam/expr" + "github.com/superdb/super/pkg/field" + "github.com/superdb/super/runtime/expr" + samexpr "github.com/superdb/super/runtime/sam/expr" "github.com/superdb/super/sbuf" "github.com/superdb/super/vector" "github.com/superdb/super/vector/vbuild" @@ -117,18 +123,18 @@ func (s *ImportStats) Copy() ImportStats { } } -func ImportComparator(sctx *super.Context, pool *Pool) *expr.Comparator { - var exprs []expr.SortExpr +func ImportComparator(sctx *super.Context, pool *Pool) *samexpr.Comparator { + var exprs []samexpr.SortExpr for _, s := range pool.SortKeys { - exprs = append(exprs, expr.NewSortExpr(expr.NewDottedExpr(sctx, s.Key), s.Order, s.Order.NullsMax(true))) + exprs = append(exprs, samexpr.NewSortExpr(samexpr.NewDottedExpr(sctx, s.Key), s.Order, s.Order.NullsMax(true))) } var o order.Which if !pool.SortKeys.IsNil() { o = pool.SortKeys.Primary().Order } // valueAsBytes establishes a total order. - exprs = append(exprs, expr.NewSortExpr(&valueAsBytes{}, o, o.NullsMax(true))) - return expr.NewComparator(exprs...) + exprs = append(exprs, samexpr.NewSortExpr(&valueAsBytes{}, o, o.NullsMax(true))) + return samexpr.NewComparator(exprs...) } type valueAsBytes struct{} @@ -138,8 +144,11 @@ func (v *valueAsBytes) Eval(val super.Value) super.Value { } func (w *Writer) sort(vec vector.Any) (vector.Any, super.Value, super.Value) { + if vec, minVal, maxVal := w.fastSort(vec); vec != nil { + return vec, minVal, maxVal + } c := ImportComparator(w.sctx, w.pool) - primaryKey := expr.NewDottedExpr(w.sctx, w.pool.SortKeys.Primary().Key) + primaryKey := samexpr.NewDottedExpr(w.sctx, w.pool.SortKeys.Primary().Key) reader := c.SortStableReader(sbuf.Materialize(vec).Values()) out := vector.NewDynamicValueBuilder() var minVal, maxVal super.Value @@ -162,3 +171,170 @@ func (w *Writer) sort(vec vector.Any) (vector.Any, super.Value, super.Value) { out.Write(*val) } } + +func (w *Writer) fastSort(vec vector.Any) (vector.Any, super.Value, super.Value) { + if len(w.pool.SortKeys) != 1 { + return nil, super.Value{}, super.Value{} + } + sortKey := w.pool.SortKeys[0] + return w.fastSortWithKey(vec, w.deref(vec, sortKey.Key), sortKey.Order) +} + +func (w *Writer) fastSortWithKey(vec, key vector.Any, o order.Which) (vector.Any, super.Value, super.Value) { + switch key := key.(type) { + case *vector.Named: + return w.fastSortWithKey(vec, key.Any, o) + case *vector.Int: + idx := make([]uint32, len(key.Values)) + for k := range idx { + idx[k] = uint32(k) + } + slices.SortStableFunc(idx, func(a, b uint32) int { + if o == order.Desc { + a, b = b, a + } + return cmp.Compare(key.Values[a], key.Values[b]) + }) + minVal := key.Values[0] + maxVal := key.Values[0] + for _, val := range key.Values[1:] { + if val < minVal { + minVal = val + } else if val > maxVal { + maxVal = val + } + } + if o == order.Desc { + minVal, maxVal = maxVal, minVal + } + return vector.NewView(vec, idx), super.NewInt(key.Type(), minVal), super.NewInt(key.Type(), maxVal) + case *vector.Uint: + idx := make([]uint32, len(key.Values)) + for k := range idx { + idx[k] = uint32(k) + } + slices.SortStableFunc(idx, func(a, b uint32) int { + if o == order.Desc { + a, b = b, a + } + return cmp.Compare(key.Values[a], key.Values[b]) + }) + minVal := key.Values[0] + maxVal := key.Values[0] + for _, val := range key.Values[1:] { + if val < minVal { + minVal = val + } else if val > maxVal { + maxVal = val + } + } + if o == order.Desc { + minVal, maxVal = maxVal, minVal + } + return vector.NewView(vec, idx), super.NewUint(key.Type(), minVal), super.NewUint(key.Type(), maxVal) + case *vector.Float: + idx := make([]uint32, len(key.Values)) + for k := range idx { + idx[k] = uint32(k) + } + slices.SortStableFunc(idx, func(a, b uint32) int { + if o == order.Desc { + a, b = b, a + } + return cmp.Compare(key.Values[a], key.Values[b]) + }) + minVal := key.Values[0] + maxVal := key.Values[0] + for _, val := range key.Values[1:] { + if val < minVal { + minVal = val + } else if val > maxVal { + maxVal = val + } + } + if o == order.Desc { + minVal, maxVal = maxVal, minVal + } + return vector.NewView(vec, idx), super.NewFloat(key.Type(), minVal), super.NewFloat(key.Type(), maxVal) + case *vector.String: + idx := make([]uint32, key.Len()) + for k := range idx { + idx[k] = uint32(k) + } + slices.SortStableFunc(idx, func(a, b uint32) int { + if o == order.Desc { + a, b = b, a + } + return strings.Compare(key.Value(a), key.Value(b)) + }) + minVal := key.Value(0) + maxVal := key.Value(0) + for k := range vec.Len() - 1 { + off := k + 1 + if s := key.Value(off); s < minVal { + minVal = s + } else if s > maxVal { + maxVal = s + } + } + if o == order.Desc { + minVal, maxVal = maxVal, minVal + } + return vector.NewView(vec, idx), super.NewString(minVal), super.NewString(maxVal) + case *vector.Bytes: + idx := make([]uint32, key.Len()) + for k := range idx { + idx[k] = uint32(k) + } + slices.SortStableFunc(idx, func(a, b uint32) int { + if o == order.Desc { + a, b = b, a + } + return bytes.Compare(key.Value(a), key.Value(b)) + }) + minVal := key.Value(0) + maxVal := key.Value(0) + for k := range vec.Len() - 1 { + off := k + 1 + if b := key.Value(off); bytes.Compare(b, minVal) < 0 { + minVal = b + } else if bytes.Compare(b, maxVal) > 0 { + maxVal = b + } + } + if o == order.Desc { + minVal, maxVal = maxVal, minVal + } + return vector.NewView(vec, idx), super.NewBytes(minVal), super.NewBytes(maxVal) + case *vector.IP: + idx := make([]uint32, len(key.Values)) + for k := range idx { + idx[k] = uint32(k) + } + slices.SortStableFunc(idx, func(a, b uint32) int { + if o == order.Desc { + a, b = b, a + } + return key.Values[a].Compare(key.Values[b]) + }) + minVal := key.Values[0] + maxVal := key.Values[0] + for _, val := range key.Values[1:] { + if val.Compare(minVal) < 0 { + minVal = val + } else if val.Compare(maxVal) > 0 { + maxVal = val + } + } + if o == order.Desc { + minVal, maxVal = maxVal, minVal + } + return vector.NewView(vec, idx), super.NewIP(minVal), super.NewIP(maxVal) + } + return nil, super.Value{}, super.Value{} +} + +func (w *Writer) deref(vec vector.Any, path field.Path) vector.Any { + e := expr.NewDottedExpr(w.sctx, path.Chain()) + return e.Eval(vec) +} diff --git a/db/ztests/seek-index-overlap.yaml b/db/ztests/seek-index-overlap.yaml index 8b0adbb72..e4d6390f6 100644 --- a/db/ztests/seek-index-overlap.yaml +++ b/db/ztests/seek-index-overlap.yaml @@ -25,6 +25,6 @@ outputs: {count:1150} - name: stderr data: | - {bytes_read:33519,bytes_matched:33519,records_read:1150,records_matched:1150} + {bytes_read:33503,bytes_matched:33503,records_read:1150,records_matched:1150} === - {bytes_read:35654,bytes_matched:35654,records_read:1150,records_matched:1150} + {bytes_read:35634,bytes_matched:35634,records_read:1150,records_matched:1150} From 1c17430193cc971ee5e31a595f045dd7b1a73eb5 Mon Sep 17 00:00:00 2001 From: Steven McCanne Date: Thu, 8 Oct 2026 11:25:55 -0700 Subject: [PATCH 3/3] fix rebase --- runtime/op/poolscanner/poolscanner.go | 200 -------------------------- 1 file changed, 200 deletions(-) delete mode 100644 runtime/op/poolscanner/poolscanner.go diff --git a/runtime/op/poolscanner/poolscanner.go b/runtime/op/poolscanner/poolscanner.go deleted file mode 100644 index e67f786d4..000000000 --- a/runtime/op/poolscanner/poolscanner.go +++ /dev/null @@ -1,200 +0,0 @@ -package poolscanner - -import ( - "context" - "errors" - - "github.com/superdb/super" - "github.com/superdb/super/db" - "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" - "github.com/superdb/super/vector" - "github.com/superdb/super/vector/vio" -) - -// 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 - scanner vio.Puller - where expr.Evaluator - pruner samexpr.Evaluator - rctx *runtime.Context - pool *db.Pool - progress *vio.Progress - unmarshaler *super.Unmarshaler - done bool - err error -} - -func NewPoolScanner(rctx *runtime.Context, parent sbuf.Puller, pool *db.Pool, where expr.Evaluator, pruner samexpr.Evaluator, progress *vio.Progress) *PoolScanner { - return &PoolScanner{ - rctx: rctx, - parent: parent, - where: where, - pruner: pruner, - pool: pool, - progress: progress, - unmarshaler: super.NewUnmarshaler(), - } -} - -func (p *PoolScanner) Pull(done bool) (vector.Any, error) { - if p.done { - return nil, p.err - } - if done { - if p.scanner != nil { - _, err := p.scanner.Pull(true) - p.close(err) - p.scanner = nil - } - return nil, p.err - } - for { - if p.scanner == nil { - batch, err := p.parent.Pull(false) - if batch == nil || err != nil { - p.close(err) - return nil, err - } - vals := batch.Values() - if len(vals) != 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]) - if err != nil { - p.close(err) - return nil, err - } - } - vec, err := p.scanner.Pull(false) - if err != nil { - p.close(err) - return nil, err - } - if vec != nil { - return vec, nil - } - p.scanner = nil - } -} - -func (p *PoolScanner) close(err error) { - p.err = err - p.done = true -} - -func newScanner(ctx context.Context, sctx *super.Context, pool *db.Pool, u *super.Unmarshaler, pruner samexpr.Evaluator, where expr.Evaluator, progress *vio.Progress, val super.Value) (vio.Puller, *data.Object, error) { - named, ok := val.Type().(*super.TypeNamed) - if !ok { - return nil, nil, errors.New("system error: SequenceScanner encountered unnamed object") - } - var objects []*data.Object - if named.Name == "data.Object" { - var object data.Object - if err := u.Unmarshal(val, &object); err != nil { - return nil, nil, err - } - objects = []*data.Object{&object} - } else { - var part meta.Partition - if err := u.Unmarshal(val, &part); err != nil { - return nil, nil, err - } - objects = part.Objects - } - scanner, err := newObjectsScanner(ctx, sctx, pool, objects, pruner, where, progress) - return scanner, objects[0], err -} - -func newObjectsScanner(ctx context.Context, sctx *super.Context, pool *db.Pool, objects []*data.Object, pruner samexpr.Evaluator, filter expr.Evaluator, progress *vio.Progress) (vio.Puller, error) { - pullers := make([]vio.Puller, 0, len(objects)) - pullersDone := func() { - for _, puller := range pullers { - puller.Pull(true) - } - } - for _, object := range objects { - s, err := newObjectScanner(ctx, sctx, pool, object, filter, progress) - if err != nil { - pullersDone() - return nil, err - } - pullers = append(pullers, s) - } - if len(pullers) == 1 { - return pullers[0], nil - } - return merge.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) { - // XXX We will figure out how concurrency is determined and how - // the pushdown works for BSUP columns in a subsequent PR. - const concurrency = 1 - scanner, err := pool.NewReader(ctx, sctx, object, nil, concurrency) - if err != nil { - return nil, err - } - return &statScanner{ - scanner: scanner, - progress: progress, - filter: filter, - }, nil -} - -type statScanner struct { - scanner vio.ScanCloser - err error - progress *vio.Progress - filter expr.Evaluator -} - -func (s *statScanner) Pull(done bool) (vector.Any, error) { - for { - if s.scanner == nil { - return nil, s.err - } - vec, err := s.scanner.Pull(done) - if vec == nil || err != nil { - s.progress.Add(s.scanner.Progress()) - if err2 := s.scanner.Close(); err == nil { - err = err2 - } - s.err = err - s.scanner = nil - return vec, err - } - if s.filter == nil { - return vec, err - } - if masked, ok := applyMask(vec, s.filter.Eval(vec)); ok { - return masked, nil - } - } -} - -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 -}