compiler/semantic: enable type checking for all input files (#6434)

For file types with an sio.Reader that has a Type method, continue to
use that to obtain a super.Type.  For other file types, if the input can
seek, read all values from the file and fuse their types to obtain a
single type.  (Fifos and pipes are excluded for now because the runtime
must also be able to read the file.)

To disable type checking for all input files, specify the -dynamic flag
to the super and super compile commands.

Author: nwt

cli/inputflags/flags.go

@@ -17,6 +17,7 @@ import (
 
 type Flags struct {
 	anyio.ReaderOpts
+	Dynamic  bool
 	ReadMax  auto.Bytes
 	ReadSize auto.Bytes
 	Threads  int
@@ -43,6 +44,7 @@ func (f *Flags) SetFlags(fs *flag.FlagSet, validate bool) {
 	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")
 }
 
 // Init is called after flags have been parsed.

cmd/super/compile/shared.go

@@ -24,6 +24,7 @@ import (
 
 type Shared struct {
 	dag         bool
+	dynamic     bool
 	includes    queryflags.Includes
 	optimize    bool
 	parallel    int
@@ -33,6 +34,7 @@ type Shared struct {
 
 func (s *Shared) SetFlags(fs *flag.FlagSet) {
 	fs.BoolVar(&s.dag, "dag", false, "display output as DAG (implied by -O or -P)")
+	fs.BoolVar(&s.dynamic, "dynamic", false, "disable static type checking of inputs on DAG")
 	fs.Var(&s.includes, "I", "source file containing query text (may be repeated)")
 	fs.BoolVar(&s.optimize, "O", false, "display optimized DAG")
 	fs.IntVar(&s.parallel, "P", 0, "display parallelized DAG")
@@ -80,6 +82,7 @@ func (s *Shared) Run(ctx context.Context, args []string, dbFlags *dbflags.Flags,
 	}
 	rctx := runtime.DefaultContext()
 	env := exec.NewEnvironment(storage.NewLocalEngine(), root)
+	env.Dynamic = s.dynamic
 	dag, err := compiler.Analyze(rctx, ast, env, false)
 	if err != nil {
 		return err

cmd/super/root/command.go

@@ -147,6 +147,7 @@ func (c *Command) Run(args []string) error {
 		ast.PrependFileScan(args)
 	}
 	env := exec.NewEnvironment(storage.NewLocalEngine(), nil)
+	env.Dynamic = c.inputFlags.Dynamic
 	env.IgnoreOpenErrors = !c.stopErr
 	env.ReaderOpts = c.inputFlags.Options()
 	comp := compiler.NewCompilerWithEnv(env)

cmd/super/ztests/from-file-error.yaml

@@ -1,5 +1,5 @@
 script: |
-  ! super -I query.spq
+  ! super -dynamic -I query.spq
 
 inputs:
   - name: query.spq

cmd/super/ztests/stop-on-error-2.yaml

@@ -1,5 +1,5 @@
 script: |
-  super -s -e=false good.sup bad.sup
+  super -dynamic -e=false -s good.sup bad.sup
 
 inputs:
   - name: good.sup

cmd/super/ztests/stop-on-error-3.yaml

@@ -1,6 +1,6 @@
 # Second input has bad middle line (detection succeeds).
 script: |
-  ! super -s -e=false good.sup bad.sup
+  ! super -dynamic -e=false -s good.sup bad.sup
 
 inputs:
   - name: good.sup

compiler/parser/ztests/glob-mul.yaml

@@ -1,7 +1,7 @@
 script: |
-  super -s -c "? a*b" in.sup
+  super -dynamic -s -c "? a*b" in.sup
   echo ===
-  super -s -c "? a*b=s" in.sup
+  super -dynamic -s -c "? a*b=s" in.sup
 
 inputs:
   - name: in.sup

compiler/semantic/op.go

@@ -22,7 +22,7 @@ import (
 	"github.com/brimdata/super/runtime/sam/expr"
 	"github.com/brimdata/super/runtime/sam/expr/function"
 	"github.com/brimdata/super/sio"
-	"github.com/brimdata/super/sio/parquetio"
+	"github.com/brimdata/super/sio/anyio"
 	"github.com/brimdata/super/sup"
 	"github.com/segmentio/ksuid"
 )
@@ -269,8 +269,11 @@ func (t *translator) file(n ast.Node, name string, args []ast.OpArg) sem.Op {
 }
 
 func (t *translator) fileType(path, format string) (super.Type, error) {
+	if t.env.Dynamic {
+		return nil, nil
+	}
 	engine := t.env.Engine()
-	if engine == nil || format != "" && format != "auto" && format != "parquet" {
+	if engine == nil {
 		return nil, nil
 	}
 	uri, err := storage.ParseURI(path)
@@ -282,7 +285,27 @@ func (t *translator) fileType(path, format string) (super.Type, error) {
 		return nil, err
 	}
 	defer r.Close()
-	return parquetio.Type(t.sctx, r), nil
+	var b [1]byte
+	if _, err := r.ReadAt(b[:], 0); err != nil {
+		// r can't seek so it's a fifo or pipe.
+		return nil, nil
+	}
+	f, err := anyio.NewFile(t.sctx, r, path, anyio.ReaderOpts{Format: format})
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+	if typer, ok := f.Reader.(interface{ Type() super.Type }); ok {
+		return typer.Type(), nil
+	}
+	fuser := t.checker.newFuser()
+	for {
+		val, err := f.Read()
+		if val == nil || err != nil {
+			return fuser.Type(), err
+		}
+		fuser.fuse(val.Type())
+	}
 }
 
 func (t *translator) fromFileGlob(globLoc ast.Node, pattern string, args []ast.OpArg) sem.Op {

compiler/semantic/ztests/pipe-schema-parquet.yaml

@@ -2,7 +2,8 @@ script: |
   super -o x.parquet -f parquet x.sup
   super -o y.parquet -f parquet y.sup
   super -s -c "select x from x.parquet join y.parquet on x=y"
-  ! super -s -c "select x from x.parquet join y.sup on x=y"
+  touch z.sup
+  ! super -s -c "select x from x.parquet join z.sup on x=y"
 
 inputs:
   - name: x.sup
@@ -20,8 +21,8 @@ outputs:
   - name: stderr
     data: |
       "x": ambiguous column reference at line 1, column 39:
-      select x from x.parquet join y.sup on x=y
+      select x from x.parquet join z.sup on x=y
                                             ~
       "x": ambiguous column reference at line 1, column 8:
-      select x from x.parquet join y.sup on x=y
+      select x from x.parquet join z.sup on x=y
              ~

compiler/sfmt/ztests/decls.yaml

@@ -1,7 +1,7 @@
 script: |
   super compile -C -I test.spq
   echo "==="
-  super compile -dag -C -I test.spq
+  super compile -C -dag -dynamic -I test.spq
 
 inputs:
   - name: test.spq