Fix library UDF edge cases

Author: Caideyipi

library-udf/src/main/java/org/apache/iotdb/library/anomaly/UDTFIQR.java

@@ -90,9 +90,12 @@ public void beforeStart(UDFParameters parameters, UDTFConfigurations configurati
 
   @Override
   public void transform(Row row, PointCollector collector) throws Exception {
+    if (row.isNull(0)) {
+      return;
+    }
     if (compute.equalsIgnoreCase(STREAM_COMPUTE) && q3 > q1) {
       double v = Util.getValueAsDouble(row);
-      if (v < q1 - 1.5 * iqr || v > q3 + 1.5 * iqr) {
+      if (Double.isFinite(v) && (v < q1 - 1.5 * iqr || v > q3 + 1.5 * iqr)) {
         collector.putDouble(row.getTime(), v);
       }
     } else if (compute.equalsIgnoreCase(BATCH_COMPUTE)) {

library-udf/src/main/java/org/apache/iotdb/library/anomaly/UDTFKSigma.java

@@ -69,12 +69,19 @@ public void beforeStart(UDFParameters udfParameters, UDTFConfigurations udtfConf
     this.multipleK = udfParameters.getDoubleOrDefault("k", 3);
     this.dataType = udfParameters.getDataType(0);
     this.windowSize = udfParameters.getIntOrDefault("window", DEFAULT_WINDOW_SIZE);
+    this.mean = 0.0;
+    this.variance = 0.0;
+    this.sumX1 = 0.0;
+    this.sumX2 = 0.0;
     this.v = new CircularQueue<>(windowSize);
     this.t = new LongCircularQueue(windowSize);
   }
 
   @Override
   public void transform(Row row, PointCollector collector) throws Exception {
+    if (row.isNull(0)) {
+      return;
+    }
     double value = Util.getValueAsDouble(row);
     long timestamp = row.getTime();
     if (Double.isFinite(value) && !Double.isNaN(value)) {

library-udf/src/main/java/org/apache/iotdb/library/anomaly/UDTFMissDetect.java

@@ -55,7 +55,12 @@ public void beforeStart(UDFParameters udfp, UDTFConfigurations udtfc) throws Exc
 
   @Override
   public void transform(Row row, PointCollector collector) throws Exception {
-    detector.insert(row.getTime(), Util.getValueAsDouble(row));
+    if (!row.isNull(0)) {
+      double v = Util.getValueAsDouble(row);
+      if (Double.isFinite(v)) {
+        detector.insert(row.getTime(), v);
+      }
+    }
     while (detector.hasNext()) {
       collector.putBoolean(detector.getOutTime(), detector.getOutValue());
       detector.next();

library-udf/src/main/java/org/apache/iotdb/library/anomaly/UDTFOutlier.java

@@ -24,6 +24,7 @@
 import org.apache.iotdb.udf.api.access.Row;
 import org.apache.iotdb.udf.api.collector.PointCollector;
 import org.apache.iotdb.udf.api.customizer.config.UDTFConfigurations;
+import org.apache.iotdb.udf.api.customizer.parameter.UDFParameterValidator;
 import org.apache.iotdb.udf.api.customizer.parameter.UDFParameters;
 import org.apache.iotdb.udf.api.customizer.strategy.RowByRowAccessStrategy;
 import org.apache.iotdb.udf.api.type.Type;
@@ -47,38 +48,66 @@ public class UDTFOutlier implements UDTF {
   private ArrayList<Double> currentValueWindow = new ArrayList<>();
   private Map<Long, Double> outliers = new HashMap<>();
 
+  @Override
+  public void validate(UDFParameterValidator validator) throws Exception {
+    validator
+        .validateInputSeriesNumber(1)
+        .validateInputSeriesDataType(0, Type.INT32, Type.INT64, Type.FLOAT, Type.DOUBLE)
+        .validate(
+            x -> (int) x > 0,
+            "Parameter k should be a positive integer.",
+            validator.getParameters().getIntOrDefault("k", 3))
+        .validate(
+            x -> (double) x >= 0,
+            "Parameter r should be non-negative.",
+            validator.getParameters().getDoubleOrDefault("r", 5))
+        .validate(
+            x -> (int) x > 0,
+            "Parameter w should be a positive integer.",
+            validator.getParameters().getIntOrDefault("w", 1000))
+        .validate(
+            x -> (int) x > 0,
+            "Parameter s should be a positive integer.",
+            validator.getParameters().getIntOrDefault("s", 500));
+  }
+
   @Override
   public void beforeStart(UDFParameters udfParameters, UDTFConfigurations udtfConfigurations)
       throws Exception {
     udtfConfigurations
         .setAccessStrategy(new RowByRowAccessStrategy())
-        .setOutputDataType(udfParameters.getDataType(0));
+        .setOutputDataType(Type.DOUBLE);
     this.k = udfParameters.getIntOrDefault("k", 3);
     this.r = udfParameters.getDoubleOrDefault("r", 5);
     this.w = udfParameters.getIntOrDefault("w", 1000);
     this.s = udfParameters.getIntOrDefault("s", 500);
 
     this.i = 0;
-
-    udtfConfigurations.setAccessStrategy(new RowByRowAccessStrategy());
-    udtfConfigurations.setOutputDataType(Type.DOUBLE);
+    currentTimeWindow.clear();
+    currentValueWindow.clear();
+    outliers.clear();
   }
 
   @Override
   public void transform(Row row, PointCollector collector) throws Exception {
-    if (!row.isNull(0)) {
-      if (i >= w && (i - w) % s == 0) {
-        detect();
-      }
+    if (row.isNull(0)) {
+      return;
+    }
+    double v = Util.getValueAsDouble(row);
+    if (!Double.isFinite(v)) {
+      return;
+    }
+    if (i >= w && (i - w) % s == 0) {
+      detect();
+    }
 
-      if (i >= w) {
-        currentValueWindow.remove(0);
-        currentTimeWindow.remove(0);
-      }
-      currentTimeWindow.add(row.getTime());
-      currentValueWindow.add(Util.getValueAsDouble(row));
-      i += 1;
+    if (i >= w) {
+      currentValueWindow.remove(0);
+      currentTimeWindow.remove(0);
     }
+    currentTimeWindow.add(row.getTime());
+    currentValueWindow.add(v);
+    i += 1;
   }
 
   @Override

library-udf/src/main/java/org/apache/iotdb/library/anomaly/UDTFRange.java

@@ -63,6 +63,9 @@ public void beforeStart(UDFParameters parameters, UDTFConfigurations configurati
 
   @Override
   public void transform(Row row, PointCollector collector) throws Exception {
+    if (row.isNull(0)) {
+      return;
+    }
     int intValue;
     long longValue;
     float floatValue;

library-udf/src/main/java/org/apache/iotdb/library/anomaly/util/MissDetector.java

@@ -49,6 +49,9 @@ public class MissDetector {
   public MissDetector(RowIterator iterator, int minLength) throws Exception {
     while (iterator.hasNextRow()) {
       Row row = iterator.next();
+      if (row.isNull(0)) {
+        continue;
+      }
       double v = Util.getValueAsDouble(row);
       if (Double.isFinite(v)) {
         this.time.add(row.getTime());

library-udf/src/main/java/org/apache/iotdb/library/anomaly/util/StreamMissDetector.java

@@ -37,21 +37,23 @@ public class StreamMissDetector {
   private int minLength;
   private int state;
   private long startTime;
+  private boolean hasStartTime;
   private int missingStartIndex;
   private boolean horizon;
   private double standard;
 
   public StreamMissDetector(int minLength) {
     this.state = 0;
-    this.startTime = -1;
+    this.hasStartTime = false;
     this.minLength = minLength;
   }
 
   public void insert(long time, double value) {
     timeWindow.push(time);
     valueWindow.push(value);
-    if (startTime < 0) {
+    if (!hasStartTime) {
       startTime = time;
+      hasStartTime = true;
     }
     switch (state) {
       case 0:

library-udf/src/main/java/org/apache/iotdb/library/anomaly/util/WindowDetect.java

@@ -41,12 +41,12 @@ public WindowDetect(RowIterator dataIterator, double l, double t) throws Excepti
     ArrayList<Double> originList = new ArrayList<>();
     while (dataIterator.hasNextRow()) {
       Row row = dataIterator.next();
-      Double v = Util.getValueAsDouble(row);
       timeList.add(row.getTime());
-      if (v == null || !Double.isFinite(v)) {
+      if (row.isNull(0)) {
         originList.add(Double.NaN);
       } else {
-        originList.add(v);
+        double v = Util.getValueAsDouble(row);
+        originList.add(Double.isFinite(v) ? v : Double.NaN);
       }
     }
     time = Util.toLongArray(timeList);

library-udf/src/main/java/org/apache/iotdb/library/dprofile/UDAFIntegral.java

@@ -40,13 +40,15 @@ public class UDAFIntegral implements UDTF {
 
   long unitTime;
   long lastTime = -1;
+  boolean hasLastValue;
   double lastValue;
   double integralValue = 0;
 
   @Override
   public void validate(UDFParameterValidator validator) throws Exception {
     validator
         .validateInputSeriesNumber(1)
+        .validateInputSeriesDataType(0, Type.INT32, Type.INT64, Type.FLOAT, Type.DOUBLE)
         .validate(
             x -> (long) x > 0,
             "Unknown time unit input. Supported units are ns, us, ms, s, m, h, d.",
@@ -61,20 +63,28 @@ public void beforeStart(UDFParameters parameters, UDTFConfigurations configurati
     configurations.setAccessStrategy(new RowByRowAccessStrategy()).setOutputDataType(Type.DOUBLE);
     unitTime =
         Util.parseTime(parameters.getStringOrDefault(TIME_UNIT_KEY, TIME_UNIT_S), parameters);
+    lastTime = -1;
+    hasLastValue = false;
+    lastValue = 0;
+    integralValue = 0;
   }
 
   @Override
   public void transform(Row row, PointCollector collector) throws Exception {
+    if (row.isNull(0)) {
+      return;
+    }
     long nowTime = row.getTime();
     double nowValue = Util.getValueAsDouble(row);
     if (Double.isFinite(nowValue)) {
       // calculate the ladder-shaped area between last point and this one
       // skip and initialize the memory if no existing previous point is available
-      if (lastTime >= 0) {
+      if (hasLastValue) {
         integralValue += (lastValue + nowValue) * (nowTime - lastTime) / 2.0 / unitTime;
       }
       lastTime = nowTime;
       lastValue = nowValue;
+      hasLastValue = true;
     }
   }
 

library-udf/src/main/java/org/apache/iotdb/library/dprofile/UDAFIntegralAvg.java

@@ -37,6 +37,7 @@ public class UDAFIntegralAvg implements UDTF {
 
   long startTime = -1;
   long lastTime = -1;
+  boolean hasValue;
   double lastValue = 0;
   double integralValue = 0;
 
@@ -51,29 +52,36 @@ public void validate(UDFParameterValidator validator) throws Exception {
   public void beforeStart(UDFParameters parameters, UDTFConfigurations configurations)
       throws Exception {
     configurations.setAccessStrategy(new RowByRowAccessStrategy()).setOutputDataType(Type.DOUBLE);
+    startTime = -1;
+    lastTime = -1;
+    hasValue = false;
+    lastValue = 0;
+    integralValue = 0;
   }
 
   @Override
   public void transform(Row row, PointCollector collector) throws Exception {
+    if (row.isNull(0)) {
+      return;
+    }
     long nowTime = row.getTime();
     double nowValue = Util.getValueAsDouble(row);
-    if (startTime < 0) {
-      startTime = nowTime;
-    }
     if (Double.isFinite(nowValue)) {
-      // calculate the ladder-shaped area between last point and this one
-      // skip and initialize the memory if no existing previous point is available
-      if (lastTime >= 0) {
+      if (!hasValue) {
+        startTime = nowTime;
+      } else {
+        // calculate the ladder-shaped area between last point and this one
         integralValue += (lastValue + nowValue) * (nowTime - lastTime) / 2.0;
       }
       lastTime = nowTime;
       lastValue = nowValue;
+      hasValue = true;
     }
   }
 
   @Override
   public void terminate(PointCollector collector) throws Exception {
-    if (startTime < 0) {
+    if (!hasValue) {
       // empty input
       collector.putDouble(0, 0);
     } else if (startTime == lastTime) {