From cd28e2cdfb9116439806ad7dd034cad1f5352e8d Mon Sep 17 00:00:00 2001 From: Steven McCanne Date: Thu, 8 Oct 2026 12:25:33 -0700 Subject: [PATCH 1/2] remove support for old BSUP format --- bsup/oldbsup/batch.go | 47 --- bsup/oldbsup/buffer.go | 85 ----- bsup/oldbsup/frame.go | 51 --- bsup/oldbsup/parser.go | 226 ------------- bsup/oldbsup/reader.go | 135 -------- bsup/oldbsup/scanner.go | 332 -------------------- bsup/oldbsup/scanner_test.go | 60 ---- bsup/oldbsup/sync.go | 83 ----- bsup/oldbsup/types.go | 109 ------- bsup/oldbsup/writer.go | 186 ----------- bsup/oldbsup/ztests/complex-named.yaml | 14 - bsup/oldbsup/ztests/dev-zero.yaml | 7 - bsup/oldbsup/ztests/esc-typeid.yaml | 236 -------------- bsup/oldbsup/ztests/issue-4082.yaml | 17 - bsup/oldbsup/ztests/outer-named.yaml | 13 - bsup/oldbsup/ztests/primitive.yaml | 16 - bsup/oldbsup/ztests/pushdown-filter.yaml | 15 - bsup/oldbsup/ztests/string-set.yaml | 16 - bsup/oldbsup/ztests/union-array.yaml | 11 - bsup/oldbsup/ztests/union-nested.yaml | 14 - bsup/oldbsup/ztests/union-simple.yaml | 12 - bsup/oldbsup/ztests/zctx-named-reset-2.yaml | 19 -- bsup/oldbsup/ztests/zctx-named-reset.yaml | 22 -- bsup/oldbsup/ztests/zctx-reset.yaml | 20 -- cli/inputflags/flags.go | 2 +- cli/outputflags/flags.go | 4 +- cmd/super/db/command.go | 2 +- cmd/super/ztests/o-flag.yaml | 6 +- db/data/object.go | 2 +- db/ztests/ls.yaml | 2 +- db/ztests/s3/rm.yaml | 18 +- fuzz/fuzz.go | 11 +- service/ztests/curl-load-error.yaml | 2 +- service/ztests/drop.yaml | 2 +- service/ztests/load-garbage.yaml | 1 - service/ztests/ls-pools.yaml | 2 +- sio/anyio/lookup.go | 7 - sio/anyio/reader.go | 21 -- sio/anyio/writer.go | 3 - sio/anyio/ztests/detector-dev-zero.yaml | 1 - sio/anyio/ztests/fake-bsup.yaml | 12 - sio/anyio/ztests/huge.yaml | 1 - sio/sio.go | 4 - sio/sup_test.go | 12 +- sup/marshal_bsup_test.go | 9 +- sup/marshal_test.go | 19 +- vector/vector_test.go | 2 +- ztests/mixed-primitive-alias.yaml | 24 -- 48 files changed, 49 insertions(+), 1866 deletions(-) delete mode 100644 bsup/oldbsup/batch.go delete mode 100644 bsup/oldbsup/buffer.go delete mode 100644 bsup/oldbsup/frame.go delete mode 100644 bsup/oldbsup/parser.go delete mode 100644 bsup/oldbsup/reader.go delete mode 100644 bsup/oldbsup/scanner.go delete mode 100644 bsup/oldbsup/scanner_test.go delete mode 100644 bsup/oldbsup/sync.go delete mode 100644 bsup/oldbsup/types.go delete mode 100644 bsup/oldbsup/writer.go delete mode 100644 bsup/oldbsup/ztests/complex-named.yaml delete mode 100644 bsup/oldbsup/ztests/dev-zero.yaml delete mode 100644 bsup/oldbsup/ztests/esc-typeid.yaml delete mode 100644 bsup/oldbsup/ztests/issue-4082.yaml delete mode 100644 bsup/oldbsup/ztests/outer-named.yaml delete mode 100644 bsup/oldbsup/ztests/primitive.yaml delete mode 100644 bsup/oldbsup/ztests/pushdown-filter.yaml delete mode 100644 bsup/oldbsup/ztests/string-set.yaml delete mode 100644 bsup/oldbsup/ztests/union-array.yaml delete mode 100644 bsup/oldbsup/ztests/union-nested.yaml delete mode 100644 bsup/oldbsup/ztests/union-simple.yaml delete mode 100644 bsup/oldbsup/ztests/zctx-named-reset-2.yaml delete mode 100644 bsup/oldbsup/ztests/zctx-named-reset.yaml delete mode 100644 bsup/oldbsup/ztests/zctx-reset.yaml delete mode 100644 sio/anyio/ztests/fake-bsup.yaml delete mode 100644 ztests/mixed-primitive-alias.yaml diff --git a/bsup/oldbsup/batch.go b/bsup/oldbsup/batch.go deleted file mode 100644 index db9f97e54d..0000000000 --- a/bsup/oldbsup/batch.go +++ /dev/null @@ -1,47 +0,0 @@ -package oldbsup - -import ( - "sync" - "sync/atomic" - - "github.com/superdb/super" - "github.com/superdb/super/sbuf" -) - -type batch struct { - buf *buffer - refs int32 - vals []super.Value -} - -var _ sbuf.Batch = (*batch)(nil) - -var batchPool sync.Pool - -// newBatch takes ownership of buf but not vals. -func newBatch(buf *buffer, vals []super.Value) *batch { - b, ok := batchPool.Get().(*batch) - if !ok { - b = &batch{} - } - b.buf = buf - b.refs = 1 - b.vals = append(b.vals[:0], vals...) - return b -} - -func (b *batch) Ref() { atomic.AddInt32(&b.refs, 1) } - -func (b *batch) Unref() { - if refs := atomic.AddInt32(&b.refs, -1); refs == 0 { - if b.buf != nil { - b.buf.free() - b.buf = nil - } - batchPool.Put(b) - } else if refs < 0 { - panic("bsup-rows: negative batch reference count") - } -} - -func (b *batch) Values() []super.Value { return b.vals } diff --git a/bsup/oldbsup/buffer.go b/bsup/oldbsup/buffer.go deleted file mode 100644 index f84dfed224..0000000000 --- a/bsup/oldbsup/buffer.go +++ /dev/null @@ -1,85 +0,0 @@ -package oldbsup - -import ( - "io" - "sync" -) - -type buffer struct { - data []byte - off int -} - -var bigBuffers sync.Pool -var smallBuffers sync.Pool - -func newBuffer(length int) *buffer { - if length <= DefaultFrameThresh { - b, ok := smallBuffers.Get().(*buffer) - if !ok { - b = &buffer{data: make([]byte, DefaultFrameThresh)} - } - b.data = b.data[:length] - b.off = 0 - return b - } - b, ok := bigBuffers.Get().(*buffer) - // The capacity check will send the smaller buffers to GC and tend - // to keep buffers big enough in the pool - if !ok || cap(b.data) < length { - // Add 25% cushion - b = &buffer{data: make([]byte, length+(length>>2))} - } - b.data = b.data[:length] - b.off = 0 - return b -} - -func newBufferFromBytes(b []byte) *buffer { - buf := newBuffer(len(b)) - copy(buf.data, b) - return buf -} - -func (b *buffer) free() { - if b == nil { - return - } - if cap(b.data) == DefaultFrameThresh { - smallBuffers.Put(b) - } else if cap(b.data) > DefaultFrameThresh { - bigBuffers.Put(b) - } -} - -// Bytes is so named to avoid collision with the bytes package. -func (b *buffer) Bytes() []byte { - return b.data[b.off:] -} - -func (b *buffer) length() int { - return len(b.data) - b.off -} - -func (b *buffer) read(n int) ([]byte, error) { - var err error - if avail := b.length(); n > avail { - if avail == 0 { - return nil, io.EOF - } - err = io.ErrUnexpectedEOF - n = avail - } - off := b.off - b.off += n - return b.data[off:b.off], err -} - -func (b *buffer) ReadByte() (byte, error) { - if b.length() < 1 { - return 0, io.EOF - } - off := b.off - b.off++ - return b.data[off], nil -} diff --git a/bsup/oldbsup/frame.go b/bsup/oldbsup/frame.go deleted file mode 100644 index f7033a6535..0000000000 --- a/bsup/oldbsup/frame.go +++ /dev/null @@ -1,51 +0,0 @@ -package oldbsup - -import ( - "fmt" - - "github.com/pierrec/lz4/v4" -) - -const ( - TypesFrame = 0 - ValuesFrame = 1 - ControlFrame = 2 -) - -const ( - EOS = 0xff - ControlFormatBSUP = 0 - ControlFormatJSON = 1 - ControlFormatSUP = 2 - ControlFormatString = 3 - ControlFormatBinary = 4 -) - -type CompressionFormat int - -const CompressionFormatLZ4 CompressionFormat = 0x00 - -type frame struct { - fmt CompressionFormat - zbuf *buffer - ubuf *buffer -} - -func (f *frame) free() { - f.zbuf.free() - f.ubuf.free() -} - -func (f *frame) decompress() error { - if f.fmt != CompressionFormatLZ4 { - return fmt.Errorf("bsup-rows: unknown compression format 0x%x", f.fmt) - } - n, err := lz4.UncompressBlock(f.zbuf.data, f.ubuf.data) - if err != nil { - return fmt.Errorf("bsup-rows: %w", err) - } - if n != len(f.ubuf.data) { - return fmt.Errorf("bsup-rows: got %d uncompressed bytes, expected %d", n, len(f.ubuf.data)) - } - return nil -} diff --git a/bsup/oldbsup/parser.go b/bsup/oldbsup/parser.go deleted file mode 100644 index 9c998a4749..0000000000 --- a/bsup/oldbsup/parser.go +++ /dev/null @@ -1,226 +0,0 @@ -package oldbsup - -import ( - "errors" - "fmt" - "io" - "slices" - - "github.com/superdb/super/pkg/peeker" - "github.com/superdb/super/sbuf" - "github.com/superdb/super/scode" -) - -var errBadFormat = errors.New("malformed bsup-rows value") - -// parser decodes the framing protocol for BSUPROWS updating and resetting its -// super context in conformance with BSUPROWS framing. -type parser struct { - peeker *peeker.Reader - types *Decoder - maxSize int -} - -const Version = 6 - -func CheckVersion(code byte) error { - var version int - if (code & 0x80) != 0 { - version = int(code & 0x7f) - } - if version != Version { - return fmt.Errorf("bsup-rows version mismatch: expected %d, found %d", Version, version) - } - return nil -} - -func (p *parser) read() (frame, error) { - for { - version, err := p.peeker.ReadByte() - if err != nil { - return frame{}, err - } - if version == EOS { - // At EOS, we create a new Decoder which clears out the types slice - // mapping the local type IDs to the shared-context types. Any data - // batches concurrently being decoded by a worker will still point - // to the old types slice so all continues on just fine as - // everything gets properly mappped to the shared context - // under concurrent locking in the target super.Context. - p.types = NewDecoder(p.types.sctx) - continue - } - if err := CheckVersion(version); err != nil { - return frame{}, err - } - code, err := p.peeker.ReadByte() - if err != nil { - return frame{}, err - } - switch typ := (code >> 4) & 3; typ { - case TypesFrame: - if err := p.decodeTypes(code); err != nil { - return frame{}, err - } - case ValuesFrame: - return p.decodeValues(code) - case ControlFrame: - return frame{}, p.decodeControl(code) - default: - return frame{}, fmt.Errorf("unknown bsup-rows message frame type: %d", typ) - } - } -} - -func (p *parser) decodeTypes(code byte) error { - if (code & 0x40) != 0 { - // Compressed - f, err := p.readCompressedFrame(code) - if err != nil { - return err - } - if err := f.decompress(); err != nil { - return err - } - if err := p.types.decode(f.ubuf); err != nil { - return err - } - f.free() - return nil - } else { - // Uncompressed. - // b points into the peaker buffer, but not a problem - // as we decode everything before the next read. - f, err := p.readFrame(code) - if err != nil { - return err - } - tmpBuf := buffer{data: f} - if err := p.types.decode(&tmpBuf); err != nil { - return err - } - return nil - } -} - -func (p *parser) decodeValues(code byte) (frame, error) { - if (code & 0x40) != 0 { - // Compressed - return p.readCompressedFrame(code) - } - bytes, err := p.readFrame(code) - if err != nil { - return frame{}, err - } - return frame{ubuf: newBufferFromBytes(bytes)}, nil -} - -// decodeControl reads the next message frame as a control message and -// returns it as *sbuf.Control, which implements error. Errors are also -// return as error so reflection must be used to distringuish the cases. -func (p *parser) decodeControl(code byte) error { - var bytes []byte - if (code & 0x40) == 0 { - // b points into the peaker buffer so we copy it. - b, err := p.readFrame(code) - if err != nil { - return err - } - bytes = slices.Clone(b) - } else { - // The frame is compressed. - blk, err := p.readCompressedFrame(code) - if err != nil { - return err - } - if err := blk.decompress(); err != nil { - return err - } - bytes = slices.Clone(blk.ubuf.data) - blk.free() - } - if len(bytes) == 0 { - return errBadFormat - } - // Insert this control message into the result queue to preserve - // order between values frames and messages. Note that a back-to-back - // sequence of control messages will be processed here by the scanner - // go-routine as the workers go idle. However, this is not a critical - // performance path so we're not worried about parallelism here. - return &sbuf.Control{ - Message: &Control{ - Format: int(bytes[0]), - Bytes: bytes[1:], - }, - } -} - -func (p *parser) readFrame(code byte) ([]byte, error) { - size, err := p.decodeLength(code) - if err != nil { - return nil, err - } - if size < 1 { - return nil, fmt.Errorf("bsup-rows: frame length (%d) too small", size) - } - if size > p.maxSize { - return nil, fmt.Errorf("bsup-rows: frame length (%d) exceeds maximum allowed (%d)", size, p.maxSize) - } - b, err := p.peeker.Read(size) - if err == peeker.ErrBufferOverflow { - return nil, fmt.Errorf("large value of %d bytes exceeds maximum read buffer", size) - } - return b, err -} - -// readCompressedFrame parses the compression header and reads the compressed -// payload from the peaker into a buffer. This allows the peaker to move on -// and the worker to decompress the buffer concurrently. (A more sophisticated -// implementation could sync the peeker movement to the decode pipeline to -// avoid this copy. In this approach, compressed buffers would point into the -// peeker buffer and be released after decompression. A reference-counted double -// buffer would work nicely for this.) -func (p *parser) readCompressedFrame(code byte) (frame, error) { - n, err := p.decodeLength(code) - if err != nil { - return frame{}, err - } - format, err := p.peeker.ReadByte() - if err != nil { - return frame{}, err - } - size, err := readUvarintAsInt(p.peeker) - if err != nil { - return frame{}, err - } - if size < 1 { - return frame{}, fmt.Errorf("bsup-rows: frame length (%d) too small", size) - } - if size > p.maxSize { - return frame{}, fmt.Errorf("bsup-rows: frame length (%d) exceeds maximum allowed (%d)", size, p.maxSize) - } - // The size of the compressed buffer needs to be adjusted by the - // byte for the format and the variable-length bytes to encode - // the original size. - n -= 1 + scode.SizeOfUvarint(uint64(size)) - b, err := p.peeker.Read(n) - if err != nil && err != io.EOF { - if err == peeker.ErrBufferOverflow { - return frame{}, fmt.Errorf("large value of %d bytes exceeds maximum read buffer", n) - } - return frame{}, errBadFormat - } - return frame{ - fmt: CompressionFormat(format), - zbuf: newBufferFromBytes(b), - ubuf: newBuffer(size), - }, nil -} - -func (p *parser) decodeLength(code byte) (int, error) { - v, err := readUvarintAsInt(p.peeker) - if err != nil { - return 0, err - } - return (v << 4) | (int(code) & 0xf), nil -} diff --git a/bsup/oldbsup/reader.go b/bsup/oldbsup/reader.go deleted file mode 100644 index 49b581884f..0000000000 --- a/bsup/oldbsup/reader.go +++ /dev/null @@ -1,135 +0,0 @@ -package oldbsup - -import ( - "context" - "encoding/binary" - "fmt" - "io" - "runtime" - - "github.com/superdb/super" - "github.com/superdb/super/sbuf" - "github.com/superdb/super/sio" -) - -const ( - ReadSize = 512 * 1024 - MaxSize = 1024 * 1024 * 1024 - TypeLimit = 10000 -) - -type Reader struct { - sctx *super.Context - reader io.Reader - opts ReaderOpts - scanner sbuf.Scanner - wrap sio.Reader -} - -var _ sbuf.ScannerAble = (*Reader)(nil) - -type ReaderOpts struct { - Validate bool - Size int - Max int - Threads int -} - -type Control struct { - Format int - Bytes []byte -} - -func NewReader(sctx *super.Context, reader io.Reader) *Reader { - return NewReaderWithOpts(sctx, reader, ReaderOpts{}) -} - -func NewReaderWithOpts(sctx *super.Context, reader io.Reader, opts ReaderOpts) *Reader { - if opts.Size == 0 { - opts.Size = ReadSize - } - if opts.Max == 0 { - opts.Max = MaxSize - } - opts.Size = min(opts.Size, opts.Max) - if opts.Threads == 0 { - opts.Threads = runtime.GOMAXPROCS(0) - } - return &Reader{ - sctx: sctx, - reader: reader, - opts: opts, - } -} - -func (r *Reader) NewScanner(ctx context.Context, filter sbuf.Pushdown) (sbuf.Scanner, error) { - if r.opts.Threads == 1 { - return newScannerSync(ctx, r.sctx, r.reader, filter, r.opts) - } - return newScanner(ctx, r.sctx, r.reader, filter, r.opts) -} - -// Close guarantees that the underlying io.Reader is not read after it returns. -func (r *Reader) Close() error { - if r.scanner != nil { - r.scanner.Pull(true) - } - return nil -} - -func (r *Reader) init() error { - if r.wrap != nil { - return nil - } - //XXX ctx... seems like all NewReaders should take ctx so they - // can have cancellable goroutines? - scanner, err := r.NewScanner(context.TODO(), nil) - if err != nil { - return err - } - r.scanner = scanner - r.wrap = sbuf.PullerReader(scanner) - return nil -} - -func (r *Reader) Read() (*super.Value, error) { - // If Read is called, then this Reader is being used as a sio.Reader and - // not as a sbuf.Puller. We just wrap the scanner in a puller to - // implement the Reader interface. If it's used a sbuf.Scanner, then - // the NewScanner method will be called and Read will never happen. - if err := r.init(); err != nil { - return nil, err - } - for { - val, err := r.wrap.Read() - if err != nil { - if _, ok := err.(*sbuf.Control); ok { - continue - } - return nil, err - } - return val, err - } -} - -func (r *Reader) ReadPayload() (*super.Value, *Control, error) { - if err := r.init(); err != nil { - return nil, nil, err - } - val, err := r.wrap.Read() - if err != nil { - if zctrl, ok := err.(*sbuf.Control); ok { - ctrl, ok := zctrl.Message.(*Control) - if !ok { - return nil, nil, fmt.Errorf("bsupio internal error: unknown control type: %T", zctrl.Message) - } - return nil, ctrl, nil - } - } - return val, nil, err -} - -func readUvarintAsInt(r io.ByteReader) (int, error) { - u64, err := binary.ReadUvarint(r) - return int(u64), err -} diff --git a/bsup/oldbsup/scanner.go b/bsup/oldbsup/scanner.go deleted file mode 100644 index c7299aa386..0000000000 --- a/bsup/oldbsup/scanner.go +++ /dev/null @@ -1,332 +0,0 @@ -package oldbsup - -import ( - "context" - "fmt" - "io" - "sync" - "sync/atomic" - - "github.com/superdb/super" - "github.com/superdb/super/pkg/peeker" - "github.com/superdb/super/runtime/sam/expr" - "github.com/superdb/super/runtime/sam/op" - "github.com/superdb/super/sbuf" - "github.com/superdb/super/scode" - "github.com/superdb/super/vector/vio" -) - -type scanner struct { - ctx context.Context - cancel context.CancelFunc - parser parser - progress vio.Progress - validate bool - once sync.Once - workers []*worker - workerCh chan *worker - resultChCh chan chan op.Result - err error - eof bool -} - -func newScanner(ctx context.Context, sctx *super.Context, r io.Reader, pushdown sbuf.Pushdown, opts ReaderOpts) (sbuf.Scanner, error) { - ctx, cancel := context.WithCancel(ctx) - s := &scanner{ - ctx: ctx, - cancel: cancel, - parser: parser{ - peeker: peeker.NewReader(r, opts.Size, opts.Max), - types: NewDecoder(sctx), - maxSize: opts.Max, - }, - validate: opts.Validate, - workerCh: make(chan *worker, opts.Threads+1), - resultChCh: make(chan chan op.Result, opts.Threads+1), - } - for range opts.Threads { - var bf *expr.BufferFilter - var f expr.Evaluator - if pushdown != nil { - var err error - bf, err = pushdown.BSUPFilter() - if err != nil { - return nil, err - } - f, err = pushdown.DataFilter() - if err != nil { - return nil, err - } - } - s.workers = append(s.workers, newWorker(ctx, &s.progress, bf, f, s.validate)) - } - return s, nil -} - -func (s *scanner) Pull(done bool) (sbuf.Batch, error) { - s.once.Do(s.start) - if done { - s.cancel() - for range s.resultChCh { - // Wait for the s.parser goroutine to exit so we know it - // won't continue reading from the underlying io.Reader. - } - s.eof = true - return nil, nil - } - if s.err != nil || s.eof { - return nil, s.err - } - for { - select { - case ch, ok := <-s.resultChCh: - if !ok { - return nil, s.ctx.Err() - } - result, ok := <-ch - if !ok { - continue - } - if result.Batch == nil || result.Err != nil { - if _, ok := result.Err.(*sbuf.Control); !ok { - s.eof = true - s.err = result.Err - s.cancel() - } - } - return result.Batch, result.Err - case <-s.ctx.Done(): - return nil, s.ctx.Err() - } - } -} - -func (s *scanner) start() { - for _, w := range s.workers { - go w.run(s.workerCh) - } - go func() { - defer close(s.resultChCh) - // This is the input goroutine that reads message blocks - // from the input. Types and control messages are decoded - // in this thread and data blocks are distributed to the workers - // with the property that all types for a given data block will - // exist in the decoder types state (which in turn points to the - // shared query context) before the worker is given the buffer - // to (optionally) uncompress, filter, and decode when matched. - // When we hit end-of-stream, a new type context and types slice are - // created for the new data batches. Since all data is mapped to - // the shared context and each worker maps its values independently, - // the decode pipeline continues to operate concurrenlty without - // any problem even when faced with changing localized type state. - for { - frame, err := s.parser.read() - if err != nil { - if _, ok := err.(*sbuf.Control); ok { - s.sendControl(err) - continue - } - if err == io.EOF { - err = nil - } - s.sendControl(err) - return - } - // Grab a free worker and give it this values message frame to work on - // along with the present Decoder's local-to-shared type state. - // We queue up the worker's resultCh so batches are delivered in order. - select { - case worker := <-s.workerCh: - w := work{ - types: s.parser.types, - frame: frame, - resultCh: make(chan op.Result, 1), - } - select { - case s.resultChCh <- w.resultCh: - select { - case worker.workCh <- w: - case <-s.ctx.Done(): - close(w.resultCh) - return - } - case <-s.ctx.Done(): - return - } - case <-s.ctx.Done(): - return - } - } - }() -} - -// sendControl provides a means for the input thread to send control -// messages and error/EOF in order with the worker threads. -func (s *scanner) sendControl(err error) bool { - ch := make(chan op.Result, 1) - ch <- op.Result{Err: err} - select { - case s.resultChCh <- ch: - return true - case <-s.ctx.Done(): - return false - } -} - -func (s *scanner) Progress() vio.Progress { - return s.progress.Copy() -} - -// worker is used by both the non-threaded synchronous scanner as well as -// the threaded scanner. As long as run() is not called, scanBatch() can -// be safely used without any channel involvement. -type worker struct { - ctx context.Context - progress *vio.Progress - workCh chan work - bufferFilter *expr.BufferFilter - filter expr.Evaluator - validate bool - - typeCache super.TypeCache - vals []super.Value -} - -type work struct { - // Workers need access to the local Decoder types slice to map - // deserialized type IDs into shared-context types and we accomplish - // this using the Decoder's implementation of the super.TypeFetcher - // interface. - types super.TypeFetcher - frame frame - resultCh chan op.Result -} - -func newWorker(ctx context.Context, p *vio.Progress, bf *expr.BufferFilter, f expr.Evaluator, validate bool) *worker { - return &worker{ - ctx: ctx, - progress: p, - workCh: make(chan work), - bufferFilter: bf, - filter: f, - validate: validate, - } -} - -func (w *worker) run(workerCh chan<- *worker) { - for { - // Tell the scanner we're ready for work. - select { - case workerCh <- w: - case <-w.ctx.Done(): - return - } - // Grab the work the scanner gave us. The scanner will arrange - // to pull the result off our resultCh and preserve order. - select { - case work := <-w.workCh: - // If the buffer is compressed, decompress it. - // If not, it wasn't compressed in the original data - // stream and we handle both cases the same from - // here on out The important bit is we are doing - // the decompress and the boyer-moore short-circuit - // scan on a processor cache-friendly buffer and - // throwing it all out asap if it is not needed. - if work.frame.zbuf != nil { - if err := work.frame.decompress(); err != nil { - work.resultCh <- op.Result{Err: err} - continue - } - work.frame.zbuf.free() - } - // Either the frame was compressed or it was uncompressed. - // In either case,the uncompressed data is now in work.blk. - // We hand ownership of ubuf over to scanBatch. the zbuf - // has been freed above so no need to free work.blk. - // If the batch survives, the work.blk.ubuf will go with it - // and will get freed when the batch's Unref count hits 0. - batch, err := w.scanBatch(work.frame.ubuf, work.types) - if batch != nil || err != nil { - work.resultCh <- op.Result{Batch: batch, Err: err} - } - close(work.resultCh) - case <-w.ctx.Done(): - return - } - } -} - -func (w *worker) scanBatch(buf *buffer, types super.TypeFetcher) (sbuf.Batch, error) { - // If w.bufferFilter evaluates to false, we know buf cannot contain - // values matching w.filter. - if w.bufferFilter != nil && !w.bufferFilter.Eval(types, buf.Bytes()) { - atomic.AddInt64(&w.progress.BytesRead, int64(buf.length())) - buf.free() - return nil, nil - } - // Otherwise, build a batch by reading all values in the buffer. - w.typeCache.Reset(types) - w.vals = w.vals[:0] - var progress vio.Progress - for buf.length() > 0 { - var val super.Value - if err := w.decodeVal(buf, &val); err != nil { - buf.free() - return nil, err - } - if w.wantValue(val, &progress) { - w.vals = append(w.vals, val) - } - } - w.progress.Add(progress) - if len(w.vals) == 0 { - buf.free() - return nil, nil - } - return newBatch(buf, w.vals), nil -} - -func (w *worker) decodeVal(buf *buffer, valRef *super.Value) error { - id, err := readUvarintAsInt(buf) - if err != nil { - return err - } - n, err := scode.ReadTag(buf) - if err != nil { - return errBadFormat - } - var b []byte - if n == 0 { - b = []byte{} - } else if n > 0 { - b, err = buf.read(n) - if err != nil && err != io.EOF { - if err == peeker.ErrBufferOverflow { - return fmt.Errorf("large value of %d bytes exceeds maximum read buffer", n) - } - return errBadFormat - } - } - typ, err := w.typeCache.LookupType(id) - if err != nil { - return fmt.Errorf("bsup-rows: %w", err) - } - *valRef = super.NewValue(typ, b) - if w.validate { - if err := valRef.Validate(); err != nil { - return err - } - } - return nil -} - -func (w *worker) wantValue(val super.Value, progress *vio.Progress) bool { - progress.BytesRead += int64(len(val.Bytes())) - progress.RecordsRead++ - if w.filter == nil || expr.IsTrue(w.filter.Eval(val)) { - progress.BytesMatched += int64(len(val.Bytes())) - progress.RecordsMatched++ - return true - } - return false -} diff --git a/bsup/oldbsup/scanner_test.go b/bsup/oldbsup/scanner_test.go deleted file mode 100644 index 56080c92f8..0000000000 --- a/bsup/oldbsup/scanner_test.go +++ /dev/null @@ -1,60 +0,0 @@ -package oldbsup - -import ( - "bytes" - "io" - "runtime" - "strconv" - "testing" - - "github.com/stretchr/testify/require" - "github.com/superdb/super" - "github.com/superdb/super/sio" -) - -func TestScannerContext(t *testing.T) { - // We want to maximize the number of scanner goroutines running - // concurrently, so don't call t.Parallel. - count := runtime.GOMAXPROCS(0) + 1 - var bufs [][]byte - var names []string - var values []any - // Add some BSUP streams to bufs. The records in each stream have a type - // unique to that stream so that they'll only validate if read with the - // correct context. - for i := range count { - names = append(names, strconv.Itoa(i)) - values = append(values, i) - rec, err := super.NewMarshaler(super.NewContext()).MarshalCustom(names, values) - require.NoError(t, err) - var buf bytes.Buffer - w := NewWriter(sio.NopCloser(&buf)) - for range 100 { - require.NoError(t, w.Write(rec)) - } - require.NoError(t, w.EndStream()) - require.NoError(t, w.Close()) - bufs = append(bufs, buf.Bytes()) - } - // Create a validating BSUP reader that repeatedly reads the streams in bufs. - var readers []io.Reader - for range 20 { - for j := range count { - readers = append(readers, bytes.NewReader(bufs[j])) - } - } - r := NewReaderWithOpts(super.NewContext(), io.MultiReader(readers...), ReaderOpts{ - Validate: true, - }) - // Create a scanner and scan, validating each record. - s, err := r.NewScanner(t.Context(), nil) - require.NoError(t, err) - for { - batch, err := s.Pull(false) - require.NoError(t, err) - if batch == nil { - break - } - batch.Unref() - } -} diff --git a/bsup/oldbsup/sync.go b/bsup/oldbsup/sync.go deleted file mode 100644 index 7d6463bd56..0000000000 --- a/bsup/oldbsup/sync.go +++ /dev/null @@ -1,83 +0,0 @@ -package oldbsup - -import ( - "context" - "io" - - "github.com/superdb/super" - "github.com/superdb/super/pkg/peeker" - "github.com/superdb/super/runtime/sam/expr" - "github.com/superdb/super/sbuf" - "github.com/superdb/super/vector/vio" -) - -type scannerSync struct { - ctx context.Context - cancel context.CancelFunc - progress vio.Progress - worker *worker - parser parser - err error - eof bool -} - -func newScannerSync(ctx context.Context, sctx *super.Context, r io.Reader, filter sbuf.Pushdown, opts ReaderOpts) (*scannerSync, error) { - ctx, cancel := context.WithCancel(ctx) - s := &scannerSync{ - ctx: ctx, - cancel: cancel, - parser: parser{ - peeker: peeker.NewReader(r, opts.Size, opts.Max), - types: NewDecoder(sctx), - maxSize: opts.Max, - }, - } - var bf *expr.BufferFilter - var f expr.Evaluator - if filter != nil { - var err error - bf, err = filter.BSUPFilter() - if err != nil { - return nil, err - } - f, err = filter.DataFilter() - if err != nil { - return nil, err - } - } - s.worker = newWorker(ctx, &s.progress, bf, f, opts.Validate) - return s, nil -} - -func (s *scannerSync) Pull(done bool) (sbuf.Batch, error) { - if done { - s.eof = true - return nil, nil - } - if s.err != nil || s.eof { - return nil, s.err - } -again: - frame, err := s.parser.read() - if err != nil { - if err == io.EOF { - err = nil - } - return nil, err - } - if frame.zbuf != nil { - if err := frame.decompress(); err != nil { - return nil, err - } - frame.zbuf.free() - } - b, err := s.worker.scanBatch(frame.ubuf, s.parser.types) - if b == nil && err == nil { - goto again - } - return b, err -} - -func (s *scannerSync) Progress() vio.Progress { - return s.progress.Copy() -} diff --git a/bsup/oldbsup/types.go b/bsup/oldbsup/types.go deleted file mode 100644 index e8924b2ad9..0000000000 --- a/bsup/oldbsup/types.go +++ /dev/null @@ -1,109 +0,0 @@ -package oldbsup - -import ( - "errors" - "fmt" - "sync" - - "github.com/superdb/super" -) - -type Encoder struct { - defs *super.TypeDefs - off int -} - -func NewEncoder() *Encoder { - return &Encoder{ - defs: super.NewTypeDefs(), - } -} - -func (e *Encoder) Reset() { - e.defs.Reset() - e.off = 0 -} - -func (e *Encoder) Flush() { - e.off = len(e.defs.Bytes()) -} - -func (e *Encoder) Len() int { - return len(e.defs.Bytes()) - e.off -} - -func (e *Encoder) nextBuffer() []byte { - bytes := e.defs.Bytes() - b := bytes[e.off:] - e.off = len(bytes) - return b -} - -// Encode takes a type from outside this context and constructs a type from -// inside this context and emits BSUP typedefs for any type needed to construct -// the new type into the buffer provided. -func (e *Encoder) Encode(external super.Type) uint32 { - // In general, a type can have a forward reference to another named type - // that isn't defined here yet, so we need to recursively bind/patch - // named types in the local type context, which a mapper can do but it - // needs the original type context (actually just the original typedefs - // table from the query sctx) - return e.defs.LookupType(external) -} - -type Decoder struct { - sctx *super.Context - defs *super.TypeDefs - mapper *super.TypeDefsMapper - // Local type IDs are mapped to the shared-context types with the types array. - // The types slice is protected with mutex as the slice can be expanded while - // worker threads are scanning earlier batches. - mu sync.RWMutex - types []super.Type -} - -var _ super.TypeFetcher = (*Decoder)(nil) - -func NewDecoder(sctx *super.Context) *Decoder { - defs := super.NewTypeDefs() - return &Decoder{ - sctx: sctx, - defs: defs, - mapper: super.NewTypeDefsMapper(sctx, defs), - } -} - -func (d *Decoder) decode(b *buffer) error { - before := d.defs.NTypes() - if ok := d.defs.AppendBytes(b.Bytes()); !ok { - return errors.New("corrupt BSUP types frame") - } - after := d.defs.NTypes() - for id := before; id < after; id++ { - typ := d.mapper.LookupType(uint32(id + super.IDTypeComplex)) - if typ == nil { - return errors.New("corrupt BSUP types frame") - } - // Even though type decoding is single threaded, workers processing a - // previous batch can be accessing the types map (via LookupType) while - // the single thread is extending it so these accesses are protected - // with the mutex. - d.mu.Lock() - d.types = append(d.types, typ) - d.mu.Unlock() - } - return nil -} - -func (d *Decoder) LookupType(id int) (super.Type, error) { - if id < super.IDTypeComplex { - return super.LookupPrimitiveByID(id) - } - d.mu.RLock() - defer d.mu.RUnlock() - off := id - super.IDTypeComplex - if off < len(d.types) { - return d.types[off], nil - } - return nil, fmt.Errorf("no type found for type id %d", id) -} diff --git a/bsup/oldbsup/writer.go b/bsup/oldbsup/writer.go deleted file mode 100644 index fe337064e0..0000000000 --- a/bsup/oldbsup/writer.go +++ /dev/null @@ -1,186 +0,0 @@ -package oldbsup - -import ( - "encoding/binary" - "io" - "slices" - - "github.com/pierrec/lz4/v4" - "github.com/superdb/super" - "github.com/superdb/super/sbuf" - "github.com/superdb/super/scode" - "github.com/superdb/super/vector" -) - -// DefaultFrameThresh is a reasonable default for WriterOpts.FrameThresh. -const DefaultFrameThresh = 512 * 1024 - -type Writer struct { - writer io.WriteCloser - position int64 - flushed int64 - compressor *compressor - - types *Encoder - values []byte - header []byte -} - -// NewWriterWithOpts returns a writer to w with opts. -func NewWriter(w io.WriteCloser) *Writer { - return &Writer{ - writer: w, - compressor: &compressor{}, - types: NewEncoder(), - } -} - -func (w *Writer) Push(vec vector.Any) error { - return sbuf.WriteVec(w, vec) -} - -func (w *Writer) DisableCompression() { - w.compressor = nil -} - -func (w *Writer) Close() error { - err := w.EndStream() - if closeErr := w.writer.Close(); err == nil { - err = closeErr - } - return err -} - -func (w *Writer) write(p []byte) error { - n, err := w.writer.Write(p) - if err != nil { - return err - } - w.position += int64(n) - return nil -} - -// Position may be called after EndStream to get a seekable offset into the -// output for the next stream. Calling Position at any other team returns -// unusable seek offsets. -func (w *Writer) Position() int64 { - return w.position -} - -func (w *Writer) EndStream() error { - // Flush any compression state and write the EOS afterward the - // compressed block since the buffer-filter may skip entire - // compressed before and we would otherwise miss the EOS marker. - if err := w.flush(); err != nil { - return err - } - if w.flushed != w.position { - if err := w.write([]byte{EOS}); err != nil { - return err - } - w.flushed = w.position - } - w.types.Reset() - return nil -} - -func (w *Writer) Write(val super.Value) error { - id := w.types.Encode(val.Type()) - w.values = binary.AppendUvarint(w.values, uint64(id)) - w.values = scode.Append(w.values, val.Bytes()) - if thresh := DefaultFrameThresh; len(w.values) >= thresh || w.types.Len() >= thresh { - return w.flush() - } - return nil -} - -func (w *Writer) WriteControl(b []byte, format uint8) error { - // Flush the compressor since we need to preserve the interleaving - // order of control messages and BSUPROW data. - if err := w.flush(); err != nil { - return err - } - // Yuck. - bytes := make([]byte, len(b)+1) - bytes[0] = format - copy(bytes[1:], b) - return w.writeBlock(ControlFrame, bytes) -} - -func (w *Writer) flush() error { - if err := w.writeBlock(TypesFrame, w.types.nextBuffer()); err != nil { - return nil - } - if err := w.writeBlock(ValuesFrame, w.values); err != nil { - return nil - } - w.values = w.values[:0] - return nil -} - -func (w *Writer) writeBlock(blockType int, b []byte) error { - if len(b) == 0 { - return nil - } - if w.compressor != nil { - sbuf, err := w.compressor.compress(b) - if err != nil { - return err - } - if sbuf != nil { - if err := w.writeCompHeader(blockType, len(b), len(sbuf)); err != nil { - return err - } - return w.write(sbuf) - } - } - if err := w.writeHeader(blockType, len(b)); err != nil { - return err - } - return w.write(b) -} - -func (w *Writer) writeHeader(blockType, size int) error { - version := Version | 0x80 - code := blockType<<4 | (size & 0xf) - w.header = append(w.header[:0], byte(version)) - w.header = append(w.header, byte(code)) - w.header = binary.AppendUvarint(w.header, uint64(size>>4)) - return w.write(w.header) -} - -func (w *Writer) writeCompHeader(blockType, size, zlen int) error { - zlen += 1 + scode.SizeOfUvarint(uint64(size)) - version := Version | 0x80 - code := (blockType << 4) | (zlen & 0xf) | 0x40 - w.header = append(w.header[:0], byte(version)) - w.header = append(w.header, byte(code)) - w.header = binary.AppendUvarint(w.header, uint64(zlen>>4)) - w.header = append(w.header, byte(CompressionFormatLZ4)) - w.header = binary.AppendUvarint(w.header, uint64(size)) - return w.write(w.header) -} - -type compressor struct { - compressor lz4.Compressor - sbuf []byte -} - -func (c *compressor) compress(b []byte) ([]byte, error) { - if c == nil || len(b) == 0 { - return nil, nil - } - c.sbuf = slices.Grow(c.sbuf[:0], len(b)) - sbuf := c.sbuf[:len(b)] - zlen, err := c.compressor.CompressBlock(b, sbuf) - if err != nil && err != lz4.ErrInvalidSourceShortBuffer { - return nil, err - } - if zlen > 0 { - // Compression succeeded and the compressed value message block - // is smaller than the buffered messages, so write the - // compressed value message block. - return sbuf[:zlen], nil - } - return nil, nil -} diff --git a/bsup/oldbsup/ztests/complex-named.yaml b/bsup/oldbsup/ztests/complex-named.yaml deleted file mode 100644 index e7197862ac..0000000000 --- a/bsup/oldbsup/ztests/complex-named.yaml +++ /dev/null @@ -1,14 +0,0 @@ -# Do a round-trip into serialized BSUP to make sure named types are encoded properly. - -script: super - | super -s - - -inputs: - - name: stdin - data: &stdin | - type bar=string - ["hello"::bar,"world"::bar] - {a:["hello"::bar,"world"::bar]} - -outputs: - - name: stdout - data: *stdin diff --git a/bsup/oldbsup/ztests/dev-zero.yaml b/bsup/oldbsup/ztests/dev-zero.yaml deleted file mode 100644 index 2e9dc984c5..0000000000 --- a/bsup/oldbsup/ztests/dev-zero.yaml +++ /dev/null @@ -1,7 +0,0 @@ -script: | - ! super -i bsuprows /dev/zero - -outputs: - - name: stderr - data: | - /dev/zero: bsup-rows version mismatch: expected 6, found 0 diff --git a/bsup/oldbsup/ztests/esc-typeid.yaml b/bsup/oldbsup/ztests/esc-typeid.yaml deleted file mode 100644 index 6a9b88deff..0000000000 --- a/bsup/oldbsup/ztests/esc-typeid.yaml +++ /dev/null @@ -1,236 +0,0 @@ -script: super in.sup | super -s - - -inputs: - - name: in.sup - data: &input | - {f0:0} - {f1:1} - {f2:2} - {f3:3} - {f4:4} - {f5:5} - {f6:6} - {f7:7} - {f8:8} - {f9:9} - {f10:10} - {f11:11} - {f12:12} - {f13:13} - {f14:14} - {f15:15} - {f16:16} - {f17:17} - {f18:18} - {f19:19} - {f20:20} - {f21:21} - {f22:22} - {f23:23} - {f24:24} - {f25:25} - {f26:26} - {f27:27} - {f28:28} - {f29:29} - {f30:30} - {f31:31} - {f32:32} - {f33:33} - {f34:34} - {f35:35} - {f36:36} - {f37:37} - {f38:38} - {f39:39} - {f40:40} - {f41:41} - {f42:42} - {f43:43} - {f44:44} - {f45:45} - {f46:46} - {f47:47} - {f48:48} - {f49:49} - {f50:50} - {f51:51} - {f52:52} - {f53:53} - {f54:54} - {f55:55} - {f56:56} - {f57:57} - {f58:58} - {f59:59} - {f60:60} - {f61:61} - {f62:62} - {f63:63} - {f64:64} - {f65:65} - {f66:66} - {f67:67} - {f68:68} - {f69:69} - {f70:70} - {f71:71} - {f72:72} - {f73:73} - {f74:74} - {f75:75} - {f76:76} - {f77:77} - {f78:78} - {f79:79} - {f80:80} - {f81:81} - {f82:82} - {f83:83} - {f84:84} - {f85:85} - {f86:86} - {f87:87} - {f88:88} - {f89:89} - {f90:90} - {f91:91} - {f92:92} - {f93:93} - {f94:94} - {f95:95} - {f96:96} - {f97:97} - {f98:98} - {f99:99} - {f100:100} - {f101:101} - {f102:102} - {f103:103} - {f104:104} - {f105:105} - {f106:106} - {f107:107} - {f108:108} - {f109:109} - {f110:110} - {f111:111} - {f112:112} - {f113:113} - {f114:114} - {f115:115} - {f116:116} - {f117:117} - {f118:118} - {f119:119} - {f120:120} - {f121:121} - {f122:122} - {f123:123} - {f124:124} - {f125:125} - {f126:126} - {f127:127} - {f128:128} - {f129:129} - {f130:130} - {f131:131} - {f132:132} - {f133:133} - {f134:134} - {f135:135} - {f136:136} - {f137:137} - {f138:138} - {f139:139} - {f140:140} - {f141:141} - {f142:142} - {f143:143} - {f144:144} - {f145:145} - {f146:146} - {f147:147} - {f148:148} - {f149:149} - {f150:150} - {f151:151} - {f152:152} - {f153:153} - {f154:154} - {f155:155} - {f156:156} - {f157:157} - {f158:158} - {f159:159} - {f160:160} - {f161:161} - {f162:162} - {f163:163} - {f164:164} - {f165:165} - {f166:166} - {f167:167} - {f168:168} - {f169:169} - {f170:170} - {f171:171} - {f172:172} - {f173:173} - {f174:174} - {f175:175} - {f176:176} - {f177:177} - {f178:178} - {f179:179} - {f180:180} - {f181:181} - {f182:182} - {f183:183} - {f184:184} - {f185:185} - {f186:186} - {f187:187} - {f188:188} - {f189:189} - {f190:190} - {f191:191} - {f192:192} - {f193:193} - {f194:194} - {f195:195} - {f196:196} - {f197:197} - {f198:198} - {f199:199} - {f200:200} - {f201:201} - {f202:202} - {f203:203} - {f204:204} - {f205:205} - {f206:206} - {f207:207} - {f208:208} - {f209:209} - {f210:210} - {f211:211} - {f212:212} - {f213:213} - {f214:214} - {f215:215} - {f216:216} - {f217:217} - {f218:218} - {f219:219} - {f220:220} - {f221:221} - {f222:222} - {f223:223} - {f224:224} - {f225:225} - {f226:226} - -outputs: - - name: stdout - data: *input diff --git a/bsup/oldbsup/ztests/issue-4082.yaml b/bsup/oldbsup/ztests/issue-4082.yaml deleted file mode 100644 index b972d8c033..0000000000 --- a/bsup/oldbsup/ztests/issue-4082.yaml +++ /dev/null @@ -1,17 +0,0 @@ -script: | - ! super -i bsuprows - - -inputs: - - name: stdin - # Generated with `bash -c 'super -o f.bsuprows - <<<"[1,2]" && base64 -b 70 f.bsuprows'`. - data: !!binary | - Vk5HAA0AAAC7AAAAAAAAACgAAAAAAAAAAQAAAIFGCQCXAfQJAAQGT2Zmc2V0AwAGTGVuZ3 - RoAwAJTWVtDADyRRFDb21wcmVzc2lvbkZvcm1hdAAABwdTZWdtZW50HgAFBFR5cGUcAAhM - b2NhdGlvbh8AA01pbgkAA01heAkABUNvdW50AgAHA0ludCAHAklEAgADBlwA8A4CAAdMZW - 5ndGhzHwAGVmFsdWVzIgAHBUFycmF5I4EeASERAgkIAhACEAIYAQICAgQCAiQLAgEHAQII - AhABAf8CAAAAAwAAAICAAgADAAAAAgAAAAIAAACAgICAgIABAQIAAAAAAAAA - -outputs: - - name: stderr - data: | - stdio:stdin: bsup-rows version mismatch: expected 6, found 0 diff --git a/bsup/oldbsup/ztests/outer-named.yaml b/bsup/oldbsup/ztests/outer-named.yaml deleted file mode 100644 index 2a741da7e8..0000000000 --- a/bsup/oldbsup/ztests/outer-named.yaml +++ /dev/null @@ -1,13 +0,0 @@ -script: super - | super -s -|super - | super -s - - -inputs: - - name: stdin - data: &stdin | - type typ=int64 - 0::typ - type typ2={x:int64} - {x:0}::typ2 - -outputs: - - name: stdout - data: *stdin diff --git a/bsup/oldbsup/ztests/primitive.yaml b/bsup/oldbsup/ztests/primitive.yaml deleted file mode 100644 index 3d8f7d2676..0000000000 --- a/bsup/oldbsup/ztests/primitive.yaml +++ /dev/null @@ -1,16 +0,0 @@ -script: | - super -f bsuprows -i sup - | super -i bsuprows -s - - -inputs: - - name: stdin - data: | - 1 - "hello" - error("foo") - -outputs: - - name: stdout - data: | - 1 - "hello" - error("foo") diff --git a/bsup/oldbsup/ztests/pushdown-filter.yaml b/bsup/oldbsup/ztests/pushdown-filter.yaml deleted file mode 100644 index 72d6483b4a..0000000000 --- a/bsup/oldbsup/ztests/pushdown-filter.yaml +++ /dev/null @@ -1,15 +0,0 @@ -# Test that union true values get de-unioned in filter pushdowns. -script: | - super -f bsuprows in.sup > test.bsuprows - super -s -c 'from test.bsuprows | where this' - -inputs: - - name: in.sup - data: | - true::(bool|null) - false::(bool|null) - -outputs: - - name: stdout - data: | - true::(bool|null) diff --git a/bsup/oldbsup/ztests/string-set.yaml b/bsup/oldbsup/ztests/string-set.yaml deleted file mode 100644 index 21bd3037a8..0000000000 --- a/bsup/oldbsup/ztests/string-set.yaml +++ /dev/null @@ -1,16 +0,0 @@ -script: super - | super -s - - -inputs: - - name: stdin - data: &stdin | - {f:null::(null|set[string])} - {f:set[]::set[string]} - {f:set[null]::set[string|null]} - {f:set["dup"]} - {f:set["a","z"]} - {f:set["a","z"]} - {f:set["a","z",null,"dup"]} - -outputs: - - name: stdout - data: *stdin diff --git a/bsup/oldbsup/ztests/union-array.yaml b/bsup/oldbsup/ztests/union-array.yaml deleted file mode 100644 index c44f72baf6..0000000000 --- a/bsup/oldbsup/ztests/union-array.yaml +++ /dev/null @@ -1,11 +0,0 @@ -script: super - | super -s - - -inputs: - - name: stdin - data: &stdin | - {a:["asdfasdf",null,100]} - {a:null::(null|[int64|string|null])} - -outputs: - - name: stdout - data: *stdin diff --git a/bsup/oldbsup/ztests/union-nested.yaml b/bsup/oldbsup/ztests/union-nested.yaml deleted file mode 100644 index 14fea8209c..0000000000 --- a/bsup/oldbsup/ztests/union-nested.yaml +++ /dev/null @@ -1,14 +0,0 @@ -script: super - | super -s - - -inputs: - - name: stdin - data: &stdin | - type foo=int64|string - {a:"\"hello\""::(foo|string|[int64]|[string])} - {a:[1,2]::(foo|string|[int64]|[string])} - {a:["\"a\"","\"b\""]::(foo|string|[int64]|[string])} - {a:123::foo::(foo|string|[int64]|[string])} - -outputs: - - name: stdout - data: *stdin diff --git a/bsup/oldbsup/ztests/union-simple.yaml b/bsup/oldbsup/ztests/union-simple.yaml deleted file mode 100644 index f298b0e0bb..0000000000 --- a/bsup/oldbsup/ztests/union-simple.yaml +++ /dev/null @@ -1,12 +0,0 @@ -script: super - | super -s - - -inputs: - - name: stdin - data: &stdin | - {a:"asdfasdf"::(int64|string)} - {a:1::(int64|string)} - {a:null::(int64|null)} - -outputs: - - name: stdout - data: *stdin diff --git a/bsup/oldbsup/ztests/zctx-named-reset-2.yaml b/bsup/oldbsup/ztests/zctx-named-reset-2.yaml deleted file mode 100644 index 91e0adc633..0000000000 --- a/bsup/oldbsup/ztests/zctx-named-reset-2.yaml +++ /dev/null @@ -1,19 +0,0 @@ -# Test that type names are properly reset and reusable after stream boundaries. -script: | - super -c "head 1" in.sup > t1.bsuprows - super -c "tail 2" in.sup > t2.bsuprows - super -s -c "where proto.is_ok() | count() by proto" t1.bsuprows t2.bsuprows - -inputs: - - name: in.sup - data: | - {ts:2015-03-05T14:25:12.963801Z} - type zenum=string - {ts:2015-03-05T14:25:14.419939Z,proto:"udp"::zenum} - {ts:2015-03-05T14:25:14.419939Z,proto:"udp"::zenum} - -outputs: - - name: stdout - data: | - type zenum=string - {proto:"udp"::zenum,count:2} diff --git a/bsup/oldbsup/ztests/zctx-named-reset.yaml b/bsup/oldbsup/ztests/zctx-named-reset.yaml deleted file mode 100644 index bc25cd537f..0000000000 --- a/bsup/oldbsup/ztests/zctx-named-reset.yaml +++ /dev/null @@ -1,22 +0,0 @@ -# Test that type names are properly reset and reusable after stream boundaries. - -script: | - super -c "head 1" in.sup > t1.bsuprows - super -c "tail 2" in.sup > t2.bsuprows - cat t1.bsuprows t2.bsuprows |super - | super -s - - -inputs: - - name: in.sup - data: | - type zenum=string - {ts:2015-03-05T14:25:12.963801Z} - {ts:2015-03-05T14:25:14.419939Z,proto:"udp"::zenum} - {ts:2015-03-05T14:25:14.419939Z,proto:"udp"::zenum} - -outputs: - - name: stdout - data: | - {ts:2015-03-05T14:25:12.963801Z} - type zenum=string - {ts:2015-03-05T14:25:14.419939Z,proto:"udp"::zenum} - {ts:2015-03-05T14:25:14.419939Z,proto:"udp"::zenum} diff --git a/bsup/oldbsup/ztests/zctx-reset.yaml b/bsup/oldbsup/ztests/zctx-reset.yaml deleted file mode 100644 index cbbc47d4fb..0000000000 --- a/bsup/oldbsup/ztests/zctx-reset.yaml +++ /dev/null @@ -1,20 +0,0 @@ -# Test that type contexts are properly reset and reusable after stream boundaries - -script: | - super in.sup > s.bsuprows - super in.sup >> s.bsuprows - super -s s.bsuprows - -inputs: - - name: in.sup - data: | - {a:"hello"} - {b:10} - -outputs: - - name: stdout - data: | - {a:"hello"} - {b:10} - {a:"hello"} - {b:10} diff --git a/cli/inputflags/flags.go b/cli/inputflags/flags.go index e6ddbcc547..f37100eb7f 100644 --- a/cli/inputflags/flags.go +++ b/cli/inputflags/flags.go @@ -25,7 +25,7 @@ func (f *Flags) SetFlags(fs *flag.FlagSet) { }) fs.BoolVar(&f.Dynamic, "dynamic", false, "disable static type checking of inputs") - fs.StringVar(&opts.Format, "i", "auto", "format of input data [auto,arrows,bsuprows,bsup,csv,json,line,parquet,sup,tsv,zeek]") + fs.StringVar(&opts.Format, "i", "auto", "format of input data [auto,arrows,bsup,csv,json,line,parquet,sup,tsv,zeek]") fs.BoolVar(&f.Static, "static", false, "force static type checking of inputs") } diff --git a/cli/outputflags/flags.go b/cli/outputflags/flags.go index 05157c5279..34eece7312 100644 --- a/cli/outputflags/flags.go +++ b/cli/outputflags/flags.go @@ -68,7 +68,7 @@ func (f *Flags) SetFormatFlags(fs *flag.FlagSet) { f.Format = "bsup" } fUsage := fmt.Sprintf( - "format for output data [arrows,bsuprows,bsup,csv,db,json,line,parquet,sup,table,tsv,zeek] (default %s)", + "format for output data [arrows,bsup,csv,db,json,line,parquet,sup,table,tsv,zeek] (default %s)", f.Format) fs.Func("f", fUsage, func(s string) error { f.Format = s @@ -124,7 +124,7 @@ func (f *Flags) Init() error { } func isBinary(fmt string) bool { - return fmt == "arrows" || fmt == "bsuprows" || fmt == "bsup" || fmt == "parquet" + return fmt == "arrows" || fmt == "bsup" || fmt == "parquet" } func (f *Flags) FileName() string { diff --git a/cmd/super/db/command.go b/cmd/super/db/command.go index a81bf81b5c..9cff1b8200 100644 --- a/cmd/super/db/command.go +++ b/cmd/super/db/command.go @@ -47,7 +47,7 @@ func New(parent charm.Command, f *flag.FlagSet) (charm.Command, error) { } func (c *Command) SetLeafFlags(f *flag.FlagSet) { - c.outputFlags.Format = "bsuprows" + c.outputFlags.Format = "bsup" c.outputFlags.SetFlags(f) c.queryFlags.SetFlags(f) c.runtimeFlags.SetFlags(f) diff --git a/cmd/super/ztests/o-flag.yaml b/cmd/super/ztests/o-flag.yaml index ccdb49ef53..1788d2b539 100644 --- a/cmd/super/ztests/o-flag.yaml +++ b/cmd/super/ztests/o-flag.yaml @@ -1,8 +1,8 @@ script: | - for ext in arrows bsuprows bsup csv json jsonl ndjson parquet sup text tsv txt; do + for ext in arrows bsup csv json jsonl ndjson parquet sup text tsv txt; do super -o out.$ext in.sup done - for ext in arrows bsuprows bsup parquet; do + for ext in arrows bsup parquet; do super -i $ext -o out.$ext.sup out.$ext done # An unknown extention yields the default output format (BSUP). @@ -28,8 +28,6 @@ inputs: outputs: - name: out.arrows.sup data: *in_sup - - name: out.bsuprows.sup - data: *in_sup - name: out.bsup.sup data: *in_sup - name: out.csv diff --git a/db/data/object.go b/db/data/object.go index 8e25ad84c4..b511d8095f 100644 --- a/db/data/object.go +++ b/db/data/object.go @@ -39,7 +39,7 @@ func (k FileKind) Description() string { } } -var fileRegex = regexp.MustCompile(`([0-9A-Za-z]{27}-(data|meta)).bsuprows$`) +var fileRegex = regexp.MustCompile(`([0-9A-Za-z]{27}-(data|meta)).bsup$`) // XXX this won't work right until we integrate segID func FileMatch(s string) (kind FileKind, id uuid.UUID, ok bool) { diff --git a/db/ztests/ls.yaml b/db/ztests/ls.yaml index ade0929d59..ae8761117d 100644 --- a/db/ztests/ls.yaml +++ b/db/ztests/ls.yaml @@ -3,7 +3,7 @@ script: | super db init -q super db create -q logs super db load -q -use logs babble.sup - super db ls -f bsuprows | super -S -c "drop id,ts" - + super db ls -f bsup | super -S -c "drop id,ts" - echo === super db -S -c "from logs@main:objects | drop id | min:=defuse(min),max:=defuse(max)" diff --git a/db/ztests/s3/rm.yaml b/db/ztests/s3/rm.yaml index 20ae4f7d96..92141afd79 100644 --- a/db/ztests/s3/rm.yaml +++ b/db/ztests/s3/rm.yaml @@ -5,13 +5,13 @@ script: | source s3.sh aws s3api create-bucket --bucket bucket super db import -R logs -data s3://bucket/db babble.sup - super db map -R logs -o count.bsuprows "count()" + super db map -R logs -o count.bsup "count()" echo === super db ls -R logs -ranges -l echo === - super db rm -R logs -ranges count.bsuprows + super db rm -R logs -ranges count.bsup echo === - super db rm -R logs -ranges count.bsuprows + super db rm -R logs -ranges count.bsup inputs: - name: babble.sup @@ -23,11 +23,11 @@ outputs: - name: stdout data: | === - [1587518620062237300-1587513611063914690]/count.bsuprows - [1587513592062544400-1587508830068523240]/count.bsuprows + [1587518620062237300-1587513611063914690]/count.bsup + [1587513592062544400-1587508830068523240]/count.bsup === - [1587518620062237300-1587513611063914690]/count.bsuprows: removed - [1587513592062544400-1587508830068523240]/count.bsuprows: removed + [1587518620062237300-1587513611063914690]/count.bsup: removed + [1587513592062544400-1587508830068523240]/count.bsup: removed === - [1587518620062237300-1587513611063914690]/count.bsuprows: not found - [1587513592062544400-1587508830068523240]/count.bsuprows: not found + [1587518620062237300-1587513611063914690]/count.bsup: not found + [1587513592062544400-1587508830068523240]/count.bsup: not found diff --git a/fuzz/fuzz.go b/fuzz/fuzz.go index 20ccbfdd68..525c15eee2 100644 --- a/fuzz/fuzz.go +++ b/fuzz/fuzz.go @@ -14,7 +14,6 @@ import ( "github.com/stretchr/testify/require" "github.com/superdb/super" "github.com/superdb/super/bsup" - "github.com/superdb/super/bsup/oldbsup" "github.com/superdb/super/compiler" "github.com/superdb/super/compiler/optimizer" "github.com/superdb/super/compiler/optimizer/demand" @@ -32,12 +31,14 @@ import ( "github.com/x448/float16" ) -func ReadBSUPRows(sctx *super.Context, bs []byte) ([]super.Value, error) { +func ReadBSUPRows(ctx context.Context, sctx *super.Context, bs []byte) ([]super.Value, error) { bytesReader := bytes.NewReader(bs) - reader := oldbsup.NewReader(sctx, bytesReader) - defer reader.Close() + reader, err := bsupio.NewReader(ctx, sctx, bytesReader, nil, 1) + if err != nil { + return nil, err + } var a sbuf.Array - err := sio.Copy(&a, reader) + err = sio.Copy(&a, sbuf.PullerReader(sbuf.NewMaterializer(reader))) if err != nil { return nil, err } diff --git a/service/ztests/curl-load-error.yaml b/service/ztests/curl-load-error.yaml index a560467339..32bae2453d 100644 --- a/service/ztests/curl-load-error.yaml +++ b/service/ztests/curl-load-error.yaml @@ -16,7 +16,7 @@ inputs: outputs: - name: stdout data: | - {"type":"Error","kind":"invalid operation","error":"format detection error\n\tarrows: schema message length exceeds 1 MiB\n\tbsup: unknown header magic: \"This\"\n\tbsuprows: bsup-rows version mismatch: expected 6, found 0\n\tcsv: line 1: EOF\n\tjson: line 1: invalid JSON value\n\tline: auto-detection not supported\n\tparquet: invalid header\n\tsup: line 1: syntax error\n\ttsv: line 1: EOF\n\tzeek: line 1: bad types/fields definition in zeek header"} + {"type":"Error","kind":"invalid operation","error":"format detection error\n\tarrows: schema message length exceeds 1 MiB\n\tbsup: unknown header magic: \"This\"\n\tcsv: line 1: EOF\n\tjson: line 1: invalid JSON value\n\tline: auto-detection not supported\n\tparquet: invalid header\n\tsup: line 1: syntax error\n\ttsv: line 1: EOF\n\tzeek: line 1: bad types/fields definition in zeek header"} code 400 {"type":"Error","kind":"invalid operation","error":"unsupported MIME type: unsupported"} code 400 diff --git a/service/ztests/drop.yaml b/service/ztests/drop.yaml index a44d6d8d52..448ac62600 100644 --- a/service/ztests/drop.yaml +++ b/service/ztests/drop.yaml @@ -5,7 +5,7 @@ script: | super db create -q p3 super db drop -f p3 echo === | tee /dev/stderr - super db ls -f bsuprows | super -s -c "cut name | sort name" - + super db ls -f bsup | super -s -c "cut name | sort name" - echo === | tee /dev/stderr ! super db drop p3 ! super db drop -db http://127.0.0.1:1 p3 diff --git a/service/ztests/load-garbage.yaml b/service/ztests/load-garbage.yaml index 296d0e024d..9dad0ab80c 100644 --- a/service/ztests/load-garbage.yaml +++ b/service/ztests/load-garbage.yaml @@ -15,7 +15,6 @@ outputs: stdio:stdin: format detection error arrows: schema message length exceeds 1 MiB bsup: unknown header magic: "This" - bsuprows: bsup-rows version mismatch: expected 6, found 0 csv: line 1: delimiter ',' not found json: line 1: invalid JSON value line: auto-detection not supported diff --git a/service/ztests/ls-pools.yaml b/service/ztests/ls-pools.yaml index 8aefbf723a..47feb96517 100644 --- a/service/ztests/ls-pools.yaml +++ b/service/ztests/ls-pools.yaml @@ -11,7 +11,7 @@ script: | # Verify that output flags are honored. super db ls -f sup | super -i sup -f sup -c "cut name | sort name" - echo === - super db ls -f bsuprows | super -i bsuprows -f table -c "cut name | sort name" - + super db ls -f bsup | super -i bsup -f table -c "cut name | sort name" - echo === inputs: diff --git a/sio/anyio/lookup.go b/sio/anyio/lookup.go index 9e4bc06232..2fb6734dd6 100644 --- a/sio/anyio/lookup.go +++ b/sio/anyio/lookup.go @@ -6,7 +6,6 @@ import ( "io" "github.com/superdb/super" - "github.com/superdb/super/bsup/oldbsup" "github.com/superdb/super/sbuf" "github.com/superdb/super/sio" "github.com/superdb/super/sio/arrowio" @@ -28,12 +27,6 @@ func lookupReader(ctx context.Context, sctx *super.Context, r io.Reader, opts Re return nil, err } return newVioPuller(sctx, r), nil - case "bsuprows": - scanner, err := oldbsup.NewReader(sctx, r).NewScanner(ctx, opts.Pushdown) - if err != nil { - return nil, err - } - return sbuf.NewDematerializer(sctx, scanner), nil case "bsup": return bsupio.NewReader(ctx, sctx, r, opts.Pushdown, opts.ConcurrentReaders) case "csv": diff --git a/sio/anyio/reader.go b/sio/anyio/reader.go index 40d4b71cd3..40dd5b5ce5 100644 --- a/sio/anyio/reader.go +++ b/sio/anyio/reader.go @@ -12,7 +12,6 @@ import ( "github.com/superdb/super" "github.com/superdb/super/bsup" - "github.com/superdb/super/bsup/oldbsup" "github.com/superdb/super/sbuf" "github.com/superdb/super/sio" "github.com/superdb/super/sio/arrowio" @@ -86,25 +85,6 @@ func NewReader(ctx context.Context, sctx *super.Context, r io.Reader, opts Reade } track.Reset() - // XXX BSUP row options will be updated in a subsequent PR. - // For the matching reader, force validation to true so we are extra - // careful about auto-matching BSUP. Then, once matched, relaxed - // validation to the user setting in the actual reader returned. - //bsupOpts := opts.BSUP - //bsupOpts.Validate = true - bsupRowsReader := oldbsup.NewReader(super.NewContext(), track) - bsupRowsErr := match(bsupRowsReader, "bsuprows", 1) - // Close bsupReader to ensure that it does not continue to call track.Read. - bsupRowsReader.Close() - if bsupRowsErr == nil { - scanner, err := oldbsup.NewReader(sctx, track.Reader()).NewScanner(ctx, opts.Pushdown) - if err != nil { - return nil, err - } - return sbuf.NewDematerializer(sctx, scanner), nil - } - track.Reset() - csvErr := isCSVStream(track, ',', "csv") if csvErr == nil { return newVioPuller(sctx, csvio.NewReader(sctx, track.Reader(), csvio.ReaderOpts{Delim: ','})), nil @@ -121,7 +101,6 @@ func NewReader(ctx context.Context, sctx *super.Context, r io.Reader, opts Reade return nil, joinErrs([]error{ arrowsErr, bsupErr, - bsupRowsErr, csvErr, jsonErr, lineErr, diff --git a/sio/anyio/writer.go b/sio/anyio/writer.go index 986ad57c2e..4889ed95d3 100644 --- a/sio/anyio/writer.go +++ b/sio/anyio/writer.go @@ -6,7 +6,6 @@ import ( "github.com/superdb/super" "github.com/superdb/super/bsup" - "github.com/superdb/super/bsup/oldbsup" "github.com/superdb/super/runtime/expr" "github.com/superdb/super/sio/arrowio" "github.com/superdb/super/sio/csvio" @@ -35,8 +34,6 @@ func NewWriter(w io.WriteCloser, opts WriterOpts) (vio.PushCloser, error) { switch opts.Format { case "arrows": return newDefuser(arrowio.NewWriter(w)), nil - case "bsuprows": // XXX this will change to use -rows flag, e.g., -f bsup -rows - return oldbsup.NewWriter(w), nil case "bsup": return bsup.NewWriterWithOpts(w, opts.BSUP), nil case "csv": diff --git a/sio/anyio/ztests/detector-dev-zero.yaml b/sio/anyio/ztests/detector-dev-zero.yaml index b2fc53fb3d..28b6db3d91 100644 --- a/sio/anyio/ztests/detector-dev-zero.yaml +++ b/sio/anyio/ztests/detector-dev-zero.yaml @@ -7,7 +7,6 @@ outputs: /dev/zero: format detection error arrows: arrow/ipc: could not read message schema: EOF bsup: unknown header magic: "\u0000\u0000\u0000\u0000" - bsuprows: bsup-rows version mismatch: expected 6, found 0 csv: line 1: bufio: buffer full json: line 1: value exceeded max buffer size line: auto-detection not supported diff --git a/sio/anyio/ztests/fake-bsup.yaml b/sio/anyio/ztests/fake-bsup.yaml deleted file mode 100644 index 63a5262454..0000000000 --- a/sio/anyio/ztests/fake-bsup.yaml +++ /dev/null @@ -1,12 +0,0 @@ -script: | - ! super -i bsuprows -s - - -inputs: - - name: stdin - data: !!binary | - BQAAAAQF4CRQAHhACAv8= - -outputs: - - name: stderr - data: | - stdio:stdin: bsup-rows version mismatch: expected 6, found 0 diff --git a/sio/anyio/ztests/huge.yaml b/sio/anyio/ztests/huge.yaml index bec60ce4db..07789da295 100644 --- a/sio/anyio/ztests/huge.yaml +++ b/sio/anyio/ztests/huge.yaml @@ -13,7 +13,6 @@ outputs: stdio:stdin: format detection error arrows: schema message length exceeds 1 MiB bsup: unknown header magic: " \n \n" - bsuprows: bsup-rows version mismatch: expected 6, found 0 csv: line 1: delimiter ',' not found json: buffer exceeded max size trying to infer input format line: auto-detection not supported diff --git a/sio/sio.go b/sio/sio.go index 06392fa9fb..677dd0bba8 100644 --- a/sio/sio.go +++ b/sio/sio.go @@ -12,8 +12,6 @@ import ( func Extension(format string) string { switch format { - case "bsuprows": - return ".bsuprows" case "bsup": return ".bsup" case "csv": @@ -37,8 +35,6 @@ func FormatFromPath(path string) string { switch filepath.Ext(path) { case ".arrows": return "arrows" - case ".bsuprows": - return "bsuprows" case ".bsup": return "bsup" case ".csv": diff --git a/sio/sup_test.go b/sio/sup_test.go index ccd63514b9..df72235c67 100644 --- a/sio/sup_test.go +++ b/sio/sup_test.go @@ -11,8 +11,10 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/superdb/super" - "github.com/superdb/super/bsup/oldbsup" + "github.com/superdb/super/bsup" + "github.com/superdb/super/sbuf" "github.com/superdb/super/sio" + "github.com/superdb/super/sio/bsupio" "github.com/superdb/super/sio/supio" ) @@ -29,15 +31,15 @@ func boomerang(t *testing.T, logs string, compress bool) { in := []byte(strings.TrimSpace(logs) + "\n") supSrc := supio.NewReader(super.NewContext(), bytes.NewReader(in)) var rawBSUP Output - rawDst := oldbsup.NewWriter(&rawBSUP) + rawDst := bsup.NewRowWriter(&rawBSUP) require.NoError(t, sio.Copy(rawDst, supSrc)) require.NoError(t, rawDst.Close()) var out Output - rawSrc := oldbsup.NewReader(super.NewContext(), &rawBSUP) - defer rawSrc.Close() + rawSrc, err := bsupio.NewReader(t.Context(), super.NewContext(), &rawBSUP, nil, 1) + require.NoError(t, err) supDst := supio.NewWriter(&out, supio.WriterOpts{}) - err := sio.Copy(supDst, rawSrc) + err = sio.Copy(supDst, sbuf.PullerReader(sbuf.NewMaterializer(rawSrc))) if assert.NoError(t, err) { assert.Equal(t, in, out.Bytes()) } diff --git a/sup/marshal_bsup_test.go b/sup/marshal_bsup_test.go index f27aee260b..ee04962634 100644 --- a/sup/marshal_bsup_test.go +++ b/sup/marshal_bsup_test.go @@ -11,9 +11,10 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/superdb/super" - "github.com/superdb/super/bsup/oldbsup" + "github.com/superdb/super/bsup" "github.com/superdb/super/pkg/nano" "github.com/superdb/super/sio" + "github.com/superdb/super/sio/bsupio" "github.com/superdb/super/sup" "github.com/x448/float16" ) @@ -23,12 +24,12 @@ func boomerang(t *testing.T, in any, out any) { rec, err := super.NewMarshaler(sctx).Marshal(in) require.NoError(t, err) var buf bytes.Buffer - zw := oldbsup.NewWriter(sio.NopCloser(&buf)) + zw := bsup.NewRowWriter(sio.NopCloser(&buf)) err = zw.Write(rec) require.NoError(t, err) require.NoError(t, zw.Close()) - zr := oldbsup.NewReader(sctx, &buf) - defer zr.Close() + zr, err := bsupio.NewValueReader(t.Context(), sctx, &buf) + require.NoError(t, err) val, err := zr.Read() require.NoError(t, err) err = super.Unmarshal(*val, out) diff --git a/sup/marshal_test.go b/sup/marshal_test.go index d8f6c75626..de795f0f5a 100644 --- a/sup/marshal_test.go +++ b/sup/marshal_test.go @@ -10,8 +10,10 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/superdb/super" - "github.com/superdb/super/bsup/oldbsup" + "github.com/superdb/super/bsup" + "github.com/superdb/super/sbuf" "github.com/superdb/super/sio" + "github.com/superdb/super/sio/bsupio" "github.com/superdb/super/sup" ) @@ -170,14 +172,13 @@ func TestMixedTypeArrayInsideRecord(t *testing.T) { require.NoError(t, err) var buffer bytes.Buffer - writer := oldbsup.NewWriter(sio.NopCloser(&buffer)) + writer := bsup.NewRowWriter(sio.NopCloser(&buffer)) recExpected := super.NewValue(zv.Type(), zv.Bytes()) writer.Write(recExpected) writer.Close() - reader := oldbsup.NewReader(super.NewContext(), &buffer) - defer reader.Close() - recActual, err := reader.Read() + reader, err := bsupio.NewReader(t.Context(), super.NewContext(), &buffer, nil, 1) + recActual, err := sbuf.PullerReader(sbuf.NewMaterializer(reader)).Read() exp := sup.FormatValue(recExpected) actual := sup.FormatValue(*recActual) assert.Equal(t, exp, actual) @@ -238,14 +239,14 @@ func TestMixedTypeArrayOfStructWithInterface(t *testing.T) { require.NoError(t, err) var buffer bytes.Buffer - writer := oldbsup.NewWriter(sio.NopCloser(&buffer)) + writer := bsup.NewRowWriter(sio.NopCloser(&buffer)) recExpected := super.NewValue(zv.Type(), zv.Bytes()) writer.Write(recExpected) writer.Close() - reader := oldbsup.NewReader(super.NewContext(), &buffer) - defer reader.Close() - recActual, err := reader.Read() + reader, err := bsupio.NewReader(t.Context(), super.NewContext(), &buffer, nil, 1) + require.NoError(t, err) + recActual, err := sbuf.PullerReader(sbuf.NewMaterializer(reader)).Read() require.NoError(t, err) exp := sup.FormatValue(recExpected) actual := sup.FormatValue(*recActual) diff --git a/vector/vector_test.go b/vector/vector_test.go index e1abf8959f..7c577d0afa 100644 --- a/vector/vector_test.go +++ b/vector/vector_test.go @@ -54,7 +54,7 @@ func BenchmarkReadBSUPRows(b *testing.B) { bs := buf.Bytes() for b.Loop() { - valuesOut, err := fuzz.ReadBSUPRows(super.NewContext(), bs) + valuesOut, err := fuzz.ReadBSUPRows(b.Context(), super.NewContext(), bs) if err != nil { panic(err) } diff --git a/ztests/mixed-primitive-alias.yaml b/ztests/mixed-primitive-alias.yaml deleted file mode 100644 index 30ebf1e9b3..0000000000 --- a/ztests/mixed-primitive-alias.yaml +++ /dev/null @@ -1,24 +0,0 @@ -script: | - super -s a.log b.bsuprows - -inputs: - - name: a.log - data: | - #separator \x09 - #set_separator , - #empty_field (empty) - #unset_field - - #path conn - #fields orig_p - #types port - 80 - # type port=uint16 {src_port:81::port} - - name: b.bsuprows - data: !!binary dHlwZSBwb3J0PXVpbnQxNiB7c3JjX3BvcnQ6ODE6OnBvcnR9Cg== - -outputs: - - name: stdout - data: | - type port=uint16 - {_path:"conn",orig_p:80::port::(port|null)} - {src_port:81::port} From 11681e82a476cb4c1125db2018b22f84b8ebb326 Mon Sep 17 00:00:00 2001 From: Steven McCanne Date: Thu, 8 Oct 2026 13:26:46 -0700 Subject: [PATCH 2/2] address PR feedback --- fuzz/fuzz.go | 4 ++-- sio/sup_test.go | 5 ++--- sup/marshal_test.go | 11 ++++++----- 3 files changed, 10 insertions(+), 10 deletions(-) diff --git a/fuzz/fuzz.go b/fuzz/fuzz.go index 525c15eee2..21969f5abb 100644 --- a/fuzz/fuzz.go +++ b/fuzz/fuzz.go @@ -33,12 +33,12 @@ import ( func ReadBSUPRows(ctx context.Context, sctx *super.Context, bs []byte) ([]super.Value, error) { bytesReader := bytes.NewReader(bs) - reader, err := bsupio.NewReader(ctx, sctx, bytesReader, nil, 1) + reader, err := bsupio.NewValueReader(ctx, sctx, bytesReader) if err != nil { return nil, err } var a sbuf.Array - err = sio.Copy(&a, sbuf.PullerReader(sbuf.NewMaterializer(reader))) + err = sio.Copy(&a, reader) if err != nil { return nil, err } diff --git a/sio/sup_test.go b/sio/sup_test.go index df72235c67..12242eaad2 100644 --- a/sio/sup_test.go +++ b/sio/sup_test.go @@ -12,7 +12,6 @@ import ( "github.com/stretchr/testify/require" "github.com/superdb/super" "github.com/superdb/super/bsup" - "github.com/superdb/super/sbuf" "github.com/superdb/super/sio" "github.com/superdb/super/sio/bsupio" "github.com/superdb/super/sio/supio" @@ -36,10 +35,10 @@ func boomerang(t *testing.T, logs string, compress bool) { require.NoError(t, rawDst.Close()) var out Output - rawSrc, err := bsupio.NewReader(t.Context(), super.NewContext(), &rawBSUP, nil, 1) + rawSrc, err := bsupio.NewValueReader(t.Context(), super.NewContext(), &rawBSUP) require.NoError(t, err) supDst := supio.NewWriter(&out, supio.WriterOpts{}) - err = sio.Copy(supDst, sbuf.PullerReader(sbuf.NewMaterializer(rawSrc))) + err = sio.Copy(supDst, rawSrc) if assert.NoError(t, err) { assert.Equal(t, in, out.Bytes()) } diff --git a/sup/marshal_test.go b/sup/marshal_test.go index de795f0f5a..921686726f 100644 --- a/sup/marshal_test.go +++ b/sup/marshal_test.go @@ -11,7 +11,6 @@ import ( "github.com/stretchr/testify/require" "github.com/superdb/super" "github.com/superdb/super/bsup" - "github.com/superdb/super/sbuf" "github.com/superdb/super/sio" "github.com/superdb/super/sio/bsupio" "github.com/superdb/super/sup" @@ -177,8 +176,10 @@ func TestMixedTypeArrayInsideRecord(t *testing.T) { writer.Write(recExpected) writer.Close() - reader, err := bsupio.NewReader(t.Context(), super.NewContext(), &buffer, nil, 1) - recActual, err := sbuf.PullerReader(sbuf.NewMaterializer(reader)).Read() + reader, err := bsupio.NewValueReader(t.Context(), super.NewContext(), &buffer) + require.NoError(t, err) + recActual, err := reader.Read() + require.NoError(t, err) exp := sup.FormatValue(recExpected) actual := sup.FormatValue(*recActual) assert.Equal(t, exp, actual) @@ -244,9 +245,9 @@ func TestMixedTypeArrayOfStructWithInterface(t *testing.T) { writer.Write(recExpected) writer.Close() - reader, err := bsupio.NewReader(t.Context(), super.NewContext(), &buffer, nil, 1) + reader, err := bsupio.NewValueReader(t.Context(), super.NewContext(), &buffer) require.NoError(t, err) - recActual, err := sbuf.PullerReader(sbuf.NewMaterializer(reader)).Read() + recActual, err := reader.Read() require.NoError(t, err) exp := sup.FormatValue(recExpected) actual := sup.FormatValue(*recActual)