Increase timeseries function (#134934)

* Implementation of increase function for timeseries applications

* rebasing aigu

* fixup

* fixup

* Test repro

* cleanup and ready to go

* Fix formatting in X-RateGroupingAggregatorFunction.java.st

This was causing an infinite CI loop where the formatter and the code
generator were fighting.

* [CI] Update transport version definitions

* adding capability for tsfn increase

* comments and fixup

* csvtests

* fixup?

* fixup

* fixcsvtests

---------

Co-authored-by: Jordan Powers <[email protected]>
Co-authored-by: elasticsearchmachine <[email protected]>

Author: 3 people

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/aggregation/RateDoubleGroupingAggregatorFunction.java

@@ -36,6 +36,13 @@ import java.util.List;
 public final class Rate$Type$GroupingAggregatorFunction implements GroupingAggregatorFunction {
 
     public static final class FunctionSupplier implements AggregatorFunctionSupplier {
+        // Overriding constructor to support isRateOverTime flag
+        private final boolean isRateOverTime;
+
+        public FunctionSupplier(boolean isRateOverTime) {
+            this.isRateOverTime = isRateOverTime;
+        }
+
         @Override
         public List<IntermediateStateDesc> nonGroupingIntermediateStateDesc() {
             throw new UnsupportedOperationException("non-grouping aggregator is not supported");
@@ -53,7 +60,7 @@ public final class Rate$Type$GroupingAggregatorFunction implements GroupingAggre
 
         @Override
         public Rate$Type$GroupingAggregatorFunction groupingAggregator(DriverContext driverContext, List<Integer> channels) {
-            return new Rate$Type$GroupingAggregatorFunction(channels, driverContext);
+            return new Rate$Type$GroupingAggregatorFunction(channels, driverContext, isRateOverTime);
         }
 
         @Override
@@ -74,11 +81,13 @@ public final class Rate$Type$GroupingAggregatorFunction implements GroupingAggre
     private final DriverContext driverContext;
     private final BigArrays bigArrays;
     private ObjectArray<ReducedState> reducedStates;
+    private final boolean isRateOverTime;
 
-    public Rate$Type$GroupingAggregatorFunction(List<Integer> channels, DriverContext driverContext) {
+    public Rate$Type$GroupingAggregatorFunction(List<Integer> channels, DriverContext driverContext, boolean isRateOverTime) {
         this.channels = channels;
         this.driverContext = driverContext;
         this.bigArrays = driverContext.bigArrays();
+        this.isRateOverTime = isRateOverTime;
         ObjectArray<Buffer> buffers = driverContext.bigArrays().newObjectArray(256);
         try {
             this.reducedStates = driverContext.bigArrays().newObjectArray(256);
@@ -565,9 +574,9 @@ public final class Rate$Type$GroupingAggregatorFunction implements GroupingAggre
                 }
                 final double rate;
                 if (evalContext instanceof TimeSeriesGroupingAggregatorEvaluationContext tsContext) {
-                    rate = extrapolateRate(state, tsContext.rangeStartInMillis(group), tsContext.rangeEndInMillis(group));
+                    rate = extrapolateRate(state, tsContext.rangeStartInMillis(group), tsContext.rangeEndInMillis(group), isRateOverTime);
                 } else {
-                    rate = computeRateWithoutExtrapolate(state);
+                    rate = computeRateWithoutExtrapolate(state, isRateOverTime);
                 }
                 rates.appendDouble(rate);
             }
@@ -636,13 +645,17 @@ public final class Rate$Type$GroupingAggregatorFunction implements GroupingAggre
         }
     }
 
-    private static double computeRateWithoutExtrapolate(ReducedState state) {
+    private static double computeRateWithoutExtrapolate(ReducedState state, boolean isRateOverTime) {
         assert state.samples >= 2 : "rate requires at least two samples; got " + state.samples;
         final long firstTS = state.intervals[state.intervals.length - 1].t2;
         final long lastTS = state.intervals[0].t1;
         double firstValue = state.intervals[state.intervals.length - 1].v2;
         double lastValue = state.intervals[0].v1 + state.resets;
-        return (lastValue - firstValue) * 1000.0 / (lastTS - firstTS);
+        if (isRateOverTime) {
+            return (lastValue - firstValue) * 1000.0 / (lastTS - firstTS);
+        } else {
+            return lastValue - firstValue;
+        }
     }
 
     /**
@@ -654,7 +667,7 @@ public final class Rate$Type$GroupingAggregatorFunction implements GroupingAggre
      * We still extrapolate the rate in this case, but not all the way to the boundary, only by half of the average duration between
      * samples (which is our guess for where the series actually starts or ends).
      */
-    private static double extrapolateRate(ReducedState state, long rangeStart, long rangeEnd) {
+    private static double extrapolateRate(ReducedState state, long rangeStart, long rangeEnd, boolean isRateOverTime) {
         assert state.samples >= 2 : "rate requires at least two samples; got " + state.samples;
         final long firstTS = state.intervals[state.intervals.length - 1].t2;
         final long lastTS = state.intervals[0].t1;
@@ -677,6 +690,10 @@ public final class Rate$Type$GroupingAggregatorFunction implements GroupingAggre
             }
             lastValue = lastValue + endGap * slope;
         }
-        return (lastValue - firstValue) * 1000.0 / (rangeEnd - rangeStart);
+        if (isRateOverTime) {
+            return (lastValue - firstValue) * 1000.0 / (rangeEnd - rangeStart);
+        } else {
+            return lastValue - firstValue;
+        }
     }
 }