refactor: general refactors around removing unused functions and types (#5094)

* refactor: remove spec json marshal and unmarshal functions

The spec marshaling and unmarshaling functions are unused and don't
work. Removing the code related to this.

* refactor: remove all calls to RegisterOpSpec

This removes all calls to `RegisterOpSpec` with the intent to deprecate
and remove the function.

This function doesn't do anything in the current version of Flux. A long
time ago, it was possible to serialize and deserialize the flux spec
which was generated from the interpreter.

We removed this ability from everything except for the interpreter
because it was unreliable and led to bugs. Since this code was only
useful for the serialization and deserialization, and this is long
unsupported and doesn't work, I'm finally removing the setup function
that enabled this capability since it doesn't work and has led mostly to
confusion about what's necessary when creating a source/transformation.

* refactor: remove unused spec format functions

* refactor: remove the unused priority type

Author: jsternberg

format.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"log"
 	"time"
 
 	"github.com/influxdata/flux"
@@ -50,8 +49,6 @@ func AddCompilerMappings(mappings flux.CompilerMappings) error {
 type CompileOption func(*compileOptions)
 
 type compileOptions struct {
-	verbose bool
-
 	extern flux.ASTHandle
 
 	planOptions struct {
@@ -92,12 +89,6 @@ func applyOptions(opts ...CompileOption) *compileOptions {
 // NOTE: compileOptions can be used only when invoking Compile* functions.
 // They can't be used when unmarshaling a Compiler and invoking its Compile method.
 
-func Verbose(v bool) CompileOption {
-	return func(o *compileOptions) {
-		o.verbose = v
-	}
-}
-
 // Compile evaluates a Flux script producing a flux.Program.
 // now parameter must be non-zero, that is the default now time should be set before compiling.
 func Compile(q string, runtime flux.Runtime, now time.Time, opts ...CompileOption) (*AstProgram, error) {
@@ -129,9 +120,6 @@ func CompileTableObject(ctx context.Context, to *flux.TableObject, now time.Time
 	if err != nil {
 		return nil, err
 	}
-	if o.verbose {
-		log.Println("Query Spec: ", flux.Formatted(s, flux.FmtJSON))
-	}
 	ps, err := buildPlan(ctx, s, o)
 	if err != nil {
 		return nil, err
@@ -499,9 +487,6 @@ func (p *AstProgram) Start(ctx context.Context, alloc memory.Allocator) (flux.Qu
 
 	// Planning.
 	s, cctx := opentracing.StartSpanFromContext(ctx, "plan")
-	if p.opts.verbose {
-		log.Println("Query Spec: ", flux.Formatted(sp, flux.FmtJSON))
-	}
 	if err := p.updateOpts(scope); err != nil {
 		return nil, errors.Wrap(err, codes.Inherit, "error in reading options while starting program")
 	}

lang/compiler.go

@@ -1,10 +1,6 @@
 package flux
 
 import (
-	"encoding/json"
-
-	"github.com/influxdata/flux/codes"
-	"github.com/influxdata/flux/internal/errors"
 	"github.com/influxdata/flux/interpreter"
 )
 
@@ -15,58 +11,6 @@ type Operation struct {
 	Source OperationSource `json:"source"`
 }
 
-func (o *Operation) UnmarshalJSON(data []byte) error {
-	type Alias Operation
-	raw := struct {
-		*Alias
-		Kind OperationKind   `json:"kind"`
-		Spec json.RawMessage `json:"spec"`
-	}{}
-	err := json.Unmarshal(data, &raw)
-	if err != nil {
-		return err
-	}
-	if raw.Alias != nil {
-		*o = *(*Operation)(raw.Alias)
-	}
-	spec, err := unmarshalOpSpec(raw.Kind, raw.Spec)
-	if err != nil {
-		return errors.Wrapf(err, codes.Inherit, "failed to unmarshal operation %q", o.ID)
-	}
-	o.Spec = spec
-	return nil
-}
-
-func unmarshalOpSpec(k OperationKind, data []byte) (OperationSpec, error) {
-	createOpSpec, ok := kindToOp[k]
-	if !ok {
-		return nil, errors.Newf(codes.Invalid, "unknown operation spec kind %v", k)
-	}
-	spec := createOpSpec()
-
-	if len(data) > 0 {
-		err := json.Unmarshal(data, spec)
-		if err != nil {
-			return nil, err
-		}
-	}
-	return spec, nil
-}
-
-func (o Operation) MarshalJSON() ([]byte, error) {
-	type Alias Operation
-	raw := struct {
-		Kind OperationKind `json:"kind"`
-		Alias
-	}{
-		Kind:  o.Spec.Kind(),
-		Alias: (Alias)(o),
-	}
-	return json.Marshal(raw)
-}
-
-type NewOperationSpec func() OperationSpec
-
 // OperationSpec specifies an operation as part of a query.
 type OperationSpec interface {
 	// Kind returns the kind of the operation.
@@ -84,24 +28,3 @@ type OperationID string
 
 // OperationKind denotes the kind of operations.
 type OperationKind string
-
-var kindToOp = make(map[OperationKind]NewOperationSpec)
-
-// RegisterOpSpec registers an operation spec with a given kind.
-// k is a label that uniquely identifies this operation. If the kind has already been registered the call panics.
-// c is a function reference that creates a new, default-initialized opSpec for the given kind.
-// TODO:(nathanielc) make this part of RegisterMethod/RegisterFunction
-func RegisterOpSpec(k OperationKind, c NewOperationSpec) {
-	if kindToOp[k] != nil {
-		panic(errors.Newf(codes.Internal, "duplicate registration for operation kind %v", k))
-	}
-	kindToOp[k] = c
-}
-
-func NumberOfOperations() int {
-	return len(kindToOp)
-}
-
-func OperationSpecNewFn(k OperationKind) NewOperationSpec {
-	return kindToOp[k]
-}

operation.go

@@ -1,19 +1,7 @@
 package flux
 
-import (
-	"math"
-	"strconv"
-
-	"github.com/influxdata/flux/codes"
-	"github.com/influxdata/flux/internal/errors"
-)
-
 // ResourceManagement defines how the query should consume avaliable resources.
 type ResourceManagement struct {
-	// Priority or the query.
-	// Queries with a lower value will move to the front of the priority queue.
-	// A zero value indicates the highest priority.
-	Priority Priority `json:"priority"`
 	// ConcurrencyQuota is the number of concurrency workers allowed to process this query.
 	// A zero value indicates the planner can pick the optimal concurrency.
 	ConcurrencyQuota int `json:"concurrency_quota"`
@@ -22,42 +10,3 @@ type ResourceManagement struct {
 	// A zero value indicates unlimited.
 	MemoryBytesQuota int64 `json:"memory_bytes_quota"`
 }
-
-// Priority is an integer that represents the query priority.
-// Any positive 32bit integer value may be used.
-// Special constants are provided to represent the extreme high and low priorities.
-type Priority int32
-
-const (
-	// High is the highest possible priority = 0
-	High Priority = 0
-	// Low is the lowest possible priority = MaxInt32
-	Low Priority = math.MaxInt32
-)
-
-func (p Priority) MarshalText() ([]byte, error) {
-	switch p {
-	case Low:
-		return []byte("low"), nil
-	case High:
-		return []byte("high"), nil
-	default:
-		return []byte(strconv.FormatInt(int64(p), 10)), nil
-	}
-}
-
-func (p *Priority) UnmarshalText(txt []byte) error {
-	switch s := string(txt); s {
-	case "low":
-		*p = Low
-	case "high":
-		*p = High
-	default:
-		i, err := strconv.ParseInt(s, 10, 32)
-		if err != nil {
-			return errors.Wrap(err, codes.Invalid, "invalid priority, must be an integer or 'low','high'")
-		}
-		*p = Priority(i)
-	}
-	return nil
-}

querytest/operations.go

@@ -28,7 +28,6 @@ type DurationOpSpec struct {
 func init() {
 	durationSignature := runtime.MustLookupBuiltinType(pkgPath, DurationKind)
 	runtime.RegisterPackageValue(pkgPath, DurationKind, flux.MustValue(flux.FunctionValue(DurationKind, createDurationOpSpec, durationSignature)))
-	flux.RegisterOpSpec(DurationKind, newDurationOp)
 	plan.RegisterProcedureSpec(DurationKind, newDurationProcedure, DurationKind)
 	execute.RegisterTransformation(DurationKind, createDurationTransformation)
 }
@@ -85,10 +84,6 @@ func createDurationOpSpec(args flux.Arguments, a *flux.Administration) (flux.Ope
 	return spec, nil
 }
 
-func newDurationOp() flux.OperationSpec {
-	return new(DurationOpSpec)
-}
-
 func (s *DurationOpSpec) Kind() flux.OperationKind {
 	return DurationKind
 }

resource_management.go

@@ -167,17 +167,6 @@ func TestDuration_NewQuery(t *testing.T) {
 	}
 }
 
-func TestDurationOperation_Marshaling(t *testing.T) {
-	data := []byte(`{"id":"duration","kind":"duration","spec":{"timeColumn": "_time"}}`)
-	op := &flux.Operation{
-		ID: "duration",
-		Spec: &events.DurationOpSpec{
-			TimeColumn: "_time",
-		},
-	}
-	querytest.OperationMarshalingTestHelper(t, data, op)
-}
-
 func TestDuration_PassThrough(t *testing.T) {
 	executetest.TransformationPassThroughTestHelper(t, func(d execute.Dataset, c execute.TableBuilderCache) execute.Transformation {
 		s := events.NewDurationTransformation(

stdlib/contrib/tomhollingworth/events/duration.go

@@ -33,7 +33,6 @@ const (
 func init() {
 	fromCSVSignature := runtime.MustLookupBuiltinType("csv", "from")
 	runtime.RegisterPackageValue("csv", "from", flux.MustValue(flux.FunctionValue(FromCSVKind, createFromCSVOpSpec, fromCSVSignature)))
-	flux.RegisterOpSpec(FromCSVKind, newFromCSVOp)
 	plan.RegisterProcedureSpec(FromCSVKind, newFromCSVProcedure, FromCSVKind)
 	execute.RegisterSource(FromCSVKind, createFromCSVSource)
 }
@@ -73,10 +72,6 @@ func createFromCSVOpSpec(args flux.Arguments, a *flux.Administration) (flux.Oper
 	return spec, nil
 }
 
-func newFromCSVOp() flux.OperationSpec {
-	return new(FromCSVOpSpec)
-}
-
 func (s *FromCSVOpSpec) Kind() flux.OperationKind {
 	return FromCSVKind
 }

stdlib/contrib/tomhollingworth/events/duration_test.go

@@ -255,18 +255,6 @@ func TestFromCSV_NewQuery(t *testing.T) {
 	}
 }
 
-func TestFromCSVOperation_Marshaling(t *testing.T) {
-	data := []byte(`{"id":"fromCSV","kind":"fromCSV","spec":{"csv":"1,2","mode":"annotations"}}`)
-	op := &flux.Operation{
-		ID: "fromCSV",
-		Spec: &csv.FromCSVOpSpec{
-			CSV:  "1,2",
-			Mode: "annotations",
-		},
-	}
-	querytest.OperationMarshalingTestHelper(t, data, op)
-}
-
 func TestFromCSV_Run(t *testing.T) {
 	spec := &csv.FromCSVProcedureSpec{
 		CSV: `#datatype,string,long,dateTime:RFC3339,dateTime:RFC3339,dateTime:RFC3339,string,string,double

stdlib/csv/from.go

@@ -26,7 +26,6 @@ If a builtin OperationSpec is not found, then it will check for functions define
 The following registrations are typically executed in the function's init() for the query phase to execute properly:
 
 	flux.RegisterPackageValue(pkgpath, name string, value values.Value)
-	flux.RegisterOpSpec(k flux.OperationKind, c flux.NewOperationSpec)
 
 Note that to register a function value with a package, the value passed into flux.RegisterPackageValue is computed using the
 followingfunction: