refactor(query): allow code to compile despite BucketsAccessed (#13282)

BucketsAccessed doesn't work currently with a private flux.Spec.
See this issue: https://github.com/influxdata/influxdb/issues/13278

This set of changes just allows code to compile until #13278 is fixed.
Note that preauthorization is not working in the meantime.

Fixes #13275.

query/influxql/compiler.go

@@ -62,21 +62,11 @@ func (c *Compiler) Compile(ctx context.Context) (flux.Program, error) {
 		return nil, err
 	}
 
-	fs, err := flux.CompileAST(ctx, astPkg, now)
-	if err != nil {
-		return nil, err
+	astCompiler := lang.ASTCompiler{
+		AST: astPkg,
+		Now: now,
 	}
-
-	pb := plan.PlannerBuilder{}
-	pb.AddLogicalOptions(c.logicalPlannerOptions...)
-	planner := pb.Build()
-	ps, err := planner.Plan(fs)
-	if err != nil {
-		return nil, err
-	}
-	return &lang.Program{
-		PlanSpec: ps,
-	}, nil
+	return astCompiler.Compile(ctx)
 }
 
 func (c *Compiler) CompilerType() flux.CompilerType {

query/preauthorizer.go

@@ -2,10 +2,10 @@ package query
 
 import (
 	"context"
-
-	"github.com/influxdata/flux"
-	platform "github.com/influxdata/influxdb"
 	"github.com/pkg/errors"
+
+	"github.com/influxdata/flux/ast"
+	platform "github.com/influxdata/influxdb"
 )
 
 // PreAuthorizer provides a method for ensuring that the buckets accessed by a query spec
@@ -13,8 +13,8 @@ import (
 // callers to fail early for operations that are not allowed.  However, it's still possible
 // for authorization to be denied at runtime even if this check passes.
 type PreAuthorizer interface {
-	PreAuthorize(ctx context.Context, spec *flux.Spec, auth platform.Authorizer, orgID *platform.ID) error
-	RequiredPermissions(ctx context.Context, spec *flux.Spec, orgID *platform.ID) ([]platform.Permission, error)
+	PreAuthorize(ctx context.Context, ast *ast.Package, auth platform.Authorizer, orgID *platform.ID) error
+	RequiredPermissions(ctx context.Context, ast *ast.Package, orgID *platform.ID) ([]platform.Permission, error)
 }
 
 // NewPreAuthorizer creates a new PreAuthorizer
@@ -28,12 +28,11 @@ type preAuthorizer struct {
 
 // PreAuthorize finds all the buckets read and written by the given spec, and ensures that execution is allowed
 // given the Authorizer.  Returns nil on success, and an error with an appropriate message otherwise.
-func (a *preAuthorizer) PreAuthorize(ctx context.Context, spec *flux.Spec, auth platform.Authorizer, orgID *platform.ID) error {
-	readBuckets, writeBuckets, err := BucketsAccessed(spec, orgID)
-
-	if err != nil {
-		return errors.Wrap(err, "could not retrieve buckets for query.Spec")
-	}
+func (a *preAuthorizer) PreAuthorize(ctx context.Context, ast *ast.Package, auth platform.Authorizer, orgID *platform.ID) error {
+	// TODO(cwolff): re-enable the ability to pre-authorize by determining the buckets accessed by a Flux script
+	//  See https://github.com/influxdata/influxdb/issues/13278
+	readBuckets := make([]platform.BucketFilter, 0)
+	writeBuckets := make([]platform.BucketFilter, 0)
 
 	for _, readBucketFilter := range readBuckets {
 		bucket, err := a.bucketService.FindBucket(ctx, readBucketFilter)
@@ -75,13 +74,11 @@ func (a *preAuthorizer) PreAuthorize(ctx context.Context, spec *flux.Spec, auth
 
 // RequiredPermissions returns a slice of permissions required for the query contained in spec.
 // This method also validates that the buckets exist.
-func (a *preAuthorizer) RequiredPermissions(ctx context.Context, spec *flux.Spec, orgID *platform.ID) ([]platform.Permission, error) {
-	readBuckets, writeBuckets, err := BucketsAccessed(spec, orgID)
-
-	if err != nil {
-		return nil, errors.Wrap(err, "could not retrieve buckets for query.Spec")
-	}
-
+func (a *preAuthorizer) RequiredPermissions(ctx context.Context, ast *ast.Package, orgID *platform.ID) ([]platform.Permission, error) {
+	// TODO(cwolff): re-enable the ability to pre-authorize by determining the buckets accessed by a Flux script
+	//  See https://github.com/influxdata/influxdb/issues/13278
+	readBuckets := make([]platform.BucketFilter, 0)
+	writeBuckets := make([]platform.BucketFilter, 0)
 	ps := make([]platform.Permission, 0, len(readBuckets)+len(writeBuckets))
 	for _, readBucketFilter := range readBuckets {
 		bucket, err := a.bucketService.FindBucket(ctx, readBucketFilter)

query/preauthorizer_test.go

@@ -3,7 +3,6 @@ package query_test
 import (
 	"context"
 	"testing"
-	"time"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/influxdata/flux"
@@ -29,8 +28,8 @@ func newBucketServiceWithOneBucket(bucket platform.Bucket) platform.BucketServic
 }
 
 func TestPreAuthorizer_PreAuthorize(t *testing.T) {
+	t.Skip("Re-enable when pre-authorizer works again")
 	ctx := context.Background()
-	now := time.Now().UTC()
 	// fresh pre-authorizer
 	auth := &platform.Authorization{Status: platform.Active}
 	emptyBucketService := mock.NewBucketService()
@@ -38,11 +37,11 @@ func TestPreAuthorizer_PreAuthorize(t *testing.T) {
 
 	// Try to pre-authorize invalid bucketID
 	q := `from(bucketID:"invalid") |> range(start:-2h) |> yield()`
-	spec, err := flux.Compile(ctx, q, now)
+	ast, err := flux.Parse(q)
 	if err != nil {
 		t.Fatalf("Error compiling query: %v", err)
 	}
-	err = preAuthorizer.PreAuthorize(ctx, spec, auth, nil)
+	err = preAuthorizer.PreAuthorize(ctx, ast, auth, nil)
 	if diagnostic := cmp.Diff("bucket service returned nil bucket", err.Error()); diagnostic != "" {
 		t.Errorf("Authorize message mismatch: -want/+got:\n%v", diagnostic)
 	}
@@ -50,11 +49,11 @@ func TestPreAuthorizer_PreAuthorize(t *testing.T) {
 	// Try to pre-authorize a valid from with bucket service with no buckets
 	// and no authorization
 	q = `from(bucket:"my_bucket") |> range(start:-2h) |> yield()`
-	spec, err = flux.Compile(ctx, q, now)
+	ast, err = flux.Parse(q)
 	if err != nil {
 		t.Fatalf("Error compiling query: %v", err)
 	}
-	err = preAuthorizer.PreAuthorize(ctx, spec, auth, nil)
+	err = preAuthorizer.PreAuthorize(ctx, ast, auth, nil)
 	if diagnostic := cmp.Diff("bucket service returned nil bucket", err.Error()); diagnostic != "" {
 		t.Errorf("Authorize message mismatch: -want/+got:\n%v", diagnostic)
 	}
@@ -73,7 +72,7 @@ func TestPreAuthorizer_PreAuthorize(t *testing.T) {
 	})
 
 	preAuthorizer = query.NewPreAuthorizer(bucketService)
-	err = preAuthorizer.PreAuthorize(ctx, spec, auth, &orgID)
+	err = preAuthorizer.PreAuthorize(ctx, ast, auth, &orgID)
 	if diagnostic := cmp.Diff(`no read permission for bucket: "my_bucket"`, err.Error()); diagnostic != "" {
 		t.Errorf("Authorize message mismatch: -want/+got:\n%v", diagnostic)
 	}
@@ -88,13 +87,14 @@ func TestPreAuthorizer_PreAuthorize(t *testing.T) {
 		Permissions: []platform.Permission{*p},
 	}
 
-	err = preAuthorizer.PreAuthorize(ctx, spec, auth, &orgID)
+	err = preAuthorizer.PreAuthorize(ctx, ast, auth, &orgID)
 	if err != nil {
 		t.Errorf("Expected successful authorization, but got error: \"%v\"", err.Error())
 	}
 }
 
 func TestPreAuthorizer_RequiredPermissions(t *testing.T) {
+	t.Skip("Re-enable when pre-authorizer works again")
 	ctx := context.Background()
 
 	i := inmem.NewService()
@@ -113,13 +113,13 @@ func TestPreAuthorizer_RequiredPermissions(t *testing.T) {
 	}
 
 	const script = `from(bucket:"b-from") |> range(start:-1m) |> to(bucket:"b-to", org:"o")`
-	spec, err := flux.Compile(ctx, script, time.Now())
+	ast, err := flux.Parse(script)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	preAuthorizer := query.NewPreAuthorizer(i)
-	perms, err := preAuthorizer.RequiredPermissions(ctx, spec, &o.ID)
+	perms, err := preAuthorizer.RequiredPermissions(ctx, ast, &o.ID)
 	if err != nil {
 		t.Fatal(err)
 	}

query/querytest/compile.go

@@ -1,11 +1,8 @@
 package querytest
 
 import (
-	"context"
-	"testing"
-	"time"
-
 	"fmt"
+	"testing"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
@@ -13,7 +10,6 @@ import (
 	"github.com/influxdata/flux/semantic/semantictest"
 	"github.com/influxdata/flux/stdlib/universe"
 	platform "github.com/influxdata/influxdb"
-	"github.com/influxdata/influxdb/query"
 )
 
 type BucketAwareQueryTestCase struct {
@@ -34,43 +30,44 @@ var opts = append(
 )
 
 func BucketAwareQueryTestHelper(t *testing.T, tc BucketAwareQueryTestCase) {
+	t.Skip("BucketsAccessed needs re-implementing; see https://github.com/influxdata/influxdb/issues/13278")
 	t.Helper()
 
-	now := time.Now().UTC()
-	got, err := flux.Compile(context.Background(), tc.Raw, now)
-	if (err != nil) != tc.WantErr {
-		t.Errorf("error compiling spec error: %v, wantErr %v", err, tc.WantErr)
-		return
-	}
-	if tc.WantErr {
-		return
-	}
-	if tc.Want != nil {
-		tc.Want.Now = now
-		if !cmp.Equal(tc.Want, got, opts...) {
-			t.Errorf("unexpected specs -want/+got %s", cmp.Diff(tc.Want, got, opts...))
-		}
-	}
-
-	var gotReadBuckets, gotWriteBuckets []platform.BucketFilter
-	if tc.WantReadBuckets != nil || tc.WantWriteBuckets != nil {
-		gotReadBuckets, gotWriteBuckets, err = query.BucketsAccessed(got, nil)
-		if err != nil {
-			t.Fatal(err)
-		}
-	}
-
-	if tc.WantReadBuckets != nil {
-		if diagnostic := verifyBuckets(*tc.WantReadBuckets, gotReadBuckets); diagnostic != "" {
-			t.Errorf("Could not verify read buckets: %v", diagnostic)
-		}
-	}
-
-	if tc.WantWriteBuckets != nil {
-		if diagnostic := verifyBuckets(*tc.WantWriteBuckets, gotWriteBuckets); diagnostic != "" {
-			t.Errorf("Could not verify write buckets: %v", diagnostic)
-		}
-	}
+	//now := time.Now().UTC()
+	//got, err := flux.Compile(context.Background(), tc.Raw, now)
+	//if (err != nil) != tc.WantErr {
+	//	t.Errorf("error compiling spec error: %v, wantErr %v", err, tc.WantErr)
+	//	return
+	//}
+	//if tc.WantErr {
+	//	return
+	//}
+	//if tc.Want != nil {
+	//	tc.Want.Now = now
+	//	if !cmp.Equal(tc.Want, got, opts...) {
+	//		t.Errorf("unexpected specs -want/+got %s", cmp.Diff(tc.Want, got, opts...))
+	//	}
+	//}
+	//
+	//var gotReadBuckets, gotWriteBuckets []platform.BucketFilter
+	//if tc.WantReadBuckets != nil || tc.WantWriteBuckets != nil {
+	//	gotReadBuckets, gotWriteBuckets, err = query.BucketsAccessed(got, nil)
+	//	if err != nil {
+	//		t.Fatal(err)
+	//	}
+	//}
+	//
+	//if tc.WantReadBuckets != nil {
+	//	if diagnostic := verifyBuckets(*tc.WantReadBuckets, gotReadBuckets); diagnostic != "" {
+	//		t.Errorf("Could not verify read buckets: %v", diagnostic)
+	//	}
+	//}
+	//
+	//if tc.WantWriteBuckets != nil {
+	//	if diagnostic := verifyBuckets(*tc.WantWriteBuckets, gotWriteBuckets); diagnostic != "" {
+	//		t.Errorf("Could not verify write buckets: %v", diagnostic)
+	//	}
+	//}
 }
 
 func verifyBuckets(wantBuckets, gotBuckets []platform.BucketFilter) string {

task/validator.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"time"
 
 	"github.com/influxdata/flux"
 	platform "github.com/influxdata/influxdb"
@@ -337,7 +336,7 @@ func (ts *taskServiceValidator) validateBucket(ctx context.Context, script strin
 		return err
 	}
 
-	spec, err := flux.Compile(ctx, script, time.Now())
+	ast, err := flux.Parse(script)
 	if err != nil {
 		return platform.NewError(
 			platform.WithErrorErr(err),
@@ -345,7 +344,7 @@ func (ts *taskServiceValidator) validateBucket(ctx context.Context, script strin
 			platform.WithErrorCode(platform.EInvalid))
 	}
 
-	if err := ts.preAuth.PreAuthorize(ctx, spec, auth, &orgID); err != nil {
+	if err := ts.preAuth.PreAuthorize(ctx, ast, auth, &orgID); err != nil {
 		ts.logger.With(loggerFields...).Info("Task failed preauthorization check",
 			zap.String("user_id", auth.GetUserID().String()),
 			zap.String("org_id", orgID.String()),