diff --git a/compiler/compiler.go b/compiler/compiler.go index d8e76e6a7..141522603 100644 --- a/compiler/compiler.go +++ b/compiler/compiler.go @@ -60,7 +60,7 @@ func (l *compiler) NewDeleteQuery(rctx *runtime.Context, ast *parser.AST, head * if err = optimizer.New(rctx, l.env).OptimizeDeleter(main, Parallelism); err != nil { return nil, err } - outputs, debugs, b, err := BuildWithBuilder(rctx, main, l.env, nil) + outputs, debugs, b, err := BuildWithBuilder(rctx, main, l.env) if err != nil { return nil, err } diff --git a/compiler/package.go b/compiler/package.go index db877cb10..9f15ab87f 100644 --- a/compiler/package.go +++ b/compiler/package.go @@ -43,18 +43,18 @@ func Optimize(ctx context.Context, main *dag.Main, env *exec.Environment, parall return nil } -func Build(rctx *runtime.Context, main *dag.Main, env *exec.Environment, readers []sio.Reader) (map[string]vio.Puller, *op.DebugChans, vio.Meter, error) { +func Build(rctx *runtime.Context, main *dag.Main, env *exec.Environment) (map[string]vio.Puller, *op.DebugChans, vio.Meter, error) { b := rungen.NewBuilder(rctx, env) - outputs, debugs, err := b.Build(main, readers...) + outputs, debugs, err := b.Build(main) if err != nil { return nil, nil, nil, err } return outputs, debugs, b.Meter(), nil } -func BuildWithBuilder(rctx *runtime.Context, main *dag.Main, env *exec.Environment, readers []sio.Reader) (map[string]vio.Puller, *op.DebugChans, *rungen.Builder, error) { +func BuildWithBuilder(rctx *runtime.Context, main *dag.Main, env *exec.Environment) (map[string]vio.Puller, *op.DebugChans, *rungen.Builder, error) { b := rungen.NewBuilder(rctx, env) - outputs, debugs, err := b.Build(main, readers...) + outputs, debugs, err := b.Build(main) if err != nil { return nil, nil, nil, err } @@ -76,7 +76,7 @@ func CompileWithAST(rctx *runtime.Context, ast *parser.AST, env *exec.Environmen return nil, err } } - outputs, debugs, meter, err := Build(rctx, main, env, nil) + outputs, debugs, meter, err := Build(rctx, main, env) if err != nil { return nil, err } diff --git a/compiler/rungen/op.go b/compiler/rungen/op.go index f9defb5ef..8dc4c245c 100644 --- a/compiler/rungen/op.go +++ b/compiler/rungen/op.go @@ -42,7 +42,6 @@ import ( vamexpr "github.com/brimdata/super/runtime/vam/expr" vamop "github.com/brimdata/super/runtime/vam/op" "github.com/brimdata/super/sbuf" - "github.com/brimdata/super/sio" "github.com/brimdata/super/vector/vio" "github.com/segmentio/ksuid" ) @@ -54,7 +53,6 @@ type Builder struct { mctx *super.Context mapper *super.TypeDefsMapper env *exec.Environment - readers []sio.Reader progress *vio.Progress debugs *vamop.DebugChans channels map[string][]vio.Puller @@ -83,13 +81,11 @@ func NewBuilder(rctx *runtime.Context, env *exec.Environment) *Builder { } } -// Build builds a flowgraph for main. If main contains a dag.DefaultSource, it -// will read from readers. -func (b *Builder) Build(main *dag.Main, readers ...sio.Reader) (map[string]vio.Puller, *vamop.DebugChans, error) { +// Build builds a flowgraph for main. +func (b *Builder) Build(main *dag.Main) (map[string]vio.Puller, *vamop.DebugChans, error) { if !isEntry(main.Body) { return nil, nil, errors.New("internal error: DAG entry point is not a data source") } - b.readers = readers if len(main.Types) != 0 { defs, ok := super.NewTypeDefsFromBytes(main.Types) if !ok { diff --git a/runtime/sam/expr/filter_test.go b/runtime/sam/expr/filter_test.go index 255c4e0dc..773d2e808 100644 --- a/runtime/sam/expr/filter_test.go +++ b/runtime/sam/expr/filter_test.go @@ -63,7 +63,7 @@ func runCasesHelper(t *testing.T, record string, cases []testcase, expectBufferF require.NoError(t, err, "filter: %q", c.filter) err = compiler.Optimize(rctx, main, env, 0) require.NoError(t, err, "filter: %q", c.filter) - _, _, builder, err := compiler.BuildWithBuilder(rctx, main, env, nil) + _, _, builder, err := compiler.BuildWithBuilder(rctx, main, env) require.NoError(t, err, "filter: %q", c.filter) scan, ok := main.Body[0].(*dag.FileScan) require.True(t, ok)