Skip to content

Commit

Permalink
Use CSUP files in vector runtime
Browse files Browse the repository at this point in the history
The commit adds the ability to use CSUP files in vector runtime. It also
implements the NewConcurrentPuller protocol to enable parallel read of a
CSUP file in vector runtime.
  • Loading branch information
mattnibs committed Nov 27, 2024
1 parent 46a0f43 commit 31b0741
Show file tree
Hide file tree
Showing 3 changed files with 85 additions and 5 deletions.
9 changes: 7 additions & 2 deletions compiler/semantic/op.go
Original file line number Diff line number Diff line change
Expand Up @@ -237,8 +237,13 @@ func (a *analyzer) formatArg(args ast.FromArgs) string {

func (a *analyzer) semFile(name string, args ast.FromArgs) dag.Op {
format := a.formatArg(args)
if format == "" && strings.HasSuffix(name, ".parquet") {
format = "parquet"
if format == "" {
switch filepath.Ext(name) {
case ".parquet":
format = "parquet"
case ".csup":
format = "csup"
}
}
return &dag.FileScan{
Kind: "FileScan",
Expand Down
12 changes: 9 additions & 3 deletions runtime/exec/environment.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ import (
"github.com/brimdata/super/zbuf"
"github.com/brimdata/super/zio/anyio"
"github.com/brimdata/super/zio/parquetio"
"github.com/brimdata/super/zio/vngio"
"github.com/segmentio/ksuid"
)

Expand Down Expand Up @@ -124,8 +125,8 @@ func (c *closePuller) Pull(done bool) (zbuf.Batch, error) {
}

func (e *Environment) VectorOpen(ctx context.Context, zctx *super.Context, path, format string, fields []field.Path) (vector.Puller, error) {
if format != "parquet" {
return nil, fmt.Errorf("vector runtime supports only Parquet files")
if format != "parquet" && format != "csup" {
return nil, fmt.Errorf("vector runtime supports only Parquet and CSUP files")
}
if path == "-" {
path = "stdio:stdin"
Expand All @@ -138,7 +139,12 @@ func (e *Environment) VectorOpen(ctx context.Context, zctx *super.Context, path,
if err != nil {
return nil, err
}
puller, err := parquetio.NewVectorReader(ctx, zctx, r, fields)
var puller vector.Puller
if format == "parquet" {
puller, err = parquetio.NewVectorReader(ctx, zctx, r, fields)
} else {
puller, err = vngio.NewVectorReader(ctx, zctx, r, fields)
}
if err != nil {
r.Close()
return nil, err
Expand Down
69 changes: 69 additions & 0 deletions zio/vngio/vectorreader.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
package vngio

import (
"context"
"errors"
"io"
"sync/atomic"

"github.com/brimdata/super"
"github.com/brimdata/super/pkg/field"
"github.com/brimdata/super/runtime/vcache"
"github.com/brimdata/super/vector"
"github.com/brimdata/super/vng"
)

type VectorReader struct {
ctx context.Context
zctx *super.Context

nextObject *atomic.Int64
objects []*vng.Object
projection vcache.Path
readerAt io.ReaderAt
}

func NewVectorReader(ctx context.Context, zctx *super.Context, r io.Reader, fields []field.Path) (*VectorReader, error) {
ra, ok := r.(io.ReaderAt)
if !ok {
return nil, errors.New("Super Columnar requires a seekable input")
}
objects, err := readObjects(ra)
if err != nil {
return nil, err
}
return &VectorReader{
ctx: ctx,
zctx: zctx,
nextObject: &atomic.Int64{},
objects: objects,
projection: vcache.NewProjection(fields),
readerAt: ra,
}, nil
}

func (v *VectorReader) NewConcurrentPuller() vector.Puller {
return v
}

func (v *VectorReader) Pull(done bool) (vector.Any, error) {
if done {
return nil, nil
}
if err := v.ctx.Err(); err != nil {
return nil, err
}
n := int(v.nextObject.Add(1) - 1)
if n >= len(v.objects) {
return nil, nil
}
o := v.objects[n]
return vcache.NewObjectFromVNG(o).Fetch(v.zctx, v.projection)
}

func (v *VectorReader) close() error {
if closer, ok := v.readerAt.(io.Closer); ok {
return closer.Close() // coffee is for closers
}
return nil
}

0 comments on commit 31b0741

Please sign in to comment.