You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by le...@apache.org on 2023/10/23 18:42:25 UTC
[iotdb] branch research/M4-visualization updated (d3ea2fd15e1 -> 46963054960)
This is an automated email from the ASF dual-hosted git repository.
leirui pushed a change to branch research/M4-visualization
in repository https://gitbox.apache.org/repos/asf/iotdb.git
from d3ea2fd15e1 add license
new 0dc6b40453f add LTTB
new 302007323e3 add MinMax
new 753a43d3969 fix
new 3d47731cc10 minmax-lsm 10%
new 4520f872b66 add
new d5855e8361a save query result
new 6a7b6f1db66 fix
new 85af5da3b1c add tracing info for updated disk points
new 46963054960 update
The 9 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../main/java/org/apache/iotdb/tool/ExportCsv.java | 12 +-
server/pom.xml | 17 ++
.../resources/conf/iotdb-engine.properties | 1 +
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 2 +-
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 8 +
.../apache/iotdb/db/query/control/TracingInfo.java | 10 +
.../iotdb/db/query/control/TracingManager.java | 14 +
.../groupby/GroupByWithoutValueFilterDataSet.java | 40 ++-
...r4CPV.java => LocalGroupByExecutor4MinMax.java} | 331 +++++----------------
.../reader/universal/PriorityMergeReader.java | 7 +
.../db/query/udf/builtin/BuiltinFunction.java | 4 +-
.../builtin/{UDTFM4MAC.java => UDTFMinMax.java} | 162 +---------
.../iotdb/db/query/udf/builtin/UDTFSample.java | 194 ++++++++++++
.../iotdb/db/query/udf/util/NoNumberException.java | 16 +-
.../org/apache/iotdb/db/query/udf/util/Util.java | 309 +++++++++++++++++++
.../apache/iotdb/db/tools/TsFileSketchTool.java | 3 +-
.../integration/m4/{MyTest5.java => MyTest6.java} | 125 +-------
.../integration/m4/{MyTest3.java => MyTest7.java} | 153 ++++------
.../integration/m4/{MyTest1.java => MyTest8.java} | 134 ++-------
.../org/apache/iotdb/session/SessionDataSet.java | 2 +-
.../iotdb/tsfile/common/conf/TSFileConfig.java | 10 +
.../iotdb/tsfile/common/conf/TSFileDescriptor.java | 4 +
.../iotdb/tsfile/read/common/ChunkSuit4CPV.java | 1 +
.../iotdb/tsfile/read/common/IOMonitor2.java | 2 +
24 files changed, 792 insertions(+), 769 deletions(-)
copy server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/{LocalGroupByExecutor4CPV.java => LocalGroupByExecutor4MinMax.java} (73%)
copy server/src/main/java/org/apache/iotdb/db/query/udf/builtin/{UDTFM4MAC.java => UDTFMinMax.java} (71%)
create mode 100644 server/src/main/java/org/apache/iotdb/db/query/udf/builtin/UDTFSample.java
copy tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/factory/FilterType.java => server/src/main/java/org/apache/iotdb/db/query/udf/util/NoNumberException.java (77%)
create mode 100644 server/src/main/java/org/apache/iotdb/db/query/udf/util/Util.java
copy server/src/test/java/org/apache/iotdb/db/integration/m4/{MyTest5.java => MyTest6.java} (62%)
copy server/src/test/java/org/apache/iotdb/db/integration/m4/{MyTest3.java => MyTest7.java} (76%)
copy server/src/test/java/org/apache/iotdb/db/integration/m4/{MyTest1.java => MyTest8.java} (81%)
[iotdb] 03/09: fix
Posted by le...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
leirui pushed a commit to branch research/M4-visualization
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 753a43d3969cac80b7c98522476819947bef3c21
Author: Lei Rui <10...@qq.com>
AuthorDate: Thu Sep 28 00:00:12 2023 +0800
fix
---
server/pom.xml | 8 --------
1 file changed, 8 deletions(-)
diff --git a/server/pom.xml b/server/pom.xml
index 775937ae168..3ca3b695285 100644
--- a/server/pom.xml
+++ b/server/pom.xml
@@ -278,14 +278,6 @@
</execution>
</executions>
</plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-compiler-plugin</artifactId>
- <configuration>
- <source>9</source>
- <target>9</target>
- </configuration>
- </plugin>
</plugins>
</build>
<profiles>
[iotdb] 02/09: add MinMax
Posted by le...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
leirui pushed a commit to branch research/M4-visualization
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 302007323e314ee2e08b77263904c766c8c9f00c
Author: Lei Rui <10...@qq.com>
AuthorDate: Wed Sep 27 22:46:13 2023 +0800
add MinMax
---
.../db/query/udf/builtin/BuiltinFunction.java | 3 +-
.../iotdb/db/query/udf/builtin/UDTFMinMax.java | 333 ++++++++++++++++++++-
.../apache/iotdb/db/integration/m4/MyTest6.java | 1 +
.../integration/m4/{MyTest6.java => MyTest7.java} | 178 +++++++----
4 files changed, 456 insertions(+), 59 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/BuiltinFunction.java b/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/BuiltinFunction.java
index ebde0642972..5a9639faaf4 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/BuiltinFunction.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/BuiltinFunction.java
@@ -49,7 +49,8 @@ public enum BuiltinFunction {
BOTTOM_K("BOTTOM_K", UDTFBottomK.class),
M4("M4", UDTFM4MAC.class),
M4_TW("M4_TW", UDTFM4.class),
- Sample("SAMPLE", UDTFSample.class);
+ Sample("SAMPLE", UDTFSample.class),
+ MinMax("MINMAX", UDTFMinMax.class);
;
private final String functionName;
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/UDTFMinMax.java b/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/UDTFMinMax.java
index da6dee8a979..6a8b1e1695a 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/UDTFMinMax.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/UDTFMinMax.java
@@ -1,3 +1,334 @@
+/*
+ * 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.query.udf.builtin;
-public class UDTFMinMax {}
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.query.udf.api.UDTF;
+import org.apache.iotdb.db.query.udf.api.access.Row;
+import org.apache.iotdb.db.query.udf.api.collector.PointCollector;
+import org.apache.iotdb.db.query.udf.api.customizer.config.UDTFConfigurations;
+import org.apache.iotdb.db.query.udf.api.customizer.parameter.UDFParameterValidator;
+import org.apache.iotdb.db.query.udf.api.customizer.parameter.UDFParameters;
+import org.apache.iotdb.db.query.udf.api.customizer.strategy.RowByRowAccessStrategy;
+import org.apache.iotdb.db.query.udf.api.exception.UDFException;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.IOMonitor2;
+import org.apache.iotdb.tsfile.read.common.IOMonitor2.DataSetType;
+
+import java.io.IOException;
+
+public class UDTFMinMax implements UDTF {
+
+ protected TSDataType dataType;
+ protected long tqs;
+ protected long tqe;
+ protected int w;
+
+ private long bottomTime;
+ private long topTime;
+
+ private int intMaxV;
+ private long longMaxV;
+ private float floatMaxV;
+ private double doubleMaxV;
+
+ private int intMinV;
+ private long longMinV;
+ private float floatMinV;
+ private double doubleMinV;
+
+ private String[] result;
+ private int idx;
+
+ private void init() {
+ this.bottomTime = 0;
+ this.topTime = 0;
+
+ this.intMinV = Integer.MAX_VALUE;
+ this.longMinV = Long.MAX_VALUE;
+ this.floatMinV = Float.MAX_VALUE;
+ this.doubleMinV = Double.MAX_VALUE;
+
+ this.intMaxV = Integer.MIN_VALUE;
+ this.longMaxV = Long.MIN_VALUE;
+ this.floatMaxV = Float.MIN_VALUE;
+ this.doubleMaxV = Double.MIN_VALUE;
+ }
+
+ @Override
+ public void validate(UDFParameterValidator validator) throws UDFException {
+ IOMonitor2.dataSetType = DataSetType.UDTFAlignByTimeDataSet_M4_POINT;
+ validator
+ .validateInputSeriesNumber(1)
+ .validateInputSeriesDataType(
+ 0, TSDataType.INT32, TSDataType.INT64, TSDataType.FLOAT, TSDataType.DOUBLE)
+ .validateRequiredAttribute("tqs")
+ .validateRequiredAttribute("tqe")
+ .validateRequiredAttribute("w");
+ }
+
+ @Override
+ public void beforeStart(UDFParameters parameters, UDTFConfigurations configurations)
+ throws MetadataException {
+ dataType = parameters.getDataType(0);
+ tqs = parameters.getLong("tqs"); // closed
+ tqe = parameters.getLong("tqe"); // open
+ w = parameters.getInt("w");
+ if ((tqe - tqs) % w != 0) {
+ throw new MetadataException("You should make tqe-tqs integer divide w");
+ }
+ configurations
+ .setAccessStrategy(new RowByRowAccessStrategy())
+ .setOutputDataType(TSDataType.TEXT);
+ init();
+ this.idx = -1;
+ result = new String[w];
+ for (int i = 0; i < w; i++) {
+ result[i] = "empty";
+ }
+ }
+
+ @Override
+ public void transform(Row row, PointCollector collector)
+ throws QueryProcessException, IOException {
+ if (idx < 0) {
+ idx = 0; // means at least not all empty
+ }
+ switch (dataType) {
+ case INT32:
+ transformInt(row.getTime(), row.getInt(0));
+ break;
+ case INT64:
+ transformLong(row.getTime(), row.getLong(0));
+ break;
+ case FLOAT:
+ transformFloat(row.getTime(), row.getFloat(0));
+ break;
+ case DOUBLE:
+ transformDouble(row.getTime(), row.getDouble(0));
+ break;
+ default:
+ break;
+ }
+ }
+
+ protected void transformInt(long time, int value) throws IOException {
+ long intervalLen = (tqe - tqs) / w;
+ int pos = (int) Math.floor((time - tqs) * 1.0 / intervalLen);
+ if (pos >= w) {
+ throw new IOException("Make sure the range time filter is time>=tqs and time<tqe");
+ }
+
+ if (pos > idx) {
+ result[idx] =
+ "BottomPoint=("
+ + bottomTime
+ + ","
+ + intMinV
+ + "), "
+ + "TopPoint=("
+ + topTime
+ + ","
+ + intMaxV
+ + ")";
+ idx = pos;
+ init(); // clear environment for this new interval
+ }
+ // update for the current interval
+ if (value < intMinV) {
+ bottomTime = time;
+ intMinV = value;
+ }
+ if (value > intMaxV) {
+ topTime = time;
+ intMaxV = value;
+ }
+ }
+
+ protected void transformLong(long time, long value) throws IOException {
+ long intervalLen = (tqe - tqs) / w;
+ int pos = (int) Math.floor((time - tqs) * 1.0 / intervalLen);
+
+ if (pos >= w) {
+ throw new IOException("Make sure the range time filter is time>=tqs and time<tqe");
+ }
+
+ if (pos > idx) {
+ result[idx] =
+ "BottomPoint=("
+ + bottomTime
+ + ","
+ + longMinV
+ + "), "
+ + "TopPoint=("
+ + topTime
+ + ","
+ + longMaxV
+ + ")";
+ idx = pos;
+ init(); // clear environment for this new interval
+ }
+
+ if (value < longMinV) {
+ bottomTime = time;
+ longMinV = value;
+ }
+ if (value > longMaxV) {
+ topTime = time;
+ longMaxV = value;
+ }
+ }
+
+ protected void transformFloat(long time, float value) throws IOException {
+ long intervalLen = (tqe - tqs) / w;
+ int pos = (int) Math.floor((time - tqs) * 1.0 / intervalLen);
+
+ if (pos >= w) {
+ throw new IOException("Make sure the range time filter is time>=tqs and time<tqe");
+ }
+
+ if (pos > idx) {
+ result[idx] =
+ "BottomPoint=("
+ + bottomTime
+ + ","
+ + floatMinV
+ + "), "
+ + "TopPoint=("
+ + topTime
+ + ","
+ + floatMaxV
+ + ")";
+ idx = pos;
+ init(); // clear environment for this new interval
+ }
+
+ if (value < floatMinV) {
+ bottomTime = time;
+ floatMinV = value;
+ }
+ if (value > floatMaxV) {
+ topTime = time;
+ floatMaxV = value;
+ }
+ }
+
+ protected void transformDouble(long time, double value) throws IOException {
+ long intervalLen = (tqe - tqs) / w;
+ int pos = (int) Math.floor((time - tqs) * 1.0 / intervalLen);
+
+ if (pos >= w) {
+ throw new IOException("Make sure the range time filter is time>=tqs and time<tqe");
+ }
+
+ if (pos > idx) {
+ result[idx] =
+ "BottomPoint=("
+ + bottomTime
+ + ","
+ + doubleMinV
+ + "), "
+ + "TopPoint=("
+ + topTime
+ + ","
+ + doubleMaxV
+ + ")";
+ idx = pos;
+ init(); // clear environment for this new interval
+ }
+
+ if (value < doubleMinV) {
+ bottomTime = time;
+ doubleMinV = value;
+ }
+ if (value > doubleMaxV) {
+ topTime = time;
+ doubleMaxV = value;
+ }
+ }
+
+ @Override
+ public void terminate(PointCollector collector) throws IOException, QueryProcessException {
+ if (idx >= 0) { // means at least not all empty
+ // record the last interval (not necessarily idx=w-1) in the transform stage
+ switch (dataType) {
+ case INT32:
+ result[idx] =
+ "BottomPoint=("
+ + bottomTime
+ + ","
+ + intMinV
+ + "), "
+ + "TopPoint=("
+ + topTime
+ + ","
+ + intMaxV
+ + ")";
+ break;
+ case INT64:
+ result[idx] =
+ "BottomPoint=("
+ + bottomTime
+ + ","
+ + longMinV
+ + "), "
+ + "TopPoint=("
+ + topTime
+ + ","
+ + longMaxV
+ + ")";
+ break;
+ case FLOAT:
+ result[idx] =
+ "BottomPoint=("
+ + bottomTime
+ + ","
+ + floatMinV
+ + "), "
+ + "TopPoint=("
+ + topTime
+ + ","
+ + floatMaxV
+ + ")";
+ break;
+ case DOUBLE:
+ result[idx] =
+ "BottomPoint=("
+ + bottomTime
+ + ","
+ + doubleMinV
+ + "), "
+ + "TopPoint=("
+ + topTime
+ + ","
+ + doubleMaxV
+ + ")";
+ break;
+ default:
+ break;
+ }
+ }
+ // collect result
+ for (int i = 0; i < w; i++) {
+ long startInterval = tqs + (tqe - tqs) / w * i;
+ collector.putString(startInterval, result[i]);
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest6.java b/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest6.java
index 46de9698d4d..c4442855377 100644
--- a/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest6.java
+++ b/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest6.java
@@ -41,6 +41,7 @@ import java.util.Locale;
import static org.junit.Assert.fail;
public class MyTest6 {
+ // test LTTB
private static final String TIMESTAMP_STR = "Time";
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest6.java b/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest7.java
similarity index 56%
copy from server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest6.java
copy to server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest7.java
index 46de9698d4d..e690338183e 100644
--- a/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest6.java
+++ b/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest7.java
@@ -34,14 +34,13 @@ import org.junit.Test;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
-import java.sql.SQLException;
import java.sql.Statement;
import java.util.Locale;
import static org.junit.Assert.fail;
-public class MyTest6 {
-
+public class MyTest7 {
+ // test MinMax
private static final String TIMESTAMP_STR = "Time";
private static String[] creationSqls =
@@ -89,83 +88,148 @@ public class MyTest6 {
}
@Test
- public void testM4Function() {
- // create timeseries
+ public void test1() { // test UDF MAC
+ prepareData1();
+
+ String[] res =
+ new String[] {
+ "0,BottomPoint=(1,5.0), TopPoint=(10,30.0)",
+ "25,BottomPoint=(25,8.0), TopPoint=(30,40.0)",
+ "50,BottomPoint=(52,8.0), TopPoint=(54,18.0)",
+ "75,empty"
+ };
try (Connection connection =
DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
Statement statement = connection.createStatement()) {
- statement.execute("SET STORAGE GROUP TO root.m4");
- statement.execute("CREATE TIMESERIES root.m4.d1.s1 with datatype=double,encoding=PLAIN");
- statement.execute("CREATE TIMESERIES root.m4.d1.s2 with datatype=INT64,encoding=PLAIN");
- } catch (SQLException throwable) {
- fail(throwable.getMessage());
+ long tqs = 0L;
+ long tqe = 100L;
+ int w = 4;
+ boolean hasResultSet =
+ statement.execute(
+ String.format(
+ "select MinMax(s0,'tqs'='%1$d','tqe'='%2$d','w'='%3$d') from root.vehicle.d0 where "
+ + "time>=%1$d and time<%2$d",
+ tqs, tqe, w));
+
+ String columnName =
+ "MinMax(root.vehicle.d0.s0, \"tqs\"=\"%d\", \"tqe\"=\"%d\", \"w\"=\"%d\")";
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ int i = 0;
+ while (resultSet.next()) {
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(String.format(columnName, tqs, tqe, w));
+ System.out.println(ans);
+ Assert.assertEquals(res[i++], ans);
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
}
+ }
+
+ @Test
+ public void test2() { // test UDF MAC extreme case: empty from the beginning
+ prepareData2();
- // insert data
- String insertTemplate = "INSERT INTO root.m4.d1(timestamp,%s)" + " VALUES(%d,%d)";
+ String[] res = new String[] {"0,empty", "50,empty"};
try (Connection connection =
DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
Statement statement = connection.createStatement()) {
- // "root.m4.d1.s1" data illustration:
- // https://user-images.githubusercontent.com/33376433/151985070-73158010-8ba0-409d-a1c1-df69bad1aaee.png
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 1, 5));
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 2, 15));
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 20, 1));
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 25, 8));
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 54, 3));
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 120, 8));
- statement.execute("FLUSH");
+ long tqs = 0L;
+ long tqe = 100L;
+ int w = 2;
+ boolean hasResultSet =
+ statement.execute(
+ String.format(
+ "select MinMax(s0,'tqs'='%1$d','tqe'='%2$d','w'='%3$d') from root.vehicle.d0 where "
+ + "time>=%1$d and time<%2$d",
+ tqs, tqe, w));
+
+ String columnName =
+ "MinMax(root.vehicle.d0.s0, \"tqs\"=\"%d\", \"tqe\"=\"%d\", \"w\"=\"%d\")";
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ int i = 0;
+ while (resultSet.next()) {
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(String.format(columnName, tqs, tqe, w));
+ System.out.println(ans);
+ Assert.assertEquals(res[i++], ans);
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ private static void prepareData1() {
+ // data:
+ // https://user-images.githubusercontent.com/33376433/151985070-73158010-8ba0-409d-a1c1-df69bad1aaee.png
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ for (String sql : creationSqls) {
+ statement.execute(sql);
+ }
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 5, 10));
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 8, 8));
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 10, 30));
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 20, 20));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 1, 5));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 2, 15));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 20, 1));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 25, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 54, 3));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 120, 8));
statement.execute("FLUSH");
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 27, 20));
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 30, 40));
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 35, 10));
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 40, 20));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 5, 10));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 8, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 10, 30));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 20, 20));
statement.execute("FLUSH");
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 33, 9));
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 45, 30));
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 52, 8));
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 54, 18));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 27, 20));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 30, 40));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 35, 10));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 40, 20));
statement.execute("FLUSH");
- // "root.m4.d1.s2" data: constant value 1
- for (int i = 0; i < 100; i++) {
- statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s2", i, 1));
- }
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 33, 9));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 45, 30));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 52, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 54, 18));
statement.execute("FLUSH");
+
} catch (Exception e) {
e.printStackTrace();
}
-
- // query tests
- test_sample();
}
- private void test_sample() {
- String[] res = new String[] {"1,5.0", "10,30.0", "33,9.0", "120,8.0"};
-
- String sql = "select Sample(s1, 'method'='triangle','k'='4') from root.m4.d1";
+ private static void prepareData2() {
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
- try (Connection conn =
- DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
- Statement statement = conn.createStatement()) {
- ResultSet resultSet = statement.executeQuery(sql);
- int count = 0;
- while (resultSet.next()) {
- String str = resultSet.getString(1) + "," + resultSet.getString(2);
- Assert.assertEquals(res[count], str);
- count++;
- System.out.println(str);
+ for (String sql : creationSqls) {
+ statement.execute(sql);
}
- Assert.assertEquals(res.length, count);
- } catch (SQLException throwable) {
- fail(throwable.getMessage());
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 200, 5));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 300, 15));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 400, 1));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 500, 8));
+ statement.execute("FLUSH");
+
+ } catch (Exception e) {
+ e.printStackTrace();
}
}
}
[iotdb] 07/09: fix
Posted by le...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
leirui pushed a commit to branch research/M4-visualization
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 6a7b6f1db66e4af2360f4698f977c3bce05d4def
Author: Lei Rui <10...@qq.com>
AuthorDate: Sun Oct 1 14:36:39 2023 +0800
fix
---
cli/src/main/java/org/apache/iotdb/tool/ExportCsv.java | 12 ++++++++----
1 file changed, 8 insertions(+), 4 deletions(-)
diff --git a/cli/src/main/java/org/apache/iotdb/tool/ExportCsv.java b/cli/src/main/java/org/apache/iotdb/tool/ExportCsv.java
index d645447db71..92929673902 100644
--- a/cli/src/main/java/org/apache/iotdb/tool/ExportCsv.java
+++ b/cli/src/main/java/org/apache/iotdb/tool/ExportCsv.java
@@ -40,7 +40,11 @@ import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVPrinter;
import org.apache.commons.csv.QuoteMode;
-import java.io.*;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.PrintWriter;
import java.time.Instant;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
@@ -342,9 +346,9 @@ public class ExportCsv extends AbstractCsvTool {
while (sessionDataSet.hasNext()) {
RowRecord rowRecord = sessionDataSet.next();
ArrayList<String> record = new ArrayList<>();
- if (rowRecord.getTimestamp() != 0) {
- record.add(timeTrans(rowRecord.getTimestamp()));
- }
+ // if (rowRecord.getTimestamp() != 0) {
+ record.add(timeTrans(rowRecord.getTimestamp()));
+ // }
rowRecord
.getFields()
.forEach(
[iotdb] 04/09: minmax-lsm 10%
Posted by le...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
leirui pushed a commit to branch research/M4-visualization
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 3d47731cc10999efe6e158c9ce9ec24ec5190355
Author: Lei Rui <10...@qq.com>
AuthorDate: Sat Sep 30 13:56:10 2023 +0800
minmax-lsm 10%
---
.../resources/conf/iotdb-engine.properties | 1 +
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 8 +
.../groupby/GroupByWithoutValueFilterDataSet.java | 40 +-
.../groupby/LocalGroupByExecutor4MinMax.java | 966 +++++++++++++++++++++
.../iotdb/tsfile/common/conf/TSFileConfig.java | 10 +
.../iotdb/tsfile/common/conf/TSFileDescriptor.java | 4 +
.../iotdb/tsfile/read/common/IOMonitor2.java | 2 +
7 files changed, 1015 insertions(+), 16 deletions(-)
diff --git a/server/src/assembly/resources/conf/iotdb-engine.properties b/server/src/assembly/resources/conf/iotdb-engine.properties
index 8b202feffaf..95228d09694 100644
--- a/server/src/assembly/resources/conf/iotdb-engine.properties
+++ b/server/src/assembly/resources/conf/iotdb-engine.properties
@@ -605,6 +605,7 @@ enable_unseq_compaction=false
####################
### Configurations for tsfile-format
####################
+enableMinMaxLSM=false
use_Statistics=true
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index b7a80faebae..4711470e002 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -896,6 +896,14 @@ public class IoTDBDescriptor {
}
private void loadTsFileProps(Properties properties) {
+ TSFileDescriptor.getInstance()
+ .getConfig()
+ .setEnableMinMaxLSM(
+ Boolean.parseBoolean(
+ properties.getProperty(
+ "enableMinMaxLSM",
+ Boolean.toString(
+ TSFileDescriptor.getInstance().getConfig().isEnableMinMaxLSM()))));
TSFileDescriptor.getInstance()
.getConfig()
.setUseStatistics(
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
index bd02b795e77..b23d9429e1f 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
@@ -201,24 +201,32 @@ public class GroupByWithoutValueFilterDataSet extends GroupByEngineDataSet {
boolean ascending)
throws StorageEngineException, QueryProcessException {
if (CONFIG.isEnableCPV()) {
- if (TSFileDescriptor.getInstance().getConfig().isUseTimeIndex()
- && TSFileDescriptor.getInstance().getConfig().isUseValueIndex()) {
+ if (TSFileDescriptor.getInstance().getConfig().isEnableMinMaxLSM()) { // MinMax-LSM
IOMonitor2.dataSetType =
- DataSetType.GroupByWithoutValueFilterDataSet_LocalGroupByExecutor4CPV_UseIndex;
- } else if (!TSFileDescriptor.getInstance().getConfig().isUseTimeIndex()
- && TSFileDescriptor.getInstance().getConfig().isUseValueIndex()) {
- IOMonitor2.dataSetType =
- DataSetType.GroupByWithoutValueFilterDataSet_LocalGroupByExecutor4CPV_NoTimeIndex;
- } else if (TSFileDescriptor.getInstance().getConfig().isUseTimeIndex()
- && !TSFileDescriptor.getInstance().getConfig().isUseValueIndex()) {
- IOMonitor2.dataSetType =
- DataSetType.GroupByWithoutValueFilterDataSet_LocalGroupByExecutor4CPV_NoValueIndex;
- } else {
- IOMonitor2.dataSetType =
- DataSetType.GroupByWithoutValueFilterDataSet_LocalGroupByExecutor4CPV_NoTimeValueIndex;
+ DataSetType.GroupByWithoutValueFilterDataSet_LocalGroupByExecutor4CPV_EnableMinMaxLSM;
+ return new LocalGroupByExecutor4MinMax(
+ path, allSensors, dataType, context, timeFilter, fileFilter, ascending);
+ } else { // M4-LSM
+ if (TSFileDescriptor.getInstance().getConfig().isUseTimeIndex()
+ && TSFileDescriptor.getInstance().getConfig().isUseValueIndex()) {
+ IOMonitor2.dataSetType =
+ DataSetType.GroupByWithoutValueFilterDataSet_LocalGroupByExecutor4CPV_UseIndex;
+ } else if (!TSFileDescriptor.getInstance().getConfig().isUseTimeIndex()
+ && TSFileDescriptor.getInstance().getConfig().isUseValueIndex()) {
+ IOMonitor2.dataSetType =
+ DataSetType.GroupByWithoutValueFilterDataSet_LocalGroupByExecutor4CPV_NoTimeIndex;
+ } else if (TSFileDescriptor.getInstance().getConfig().isUseTimeIndex()
+ && !TSFileDescriptor.getInstance().getConfig().isUseValueIndex()) {
+ IOMonitor2.dataSetType =
+ DataSetType.GroupByWithoutValueFilterDataSet_LocalGroupByExecutor4CPV_NoValueIndex;
+ } else {
+ IOMonitor2.dataSetType =
+ DataSetType
+ .GroupByWithoutValueFilterDataSet_LocalGroupByExecutor4CPV_NoTimeValueIndex;
+ }
+ return new LocalGroupByExecutor4CPV(
+ path, allSensors, dataType, context, timeFilter, fileFilter, ascending);
}
- return new LocalGroupByExecutor4CPV(
- path, allSensors, dataType, context, timeFilter, fileFilter, ascending);
} else { // enableCPV=false
if (TSFileDescriptor.getInstance().getConfig().isUseStatistics()) {
IOMonitor2.dataSetType =
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor4MinMax.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor4MinMax.java
new file mode 100644
index 00000000000..ff074f588cb
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor4MinMax.java
@@ -0,0 +1,966 @@
+/*
+ * 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.query.dataset.groupby;
+
+import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.query.filter.TsFileFilter;
+import org.apache.iotdb.db.query.reader.series.SeriesReader;
+import org.apache.iotdb.db.query.reader.universal.PriorityMergeReader.MergeReaderPriority;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.statistics.DoubleStatistics;
+import org.apache.iotdb.tsfile.file.metadata.statistics.FloatStatistics;
+import org.apache.iotdb.tsfile.file.metadata.statistics.IntegerStatistics;
+import org.apache.iotdb.tsfile.file.metadata.statistics.LongStatistics;
+import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
+import org.apache.iotdb.tsfile.read.common.ChunkSuit4CPV;
+import org.apache.iotdb.tsfile.read.common.TimeRange;
+import org.apache.iotdb.tsfile.read.filter.GroupByFilter;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.reader.page.PageReader;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.Map;
+import java.util.Set;
+
+// This is the MFGroupByExecutor in M4-LSM paper.
+public class LocalGroupByExecutor4MinMax implements GroupByExecutor {
+
+ private static final Logger M4_CHUNK_METADATA = LoggerFactory.getLogger("M4_CHUNK_METADATA");
+
+ // Aggregate result buffer of this path
+ private final List<AggregateResult> results = new ArrayList<>();
+
+ private List<ChunkSuit4CPV> currentChunkList;
+ private final List<ChunkSuit4CPV> futureChunkList = new ArrayList<>();
+
+ // this is designed to keep the split chunk from futureChunkList, not destroying the sorted order
+ // of futureChunkList
+ private Map<Integer, List<ChunkSuit4CPV>> splitChunkList = new HashMap<>();
+
+ private Filter timeFilter;
+
+ private TSDataType tsDataType;
+
+ public LocalGroupByExecutor4MinMax(
+ PartialPath path,
+ Set<String> allSensors,
+ TSDataType dataType,
+ QueryContext context,
+ Filter timeFilter,
+ TsFileFilter fileFilter,
+ boolean ascending)
+ throws StorageEngineException, QueryProcessException {
+ // long start = System.nanoTime();
+
+ this.tsDataType = dataType;
+
+ // get all data sources
+ QueryDataSource queryDataSource =
+ QueryResourceManager.getInstance().getQueryDataSource(path, context, this.timeFilter);
+
+ // update filter by TTL
+ this.timeFilter = queryDataSource.updateFilterUsingTTL(timeFilter);
+
+ SeriesReader seriesReader =
+ new SeriesReader(
+ path,
+ allSensors,
+ dataType,
+ context,
+ queryDataSource,
+ timeFilter,
+ null,
+ fileFilter,
+ ascending);
+
+ // unpackAllOverlappedFilesToTimeSeriesMetadata
+ try {
+ // : this might be bad to load all chunk metadata at first
+ futureChunkList.addAll(seriesReader.getAllChunkMetadatas4CPV());
+ // order futureChunkList by chunk startTime
+ futureChunkList.sort(
+ new Comparator<ChunkSuit4CPV>() {
+ public int compare(ChunkSuit4CPV o1, ChunkSuit4CPV o2) {
+ return ((Comparable) (o1.getChunkMetadata().getStartTime()))
+ .compareTo(o2.getChunkMetadata().getStartTime());
+ }
+ });
+
+ if (M4_CHUNK_METADATA.isDebugEnabled()) {
+ if (timeFilter instanceof GroupByFilter) {
+ M4_CHUNK_METADATA.debug(
+ "M4_QUERY_PARAM,{},{},{}",
+ ((GroupByFilter) timeFilter).getStartTime(),
+ ((GroupByFilter) timeFilter).getEndTime(),
+ ((GroupByFilter) timeFilter).getInterval());
+ }
+ for (ChunkSuit4CPV chunkSuit4CPV : futureChunkList) {
+ Statistics statistics = chunkSuit4CPV.getChunkMetadata().getStatistics();
+ long FP_t = statistics.getStartTime();
+ long LP_t = statistics.getEndTime();
+ long BP_t = statistics.getBottomTimestamp();
+ long TP_t = statistics.getTopTimestamp();
+ switch (statistics.getType()) {
+ case INT32:
+ int FP_v_int = ((IntegerStatistics) statistics).getFirstValue();
+ int LP_v_int = ((IntegerStatistics) statistics).getLastValue();
+ int BP_v_int = ((IntegerStatistics) statistics).getMinValue();
+ int TP_v_int = ((IntegerStatistics) statistics).getMaxValue();
+ M4_CHUNK_METADATA.debug(
+ "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}",
+ FP_t,
+ LP_t,
+ BP_t,
+ TP_t,
+ FP_v_int,
+ LP_v_int,
+ BP_v_int,
+ TP_v_int,
+ chunkSuit4CPV.getChunkMetadata().getVersion(),
+ chunkSuit4CPV.getChunkMetadata().getOffsetOfChunkHeader(),
+ statistics.getCount());
+ break;
+ case INT64:
+ long FP_v_long = ((LongStatistics) statistics).getFirstValue();
+ long LP_v_long = ((LongStatistics) statistics).getLastValue();
+ long BP_v_long = ((LongStatistics) statistics).getMinValue();
+ long TP_v_long = ((LongStatistics) statistics).getMaxValue();
+ M4_CHUNK_METADATA.debug(
+ "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}",
+ FP_t,
+ LP_t,
+ BP_t,
+ TP_t,
+ FP_v_long,
+ LP_v_long,
+ BP_v_long,
+ TP_v_long,
+ chunkSuit4CPV.getChunkMetadata().getVersion(),
+ chunkSuit4CPV.getChunkMetadata().getOffsetOfChunkHeader(),
+ statistics.getCount());
+ break;
+ case FLOAT:
+ float FP_v_float = ((FloatStatistics) statistics).getFirstValue();
+ float LP_v_float = ((FloatStatistics) statistics).getLastValue();
+ float BP_v_float = ((FloatStatistics) statistics).getMinValue();
+ float TP_v_float = ((FloatStatistics) statistics).getMaxValue();
+ M4_CHUNK_METADATA.debug(
+ "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}",
+ FP_t,
+ LP_t,
+ BP_t,
+ TP_t,
+ FP_v_float,
+ LP_v_float,
+ BP_v_float,
+ TP_v_float,
+ chunkSuit4CPV.getChunkMetadata().getVersion(),
+ chunkSuit4CPV.getChunkMetadata().getOffsetOfChunkHeader(),
+ statistics.getCount());
+ break;
+ case DOUBLE:
+ double FP_v_double = ((DoubleStatistics) statistics).getFirstValue();
+ double LP_v_double = ((DoubleStatistics) statistics).getLastValue();
+ double BP_v_double = ((DoubleStatistics) statistics).getMinValue();
+ double TP_v_double = ((DoubleStatistics) statistics).getMaxValue();
+ M4_CHUNK_METADATA.debug(
+ "M4_CHUNK_METADATA,{},{},{},{},{},{},{},{},{},{},{}",
+ FP_t,
+ LP_t,
+ BP_t,
+ TP_t,
+ FP_v_double,
+ LP_v_double,
+ BP_v_double,
+ TP_v_double,
+ chunkSuit4CPV.getChunkMetadata().getVersion(),
+ chunkSuit4CPV.getChunkMetadata().getOffsetOfChunkHeader(),
+ statistics.getCount());
+ break;
+ default:
+ throw new QueryProcessException("unsupported data type!");
+ }
+ }
+ }
+
+ } catch (IOException e) {
+ throw new QueryProcessException(e.getMessage());
+ }
+
+ // IOMonitor2.addMeasure(Operation.M4_LSM_INIT_LOAD_ALL_CHUNKMETADATAS, System.nanoTime() -
+ // start);
+ }
+
+ @Override
+ public void addAggregateResult(AggregateResult aggrResult) {
+ results.add(aggrResult);
+ }
+
+ private void getCurrentChunkListFromFutureChunkList(
+ long curStartTime, long curEndTime, long startTime, long endTime, long interval)
+ throws IOException {
+ // IOMonitor2.M4_LSM_status = Operation.M4_LSM_MERGE_M4_TIME_SPAN;
+
+ // empty currentChunkList
+ currentChunkList = new ArrayList<>();
+
+ // get related chunks from splitChunkList
+ int curIdx = (int) Math.floor((curStartTime - startTime) * 1.0 / interval);
+ if (splitChunkList.get(curIdx) != null) {
+ currentChunkList.addAll(splitChunkList.get(curIdx));
+ // when to free splitChunkList memory
+ }
+
+ // iterate futureChunkList
+ ListIterator<ChunkSuit4CPV> itr = futureChunkList.listIterator();
+ while (itr.hasNext()) {
+ ChunkSuit4CPV chunkSuit4CPV = (ChunkSuit4CPV) (itr.next());
+ ChunkMetadata chunkMetadata = chunkSuit4CPV.getChunkMetadata();
+ long chunkMinTime = chunkMetadata.getStartTime();
+ long chunkMaxTime = chunkMetadata.getEndTime();
+ if (chunkMinTime >= curEndTime && chunkMinTime < endTime) {
+ // the chunk falls on the right side of the current M4 interval Ii,
+ // and since futureChunkList is ordered by the startTime of chunkMetadata,
+ // the loop can be terminated early.
+ break;
+ } else if (chunkMaxTime < curStartTime || chunkMinTime >= endTime) {
+ // the chunk falls on the left side of the current M4 interval Ii
+ // or the chunk falls on the right side of the total query range
+ itr.remove();
+ } else if (chunkMinTime >= curStartTime && chunkMaxTime < curEndTime) {
+ // the chunk falls completely within the current M4 interval Ii
+ currentChunkList.add(chunkSuit4CPV);
+ itr.remove();
+ } else {
+ // the chunk partially overlaps in time with the current M4 interval Ii.
+ // load this chunk, split it on deletes and all w intervals.
+ // add to currentChunkList and futureChunkList.
+ itr.remove();
+ // B: loads chunk data from disk to memory
+ // C: decompress page data, split time&value buffers
+ PageReader pageReader =
+ FileLoaderUtils.loadPageReaderList4CPV(
+ chunkSuit4CPV.getChunkMetadata(), this.timeFilter);
+ // ATTENTION: YOU HAVE TO ENSURE THAT THERE IS ONLY ONE PAGE IN A CHUNK,
+ // BECAUSE THE WHOLE IMPLEMENTATION IS BASED ON THIS ASSUMPTION.
+ // OTHERWISE, PAGEREADER IS FOR THE FIRST PAGE IN THE CHUNK WHILE
+ // STEPREGRESS IS FOR THE LAST PAGE IN THE CHUNK (THE MERGE OF STEPREGRESS IS ASSIGN
+ // DIRECTLY), WHICH WILL INTRODUCE BUGS!
+
+ // chunk data read operation (b) get the closest data point after or before a timestamp
+ pageReader.split4CPV(
+ startTime,
+ endTime,
+ interval,
+ curStartTime,
+ currentChunkList,
+ splitChunkList,
+ chunkMetadata);
+ }
+ }
+ }
+
+ /**
+ * @param curStartTime closed
+ * @param curEndTime open
+ * @param startTime closed
+ * @param endTime open
+ */
+ @Override
+ public List<AggregateResult> calcResult(
+ long curStartTime, long curEndTime, long startTime, long endTime, long interval)
+ throws IOException {
+ // clear result cache
+ for (AggregateResult result : results) {
+ result.reset();
+ }
+
+ // long start = System.nanoTime();
+ getCurrentChunkListFromFutureChunkList(curStartTime, curEndTime, startTime, endTime, interval);
+ // IOMonitor2.addMeasure(Operation.M4_LSM_MERGE_M4_TIME_SPAN, System.nanoTime() - start);
+
+ if (currentChunkList.size() == 0) {
+ return results;
+ }
+
+ // start = System.nanoTime();
+ calculateBottomPoint(currentChunkList, startTime, endTime, interval, curStartTime);
+ // IOMonitor2.addMeasure(Operation.M4_LSM_BP, System.nanoTime() - start);
+
+ // start = System.nanoTime();
+ calculateTopPoint(currentChunkList, startTime, endTime, interval, curStartTime);
+ // IOMonitor2.addMeasure(Operation.M4_LSM_TP, System.nanoTime() - start);
+
+ return results;
+ }
+
+ private void calculateBottomPoint(
+ List<ChunkSuit4CPV> currentChunkList,
+ long startTime,
+ long endTime,
+ long interval,
+ long curStartTime)
+ throws IOException {
+ // IOMonitor2.M4_LSM_status = Operation.M4_LSM_BP;
+ // check size>0 because after updateBPTP because empty ChunkSuit4CPV will be removed from
+ // currentChunkList
+ while (currentChunkList.size() > 0) { // loop 1
+ // sorted by bottomValue, find BP candidate set
+ // double check the sort order logic for different aggregations
+ currentChunkList.sort(
+ (o1, o2) -> {
+ return ((Comparable) (o1.getStatistics().getMinValue()))
+ .compareTo(o2.getStatistics().getMinValue());
+ // NOTE here get statistics from ChunkSuit4CPV, not from ChunkSuit4CPV.ChunkMetadata
+ });
+ // NOTE here get statistics from ChunkSuit4CPV, not from ChunkSuit4CPV.ChunkMetadata
+ Object value = currentChunkList.get(0).getStatistics().getMinValue();
+ List<ChunkSuit4CPV> candidateSet = new ArrayList<>();
+ for (ChunkSuit4CPV chunkSuit4CPV : currentChunkList) {
+ // NOTE here get statistics from ChunkSuit4CPV, not from ChunkSuit4CPV.ChunkMetadata
+ if (chunkSuit4CPV.getStatistics().getMinValue().equals(value)) {
+ candidateSet.add(chunkSuit4CPV);
+ } else {
+ break; // note that this is an early break since currentChunkList is sorted
+ }
+ }
+
+ // check, whether nonLazyLoad remove affects candidateSet
+ List<ChunkSuit4CPV> nonLazyLoad = new ArrayList<>(candidateSet);
+ // double check the sort order logic for version
+ nonLazyLoad.sort(
+ (o1, o2) ->
+ new MergeReaderPriority(
+ o2.getChunkMetadata().getVersion(),
+ o2.getChunkMetadata().getOffsetOfChunkHeader())
+ .compareTo(
+ new MergeReaderPriority(
+ o1.getChunkMetadata().getVersion(),
+ o1.getChunkMetadata().getOffsetOfChunkHeader())));
+ while (true) { // loop 2
+ // if there is no chunk for lazy loading, then load all chunks in candidateSet,
+ // and apply deleteIntervals, deleting BP no matter out of deletion or update
+ if (nonLazyLoad.size() == 0) {
+ for (ChunkSuit4CPV chunkSuit4CPV : candidateSet) {
+ // Note the pass of delete intervals
+ if (chunkSuit4CPV.getPageReader() == null) {
+ PageReader pageReader =
+ FileLoaderUtils.loadPageReaderList4CPV(
+ chunkSuit4CPV.getChunkMetadata(), this.timeFilter);
+ // ATTENTION: YOU HAVE TO ENSURE THAT THERE IS ONLY ONE PAGE IN A CHUNK,
+ // BECAUSE THE WHOLE IMPLEMENTATION IS BASED ON THIS ASSUMPTION.
+ // OTHERWISE, PAGEREADER IS FOR THE FIRST PAGE IN THE CHUNK WHILE
+ // STEPREGRESS IS FOR THE LAST PAGE IN THE CHUNK (THE MERGE OF STEPREGRESS IS ASSIGN
+ // DIRECTLY), WHICH WILL INTRODUCE BUGS!
+ chunkSuit4CPV.setPageReader(pageReader);
+ } else {
+ // Note the pass of delete intervals, especially deleting the non-latest candidate
+ // point.
+ // pageReader does not refer to the same deleteInterval as those in chunkMetadata
+ // after chunkMetadata executes insertIntoSortedDeletions
+ chunkSuit4CPV
+ .getPageReader()
+ .setDeleteIntervalList(chunkSuit4CPV.getChunkMetadata().getDeleteIntervalList());
+ }
+ // chunk data read operation (c)
+ // chunkSuit4CPV.getPageReader().updateBPTP(chunkSuit4CPV);
+ chunkSuit4CPV.getPageReader().updateBP_withValueIndex(chunkSuit4CPV);
+ // check if empty
+ if (chunkSuit4CPV.statistics.getCount() == 0) {
+ currentChunkList.remove(chunkSuit4CPV);
+ }
+ }
+ break; // exit loop 2, enter loop 1
+ }
+ // otherwise, extract the next new candidate point from the candidate set with the lazy load
+ // strategy
+ ChunkSuit4CPV candidate = nonLazyLoad.get(0); // sorted by version
+ MergeReaderPriority candidateVersion =
+ new MergeReaderPriority(
+ candidate.getChunkMetadata().getVersion(),
+ candidate.getChunkMetadata().getOffsetOfChunkHeader());
+ long candidateTimestamp = candidate.getStatistics().getBottomTimestamp(); // check
+ Object candidateValue = candidate.getStatistics().getMinValue(); // check
+
+ // verify if this candidate point is deleted
+ boolean isDeletedItself = false;
+ if (candidateTimestamp < curStartTime || candidateTimestamp >= curStartTime + interval) {
+ isDeletedItself = true;
+ } else {
+ isDeletedItself =
+ PageReader.isDeleted(
+ candidateTimestamp, candidate.getChunkMetadata().getDeleteIntervalList());
+ }
+ if (isDeletedItself) {
+ // the candidate point is deleted, then label the chunk as already lazy loaded, and back
+ // to loop 2
+ nonLazyLoad.remove(candidate);
+ // check this can really remove the element
+ // check whether nonLazyLoad remove affects candidateSet
+ // check nonLazyLoad sorted by version number from high to low
+ continue; // back to loop 2
+
+ } else { // not deleted
+ boolean isUpdate = false;
+ // find overlapping chunks with higher versions
+ List<ChunkSuit4CPV> overlaps = new ArrayList<>();
+ for (ChunkSuit4CPV chunkSuit4CPV : currentChunkList) {
+ ChunkMetadata chunkMetadata = chunkSuit4CPV.getChunkMetadata();
+ MergeReaderPriority version =
+ new MergeReaderPriority(
+ chunkMetadata.getVersion(), chunkMetadata.getOffsetOfChunkHeader());
+ if (version.compareTo(candidateVersion) <= 0) { // including bottomChunkMetadata
+ continue;
+ }
+ if (candidateTimestamp < chunkSuit4CPV.getStatistics().getStartTime()
+ || candidateTimestamp > chunkSuit4CPV.getStatistics().getEndTime()) {
+ continue;
+ }
+ if (candidateTimestamp == chunkSuit4CPV.getStatistics().getStartTime()
+ || candidateTimestamp == chunkSuit4CPV.getStatistics().getEndTime()) {
+ isUpdate = true;
+ // this case does not need to execute chunk data read operation (a),
+ // because definitely overwrite
+ break;
+ }
+ overlaps.add(chunkSuit4CPV);
+ }
+
+ if (!isUpdate && overlaps.size() == 0) {
+ // no overlaps, then the candidate point is not updated, then it is the final result
+ results
+ .get(0)
+ .updateResultUsingValues(
+ new long[] {candidateTimestamp}, 1, new Object[] {candidateValue});
+ return; // finished
+ } else if (!isUpdate) {
+ // verify whether the candidate point is updated
+ for (ChunkSuit4CPV chunkSuit4CPV : overlaps) {
+ if (chunkSuit4CPV.getPageReader() == null) {
+ PageReader pageReader =
+ FileLoaderUtils.loadPageReaderList4CPV(
+ chunkSuit4CPV.getChunkMetadata(), this.timeFilter);
+ // ATTENTION: YOU HAVE TO ENSURE THAT THERE IS ONLY ONE PAGE IN A CHUNK,
+ // BECAUSE THE WHOLE IMPLEMENTATION IS BASED ON THIS ASSUMPTION.
+ // OTHERWISE, PAGEREADER IS FOR THE FIRST PAGE IN THE CHUNK WHILE
+ // STEPREGRESS IS FOR THE LAST PAGE IN THE CHUNK (THE MERGE OF STEPREGRESS IS
+ // ASSIGN DIRECTLY), WHICH WILL INTRODUCE BUGS!
+ chunkSuit4CPV.setPageReader(pageReader);
+ }
+ // chunk data read operation (a): check existence of data point at a timestamp
+ isUpdate = chunkSuit4CPV.checkIfExist(candidateTimestamp);
+ if (isUpdate) {
+ // since the candidate point is updated, early break
+ break;
+ }
+ }
+ }
+ if (!isUpdate) {
+ // the candidate point is not updated, then it is the final result
+ results
+ .get(0)
+ .updateResultUsingValues(
+ new long[] {candidateTimestamp}, 1, new Object[] {candidateValue});
+ return; // finished
+ } else {
+ // the candidate point is updated, then label the chunk as already lazy loaded,
+ // add the deletion of the candidate point in deleteInterval, and back to loop 2
+ if (candidate.getChunkMetadata().getDeleteIntervalList() == null) {
+ List<TimeRange> tmp = new ArrayList<>();
+ tmp.add(new TimeRange(candidateTimestamp, candidateTimestamp));
+ candidate.getChunkMetadata().setDeleteIntervalList(tmp);
+ } else {
+ candidate
+ .getChunkMetadata()
+ .insertIntoSortedDeletions(candidateTimestamp, candidateTimestamp); // check
+ }
+ nonLazyLoad.remove(candidate);
+ // check this can really remove the element
+ // check whether nonLazyLoad remove affects candidateSet
+ // check nonLazyLoad sorted by version number from high to low
+ continue; // back to loop 2
+ }
+ }
+ }
+ }
+ }
+
+ private void calculateTopPoint(
+ List<ChunkSuit4CPV> currentChunkList,
+ long startTime,
+ long endTime,
+ long interval,
+ long curStartTime)
+ throws IOException {
+ // IOMonitor2.M4_LSM_status = Operation.M4_LSM_TP;
+ // check size>0 because after updateBPTP empty ChunkSuit4CPV will be removed from
+ // currentChunkList
+ while (currentChunkList.size() > 0) { // loop 1
+ // sorted by topValue, find TP candidate set
+ currentChunkList.sort(
+ new Comparator<ChunkSuit4CPV>() { // double check the sort order logic for different
+ // aggregations
+ public int compare(ChunkSuit4CPV o1, ChunkSuit4CPV o2) {
+ return ((Comparable) (o2.getStatistics().getMaxValue()))
+ .compareTo(o1.getStatistics().getMaxValue());
+ // NOTE here get statistics from ChunkSuit4CPV, not from ChunkSuit4CPV.ChunkMetadata,
+ // because statistics of ChunkSuit4CPV is updated, while statistics of
+ // ChunkSuit4CPV.ChunkMetadata
+ // is fixed.
+ }
+ });
+ // NOTE here get statistics from ChunkSuit4CPV, not from ChunkSuit4CPV.ChunkMetadata
+ Object value = currentChunkList.get(0).getStatistics().getMaxValue();
+ List<ChunkSuit4CPV> candidateSet = new ArrayList<>();
+ for (ChunkSuit4CPV chunkSuit4CPV : currentChunkList) {
+ // NOTE here get statistics from ChunkSuit4CPV, not from ChunkSuit4CPV.ChunkMetadata
+ if (chunkSuit4CPV.getStatistics().getMaxValue().equals(value)) {
+ candidateSet.add(chunkSuit4CPV);
+ } else {
+ break; // note that this is an early break since currentChunkList is sorted
+ }
+ }
+
+ List<ChunkSuit4CPV> nonLazyLoad = new ArrayList<>(candidateSet);
+ // check, whether nonLazyLoad remove affects candidateSet
+ nonLazyLoad.sort(
+ new Comparator<ChunkSuit4CPV>() { // double check the sort order logic for version
+ public int compare(ChunkSuit4CPV o1, ChunkSuit4CPV o2) {
+ return new MergeReaderPriority(
+ o2.getChunkMetadata().getVersion(),
+ o2.getChunkMetadata().getOffsetOfChunkHeader())
+ .compareTo(
+ new MergeReaderPriority(
+ o1.getChunkMetadata().getVersion(),
+ o1.getChunkMetadata().getOffsetOfChunkHeader()));
+ }
+ });
+ while (true) { // loop 2
+ // if there is no chunk for lazy loading, then load all chunks in candidateSet,
+ // and apply deleteIntervals, deleting TP no matter out of deletion or update
+ if (nonLazyLoad.size() == 0) {
+ for (ChunkSuit4CPV chunkSuit4CPV : candidateSet) {
+ // Note the pass of delete intervals
+ if (chunkSuit4CPV.getPageReader() == null) {
+ PageReader pageReader =
+ FileLoaderUtils.loadPageReaderList4CPV(
+ chunkSuit4CPV.getChunkMetadata(), this.timeFilter);
+ // ATTENTION: YOU HAVE TO ENSURE THAT THERE IS ONLY ONE PAGE IN A CHUNK,
+ // BECAUSE THE WHOLE IMPLEMENTATION IS BASED ON THIS ASSUMPTION.
+ // OTHERWISE, PAGEREADER IS FOR THE FIRST PAGE IN THE CHUNK WHILE
+ // STEPREGRESS IS FOR THE LAST PAGE IN THE CHUNK (THE MERGE OF STEPREGRESS IS ASSIGN
+ // DIRECTLY), WHICH WILL INTRODUCE BUGS!
+ chunkSuit4CPV.setPageReader(pageReader);
+ } else {
+ // Note the pass of delete intervals, especially deleting the non-latest candidate
+ // point.
+ // pageReader does not refer to the same deleteInterval as those in chunkMetadata
+ // after chunkMetadata executes insertIntoSortedDeletions
+ chunkSuit4CPV
+ .getPageReader()
+ .setDeleteIntervalList(chunkSuit4CPV.getChunkMetadata().getDeleteIntervalList());
+ }
+ // chunk data read operation (c)
+ // chunkSuit4CPV.getPageReader().updateBPTP(chunkSuit4CPV);
+ chunkSuit4CPV.getPageReader().updateTP_withValueIndex(chunkSuit4CPV); //
+ // check if empty
+ if (chunkSuit4CPV.statistics.getCount() == 0) {
+ currentChunkList.remove(chunkSuit4CPV);
+ }
+ }
+ break; // exit loop 2, enter loop 1
+ }
+ // otherwise, extract the next new candidate point from the candidate set with the lazy load
+ // strategy
+ ChunkSuit4CPV candidate = nonLazyLoad.get(0); // sorted by version
+ MergeReaderPriority candidateVersion =
+ new MergeReaderPriority(
+ candidate.getChunkMetadata().getVersion(),
+ candidate.getChunkMetadata().getOffsetOfChunkHeader());
+ // NOTE here get statistics from ChunkSuit4CPV, not from ChunkSuit4CPV.ChunkMetadata,
+ // because statistics of ChunkSuit4CPV is updated, while statistics of
+ // ChunkSuit4CPV.ChunkMetadata
+ // is fixed.
+ long candidateTimestamp = candidate.getStatistics().getTopTimestamp(); // check
+ Object candidateValue = candidate.getStatistics().getMaxValue(); // check
+
+ // verify if this candidate point is deleted
+ boolean isDeletedItself = false;
+ if (candidateTimestamp < curStartTime || candidateTimestamp >= curStartTime + interval) {
+ isDeletedItself = true;
+ } else {
+ isDeletedItself =
+ PageReader.isDeleted(
+ candidateTimestamp, candidate.getChunkMetadata().getDeleteIntervalList());
+ }
+ if (isDeletedItself) {
+ // the candidate point is deleted, then label the chunk as already lazy loaded, and back
+ // to loop 2
+ nonLazyLoad.remove(candidate);
+ // check this can really remove the element
+ // check whether nonLazyLoad remove affects candidateSet
+ // check nonLazyLoad sorted by version number from high to low
+ continue; // back to loop 2
+
+ } else { // not deleted
+ boolean isUpdate = false;
+ // find overlapping chunks with higher versions
+ List<ChunkSuit4CPV> overlaps = new ArrayList<>();
+ for (ChunkSuit4CPV chunkSuit4CPV : currentChunkList) {
+ ChunkMetadata chunkMetadata = chunkSuit4CPV.getChunkMetadata();
+ MergeReaderPriority version =
+ new MergeReaderPriority(
+ chunkMetadata.getVersion(), chunkMetadata.getOffsetOfChunkHeader());
+ if (version.compareTo(candidateVersion) <= 0) { // including topChunkMetadata
+ continue;
+ }
+ if (candidateTimestamp < chunkMetadata.getStartTime()
+ || candidateTimestamp > chunkMetadata.getEndTime()) {
+ continue;
+ }
+ if (candidateTimestamp == chunkSuit4CPV.getStatistics().getStartTime()
+ || candidateTimestamp == chunkSuit4CPV.getStatistics().getEndTime()) {
+ isUpdate = true; // note that here overlaps does not add.
+ // this case does not need to execute chunk data read operation (a),
+ // because definitely overwrite
+ break;
+ }
+ overlaps.add(chunkSuit4CPV);
+ }
+
+ if (!isUpdate && overlaps.size() == 0) {
+ // no overlaps, then the candidate point is not updated, then it is the final result
+ results
+ .get(1)
+ .updateResultUsingValues(
+ new long[] {candidateTimestamp}, 1, new Object[] {candidateValue});
+ return; // finished
+ } else if (!isUpdate) {
+ // verify whether the candidate point is updated
+ for (ChunkSuit4CPV chunkSuit4CPV : overlaps) {
+ if (chunkSuit4CPV.getPageReader() == null) {
+ PageReader pageReader =
+ FileLoaderUtils.loadPageReaderList4CPV(
+ chunkSuit4CPV.getChunkMetadata(), this.timeFilter);
+ // ATTENTION: YOU HAVE TO ENSURE THAT THERE IS ONLY ONE PAGE IN A CHUNK,
+ // BECAUSE THE WHOLE IMPLEMENTATION IS BASED ON THIS ASSUMPTION.
+ // OTHERWISE, PAGEREADER IS FOR THE FIRST PAGE IN THE CHUNK WHILE
+ // STEPREGRESS IS FOR THE LAST PAGE IN THE CHUNK (THE MERGE OF STEPREGRESS IS
+ // ASSIGN DIRECTLY), WHICH WILL INTRODUCE BUGS!
+ chunkSuit4CPV.setPageReader(pageReader);
+ }
+ // chunk data read operation (a): check existence of data point at a timestamp
+ isUpdate = chunkSuit4CPV.checkIfExist(candidateTimestamp);
+ if (isUpdate) {
+ // since the candidate point is updated, early break
+ break;
+ }
+ }
+ }
+ if (!isUpdate) {
+ // the candidate point is not updated, then it is the final result
+ results
+ .get(1)
+ .updateResultUsingValues(
+ new long[] {candidateTimestamp}, 1, new Object[] {candidateValue});
+ return; // finished
+ } else {
+ // the candidate point is updated, then label the chunk as already lazy loaded,
+ // add the deletion of the candidate point in deleteInterval, and back to loop 2
+ if (candidate.getChunkMetadata().getDeleteIntervalList() == null) {
+ List<TimeRange> tmp = new ArrayList<>();
+ tmp.add(new TimeRange(candidateTimestamp, candidateTimestamp));
+ candidate.getChunkMetadata().setDeleteIntervalList(tmp);
+ } else {
+ candidate
+ .getChunkMetadata()
+ .insertIntoSortedDeletions(candidateTimestamp, candidateTimestamp); // check
+ }
+ nonLazyLoad.remove(candidate);
+ // check this can really remove the element
+ // check whether nonLazyLoad remove affects candidateSet
+ // check nonLazyLoad sorted by version number from high to low
+ continue; // back to loop 2
+ }
+ }
+ }
+ }
+ }
+
+ private void calculateFirstPoint(
+ List<ChunkSuit4CPV> currentChunkList,
+ long startTime,
+ long endTime,
+ long interval,
+ long curStartTime)
+ throws IOException {
+ // IOMonitor2.M4_LSM_status = Operation.M4_LSM_FP;
+ while (currentChunkList.size() > 0) { // loop 1
+ // sorted by startTime and version, find FP candidate
+ currentChunkList.sort(
+ new Comparator<ChunkSuit4CPV>() { // double check the sort order logic for different
+ // aggregations
+ public int compare(ChunkSuit4CPV o1, ChunkSuit4CPV o2) {
+ // NOTE here get statistics from ChunkSuit4CPV, not from ChunkSuit4CPV.ChunkMetadata
+ int res =
+ ((Comparable) (o1.getStatistics().getStartTime()))
+ .compareTo(o2.getStatistics().getStartTime());
+ if (res != 0) {
+ return res;
+ } else {
+ return new MergeReaderPriority(
+ o2.getChunkMetadata().getVersion(),
+ o2.getChunkMetadata().getOffsetOfChunkHeader())
+ .compareTo(
+ new MergeReaderPriority(
+ o1.getChunkMetadata().getVersion(),
+ o1.getChunkMetadata().getOffsetOfChunkHeader()));
+ }
+ }
+ });
+
+ ChunkSuit4CPV susp_candidate = currentChunkList.get(0);
+ if (susp_candidate.isLazyLoad()) {
+ // means the chunk is already lazy loaded, then load the chunk, apply deletes, update
+ // statistics,
+ // cancel the lazy loaded mark, and back to loop 1
+ if (susp_candidate.getPageReader() == null) {
+ PageReader pageReader =
+ FileLoaderUtils.loadPageReaderList4CPV(
+ susp_candidate.getChunkMetadata(), this.timeFilter);
+ // ATTENTION: YOU HAVE TO ENSURE THAT THERE IS ONLY ONE PAGE IN A CHUNK,
+ // BECAUSE THE WHOLE IMPLEMENTATION IS BASED ON THIS ASSUMPTION.
+ // OTHERWISE, PAGEREADER IS FOR THE FIRST PAGE IN THE CHUNK WHILE
+ // STEPREGRESS IS FOR THE LAST PAGE IN THE CHUNK (THE MERGE OF STEPREGRESS IS ASSIGN
+ // DIRECTLY), WHICH WILL INTRODUCE BUGS!
+ susp_candidate.setPageReader(pageReader);
+ }
+ // chunk data read operation (b): get the closest data point after or before a timestamp
+ susp_candidate.updateFPwithTheClosetPointEqualOrAfter(
+ susp_candidate.getStatistics().getStartTime());
+ susp_candidate.setLazyLoad(false); // DO NOT FORGET THIS!!!
+ continue; // back to loop 1
+ } else {
+ // the chunk has not been lazy loaded, then verify whether the candidate point is deleted
+ // Note the higher versions of deletes are guaranteed by
+ // QueryUtils.modifyChunkMetaData(chunkMetadataList,pathModifications)
+ // NOTE here get statistics from ChunkSuit4CPV, not from ChunkSuit4CPV.ChunkMetadata
+ long candidateTimestamp = susp_candidate.getStatistics().getStartTime(); // check
+ Object candidateValue = susp_candidate.getStatistics().getFirstValue(); // check
+
+ boolean isDeletedItself = false;
+ long deleteEndTime = -1;
+ List<TimeRange> deleteIntervalList =
+ susp_candidate.getChunkMetadata().getDeleteIntervalList();
+ if (deleteIntervalList != null) {
+ int deleteCursor = 0;
+ while (deleteCursor < deleteIntervalList.size()) {
+ if (deleteIntervalList.get(deleteCursor).getMax() < candidateTimestamp) {
+ deleteCursor++;
+ } else if (deleteIntervalList.get(deleteCursor).contains(candidateTimestamp)) {
+ isDeletedItself = true;
+ deleteEndTime = deleteIntervalList.get(deleteCursor).getMax();
+ break; // since delete intervals are already sorted and merged
+ } else {
+ break; // since delete intervals are already sorted and merged
+ }
+ }
+ }
+ if (isDeletedItself) {
+ // deleteEndTime may be after the current endTime,
+ // because deleteStartTime can be after the startTime of the whole chunk
+ if (deleteEndTime
+ >= susp_candidate.getStatistics().getEndTime()) { // NOTE here calculate FP
+ // deleted as a whole
+ currentChunkList.remove(susp_candidate);
+ } else {
+ // the candidate point is deleted, then label the chunk as already lazy loaded,
+ // update chunkStartTime without loading data, and back to loop 1
+ susp_candidate.setLazyLoad(true);
+ // NOTE here get statistics from ChunkSuit4CPV, not from ChunkSuit4CPV.ChunkMetadata
+ susp_candidate.getStatistics().setStartTime(deleteEndTime + 1); // check
+ // +1 is because delete is closed interval
+ }
+ continue; // back to loop 1
+ } else {
+ // the candidate point is not deleted, then it is the final result
+ results
+ .get(0)
+ .updateResultUsingValues(
+ new long[] {candidateTimestamp}, 1, new Object[] {candidateValue});
+ results
+ .get(2)
+ .updateResultUsingValues(
+ new long[] {candidateTimestamp}, 1, new Object[] {candidateValue});
+ return;
+ }
+ }
+ }
+ }
+
+ private void calculateLastPoint(
+ List<ChunkSuit4CPV> currentChunkList,
+ long startTime,
+ long endTime,
+ long interval,
+ long curStartTime)
+ throws IOException {
+ // IOMonitor2.M4_LSM_status = Operation.M4_LSM_LP;
+ while (currentChunkList.size() > 0) { // loop 1
+ // sorted by endTime and version, find LP candidate
+ currentChunkList.sort(
+ new Comparator<ChunkSuit4CPV>() {
+ // aggregations
+ public int compare(ChunkSuit4CPV o1, ChunkSuit4CPV o2) {
+ int res =
+ ((Comparable) (o2.getStatistics().getEndTime()))
+ .compareTo(o1.getStatistics().getEndTime());
+ // NOTE here get statistics from ChunkSuit4CPV, not from ChunkSuit4CPV.ChunkMetadata
+ if (res != 0) {
+ return res;
+ } else {
+ return new MergeReaderPriority(
+ o2.getChunkMetadata().getVersion(),
+ o2.getChunkMetadata().getOffsetOfChunkHeader())
+ .compareTo(
+ new MergeReaderPriority(
+ o1.getChunkMetadata().getVersion(),
+ o1.getChunkMetadata().getOffsetOfChunkHeader()));
+ }
+ }
+ });
+
+ ChunkSuit4CPV susp_candidate = currentChunkList.get(0);
+ if (susp_candidate.isLazyLoad()) {
+ // means the chunk is already lazy loaded, then load the chunk, apply deletes, update
+ // statistics,
+ // cancel the lazy loaded mark, and back to loop 1
+ if (susp_candidate.getPageReader() == null) {
+ PageReader pageReader =
+ FileLoaderUtils.loadPageReaderList4CPV(
+ susp_candidate.getChunkMetadata(), this.timeFilter);
+ // ATTENTION: YOU HAVE TO ENSURE THAT THERE IS ONLY ONE PAGE IN A CHUNK,
+ // BECAUSE THE WHOLE IMPLEMENTATION IS BASED ON THIS ASSUMPTION.
+ // OTHERWISE, PAGEREADER IS FOR THE FIRST PAGE IN THE CHUNK WHILE
+ // STEPREGRESS IS FOR THE LAST PAGE IN THE CHUNK (THE MERGE OF STEPREGRESS IS ASSIGN
+ // DIRECTLY), WHICH WILL INTRODUCE BUGS!
+ susp_candidate.setPageReader(pageReader);
+ }
+ // update LP equal to or before statistics.getEndTime
+ // (b) get the closest data point after or before a timestamp
+ susp_candidate.updateLPwithTheClosetPointEqualOrBefore(
+ susp_candidate.getStatistics().getEndTime()); // DEBUG
+ susp_candidate.setLazyLoad(false); // DO NOT FORGET THIS!!!
+ continue; // back to loop 1
+ } else {
+ // the chunk has not been lazy loaded, then verify whether the candidate point is deleted
+ // Note the higher versions of deletes are guaranteed by
+ // QueryUtils.modifyChunkMetaData(chunkMetadataList,pathModifications)
+ // NOTE here get statistics from ChunkSuit4CPV, not from ChunkSuit4CPV.ChunkMetadata
+ long candidateTimestamp = susp_candidate.getStatistics().getEndTime(); // check
+ Object candidateValue = susp_candidate.getStatistics().getLastValue(); // check
+
+ boolean isDeletedItself = false;
+ long deleteStartTime = Long.MAX_VALUE; // check
+ List<TimeRange> deleteIntervalList =
+ susp_candidate.getChunkMetadata().getDeleteIntervalList();
+ if (deleteIntervalList != null) {
+ int deleteCursor = 0;
+ while (deleteCursor < deleteIntervalList.size()) {
+ if (deleteIntervalList.get(deleteCursor).getMax() < candidateTimestamp) {
+ deleteCursor++;
+ } else if (deleteIntervalList.get(deleteCursor).contains(candidateTimestamp)) {
+ isDeletedItself = true;
+ deleteStartTime = deleteIntervalList.get(deleteCursor).getMin();
+ break; // since delete intervals are already sorted and merged
+ } else {
+ break; // since delete intervals are already sorted and merged
+ }
+ }
+ }
+ if (isDeletedItself) {
+ // deleteStartTime may be before the current startTime,
+ // because deleteEndTime can be before the endTime of the whole chunk
+ if (deleteStartTime <= susp_candidate.getStatistics().getStartTime()) {
+ // NOTE here calculate LP.
+ // deleted as a whole
+ currentChunkList.remove(susp_candidate);
+ } else {
+ susp_candidate.setLazyLoad(true);
+ // NOTE here get statistics from ChunkSuit4CPV, not from
+ // ChunkSuit4CPV.ChunkMetadata
+ susp_candidate.getStatistics().setEndTime(deleteStartTime - 1);
+ // -1 is because delete is closed interval
+ // check
+ }
+ continue; // back to loop 1
+ } else {
+ // the candidate point is not deleted, then it is the final result
+ results
+ .get(1)
+ .updateResultUsingValues(
+ new long[] {candidateTimestamp}, 1, new Object[] {candidateValue});
+ results
+ .get(3)
+ .updateResultUsingValues(
+ new long[] {candidateTimestamp}, 1, new Object[] {candidateValue});
+ return;
+ }
+ }
+ }
+ }
+
+ @Override
+ public Pair<Long, Object> peekNextNotNullValue(long nextStartTime, long nextEndTime)
+ throws IOException {
+ throw new IOException("no implemented");
+ }
+
+ @Override
+ public List<AggregateResult> calcResult(long curStartTime, long curEndTime)
+ throws IOException, QueryProcessException {
+ throw new IOException("no implemented");
+ }
+
+ public List<ChunkSuit4CPV> getCurrentChunkList() {
+ return currentChunkList;
+ }
+
+ public List<ChunkSuit4CPV> getFutureChunkList() {
+ return futureChunkList;
+ }
+}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/common/conf/TSFileConfig.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/common/conf/TSFileConfig.java
index 337f990d42f..c745982c5a4 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/common/conf/TSFileConfig.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/common/conf/TSFileConfig.java
@@ -27,6 +27,8 @@ import java.nio.charset.Charset;
/** TSFileConfig is a configure class. Every variables is public and has default value. */
public class TSFileConfig implements Serializable {
+ private boolean enableMinMaxLSM = false;
+
private boolean useStatistics = true;
private boolean useTimeIndex = true;
@@ -157,6 +159,14 @@ public class TSFileConfig implements Serializable {
public TSFileConfig() {}
+ public boolean isEnableMinMaxLSM() {
+ return enableMinMaxLSM;
+ }
+
+ public void setEnableMinMaxLSM(boolean enableMinMaxLSM) {
+ this.enableMinMaxLSM = enableMinMaxLSM;
+ }
+
public boolean isUseStatistics() {
return useStatistics;
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/common/conf/TSFileDescriptor.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/common/conf/TSFileDescriptor.java
index e4d874e4e76..3aa322b4e89 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/common/conf/TSFileDescriptor.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/common/conf/TSFileDescriptor.java
@@ -105,6 +105,10 @@ public class TSFileDescriptor {
Properties properties = new Properties();
try {
properties.load(inputStream);
+ conf.setEnableMinMaxLSM(
+ Boolean.parseBoolean(
+ properties.getProperty(
+ "enableMinMaxLSM", Boolean.toString(conf.isEnableMinMaxLSM()))));
conf.setUseStatistics(
Boolean.parseBoolean(
properties.getProperty("use_Statistics", Boolean.toString(conf.isUseStatistics()))));
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/IOMonitor2.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/IOMonitor2.java
index fd160b3fd6f..c20a9de8994 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/IOMonitor2.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/IOMonitor2.java
@@ -30,6 +30,8 @@ public class IOMonitor2 {
GroupByWithoutValueFilterDataSet_LocalGroupByExecutor4CPV_NoTimeIndex, // cpv_noTimeIndex
GroupByWithoutValueFilterDataSet_LocalGroupByExecutor4CPV_NoValueIndex, // cpv_noValueIndex
GroupByWithoutValueFilterDataSet_LocalGroupByExecutor4CPV_NoTimeValueIndex, // cpv_noTimeValueIndex
+
+ GroupByWithoutValueFilterDataSet_LocalGroupByExecutor4CPV_EnableMinMaxLSM, // MinMax-LSM
GroupByWithoutValueFilterDataSet_LocalGroupByExecutor_UseStatistics, // moc
GroupByWithoutValueFilterDataSet_LocalGroupByExecutor_NotUseStatistics // mac_groupBy
}
[iotdb] 09/09: update
Posted by le...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
leirui pushed a commit to branch research/M4-visualization
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 46963054960006e020a4ba3d1886ef95a9390e64
Author: Lei Rui <10...@qq.com>
AuthorDate: Wed Oct 18 16:27:35 2023 +0800
update
---
.../groupby/LocalGroupByExecutor4MinMax.java | 311 +++++----------------
.../apache/iotdb/db/integration/m4/MyTest8.java | 3 +
.../iotdb/tsfile/read/common/ChunkSuit4CPV.java | 1 +
3 files changed, 72 insertions(+), 243 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor4MinMax.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor4MinMax.java
index ff074f588cb..44f29f9d897 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor4MinMax.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor4MinMax.java
@@ -266,30 +266,30 @@ public class LocalGroupByExecutor4MinMax implements GroupByExecutor {
currentChunkList.add(chunkSuit4CPV);
itr.remove();
} else {
+ // TODO modify logic
// the chunk partially overlaps in time with the current M4 interval Ii.
// load this chunk, split it on deletes and all w intervals.
// add to currentChunkList and futureChunkList.
itr.remove();
- // B: loads chunk data from disk to memory
- // C: decompress page data, split time&value buffers
- PageReader pageReader =
- FileLoaderUtils.loadPageReaderList4CPV(
- chunkSuit4CPV.getChunkMetadata(), this.timeFilter);
- // ATTENTION: YOU HAVE TO ENSURE THAT THERE IS ONLY ONE PAGE IN A CHUNK,
- // BECAUSE THE WHOLE IMPLEMENTATION IS BASED ON THIS ASSUMPTION.
- // OTHERWISE, PAGEREADER IS FOR THE FIRST PAGE IN THE CHUNK WHILE
- // STEPREGRESS IS FOR THE LAST PAGE IN THE CHUNK (THE MERGE OF STEPREGRESS IS ASSIGN
- // DIRECTLY), WHICH WILL INTRODUCE BUGS!
-
- // chunk data read operation (b) get the closest data point after or before a timestamp
- pageReader.split4CPV(
- startTime,
- endTime,
- interval,
- curStartTime,
- currentChunkList,
- splitChunkList,
- chunkMetadata);
+ int numberOfSpans =
+ (int)
+ Math.floor(
+ (Math.min(chunkMetadata.getEndTime(), endTime - 1) - curStartTime)
+ * 1.0
+ / interval)
+ + 1;
+ for (int n = 0; n < numberOfSpans; n++) {
+ ChunkSuit4CPV newChunkSuit4CPV = new ChunkSuit4CPV(chunkMetadata, null, true);
+ newChunkSuit4CPV.needsUpdateStartEndPos =
+ true; // note this, because this chunk does not fall within a time span, but cross
+ if (n == 0) {
+ currentChunkList.add(newChunkSuit4CPV);
+ } else {
+ int globalIdx = curIdx + n; // note splitChunkList uses global idx key
+ splitChunkList.computeIfAbsent(globalIdx, k -> new ArrayList<>());
+ splitChunkList.get(globalIdx).add(newChunkSuit4CPV);
+ }
+ }
}
}
}
@@ -396,6 +396,30 @@ public class LocalGroupByExecutor4MinMax implements GroupByExecutor {
.getPageReader()
.setDeleteIntervalList(chunkSuit4CPV.getChunkMetadata().getDeleteIntervalList());
}
+ // TODO add logic
+ if (chunkSuit4CPV
+ .needsUpdateStartEndPos) { // otherwise the chunk falls completely with the current
+ // time span
+ chunkSuit4CPV.needsUpdateStartEndPos = false;
+ // update startPos & endPos by dealing with the current time span virtual deletes
+ long leftEndIncluded = curStartTime;
+ long rightEndExcluded = curStartTime + interval;
+ int FP_pos = -1;
+ int LP_pos = -1;
+ if (leftEndIncluded > chunkSuit4CPV.statistics.getStartTime()) {
+ FP_pos = chunkSuit4CPV.updateFPwithTheClosetPointEqualOrAfter(leftEndIncluded);
+ }
+ if (rightEndExcluded <= chunkSuit4CPV.statistics.getEndTime()) {
+ // -1 is because right end is excluded end
+ LP_pos =
+ chunkSuit4CPV.updateLPwithTheClosetPointEqualOrBefore(rightEndExcluded - 1);
+ }
+ if (FP_pos != -1 && LP_pos != -1 && FP_pos > LP_pos) {
+ // means the chunk has no point in this span
+ currentChunkList.remove(chunkSuit4CPV);
+ }
+ }
+
// chunk data read operation (c)
// chunkSuit4CPV.getPageReader().updateBPTP(chunkSuit4CPV);
chunkSuit4CPV.getPageReader().updateBP_withValueIndex(chunkSuit4CPV);
@@ -594,6 +618,30 @@ public class LocalGroupByExecutor4MinMax implements GroupByExecutor {
.getPageReader()
.setDeleteIntervalList(chunkSuit4CPV.getChunkMetadata().getDeleteIntervalList());
}
+ // TODO add logic
+ if (chunkSuit4CPV
+ .needsUpdateStartEndPos) { // otherwise the chunk falls completely with the current
+ // time span
+ chunkSuit4CPV.needsUpdateStartEndPos = false;
+ // update startPos & endPos by dealing with the current time span virtual deletes
+ long leftEndIncluded = curStartTime;
+ long rightEndExcluded = curStartTime + interval;
+ int FP_pos = -1;
+ int LP_pos = -1;
+ if (leftEndIncluded > chunkSuit4CPV.statistics.getStartTime()) {
+ FP_pos = chunkSuit4CPV.updateFPwithTheClosetPointEqualOrAfter(leftEndIncluded);
+ }
+ if (rightEndExcluded <= chunkSuit4CPV.statistics.getEndTime()) {
+ // -1 is because right end is excluded end
+ LP_pos =
+ chunkSuit4CPV.updateLPwithTheClosetPointEqualOrBefore(rightEndExcluded - 1);
+ }
+ if (FP_pos != -1 && LP_pos != -1 && FP_pos > LP_pos) {
+ // means the chunk has no point in this span
+ currentChunkList.remove(chunkSuit4CPV);
+ }
+ }
+
// chunk data read operation (c)
// chunkSuit4CPV.getPageReader().updateBPTP(chunkSuit4CPV);
chunkSuit4CPV.getPageReader().updateTP_withValueIndex(chunkSuit4CPV); //
@@ -721,229 +769,6 @@ public class LocalGroupByExecutor4MinMax implements GroupByExecutor {
}
}
- private void calculateFirstPoint(
- List<ChunkSuit4CPV> currentChunkList,
- long startTime,
- long endTime,
- long interval,
- long curStartTime)
- throws IOException {
- // IOMonitor2.M4_LSM_status = Operation.M4_LSM_FP;
- while (currentChunkList.size() > 0) { // loop 1
- // sorted by startTime and version, find FP candidate
- currentChunkList.sort(
- new Comparator<ChunkSuit4CPV>() { // double check the sort order logic for different
- // aggregations
- public int compare(ChunkSuit4CPV o1, ChunkSuit4CPV o2) {
- // NOTE here get statistics from ChunkSuit4CPV, not from ChunkSuit4CPV.ChunkMetadata
- int res =
- ((Comparable) (o1.getStatistics().getStartTime()))
- .compareTo(o2.getStatistics().getStartTime());
- if (res != 0) {
- return res;
- } else {
- return new MergeReaderPriority(
- o2.getChunkMetadata().getVersion(),
- o2.getChunkMetadata().getOffsetOfChunkHeader())
- .compareTo(
- new MergeReaderPriority(
- o1.getChunkMetadata().getVersion(),
- o1.getChunkMetadata().getOffsetOfChunkHeader()));
- }
- }
- });
-
- ChunkSuit4CPV susp_candidate = currentChunkList.get(0);
- if (susp_candidate.isLazyLoad()) {
- // means the chunk is already lazy loaded, then load the chunk, apply deletes, update
- // statistics,
- // cancel the lazy loaded mark, and back to loop 1
- if (susp_candidate.getPageReader() == null) {
- PageReader pageReader =
- FileLoaderUtils.loadPageReaderList4CPV(
- susp_candidate.getChunkMetadata(), this.timeFilter);
- // ATTENTION: YOU HAVE TO ENSURE THAT THERE IS ONLY ONE PAGE IN A CHUNK,
- // BECAUSE THE WHOLE IMPLEMENTATION IS BASED ON THIS ASSUMPTION.
- // OTHERWISE, PAGEREADER IS FOR THE FIRST PAGE IN THE CHUNK WHILE
- // STEPREGRESS IS FOR THE LAST PAGE IN THE CHUNK (THE MERGE OF STEPREGRESS IS ASSIGN
- // DIRECTLY), WHICH WILL INTRODUCE BUGS!
- susp_candidate.setPageReader(pageReader);
- }
- // chunk data read operation (b): get the closest data point after or before a timestamp
- susp_candidate.updateFPwithTheClosetPointEqualOrAfter(
- susp_candidate.getStatistics().getStartTime());
- susp_candidate.setLazyLoad(false); // DO NOT FORGET THIS!!!
- continue; // back to loop 1
- } else {
- // the chunk has not been lazy loaded, then verify whether the candidate point is deleted
- // Note the higher versions of deletes are guaranteed by
- // QueryUtils.modifyChunkMetaData(chunkMetadataList,pathModifications)
- // NOTE here get statistics from ChunkSuit4CPV, not from ChunkSuit4CPV.ChunkMetadata
- long candidateTimestamp = susp_candidate.getStatistics().getStartTime(); // check
- Object candidateValue = susp_candidate.getStatistics().getFirstValue(); // check
-
- boolean isDeletedItself = false;
- long deleteEndTime = -1;
- List<TimeRange> deleteIntervalList =
- susp_candidate.getChunkMetadata().getDeleteIntervalList();
- if (deleteIntervalList != null) {
- int deleteCursor = 0;
- while (deleteCursor < deleteIntervalList.size()) {
- if (deleteIntervalList.get(deleteCursor).getMax() < candidateTimestamp) {
- deleteCursor++;
- } else if (deleteIntervalList.get(deleteCursor).contains(candidateTimestamp)) {
- isDeletedItself = true;
- deleteEndTime = deleteIntervalList.get(deleteCursor).getMax();
- break; // since delete intervals are already sorted and merged
- } else {
- break; // since delete intervals are already sorted and merged
- }
- }
- }
- if (isDeletedItself) {
- // deleteEndTime may be after the current endTime,
- // because deleteStartTime can be after the startTime of the whole chunk
- if (deleteEndTime
- >= susp_candidate.getStatistics().getEndTime()) { // NOTE here calculate FP
- // deleted as a whole
- currentChunkList.remove(susp_candidate);
- } else {
- // the candidate point is deleted, then label the chunk as already lazy loaded,
- // update chunkStartTime without loading data, and back to loop 1
- susp_candidate.setLazyLoad(true);
- // NOTE here get statistics from ChunkSuit4CPV, not from ChunkSuit4CPV.ChunkMetadata
- susp_candidate.getStatistics().setStartTime(deleteEndTime + 1); // check
- // +1 is because delete is closed interval
- }
- continue; // back to loop 1
- } else {
- // the candidate point is not deleted, then it is the final result
- results
- .get(0)
- .updateResultUsingValues(
- new long[] {candidateTimestamp}, 1, new Object[] {candidateValue});
- results
- .get(2)
- .updateResultUsingValues(
- new long[] {candidateTimestamp}, 1, new Object[] {candidateValue});
- return;
- }
- }
- }
- }
-
- private void calculateLastPoint(
- List<ChunkSuit4CPV> currentChunkList,
- long startTime,
- long endTime,
- long interval,
- long curStartTime)
- throws IOException {
- // IOMonitor2.M4_LSM_status = Operation.M4_LSM_LP;
- while (currentChunkList.size() > 0) { // loop 1
- // sorted by endTime and version, find LP candidate
- currentChunkList.sort(
- new Comparator<ChunkSuit4CPV>() {
- // aggregations
- public int compare(ChunkSuit4CPV o1, ChunkSuit4CPV o2) {
- int res =
- ((Comparable) (o2.getStatistics().getEndTime()))
- .compareTo(o1.getStatistics().getEndTime());
- // NOTE here get statistics from ChunkSuit4CPV, not from ChunkSuit4CPV.ChunkMetadata
- if (res != 0) {
- return res;
- } else {
- return new MergeReaderPriority(
- o2.getChunkMetadata().getVersion(),
- o2.getChunkMetadata().getOffsetOfChunkHeader())
- .compareTo(
- new MergeReaderPriority(
- o1.getChunkMetadata().getVersion(),
- o1.getChunkMetadata().getOffsetOfChunkHeader()));
- }
- }
- });
-
- ChunkSuit4CPV susp_candidate = currentChunkList.get(0);
- if (susp_candidate.isLazyLoad()) {
- // means the chunk is already lazy loaded, then load the chunk, apply deletes, update
- // statistics,
- // cancel the lazy loaded mark, and back to loop 1
- if (susp_candidate.getPageReader() == null) {
- PageReader pageReader =
- FileLoaderUtils.loadPageReaderList4CPV(
- susp_candidate.getChunkMetadata(), this.timeFilter);
- // ATTENTION: YOU HAVE TO ENSURE THAT THERE IS ONLY ONE PAGE IN A CHUNK,
- // BECAUSE THE WHOLE IMPLEMENTATION IS BASED ON THIS ASSUMPTION.
- // OTHERWISE, PAGEREADER IS FOR THE FIRST PAGE IN THE CHUNK WHILE
- // STEPREGRESS IS FOR THE LAST PAGE IN THE CHUNK (THE MERGE OF STEPREGRESS IS ASSIGN
- // DIRECTLY), WHICH WILL INTRODUCE BUGS!
- susp_candidate.setPageReader(pageReader);
- }
- // update LP equal to or before statistics.getEndTime
- // (b) get the closest data point after or before a timestamp
- susp_candidate.updateLPwithTheClosetPointEqualOrBefore(
- susp_candidate.getStatistics().getEndTime()); // DEBUG
- susp_candidate.setLazyLoad(false); // DO NOT FORGET THIS!!!
- continue; // back to loop 1
- } else {
- // the chunk has not been lazy loaded, then verify whether the candidate point is deleted
- // Note the higher versions of deletes are guaranteed by
- // QueryUtils.modifyChunkMetaData(chunkMetadataList,pathModifications)
- // NOTE here get statistics from ChunkSuit4CPV, not from ChunkSuit4CPV.ChunkMetadata
- long candidateTimestamp = susp_candidate.getStatistics().getEndTime(); // check
- Object candidateValue = susp_candidate.getStatistics().getLastValue(); // check
-
- boolean isDeletedItself = false;
- long deleteStartTime = Long.MAX_VALUE; // check
- List<TimeRange> deleteIntervalList =
- susp_candidate.getChunkMetadata().getDeleteIntervalList();
- if (deleteIntervalList != null) {
- int deleteCursor = 0;
- while (deleteCursor < deleteIntervalList.size()) {
- if (deleteIntervalList.get(deleteCursor).getMax() < candidateTimestamp) {
- deleteCursor++;
- } else if (deleteIntervalList.get(deleteCursor).contains(candidateTimestamp)) {
- isDeletedItself = true;
- deleteStartTime = deleteIntervalList.get(deleteCursor).getMin();
- break; // since delete intervals are already sorted and merged
- } else {
- break; // since delete intervals are already sorted and merged
- }
- }
- }
- if (isDeletedItself) {
- // deleteStartTime may be before the current startTime,
- // because deleteEndTime can be before the endTime of the whole chunk
- if (deleteStartTime <= susp_candidate.getStatistics().getStartTime()) {
- // NOTE here calculate LP.
- // deleted as a whole
- currentChunkList.remove(susp_candidate);
- } else {
- susp_candidate.setLazyLoad(true);
- // NOTE here get statistics from ChunkSuit4CPV, not from
- // ChunkSuit4CPV.ChunkMetadata
- susp_candidate.getStatistics().setEndTime(deleteStartTime - 1);
- // -1 is because delete is closed interval
- // check
- }
- continue; // back to loop 1
- } else {
- // the candidate point is not deleted, then it is the final result
- results
- .get(1)
- .updateResultUsingValues(
- new long[] {candidateTimestamp}, 1, new Object[] {candidateValue});
- results
- .get(3)
- .updateResultUsingValues(
- new long[] {candidateTimestamp}, 1, new Object[] {candidateValue});
- return;
- }
- }
- }
- }
-
@Override
public Pair<Long, Object> peekNextNotNullValue(long nextStartTime, long nextEndTime)
throws IOException {
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest8.java b/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest8.java
index e3a8a2c3b53..df261cfc175 100644
--- a/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest8.java
+++ b/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest8.java
@@ -41,6 +41,7 @@ import java.util.Locale;
import static org.junit.Assert.fail;
public class MyTest8 {
+
// test MinMax-LSM
private static final String TIMESTAMP_STR = "Time";
@@ -65,6 +66,7 @@ public class MyTest8 {
@Before
public void setUp() throws Exception {
TSFileDescriptor.getInstance().getConfig().setEnableMinMaxLSM(true);
+
TSFileDescriptor.getInstance().getConfig().setTimeEncoder("PLAIN");
originalCompactionStrategy = config.getCompactionStrategy();
config.setCompactionStrategy(CompactionStrategy.NO_COMPACTION);
@@ -84,6 +86,7 @@ public class MyTest8 {
@After
public void tearDown() throws Exception {
+ TSFileDescriptor.getInstance().getConfig().setEnableMinMaxLSM(false);
EnvironmentUtils.cleanEnv();
config.setCompactionStrategy(originalCompactionStrategy);
config.setEnableCPV(originalEnableCPV);
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/ChunkSuit4CPV.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/ChunkSuit4CPV.java
index ccab972652f..9be0b03a15d 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/ChunkSuit4CPV.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/ChunkSuit4CPV.java
@@ -45,6 +45,7 @@ public class ChunkSuit4CPV {
public Statistics statistics; // dynamically updated, includes FP/LP/BP/TP info
// [startPos,endPos] definitely for curStartTime interval, thanks to split4CPV
+ public boolean needsUpdateStartEndPos = false;
public int startPos = -1; // the first point position, starting from 0
public int endPos = -1; // the last point position, starting from 0
[iotdb] 05/09: add
Posted by le...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
leirui pushed a commit to branch research/M4-visualization
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 4520f872b6669c06fe034ac17ab43ee3eebf938c
Author: Lei Rui <10...@qq.com>
AuthorDate: Sat Sep 30 15:08:07 2023 +0800
add
---
.../apache/iotdb/db/integration/m4/MyTest8.java | 636 +++++++++++++++++++++
1 file changed, 636 insertions(+)
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest8.java b/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest8.java
new file mode 100644
index 00000000000..c9187d9d53b
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest8.java
@@ -0,0 +1,636 @@
+/*
+ * 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.integration.m4;
+
+import static org.junit.Assert.fail;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.Locale;
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.compaction.CompactionStrategy;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.jdbc.Config;
+import org.apache.iotdb.jdbc.IoTDBStatement;
+import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class MyTest8 {
+ // test MinMax-LSM
+ private static final String TIMESTAMP_STR = "Time";
+
+ private static String[] creationSqls =
+ new String[] {
+ "SET STORAGE GROUP TO root.vehicle.d0",
+ "CREATE TIMESERIES root.vehicle.d0.s0 WITH DATATYPE=INT64, ENCODING=PLAIN",
+ // IoTDB int data type does not support plain encoding, so use long data type
+ };
+
+ private final String d0s0 = "root.vehicle.d0.s0";
+
+ private static final String insertTemplate =
+ "INSERT INTO root.vehicle.d0(timestamp,s0)" + " VALUES(%d,%d)";
+
+ private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ private static boolean originalEnableCPV;
+
+ private static boolean originalUseMad;
+ private static CompactionStrategy originalCompactionStrategy;
+
+ @Before
+ public void setUp() throws Exception {
+ TSFileDescriptor.getInstance().getConfig().setEnableMinMaxLSM(true);
+ TSFileDescriptor.getInstance().getConfig().setTimeEncoder("PLAIN");
+ originalCompactionStrategy = config.getCompactionStrategy();
+ config.setCompactionStrategy(CompactionStrategy.NO_COMPACTION);
+
+ originalEnableCPV = config.isEnableCPV();
+ config.setEnableCPV(true); // CPV
+
+ originalUseMad = TSFileDescriptor.getInstance().getConfig().isUseMad();
+ TSFileDescriptor.getInstance().getConfig().setUseMad(true);
+
+ TSFileDescriptor.getInstance().getConfig().setErrorParam(10);
+
+ EnvironmentUtils.envSetUp();
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ config.setTimestampPrecision("ms");
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvironmentUtils.cleanEnv();
+ config.setCompactionStrategy(originalCompactionStrategy);
+ config.setEnableCPV(originalEnableCPV);
+ // TSFileDescriptor.getInstance().getConfig().setUseTimeIndex(originalUseChunkIndex);
+ TSFileDescriptor.getInstance().getConfig().setUseMad(originalUseMad);
+ }
+
+ @Test
+ public void test1() throws Exception {
+ prepareData1();
+
+ String[] res =
+ new String[] {
+ "0,5[1],30[10]",
+ "25,8[25],40[30]",
+ "50,8[52],18[54]",
+ "75,null,null"
+ };
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ /*
+ * Sql format: SELECT min_time(s0), max_time(s0), first_value(s0), last_value(s0), min_value(s0),
+ * max_value(s0) ROM root.xx group by ([tqs,tqe),IntervalLength). Requirements: (1) Don't change the
+ * sequence of the above six aggregates (2) Make sure (tqe-tqs) is divisible by IntervalLength. (3)
+ * Assume each chunk has only one page.
+ */
+ boolean hasResultSet =
+ statement.execute(
+ "SELECT min_value(s0), max_value(s0)"
+ + " FROM root.vehicle.d0 group by ([0,100),25ms)");
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ int i = 0;
+ while (resultSet.next()) {
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(String.format("min_value(%s)", d0s0))
+ + ","
+ + resultSet.getString(String.format("max_value(%s)", d0s0));
+ System.out.println(ans);
+ Assert.assertEquals(res[i++], ans);
+ }
+ }
+ System.out.println(((IoTDBStatement) statement).executeFinish());
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void test5() throws Exception {
+ prepareData5();
+
+ String[] res =
+ new String[] {
+ "0,5[1],30[10]",
+ "25,8[25],40[30]",
+ "50,null,null",
+ "75,null,null"
+ };
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ boolean hasResultSet =
+ statement.execute(
+ "SELECT min_value(s0), max_value(s0)"
+ + " FROM root.vehicle.d0 group by ([0,100),25ms)");
+
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ int i = 0;
+ while (resultSet.next()) {
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(String.format("min_value(%s)", d0s0))
+ + ","
+ + resultSet.getString(String.format("max_value(%s)", d0s0));
+ System.out.println(ans);
+ Assert.assertEquals(res[i++], ans);
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ private static void prepareData1() {
+ // data:
+ // https://user-images.githubusercontent.com/33376433/151985070-73158010-8ba0-409d-a1c1-df69bad1aaee.png
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ for (String sql : creationSqls) {
+ statement.execute(sql);
+ }
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 1, 5));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 2, 15));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 20, 1));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 25, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 54, 3));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 120, 8));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 5, 10));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 8, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 10, 30));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 20, 20));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 27, 20));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 30, 40));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 35, 10));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 40, 20));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 33, 9));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 45, 30));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 52, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 54, 18));
+ statement.execute("FLUSH");
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ private static void prepareData5() {
+ // data:
+ // https://user-images.githubusercontent.com/33376433/151985070-73158010-8ba0-409d-a1c1-df69bad1aaee.png
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ for (String sql : creationSqls) {
+ statement.execute(sql);
+ }
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 1, 5));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 2, 15));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 20, 1));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 25, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 54, 3));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 120, 8));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 5, 10));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 8, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 10, 30));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 20, 20));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 27, 20));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 30, 40));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 35, 10));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 40, 20));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 33, 9));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 45, 30));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 52, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 54, 18));
+ statement.execute("FLUSH");
+
+ statement.execute("delete from root.vehicle.d0.s0 where time>=52 and time<=54");
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ @Test
+ public void test2() { // add deletes
+ prepareData2();
+
+ String[] res =
+ new String[] {
+ "0,5[1],30[10]",
+ "25,8[25],20[27]",
+ "50,null,null",
+ "75,null,null"
+ };
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ boolean hasResultSet =
+ statement.execute(
+ "SELECT min_value(s0), max_value(s0)"
+ + " FROM root.vehicle.d0 group by ([0,100),25ms)"); // don't change the
+ // sequence!!!
+
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ int i = 0;
+ while (resultSet.next()) {
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(String.format("min_value(%s)", d0s0))
+ + ","
+ + resultSet.getString(String.format("max_value(%s)", d0s0));
+ System.out.println(ans);
+ Assert.assertEquals(res[i++], ans);
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ private static void prepareData2() {
+ // data:
+ // https://user-images.githubusercontent.com/33376433/151995378-07a2f8df-5cac-499a-ae88-e3b017eee07a.png
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ for (String sql : creationSqls) {
+ statement.execute(sql);
+ }
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 1, 5));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 2, 15));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 20, 1));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 25, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 54, 3));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 120, 8));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 5, 10));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 8, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 10, 30));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 20, 20));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 27, 20));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 30, 40));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 35, 10));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 40, 20));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 33, 9));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 45, 30));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 52, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 54, 18));
+ statement.execute("FLUSH");
+
+ statement.execute("delete from root.vehicle.d0.s0 where time>=28 and time<=60");
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ @Test
+ public void test2_2() { // use data2 but change the sql from group by ([0,100),25ms) to group by
+ // ([0,150),25ms)
+ prepareData2();
+
+ String[] res =
+ new String[] {
+ "0,5[1],30[10]",
+ "25,8[25],20[27]",
+ "50,null,null",
+ "75,null,null",
+ "100,8[120],8[120]",
+ "125,null,null"
+ };
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ boolean hasResultSet =
+ statement.execute(
+ "SELECT min_value(s0), max_value(s0)"
+ + " FROM root.vehicle.d0 group by ([0,150),25ms)"); // don't change the
+ // sequence!!!
+
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ int i = 0;
+ while (resultSet.next()) {
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(String.format("min_value(%s)", d0s0))
+ + ","
+ + resultSet.getString(String.format("max_value(%s)", d0s0));
+ System.out.println(ans);
+ Assert.assertEquals(res[i++], ans);
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void test3() { // all seq
+ prepareData3();
+
+ String[] res =
+ new String[] {
+ "0,1[10],10[2]",
+ "25,2[40],8[30]",
+ "50,4[72],20[62]",
+ "75,1[90],11[80]"
+ };
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ boolean hasResultSet =
+ statement.execute(
+ "SELECT min_value(s0), max_value(s0)"
+ + " FROM root.vehicle.d0 group by ([0,100),25ms)"); // don't change the
+ // sequence!!!
+
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ int i = 0;
+ while (resultSet.next()) {
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(String.format("min_value(%s)", d0s0))
+ + ","
+ + resultSet.getString(String.format("max_value(%s)", d0s0));
+ System.out.println(ans);
+ Assert.assertEquals(res[i++], ans);
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ private static void prepareData3() {
+ // data:
+ // https://user-images.githubusercontent.com/33376433/152003603-6b4e7494-00ff-47e4-bf6e-cab3c8600ce2.png
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ for (String sql : creationSqls) {
+ statement.execute(sql);
+ }
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 1, 5));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 2, 10));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 10, 1));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 20, 5));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 22, 4));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 30, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 40, 2));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 55, 5));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 60, 15));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 62, 20));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 65, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 70, 18));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 72, 4));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 80, 11));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 90, 1));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 105, 7));
+ statement.execute("FLUSH");
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ @Test
+ public void test3_2() { // not seq but no overlap
+ prepareData3_2();
+
+ String[] res =
+ new String[] {
+ "0,1[10],10[2]",
+ "25,2[40],8[30]",
+ "50,4[72],20[62]",
+ "75,1[90],11[80]"
+ };
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ boolean hasResultSet =
+ statement.execute(
+ "SELECT min_value(s0), max_value(s0)"
+ + " FROM root.vehicle.d0 group by ([0,100),25ms)"); // don't change the
+ // sequence!!!
+
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ int i = 0;
+ while (resultSet.next()) {
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(String.format("min_value(%s)", d0s0))
+ + ","
+ + resultSet.getString(String.format("max_value(%s)", d0s0));
+ System.out.println(ans);
+ Assert.assertEquals(res[i++], ans);
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ private static void prepareData3_2() {
+ // data:
+ // https://user-images.githubusercontent.com/33376433/152003603-6b4e7494-00ff-47e4-bf6e-cab3c8600ce2.png
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ for (String sql : creationSqls) {
+ statement.execute(sql);
+ }
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 72, 4));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 80, 11));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 90, 1));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 105, 7));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 1, 5));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 2, 10));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 10, 1));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 20, 5));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 22, 4));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 30, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 40, 2));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 55, 5));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 60, 15));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 62, 20));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 65, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 70, 18));
+ statement.execute("FLUSH");
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ @Test
+ public void test4() { // delete sequence move forward
+ prepareData4();
+
+ String[] res =
+ new String[] {
+ "0,5[1],30[10]",
+ "25,8[25],30[45]",
+ "50,8[52],18[54]",
+ "75,null,null"
+ };
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ boolean hasResultSet =
+ statement.execute(
+ "SELECT min_value(s0), max_value(s0)"
+ + " FROM root.vehicle.d0 group by ([0,100),25ms)"); // don't change the
+ // sequence!!!
+
+ Assert.assertTrue(hasResultSet);
+ try (ResultSet resultSet = statement.getResultSet()) {
+ int i = 0;
+ while (resultSet.next()) {
+ String ans =
+ resultSet.getString(TIMESTAMP_STR)
+ + ","
+ + resultSet.getString(String.format("min_value(%s)", d0s0))
+ + ","
+ + resultSet.getString(String.format("max_value(%s)", d0s0));
+ System.out.println(ans);
+ Assert.assertEquals(res[i++], ans);
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ private static void prepareData4() {
+ // data:
+ // https://user-images.githubusercontent.com/33376433/152006061-f1d95952-3f5c-4d88-b34e-45d3bb61b600.png
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ for (String sql : creationSqls) {
+ statement.execute(sql);
+ }
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 1, 5));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 2, 15));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 20, 1));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 25, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 54, 3));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 120, 8));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 5, 10));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 8, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 10, 30));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 20, 20));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 27, 20));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 30, 40));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 35, 10));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 40, 20));
+ statement.execute("FLUSH");
+
+ statement.execute("delete from root.vehicle.d0.s0 where time>=28 and time<=42");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 33, 9));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 45, 30));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 52, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, 54, 18));
+ statement.execute("FLUSH");
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+}
[iotdb] 01/09: add LTTB
Posted by le...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
leirui pushed a commit to branch research/M4-visualization
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 0dc6b40453f1e00913d9ec3e8f26101beb9fd2f1
Author: Lei Rui <10...@qq.com>
AuthorDate: Wed Sep 27 20:58:18 2023 +0800
add LTTB
---
server/pom.xml | 25 ++
.../db/query/udf/builtin/BuiltinFunction.java | 3 +-
.../iotdb/db/query/udf/builtin/UDTFMinMax.java | 3 +
.../iotdb/db/query/udf/builtin/UDTFSample.java | 194 +++++++++++++
.../iotdb/db/query/udf/util/NoNumberException.java | 30 ++
.../org/apache/iotdb/db/query/udf/util/Util.java | 309 +++++++++++++++++++++
.../apache/iotdb/db/integration/m4/MyTest6.java | 171 ++++++++++++
7 files changed, 734 insertions(+), 1 deletion(-)
diff --git a/server/pom.xml b/server/pom.xml
index 5c4f833d826..775937ae168 100644
--- a/server/pom.xml
+++ b/server/pom.xml
@@ -29,12 +29,29 @@
</parent>
<artifactId>iotdb-server</artifactId>
<name>IoTDB Server</name>
+ <repositories>
+ <repository>
+ <id>jitpack.io</id>
+ <url>https://jitpack.io</url>
+ </repository>
+ </repositories>
<properties>
<iotdb.test.skip>false</iotdb.test.skip>
<iotdb.it.skip>${iotdb.test.skip}</iotdb.it.skip>
<iotdb.ut.skip>${iotdb.test.skip}</iotdb.ut.skip>
</properties>
<dependencies>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-math3</artifactId>
+ <version>3.5</version>
+ </dependency>
+ <!-- Sampling -->
+ <dependency>
+ <groupId>com.github.ggalmazor</groupId>
+ <artifactId>lt_downsampling_java8</artifactId>
+ <version>0.0.6</version>
+ </dependency>
<dependency>
<groupId>org.apache.iotdb</groupId>
<artifactId>service-rpc</artifactId>
@@ -261,6 +278,14 @@
</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <source>9</source>
+ <target>9</target>
+ </configuration>
+ </plugin>
</plugins>
</build>
<profiles>
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/BuiltinFunction.java b/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/BuiltinFunction.java
index 216d912d393..ebde0642972 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/BuiltinFunction.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/BuiltinFunction.java
@@ -48,7 +48,8 @@ public enum BuiltinFunction {
TOP_K("TOP_K", UDTFTopK.class),
BOTTOM_K("BOTTOM_K", UDTFBottomK.class),
M4("M4", UDTFM4MAC.class),
- M4_TW("M4_TW", UDTFM4.class);
+ M4_TW("M4_TW", UDTFM4.class),
+ Sample("SAMPLE", UDTFSample.class);
;
private final String functionName;
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/UDTFMinMax.java b/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/UDTFMinMax.java
new file mode 100644
index 00000000000..da6dee8a979
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/UDTFMinMax.java
@@ -0,0 +1,3 @@
+package org.apache.iotdb.db.query.udf.builtin;
+
+public class UDTFMinMax {}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/UDTFSample.java b/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/UDTFSample.java
new file mode 100644
index 00000000000..e9bf4d5a009
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/builtin/UDTFSample.java
@@ -0,0 +1,194 @@
+/*
+ * 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.query.udf.builtin;
+
+import org.apache.iotdb.db.query.udf.api.UDTF;
+import org.apache.iotdb.db.query.udf.api.access.Row;
+import org.apache.iotdb.db.query.udf.api.access.RowIterator;
+import org.apache.iotdb.db.query.udf.api.access.RowWindow;
+import org.apache.iotdb.db.query.udf.api.collector.PointCollector;
+import org.apache.iotdb.db.query.udf.api.customizer.config.UDTFConfigurations;
+import org.apache.iotdb.db.query.udf.api.customizer.parameter.UDFParameterValidator;
+import org.apache.iotdb.db.query.udf.api.customizer.parameter.UDFParameters;
+import org.apache.iotdb.db.query.udf.api.customizer.strategy.RowByRowAccessStrategy;
+import org.apache.iotdb.db.query.udf.api.customizer.strategy.SlidingSizeWindowAccessStrategy;
+import org.apache.iotdb.db.query.udf.util.NoNumberException;
+import org.apache.iotdb.db.query.udf.util.Util;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+
+import com.github.ggalmazor.ltdownsampling.LTThreeBuckets;
+import com.github.ggalmazor.ltdownsampling.Point;
+import org.apache.commons.lang3.tuple.Pair;
+
+import java.math.BigDecimal;
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Random;
+
+/** This function samples data by pool sampling. */
+public class UDTFSample implements UDTF {
+
+ enum Method {
+ ISOMETRIC,
+ RESERVOIR,
+ TRIANGLE
+ }
+
+ private int k; // sample numbers
+ private Method method;
+ private static final String METHOD_RESERVOIR = "reservoir";
+ // These variables occurs in pool sampling
+ private Pair<Long, Object>[] samples; // sampled data
+ private int num = 0; // number of points already sampled
+ private Random random;
+ private TSDataType dataType;
+
+ @Override
+ public void validate(UDFParameterValidator validator) throws Exception {
+ validator
+ .validateInputSeriesNumber(1)
+ .validate(
+ k -> (int) k > 0,
+ "k should be a positive integer.",
+ validator.getParameters().getIntOrDefault("k", 1))
+ .validate(
+ method ->
+ "isometric".equalsIgnoreCase((String) method)
+ || METHOD_RESERVOIR.equalsIgnoreCase((String) method)
+ || "triangle".equalsIgnoreCase((String) method),
+ "Illegal sampling method.",
+ validator.getParameters().getStringOrDefault("method", METHOD_RESERVOIR));
+ }
+
+ @Override
+ public void beforeStart(UDFParameters parameters, UDTFConfigurations configurations)
+ throws Exception {
+ this.k = parameters.getIntOrDefault("k", 1);
+ this.dataType = parameters.getDataType(0);
+ String methodIn = parameters.getStringOrDefault("method", METHOD_RESERVOIR);
+ if ("triangle".equalsIgnoreCase(methodIn)) {
+ this.method = Method.TRIANGLE;
+ } else if ("isometric".equalsIgnoreCase(methodIn)) {
+ this.method = Method.ISOMETRIC;
+ } else {
+ this.method = Method.RESERVOIR;
+ }
+ if (this.method == Method.ISOMETRIC || this.method == Method.TRIANGLE) {
+ configurations
+ .setAccessStrategy(new SlidingSizeWindowAccessStrategy(Integer.MAX_VALUE))
+ .setOutputDataType(parameters.getDataType(0));
+ } else {
+ configurations
+ .setAccessStrategy(new RowByRowAccessStrategy())
+ .setOutputDataType(parameters.getDataType(0));
+ this.samples = new Pair[this.k];
+ this.random = new Random();
+ }
+ }
+
+ @Override
+ public void transform(Row row, PointCollector collector) throws Exception {
+ // pool sampling
+ int x;
+ if (this.num < this.k) {
+ x = this.num;
+ } else {
+ x = random.nextInt(num + 1);
+ }
+ if (x < this.k) {
+ Object v = Util.getValueAsObject(row);
+ Long t = row.getTime();
+ this.samples[x] = Pair.of(t, v);
+ }
+ this.num++;
+ }
+
+ @Override
+ public void transform(RowWindow rowWindow, PointCollector collector) throws Exception {
+ // equal-distance sampling
+ int n = rowWindow.windowSize();
+
+ if (this.k < n) {
+ if (this.method == Method.TRIANGLE) {
+ List<Point> input = new LinkedList<>();
+ for (int i = 0; i < n; i++) {
+ Row row = rowWindow.getRow(i);
+ BigDecimal time = BigDecimal.valueOf(row.getTime());
+ BigDecimal data = BigDecimal.valueOf(Util.getValueAsDouble(row));
+ input.add(new Point(time, data));
+ }
+ if (k > 2) {
+ // The first and last element will always be sampled so the buckets is k - 2
+ List<Point> output = LTThreeBuckets.sorted(input, k - 2);
+ for (Point p : output) {
+ switch (dataType) {
+ case INT32:
+ Util.putValue(collector, dataType, p.getX().longValue(), p.getY().intValue());
+ break;
+ case INT64:
+ Util.putValue(collector, dataType, p.getX().longValue(), p.getY().longValue());
+ break;
+ case FLOAT:
+ Util.putValue(collector, dataType, p.getX().longValue(), p.getY().floatValue());
+ break;
+ case DOUBLE:
+ Util.putValue(collector, dataType, p.getX().longValue(), p.getY().doubleValue());
+ break;
+ default:
+ throw new NoNumberException();
+ }
+ }
+ } else { // For corner case of k == 1 and k == 2
+ Row row = rowWindow.getRow(0); // Put first element
+ Util.putValue(collector, dataType, row.getTime(), Util.getValueAsObject(row));
+ if (k == 2) {
+ row = rowWindow.getRow(n - 1); // Put last element
+ Util.putValue(collector, dataType, row.getTime(), Util.getValueAsObject(row));
+ }
+ }
+ } else { // Method.ISOMETRIC
+ for (long i = 0; i < this.k; i++) {
+ long j = Math.floorDiv(i * n, k); // avoid intermediate result overflows
+ Row row = rowWindow.getRow((int) j);
+ Util.putValue(collector, dataType, row.getTime(), Util.getValueAsObject(row));
+ }
+ }
+ } else { // when k is larger than series length, output all points
+ RowIterator iterator = rowWindow.getRowIterator();
+ while (iterator.hasNextRow()) {
+ Row row = iterator.next();
+ Util.putValue(collector, dataType, row.getTime(), Util.getValueAsObject(row));
+ }
+ }
+ }
+
+ @Override
+ public void terminate(PointCollector pc) throws Exception {
+ if (samples != null) { // for pool sampling only
+ int m = Math.min(num, k);
+ Arrays.sort(samples, 0, m);
+ for (int i = 0; i < m; i++) {
+ Pair<Long, Object> p = samples[i];
+ Util.putValue(pc, dataType, p.getLeft(), p.getRight());
+ }
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/util/NoNumberException.java b/server/src/main/java/org/apache/iotdb/db/query/udf/util/NoNumberException.java
new file mode 100644
index 00000000000..7fd4802e7e1
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/util/NoNumberException.java
@@ -0,0 +1,30 @@
+/*
+ * 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.query.udf.util;
+
+/** throw when the value got is not numeric. */
+public class NoNumberException extends Exception {
+
+ @Override
+ public String toString() {
+ String s = "The value of the input time series is not numeric.\n";
+ return s + super.toString();
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/util/Util.java b/server/src/main/java/org/apache/iotdb/db/query/udf/util/Util.java
new file mode 100644
index 00000000000..0528681a42b
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/util/Util.java
@@ -0,0 +1,309 @@
+/*
+ * 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.query.udf.util;
+
+import org.apache.iotdb.db.query.udf.api.access.Row;
+import org.apache.iotdb.db.query.udf.api.collector.PointCollector;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+
+import org.apache.commons.math3.stat.descriptive.rank.Median;
+import org.eclipse.collections.api.tuple.primitive.LongIntPair;
+import org.eclipse.collections.impl.map.mutable.primitive.LongIntHashMap;
+
+import java.io.IOException;
+import java.util.List;
+
+/** This class offers functions of getting and putting values from iotdb interface. */
+public class Util {
+
+ private Util() {
+ throw new IllegalStateException("Utility class");
+ }
+
+ /**
+ * Get value from specific column from Row, and cast to double. Make sure never get null from Row.
+ *
+ * @param row data row
+ * @param index the column index
+ * @return value of specific column from Row
+ * @throws NoNumberException when getting a no number datatype
+ */
+ public static double getValueAsDouble(Row row, int index) throws IOException, NoNumberException {
+ double ans = 0;
+ switch (row.getDataType(index)) {
+ case INT32:
+ ans = row.getInt(index);
+ break;
+ case INT64:
+ ans = row.getLong(index);
+ break;
+ case FLOAT:
+ ans = row.getFloat(index);
+ break;
+ case DOUBLE:
+ ans = row.getDouble(index);
+ break;
+ default:
+ throw new NoNumberException();
+ }
+ return ans;
+ }
+
+ /**
+ * Get value from 0th column from Row, and cast to double. Make sure never get null from Row.
+ *
+ * @param row data row
+ * @return value from 0th column from Row
+ * @throws NoNumberException when getting a no number datatype
+ */
+ public static double getValueAsDouble(Row row) throws IOException, NoNumberException {
+ return getValueAsDouble(row, 0);
+ }
+
+ /**
+ * Get value from 0th column from Row, and cast to Object.
+ *
+ * @param row data row
+ * @return value from 0th column from Row
+ */
+ public static Object getValueAsObject(Row row) throws IOException {
+ Object ans = 0;
+ switch (row.getDataType(0)) {
+ case INT32:
+ ans = row.getInt(0);
+ break;
+ case INT64:
+ ans = row.getLong(0);
+ break;
+ case FLOAT:
+ ans = row.getFloat(0);
+ break;
+ case DOUBLE:
+ ans = row.getDouble(0);
+ break;
+ case BOOLEAN:
+ ans = row.getBoolean(0);
+ break;
+ case TEXT:
+ ans = row.getString(0);
+ break;
+ default:
+ break;
+ }
+ return ans;
+ }
+
+ /**
+ * Add new data point to PointCollector.
+ *
+ * @param pc PointCollector
+ * @param type datatype
+ * @param t timestamp
+ * @param o value in Object type
+ */
+ public static void putValue(PointCollector pc, TSDataType type, long t, Object o)
+ throws IOException {
+ switch (type) {
+ case INT32:
+ pc.putInt(t, (Integer) o);
+ break;
+ case INT64:
+ pc.putLong(t, (Long) o);
+ break;
+ case FLOAT:
+ pc.putFloat(t, (Float) o);
+ break;
+ case DOUBLE:
+ pc.putDouble(t, (Double) o);
+ break;
+ case BOOLEAN:
+ pc.putBoolean(t, (Boolean) o);
+ break;
+ default:
+ break;
+ }
+ }
+
+ /**
+ * cast {@code ArrayList<Double>} to {@code double[]}.
+ *
+ * @param list ArrayList to cast
+ * @return cast result
+ */
+ public static double[] toDoubleArray(List<Double> list) {
+ return list.stream().mapToDouble(Double::valueOf).toArray();
+ }
+
+ /**
+ * cast {@code ArrayList<Long>} to {@code long[]}.
+ *
+ * @param list ArrayList to cast
+ * @return cast result
+ */
+ public static long[] toLongArray(List<Long> list) {
+ return list.stream().mapToLong(Long::valueOf).toArray();
+ }
+
+ /**
+ * calculate median absolute deviation of input series. 1.4826 is multiplied in order to achieve
+ * asymptotic normality. Note: 1.4826 = 1/qnorm(3/4)
+ *
+ * @param value input series
+ * @return median absolute deviation MAD
+ */
+ public static double mad(double[] value) {
+ Median median = new Median();
+ double mid = median.evaluate(value);
+ double[] d = new double[value.length];
+ for (int i = 0; i < value.length; i++) {
+ d[i] = Math.abs(value[i] - mid);
+ }
+ return 1.4826 * median.evaluate(d);
+ }
+
+ /**
+ * calculate 1-order difference of input series.
+ *
+ * @param origin original series
+ * @return 1-order difference
+ */
+ public static double[] variation(double[] origin) {
+ int n = origin.length;
+ double[] variance = new double[n - 1];
+ for (int i = 0; i < n - 1; i++) {
+ variance[i] = origin[i + 1] - origin[i];
+ }
+ return variance;
+ }
+
+ /**
+ * calculate 1-order difference of input series.
+ *
+ * @param origin original series
+ * @return 1-order difference
+ */
+ public static double[] variation(long[] origin) {
+ int n = origin.length;
+ double[] variance = new double[n - 1];
+ for (int i = 0; i < n - 1; i++) {
+ variance[i] = (origin[i + 1] - origin[i]);
+ }
+ return variance;
+ }
+
+ /**
+ * calculate 1-order difference of input series.
+ *
+ * @param origin original series
+ * @return 1-order difference
+ */
+ public static int[] variation(int[] origin) {
+ int n = origin.length;
+ int[] variance = new int[n - 1];
+ for (int i = 0; i < n - 1; i++) {
+ variance[i] = origin[i + 1] - origin[i];
+ }
+ return variance;
+ }
+
+ /**
+ * calculate speed (1-order derivative with backward difference).
+ *
+ * @param origin value series
+ * @param time timestamp series
+ * @return speed series
+ */
+ public static double[] speed(double[] origin, double[] time) {
+ int n = origin.length;
+ double[] speed = new double[n - 1];
+ for (int i = 0; i < n - 1; i++) {
+ speed[i] = (origin[i + 1] - origin[i]) / (time[i + 1] - time[i]);
+ }
+ return speed;
+ }
+
+ /**
+ * calculate speed (1-order derivative with backward difference).
+ *
+ * @param origin value series
+ * @param time timestamp series
+ * @return speed series
+ */
+ public static double[] speed(double[] origin, long[] time) {
+ int n = origin.length;
+ double[] speed = new double[n - 1];
+ for (int i = 0; i < n - 1; i++) {
+ speed[i] = (origin[i + 1] - origin[i]) / (time[i + 1] - time[i]);
+ }
+ return speed;
+ }
+
+ /**
+ * computes mode.
+ *
+ * @param values input series
+ * @return mode
+ */
+ public static long mode(long[] values) {
+ LongIntHashMap map = new LongIntHashMap();
+ for (long v : values) {
+ map.addToValue(v, 1);
+ }
+ long key = 0;
+ int maxValue = 0;
+ for (LongIntPair p : map.keyValuesView()) {
+ if (p.getTwo() > maxValue) {
+ key = p.getOne();
+ maxValue = p.getTwo();
+ }
+ }
+ return key;
+ }
+
+ /**
+ * cast String to timestamp.
+ *
+ * @param s input string
+ * @return timestamp
+ */
+ public static long parseTime(String s) {
+ long unit = 0;
+ s = s.toLowerCase();
+ s = s.replace(" ", "");
+ if (s.endsWith("ms")) {
+ unit = 1;
+ s = s.substring(0, s.length() - 2);
+ } else if (s.endsWith("s")) {
+ unit = 1000;
+ s = s.substring(0, s.length() - 1);
+ } else if (s.endsWith("m")) {
+ unit = 60 * 1000L;
+ s = s.substring(0, s.length() - 1);
+ } else if (s.endsWith("h")) {
+ unit = 60 * 60 * 1000L;
+ s = s.substring(0, s.length() - 1);
+ } else if (s.endsWith("d")) {
+ unit = 24 * 60 * 60 * 1000L;
+ s = s.substring(0, s.length() - 1);
+ }
+ double v = Double.parseDouble(s);
+ return (long) (unit * v);
+ }
+}
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest6.java b/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest6.java
new file mode 100644
index 00000000000..46de9698d4d
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest6.java
@@ -0,0 +1,171 @@
+/*
+ * 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.integration.m4;
+
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.compaction.CompactionStrategy;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.jdbc.Config;
+import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Locale;
+
+import static org.junit.Assert.fail;
+
+public class MyTest6 {
+
+ private static final String TIMESTAMP_STR = "Time";
+
+ private static String[] creationSqls =
+ new String[] {
+ "SET STORAGE GROUP TO root.vehicle.d0",
+ "CREATE TIMESERIES root.vehicle.d0.s0 WITH DATATYPE=DOUBLE,ENCODING=PLAIN",
+ };
+
+ private final String d0s0 = "root.vehicle.d0.s0";
+
+ private static final String insertTemplate =
+ "INSERT INTO root.vehicle.d0(timestamp,s0)" + " VALUES(%d,%d)";
+
+ private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ private static boolean originalEnableCPV;
+ private static CompactionStrategy originalCompactionStrategy;
+ private static boolean originalUseChunkIndex;
+
+ @Before
+ public void setUp() throws Exception {
+ TSFileDescriptor.getInstance().getConfig().setTimeEncoder("PLAIN");
+ originalCompactionStrategy = config.getCompactionStrategy();
+ config.setCompactionStrategy(CompactionStrategy.NO_COMPACTION);
+
+ originalEnableCPV = config.isEnableCPV();
+ // config.setEnableCPV(false); // MOC
+ config.setEnableCPV(true); // CPV
+
+ originalUseChunkIndex = TSFileDescriptor.getInstance().getConfig().isUseTimeIndex();
+ TSFileDescriptor.getInstance().getConfig().setUseTimeIndex(false);
+
+ EnvironmentUtils.envSetUp();
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ config.setTimestampPrecision("ms");
+
+ TSFileDescriptor.getInstance().getConfig().setErrorParam(10);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvironmentUtils.cleanEnv();
+ config.setCompactionStrategy(originalCompactionStrategy);
+ config.setEnableCPV(originalEnableCPV);
+ TSFileDescriptor.getInstance().getConfig().setUseTimeIndex(originalUseChunkIndex);
+ }
+
+ @Test
+ public void testM4Function() {
+ // create timeseries
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.m4");
+ statement.execute("CREATE TIMESERIES root.m4.d1.s1 with datatype=double,encoding=PLAIN");
+ statement.execute("CREATE TIMESERIES root.m4.d1.s2 with datatype=INT64,encoding=PLAIN");
+ } catch (SQLException throwable) {
+ fail(throwable.getMessage());
+ }
+
+ // insert data
+ String insertTemplate = "INSERT INTO root.m4.d1(timestamp,%s)" + " VALUES(%d,%d)";
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ // "root.m4.d1.s1" data illustration:
+ // https://user-images.githubusercontent.com/33376433/151985070-73158010-8ba0-409d-a1c1-df69bad1aaee.png
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 1, 5));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 2, 15));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 20, 1));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 25, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 54, 3));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 120, 8));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 5, 10));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 8, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 10, 30));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 20, 20));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 27, 20));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 30, 40));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 35, 10));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 40, 20));
+ statement.execute("FLUSH");
+
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 33, 9));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 45, 30));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 52, 8));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s1", 54, 18));
+ statement.execute("FLUSH");
+
+ // "root.m4.d1.s2" data: constant value 1
+ for (int i = 0; i < 100; i++) {
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, "s2", i, 1));
+ }
+ statement.execute("FLUSH");
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+
+ // query tests
+ test_sample();
+ }
+
+ private void test_sample() {
+ String[] res = new String[] {"1,5.0", "10,30.0", "33,9.0", "120,8.0"};
+
+ String sql = "select Sample(s1, 'method'='triangle','k'='4') from root.m4.d1";
+
+ try (Connection conn =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = conn.createStatement()) {
+ ResultSet resultSet = statement.executeQuery(sql);
+ int count = 0;
+ while (resultSet.next()) {
+ String str = resultSet.getString(1) + "," + resultSet.getString(2);
+ Assert.assertEquals(res[count], str);
+ count++;
+ System.out.println(str);
+ }
+ Assert.assertEquals(res.length, count);
+ } catch (SQLException throwable) {
+ fail(throwable.getMessage());
+ }
+ }
+}
[iotdb] 06/09: save query result
Posted by le...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
leirui pushed a commit to branch research/M4-visualization
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d5855e8361a80f000b233978a7085b9bef431cda
Author: Lei Rui <10...@qq.com>
AuthorDate: Sun Oct 1 00:01:12 2023 +0800
save query result
---
.../apache/iotdb/db/integration/m4/MyTest8.java | 58 ++++++----------------
.../org/apache/iotdb/session/SessionDataSet.java | 2 +-
2 files changed, 16 insertions(+), 44 deletions(-)
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest8.java b/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest8.java
index c9187d9d53b..e3a8a2c3b53 100644
--- a/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest8.java
+++ b/server/src/test/java/org/apache/iotdb/db/integration/m4/MyTest8.java
@@ -19,13 +19,6 @@
package org.apache.iotdb.db.integration.m4;
-import static org.junit.Assert.fail;
-
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.ResultSet;
-import java.sql.Statement;
-import java.util.Locale;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.compaction.CompactionStrategy;
@@ -33,11 +26,20 @@ import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.jdbc.Config;
import org.apache.iotdb.jdbc.IoTDBStatement;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
+
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.Locale;
+
+import static org.junit.Assert.fail;
+
public class MyTest8 {
// test MinMax-LSM
private static final String TIMESTAMP_STR = "Time";
@@ -94,12 +96,7 @@ public class MyTest8 {
prepareData1();
String[] res =
- new String[] {
- "0,5[1],30[10]",
- "25,8[25],40[30]",
- "50,8[52],18[54]",
- "75,null,null"
- };
+ new String[] {"0,5[1],30[10]", "25,8[25],40[30]", "50,8[52],18[54]", "75,null,null"};
try (Connection connection =
DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
Statement statement = connection.createStatement()) {
@@ -139,12 +136,7 @@ public class MyTest8 {
prepareData5();
String[] res =
- new String[] {
- "0,5[1],30[10]",
- "25,8[25],40[30]",
- "50,null,null",
- "75,null,null"
- };
+ new String[] {"0,5[1],30[10]", "25,8[25],40[30]", "50,null,null", "75,null,null"};
try (Connection connection =
DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
Statement statement = connection.createStatement()) {
@@ -266,12 +258,7 @@ public class MyTest8 {
prepareData2();
String[] res =
- new String[] {
- "0,5[1],30[10]",
- "25,8[25],20[27]",
- "50,null,null",
- "75,null,null"
- };
+ new String[] {"0,5[1],30[10]", "25,8[25],20[27]", "50,null,null", "75,null,null"};
try (Connection connection =
DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
Statement statement = connection.createStatement()) {
@@ -394,12 +381,7 @@ public class MyTest8 {
prepareData3();
String[] res =
- new String[] {
- "0,1[10],10[2]",
- "25,2[40],8[30]",
- "50,4[72],20[62]",
- "75,1[90],11[80]"
- };
+ new String[] {"0,1[10],10[2]", "25,2[40],8[30]", "50,4[72],20[62]", "75,1[90],11[80]"};
try (Connection connection =
DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
Statement statement = connection.createStatement()) {
@@ -474,12 +456,7 @@ public class MyTest8 {
prepareData3_2();
String[] res =
- new String[] {
- "0,1[10],10[2]",
- "25,2[40],8[30]",
- "50,4[72],20[62]",
- "75,1[90],11[80]"
- };
+ new String[] {"0,1[10],10[2]", "25,2[40],8[30]", "50,4[72],20[62]", "75,1[90],11[80]"};
try (Connection connection =
DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
Statement statement = connection.createStatement()) {
@@ -554,12 +531,7 @@ public class MyTest8 {
prepareData4();
String[] res =
- new String[] {
- "0,5[1],30[10]",
- "25,8[25],30[45]",
- "50,8[52],18[54]",
- "75,null,null"
- };
+ new String[] {"0,5[1],30[10]", "25,8[25],30[45]", "50,8[52],18[54]", "75,null,null"};
try (Connection connection =
DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
Statement statement = connection.createStatement()) {
diff --git a/session/src/main/java/org/apache/iotdb/session/SessionDataSet.java b/session/src/main/java/org/apache/iotdb/session/SessionDataSet.java
index bb0753223b1..4e3c5869f07 100644
--- a/session/src/main/java/org/apache/iotdb/session/SessionDataSet.java
+++ b/session/src/main/java/org/apache/iotdb/session/SessionDataSet.java
@@ -41,7 +41,7 @@ import static org.apache.iotdb.rpc.IoTDBRpcDataSet.START_INDEX;
public class SessionDataSet {
- private final IoTDBRpcDataSet ioTDBRpcDataSet;
+ public final IoTDBRpcDataSet ioTDBRpcDataSet;
private String finishResult;
public SessionDataSet(
[iotdb] 08/09: add tracing info for updated disk points
Posted by le...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
leirui pushed a commit to branch research/M4-visualization
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 85af5da3b1c49b204d27de678d8bc1bf2d31bef7
Author: Lei Rui <10...@qq.com>
AuthorDate: Thu Oct 5 21:19:49 2023 +0800
add tracing info for updated disk points
---
.../main/java/org/apache/iotdb/db/conf/IoTDBConfig.java | 2 +-
.../org/apache/iotdb/db/query/control/TracingInfo.java | 10 ++++++++++
.../org/apache/iotdb/db/query/control/TracingManager.java | 14 ++++++++++++++
.../db/query/reader/universal/PriorityMergeReader.java | 7 +++++++
.../java/org/apache/iotdb/db/tools/TsFileSketchTool.java | 3 ++-
5 files changed, 34 insertions(+), 2 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index cee0f8e24b1..f45461a0690 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -153,7 +153,7 @@ public class IoTDBConfig {
private int flushWalThreshold = 10000;
/** this variable set timestamp precision as millisecond, microsecond or nanosecond */
- private String timestampPrecision = "ns";
+ private String timestampPrecision = "ms";
/**
* The cycle when write ahead log is periodically forced to be written to disk(in milliseconds) If
diff --git a/server/src/main/java/org/apache/iotdb/db/query/control/TracingInfo.java b/server/src/main/java/org/apache/iotdb/db/query/control/TracingInfo.java
index a72c9f11020..4809631b511 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/control/TracingInfo.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/control/TracingInfo.java
@@ -35,6 +35,8 @@ public class TracingInfo {
private long totalChunkPoints = 0;
private int totalPageNum = 0;
private int overlappedPageNum = 0;
+
+ private long updatedPointNum = 0;
private Set<TsFileResource> seqFileSet = new HashSet<>();
private Set<TsFileResource> unSeqFileSet = new HashSet<>();
@@ -61,6 +63,10 @@ public class TracingInfo {
this.overlappedPageNum++;
}
+ public void addUpdatedPointNum() {
+ this.updatedPointNum++;
+ }
+
public int getTotalPageNum() {
return totalPageNum;
}
@@ -69,6 +75,10 @@ public class TracingInfo {
return overlappedPageNum;
}
+ public long getUpdatedPointNum() {
+ return updatedPointNum;
+ }
+
public Set<TsFileResource> getSeqFileSet() {
return seqFileSet;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/control/TracingManager.java b/server/src/main/java/org/apache/iotdb/db/query/control/TracingManager.java
index b2dd13a891a..ba170a6f0e5 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/control/TracingManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/control/TracingManager.java
@@ -123,6 +123,8 @@ public class TracingManager {
writeChunksInfo(queryId, tracingInfo.getTotalChunkNum(), tracingInfo.getTotalChunkPoints());
writeOverlappedPageInfo(
queryId, tracingInfo.getTotalPageNum(), tracingInfo.getOverlappedPageNum());
+ writeUpdatedPointNumInfo(
+ queryId, tracingInfo.getTotalChunkPoints(), tracingInfo.getUpdatedPointNum());
}
public void writeTsFileInfo(
@@ -185,6 +187,18 @@ public class TracingManager {
writer.write(builder.toString());
}
+ public void writeUpdatedPointNumInfo(long queryId, long totalChunkPoints, long updatedPointNum)
+ throws IOException {
+ StringBuilder builder =
+ new StringBuilder(String.format(QUERY_ID, queryId))
+ .append(" - Rate of updated points: ")
+ .append(String.format("%.1f%%, ", (double) updatedPointNum / totalChunkPoints * 100))
+ .append(
+ String.format(
+ "%d updated points in total %d points.\n", updatedPointNum, totalChunkPoints));
+ writer.write(builder.toString());
+ }
+
public void writeEndTime(long queryId) throws IOException {
long endTime = System.currentTimeMillis();
StringBuilder builder =
diff --git a/server/src/main/java/org/apache/iotdb/db/query/reader/universal/PriorityMergeReader.java b/server/src/main/java/org/apache/iotdb/db/query/reader/universal/PriorityMergeReader.java
index 870ce0450ab..7f20590f89e 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/reader/universal/PriorityMergeReader.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/reader/universal/PriorityMergeReader.java
@@ -40,6 +40,8 @@ public class PriorityMergeReader implements IPointReader {
protected PriorityQueue<Element> heap;
+ private QueryContext context;
+
public PriorityMergeReader() {
heap =
new PriorityQueue<>(
@@ -77,6 +79,7 @@ public class PriorityMergeReader implements IPointReader {
public void addReader(
IPointReader reader, MergeReaderPriority priority, long endTime, QueryContext context)
throws IOException {
+ this.context = context;
if (reader.hasNextTimeValuePair()) {
heap.add(new Element(reader, reader.nextTimeValuePair(), priority));
currentReadStopTime = Math.max(currentReadStopTime, endTime);
@@ -152,6 +155,10 @@ public class PriorityMergeReader implements IPointReader {
protected void updateHeap(long topTime, long topNextTime) throws IOException {
while (!heap.isEmpty() && heap.peek().currTime() == topTime) {
+ if (IoTDBDescriptor.getInstance().getConfig().isEnablePerformanceTracing()) {
+ // TODO tracing
+ TracingManager.getInstance().getTracingInfo(context.getQueryId()).addUpdatedPointNum();
+ }
Element e = heap.poll();
if (!e.hasNext()) {
e.reader.close();
diff --git a/server/src/main/java/org/apache/iotdb/db/tools/TsFileSketchTool.java b/server/src/main/java/org/apache/iotdb/db/tools/TsFileSketchTool.java
index 6cdb37b8ab5..568a181adb5 100644
--- a/server/src/main/java/org/apache/iotdb/db/tools/TsFileSketchTool.java
+++ b/server/src/main/java/org/apache/iotdb/db/tools/TsFileSketchTool.java
@@ -308,7 +308,8 @@ public class TsFileSketchTool {
}
private static Pair<String, String> checkArgs(String[] args) {
- String filename = "D:\\plain-plain-noindex.tsfile";
+ String filename =
+ "D:\\github\\mid\\iotdb-server-0.12.4\\data\\data\\unsequence\\root.game\\0\\0\\1696508586460-2-0-0.tsfile";
String outFile = "TsFile_sketch_view.txt";
if (args.length == 1) {
filename = args[0];