build(flux): update flux to v0.172.0 (#23494)

* build(flux): update flux to v0.172.0

* refactor(fluxtest): updates the flux test executor to the new interface

go.mod

@@ -29,7 +29,7 @@ require (
 	github.com/google/go-jsonnet v0.17.0
 	github.com/hashicorp/vault/api v1.0.2
 	github.com/influxdata/cron v0.0.0-20201006132531-4bb0a200dcbe
-	github.com/influxdata/flux v0.171.0
+	github.com/influxdata/flux v0.172.0
 	github.com/influxdata/httprouter v1.3.1-0.20191122104820-ee83e2772f69
 	github.com/influxdata/influx-cli/v2 v2.2.1-0.20220318222112-88ba3464cd07
 	github.com/influxdata/influxql v1.1.1-0.20211004132434-7e7d61973256

go.sum

@@ -497,16 +497,14 @@ github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NH
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/influxdata/cron v0.0.0-20201006132531-4bb0a200dcbe h1:7j4SdN/BvQwN6WoUq7mv0kg5U9NhnFBxPGMafYRKym0=
 github.com/influxdata/cron v0.0.0-20201006132531-4bb0a200dcbe/go.mod h1:XabtPPW2qsCg0tl+kjaPU+cFS+CjQXEXbT1VJvHT4og=
-github.com/influxdata/flux v0.171.0 h1:9s0MA0bGXPRmzeAvZPYl1412qYSdeTNQb1cgW83nu2M=
-github.com/influxdata/flux v0.171.0/go.mod h1:fNtcZ8tqtVDjwWYcPRvCdlY5t3n+NYCc5xunKCmigQA=
+github.com/influxdata/flux v0.172.0 h1:aDLGOsQAVGlABAOHRpq9Nuog9Gv3bDN2hcnhW50kWWk=
+github.com/influxdata/flux v0.172.0/go.mod h1:fNtcZ8tqtVDjwWYcPRvCdlY5t3n+NYCc5xunKCmigQA=
 github.com/influxdata/gosnowflake v1.6.9 h1:BhE39Mmh8bC+Rvd4QQsP2gHypfeYIH1wqW1AjGWxxrE=
 github.com/influxdata/gosnowflake v1.6.9/go.mod h1:9W/BvCXOKx2gJtQ+jdi1Vudev9t9/UDOEHnlJZ/y1nU=
 github.com/influxdata/httprouter v1.3.1-0.20191122104820-ee83e2772f69 h1:WQsmW0fXO4ZE/lFGIE84G6rIV5SJN3P3sjIXAP1a8eU=
 github.com/influxdata/httprouter v1.3.1-0.20191122104820-ee83e2772f69/go.mod h1:pwymjR6SrP3gD3pRj9RJwdl1j5s3doEEV8gS4X9qSzA=
 github.com/influxdata/influx-cli/v2 v2.2.1-0.20220318222112-88ba3464cd07 h1:qfj5kTFYg5KhePA0A5BzFV6zxW0yLYF5K0O7t3drqn8=
 github.com/influxdata/influx-cli/v2 v2.2.1-0.20220318222112-88ba3464cd07/go.mod h1:p1X8Ga67SzLC35qmwvTCmWXdpZOTHSWWMXJ0zwRTW50=
-github.com/influxdata/influxdb v1.9.7 h1:asjvZJ8NFFmxkSw+kOJj1ItGLQdU1nvRQE3jvdQXeRU=
-github.com/influxdata/influxdb v1.9.7/go.mod h1:YZMcI9MYeMGLcg7Td7z5YRk52tL85r5bF4qX6WCnSt4=
 github.com/influxdata/influxdb-client-go/v2 v2.3.1-0.20210518120617-5d1fff431040 h1:MBLCfcSsUyFPDJp6T7EoHp/Ph3Jkrm4EuUKLD2rUWHg=
 github.com/influxdata/influxdb-client-go/v2 v2.3.1-0.20210518120617-5d1fff431040/go.mod h1:vLNHdxTJkIf2mSLvGrpj8TCcISApPoXkaxP8g9uRlW8=
 github.com/influxdata/influxdb1-client v0.0.0-20191209144304-8bf82d3c094d/go.mod h1:qj24IKcXYK6Iy9ceXlo3Tc+vtHo9lIhSX5JddghvEPo=

internal/cmd/fluxtest-harness-influxdb/test.go

@@ -10,12 +10,10 @@ import (
 	"io"
 	"os"
 	"os/exec"
-	"strings"
 
 	"github.com/influxdata/flux"
 	"github.com/influxdata/flux/ast"
 	"github.com/influxdata/flux/cmd/flux/cmd"
-	"github.com/influxdata/flux/execute/table"
 	"github.com/influxdata/flux/lang"
 	"github.com/influxdata/flux/parser"
 	"github.com/influxdata/influxdb/v2"
@@ -78,7 +76,7 @@ func (t *testExecutor) Close() error {
 	return nil
 }
 
-func (t *testExecutor) Run(pkg *ast.Package) error {
+func (t *testExecutor) Run(pkg *ast.Package, fn cmd.TestResultFunc) error {
 	l := t.l.Launcher
 	b := &influxdb.Bucket{
 		OrgID: t.l.Org.ID,
@@ -108,15 +106,26 @@ func (t *testExecutor) Run(pkg *ast.Package) error {
 
 	// During the first execution, we are performing the writes
 	// that are in the testcase.
-	err := t.executeWithOptions(bucketOpt, orgOpt, t.writeOptAST, pkg)
+	err := t.executeWithOptions(bucketOpt, orgOpt, t.writeOptAST, pkg, func(ctx context.Context, results flux.ResultIterator) error {
+		for results.More() {
+			res := results.Next()
+			if err := res.Tables().Do(func(table flux.Table) error {
+				table.Done()
+				return nil
+			}); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
 	if err != nil {
 		// Some test assertions can fail in the first pass, so those errors do not fail the test case.
 		// However those errors can be useful when the error is unexpected, therefore we simply log the error here.
 		fmt.Printf("Error from write pass: %s\n", err)
 	}
 
 	// Execute the read pass.
-	err = t.executeWithOptions(bucketOpt, orgOpt, t.readOptAST, pkg)
+	err = t.executeWithOptions(bucketOpt, orgOpt, t.readOptAST, pkg, fn)
 	if flags.wait {
 		// TODO(nathanielc): When the executor is given access to the test name,
 		// make the configName a function of the test name.
@@ -148,7 +157,7 @@ func (t *testExecutor) Run(pkg *ast.Package) error {
 	return err
 }
 
-func (t *testExecutor) executeWithOptions(bucketOpt, orgOpt *ast.OptionStatement, optionsAST *ast.File, pkg *ast.Package) error {
+func (t *testExecutor) executeWithOptions(bucketOpt, orgOpt *ast.OptionStatement, optionsAST *ast.File, pkg *ast.Package, fn cmd.TestResultFunc) error {
 	options := optionsAST.Copy().(*ast.File)
 	options.Body = append([]ast.Statement{bucketOpt, orgOpt}, options.Body...)
 
@@ -172,26 +181,7 @@ func (t *testExecutor) executeWithOptions(bucketOpt, orgOpt *ast.OptionStatement
 	}
 	defer r.Release()
 
-	var output strings.Builder
-	for r.More() {
-		v := r.Next()
-
-		if err := v.Tables().Do(func(tbl flux.Table) error {
-			// The data returned here is the result of `testing.diff`, so any result means that
-			// a comparison of two tables showed inequality. Capture that inequality as part of the error.
-			// XXX: rockstar (08 Dec 2020) - This could use some ergonomic work, as the diff testOutput
-			// is not exactly "human readable."
-			_, _ = fmt.Fprint(&output, table.Stringify(tbl))
-			return nil
-		}); err != nil {
-			return err
-		}
-	}
-	if output.Len() > 0 {
-		return errors.New(output.String())
-	}
-	r.Release()
-	return r.Err()
+	return fn(t.ctx, r)
 }
 
 // This options definition puts to() in the path of the CSV input. The tests

preview.flux

@@ -0,0 +1,8 @@
+import "experimental/influxdb"
+import "internal/debug"
+
+influxdb.preview(bucket: "preview-test")
+|> range(start: -1d)
+|> debug.pass()
+|> group()
+|> aggregateWindow(every: 1m, fn: mean)

seed.flux

@@ -0,0 +1,4 @@
+import "internal/gen"
+
+gen.tables(n: 50000, tags: [{name: "_measurement", cardinality: 1}, {name: "_field", cardinality: 1}, {name: "t0", cardinality: 10}, {name: "t1", cardinality: 5}])
+  |> to(bucket: "preview-test")

test.flux

@@ -0,0 +1,8 @@
+import "array"
+import "profiler"
+import "internal/gen"
+import "runtime"
+
+option profiler.enabledProfilers = ["operator"]
+
+array.from(rows: [{version: runtime.version()}])