add the notion of bounds to plan nodes (#142)

Author: jlapacik

bounds.go

@@ -5,12 +5,13 @@ import "time"
 type Bounds struct {
 	Start Time
 	Stop  Time
+	Now   time.Time
 }
 
 // IsEmpty reports whether the given bounds
 // are empty, i.e., if start >= stop.
-func (b Bounds) IsEmpty(now time.Time) bool {
-	return b.Start.Time(now).Equal(b.Stop.Time(now)) || b.Start.Time(now).After(b.Stop.Time(now))
+func (b Bounds) IsEmpty() bool {
+	return b.Start.Time(b.Now).Equal(b.Stop.Time(b.Now)) || b.Start.Time(b.Now).After(b.Stop.Time(b.Now))
 }
 
 // HasZero returns true if the given bounds contain a Go zero time value as either Start or Stop.

bounds_test.go

@@ -73,53 +73,53 @@ func TestBounds_IsEmpty(t *testing.T) {
 	}{
 		{
 			name: "empty bounds / start == stop",
-			now:  time.Date(2018, time.August, 14, 11, 0, 0, 0, time.UTC),
 			bounds: flux.Bounds{
 				Start: flux.Now,
 				Stop:  flux.Now,
+				Now:   time.Date(2018, time.August, 14, 11, 0, 0, 0, time.UTC),
 			},
 			want: true,
 		},
 		{
 			name: "empty bounds / absolute now == relative now",
-			now:  time.Date(2018, time.August, 14, 11, 0, 0, 0, time.UTC),
 			bounds: flux.Bounds{
 				Start: flux.Now,
 				Stop: flux.Time{
 					Absolute: time.Date(2018, time.August, 14, 11, 0, 0, 0, time.UTC),
 				},
+				Now: time.Date(2018, time.August, 14, 11, 0, 0, 0, time.UTC),
 			},
 			want: true,
 		},
 		{
 			name: "start > stop",
-			now:  time.Date(2018, time.August, 14, 11, 0, 0, 0, time.UTC),
 			bounds: flux.Bounds{
 				Start: flux.Time{
 					IsRelative: true,
 					Relative:   time.Hour,
 				},
 				Stop: flux.Now,
+				Now:  time.Date(2018, time.August, 14, 11, 0, 0, 0, time.UTC),
 			},
 			want: true,
 		},
 		{
 			name: "start < stop",
-			now:  time.Date(2018, time.August, 14, 11, 0, 0, 0, time.UTC),
 			bounds: flux.Bounds{
 				Start: flux.Time{
 					IsRelative: true,
 					Relative:   -1 * time.Hour,
 				},
 				Stop: flux.Now,
+				Now:  time.Date(2018, time.August, 14, 11, 0, 0, 0, time.UTC),
 			},
 			want: false,
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := tt.bounds.IsEmpty(tt.now)
+			got := tt.bounds.IsEmpty()
 			if got != tt.want {
 				t.Errorf("unexpected result for bounds.IsEmpty(): got %t, want %t", got, tt.want)
 			}

execute/executor.go

@@ -138,11 +138,21 @@ func (v *createExecutionNodeVisitor) Visit(node plan.PlanNode) error {
 	kind := spec.Kind()
 	id := plan.ProcedureIDFromNodeID(node.ID())
 
+	// Add explicit stream context if bounds are set on this node
+	var streamContext streamContext
+	if node.Bounds() != nil {
+		streamContext.bounds = &Bounds{
+			Start: node.Bounds().Start,
+			Stop:  node.Bounds().Stop,
+		}
+	}
+
 	// Build execution context
 	ec := executionContext{
-		ctx:     v.ctx,
-		es:      v.es,
-		parents: make([]DatasetID, len(node.Predecessors())),
+		ctx:           v.ctx,
+		es:            v.es,
+		parents:       make([]DatasetID, len(node.Predecessors())),
+		streamContext: streamContext,
 	}
 
 	for i, pred := range node.Predecessors() {

functions/transformations/range.go

@@ -8,6 +8,7 @@ import (
 	plan "github.com/influxdata/flux/planner"
 	"github.com/influxdata/flux/semantic"
 	"github.com/influxdata/flux/values"
+	"github.com/pkg/errors"
 )
 
 const RangeKind = "range"
@@ -100,6 +101,18 @@ type RangeProcedureSpec struct {
 	StopCol  string
 }
 
+// TimeBounds implements plan.BoundsAwareProcedureSpec
+func (s *RangeProcedureSpec) TimeBounds(predecessorBounds *plan.Bounds) *plan.Bounds {
+	bounds := &plan.Bounds{
+		Start: values.ConvertTime(s.Bounds.Start.Time(s.Bounds.Now)),
+		Stop:  values.ConvertTime(s.Bounds.Stop.Time(s.Bounds.Now)),
+	}
+	if predecessorBounds != nil {
+		bounds = bounds.Intersect(predecessorBounds)
+	}
+	return bounds
+}
+
 func newRangeProcedure(qs flux.OperationSpec, pa plan.Administration) (plan.ProcedureSpec, error) {
 	spec, ok := qs.(*RangeOpSpec)
 
@@ -111,11 +124,21 @@ func newRangeProcedure(qs flux.OperationSpec, pa plan.Administration) (plan.Proc
 		spec.TimeCol = execute.DefaultTimeColLabel
 	}
 
+	bounds := flux.Bounds{
+		Start: spec.Start,
+		Stop:  spec.Stop,
+		Now:   pa.Now(),
+	}
+
+	if bounds.HasZero() {
+		return nil, errors.New(`cannot pass zero time to 'range'`)
+	}
+	if bounds.IsEmpty() {
+		return nil, errors.New("cannot query an empty range")
+	}
+
 	return &RangeProcedureSpec{
-		Bounds: flux.Bounds{
-			Start: spec.Start,
-			Stop:  spec.Stop,
-		},
+		Bounds:   bounds,
 		TimeCol:  spec.TimeCol,
 		StartCol: spec.StartCol,
 		StopCol:  spec.StopCol,
@@ -142,12 +165,9 @@ func createRangeTransformation(id execute.DatasetID, mode execute.AccumulationMo
 	cache := execute.NewTableBuilderCache(a.Allocator())
 	d := execute.NewDataset(id, mode, cache)
 
-	bounds := execute.Bounds{
-		Start: a.ResolveTime(s.Bounds.Start),
-		Stop:  a.ResolveTime(s.Bounds.Stop),
-	}
+	bounds := a.StreamContext().Bounds()
 
-	t, err := NewRangeTransformation(d, cache, s, bounds)
+	t, err := NewRangeTransformation(d, cache, s, *bounds)
 	if err != nil {
 		return nil, nil, err
 	}

functions/transformations/shift.go

@@ -2,6 +2,7 @@ package transformations
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/influxdata/flux"
 	"github.com/influxdata/flux/execute"
@@ -73,9 +74,18 @@ type ShiftProcedureSpec struct {
 	plan.DefaultCost
 	Shift   flux.Duration
 	Columns []string
+	Now     time.Time
 }
 
-func newShiftProcedure(qs flux.OperationSpec, _ plan.Administration) (plan.ProcedureSpec, error) {
+// TimeBounds implements plan.BoundsAwareProcedureSpec
+func (s *ShiftProcedureSpec) TimeBounds(predecessorBounds *plan.Bounds) *plan.Bounds {
+	if predecessorBounds != nil {
+		return predecessorBounds.Shift(values.Duration(s.Shift))
+	}
+	return nil
+}
+
+func newShiftProcedure(qs flux.OperationSpec, pa plan.Administration) (plan.ProcedureSpec, error) {
 	spec, ok := qs.(*ShiftOpSpec)
 	if !ok {
 		return nil, fmt.Errorf("invalid spec type %T", qs)
@@ -84,6 +94,7 @@ func newShiftProcedure(qs flux.OperationSpec, _ plan.Administration) (plan.Proce
 	return &ShiftProcedureSpec{
 		Shift:   spec.Shift,
 		Columns: spec.Columns,
+		Now:     pa.Now(),
 	}, nil
 }
 

planner/bounds.go

@@ -0,0 +1,110 @@
+package planner
+
+import "github.com/influxdata/flux/values"
+
+// EmptyBounds is a time range containing only a single point
+var EmptyBounds = &Bounds{
+	Start: values.Time(0),
+	Stop:  values.Time(0),
+}
+
+// Bounds is a range of time
+type Bounds struct {
+	Start values.Time
+	Stop  values.Time
+}
+
+// BoundsAwareProcedureSpec is any procedure
+// that modifies the time bounds of its data.
+type BoundsAwareProcedureSpec interface {
+	TimeBounds(predecessorBounds *Bounds) *Bounds
+}
+
+// ComputeBounds computes the time bounds for a
+// plan node from the bounds of its predecessors.
+func ComputeBounds(node PlanNode) error {
+	var bounds *Bounds
+
+	for _, pred := range node.Predecessors() {
+
+		if pred.Bounds() != nil && bounds == nil {
+			bounds = pred.Bounds()
+		}
+		if pred.Bounds() != nil && bounds != nil {
+			bounds = bounds.Union(pred.Bounds())
+		}
+	}
+
+	if s, ok := node.ProcedureSpec().(BoundsAwareProcedureSpec); ok {
+		bounds = s.TimeBounds(bounds)
+	}
+	node.SetBounds(bounds)
+	return nil
+}
+
+// IsEmpty reports whether the given bounds contain at most a single point
+func (b *Bounds) IsEmpty() bool {
+	return b.Start >= b.Stop
+}
+
+// Contains reports whether a given time is contained within the time range
+func (b *Bounds) Contains(t values.Time) bool {
+	return t >= b.Start && t < b.Stop
+}
+
+// Overlaps reports whether two given bounds have overlapping time ranges
+func (b *Bounds) Overlaps(o *Bounds) bool {
+	return b.Contains(o.Start) ||
+		(b.Contains(o.Stop) && o.Stop > b.Start) ||
+		o.Contains(b.Start)
+}
+
+// Union returns the smallest bounds which contain both input bounds.
+// It returns empty bounds if one of the input bounds are empty.
+func (b *Bounds) Union(o *Bounds) *Bounds {
+	if b.IsEmpty() || o.IsEmpty() {
+		return EmptyBounds
+	}
+	u := new(Bounds)
+
+	u.Start = b.Start
+	if o.Start < b.Start {
+		u.Start = o.Start
+	}
+
+	u.Stop = b.Stop
+	if o.Stop > b.Stop {
+		u.Stop = o.Stop
+	}
+
+	return u
+}
+
+// Intersect returns the intersection of two bounds.
+// It returns empty bounds if one of the input bounds are empty.
+func (b *Bounds) Intersect(o *Bounds) *Bounds {
+	if b.IsEmpty() || o.IsEmpty() || !b.Overlaps(o) {
+		return EmptyBounds
+	}
+	i := new(Bounds)
+
+	i.Start = b.Start
+	if o.Start > b.Start {
+		i.Start = o.Start
+	}
+
+	i.Stop = b.Stop
+	if o.Stop < b.Stop {
+		i.Stop = o.Stop
+	}
+
+	return i
+}
+
+// Shift moves the start and stop values of a time range by a specified duration
+func (b *Bounds) Shift(d values.Duration) *Bounds {
+	return &Bounds{
+		Start: b.Start.Add(d),
+		Stop:  b.Stop.Add(d),
+	}
+}

planner/bounds_test.go

@@ -2,6 +2,7 @@ package planner
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/influxdata/flux"
 )
@@ -67,7 +68,7 @@ func WithLogicalRule(rule Rule) LogicalOption {
 
 // Plan translates the given flux.Spec to a plan and transforms it by applying rules.
 func (l *logicalPlanner) Plan(spec *flux.Spec) (*PlanSpec, error) {
-	logicalPlan, err := createLogicalPlan(spec, l)
+	logicalPlan, err := createLogicalPlan(spec)
 	if err != nil {
 		return nil, err
 	}
@@ -80,14 +81,24 @@ func (l *logicalPlanner) Plan(spec *flux.Spec) (*PlanSpec, error) {
 	return newLogicalPlan, nil
 }
 
-func (logicalPlanner) ConvertID(oid flux.OperationID) ProcedureID {
+// TODO: This is unnecessary if we no longer need ProcedureIDs
+type administration struct {
+	now time.Time
+}
+
+func (a administration) Now() time.Time {
+	return a.now
+}
+
+func (a administration) ConvertID(oid flux.OperationID) ProcedureID {
 	return ProcedureIDFromOperationID(oid)
 }
 
 // LogicalPlanNode consists of the input and output edges and a procedure spec
 // that describes what the node does.
 type LogicalPlanNode struct {
 	edges
+	bounds
 	id   NodeID
 	Spec ProcedureSpec
 }
@@ -116,15 +127,16 @@ func (lpn *LogicalPlanNode) ShallowCopy() PlanNode {
 }
 
 // createLogicalPlan creates a logical query plan from a flux spec
-func createLogicalPlan(spec *flux.Spec, a Administration) (*PlanSpec, error) {
+func createLogicalPlan(spec *flux.Spec) (*PlanSpec, error) {
 	nodes := make(map[flux.OperationID]PlanNode, len(spec.Operations))
+	admin := administration{now: spec.Now}
 
 	plan := NewPlanSpec()
 	plan.Resources = spec.Resources
 	plan.Now = spec.Now
 
 	v := &fluxSpecVisitor{
-		a:     a,
+		a:     admin,
 		spec:  spec,
 		plan:  plan,
 		nodes: nodes,

planner/logical.go

@@ -14,13 +14,13 @@ import (
 	"github.com/influxdata/flux/semantic"
 )
 
-func compile(fluxText string) (*flux.Spec, error) {
-	now := time.Now().UTC()
+func compile(fluxText string, now time.Time) (*flux.Spec, error) {
 	return flux.Compile(context.Background(), fluxText, now)
 }
 
 // Test the translation of Flux query to logical plan
 func TestFluxSpecToLogicalPlan(t *testing.T) {
+	now := time.Now().UTC()
 	testcases := []struct {
 		// Name of the test
 		name string
@@ -48,6 +48,7 @@ func TestFluxSpecToLogicalPlan(t *testing.T) {
 							Stop: flux.Time{
 								IsRelative: true,
 							},
+							Now: now,
 						},
 						TimeCol:  "_time",
 						StartCol: "_start",
@@ -76,6 +77,7 @@ func TestFluxSpecToLogicalPlan(t *testing.T) {
 							Stop: flux.Time{
 								IsRelative: true,
 							},
+							Now: now,
 						},
 						TimeCol:  "_time",
 						StartCol: "_start",
@@ -105,7 +107,7 @@ func TestFluxSpecToLogicalPlan(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
-			spec, err := compile(tc.query)
+			spec, err := compile(tc.query, now)
 
 			if err != nil {
 				t.Fatal(err)
@@ -308,7 +310,7 @@ func TestLogicalPlanner(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
-			fluxSpec, err := compile(tc.flux)
+			fluxSpec, err := compile(tc.flux, time.Now().UTC())
 			if err != nil {
 				t.Fatalf("could not compile flux query: %v", err)
 			}

planner/logical_test.go

@@ -51,6 +51,11 @@ func (pp *physicalPlanner) Plan(spec *PlanSpec) (*PlanSpec, error) {
 		return nil, err
 	}
 
+	// Compute time bounds for nodes in the plan
+	if err := final.BottomUpWalk(ComputeBounds); err != nil {
+		return nil, err
+	}
+
 	// Update memory quota
 	if final.Resources.MemoryBytesQuota == 0 {
 		final.Resources.MemoryBytesQuota = pp.defaultMemoryLimit
@@ -132,6 +137,7 @@ type PhysicalProcedureSpec interface {
 // PhysicalPlanNode represents a physical operation in a plan.
 type PhysicalPlanNode struct {
 	edges
+	bounds
 	id   NodeID
 	Spec PhysicalProcedureSpec
 

planner/physical.go

@@ -2,6 +2,7 @@ package plantest
 
 import (
 	"fmt"
+
 	"github.com/google/go-cmp/cmp"
 	"github.com/influxdata/flux/planner"
 	"github.com/influxdata/flux/semantic/semantictest"
@@ -91,6 +92,12 @@ func cmpPlanNode(p, q planner.PlanNode) error {
 		return fmt.Errorf("wanted %s, but got %s", p.Kind(), q.Kind())
 	}
 
+	// Both nodes must have the same time bounds
+	if !cmp.Equal(p.Bounds(), q.Bounds()) {
+		return fmt.Errorf("plan nodes have different bounds -want(%s)/+got(%s) %s",
+			p.ID(), q.ID(), cmp.Diff(p.Bounds(), q.Bounds()))
+	}
+
 	// The specifications of both procedures must be the same
 	if !cmp.Equal(p.ProcedureSpec(), q.ProcedureSpec(), semantictest.CmpOptions...) {
 		return fmt.Errorf("procedure specs not equal -want(%s)/+got(%s) %s",

planner/plantest/cmp.go

@@ -2,6 +2,7 @@ package planner
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/influxdata/flux"
 	uuid "github.com/satori/go.uuid"
@@ -18,6 +19,7 @@ func (id ProcedureID) String() string {
 
 type Administration interface {
 	ConvertID(flux.OperationID) ProcedureID
+	Now() time.Time
 }
 
 func ProcedureIDFromOperationID(id flux.OperationID) ProcedureID {
@@ -28,6 +30,11 @@ func ProcedureIDFromNodeID(id NodeID) ProcedureID {
 	return ProcedureID(uuid.NewV5(RootUUID, string(id)))
 }
 
+// TODO: Is it necessary to pass in an Administration?
+// Currently Administration only converts IDs and provides
+// access to the now time. If it is determined that there
+// is no need for ProcedureIDs then we could probably just
+// pass in the now time directly.
 type CreateProcedureSpec func(flux.OperationSpec, Administration) (ProcedureSpec, error)
 
 var kindToProcedure = make(map[ProcedureKind]CreateProcedureSpec)

planner/registration.go

@@ -15,6 +15,9 @@ type PlanNode interface {
 	// Returns an identifier for this plan node
 	ID() NodeID
 
+	// Returns the time bounds for this plan node
+	Bounds() *Bounds
+
 	// Plan nodes executed immediately before this node
 	Predecessors() []PlanNode
 
@@ -29,6 +32,7 @@ type PlanNode interface {
 
 	// Helper methods for manipulating a plan
 	// These methods are used during planning
+	SetBounds(bounds *Bounds)
 	AddSuccessors(...PlanNode)
 	AddPredecessors(...PlanNode)
 	RemovePredecessor(PlanNode)
@@ -157,6 +161,18 @@ type ProcedureSpec interface {
 // ProcedureKind denotes the kind of operation
 type ProcedureKind string
 
+type bounds struct {
+	value *Bounds
+}
+
+func (b *bounds) SetBounds(bounds *Bounds) {
+	b.value = bounds
+}
+
+func (b *bounds) Bounds() *Bounds {
+	return b.value
+}
+
 type edges struct {
 	predecessors []PlanNode
 	successors   []PlanNode