From 91e1bf5d0f680ff0e7e86b2243e4cdf9562b8b5b Mon Sep 17 00:00:00 2001 From: Noah Treuhaft Date: Thu, 15 Oct 2026 20:43:36 -0400 Subject: [PATCH] runtime: convert uniq operator to vio.Puller --- compiler/rungen/op.go | 3 -- compiler/rungen/vop.go | 2 + runtime/op/uniq.go | 88 +++++++++++++++++++++++++++++++++++ runtime/sam/op/uniq/uniq.go | 91 ------------------------------------- 4 files changed, 90 insertions(+), 94 deletions(-) create mode 100644 runtime/op/uniq.go delete mode 100644 runtime/sam/op/uniq/uniq.go diff --git a/compiler/rungen/op.go b/compiler/rungen/op.go index 05157ddbfb..89b9e12fc8 100644 --- a/compiler/rungen/op.go +++ b/compiler/rungen/op.go @@ -18,7 +18,6 @@ import ( "github.com/superdb/super/runtime/op" samexpr "github.com/superdb/super/runtime/sam/expr" "github.com/superdb/super/runtime/sam/op/top" - "github.com/superdb/super/runtime/sam/op/uniq" "github.com/superdb/super/sbuf" "github.com/superdb/super/vector" "github.com/superdb/super/vector/vio" @@ -109,8 +108,6 @@ func (b *Builder) compileLeaf(o dag.Op, parent sbuf.Puller) (sbuf.Puller, error) return nil, err } return top.New(b.sctx(), parent, v.Limit, exprs, v.Reverse), nil - case *dag.UniqOp: - return uniq.New(b.rctx, parent, v.Cflag), nil default: return nil, fmt.Errorf("unknown DAG operator type: %v", v) } diff --git a/compiler/rungen/vop.go b/compiler/rungen/vop.go index a998ce4720..4024462537 100644 --- a/compiler/rungen/vop.go +++ b/compiler/rungen/vop.go @@ -411,6 +411,8 @@ func (b *Builder) compileVamLeaf(o dag.Op, parent vio.Puller) (vio.Puller, error return op.NewTableScanner(b.rctx, slicer, table, nil, nil, b.progress), nil case *dag.TableMetaScan: return op.NewTableMetaScanner(b.rctx.Context, b.sctx(), b.env.DB(), o.ID, o.Meta) + case *dag.UniqOp: + return op.NewUniq(b.sctx(), parent, o.Cflag), nil case *dag.UnnestOp: e, err := b.compileVamExpr(o.Expr) if err != nil { diff --git a/runtime/op/uniq.go b/runtime/op/uniq.go new file mode 100644 index 0000000000..c90c1e8ed9 --- /dev/null +++ b/runtime/op/uniq.go @@ -0,0 +1,88 @@ +package op + +import ( + "github.com/superdb/super" + "github.com/superdb/super/sbuf" + "github.com/superdb/super/scode" + "github.com/superdb/super/vector" + "github.com/superdb/super/vector/vio" +) + +type Uniq struct { + sctx *super.Context + parent vio.Puller + withCount bool + + count int64 + eos bool + last super.Value + pullBuilder scode.Builder + wrapBuilder scode.Builder +} + +func NewUniq(sctx *super.Context, parent vio.Puller, withCount bool) *Uniq { + return &Uniq{ + sctx: sctx, + parent: parent, + withCount: withCount, + } +} + +func (u *Uniq) Pull(done bool) (vector.Any, error) { + if u.eos { + u.eos = false + return nil, nil + } + b := vector.NewDynamicValueBuilder() + for { + vec, err := u.parent.Pull(done) + if err != nil { + return nil, err + } + if vec == nil { + if u.count == 0 { + return nil, nil + } + val := u.wrap(u.last) + u.count = 0 + u.eos = true + // The zero value is not a valid super.Value. + u.last = super.Value{} + return sbuf.Dematerialize(u.sctx, val), nil + } + for i := range vec.Len() { + u.writeUniq(b, vector.ValueAt(&u.pullBuilder, vec, i)) + } + if vec := b.Build(u.sctx); vec.Len() > 0 { + return vec, nil + } + } +} + +func (u *Uniq) writeUniq(b *vector.DynamicValueBuilder, val super.Value) { + if u.count == 0 { + u.count = 1 + u.last = val.Copy() + } else if val.Equal(u.last) { + u.count++ + } else { + b.Write(u.wrap(u.last)) + u.count = 1 + u.last = val.Copy() + } +} + +func (u *Uniq) wrap(val super.Value) super.Value { + if !u.withCount { + return val + } + b := u.wrapBuilder + b.Truncate() + b.Append(val.Bytes()) + b.Append(super.EncodeInt(u.count)) + typ := u.sctx.MustLookupTypeRecord([]super.Field{ + super.NewField("value", val.Type()), + super.NewField("count", super.TypeInt64), + }) + return super.NewValue(typ, b.Bytes()) +} diff --git a/runtime/sam/op/uniq/uniq.go b/runtime/sam/op/uniq/uniq.go deleted file mode 100644 index e236d81813..0000000000 --- a/runtime/sam/op/uniq/uniq.go +++ /dev/null @@ -1,91 +0,0 @@ -package uniq - -import ( - "bytes" - - "github.com/superdb/super" - "github.com/superdb/super/runtime" - "github.com/superdb/super/sbuf" - "github.com/superdb/super/scode" -) - -type Op struct { - rctx *runtime.Context - parent sbuf.Puller - builder scode.Builder - cflag bool - count int64 - last *super.Value - eos bool -} - -func New(rctx *runtime.Context, parent sbuf.Puller, cflag bool) *Op { - return &Op{ - rctx: rctx, - parent: parent, - cflag: cflag, - } -} - -func (o *Op) wrap(t *super.Value) super.Value { - if o.cflag { - o.builder.Reset() - o.builder.Append(t.Bytes()) - o.builder.Append(super.EncodeInt(o.count)) - typ := o.rctx.Sctx.MustLookupTypeRecord([]super.Field{ - super.NewField("value", t.Type()), - super.NewField("count", super.TypeInt64), - }) - return super.NewValue(typ, o.builder.Bytes()).Copy() - } - return *t -} - -func (o *Op) appendUniq(out []super.Value, t *super.Value) []super.Value { - if o.count == 0 { - o.last = t.Copy().Ptr() - o.count = 1 - return out - } else if bytes.Equal(t.Bytes(), o.last.Bytes()) { - o.count++ - return out - } - out = append(out, o.wrap(o.last)) - o.last = t.Copy().Ptr() - o.count = 1 - return out -} - -// uniq is a little bit complicated because we have to check uniqueness -// across records between calls to Pull. -func (o *Op) Pull(done bool) (sbuf.Batch, error) { - if o.eos { - o.eos = false - return nil, nil - } - for { - batch, err := o.parent.Pull(done) - if err != nil { - return nil, err - } - if batch == nil { - if o.last == nil { - return nil, nil - } - o.eos = true - t := o.wrap(o.last) - o.count = 0 - o.last = nil - return sbuf.NewArray([]super.Value{t}), nil - } - var out []super.Value - vals := batch.Values() - for i := range vals { - out = o.appendUniq(out, &vals[i]) - } - batch.Unref() - if len(out) > 0 { - return sbuf.NewArray(out), nil - } - } -}