simplfy compiler patterns

cmd/super/compile/command.go

@@ -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
 }
@@ -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)
 }

cmd/super/compile/shared.go

@@ -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"
@@ -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")
 	}
@@ -62,11 +61,14 @@ 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...)
+	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 {
@@ -76,34 +78,30 @@ 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)
+	rctx := runtime.DefaultContext()
+	env := exec.NewEnvironment(nil, lk)
+	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 {

cmd/super/db/compile/command.go

@@ -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)
 }

cmd/super/dev/vector/search/command.go

@@ -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"
 )
 
@@ -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
 	}

cmd/super/root/command.go

@@ -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

compiler/compiler.go

@@ -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"
+}

compiler/describe/analyze.go

@@ -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"
 )
 
@@ -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
@@ -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 {