Update 3 UDFs: Percentile, Quantile and Cluster

integration-test/src/test/java/org/apache/iotdb/libudf/it/dlearn/DLearnIT.java

@@ -94,6 +94,11 @@ private static void createTimeSeries() {
               + "datatype=double, "
               + "encoding=plain, "
               + "compression=uncompressed");
+      statement.addBatch(
+          "create timeseries root.vehicle.d3.s1 with "
+              + "datatype=int32, "
+              + "encoding=plain, "
+              + "compression=uncompressed");
       statement.executeBatch();
     } catch (SQLException throwable) {
       fail(throwable.getMessage());
@@ -168,6 +173,16 @@ private static void generateData() {
           String.format(
               "insert into root.vehicle.d2(timestamp,s1,s2,s3,s4) values(%d,%d,%d,%d,%d)",
               900, 4, 4, 4, 4));
+      // Toy series for cluster UDF (l=3, k=2): windows [1,2,3], [10,20,30], [1,5,1]. With
+      // norm=false,
+      // k-means groups the first two windows; k-shape / medoidshape group windows 0 and 2
+      // (shape-related).
+      int[] toy = {1, 2, 3, 10, 20, 30, 1, 5, 1};
+      for (int i = 0; i < toy.length; i++) {
+        statement.addBatch(
+            String.format(
+                "insert into root.vehicle.d3(timestamp,s1) values(%d,%d)", (i + 1) * 100, toy[i]));
+      }
       statement.executeBatch();
     } catch (SQLException throwable) {
       fail(throwable.getMessage());
@@ -179,6 +194,7 @@ private static void registerUDF() {
         Statement statement = connection.createStatement()) {
       statement.execute("create function iqr as 'org.apache.iotdb.library.anomaly.UDTFIQR'");
       statement.execute("create function ar as 'org.apache.iotdb.library.dlearn.UDTFAR'");
+      statement.execute("create function cluster as 'org.apache.iotdb.library.dlearn.UDTFCluster'");
     } catch (SQLException throwable) {
       fail(throwable.getMessage());
     }
@@ -308,4 +324,70 @@ public void testAR4() {
       fail(throwable.getMessage());
     }
   }
+
+  @Test
+  public void testCluster1() {
+    String sqlStr =
+        "select cluster(d3.s1, 'l'='3', 'k'='2', 'method'='kmeans', 'norm'='false', "
+            + "'maxiter'='50', 'output'='label') from root.vehicle";
+    try (Connection connection = EnvFactory.getEnv().getConnection();
+        Statement statement = connection.createStatement();
+        ResultSet resultSet = statement.executeQuery(sqlStr)) {
+      resultSet.next();
+      int l0 = resultSet.getInt(2);
+      resultSet.next();
+      int l1 = resultSet.getInt(2);
+      resultSet.next();
+      int l2 = resultSet.getInt(2);
+      Assert.assertFalse(resultSet.next());
+      Assert.assertEquals(l0, l2);
+      Assert.assertNotEquals(l0, l1);
+    } catch (SQLException throwable) {
+      fail(throwable.getMessage());
+    }
+  }
+
+  @Test
+  public void testCluster2() {
+    String sqlStr =
+        "select cluster(d3.s1, 'l'='3', 'k'='2', 'method'='kshape', 'norm'='true', "
+            + "'maxiter'='50', 'output'='label') from root.vehicle";
+    try (Connection connection = EnvFactory.getEnv().getConnection();
+        Statement statement = connection.createStatement();
+        ResultSet resultSet = statement.executeQuery(sqlStr)) {
+      resultSet.next();
+      int l0 = resultSet.getInt(2);
+      resultSet.next();
+      int l1 = resultSet.getInt(2);
+      resultSet.next();
+      int l2 = resultSet.getInt(2);
+      Assert.assertFalse(resultSet.next());
+      Assert.assertEquals(l0, l1);
+      Assert.assertNotEquals(l0, l2);
+    } catch (SQLException throwable) {
+      fail(throwable.getMessage());
+    }
+  }
+
+  @Test
+  public void testCluster3() {
+    String sqlStr =
+        "select cluster(d3.s1, 'l'='3', 'k'='2', 'method'='medoidshape', 'norm'='true', "
+            + "'sample_rate'='1', 'maxiter'='50', 'output'='label') from root.vehicle";
+    try (Connection connection = EnvFactory.getEnv().getConnection();
+        Statement statement = connection.createStatement();
+        ResultSet resultSet = statement.executeQuery(sqlStr)) {
+      resultSet.next();
+      int l0 = resultSet.getInt(2);
+      resultSet.next();
+      int l1 = resultSet.getInt(2);
+      resultSet.next();
+      int l2 = resultSet.getInt(2);
+      Assert.assertFalse(resultSet.next());
+      Assert.assertEquals(l0, l1);
+      Assert.assertNotEquals(l0, l2);
+    } catch (SQLException throwable) {
+      fail(throwable.getMessage());
+    }
+  }
 }

library-udf/src/assembly/tools/register-UDF.sh

@@ -102,6 +102,7 @@ pass=root
 
 # Machine Learning
 ../sbin/start-cli.sh -h $host -p $rpcPort -u $user -pw $pass -e "create function ar as 'org.apache.iotdb.library.dlearn.UDTFAR'"
+../sbin/start-cli.sh -h $host -p $rpcPort -u $user -pw $pass -e "create function cluster as 'org.apache.iotdb.library.dlearn.UDTFCluster'"
 
 # Match
 ../sbin/start-cli.sh -h $host -p $rpcPort -u $user -pw $pass -e "create function pattern_match as 'org.apache.iotdb.library.match.UDAFPatternMatch'"

library-udf/src/main/java/org/apache/iotdb/library/dlearn/UDTFCluster.java

@@ -0,0 +1,216 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.library.dlearn;
+
+import org.apache.iotdb.library.dlearn.util.cluster.KMeans;
+import org.apache.iotdb.library.dlearn.util.cluster.KShape;
+import org.apache.iotdb.library.dlearn.util.cluster.MedoidShape;
+import org.apache.iotdb.library.util.Util;
+import org.apache.iotdb.udf.api.UDTF;
+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.exception.UDFException;
+import org.apache.iotdb.udf.api.type.Type;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Clusters a time series by partitioning it into non-overlapping subsequences of length l.
+ * Parameters: l, k, method (default kmeans), norm, maxiter, output; medoidshape also uses
+ * sample_rate (greedy sampling ratio; use 1 when the window count is small). Requires at least k
+ * windows.
+ */
+public class UDTFCluster implements UDTF {
+
+  private static final String METHOD_KMEANS = "kmeans";
+  private static final String METHOD_KSHAPE = "kshape";
+  private static final String METHOD_MEDOIDSHAPE = "medoidshape";
+
+  private static final String OUTPUT_LABEL = "label";
+  private static final String OUTPUT_CENTROID = "centroid";
+
+  private static final int DEFAULT_MAX_ITER = 200;
+  private static final double DEFAULT_SAMPLE_RATE = 0.3;
+  private static final String DEFAULT_METHOD = METHOD_KMEANS;
+
+  private int l;
+  private int k;
+  private String method;
+  private boolean norm;
+  private int maxIter;
+  private String output;
+  private double sampleRate;
+
+  private final List<Long> timestamps = new ArrayList<>();
+  private final List<Double> values = new ArrayList<>();
+
+  @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 l must be a positive integer.",
+            validator.getParameters().getInt("l"))
+        .validate(
+            x -> (int) x >= 2,
+            "Parameter k must be at least 2.",
+            validator.getParameters().getInt("k"))
+        .validate(
+            x -> {
+              String m = ((String) x).toLowerCase();
+              return METHOD_KMEANS.equals(m)
+                  || METHOD_KSHAPE.equals(m)
+                  || METHOD_MEDOIDSHAPE.equals(m);
+            },
+            "Parameter method must be one of: kmeans, kshape, medoidshape.",
+            validator.getParameters().getStringOrDefault("method", DEFAULT_METHOD))
+        .validate(
+            x -> (int) x >= 1,
+            "Parameter maxiter must be a positive integer.",
+            validator.getParameters().getIntOrDefault("maxiter", DEFAULT_MAX_ITER))
+        .validate(
+            x -> {
+              String o = ((String) x).toLowerCase();
+              return OUTPUT_LABEL.equals(o) || OUTPUT_CENTROID.equals(o);
+            },
+            "Parameter output must be label or centroid.",
+            validator.getParameters().getStringOrDefault("output", OUTPUT_LABEL))
+        .validate(
+            x -> {
+              double d = ((Number) x).doubleValue();
+              return d > 0 && d <= 1.0;
+            },
+            "Parameter sample_rate must be in (0, 1].",
+            validator.getParameters().getDoubleOrDefault("sample_rate", DEFAULT_SAMPLE_RATE));
+  }
+
+  @Override
+  public void beforeStart(UDFParameters parameters, UDTFConfigurations configurations)
+      throws Exception {
+    this.output = parameters.getStringOrDefault("output", OUTPUT_LABEL).toLowerCase();
+    if (OUTPUT_CENTROID.equals(output)) {
+      configurations.setAccessStrategy(new RowByRowAccessStrategy()).setOutputDataType(Type.DOUBLE);
+    } else {
+      configurations.setAccessStrategy(new RowByRowAccessStrategy()).setOutputDataType(Type.INT32);
+    }
+    this.l = parameters.getInt("l");
+    this.k = parameters.getInt("k");
+    this.method = parameters.getStringOrDefault("method", DEFAULT_METHOD).toLowerCase();
+    this.norm = parameters.getBooleanOrDefault("norm", true);
+    this.maxIter = parameters.getIntOrDefault("maxiter", DEFAULT_MAX_ITER);
+    this.sampleRate = parameters.getDoubleOrDefault("sample_rate", DEFAULT_SAMPLE_RATE);
+    timestamps.clear();
+    values.clear();
+  }
+
+  @Override
+  public void transform(Row row, PointCollector collector) throws Exception {
+    if (!row.isNull(0)) {
+      timestamps.add(row.getTime());
+      values.add(Util.getValueAsDouble(row));
+    }
+  }
+
+  @Override
+  public void terminate(PointCollector collector) throws Exception {
+    int n = values.size();
+    if (n < l) {
+      throw new UDFException(
+          "Time series length must be at least l; got " + n + " points, l=" + l + ".");
+    }
+    int numWindows = n / l;
+    if (numWindows < k) {
+      throw new UDFException(
+          "Not enough non-overlapping windows: got "
+              + numWindows
+              + " windows, need at least k="
+              + k
+              + ".");
+    }
+
+    double[][] windows = new double[numWindows][l];
+    long[] windowStartTime = new long[numWindows];
+    for (int w = 0; w < numWindows; w++) {
+      windowStartTime[w] = timestamps.get(w * l);
+      for (int j = 0; j < l; j++) {
+        windows[w][j] = values.get(w * l + j);
+      }
+    }
+
+    if (OUTPUT_LABEL.equals(output)) {
+      int[] labels;
+      if (METHOD_KMEANS.equals(method)) {
+        KMeans km = new KMeans();
+        km.fit(windows, k, norm, maxIter);
+        labels = km.getLabels();
+      } else if (METHOD_KSHAPE.equals(method)) {
+        KShape ks = new KShape();
+        ks.fit(windows, k, norm, maxIter);
+        labels = ks.getLabels();
+      } else if (METHOD_MEDOIDSHAPE.equals(method)) {
+        MedoidShape ms = new MedoidShape();
+        ms.setSampleRate(sampleRate);
+        ms.fit(windows, k, norm, maxIter);
+        labels = ms.getLabels();
+      } else {
+        throw new UDFException("Unsupported method: " + method);
+      }
+      for (int w = 0; w < numWindows; w++) {
+        collector.putInt(windowStartTime[w], labels[w]);
+      }
+    } else {
+      double[][] centroids;
+      if (METHOD_KMEANS.equals(method)) {
+        KMeans km = new KMeans();
+        km.fit(windows, k, norm, maxIter);
+        centroids = km.getCentroids();
+      } else if (METHOD_KSHAPE.equals(method)) {
+        KShape ks = new KShape();
+        ks.fit(windows, k, norm, maxIter);
+        centroids = ks.getCentroids();
+      } else if (METHOD_MEDOIDSHAPE.equals(method)) {
+        MedoidShape ms = new MedoidShape();
+        ms.setSampleRate(sampleRate);
+        ms.fit(windows, k, norm, maxIter);
+        centroids = ms.getCentroids();
+      } else {
+        throw new UDFException("Unsupported method: " + method);
+      }
+      emitConcatenatedCentroids(collector, centroids);
+    }
+  }
+
+  private static void emitConcatenatedCentroids(PointCollector collector, double[][] centroids)
+      throws Exception {
+    long t = 0L;
+    for (double[] row : centroids) {
+      for (double v : row) {
+        collector.putDouble(t++, v);
+      }
+    }
+  }
+}