This is an automated email from the ASF dual-hosted git repository.
yongzao pushed a commit to branch TOP-K-DTW
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/TOP-K-DTW by this push:
new 30df81cfaf4 AddIT and Debug
30df81cfaf4 is described below
commit 30df81cfaf48499d39e865fde488973904a49484
Author: YongzaoDan <[email protected]>
AuthorDate: Fri Jun 2 16:47:48 2023 +0800
AddIT and Debug
---
.../BuiltinTimeSeriesGeneratingFunctionEnum.java | 3 +-
.../apache/iotdb/db/it/udf/IoTDBUDTFTopKDTWIT.java | 188 +++++++++++++++++++++
.../iotdb/commons/udf/builtin/UDTFTopKDTW.java | 34 ++++
.../udf/builtin/UDTFTopKDTWSlidingWindow.java | 35 +++-
4 files changed, 250 insertions(+), 10 deletions(-)
diff --git
a/integration-test/src/main/java/org/apache/iotdb/itbase/constant/BuiltinTimeSeriesGeneratingFunctionEnum.java
b/integration-test/src/main/java/org/apache/iotdb/itbase/constant/BuiltinTimeSeriesGeneratingFunctionEnum.java
index 337bdb97633..38ab236a27e 100644
---
a/integration-test/src/main/java/org/apache/iotdb/itbase/constant/BuiltinTimeSeriesGeneratingFunctionEnum.java
+++
b/integration-test/src/main/java/org/apache/iotdb/itbase/constant/BuiltinTimeSeriesGeneratingFunctionEnum.java
@@ -73,7 +73,8 @@ public enum BuiltinTimeSeriesGeneratingFunctionEnum {
EQUAL_SIZE_BUCKET_OUTLIER_SAMPLE("EQUAL_SIZE_BUCKET_OUTLIER_SAMPLE"),
JEXL("JEXL"),
MASTER_REPAIR("MASTER_REPAIR"),
- M4("M4");
+ M4("M4"),
+ TOP_K_DTW_SLIDING_WINDOW("TOP_K_DTW_SLIDING_WINDOW");
private final String functionName;
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/udf/IoTDBUDTFTopKDTWIT.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/udf/IoTDBUDTFTopKDTWIT.java
new file mode 100644
index 00000000000..902f0b01c35
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/db/it/udf/IoTDBUDTFTopKDTWIT.java
@@ -0,0 +1,188 @@
+/*
+ * 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.db.it.udf;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({LocalStandaloneIT.class})
+public class IoTDBUDTFTopKDTWIT {
+
+ private static final long TIMESTAMP_STRIDE = 1000L;
+ private static final long DATA_START_TIME = 954475200000L;
+
+ private static final int PATTERN_COUNT = 10;
+ private static final int PATTERN_LENGTH = 100;
+ private static final double PATTERN_RANGE = 100.0;
+
+ private static final int SERIES_LENGTH = 1000;
+ private static final double SERIES_RANGE = 1000.0;
+
+ private static long[] correctStartTimes;
+ private static long[] correctEndTimes;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvFactory.getEnv().initClusterEnvironment();
+ createTimeSeries();
+ generateData();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+
+ private static void createTimeSeries() {
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ statement.addBatch("create database root.database");
+ statement.addBatch(
+ "create timeseries root.database.device.s with "
+ + "datatype=double, "
+ + "encoding=plain, "
+ + "compression=uncompressed");
+ statement.addBatch(
+ "create timeseries root.database.device.p with "
+ + "datatype=double, "
+ + "encoding=plain, "
+ + "compression=uncompressed");
+ statement.executeBatch();
+ } catch (SQLException throwable) {
+ fail(throwable.getMessage());
+ }
+ }
+
+ private static void generateData() {
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ Random random = new Random();
+
+ double[] pattern = new double[PATTERN_LENGTH];
+ for (int i = 0; i < PATTERN_LENGTH; i++) {
+ // Construct pattern series with 100 points,
+ // each point is a random double in [-100, 100]
+ pattern[i] = PATTERN_RANGE * (2.0 * random.nextDouble() - 1.0);
+ statement.addBatch(
+ String.format(
+ "insert into root.database.device(timestamp, p) values(%d,
%f)",
+ i * TIMESTAMP_STRIDE, pattern[i]));
+ }
+
+ long currentTime = DATA_START_TIME;
+ correctStartTimes = new long[PATTERN_COUNT];
+ correctEndTimes = new long[PATTERN_COUNT];
+ for (int i = 0; i < PATTERN_COUNT; i++) {
+ // Construct series with 1000 points,
+ // each point is a random double in [-1000, 1000]
+ for (int j = 0; j < SERIES_LENGTH; j++) {
+ statement.addBatch(
+ String.format(
+ "insert into root.database.device(timestamp, s) values(%d,
%f)",
+ currentTime, SERIES_RANGE * (2.0 * random.nextDouble() -
1.0)));
+ currentTime += TIMESTAMP_STRIDE;
+ }
+
+ correctStartTimes[i] = currentTime;
+ for (int j = 0; j < PATTERN_LENGTH; j++) {
+ // Construct a similar pattern
+ int repeatTime = (int) Math.abs(random.nextGaussian()) + 1;
+ if (j == 0 || j == PATTERN_LENGTH - 1) {
+ // Make sure the first and last points are the same
+ repeatTime = 1;
+ }
+ for (int k = 0; k < repeatTime; k++) {
+ statement.addBatch(
+ String.format(
+ "insert into root.database.device(timestamp, s) values(%d,
%f)",
+ currentTime, pattern[j]));
+ currentTime += TIMESTAMP_STRIDE;
+ }
+ }
+ correctEndTimes[i] = currentTime - TIMESTAMP_STRIDE;
+ }
+
+ statement.executeBatch();
+
+ } catch (SQLException throwable) {
+ fail(throwable.getMessage());
+ }
+ }
+
+ @Test
+ public void testTopKDTWSlidingWindow() {
+ String sqlStr =
+ String.format(
+ "select top_k_dtw_sliding_window(s, p, 'k'='%d') from
root.database.device",
+ PATTERN_COUNT);
+
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+
+ ResultSet resultSet = statement.executeQuery(sqlStr);
+ checkResultSet(resultSet);
+
+ } catch (SQLException throwable) {
+ fail(throwable.getMessage());
+ }
+ }
+
+ // TODO: QGRAM Test
+
+ private void checkResultSet(ResultSet resultSet) throws SQLException {
+ List<Long> startTimes = new ArrayList<>();
+ List<Long> endTimes = new ArrayList<>();
+ List<Double> distances = new ArrayList<>();
+ while (resultSet.next()) {
+ String[] result = resultSet.getString(2).split(",");
+ startTimes.add(Long.parseLong(result[0].substring(result[0].indexOf('=')
+ 1)));
+ endTimes.add(Long.parseLong(result[1].substring(result[1].indexOf('=') +
1)));
+ distances.add(
+ Double.parseDouble(
+ result[2].substring(result[2].indexOf('=') + 1,
result[2].length() - 1)));
+ }
+ Assert.assertEquals(PATTERN_COUNT, startTimes.size());
+
+ for (int i = 0; i < PATTERN_COUNT; i++) {
+ Assert.assertEquals(correctStartTimes[i], startTimes.get(i).longValue());
+ Assert.assertEquals(correctEndTimes[i], endTimes.get(i).longValue());
+ Assert.assertEquals(0.0, distances.get(i), 1e-6);
+ }
+ }
+}
diff --git
a/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFTopKDTW.java
b/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFTopKDTW.java
index 9a9cb3be1f4..c0755b95416 100644
---
a/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFTopKDTW.java
+++
b/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFTopKDTW.java
@@ -1,6 +1,26 @@
+/*
+ * 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.commons.udf.builtin;
import org.apache.iotdb.udf.api.UDTF;
+import org.apache.iotdb.udf.api.access.Row;
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;
@@ -11,10 +31,14 @@ import org.apache.iotdb.udf.api.type.Type;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
public abstract class UDTFTopKDTW implements UDTF {
protected static final Logger LOGGER =
LoggerFactory.getLogger(UDTFTopKDTW.class);
+ protected static final double EPS = 1e-6;
+
private static final int DEFAULT_BATCH_SIZE = 65536;
private static final String K = "k";
@@ -48,4 +72,14 @@ public abstract class UDTFTopKDTW implements UDTF {
.setAccessStrategy(new SlidingSizeWindowAccessStrategy(batchSize))
.setOutputDataType(Type.TEXT);
}
+
+ protected double safelyReadDoubleValue(Row row, int columnIndex) throws
IOException {
+ switch (row.getDataType(columnIndex)) {
+ case FLOAT:
+ return row.getFloat(columnIndex);
+ case DOUBLE:
+ default:
+ return row.getDouble(columnIndex);
+ }
+ }
}
diff --git
a/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFTopKDTWSlidingWindow.java
b/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFTopKDTWSlidingWindow.java
index 94ba1e16d3d..67ae513b86d 100644
---
a/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFTopKDTWSlidingWindow.java
+++
b/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFTopKDTWSlidingWindow.java
@@ -75,8 +75,20 @@ public class UDTFTopKDTWSlidingWindow extends UDTFTopKDTW {
private DTWPath[] dtwBefore;
private DTWPath[] dtwCurrent;
- private final PriorityQueue<DTWPath> topK =
- new PriorityQueue<>((o1, o2) -> Double.compare(o2.distance,
o1.distance));
+ private final PriorityQueue<DTWPath> topK = new
PriorityQueue<>(this::compareDTWPath);
+
+ private int compareDTWPath(DTWPath p1, DTWPath p2) {
+ if (Math.abs(p1.distance - p2.distance) > EPS) {
+ // The bigger the distance, the higher the priority
+ return Double.compare(p2.distance, p1.distance);
+ } else if (p1.startTime != p2.startTime) {
+ // The bigger the start time, the higher the priority
+ return Long.compare(p2.startTime, p1.startTime);
+ } else {
+ // The bigger the end time, the higher the priority
+ return Long.compare(p2.endTime, p1.endTime);
+ }
+ }
@Override
public void transform(RowWindow rowWindow, PointCollector collector) throws
Exception {
@@ -89,7 +101,7 @@ public class UDTFTopKDTWSlidingWindow extends UDTFTopKDTW {
if (row.isNull(COLUMN_P)) {
break;
}
- patternList.add(new DTWPoint(row.getTime(), row.getDouble(COLUMN_P)));
+ patternList.add(new DTWPoint(row.getTime(), safelyReadDoubleValue(row,
COLUMN_P)));
}
pattern = patternList.toArray(new DTWPoint[0]);
}
@@ -101,13 +113,14 @@ public class UDTFTopKDTWSlidingWindow extends UDTFTopKDTW
{
continue;
}
- double value = row.getDouble(COLUMN_S);
+ long currentTime = row.getTime();
+ double value = safelyReadDoubleValue(row, COLUMN_S);
dtwCurrent = new DTWPath[pattern.length];
for (int i = 0; i < pattern.length; i++) {
double currentDistance = Math.abs(value - pattern[i].value);
if (i == 0) {
// Start a new DTW path
- dtwCurrent[i] = new DTWPath(pattern[i].time, pattern[i].time,
currentDistance);
+ dtwCurrent[i] = new DTWPath(currentTime, currentTime,
currentDistance);
continue;
}
@@ -121,14 +134,14 @@ public class UDTFTopKDTWSlidingWindow extends UDTFTopKDTW
{
dtwCurrent[i] = dtwBefore[i - 1].copy();
}
}
- dtwCurrent[i].endTime = pattern[i].time;
+ dtwCurrent[i].endTime = currentTime;
dtwCurrent[i].distance += currentDistance;
}
dtwBefore = Arrays.copyOf(dtwCurrent, dtwCurrent.length);
DTWPath currentPath = dtwCurrent[dtwCurrent.length - 1];
if (topK.size() < k) {
topK.offer(currentPath);
- } else if (topK.peek().distance > currentPath.distance) {
+ } else if (compareDTWPath(currentPath, topK.peek()) > 0) {
topK.poll();
topK.offer(currentPath);
}
@@ -137,8 +150,12 @@ public class UDTFTopKDTWSlidingWindow extends UDTFTopKDTW {
@Override
public void terminate(PointCollector collector) throws Exception {
- while (!topK.isEmpty()) {
- DTWPath path = topK.poll();
+ int topSize = topK.size();
+ DTWPath[] result = new DTWPath[topSize];
+ for (int i = topSize - 1; i >= 0; i--) {
+ result[i] = topK.poll();
+ }
+ for (DTWPath path : result) {
collector.putString(path.startTime, path.toString());
}
}
