Implementation of increase function for timeseries applications

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/aggregation/X-RateGroupingAggregatorFunction.java.st

@@ -36,6 +36,12 @@ 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 +59,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 +80,12 @@ public final class Rate$Type$GroupingAggregatorFunction implements GroupingAggre
     private final DriverContext driverContext;
     private final BigArrays bigArrays;
     private ObjectArray<ReducedState> reducedStates;
-
-    public Rate$Type$GroupingAggregatorFunction(List<Integer> channels, DriverContext driverContext) {
+    private final boolean isRateOverTime;
+    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);
@@ -550,7 +557,7 @@ 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);
                 }
@@ -666,7 +673,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) {
         final int len = state.timestamps.length;
         assert len >= 2 : "rate requires at least two samples; got " + len;
         final long firstTS = state.timestamps[state.timestamps.length - 1];
@@ -696,7 +703,11 @@ 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;
+        }
     }
 
     // TODO: copied from old rate - simplify this or explain why we need it?