feat: add feature flag for fill() (#18445)

flags.yml

@@ -109,3 +109,10 @@
   default: false
   contact: Ariel Salem / Monitoring Team
   lifetime: temporary
+
+- name: Memory Optimized Fill
+  description: Enable the memory optimized fill()
+  key: memoryOptimizedFill
+  default: false
+  contact: Query Team
+  lifetime: temporary

kit/feature/doc.go

@@ -40,7 +40,7 @@
 //   contact: My Name
 // ```
 //
-// My flag type is inferred to be boolean by my defaulf of `false` when I run
+// My flag type is inferred to be boolean by my default of `false` when I run
 // `make flags` and the `feature` package now includes `func MyFeature() BoolFlag`.
 //
 // I use this to control my backend code with
@@ -69,7 +69,7 @@
 // ```
 //
 // ```
-// influxd --feature-flags flag1:value1,flag2:value2
+// influxd --feature-flags flag1=value1,flag2=value2
 // ```
 //
 package feature

query/stdlib/influxdata/influxdb/rules.go

@@ -30,6 +30,7 @@ func init() {
 		PushDownWindowAggregateByTimeRule{},
 		PushDownBareAggregateRule{},
 		PushDownGroupAggregateRule{},
+		SwitchFillImplRule{},
 	)
 }
 
@@ -1010,3 +1011,24 @@ func canPushGroupedAggregate(ctx context.Context, pn plan.Node) bool {
 	}
 	return false
 }
+
+type SwitchFillImplRule struct{}
+
+func (SwitchFillImplRule) Name() string {
+	return "SwitchFillImplRule"
+}
+
+func (SwitchFillImplRule) Pattern() plan.Pattern {
+	return plan.Pat(universe.FillKind, plan.Any())
+}
+
+func (r SwitchFillImplRule) Rewrite(ctx context.Context, pn plan.Node) (plan.Node, bool, error) {
+	if !feature.MemoryOptimizedFill().Enabled(ctx) {
+		spec := pn.ProcedureSpec().Copy()
+		universe.UseDeprecatedImpl(spec)
+		if err := pn.ReplaceSpec(spec); err != nil {
+			return nil, false, err
+		}
+	}
+	return pn, false, nil
+}

query/stdlib/influxdata/influxdb/rules_test.go

@@ -1882,6 +1882,80 @@ func TestPushDownWindowAggregateRule(t *testing.T) {
 	}
 }
 
+func TestSwitchFillImplRule(t *testing.T) {
+	flagger := mock.NewFlagger(map[feature.Flag]interface{}{
+		feature.MemoryOptimizedFill(): true,
+	})
+	withFlagger, _ := feature.Annotate(context.Background(), flagger)
+	readRange := &influxdb.ReadRangePhysSpec{
+		Bucket: "my-bucket",
+		Bounds: flux.Bounds{
+			Start: fluxTime(5),
+			Stop:  fluxTime(10),
+		},
+	}
+	sourceSpec := &universe.DualImplProcedureSpec{
+		ProcedureSpec: &universe.FillProcedureSpec{
+			DefaultCost: plan.DefaultCost{},
+			Column:      "_value",
+			Value:       values.NewFloat(0),
+			UsePrevious: false,
+		},
+		UseDeprecated: false,
+	}
+	targetSpec := sourceSpec.Copy().(*universe.DualImplProcedureSpec)
+	universe.UseDeprecatedImpl(targetSpec)
+
+	testcases := []plantest.RuleTestCase{
+		{
+			Context: withFlagger,
+			Name:    "enable memory optimized fill",
+			Rules:   []plan.Rule{influxdb.SwitchFillImplRule{}},
+			Before: &plantest.PlanSpec{
+				Nodes: []plan.Node{
+					plan.CreatePhysicalNode("ReadRange", readRange),
+					plan.CreatePhysicalNode("fill", sourceSpec),
+				},
+				Edges: [][2]int{
+					{0, 1},
+				},
+			},
+			NoChange: true,
+		},
+		{
+			Context: context.Background(),
+			Name:    "disable memory optimized fill",
+			Rules:   []plan.Rule{influxdb.SwitchFillImplRule{}},
+			Before: &plantest.PlanSpec{
+				Nodes: []plan.Node{
+					plan.CreatePhysicalNode("ReadRange", readRange),
+					plan.CreatePhysicalNode("fill", sourceSpec),
+				},
+				Edges: [][2]int{
+					{0, 1},
+				},
+			},
+			After: &plantest.PlanSpec{
+				Nodes: []plan.Node{
+					plan.CreatePhysicalNode("ReadRange", readRange),
+					plan.CreatePhysicalNode("fill", targetSpec),
+				},
+				Edges: [][2]int{
+					{0, 1},
+				},
+			},
+		},
+	}
+
+	for _, tc := range testcases {
+		tc := tc
+		t.Run(tc.Name, func(t *testing.T) {
+			t.Parallel()
+			plantest.PhysicalRuleTestHelper(t, &tc)
+		})
+	}
+}
+
 func TestPushDownBareAggregateRule(t *testing.T) {
 	// Turn on support for window aggregate count
 	flagger := mock.NewFlagger(map[feature.Flag]interface{}{