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 <[email protected]> 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:
