Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

simplfy compiler patterns #5425

Merged
merged 2 commits into from
Nov 3, 2024
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion cmd/super/compile/command.go
Original file line number Diff line number Diff line change
Expand Up @@ -41,10 +41,12 @@ func init() {
type Command struct {
*root.Command
shared Shared
files bool
}

func New(parent charm.Command, f *flag.FlagSet) (charm.Command, error) {
c := &Command{Command: parent.(*root.Command)}
f.BoolVar(&c.files, "files", false, "compile query as if command-line input files are present)")
c.shared.SetFlags(f)
return c, nil
}
Expand All @@ -55,5 +57,5 @@ func (c *Command) Run(args []string) error {
return err
}
defer cleanup()
return c.shared.Run(ctx, args, nil, false)
return c.shared.Run(ctx, args, nil, false, c.files)
}
34 changes: 16 additions & 18 deletions cmd/super/compile/shared.go
Original file line number Diff line number Diff line change
Expand Up @@ -11,12 +11,11 @@ import (
"github.com/brimdata/super/cli/outputflags"
"github.com/brimdata/super/cli/queryflags"
"github.com/brimdata/super/compiler"
"github.com/brimdata/super/compiler/data"
"github.com/brimdata/super/compiler/describe"
"github.com/brimdata/super/compiler/parser"
"github.com/brimdata/super/lake"
"github.com/brimdata/super/pkg/storage"
"github.com/brimdata/super/runtime"
"github.com/brimdata/super/runtime/exec"
"github.com/brimdata/super/zbuf"
"github.com/brimdata/super/zfmt"
"github.com/brimdata/super/zio"
Expand All @@ -43,7 +42,7 @@ func (s *Shared) SetFlags(fs *flag.FlagSet) {
s.OutputFlags.SetFlags(fs)
}

func (s *Shared) Run(ctx context.Context, args []string, lakeFlags *lakeflags.Flags, desc bool) error {
func (s *Shared) Run(ctx context.Context, args []string, lakeFlags *lakeflags.Flags, desc, extInput bool) error {
if len(s.includes) == 0 && len(args) == 0 {
return errors.New("no query specified")
}
Expand All @@ -62,11 +61,16 @@ func (s *Shared) Run(ctx context.Context, args []string, lakeFlags *lakeflags.Fl
if len(args) == 1 {
query = args[0]
}
ast, err := parser.ParseQuery(query, s.includes...)
rctx := runtime.DefaultContext()
env := exec.NewEnvironment(nil, lk)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: Declare this closer to its use.

ast, err := compiler.Parse(query, s.includes...)
if err != nil {
return err
}
if s.optimize || s.parallel > 0 || desc {
if s.parallel > 0 {
s.optimize = true
}
if s.optimize || desc {
s.dag = true
}
if !s.dag {
Expand All @@ -76,34 +80,28 @@ func (s *Shared) Run(ctx context.Context, args []string, lakeFlags *lakeflags.Fl
}
return s.writeValue(ctx, ast.Parsed())
}
runtime, err := compiler.NewJob(runtime.DefaultContext(), ast, data.NewSource(nil, lk), false)
dag, err := compiler.Analyze(rctx, ast, env, extInput)
if err != nil {
return err
}
if desc {
description, err := describe.AnalyzeDAG(ctx, runtime.Entry(), data.NewSource(nil, lk))
description, err := describe.AnalyzeDAG(ctx, dag, env)
if err != nil {
return err
}
return s.writeValue(ctx, description)
}
if s.parallel > 0 {
if err := runtime.Optimize(); err != nil {
return err
}
if err := runtime.Parallelize(s.parallel); err != nil {
return err
}
} else if s.optimize {
if err := runtime.Optimize(); err != nil {
if s.optimize {
dag, err = compiler.Optimize(rctx, dag, env, s.parallel)
if err != nil {
return err
}
}
if s.query {
fmt.Println(zfmt.DAG(runtime.Entry()))
fmt.Println(zfmt.DAG(dag))
return nil
}
return s.writeValue(ctx, runtime.Entry())
return s.writeValue(ctx, dag)
}

func (s *Shared) writeValue(ctx context.Context, v any) error {
Expand Down
2 changes: 1 addition & 1 deletion cmd/super/db/compile/command.go
Original file line number Diff line number Diff line change
Expand Up @@ -47,5 +47,5 @@ func (c *Command) Run(args []string) error {
return err
}
defer cleanup()
return c.shared.Run(ctx, args, &c.parent.LakeFlags, c.describe)
return c.shared.Run(ctx, args, &c.parent.LakeFlags, c.describe, false)
}
4 changes: 2 additions & 2 deletions cmd/super/dev/vector/search/command.go
Original file line number Diff line number Diff line change
Expand Up @@ -10,10 +10,10 @@ import (
"github.com/brimdata/super/cli/poolflags"
"github.com/brimdata/super/cmd/super/dev/vector"
"github.com/brimdata/super/compiler"
"github.com/brimdata/super/compiler/data"
"github.com/brimdata/super/pkg/charm"
"github.com/brimdata/super/pkg/storage"
"github.com/brimdata/super/runtime"
"github.com/brimdata/super/runtime/exec"
"github.com/brimdata/super/zbuf"
)

Expand Down Expand Up @@ -66,7 +66,7 @@ func (c *Command) Run(args []string) error {
}
text := args[0]
rctx := runtime.NewContext(ctx, super.NewContext())
puller, err := compiler.VectorFilterCompile(rctx, text, data.NewSource(nil, root), head)
puller, err := compiler.VectorFilterCompile(rctx, text, exec.NewEnvironment(nil, root), head)
if err != nil {
return err
}
Expand Down
2 changes: 1 addition & 1 deletion cmd/super/root/command.go
Original file line number Diff line number Diff line change
Expand Up @@ -158,7 +158,7 @@ func (c *Command) Run(args []string) error {
if err != nil {
return err
}
comp := compiler.NewFileSystemCompiler(local)
comp := compiler.NewCompiler(local)
query, err := runtime.CompileQuery(ctx, zctx, comp, ast, readers)
if err != nil {
return err
Expand Down
73 changes: 73 additions & 0 deletions compiler/compiler.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
package compiler

import (
goruntime "runtime"

"github.com/brimdata/super/compiler/dag"
"github.com/brimdata/super/compiler/optimizer"
"github.com/brimdata/super/compiler/parser"
"github.com/brimdata/super/lake"
"github.com/brimdata/super/lakeparse"
"github.com/brimdata/super/pkg/storage"
"github.com/brimdata/super/runtime"
"github.com/brimdata/super/runtime/exec"
"github.com/brimdata/super/zio"
)

var Parallelism = goruntime.GOMAXPROCS(0) //XXX

type compiler struct {
env *exec.Environment
}

func NewCompiler(local storage.Engine) runtime.Compiler {
return &compiler{env: exec.NewEnvironment(local, nil)}
}

func NewLakeCompiler(lk *lake.Root) runtime.Compiler {
// We configure a remote storage engine into the lake compiler so that
// "from" operators that source http or s3 will work, but stdio and
// file system accesses will be rejected at open time.
return &compiler{env: exec.NewEnvironment(storage.NewRemoteEngine(), lk)}
}

func (c *compiler) NewQuery(rctx *runtime.Context, ast *parser.AST, readers []zio.Reader, parallelism int) (runtime.Query, error) {
if c.env.IsLake() {
if parallelism == 0 {
parallelism = Parallelism
}
}
return CompileWithAST(rctx, ast, c.env, true, parallelism, readers)
}

func (l *compiler) NewLakeDeleteQuery(rctx *runtime.Context, ast *parser.AST, head *lakeparse.Commitish) (runtime.DeleteQuery, error) {
if err := ast.ConvertToDeleteWhere(head.Pool, head.Branch); err != nil {
return nil, err
}
seq := ast.Parsed()
if len(seq) != 2 {
return nil, &InvalidDeleteWhereQuery{}
}
dagSeq, err := Analyze(rctx, ast, l.env, false)
if err != nil {
return nil, err
}
if _, ok := dagSeq[1].(*dag.Filter); !ok {
return nil, &InvalidDeleteWhereQuery{}
}
dagSeq, err = optimizer.New(rctx, l.env).OptimizeDeleter(dagSeq, Parallelism)
if err != nil {
return nil, err
}
outputs, b, err := BuildWithBuilder(rctx, dagSeq, l.env, nil)
if err != nil {
return nil, err
}
return exec.NewDeleteQuery(rctx, bundleOutputs(rctx, outputs), b.Deletes()), nil
}

type InvalidDeleteWhereQuery struct{}

func (InvalidDeleteWhereQuery) Error() string {
return "invalid delete where query: must be a single filter operation"
}
6 changes: 3 additions & 3 deletions compiler/describe/analyze.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,13 +5,13 @@ import (
"fmt"

"github.com/brimdata/super/compiler/dag"
"github.com/brimdata/super/compiler/data"
"github.com/brimdata/super/compiler/optimizer"
"github.com/brimdata/super/compiler/parser"
"github.com/brimdata/super/compiler/semantic"
"github.com/brimdata/super/lake"
"github.com/brimdata/super/order"
"github.com/brimdata/super/pkg/field"
"github.com/brimdata/super/runtime/exec"
"github.com/segmentio/ksuid"
)

Expand Down Expand Up @@ -54,7 +54,7 @@ type Channel struct {
Sort order.SortKeys `json:"sort"`
}

func Analyze(ctx context.Context, query string, src *data.Source) (*Info, error) {
func Analyze(ctx context.Context, query string, src *exec.Environment) (*Info, error) {
ast, err := parser.ParseQuery(query)
if err != nil {
return nil, err
Expand All @@ -66,7 +66,7 @@ func Analyze(ctx context.Context, query string, src *data.Source) (*Info, error)
return AnalyzeDAG(ctx, entry, src)
}

func AnalyzeDAG(ctx context.Context, entry dag.Seq, src *data.Source) (*Info, error) {
func AnalyzeDAG(ctx context.Context, entry dag.Seq, src *exec.Environment) (*Info, error) {
var err error
var info Info
if info.Sources, err = describeSources(ctx, src.Lake(), entry[0]); err != nil {
Expand Down
70 changes: 0 additions & 70 deletions compiler/file.go

This file was deleted.

Loading