CSUP: Allow multiple objects

vng/object.go

@@ -75,6 +75,10 @@ func (o *Object) NewReader(zctx *super.Context) (zio.Reader, error) {
 	return NewZedReader(zctx, o.meta, o.readerAt)
 }
 
+func (o *Object) Size() uint64 {
+	return HeaderSize + o.header.MetaSize + o.header.DataSize
+}
+
 func readMetadata(r io.Reader) (Metadata, error) {
 	zctx := super.NewContext()
 	zr := zngio.NewReader(zctx, r)

vng/writer.go

@@ -11,6 +11,8 @@ import (
 	"github.com/brimdata/super/zson"
 )
 
+var maxObjectSize uint32 = 120_000
+
 // Writer implements the zio.Writer interface. A Writer creates a vector
 // VNG object from a stream of super.Records.
 type Writer struct {
@@ -30,18 +32,24 @@ func NewWriter(w io.WriteCloser) *Writer {
 }
 
 func (w *Writer) Close() error {
-	firstErr := w.finalize()
+	firstErr := w.finalizeObject()
 	if err := w.writer.Close(); err != nil && firstErr == nil {
 		firstErr = err
 	}
 	return firstErr
 }
 
 func (w *Writer) Write(val super.Value) error {
-	return w.dynamic.Write(val)
+	if err := w.dynamic.Write(val); err != nil {
+		return err
+	}
+	if w.dynamic.len >= maxObjectSize {
+		return w.finalizeObject()
+	}
+	return nil
 }
 
-func (w *Writer) finalize() error {
+func (w *Writer) finalizeObject() error {
 	meta, dataSize, err := w.dynamic.Encode()
 	if err != nil {
 		return fmt.Errorf("system error: could not encode VNG metadata: %w", err)
@@ -74,5 +82,8 @@ func (w *Writer) finalize() error {
 	if err := w.dynamic.Emit(w.writer); err != nil {
 		return fmt.Errorf("system error: could not write VNG data section: %w", err)
 	}
+	// Set new dynamic so we can write the next section.
+	w.dynamic = NewDynamicEncoder()
+	w.zctx.Reset()
 	return nil
 }

zio/vngio/reader.go

@@ -3,21 +3,80 @@ package vngio
 import (
 	"errors"
 	"io"
+	"math"
 
 	"github.com/brimdata/super"
 	"github.com/brimdata/super/pkg/field"
 	"github.com/brimdata/super/vng"
 	"github.com/brimdata/super/zio"
 )
 
+type reader struct {
+	zctx     *super.Context
+	objects  []*vng.Object
+	n        int
+	readerAt io.ReaderAt
+	reader   zio.Reader
+}
+
 func NewReader(zctx *super.Context, r io.Reader, fields []field.Path) (zio.Reader, error) {
 	ra, ok := r.(io.ReaderAt)
 	if !ok {
 		return nil, errors.New("Super Columnar requires a seekable input")
 	}
-	o, err := vng.NewObject(ra)
+	objects, err := readObjects(ra)
 	if err != nil {
 		return nil, err
 	}
-	return o.NewReader(zctx)
+	return &reader{
+		zctx:     zctx,
+		objects:  objects,
+		readerAt: ra,
+	}, nil
+}
+
+func (r *reader) Read() (*super.Value, error) {
+again:
+	if r.reader == nil {
+		if r.n >= len(r.objects) {
+			return nil, nil
+		}
+		o := r.objects[r.n]
+		r.n++
+		var err error
+		if r.reader, err = o.NewReader(r.zctx); err != nil {
+			return nil, err
+		}
+	}
+	v, err := r.reader.Read()
+	if v == nil && err == nil {
+		r.reader = nil
+		goto again
+	}
+	return v, err
+}
+
+func (r *reader) Close() error {
+	if closer, ok := r.readerAt.(io.Closer); ok {
+		return closer.Close()
+	}
+	return nil
+}
+
+func readObjects(r io.ReaderAt) ([]*vng.Object, error) {
+	var objects []*vng.Object
+	var start int64
+	for {
+		// NewObject puts the right end to the passed in SectionReader and since
+		// the end is unkown just pass MaxInt64.
+		o, err := vng.NewObject(io.NewSectionReader(r, start, math.MaxInt64))
+		if err != nil {
+			if err == io.EOF && len(objects) > 0 {
+				return objects, nil
+			}
+			return nil, err
+		}
+		objects = append(objects, o)
+		start += int64(o.Size())
+	}
 }