handle null values

Author: jan-elastic

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

@@ -48,6 +48,7 @@ public String describe() {
     private final List<Page> outputPages;
     private boolean finished;
     private int outputPageIndex;
+    private Warnings warnings;
 
     public ChangePointOperator(DriverContext driverContext, int inputChannel, String sourceText, int sourceLine, int sourceColumn) {
         this.driverContext = driverContext;
@@ -59,6 +60,7 @@ public ChangePointOperator(DriverContext driverContext, int inputChannel, String
         finished = false;
         inputPages = new ArrayList<>();
         outputPages = new ArrayList<>();
+        warnings = null;
     }
 
     @Override
@@ -105,23 +107,30 @@ private void createOutputPages() {
         // 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++) {
-                // TODO: nulls
-                values[valuesIndex++] = ((Number) BlockUtils.toJavaObject(inputBlock, i)).doubleValue();
+                Object value = BlockUtils.toJavaObject(inputBlock, i);
+                if (value == null) {
+                    hasNulls = true;
+                    values[valuesIndex++] = 0;
+                } else {
+                    values[valuesIndex++] = ((Number) value).doubleValue();
+                }
             }
         }
 
         ChangeType changeType = ChangePointDetector.getChangeType(new MlAggsHelper.DoubleBucketValues(null, values));
         int changePointIndex = changeType.changePoint();
 
         if (changeType instanceof ChangeType.Indeterminable indeterminable) {
-            Warnings.createWarnings(driverContext.warningsMode(), sourceLine, sourceColumn, sourceText)
-                .registerException(new IllegalArgumentException(indeterminable.getReason()));
+            warnings(false).registerException(new IllegalArgumentException(indeterminable.getReason()));
+        }
+        if (hasNulls) {
+            warnings(true).registerException(new IllegalArgumentException("values contain nulls; treating them as zeroes"));
         }
 
-        // TODO: throw error when indeterminable, due to not enough data
         BlockFactory blockFactory = driverContext.blockFactory();
         int pageStartIndex = 0;
         for (Page inputPage : inputPages) {
@@ -159,4 +168,15 @@ private void createOutputPages() {
 
     @Override
     public void close() {}
+
+    private Warnings warnings(boolean onlyWarnings) {
+        if (warnings == null) {
+            if (onlyWarnings) {
+                this.warnings = Warnings.createOnlyWarnings(driverContext.warningsMode(), sourceLine, sourceColumn, sourceText);
+            } else {
+                this.warnings = Warnings.createWarnings(driverContext.warningsMode(), sourceLine, sourceColumn, sourceText);
+            }
+        }
+        return warnings;
+    }
 }

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

@@ -32,7 +32,7 @@ public void registerException(Exception exception) {
      * @return A warnings collector object
      */
     public static Warnings createWarnings(DriverContext.WarningsMode warningsMode, int lineNumber, int columnNumber, String sourceText) {
-        return createWarnings(warningsMode, lineNumber, columnNumber, sourceText, "treating result as null");
+        return createWarnings(warningsMode, lineNumber, columnNumber, sourceText, "evaluation of [{}] failed, treating result as null");
     }
 
     /**
@@ -50,7 +50,25 @@ public static Warnings createWarningsTreatedAsFalse(
         int columnNumber,
         String sourceText
     ) {
-        return createWarnings(warningsMode, lineNumber, columnNumber, sourceText, "treating result as false");
+        return createWarnings(warningsMode, lineNumber, columnNumber, sourceText, "evaluation of [{}] failed, treating result as false");
+    }
+
+    /**
+     * Create a new warnings object based on the given mode which warns that
+     * evaluation resulted in warnings.
+     * @param warningsMode The warnings collection strategy to use
+     * @param lineNumber The line number of the source text. Same as `source.getLineNumber()`
+     * @param columnNumber The column number of the source text. Same as `source.getColumnNumber()`
+     * @param sourceText The source text that caused the warning. Same as `source.text()`
+     * @return A warnings collector object
+     */
+    public static Warnings createOnlyWarnings(
+        DriverContext.WarningsMode warningsMode,
+        int lineNumber,
+        int columnNumber,
+        String sourceText
+    ) {
+        return createWarnings(warningsMode, lineNumber, columnNumber, sourceText, "warnings during evaluation of [{}]");
     }
 
     private static Warnings createWarnings(
@@ -80,10 +98,9 @@ private Warnings(int lineNumber, int columnNumber, String sourceText, String fir
         this.location = format("Line {}:{}: ", lineNumber, columnNumber);
         this.first = format(
             null,
-            "{}evaluation of [{}] failed, {}. Only first {} failures recorded.",
+            "{}" + first + ". Only first {} failures recorded.",
             location,
             sourceText,
-            first,
             MAX_ADDED_WARNINGS
         );
     }

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

@@ -224,3 +224,44 @@ count:long | @timestamp:datetime      | type:text | pvalue:double
 93         | 2024-05-10T00:10:00.000Z | null      | null         
 23         | 2024-05-10T00:20:00.000Z | null      | null     
 ;
+
+
+null values
+required_capability: change_point
+
+FROM k8s
+  | STATS count=COUNT() BY @timestamp=BUCKET(@timestamp, 1 MINUTE)
+  | EVAL count=count+CASE(@timestamp>="2024-05-10T00:11:00.000Z", 100, 0)
+  | EVAL count=CASE(@timestamp=="2024-05-10T00:04:00.000Z", NULL, count)
+  | EVAL count=CASE(@timestamp=="2024-05-10T00:08:00.000Z", NULL, count)
+  | CHANGE_POINT count ON @timestamp AS type, pvalue
+;
+
+warning:Line 6:3: warnings during evaluation of [CHANGE_POINT count ON @timestamp AS type, pvalue]. Only first 20 failures recorded.
+warning:Line 6:3: java.lang.IllegalArgumentException: values contain nulls; treating them as zeroes
+
+@timestamp:datetime      | count:long | type:text   | pvalue:double
+2024-05-10T00:00:00.000Z | 4          | null        | null
+2024-05-10T00:01:00.000Z | 4          | null        | null
+2024-05-10T00:02:00.000Z | 8          | null        | null
+2024-05-10T00:03:00.000Z | 8          | null        | null
+2024-05-10T00:04:00.000Z | null       | null        | null
+2024-05-10T00:05:00.000Z | 8          | null        | null
+2024-05-10T00:06:00.000Z | 10         | null        | null
+2024-05-10T00:07:00.000Z | 5          | null        | null
+2024-05-10T00:08:00.000Z | null       | null        | null
+2024-05-10T00:09:00.000Z | 20         | null        | null
+2024-05-10T00:10:00.000Z | 5          | null        | null
+2024-05-10T00:11:00.000Z | 107        | step_change | 2.1947350086494926E-22
+2024-05-10T00:12:00.000Z | 108        | null        | null
+2024-05-10T00:13:00.000Z | 109        | null        | null
+2024-05-10T00:14:00.000Z | 109        | null        | null
+2024-05-10T00:15:00.000Z | 111        | null        | null
+2024-05-10T00:16:00.000Z | 107        | null        | null
+2024-05-10T00:17:00.000Z | 115        | null        | null
+2024-05-10T00:18:00.000Z | 117        | null        | null
+2024-05-10T00:19:00.000Z | 105        | null        | null
+2024-05-10T00:20:00.000Z | 110        | null        | null
+2024-05-10T00:21:00.000Z | 104        | null        | null
+2024-05-10T00:22:00.000Z | 109        | null        | null
+;