Implement rate/increase

Author: fionaliao

integration/getting_started_with_gossiped_ring_test.go

@@ -126,7 +126,7 @@ func runTestGettingStartedWithGossipedRing(t *testing.T, mimir1 *e2emimir.MimirS
 	require.NoError(t, mimir1.WaitSumMetrics(e2e.Equals(0+blocksLoadedOffset), "cortex_bucket_store_blocks_loaded"))
 	require.NoError(t, mimir2.WaitSumMetrics(e2e.Equals(0+blocksLoadedOffset), "cortex_bucket_store_blocks_loaded"))
 
-	// Finalize blocks from ingesters to the store.
+	// Flush blocks from ingesters to the store.
 	for _, instance := range []*e2emimir.MimirService{mimir1, mimir2} {
 		res, err := e2e.DoGet("http://" + instance.HTTPEndpoint() + "/ingester/flush")
 		require.NoError(t, err)

pkg/frontend/querymiddleware/shard_active_series.go

@@ -276,7 +276,7 @@ func (s *shardActiveSeriesMiddleware) writeMergedResponse(ctx context.Context, c
 		item.Reset(labels.EmptyLabels())
 		labelBuilderPool.Put(item)
 
-		// Finalize the stream buffer if it's getting too large.
+		// Flush the stream buffer if it's getting too large.
 		if stream.Buffered() > jsoniterMaxBufferSize {
 			_ = stream.Flush()
 		}
@@ -344,7 +344,7 @@ func (s *shardActiveSeriesMiddleware) writeMergedResponseWithZeroAllocationDecod
 		// Write the value as is, since it's already a JSON array.
 		stream.WriteRaw(rawStr)
 
-		// Finalize the stream buffer if it's getting too large.
+		// Flush the stream buffer if it's getting too large.
 		if stream.Buffered() > jsoniterMaxBufferSize {
 			_ = stream.Flush()
 		}

pkg/ingester/ingester.go

@@ -580,7 +580,7 @@ func (i *Ingester) generateHeadStatisticsForAllUsers(context.Context) error {
 }
 
 // NewForFlusher is a special version of ingester used by Flusher. This
-// ingester is not ingesting anything, its only purpose is to react on Finalize
+// ingester is not ingesting anything, its only purpose is to react on Flush
 // method and flush all openened TSDBs when called.
 func NewForFlusher(cfg Config, limits *validation.Overrides, registerer prometheus.Registerer, logger log.Logger) (*Ingester, error) {
 	i, err := newIngester(cfg, limits, registerer, logger)
@@ -593,7 +593,7 @@ func NewForFlusher(cfg Config, limits *validation.Overrides, registerer promethe
 	i.limiter = NewLimiter(limits, flusherLimiterStrategy{})
 
 	// This ingester will not start any subservices (lifecycler, compaction, shipping),
-	// and will only open TSDBs, wait for Finalize to be called, and then close TSDBs again.
+	// and will only open TSDBs, wait for Flush to be called, and then close TSDBs again.
 	i.BasicService = services.NewIdleService(i.startingForFlusher, i.stoppingForFlusher)
 	return i, nil
 }

pkg/ingester/ingester_test.go

@@ -5894,7 +5894,7 @@ func Benchmark_Ingester_MetricsForLabelMatchers(b *testing.B) {
 	i := createIngesterWithSeries(b, userID, numSeries, numSamplesPerSeries, startTimestamp, step)
 	ctx := user.InjectOrgID(context.Background(), "test")
 
-	// Finalize the ingester to ensure blocks have been compacted, so we'll test
+	// Flush the ingester to ensure blocks have been compacted, so we'll test
 	// fetching labels from blocks.
 	i.Flush()
 
@@ -8432,7 +8432,7 @@ func TestIngesterNoFlushWithInFlightRequest(t *testing.T) {
 	lockState, err := db.acquireAppendLock(0)
 	require.NoError(t, err)
 
-	// Finalize handler only triggers compactions, but doesn't wait for them to finish. We cannot use ?wait=true here,
+	// Flush handler only triggers compactions, but doesn't wait for them to finish. We cannot use ?wait=true here,
 	// because it would deadlock -- flush will wait for appendLock to be released.
 	i.FlushHandler(httptest.NewRecorder(), httptest.NewRequest("POST", "/ingester/flush", nil))
 

pkg/ingester/label_names_and_values.go

@@ -162,7 +162,7 @@ func labelValuesCardinality(
 			if respSize < msgSizeThreshold {
 				continue
 			}
-			// Finalize the response when reached message threshold.
+			// Flush the response when reached message threshold.
 			if err := client.SendLabelValuesCardinalityResponse(srv, &resp); err != nil {
 				return err
 			}

pkg/mimirpb/split.go

@@ -94,7 +94,7 @@ func SplitWriteRequestByMaxMarshalSizeRW2(req *WriteRequest, reqSize, maxSize in
 		// If the next partial request doesn't have any timeseries yet, we add the series anyway, in order to avoid an infinite loop
 		// if a single timeseries is bigger than the limit.
 		if nextReqSize+seriesSize+symbolsSize > maxSize && len(nextReq.TimeseriesRW2) > 0 {
-			// Finalize the next partial request.
+			// Flush the next partial request.
 			nextReq.SymbolsRW2 = nextReqSymbols.Symbols()
 			partialReqs = append(partialReqs, nextReq)
 
@@ -118,7 +118,7 @@ func SplitWriteRequestByMaxMarshalSizeRW2(req *WriteRequest, reqSize, maxSize in
 	}
 
 	if len(nextReq.TimeseriesRW2) > 0 {
-		// Finalize the last partial request.
+		// Flush the last partial request.
 		nextReq.SymbolsRW2 = nextReqSymbols.Symbols()
 		partialReqs = append(partialReqs, nextReq)
 	}
@@ -168,7 +168,7 @@ func splitTimeseriesByMaxMarshalSize(req *WriteRequest, reqSize, maxSize int) []
 		// If the next partial request doesn't have any timeseries yet, we add the series anyway, in order to avoid an infinite loop
 		// if a single timeseries is bigger than the limit.
 		if nextReqSize+seriesSize > maxSize && nextReqTimeseriesLength > 0 {
-			// Finalize the next partial request.
+			// Flush the next partial request.
 			nextReq.Timeseries = req.Timeseries[nextReqTimeseriesStart : nextReqTimeseriesStart+nextReqTimeseriesLength]
 			partialReqs = append(partialReqs, nextReq)
 
@@ -184,7 +184,7 @@ func splitTimeseriesByMaxMarshalSize(req *WriteRequest, reqSize, maxSize int) []
 	}
 
 	if nextReqTimeseriesLength > 0 {
-		// Finalize the last partial request.
+		// Flush the last partial request.
 		nextReq.Timeseries = req.Timeseries[nextReqTimeseriesStart : nextReqTimeseriesStart+nextReqTimeseriesLength]
 		partialReqs = append(partialReqs, nextReq)
 	}
@@ -234,7 +234,7 @@ func splitMetadataByMaxMarshalSize(req *WriteRequest, reqSize, maxSize int) []*W
 		// If the next partial request doesn't have any metadata yet, we add the metadata anyway, in order to avoid an infinite loop
 		// if a single metadata is bigger than the limit.
 		if nextReqSize+metadataSize > maxSize && nextReqMetadataLength > 0 {
-			// Finalize the next partial request.
+			// Flush the next partial request.
 			nextReq.Metadata = req.Metadata[nextReqMetadataStart : nextReqMetadataStart+nextReqMetadataLength]
 			partialReqs = append(partialReqs, nextReq)
 
@@ -250,7 +250,7 @@ func splitMetadataByMaxMarshalSize(req *WriteRequest, reqSize, maxSize int) []*W
 	}
 
 	if nextReqMetadataLength > 0 {
-		// Finalize the last partial request.
+		// Flush the last partial request.
 		nextReq.Metadata = req.Metadata[nextReqMetadataStart : nextReqMetadataStart+nextReqMetadataLength]
 		partialReqs = append(partialReqs, nextReq)
 	}

pkg/streamingpromql/evaluator.go

@@ -166,7 +166,7 @@ func (e *Evaluator) Evaluate(ctx context.Context, observer EvaluationObserver) (
 	}
 
 	if e.engine.pedantic {
-		// Finalize the root operator a second time to ensure all operators behave correctly if Finalize is called multiple times.
+		// Flush the root operator a second time to ensure all operators behave correctly if Finalize is called multiple times.
 		if err := e.root.Finalize(ctx); err != nil {
 			return fmt.Errorf("pedantic mode: failed to finalize operator a second time after successfully finalizing the first time: %w", err)
 		}

pkg/streamingpromql/operators/functions/factories.go

@@ -704,7 +704,7 @@ func init() {
 	must(RegisterFunction(FUNCTION_HISTOGRAM_SUM, "histogram_sum", parser.ValueTypeVector, InstantVectorTransformationFunctionOperatorFactory("histogram_sum", HistogramSum)))
 	must(RegisterFunction(FUNCTION_HOUR, "hour", parser.ValueTypeVector, TimeTransformationFunctionOperatorFactory("hour", Hour)))
 	must(RegisterFunction(FUNCTION_IDELTA, "idelta", parser.ValueTypeVector, FunctionOverRangeVectorOperatorFactory("idelta", Idelta)))
-	must(RegisterFunction(FUNCTION_INCREASE, "increase", parser.ValueTypeVector, FunctionOverRangeVectorOperatorFactory("increase", Increase)))
+	must(RegisterFunctionWithSplitFactory(FUNCTION_INCREASE, "increase", parser.ValueTypeVector, FunctionOverRangeVectorOperatorFactory("increase", Increase), SplitIncrease))
 	must(RegisterFunction(FUNCTION_IRATE, "irate", parser.ValueTypeVector, FunctionOverRangeVectorOperatorFactory("irate", Irate)))
 	must(RegisterFunction(FUNCTION_LABEL_JOIN, "label_join", parser.ValueTypeVector, LabelJoinFunctionOperatorFactory))
 	must(RegisterFunction(FUNCTION_LABEL_REPLACE, "label_replace", parser.ValueTypeVector, LabelReplaceFunctionOperatorFactory))
@@ -722,7 +722,7 @@ func init() {
 	must(RegisterFunction(FUNCTION_PRESENT_OVER_TIME, "present_over_time", parser.ValueTypeVector, FunctionOverRangeVectorOperatorFactory("present_over_time", PresentOverTime)))
 	must(RegisterFunction(FUNCTION_QUANTILE_OVER_TIME, "quantile_over_time", parser.ValueTypeVector, QuantileOverTimeFactory))
 	must(RegisterFunction(FUNCTION_RAD, "rad", parser.ValueTypeVector, InstantVectorTransformationFunctionOperatorFactory("rad", Rad)))
-	must(RegisterFunction(FUNCTION_RATE, "rate", parser.ValueTypeVector, FunctionOverRangeVectorOperatorFactory("rate", Rate)))
+	must(RegisterFunctionWithSplitFactory(FUNCTION_RATE, "rate", parser.ValueTypeVector, FunctionOverRangeVectorOperatorFactory("rate", Rate), SplitRate))
 	must(RegisterFunction(FUNCTION_RESETS, "resets", parser.ValueTypeVector, FunctionOverRangeVectorOperatorFactory("resets", Resets)))
 	must(RegisterFunction(FUNCTION_ROUND, "round", parser.ValueTypeVector, RoundFunctionOperatorFactory))
 	must(RegisterFunction(FUNCTION_SCALAR, "scalar", parser.ValueTypeScalar, instantVectorToScalarOperatorFactory))

pkg/streamingpromql/operators/functions/function_over_instant_vector.go

@@ -21,7 +21,7 @@ type FunctionOverInstantVector struct {
 	// as an argument. We can assume this will always be the Inner operator and therefore
 	// what we use for the SeriesMetadata.
 	Inner types.InstantVectorOperator
-	// Any scalar arguments will be read once and passed to FuncDef.SeriesDataFunc.
+	// Any scalar arguments will be read once and passed to Func.SeriesDataFunc.
 	ScalarArgs               []types.ScalarOperator
 	MemoryConsumptionTracker *limiter.MemoryConsumptionTracker
 	Func                     FunctionOverInstantVectorDefinition

pkg/streamingpromql/operators/functions/function_over_range_vector.go

@@ -132,11 +132,13 @@ func (m *FunctionOverRangeVector) NextSeries(ctx context.Context) (types.Instant
 
 	for {
 		step, err := m.Inner.NextStepSamples(ctx)
+
 		// nolint:errorlint // errors.Is introduces a performance overhead, and NextStepSamples is guaranteed to return exactly EOS, never a wrapped error.
 		if err == types.EOS {
 			if m.seriesValidationFunc != nil {
 				m.seriesValidationFunc(data, m.metricNames.GetMetricNameForSeries(m.currentSeriesIndex), m.emitAnnotationFunc)
 			}
+
 			return data, nil
 		} else if err != nil {
 			return types.InstantVectorSeriesData{}, err
@@ -146,7 +148,6 @@ func (m *FunctionOverRangeVector) NextSeries(ctx context.Context) (types.Instant
 		if err != nil {
 			return types.InstantVectorSeriesData{}, err
 		}
-
 		if hasFloat {
 			if data.Floats == nil {
 				// Only get FPoint slice once we are sure we have float points.