Fix change point detection for uncertain non-stationary distributions. (#119578)

Author: jan-elastic

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/aggs/changepoint/ChangePointAggregator.java

@@ -134,7 +134,7 @@ public InternalAggregation doReduce(InternalAggregations aggregations, Aggregati
         }
 
         ChangePointBucket changePointBucket = null;
-        if (change.changePoint() >= 0) {
+        if (change.changePoint() != ChangeType.NO_CHANGE_POINT) {
             changePointBucket = extractBucket(bucketsPaths()[0], aggregations, change.changePoint()).map(
                 b -> new ChangePointBucket(b.getKey(), b.getDocCount(), b.getAggregations())
             ).orElse(null);
@@ -299,7 +299,7 @@ static TestStats testTrendVs(TestStats H0, double[] values, double[] weights) {
     static TestStats testStepChangeVs(TestStats H0, double[] values, double[] weights, int[] candidateChangePoints) {
 
         double vStep = Double.MAX_VALUE;
-        int changePoint = -1;
+        int changePoint = ChangeType.NO_CHANGE_POINT;
 
         // Initialize running stats so that they are only missing the individual changepoint values
         RunningStats lowerRange = new RunningStats();
@@ -336,7 +336,7 @@ static TestStats testStepChangeVs(TestStats H0, double[] values, double[] weight
     static TestStats testTrendChangeVs(TestStats H0, double[] values, double[] weights, int[] candidateChangePoints) {
 
         double vChange = Double.MAX_VALUE;
-        int changePoint = -1;
+        int changePoint = ChangeType.NO_CHANGE_POINT;
 
         // Initialize running stats so that they are only missing the individual changepoint values
         RunningStats lowerRange = new RunningStats();
@@ -451,7 +451,7 @@ static TestStats testDistributionChange(
     ) {
 
         double maxDiff = 0.0;
-        int changePoint = -1;
+        int changePoint = ChangeType.NO_CHANGE_POINT;
 
         // Initialize running stats so that they are only missing the individual changepoint values
         RunningStats lowerRange = new RunningStats();
@@ -484,6 +484,9 @@ static TestStats testDistributionChange(
 
         double pValue = 1;
         for (int cp : sampleData.changePoints()) {
+            if (cp == ChangeType.NO_CHANGE_POINT) {
+                continue;
+            }
             double[] x = Arrays.copyOfRange(sampleValues, 0, cp);
             double[] y = Arrays.copyOfRange(sampleValues, cp, sampleValues.length);
             double statistic = KOLMOGOROV_SMIRNOV_TEST.kolmogorovSmirnovStatistic(x, y);
@@ -520,7 +523,7 @@ record TestStats(Type type, double pValue, double var, double nParams, int chang
         }
 
         TestStats(Type type, double pValue, double var, double nParams, DataStats dataStats) {
-            this(type, pValue, var, nParams, -1, dataStats);
+            this(type, pValue, var, nParams, ChangeType.NO_CHANGE_POINT, dataStats);
         }
 
         boolean accept(double pValueThreshold) {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/aggs/changepoint/ChangeType.java

@@ -21,8 +21,10 @@
  */
 public interface ChangeType extends NamedWriteable, NamedXContentObject {
 
+    int NO_CHANGE_POINT = -1;
+
     default int changePoint() {
-        return -1;
+        return NO_CHANGE_POINT;
     }
 
     default double pValue() {

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/aggs/changepoint/ChangePointAggregatorTests.java

@@ -574,6 +574,14 @@ public void testSpikeSelectionVsChange() throws IOException {
         });
     }
 
+    public void testUncertainNonStationary() throws IOException {
+        double[] bucketValues = new double[] { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 700, 735, 715, 0 };
+        testChangeType(bucketValues, changeType -> {
+            assertThat(changeType, instanceOf(ChangeType.NonStationary.class));
+            assertThat(((ChangeType.NonStationary) changeType).getTrend(), equalTo("increasing"));
+        });
+    }
+
     void testChangeType(double[] bucketValues, Consumer<ChangeType> changeTypeAssertions) throws IOException {
         FilterAggregationBuilder dummy = AggregationBuilders.filter("dummy", new MatchAllQueryBuilder())
             .subAggregation(