changes from review and test cases

Author: jlapacik

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,12 +101,16 @@ type RangeProcedureSpec struct {
 	StopCol  string
 }
 
-// TimeBounds implements plan.BoundedProcedureSpec
-func (s *RangeProcedureSpec) TimeBounds() *plan.Bounds {
-	return &plan.Bounds{
+// 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) {
@@ -119,12 +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,
-			Now:   pa.Now(),
-		},
+		Bounds:   bounds,
 		TimeCol:  spec.TimeCol,
 		StartCol: spec.StartCol,
 		StopCol:  spec.StopCol,

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

@@ -14,10 +14,10 @@ type Bounds struct {
 	Stop  values.Time
 }
 
-// BoundedProcedureSpec represents any procedure that
-// introduces time bounds for the data it produces.
-type BoundedProcedureSpec interface {
-	TimeBounds() *Bounds
+// 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
@@ -35,15 +35,10 @@ func ComputeBounds(node PlanNode) error {
 		}
 	}
 
-	s, bdd := node.ProcedureSpec().(BoundedProcedureSpec)
-
-	if bdd && bounds != nil {
-		bounds = bounds.Intersect(s.TimeBounds())
-	}
-	if bdd && bounds == nil {
-		bounds = s.TimeBounds()
+	if s, ok := node.ProcedureSpec().(BoundsAwareProcedureSpec); ok {
+		bounds = s.TimeBounds(bounds)
 	}
-	node.AddBounds(bounds)
+	node.SetBounds(bounds)
 	return nil
 }
 
@@ -105,3 +100,11 @@ func (b *Bounds) Intersect(o *Bounds) *Bounds {
 
 	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

@@ -7,6 +7,7 @@ import (
 	"github.com/google/go-cmp/cmp"
 
 	"github.com/influxdata/flux/planner"
+	"github.com/influxdata/flux/planner/plantest"
 	"github.com/influxdata/flux/values"
 )
 
@@ -246,3 +247,263 @@ func TestBounds_IsEmpty(t *testing.T) {
 		})
 	}
 }
+
+// A BoundsAwareProcedureSpec that intersects its bounds with its predecessors' bounds
+type mockBoundsIntersectProcedureSpec struct {
+	planner.DefaultCost
+	bounds *planner.Bounds
+}
+
+func (m *mockBoundsIntersectProcedureSpec) Kind() planner.ProcedureKind {
+	return "mock-intersect-bounds"
+}
+
+func (m *mockBoundsIntersectProcedureSpec) Copy() planner.ProcedureSpec {
+	return &mockBoundsIntersectProcedureSpec{}
+}
+
+func (m *mockBoundsIntersectProcedureSpec) TimeBounds(predecessorBounds *planner.Bounds) *planner.Bounds {
+	if predecessorBounds != nil {
+		return predecessorBounds.Intersect(m.bounds)
+	}
+	return m.bounds
+}
+
+// A BoundsAwareProcedureSpec that shifts its predecessors' bounds
+type mockBoundsShiftProcedureSpec struct {
+	planner.DefaultCost
+	by values.Duration
+}
+
+func (m *mockBoundsShiftProcedureSpec) Kind() planner.ProcedureKind {
+	return "mock-shift-bounds"
+}
+
+func (m *mockBoundsShiftProcedureSpec) Copy() planner.ProcedureSpec {
+	return &mockBoundsShiftProcedureSpec{}
+}
+
+func (m *mockBoundsShiftProcedureSpec) TimeBounds(predecessorBounds *planner.Bounds) *planner.Bounds {
+	if predecessorBounds != nil {
+		return predecessorBounds.Shift(m.by)
+	}
+	return nil
+}
+
+// Create a PlanNode with id and mockBoundsIntersectProcedureSpec
+func makeBoundsNode(id string, bounds *planner.Bounds) planner.PlanNode {
+	return planner.CreatePhysicalNode(planner.NodeID(id),
+		&mockBoundsIntersectProcedureSpec{
+			bounds: bounds,
+		})
+}
+
+// Create a PlanNode with id and mockBoundsShiftProcedureSpec
+func makeShiftNode(id string, duration values.Duration) planner.PlanNode {
+	return planner.CreateLogicalNode(planner.NodeID(id),
+		&mockBoundsShiftProcedureSpec{
+			by: duration,
+		})
+}
+
+func bounds(start, stop int) *planner.Bounds {
+	return &planner.Bounds{
+		Start: values.Time(start),
+		Stop:  values.Time(stop),
+	}
+}
+
+// Test that bounds are propagated up through the plan correctly
+func TestBounds_ComputePlanBounds(t *testing.T) {
+	tests := []struct {
+		// Name of test
+		name string
+		// Nodes and edges defining plan
+		spec *plantest.PhysicalPlanSpec
+		// Map from node ID to the expected bounds for that node
+		want map[planner.NodeID]*planner.Bounds
+	}{
+		{
+			name: "no bounds",
+			spec: &plantest.PhysicalPlanSpec{
+				Nodes: []planner.PlanNode{
+					makeNode("0"),
+				},
+			},
+			want: map[planner.NodeID]*planner.Bounds{
+				"0": nil,
+			},
+		},
+		{
+			name: "single time bounds",
+			// 0 -> 1 -> 2 -> 3 -> 4
+			spec: &plantest.PhysicalPlanSpec{
+				Nodes: []planner.PlanNode{
+					makeNode("0"),
+					makeNode("1"),
+					makeBoundsNode("2", bounds(5, 10)),
+					makeNode("3"),
+					makeNode("4"),
+				},
+				Edges: [][2]int{
+					{0, 1},
+					{1, 2},
+					{2, 3},
+					{3, 4},
+				},
+			},
+			want: map[planner.NodeID]*planner.Bounds{
+				"0": nil,
+				"1": nil,
+				"2": bounds(5, 10),
+				"3": bounds(5, 10),
+				"4": bounds(5, 10)},
+		},
+		{
+			name: "multiple intersect time bounds",
+			// 0 -> 1 -> 2 -> 3 -> 4
+			spec: &plantest.PhysicalPlanSpec{
+				Nodes: []planner.PlanNode{
+					makeNode("0"),
+					makeBoundsNode("1", bounds(5, 10)),
+					makeNode("2"),
+					makeBoundsNode("3", bounds(7, 11)),
+					makeNode("4"),
+				},
+				Edges: [][2]int{
+					{0, 1},
+					{1, 2},
+					{2, 3},
+					{3, 4},
+				},
+			},
+			want: map[planner.NodeID]*planner.Bounds{
+				"0": nil,
+				"1": bounds(5, 10),
+				"2": bounds(5, 10),
+				"3": bounds(7, 10),
+				"4": bounds(7, 10)},
+		},
+		{
+			name: "shift nil time bounds",
+			// 0 -> 1 -> 2
+			spec: &plantest.PhysicalPlanSpec{
+				Nodes: []planner.PlanNode{
+					makeNode("0"),
+					makeShiftNode("1", values.Duration(5)),
+					makeNode("2"),
+				},
+				Edges: [][2]int{
+					{0, 1},
+					{1, 2},
+				},
+			},
+			want: map[planner.NodeID]*planner.Bounds{
+				"0": nil,
+				"1": nil,
+				"2": nil,
+			},
+		},
+		{
+			name: "shift bounds after intersecting bounds",
+			// 0 -> 1 -> 2 -> 3 -> 4
+			spec: &plantest.PhysicalPlanSpec{
+				Nodes: []planner.PlanNode{
+					makeNode("0"),
+					makeBoundsNode("1", bounds(5, 10)),
+					makeNode("2"),
+					makeShiftNode("3", values.Duration(5)),
+					makeNode("4"),
+				},
+				Edges: [][2]int{
+					{0, 1},
+					{1, 2},
+					{2, 3},
+					{3, 4},
+				},
+			},
+			want: map[planner.NodeID]*planner.Bounds{
+				"0": nil,
+				"1": bounds(5, 10),
+				"2": bounds(5, 10),
+				"3": bounds(10, 15),
+				"4": bounds(10, 15)},
+		},
+		{
+			name: "join",
+			//   2
+			//  / \
+			// 0   1
+			spec: &plantest.PhysicalPlanSpec{
+				Nodes: []planner.PlanNode{
+					makeBoundsNode("0", bounds(5, 10)),
+					makeBoundsNode("1", bounds(12, 20)),
+					makeNode("2"),
+				},
+				Edges: [][2]int{
+					{0, 2},
+					{1, 2},
+				},
+			},
+			want: map[planner.NodeID]*planner.Bounds{
+				"0": bounds(5, 10),
+				"1": bounds(12, 20),
+				"2": bounds(5, 20),
+			},
+		},
+		{
+			name: "yields",
+			// 3   4
+			//  \ /
+			//   1   2
+			//    \ /
+			//     0
+			spec: &plantest.PhysicalPlanSpec{
+				Nodes: []planner.PlanNode{
+					makeNode("0"),
+					makeBoundsNode("1", bounds(5, 10)),
+					makeNode("2"),
+					makeNode("3"),
+					makeNode("4"),
+				},
+				Edges: [][2]int{
+					{0, 1},
+					{0, 2},
+					{1, 3},
+					{1, 4},
+				},
+			},
+			want: map[planner.NodeID]*planner.Bounds{
+				"0": nil,
+				"1": bounds(5, 10),
+				"2": nil,
+				"3": bounds(5, 10),
+				"4": bounds(5, 10),
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			// Create plan from spec
+			thePlan := plantest.CreatePhysicalPlanSpec(tc.spec)
+
+			// Method used to compute the bounds at each node
+			if err := thePlan.BottomUpWalk(planner.ComputeBounds); err != nil {
+				t.Fatal(err)
+			}
+
+			// Map NodeID -> Bounds
+			got := make(map[planner.NodeID]*planner.Bounds)
+			thePlan.BottomUpWalk(func(n planner.PlanNode) error {
+				got[n.ID()] = n.Bounds()
+				return nil
+			})
+
+			if !cmp.Equal(tc.want, got) {
+				t.Errorf("Did not get expected time bounds, -want/+got:\n%v", cmp.Diff(tc.want, got))
+			}
+		})
+	}
+}

planner/logical.go

@@ -146,10 +146,6 @@ func createLogicalPlan(spec *flux.Spec) (*PlanSpec, error) {
 		return nil, err
 	}
 
-	if err := v.plan.BottomUpWalk(ComputeBounds); err != nil {
-		return nil, err
-	}
-
 	return v.plan, nil
 }
 

planner/physical.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

planner/plantest/cmp.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/types.go

@@ -32,7 +32,7 @@ type PlanNode interface {
 
 	// Helper methods for manipulating a plan
 	// These methods are used during planning
-	AddBounds(bounds *Bounds)
+	SetBounds(bounds *Bounds)
 	AddSuccessors(...PlanNode)
 	AddPredecessors(...PlanNode)
 	RemovePredecessor(PlanNode)
@@ -165,7 +165,7 @@ type bounds struct {
 	value *Bounds
 }
 
-func (b *bounds) AddBounds(bounds *Bounds) {
+func (b *bounds) SetBounds(bounds *Bounds) {
 	b.value = bounds
 }