[ESQL] Add a BY clause to CHANGE_POINT command

docs/changelog/145210.yaml

@@ -0,0 +1,5 @@
+area: ES|QL
+issues: []
+pr: 145210
+summary: Add a BY clause to CHANGE_POINT command
+type: feature

docs/reference/query-languages/esql/_snippets/commands/examples/change_point.csv-spec/changePointForDocsByGroup.md

@@ -0,0 +1,16 @@
+% This is generated by ESQL's CommandDocsTests. Do not edit it. See ../README.md for how to regenerate it.
+
+```esql
+ROW k=[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25], g=[0,1,2]
+| MV_EXPAND k
+| MV_EXPAND g
+| EVAL value=CASE(k<13, 0, 42)
+| CHANGE_POINT value ON k BY g
+| WHERE type IS NOT NULL
+```
+
+| k:integer | g:integer | value:integer | type:keyword | pvalue:double |
+| --- | --- | --- | --- | --- |
+| 13 | 0 | 42 | step_change | 0.0 |
+| 13 | 1 | 42 | step_change | 0.0 |
+| 13 | 2 | 42 | step_change | 0.0 |

docs/reference/query-languages/esql/_snippets/commands/layout/change_point.md

@@ -11,9 +11,22 @@ The `CHANGE_POINT` command requires a [platinum license](https://www.elastic.co/
 
 ## Syntax
 
+::::{applies-switch}
+
+:::{applies-item} { stack: ga 9.2+, "serverless": "ga"}
 ```esql
 CHANGE_POINT value [ON key] [AS type_name, pvalue_name]
 ```
+:::
+
+:::{applies-item} { "stack": "preview 9.4", "serverless": "preview" }
+```esql
+CHANGE_POINT value [ON key] [BY group] [AS type_name, pvalue_name]
+```
+:::
+
+::::
+
 
 ## Parameters
 
@@ -23,6 +36,9 @@ CHANGE_POINT value [ON key] [AS type_name, pvalue_name]
 `key`
 :   The column with the key to order the values by. If not specified, `@timestamp` is used.
 
+`group` {applies_to}`stack: preview 9.4` {applies_to}`serverless: preview`
+:   The column to group values by. When specified, change point detection is performed independently for each group.
+
 `type_name`
 :   The name of the output column with the change point type. If not specified, `type` is used.
 
@@ -44,6 +60,8 @@ The possible change point types are:
 
 ::::{note}
 There must be at least 22 values for change point detection. Any values beyond the first 1,000 are ignored.
+
+When a `BY` clause is provided, these rules apply per group. {applies_to}`stack: preview 9.4` {applies_to}`serverless: preview`
 ::::
 
 ## Examples
@@ -52,3 +70,8 @@ The following example detects a step change in a metric:
 
 :::{include} ../examples/change_point.csv-spec/changePointForDocs.md
 :::
+
+The following example detects a step change independently for each group: {applies_to}`stack: preview 9.4` {applies_to}`serverless: preview`
+
+:::{include} ../examples/change_point.csv-spec/changePointForDocsByGroup.md
+:::

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/ChangePointOperator.java

@@ -8,6 +8,7 @@
 package org.elasticsearch.compute.operator;
 
 import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.BlockUtils;
@@ -25,6 +26,7 @@
 import java.util.ArrayList;
 import java.util.Deque;
 import java.util.List;
+import java.util.Objects;
 
 /**
  * Find spikes, dips and change point in a list of values.
@@ -37,29 +39,35 @@ public class ChangePointOperator extends CompleteInputCollectorOperator {
     private static final Logger logger = LogManager.getLogger(ChangePointOperator.class);
     public static final int INPUT_VALUE_COUNT_LIMIT = 1000;
 
-    public record Factory(int channel, WarningSourceLocation source) implements OperatorFactory {
+    private record DetectedChangePoint(int index, ChangeType type) {}
+
+    public record Factory(int channel, Integer groupingChannel, WarningSourceLocation source) implements OperatorFactory {
 
         @Override
         public Operator get(DriverContext driverContext) {
-            return new ChangePointOperator(driverContext, channel, source);
+            return new ChangePointOperator(driverContext, channel, groupingChannel, source);
         }
 
         @Override
         public String describe() {
-            return "ChangePointOperator[channel=" + channel + "]";
+            return groupingChannel == null
+                ? Strings.format("ChangePointOperator[channel=%d]", channel)
+                : Strings.format("ChangePointOperator[channel=%d, groupingChannel=%d]", channel, groupingChannel);
         }
     }
 
     private final DriverContext driverContext;
     private final int channel;
+    private final Integer groupChannel;
     private final WarningSourceLocation source;
 
     private final Deque<Page> outputPages;
     private Warnings warnings;
 
-    public ChangePointOperator(DriverContext driverContext, int channel, WarningSourceLocation source) {
+    public ChangePointOperator(DriverContext driverContext, int channel, Integer groupingChannel, WarningSourceLocation source) {
         this.driverContext = driverContext;
         this.channel = channel;
+        this.groupChannel = groupingChannel;
         this.source = source;
 
         outputPages = new ArrayDeque<>();
@@ -71,25 +79,67 @@ public boolean canProduceMoreDataWithoutExtraInput() {
         return outputPages.isEmpty() == false;
     }
 
-    private void createOutputPages() {
-        int valuesCount = 0;
-        for (Page page : inputPages) {
-            valuesCount += page.getPositionCount();
-        }
-        boolean tooManyValues = valuesCount > INPUT_VALUE_COUNT_LIMIT;
-        if (tooManyValues) {
-            valuesCount = INPUT_VALUE_COUNT_LIMIT;
-        }
+    private ChangeType detectChangePoint(List<Double> values, List<Integer> bucketIndexes) {
+        MlAggsHelper.DoubleBucketValues bucketValues = new MlAggsHelper.DoubleBucketValues(
+            null,
+            values.stream().mapToDouble(Double::doubleValue).toArray(),
+            bucketIndexes.stream().mapToInt(Integer::intValue).toArray()
+        );
+        ChangeType changeType = ChangePointDetector.getChangeType(bucketValues);
+        return changeType;
+    }
 
-        List<Double> values = new ArrayList<>(valuesCount);
-        List<Integer> bucketIndexes = new ArrayList<>(valuesCount);
+    private void createOutputPages() {
+        List<Double> values = new ArrayList<>();
+        List<Integer> bucketIndexes = new ArrayList<>();
+        ArrayDeque<DetectedChangePoint> detectedChangePoints = new ArrayDeque<>();
         int valuesIndex = 0;
+        int currentGroupRowCount = 0;
+        Object previousGroupKey = groupChannel != null && inputPages.isEmpty() == false
+            ? BlockUtils.toJavaObject(inputPages.peek().getBlock(groupChannel), 0)
+            : null;
+
         boolean hasNulls = false;
         boolean hasMultivalued = false;
+        boolean hasIndeterminableChangePoint = false;
+        boolean tooManyValues = false;
+        boolean lastGroupHasRows = false;
+        String indeterminableChangePointReason = "";
         for (Page inputPage : inputPages) {
             Block inputBlock = inputPage.getBlock(channel);
-            for (int i = 0; i < inputBlock.getPositionCount() && valuesIndex < valuesCount; i++) {
+            Block groupBlock = groupChannel != null ? inputPage.getBlock(groupChannel) : null;
+            for (int i = 0; i < inputBlock.getPositionCount(); i++) {
+                if (groupBlock != null) {
+                    Object currentGroupKey = BlockUtils.toJavaObject(groupBlock, i);
+                    if (Objects.equals(currentGroupKey, previousGroupKey) == false) {
+                        if (values.isEmpty() == false || lastGroupHasRows) {
+                            var changeType = detectChangePoint(values, bucketIndexes);
+                            var changePointIndex = changeType.changePoint();
+                            if (changePointIndex >= 0) {
+                                detectedChangePoints.offer(new DetectedChangePoint(changePointIndex, changeType));
+                            }
+                            if (changeType instanceof ChangeType.Indeterminable indeterminable) {
+                                hasIndeterminableChangePoint = true;
+                                indeterminableChangePointReason = indeterminable.getReason();
+                            }
+                            values.clear();
+                            bucketIndexes.clear();
+                            lastGroupHasRows = false;
+                        }
+                        previousGroupKey = currentGroupKey;
+                        currentGroupRowCount = 0;
+                    }
+                }
+
+                if (currentGroupRowCount >= INPUT_VALUE_COUNT_LIMIT) {
+                    tooManyValues = true;
+                    valuesIndex++;
+                    continue;
+                }
+
                 Object value = BlockUtils.toJavaObject(inputBlock, i);
+                lastGroupHasRows = true;
+                currentGroupRowCount++;
                 if (value == null) {
                     hasNulls = true;
                     valuesIndex++;
@@ -103,32 +153,47 @@ private void createOutputPages() {
             }
         }
 
-        MlAggsHelper.DoubleBucketValues bucketValues = new MlAggsHelper.DoubleBucketValues(
-            null,
-            values.stream().mapToDouble(Double::doubleValue).toArray(),
-            bucketIndexes.stream().mapToInt(Integer::intValue).toArray()
-        );
-        ChangeType changeType = ChangePointDetector.getChangeType(bucketValues);
-        int changePointIndex = changeType.changePoint();
+        // flush last (or only) group; for "non-grouped" or "all-null" input this still
+        // runs to produce an "indeterminable" warning.
+        if (values.isEmpty() == false || groupChannel == null || lastGroupHasRows) {
+            var changeType = detectChangePoint(values, bucketIndexes);
+            var changePointIndex = changeType.changePoint();
+            if (changePointIndex >= 0) {
+                detectedChangePoints.offer(new DetectedChangePoint(changePointIndex, changeType));
+            }
+            if (changeType instanceof ChangeType.Indeterminable indeterminable) {
+                hasIndeterminableChangePoint = true;
+                indeterminableChangePointReason = indeterminable.getReason();
+            }
+        }
 
+        buildOutputPages(detectedChangePoints);
+        emitWarnings(tooManyValues, hasNulls, hasMultivalued, hasIndeterminableChangePoint, indeterminableChangePointReason);
+    }
+
+    private void buildOutputPages(ArrayDeque<DetectedChangePoint> detectedChangePoints) {
         BlockFactory blockFactory = driverContext.blockFactory();
         int pageStartIndex = 0;
         while (inputPages.isEmpty() == false) {
             Page inputPage = inputPages.peek();
+            int pageEndIndex = pageStartIndex + inputPage.getPositionCount();
             Page outputPage;
             Block changeTypeBlock = null;
             Block changePvalueBlock = null;
             boolean success = false;
             try {
-                if (pageStartIndex <= changePointIndex && changePointIndex < pageStartIndex + inputPage.getPositionCount()) {
+                DetectedChangePoint head = detectedChangePoints.peek();
+                if (head != null && head.index() < pageEndIndex) {
                     try (
                         BytesRefBlock.Builder changeTypeBlockBuilder = blockFactory.newBytesRefBlockBuilder(inputPage.getPositionCount());
                         DoubleBlock.Builder pvalueBlockBuilder = blockFactory.newDoubleBlockBuilder(inputPage.getPositionCount())
                     ) {
                         for (int i = 0; i < inputPage.getPositionCount(); i++) {
-                            if (pageStartIndex + i == changePointIndex) {
-                                changeTypeBlockBuilder.appendBytesRef(new BytesRef(changeType.getWriteableName()));
-                                pvalueBlockBuilder.appendDouble(changeType.pValue());
+                            if (head != null && pageStartIndex + i == head.index()) {
+                                changeTypeBlockBuilder.appendBytesRef(new BytesRef(head.type().getWriteableName()));
+                                pvalueBlockBuilder.appendDouble(head.type().pValue());
+                                detectedChangePoints.poll();
+                                head = detectedChangePoints.peek();
                             } else {
                                 changeTypeBlockBuilder.appendNull();
                                 pvalueBlockBuilder.appendNull();
@@ -152,33 +217,33 @@ private void createOutputPages() {
 
             inputPages.removeFirst();
             outputPages.add(outputPage);
-            pageStartIndex += inputPage.getPositionCount();
+            pageStartIndex = pageEndIndex;
         }
+    }
 
-        if (changeType instanceof ChangeType.Indeterminable indeterminable) {
-            if (logger.isDebugEnabled()) {
-                logger.debug("Change point indeterminable: {}", indeterminable.getReason());
-            }
-            warnings(false).registerException(new IllegalArgumentException(indeterminable.getReason()));
-        }
+    private void emitWarnings(
+        boolean tooManyValues,
+        boolean hasNulls,
+        boolean hasMultivalued,
+        boolean hasIndeterminableChangePoint,
+        String indeterminableReason
+    ) {
         if (tooManyValues) {
-            if (logger.isDebugEnabled()) {
-                logger.debug("Too many values: limit is {}, some values were ignored", INPUT_VALUE_COUNT_LIMIT);
-            }
+            logger.debug(() -> Strings.format("Too many values: limit is %d, some values were ignored", INPUT_VALUE_COUNT_LIMIT));
             warnings(true).registerException(
                 new IllegalArgumentException("too many values; keeping only first " + INPUT_VALUE_COUNT_LIMIT + " values")
             );
         }
+        if (hasIndeterminableChangePoint) {
+            logger.debug(() -> Strings.format("Change point indeterminable: %s", indeterminableReason));
+            warnings(false).registerException(new IllegalArgumentException(indeterminableReason));
+        }
         if (hasNulls) {
-            if (logger.isDebugEnabled()) {
-                logger.debug("Values contain nulls; skipping them");
-            }
+            logger.debug(() -> "Values contain nulls; skipping them");
             warnings(true).registerException(new IllegalArgumentException("values contain nulls; skipping them"));
         }
         if (hasMultivalued) {
-            if (logger.isDebugEnabled()) {
-                logger.debug("Values contain multivalued entries; skipping them");
-            }
+            logger.debug(() -> "Values contain multivalued entries; skipping them");
             warnings(true).registerException(
                 new IllegalArgumentException(
                     "values contains multivalued entries; skipping them (please consider reducing them with e.g. MV_AVG or MV_SUM)"
@@ -209,7 +274,11 @@ protected void onClose() {
 
     @Override
     public String toString() {
-        return "ChangePointOperator[channel=" + channel + "]";
+        if (groupChannel == null) {
+            return "ChangePointOperator[channel=" + channel + "]";
+        } else {
+            return "ChangePointOperator[channel=" + channel + ", groupChannel=" + groupChannel + "]";
+        }
     }
 
     private Warnings warnings(boolean onlyWarnings) {