From 83c908f37fc591bf47add3f5198e65f3568dd8d6 Mon Sep 17 00:00:00 2001 From: Steven McCanne Date: Wed, 7 Oct 2026 18:57:52 -0700 Subject: [PATCH] change oldbsup to BSUP rows in spiller --- runtime/sam/op/spill/file.go | 29 +++++++++++++---------------- runtime/sam/op/spill/peeker.go | 10 +++++----- 2 files changed, 18 insertions(+), 21 deletions(-) diff --git a/runtime/sam/op/spill/file.go b/runtime/sam/op/spill/file.go index 92c2afe77..859070ace 100644 --- a/runtime/sam/op/spill/file.go +++ b/runtime/sam/op/spill/file.go @@ -2,13 +2,15 @@ package spill import ( "bufio" + "context" "os" "github.com/superdb/super" - "github.com/superdb/super/bsup/oldbsup" + "github.com/superdb/super/bsup" "github.com/superdb/super/pkg/bufwriter" "github.com/superdb/super/pkg/fs" "github.com/superdb/super/sio" + "github.com/superdb/super/sio/bsupio" ) // File provides a means to write a sequence of Super values to temporary @@ -17,9 +19,9 @@ import ( // but can be processed in multiple passes. File implements sio.Reader and // sio.Writer. type File struct { - *oldbsup.Reader - *oldbsup.Writer - file *os.File + reader sio.Reader + writer *bsup.RowWriter + file *os.File } // NewFile returns a File. Records should be written to File via the sio.Writer @@ -27,7 +29,7 @@ type File struct { // records via the sio.Reader interface. func NewFile(f *os.File) *File { return &File{ - Writer: oldbsup.NewWriter(bufwriter.New(sio.NopCloser(f))), + writer: bsup.NewRowWriter(bufwriter.New(sio.NopCloser(f))), file: f, } } @@ -40,28 +42,23 @@ func NewFileWithPath(path string) (*File, error) { return NewFile(f), nil } -func (f *File) Rewind(sctx *super.Context) error { +func (f *File) Rewind(ctx context.Context, sctx *super.Context) error { // Close the writer to flush any pending output but since we // wrapped the file in a sio.NopCloser, the file will stay open. - if err := f.Writer.Close(); err != nil { + if err := f.writer.Close(); err != nil { return err } - f.Writer = nil + f.writer = nil if _, err := f.file.Seek(0, 0); err != nil { return err } - if f.Reader != nil { - f.Reader.Close() - } - f.Reader = oldbsup.NewReader(sctx, bufio.NewReader(f.file)) - return nil + var err error + f.reader, err = bsupio.NewValueReader(ctx, sctx, bufio.NewReader(f.file)) + return err } // CloseAndRemove closes and removes the underlying file. func (r *File) CloseAndRemove() error { - if r.Reader != nil { - r.Reader.Close() - } err := r.file.Close() if rmErr := os.Remove(r.file.Name()); err == nil { err = rmErr diff --git a/runtime/sam/op/spill/peeker.go b/runtime/sam/op/spill/peeker.go index 6c522c060..546488816 100644 --- a/runtime/sam/op/spill/peeker.go +++ b/runtime/sam/op/spill/peeker.go @@ -13,20 +13,20 @@ type peeker struct { ordinal int } -func newPeeker(ctx context.Context, sctx *super.Context, filename string, ordinal int, zr sio.Reader) (*peeker, error) { +func newPeeker(ctx context.Context, sctx *super.Context, filename string, ordinal int, reader sio.Reader) (*peeker, error) { f, err := NewFileWithPath(filename) if err != nil { return nil, err } - if err := sio.CopyWithContext(ctx, f, zr); err != nil { + if err := sio.CopyWithContext(ctx, f.writer, reader); err != nil { f.CloseAndRemove() return nil, err } - if err := f.Rewind(sctx); err != nil { + if err := f.Rewind(ctx, sctx); err != nil { f.CloseAndRemove() return nil, err } - first, err := f.Read() + first, err := f.reader.Read() if err != nil { f.CloseAndRemove() return nil, err @@ -42,7 +42,7 @@ func (p *peeker) read() (*super.Value, bool, error) { rec = rec.Copy().Ptr() } var err error - p.nextRecord, err = p.Read() + p.nextRecord, err = p.reader.Read() eof := p.nextRecord == nil && err == nil return rec, eof, err }