too much data

Author: jan-elastic

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

@@ -19,13 +19,14 @@
 import org.elasticsearch.xpack.ml.aggs.changepoint.ChangeType;
 
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
 
 public class ChangePointOperator implements Operator {
 
     // TODO: close upon failure / interrupt
 
+    public static final int INPUT_VALUE_COUNT_LIMIT = 1000;
+
     public record Factory(int inputChannel, String sourceText, int sourceLine, int sourceColumn) implements OperatorFactory {
         @Override
         public Operator get(DriverContext driverContext) {
@@ -103,14 +104,18 @@ private void createOutputPages() {
         for (Page page : inputPages) {
             valuesCount += page.getPositionCount();
         }
+        boolean tooManyValues = valuesCount > INPUT_VALUE_COUNT_LIMIT;
+        if (tooManyValues) {
+            valuesCount = INPUT_VALUE_COUNT_LIMIT;
+        }
 
         // TODO: account for this memory?
         double[] values = new double[valuesCount];
         int valuesIndex = 0;
         boolean hasNulls = false;
         for (Page inputPage : inputPages) {
             Block inputBlock = inputPage.getBlock(inputChannel);
-            for (int i = 0; i < inputBlock.getPositionCount(); i++) {
+            for (int i = 0; i < inputBlock.getPositionCount() && valuesIndex < valuesCount; i++) {
                 Object value = BlockUtils.toJavaObject(inputBlock, i);
                 if (value == null) {
                     hasNulls = true;
@@ -124,13 +129,6 @@ private void createOutputPages() {
         ChangeType changeType = ChangePointDetector.getChangeType(new MlAggsHelper.DoubleBucketValues(null, values));
         int changePointIndex = changeType.changePoint();
 
-        if (changeType instanceof ChangeType.Indeterminable indeterminable) {
-            warnings(false).registerException(new IllegalArgumentException(indeterminable.getReason()));
-        }
-        if (hasNulls) {
-            warnings(true).registerException(new IllegalArgumentException("values contain nulls; treating them as zeroes"));
-        }
-
         BlockFactory blockFactory = driverContext.blockFactory();
         int pageStartIndex = 0;
         for (Page inputPage : inputPages) {
@@ -164,6 +162,16 @@ private void createOutputPages() {
         }
 
         inputPages.clear();
+
+        if (changeType instanceof ChangeType.Indeterminable indeterminable) {
+            warnings(false).registerException(new IllegalArgumentException(indeterminable.getReason()));
+        }
+        if (tooManyValues) {
+            warnings(true).registerException(new IllegalArgumentException("too many values; keeping only first " + INPUT_VALUE_COUNT_LIMIT + " values"));
+        }
+        if (hasNulls) {
+            warnings(true).registerException(new IllegalArgumentException("values contain nulls; treating them as zeroes"));
+        }
     }
 
     @Override

x-pack/plugin/esql/qa/testFixtures/src/main/resources/change_point.csv-spec

@@ -208,24 +208,6 @@ null            | null
 ;
 
 
-not enough data
-required_capability: change_point
-
-FROM k8s
-  | STATS count=COUNT() BY @timestamp=BUCKET(@timestamp, 10 MINUTES)
-  | CHANGE_POINT count ON @timestamp AS type, pvalue
-;
-
-warning:Line 3:3: evaluation of [CHANGE_POINT count ON @timestamp AS type, pvalue] failed, treating result as null. Only first 20 failures recorded.
-warning:Line 3:3: java.lang.IllegalArgumentException: not enough buckets to calculate change_point. Requires at least [22]; found [3]
-
-count:long | @timestamp:datetime      | type:text | pvalue:double
-84         | 2024-05-10T00:00:00.000Z | null      | null         
-93         | 2024-05-10T00:10:00.000Z | null      | null         
-23         | 2024-05-10T00:20:00.000Z | null      | null     
-;
-
-
 null values
 required_capability: change_point