Merge branch 'master' into chore/update-flux

mock/reader.go

@@ -7,27 +7,27 @@ import (
 	"github.com/influxdata/influxdb/v2/query"
 )
 
-type StoreReader struct {
+type StorageReader struct {
 	ReadFilterFn    func(ctx context.Context, spec query.ReadFilterSpec, alloc *memory.Allocator) (query.TableIterator, error)
 	ReadGroupFn     func(ctx context.Context, spec query.ReadGroupSpec, alloc *memory.Allocator) (query.TableIterator, error)
 	ReadTagKeysFn   func(ctx context.Context, spec query.ReadTagKeysSpec, alloc *memory.Allocator) (query.TableIterator, error)
 	ReadTagValuesFn func(ctx context.Context, spec query.ReadTagValuesSpec, alloc *memory.Allocator) (query.TableIterator, error)
 	CloseFn         func()
 }
 
-func (s *StoreReader) ReadFilter(ctx context.Context, spec query.ReadFilterSpec, alloc *memory.Allocator) (query.TableIterator, error) {
+func (s *StorageReader) ReadFilter(ctx context.Context, spec query.ReadFilterSpec, alloc *memory.Allocator) (query.TableIterator, error) {
 	return s.ReadFilterFn(ctx, spec, alloc)
 }
 
-func (s *StoreReader) ReadGroup(ctx context.Context, spec query.ReadGroupSpec, alloc *memory.Allocator) (query.TableIterator, error) {
+func (s *StorageReader) ReadGroup(ctx context.Context, spec query.ReadGroupSpec, alloc *memory.Allocator) (query.TableIterator, error) {
 	return s.ReadGroupFn(ctx, spec, alloc)
 }
 
-func (s *StoreReader) ReadTagKeys(ctx context.Context, spec query.ReadTagKeysSpec, alloc *memory.Allocator) (query.TableIterator, error) {
+func (s *StorageReader) ReadTagKeys(ctx context.Context, spec query.ReadTagKeysSpec, alloc *memory.Allocator) (query.TableIterator, error) {
 	return s.ReadTagKeysFn(ctx, spec, alloc)
 }
 
-func (s *StoreReader) ReadTagValues(ctx context.Context, spec query.ReadTagValuesSpec, alloc *memory.Allocator) (query.TableIterator, error) {
+func (s *StorageReader) ReadTagValues(ctx context.Context, spec query.ReadTagValuesSpec, alloc *memory.Allocator) (query.TableIterator, error) {
 	return s.ReadTagValuesFn(ctx, spec, alloc)
 }
 
@@ -41,12 +41,12 @@ func (s *StorageReader) Close() {
 }
 
 type WindowAggregateStoreReader struct {
-	*StoreReader
+	*StorageReader
 	HasWindowAggregateCapabilityFn func(ctx context.Context) bool
 	ReadWindowAggregateFn          func(ctx context.Context, spec query.ReadWindowAggregateSpec, alloc *memory.Allocator) (query.TableIterator, error)
 }
 
-func (s *WindowAggregateStoreReader) HasWindowAggregateCapability(ctx context.Context, capability ...*influxdb.WindowAggregateCapability) bool {
+func (s *WindowAggregateStoreReader) HasWindowAggregateCapability(ctx context.Context, capability ...*query.WindowAggregateCapability) bool {
 	// Use the function if it exists.
 	if s.HasWindowAggregateCapabilityFn != nil {
 		return s.HasWindowAggregateCapabilityFn(ctx)

query/stdlib/influxdata/influxdb/operators.go

@@ -110,7 +110,7 @@ type ReadWindowAggregatePhysSpec struct {
 	ReadRangePhysSpec
 
 	WindowEvery int64
-	Aggregates  []string
+	Aggregates  []plan.ProcedureKind
 }
 
 func (s *ReadWindowAggregatePhysSpec) Kind() plan.ProcedureKind {

query/stdlib/influxdata/influxdb/source_test.go

@@ -201,7 +201,7 @@ func TestReadWindowAggregateSource(t *testing.T) {
 					BucketID: bucketID.String(),
 				},
 				WindowEvery: 10,
-				Aggregates: []string{
+				Aggregates: []plan.ProcedureKind{
 					universe.SumKind,
 				},
 			}

query/storage.go

@@ -24,10 +24,15 @@ type StorageReader interface {
 	Close()
 }
 
+// WindowAggregateCapability describes what is supported by WindowAggregateReader.
+type WindowAggregateCapability struct{}
+
 // WindowAggregateReader implements the WindowAggregate capability.
 type WindowAggregateReader interface {
 	// HasWindowAggregateCapability will test if this Reader source supports the ReadWindowAggregate capability.
-	HasWindowAggregateCapability(ctx context.Context) bool
+	// If WindowAggregateCapability is passed to the method, then the struct
+	// is filled with a detailed list of what the RPC call supports.
+	HasWindowAggregateCapability(ctx context.Context, capability ...*WindowAggregateCapability) bool
 
 	// ReadWindowAggregate will read a table using the WindowAggregate method.
 	ReadWindowAggregate(ctx context.Context, spec ReadWindowAggregateSpec, alloc *memory.Allocator) (TableIterator, error)

storage/flux/reader.go

@@ -80,6 +80,23 @@ func (r *storeReader) ReadGroup(ctx context.Context, spec query.ReadGroupSpec, a
 	}, nil
 }
 
+func (r *storeReader) HasWindowAggregateCapability(ctx context.Context, capability ...*query.WindowAggregateCapability) bool {
+	if aggStore, ok := r.s.(storage.WindowAggregateStore); ok {
+		return aggStore.HasWindowAggregateCapability(ctx)
+	}
+	return false
+}
+
+func (r *storeReader) ReadWindowAggregate(ctx context.Context, spec query.ReadWindowAggregateSpec, alloc *memory.Allocator) (query.TableIterator, error) {
+	return &windowAggregateIterator{
+		ctx:   ctx,
+		s:     r.s,
+		spec:  spec,
+		cache: newTagsCache(0),
+		alloc: alloc,
+	}, nil
+}
+
 func (r *storeReader) ReadTagKeys(ctx context.Context, spec query.ReadTagKeysSpec, alloc *memory.Allocator) (query.TableIterator, error) {
 	return &tagKeysIterator{
 		ctx:       ctx,
@@ -495,7 +512,7 @@ func groupKeyForGroup(kv [][]byte, spec *query.ReadGroupSpec, bnds execute.Bound
 type windowAggregateIterator struct {
 	ctx   context.Context
 	s     storage.Store
-	spec  influxdb.ReadWindowAggregateSpec
+	spec  query.ReadWindowAggregateSpec
 	stats cursors.CursorStats
 	cache *tagsCache
 	alloc *memory.Allocator
@@ -515,44 +532,35 @@ func (wai *windowAggregateIterator) Do(f func(flux.Table) error) error {
 		return err
 	}
 
-	var predicate *datatypes.Predicate
-	if wai.spec.Predicate != nil {
-		p, err := toStoragePredicate(wai.spec.Predicate)
-		if err != nil {
-			return err
-		}
-		predicate = p
-	}
-
 	var req datatypes.ReadWindowAggregateRequest
 	req.ReadSource = any
-	req.Predicate = predicate
+	req.Predicate = wai.spec.Predicate
 	req.Range.Start = int64(wai.spec.Bounds.Start)
 	req.Range.End = int64(wai.spec.Bounds.Stop)
 
 	req.WindowEvery = wai.spec.WindowEvery
 	req.Aggregate = make([]*datatypes.Aggregate, len(wai.spec.Aggregates))
 	for i, aggKind := range wai.spec.Aggregates {
-		if agg, err := determineAggregateMethod(aggKind); err != nil {
+		if agg, err := determineAggregateMethod(string(aggKind)); err != nil {
 			return err
 		} else if agg != datatypes.AggregateTypeNone {
 			req.Aggregate[i] = &datatypes.Aggregate{Type: agg}
 		}
 	}
 
-	if aggStore, ok := wai.s.(storage.WindowAggregateStore); !ok {
-		return errors.New("storage does not support window aggregate.")
-	} else {
-		rs, err := aggStore.WindowAggregate(wai.ctx, &req)
-		if err != nil {
-			return err
-		}
+	aggStore, ok := wai.s.(storage.WindowAggregateStore)
+	if !ok {
+		return errors.New("storage does not support window aggregate")
+	}
+	rs, err := aggStore.WindowAggregate(wai.ctx, &req)
+	if err != nil {
+		return err
+	}
 
-		if rs == nil {
-			return nil
-		}
-		return wai.handleRead(f, rs)
+	if rs == nil {
+		return nil
 	}
+	return wai.handleRead(f, rs)
 }
 
 func (wai *windowAggregateIterator) handleRead(f func(flux.Table) error, rs storage.ResultSet) error {