You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2021/09/17 11:54:19 UTC
[iotdb] 04/05: add windowing tests
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch nested-operations
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 069eccca51903459c3f282fed644518451451db3
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Fri Sep 17 19:11:29 2021 +0800
add windowing tests
---
.../layer/MultiInputColumnIntermediateLayer.java | 21 ++---
...InputColumnMultiReferenceIntermediateLayer.java | 21 ++---
...nputColumnSingleReferenceIntermediateLayer.java | 21 ++---
.../iotdb/db/integration/IoTDBNestedQueryIT.java | 97 +++++++++++++++++++++-
.../db/integration/IoTDBUDFWindowQueryIT.java | 6 +-
.../integration/IoTDBUDTFAlignByTimeQueryIT.java | 6 +-
.../db/integration/IoTDBUDTFNonAlignQueryIT.java | 6 +-
.../apache/iotdb/db/query/udf/example/Counter.java | 24 ++++--
8 files changed, 142 insertions(+), 60 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/MultiInputColumnIntermediateLayer.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/MultiInputColumnIntermediateLayer.java
index 89034eb..6d0675f 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/MultiInputColumnIntermediateLayer.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/MultiInputColumnIntermediateLayer.java
@@ -283,24 +283,17 @@ public class MultiInputColumnIntermediateLayer extends IntermediateLayer
@Override
public boolean next() throws IOException, QueryProcessException {
- if (displayWindowEnd <= nextWindowTimeBegin) {
- return false;
- }
- if (hasCached || !hasAtLeastOneRow) {
+ if (hasCached) {
return true;
}
+ if (!hasAtLeastOneRow || displayWindowEnd <= nextWindowTimeBegin) {
+ return false;
+ }
long nextWindowTimeEnd = Math.min(nextWindowTimeBegin + timeInterval, displayWindowEnd);
- int oldTVListSize = rowRecordList.size();
while (rowRecordList.getTime(rowRecordList.size() - 1) < nextWindowTimeEnd) {
if (!LayerCacheUtils.cacheRow(udfInputDataSet, rowRecordList)) {
- if (displayWindowEnd == Long.MAX_VALUE
- // display window end == the max timestamp of the query result set
- && oldTVListSize == rowRecordList.size()) {
- return false;
- } else {
- break;
- }
+ break;
}
}
@@ -323,8 +316,8 @@ public class MultiInputColumnIntermediateLayer extends IntermediateLayer
}
window.seek(nextIndexBegin, nextIndexEnd);
- hasCached = true;
- return true;
+ hasCached = nextIndexBegin != nextIndexEnd;
+ return hasCached;
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/SingleInputColumnMultiReferenceIntermediateLayer.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/SingleInputColumnMultiReferenceIntermediateLayer.java
index 100d77f..53ecbf2 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/SingleInputColumnMultiReferenceIntermediateLayer.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/SingleInputColumnMultiReferenceIntermediateLayer.java
@@ -276,24 +276,17 @@ public class SingleInputColumnMultiReferenceIntermediateLayer extends Intermedia
@Override
public boolean next() throws IOException, QueryProcessException {
- if (displayWindowEnd <= nextWindowTimeBegin) {
- return false;
- }
- if (hasCached || !hasAtLeastOneRow) {
+ if (hasCached) {
return true;
}
+ if (!hasAtLeastOneRow || displayWindowEnd <= nextWindowTimeBegin) {
+ return false;
+ }
long nextWindowTimeEnd = Math.min(nextWindowTimeBegin + timeInterval, displayWindowEnd);
- int oldTVListSize = tvList.size();
while (tvList.getTime(tvList.size() - 1) < nextWindowTimeEnd) {
if (!LayerCacheUtils.cachePoint(dataType, parentLayerPointReader, tvList)) {
- if (displayWindowEnd == Long.MAX_VALUE
- // display window end == the max timestamp of the query result set
- && oldTVListSize == tvList.size()) {
- return false;
- } else {
- break;
- }
+ break;
}
}
@@ -316,8 +309,8 @@ public class SingleInputColumnMultiReferenceIntermediateLayer extends Intermedia
}
window.seek(nextIndexBegin, nextIndexEnd);
- hasCached = true;
- return true;
+ hasCached = nextIndexBegin != nextIndexEnd;
+ return hasCached;
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/SingleInputColumnSingleReferenceIntermediateLayer.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/SingleInputColumnSingleReferenceIntermediateLayer.java
index 16977ae..3dc3e34 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/SingleInputColumnSingleReferenceIntermediateLayer.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/layer/SingleInputColumnSingleReferenceIntermediateLayer.java
@@ -193,24 +193,17 @@ public class SingleInputColumnSingleReferenceIntermediateLayer extends Intermedi
@Override
public boolean next() throws IOException, QueryProcessException {
- if (displayWindowEnd <= nextWindowTimeBegin) {
- return false;
- }
- if (hasCached || !hasAtLeastOneRow) {
+ if (hasCached) {
return true;
}
+ if (!hasAtLeastOneRow || displayWindowEnd <= nextWindowTimeBegin) {
+ return false;
+ }
long nextWindowTimeEnd = Math.min(nextWindowTimeBegin + timeInterval, displayWindowEnd);
- int oldTVListSize = tvList.size();
while (tvList.getTime(tvList.size() - 1) < nextWindowTimeEnd) {
if (!LayerCacheUtils.cachePoint(dataType, parentLayerPointReader, tvList)) {
- if (displayWindowEnd == Long.MAX_VALUE
- // display window end == the max timestamp of the query result set
- && oldTVListSize == tvList.size()) {
- return false;
- } else {
- break;
- }
+ break;
}
}
@@ -233,8 +226,8 @@ public class SingleInputColumnSingleReferenceIntermediateLayer extends Intermedi
}
window.seek(nextIndexBegin, nextIndexEnd);
- hasCached = true;
- return true;
+ hasCached = nextIndexBegin != nextIndexEnd;
+ return hasCached;
}
@Override
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBNestedQueryIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBNestedQueryIT.java
index 035043d..735a941 100644
--- a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBNestedQueryIT.java
+++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBNestedQueryIT.java
@@ -39,6 +39,11 @@ import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
+import static org.apache.iotdb.db.integration.IoTDBUDFWindowQueryIT.ACCESS_STRATEGY_KEY;
+import static org.apache.iotdb.db.integration.IoTDBUDFWindowQueryIT.ACCESS_STRATEGY_SLIDING_SIZE;
+import static org.apache.iotdb.db.integration.IoTDBUDFWindowQueryIT.ACCESS_STRATEGY_SLIDING_TIME;
+import static org.apache.iotdb.db.integration.IoTDBUDFWindowQueryIT.TIME_INTERVAL_KEY;
+import static org.apache.iotdb.db.integration.IoTDBUDFWindowQueryIT.WINDOW_SIZE_KEY;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
@@ -48,9 +53,9 @@ public class IoTDBNestedQueryIT {
@BeforeClass
public static void setUp() throws Exception {
- IoTDBDescriptor.getInstance().getConfig().setUdfCollectorMemoryBudgetInMB(1);
- IoTDBDescriptor.getInstance().getConfig().setUdfTransformerMemoryBudgetInMB(1);
- IoTDBDescriptor.getInstance().getConfig().setUdfReaderMemoryBudgetInMB(1);
+ IoTDBDescriptor.getInstance().getConfig().setUdfCollectorMemoryBudgetInMB(5);
+ IoTDBDescriptor.getInstance().getConfig().setUdfTransformerMemoryBudgetInMB(5);
+ IoTDBDescriptor.getInstance().getConfig().setUdfReaderMemoryBudgetInMB(5);
EnvironmentUtils.envSetUp();
Class.forName(Config.JDBC_DRIVER_NAME);
createTimeSeries();
@@ -110,6 +115,10 @@ public class IoTDBNestedQueryIT {
Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
Statement statement = connection.createStatement()) {
statement.execute("create function adder as \"org.apache.iotdb.db.query.udf.example.Adder\"");
+ statement.execute(
+ "create function time_window_counter as \"org.apache.iotdb.db.query.udf.example.Counter\"");
+ statement.execute(
+ "create function size_window_counter as \"org.apache.iotdb.db.query.udf.example.Counter\"");
} catch (SQLException throwable) {
fail(throwable.getMessage());
}
@@ -130,6 +139,8 @@ public class IoTDBNestedQueryIT {
Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
Statement statement = connection.createStatement()) {
statement.execute("drop function adder");
+ statement.execute("drop function time_window_counter");
+ statement.execute("drop function size_window_counter");
} catch (SQLException throwable) {
fail(throwable.getMessage());
}
@@ -161,6 +172,7 @@ public class IoTDBNestedQueryIT {
assertEquals(-count, Double.parseDouble(resultSet.getString(7)), 0);
assertEquals(-count, Double.parseDouble(resultSet.getString(8)), 0);
}
+
assertEquals(ITERATION_TIMES, count);
} catch (SQLException throwable) {
fail(throwable.getMessage());
@@ -183,6 +195,7 @@ public class IoTDBNestedQueryIT {
int count = 0;
while (resultSet.next()) {
++count;
+
assertEquals(count, Integer.parseInt(resultSet.getString(1)));
assertEquals(count, Double.parseDouble(resultSet.getString(2)), 0);
assertEquals(count, Double.parseDouble(resultSet.getString(3)), 0);
@@ -192,6 +205,7 @@ public class IoTDBNestedQueryIT {
Double.parseDouble(resultSet.getString(5)),
1e-5);
}
+
assertEquals(ITERATION_TIMES, count);
} catch (SQLException throwable) {
fail(throwable.getMessage());
@@ -214,12 +228,14 @@ public class IoTDBNestedQueryIT {
int count = 0;
while (resultSet.next()) {
++count;
+
assertEquals(ITERATION_TIMES, Double.parseDouble(resultSet.getString(1)), 1e-5);
assertEquals(ITERATION_TIMES, Double.parseDouble(resultSet.getString(2)), 1e-5);
assertEquals(ITERATION_TIMES, Double.parseDouble(resultSet.getString(3)), 1e-5);
assertEquals(ITERATION_TIMES, Double.parseDouble(resultSet.getString(4)), 1e-5);
assertEquals(ITERATION_TIMES, Double.parseDouble(resultSet.getString(5)), 1e-5);
}
+
assertEquals(1, count);
} catch (SQLException throwable) {
fail(throwable.getMessage());
@@ -242,6 +258,7 @@ public class IoTDBNestedQueryIT {
int count = 0;
while (resultSet.next()) {
++count;
+
assertEquals(count, Double.parseDouble(resultSet.getString(1)), 1e-5);
assertEquals(2 * count, Double.parseDouble(resultSet.getString(2)), 1e-5);
assertEquals(-2 * count, Double.parseDouble(resultSet.getString(3)), 1e-5);
@@ -249,9 +266,83 @@ public class IoTDBNestedQueryIT {
assertEquals(4 * count, Double.parseDouble(resultSet.getString(5)), 1e-5);
assertEquals(0, Double.parseDouble(resultSet.getString(6)), 1e-5);
}
+
assertEquals(ITERATION_TIMES, count);
} catch (SQLException throwable) {
fail(throwable.getMessage());
}
}
+
+ @Test
+ public void testNestedWindowingFunctionExpressions() {
+ final int[] windows =
+ new int[] {
+ 1,
+ 2,
+ 3,
+ 100,
+ 499,
+ ITERATION_TIMES - 1,
+ ITERATION_TIMES,
+ ITERATION_TIMES + 1,
+ ITERATION_TIMES + 13
+ };
+
+ for (int window : windows) {
+ String sqlStr =
+ String.format(
+ "select time_window_counter(sin(d1.s1), '%s'='%s', '%s'='%s'), time_window_counter(sin(d1.s1), cos(d2.s2) / sin(d1.s1), d1.s2, '%s'='%s', '%s'='%s'), size_window_counter(cos(d2.s2), '%s'='%s', '%s'='%s'), size_window_counter(cos(d2.s2), cos(d2.s2), '%s'='%s', '%s'='%s') from root.vehicle",
+ ACCESS_STRATEGY_KEY,
+ ACCESS_STRATEGY_SLIDING_TIME,
+ TIME_INTERVAL_KEY,
+ window,
+ ACCESS_STRATEGY_KEY,
+ ACCESS_STRATEGY_SLIDING_TIME,
+ TIME_INTERVAL_KEY,
+ window,
+ ACCESS_STRATEGY_KEY,
+ ACCESS_STRATEGY_SLIDING_SIZE,
+ WINDOW_SIZE_KEY,
+ window,
+ ACCESS_STRATEGY_KEY,
+ ACCESS_STRATEGY_SLIDING_SIZE,
+ WINDOW_SIZE_KEY,
+ window);
+
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ ResultSet resultSet = statement.executeQuery(sqlStr);
+
+ assertEquals(1 + 4, resultSet.getMetaData().getColumnCount());
+
+ int count = 0;
+ while (resultSet.next()) {
+ assertEquals((long) count * window + 1, Long.parseLong(resultSet.getString(1)));
+
+ double c2 = Double.parseDouble(resultSet.getString(2));
+ double c3 = Double.parseDouble(resultSet.getString(3));
+ double c4 = Double.parseDouble(resultSet.getString(4));
+ double c5 = Double.parseDouble(resultSet.getString(5));
+
+ ++count;
+
+ assertEquals(
+ ((ITERATION_TIMES < count * window)
+ ? window - (count * window - ITERATION_TIMES)
+ : window),
+ c2,
+ 0);
+ assertEquals(c2, c3, 0);
+ assertEquals(c2, c4, 0);
+ assertEquals(c2, c5, 0);
+ }
+
+ assertEquals(ITERATION_TIMES / window + (ITERATION_TIMES % window == 0 ? 0 : 1), count);
+ } catch (SQLException throwable) {
+ fail(throwable.getMessage());
+ }
+ }
+ }
}
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBUDFWindowQueryIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBUDFWindowQueryIT.java
index 5be7e16..0609f39 100644
--- a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBUDFWindowQueryIT.java
+++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBUDFWindowQueryIT.java
@@ -60,9 +60,9 @@ public class IoTDBUDFWindowQueryIT {
@BeforeClass
public static void setUp() throws Exception {
- IoTDBDescriptor.getInstance().getConfig().setUdfCollectorMemoryBudgetInMB(1);
- IoTDBDescriptor.getInstance().getConfig().setUdfTransformerMemoryBudgetInMB(1);
- IoTDBDescriptor.getInstance().getConfig().setUdfReaderMemoryBudgetInMB(1);
+ IoTDBDescriptor.getInstance().getConfig().setUdfCollectorMemoryBudgetInMB(5);
+ IoTDBDescriptor.getInstance().getConfig().setUdfTransformerMemoryBudgetInMB(5);
+ IoTDBDescriptor.getInstance().getConfig().setUdfReaderMemoryBudgetInMB(5);
EnvironmentUtils.envSetUp();
Class.forName(Config.JDBC_DRIVER_NAME);
createTimeSeries();
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBUDTFAlignByTimeQueryIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBUDTFAlignByTimeQueryIT.java
index 2910174..f2f7fad 100644
--- a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBUDTFAlignByTimeQueryIT.java
+++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBUDTFAlignByTimeQueryIT.java
@@ -62,9 +62,9 @@ public class IoTDBUDTFAlignByTimeQueryIT {
@BeforeClass
public static void setUp() throws Exception {
- IoTDBDescriptor.getInstance().getConfig().setUdfCollectorMemoryBudgetInMB(1);
- IoTDBDescriptor.getInstance().getConfig().setUdfTransformerMemoryBudgetInMB(1);
- IoTDBDescriptor.getInstance().getConfig().setUdfReaderMemoryBudgetInMB(1);
+ IoTDBDescriptor.getInstance().getConfig().setUdfCollectorMemoryBudgetInMB(5);
+ IoTDBDescriptor.getInstance().getConfig().setUdfTransformerMemoryBudgetInMB(5);
+ IoTDBDescriptor.getInstance().getConfig().setUdfReaderMemoryBudgetInMB(5);
EnvironmentUtils.envSetUp();
Class.forName(Config.JDBC_DRIVER_NAME);
createTimeSeries();
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBUDTFNonAlignQueryIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBUDTFNonAlignQueryIT.java
index 69ab2ee..134a73c 100644
--- a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBUDTFNonAlignQueryIT.java
+++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBUDTFNonAlignQueryIT.java
@@ -60,9 +60,9 @@ public class IoTDBUDTFNonAlignQueryIT {
@BeforeClass
public static void setUp() throws Exception {
- IoTDBDescriptor.getInstance().getConfig().setUdfCollectorMemoryBudgetInMB(1);
- IoTDBDescriptor.getInstance().getConfig().setUdfTransformerMemoryBudgetInMB(1);
- IoTDBDescriptor.getInstance().getConfig().setUdfReaderMemoryBudgetInMB(1);
+ IoTDBDescriptor.getInstance().getConfig().setUdfCollectorMemoryBudgetInMB(5);
+ IoTDBDescriptor.getInstance().getConfig().setUdfTransformerMemoryBudgetInMB(5);
+ IoTDBDescriptor.getInstance().getConfig().setUdfReaderMemoryBudgetInMB(5);
EnvironmentUtils.envSetUp();
Class.forName(Config.JDBC_DRIVER_NAME);
createTimeSeries();
diff --git a/server/src/test/java/org/apache/iotdb/db/query/udf/example/Counter.java b/server/src/test/java/org/apache/iotdb/db/query/udf/example/Counter.java
index bcfdd5f..1e7b316 100644
--- a/server/src/test/java/org/apache/iotdb/db/query/udf/example/Counter.java
+++ b/server/src/test/java/org/apache/iotdb/db/query/udf/example/Counter.java
@@ -35,7 +35,15 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
-import static org.apache.iotdb.db.integration.IoTDBUDFWindowQueryIT.*;
+import static org.apache.iotdb.db.integration.IoTDBUDFWindowQueryIT.ACCESS_STRATEGY_KEY;
+import static org.apache.iotdb.db.integration.IoTDBUDFWindowQueryIT.ACCESS_STRATEGY_ROW_BY_ROW;
+import static org.apache.iotdb.db.integration.IoTDBUDFWindowQueryIT.ACCESS_STRATEGY_SLIDING_SIZE;
+import static org.apache.iotdb.db.integration.IoTDBUDFWindowQueryIT.ACCESS_STRATEGY_SLIDING_TIME;
+import static org.apache.iotdb.db.integration.IoTDBUDFWindowQueryIT.DISPLAY_WINDOW_BEGIN_KEY;
+import static org.apache.iotdb.db.integration.IoTDBUDFWindowQueryIT.DISPLAY_WINDOW_END_KEY;
+import static org.apache.iotdb.db.integration.IoTDBUDFWindowQueryIT.SLIDING_STEP_KEY;
+import static org.apache.iotdb.db.integration.IoTDBUDFWindowQueryIT.TIME_INTERVAL_KEY;
+import static org.apache.iotdb.db.integration.IoTDBUDFWindowQueryIT.WINDOW_SIZE_KEY;
public class Counter implements UDTF {
@@ -52,11 +60,15 @@ public class Counter implements UDTF {
break;
case ACCESS_STRATEGY_SLIDING_TIME:
configurations.setAccessStrategy(
- new SlidingTimeWindowAccessStrategy(
- parameters.getLong(TIME_INTERVAL_KEY),
- parameters.getLong(SLIDING_STEP_KEY),
- parameters.getLong(DISPLAY_WINDOW_BEGIN_KEY),
- parameters.getLong(DISPLAY_WINDOW_END_KEY)));
+ parameters.hasAttribute(SLIDING_STEP_KEY)
+ && parameters.hasAttribute(DISPLAY_WINDOW_BEGIN_KEY)
+ && parameters.hasAttribute(DISPLAY_WINDOW_END_KEY)
+ ? new SlidingTimeWindowAccessStrategy(
+ parameters.getLong(TIME_INTERVAL_KEY),
+ parameters.getLong(SLIDING_STEP_KEY),
+ parameters.getLong(DISPLAY_WINDOW_BEGIN_KEY),
+ parameters.getLong(DISPLAY_WINDOW_END_KEY))
+ : new SlidingTimeWindowAccessStrategy(parameters.getLong(TIME_INTERVAL_KEY)));
break;
case ACCESS_STRATEGY_ROW_BY_ROW:
default: