Merge branch 'main' into ztest-remove-Bundle.TestName

Author: nwt

cli/inputflags/flags.go

@@ -1,81 +1,52 @@
 package inputflags
 
 import (
-	"context"
 	"errors"
 	"flag"
-	"fmt"
-	"os"
 
-	"github.com/brimdata/super"
 	"github.com/brimdata/super/cli/auto"
-	"github.com/brimdata/super/pkg/storage"
-	"github.com/brimdata/super/sio"
 	"github.com/brimdata/super/sio/anyio"
 	"github.com/brimdata/super/sio/bsupio"
 )
 
 type Flags struct {
-	anyio.ReaderOpts
-	Dynamic  bool
-	ReadMax  auto.Bytes
-	ReadSize auto.Bytes
-	Threads  int
-}
-
-func (f *Flags) Options() anyio.ReaderOpts {
-	return f.ReaderOpts
+	Dynamic      bool
+	ReaderOpts   anyio.ReaderOpts
+	bsupReadMax  auto.Bytes
+	bsupReadSize auto.Bytes
 }
 
 func (f *Flags) SetFlags(fs *flag.FlagSet, validate bool) {
-	fs.StringVar(&f.Format, "i", "auto", "format of input data [auto,arrows,bsup,csup,csv,json,line,parquet,sup,tsv,zeek,jsup]")
-	f.CSV.Delim = ','
+	f.bsupReadMax = auto.NewBytes(bsupio.MaxSize)
+	fs.Var(&f.bsupReadMax, "bsup.readmax", "maximum Super Binary read buffer size in MiB, MB, etc.")
+	f.bsupReadSize = auto.NewBytes(bsupio.ReadSize)
+	fs.Var(&f.bsupReadSize, "bsup.readsize", "target Super Binary read buffer size in MiB, MB, etc.")
+	opts := &f.ReaderOpts
+	fs.IntVar(&opts.BSUP.Threads, "bsup.threads", 0, "number of Super Binary read threads (0=GOMAXPROCS)")
+	fs.BoolVar(&opts.BSUP.Validate, "bsup.validate", validate, "validate format when reading Super Binary")
+	opts.CSV.Delim = ','
 	fs.Func("csv.delim", `CSV field delimiter (default ",")`, func(s string) error {
 		if len(s) != 1 {
 			return errors.New("CSV field delimiter must be exactly one character")
 		}
-		f.CSV.Delim = rune(s[0])
+		opts.CSV.Delim = rune(s[0])
 		return nil
 
 	})
-	fs.IntVar(&f.BSUP.Threads, "bsup.threads", 0, "number of Super Binary read threads (0=GOMAXPROCS)")
-	fs.BoolVar(&f.BSUP.Validate, "bsup.validate", validate, "validate format when reading Super Binary")
-	f.ReadMax = auto.NewBytes(bsupio.MaxSize)
-	fs.Var(&f.ReadMax, "bsup.readmax", "maximum Super Binary read buffer size in MiB, MB, etc.")
-	f.ReadSize = auto.NewBytes(bsupio.ReadSize)
-	fs.Var(&f.ReadSize, "bsup.readsize", "target Super Binary read buffer size in MiB, MB, etc.")
 	fs.BoolVar(&f.Dynamic, "dynamic", false, "disable static type checking of inputs")
+	fs.StringVar(&opts.Format, "i", "auto", "format of input data [auto,arrows,bsup,csup,csv,json,jsup,line,parquet,sup,tsv,zeek]")
 }
 
 // Init is called after flags have been parsed.
 func (f *Flags) Init() error {
-	f.BSUP.Max = int(f.ReadMax.Bytes)
-	if f.BSUP.Max < 0 {
+	bsup := &f.ReaderOpts.BSUP
+	bsup.Max = int(f.bsupReadMax.Bytes)
+	if bsup.Max < 0 {
 		return errors.New("max read buffer size must be greater than zero")
 	}
-	f.BSUP.Size = int(f.ReadSize.Bytes)
-	if f.BSUP.Size < 0 {
+	bsup.Size = int(f.bsupReadSize.Bytes)
+	if bsup.Size < 0 {
 		return errors.New("target read buffer size must be greater than zero")
 	}
 	return nil
 }
-
-func (f *Flags) Open(ctx context.Context, sctx *super.Context, engine storage.Engine, paths []string, stopOnErr bool) ([]sio.Reader, error) {
-	var readers []sio.Reader
-	for _, path := range paths {
-		if path == "-" {
-			path = "stdio:stdin"
-		}
-		file, err := anyio.Open(ctx, sctx, engine, path, f.ReaderOpts)
-		if err != nil {
-			err = fmt.Errorf("%s: %w", path, err)
-			if stopOnErr {
-				return nil, err
-			}
-			fmt.Fprintln(os.Stderr, err)
-			continue
-		}
-		readers = append(readers, file)
-	}
-	return readers, nil
-}

cmd/super/db/load/command.go

@@ -22,6 +22,7 @@ import (
 	"github.com/brimdata/super/pkg/storage"
 	"github.com/brimdata/super/pkg/units"
 	"github.com/brimdata/super/sio"
+	"github.com/brimdata/super/sio/anyio"
 	"github.com/paulbellamy/ratecounter"
 	"golang.org/x/term"
 )
@@ -79,7 +80,7 @@ func (c *Command) Run(args []string) error {
 	paths := args
 	c.engine = &engineWrap{Engine: storage.NewLocalEngine()}
 	sctx := super.NewContext()
-	readers, err := c.inputFlags.Open(ctx, sctx, c.engine, paths, false)
+	readers, err := c.open(ctx, sctx, paths)
 	if err != nil {
 		return err
 	}
@@ -116,6 +117,22 @@ func (c *Command) Run(args []string) error {
 	return nil
 }
 
+func (c *Command) open(ctx context.Context, sctx *super.Context, paths []string) ([]sio.Reader, error) {
+	var readers []sio.Reader
+	for _, path := range paths {
+		if path == "-" {
+			path = "stdio:stdin"
+		}
+		file, err := anyio.Open(ctx, sctx, c.engine, path, c.inputFlags.ReaderOpts)
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "%s: %s\n", path, err)
+			continue
+		}
+		readers = append(readers, file)
+	}
+	return readers, nil
+}
+
 func (c *Command) Display(w io.Writer) bool {
 	readBytes, completed := c.engine.status()
 	fmt.Fprintf(w, "(%d/%d) ", completed, len(c.engine.readers))

cmd/super/root/command.go

@@ -149,7 +149,7 @@ func (c *Command) Run(args []string) error {
 	env := exec.NewEnvironment(storage.NewLocalEngine(), nil)
 	env.Dynamic = c.inputFlags.Dynamic
 	env.IgnoreOpenErrors = !c.stopErr
-	env.ReaderOpts = c.inputFlags.Options()
+	env.ReaderOpts = c.inputFlags.ReaderOpts
 	comp := compiler.NewCompilerWithEnv(env)
 	query, err := runtime.CompileQuery(ctx, super.NewContext(), comp, ast, nil)
 	if err != nil {

compiler/rungen/expr.go

@@ -125,7 +125,7 @@ func (b *Builder) compileBinary(e *dag.BinaryExpr) (expr.Evaluator, error) {
 	case "<", "<=", ">", ">=":
 		return expr.NewCompareRelative(b.sctx(), lhs, rhs, op)
 	case "+", "-", "*", "/", "%":
-		return expr.NewArithmetic(b.sctx(), lhs, rhs, op)
+		return expr.NewArithmetic(b.sctx(), op, lhs, rhs)
 	default:
 		return nil, fmt.Errorf("invalid binary operator %s", op)
 	}

compiler/rungen/vexpr.go

@@ -102,9 +102,9 @@ func (b *Builder) compileVamBinary(e *dag.BinaryExpr) (vamexpr.Evaluator, error)
 	case "in":
 		return vamexpr.NewIn(b.sctx(), lhs, rhs), nil
 	case "==", "!=", "<", "<=", ">", ">=":
-		return vamexpr.NewCompare(b.sctx(), lhs, rhs, op), nil
+		return vamexpr.NewCompare(b.sctx(), op, lhs, rhs), nil
 	case "+", "-", "*", "/", "%":
-		return vamexpr.NewArith(b.sctx(), lhs, rhs, op), nil
+		return vamexpr.NewArith(b.sctx(), op, lhs, rhs), nil
 	default:
 		return nil, fmt.Errorf("invalid binary operator %s", op)
 	}

runtime/sam/expr/eval.go

@@ -428,7 +428,7 @@ var DivideByZero = errors.New("divide by zero")
 
 // NewArithmetic compiles an expression of the form "expr1 op expr2"
 // for the arithmetic operators +, -, *, /
-func NewArithmetic(sctx *super.Context, lhs, rhs Evaluator, op string) (Evaluator, error) {
+func NewArithmetic(sctx *super.Context, op string, lhs, rhs Evaluator) (Evaluator, error) {
 	n := newNumeric(sctx, lhs, rhs)
 	switch op {
 	case "+":

runtime/vam/expr/arith.go

@@ -20,7 +20,7 @@ type Arith struct {
 	rhs    Evaluator
 }
 
-func NewArith(sctx *super.Context, lhs, rhs Evaluator, op string) *Arith {
+func NewArith(sctx *super.Context, op string, lhs, rhs Evaluator) *Arith {
 	return &Arith{sctx, vector.ArithOpFromString(op), lhs, rhs}
 }
 

runtime/vam/expr/arith_test.go

@@ -35,7 +35,7 @@ func TestArithOpsAndForms(t *testing.T) {
 	for _, c := range cases {
 		f := func(expected []int64, lhs, rhs vector.Any) {
 			t.Helper()
-			cmp := NewArith(super.NewContext(), &testEval{lhs}, &testEval{rhs}, c.op)
+			cmp := NewArith(super.NewContext(), c.op, &testEval{lhs}, &testEval{rhs})
 			assert.Equal(t, expected, cmp.Eval(nil).(*vector.Int).Values, "op: %s", c.op)
 		}
 
@@ -59,7 +59,7 @@ func TestArithOpsAndForms(t *testing.T) {
 		f(c.expectedForConstLHS, Const, rhsView)
 
 		// Arithmetic on two vector.Consts returns another vector.Const.
-		cmp := NewArith(super.NewContext(), &testEval{Const}, &testEval{Const}, c.op)
+		cmp := NewArith(super.NewContext(), c.op, &testEval{Const}, &testEval{Const})
 		val := cmp.Eval(nil).(*vector.Const)
 		assert.Equal(t, uint32(3), val.Len(), "op: %s", c.op)
 		expected := super.NewInt64(c.expectedForConstLHS[0])

runtime/vam/expr/compare.go

@@ -19,7 +19,7 @@ type Compare struct {
 	rhs    Evaluator
 }
 
-func NewCompare(sctx *super.Context, lhs, rhs Evaluator, op string) *Compare {
+func NewCompare(sctx *super.Context, op string, lhs, rhs Evaluator) *Compare {
 	return &Compare{sctx, vector.CompareOpFromString(op), lhs, rhs}
 }
 

runtime/vam/expr/compare_test.go

@@ -43,7 +43,7 @@ func TestCompareOpsAndForms(t *testing.T) {
 	for _, c := range cases {
 		f := func(expected string, lhs, rhs vector.Any) {
 			t.Helper()
-			cmp := NewCompare(super.NewContext(), &testEval{lhs}, &testEval{rhs}, c.op)
+			cmp := NewCompare(super.NewContext(), c.op, &testEval{lhs}, &testEval{rhs})
 			assert.Equal(t, expected, cmp.Eval(nil).(*vector.Bool).Bits.String(), "op: %s", c.op)
 		}
 
@@ -67,11 +67,10 @@ func TestCompareOpsAndForms(t *testing.T) {
 		f(c.expectedForConstLHS, Const, rhsView)
 
 		// Comparing two vector.Consts yields another vector.Const.
-		cmp := NewCompare(super.NewContext(), &testEval{Const}, &testEval{Const}, c.op)
+		cmp := NewCompare(super.NewContext(), c.op, &testEval{Const}, &testEval{Const})
 		val := cmp.Eval(nil).(*vector.Const)
 		assert.Equal(t, uint32(3), val.Len(), "op: %s", c.op)
 		expected := super.NewBool(c.expectedForConstLHS == "111")
 		assert.Equal(t, expected, val.Value(), "op: %s", c.op)
 	}
-
 }