This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new f0b8a4b [IOTDB-836] Fix merging statistics failed when endFile (#1637)
f0b8a4b is described below
commit f0b8a4be8aec82d6662391b267b41f519e2d0c9e
Author: Zesong Sun <[email protected]>
AuthorDate: Sat Aug 22 16:47:07 2020 +0800
[IOTDB-836] Fix merging statistics failed when endFile (#1637)
* [IOTDB-836] Fix merging statistics failed when endFile
---
.../org/apache/iotdb/db/engine/StorageEngine.java | 13 ++++++
.../iotdb/db/engine/memtable/AbstractMemTable.java | 2 +
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 18 +--------
.../writelog/recover/TsFileRecoverPerformer.java | 5 +++
.../db/integration/IoTDBDeleteTimeseriesIT.java | 47 ++++++++++++++++------
.../iotdb/tsfile/write/writer/TsFileIOWriter.java | 5 ++-
6 files changed, 60 insertions(+), 30 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
index 57f97ff..a555696 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
@@ -441,6 +441,19 @@ public class StorageEngine implements IService {
}
/**
+ * delete data of timeseries "{deviceId}.{measurementId}"
+ */
+ public void deleteTimeseries(String deviceId, String measurementId)
+ throws StorageEngineException {
+ StorageGroupProcessor storageGroupProcessor = getProcessor(deviceId);
+ try {
+ storageGroupProcessor.delete(deviceId, measurementId, Long.MIN_VALUE,
Long.MAX_VALUE);
+ } catch (IOException e) {
+ throw new StorageEngineException(e.getMessage());
+ }
+ }
+
+ /**
* query data.
*/
public QueryDataSource query(SingleSeriesExpression seriesExpression,
QueryContext context,
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
b/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
index 763b8b4..5750502 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
@@ -237,6 +237,8 @@ public abstract class AbstractMemTable implements IMemTable
{
if (chunk == null) {
return;
}
+ // If startTimestamp == Long.MIN_VALUE && endTimestamp == Long.MAX_VALUE,
+ // it means that the whole timeseries is deleted
if (startTimestamp == Long.MIN_VALUE && endTimestamp == Long.MAX_VALUE) {
deviceMap.remove(measurementId);
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
index 7420dcd..068d888 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
@@ -977,9 +977,9 @@ public class PlanExecutor implements IPlanExecutor {
throws QueryProcessException {
List<Path> deletePathList = deleteTimeSeriesPlan.getPaths();
try {
- deleteDataOfTimeSeries(deletePathList);
List<String> failedNames = new LinkedList<>();
for (Path path : deletePathList) {
+ StorageEngine.getInstance().deleteTimeseries(path.getDevice(),
path.getMeasurement());
String failedTimeseries = mManager.deleteTimeseries(path.toString());
if (!failedTimeseries.isEmpty()) {
failedNames.add(failedTimeseries);
@@ -1059,22 +1059,6 @@ public class PlanExecutor implements IPlanExecutor {
return true;
}
- /**
- * Delete all data of time series in pathList.
- *
- * @param pathList deleted paths
- */
- protected void deleteDataOfTimeSeries(List<Path> pathList)
- throws QueryProcessException, StorageGroupNotSetException,
StorageEngineException {
- for (Path p : pathList) {
- DeletePlan deletePlan = new DeletePlan();
- deletePlan.addPath(p);
- deletePlan.setDeleteStartTime(Long.MIN_VALUE);
- deletePlan.setDeleteEndTime(Long.MAX_VALUE);
- processNonQuery(deletePlan);
- }
- }
-
protected QueryDataSet processAuthorQuery(AuthorPlan plan)
throws QueryProcessException {
AuthorType authorType = plan.getAuthorType();
diff --git
a/server/src/main/java/org/apache/iotdb/db/writelog/recover/TsFileRecoverPerformer.java
b/server/src/main/java/org/apache/iotdb/db/writelog/recover/TsFileRecoverPerformer.java
index 35e3e7e..8ce6cc3 100644
---
a/server/src/main/java/org/apache/iotdb/db/writelog/recover/TsFileRecoverPerformer.java
+++
b/server/src/main/java/org/apache/iotdb/db/writelog/recover/TsFileRecoverPerformer.java
@@ -46,6 +46,7 @@ import
org.apache.iotdb.db.writelog.manager.MultiFileLogNodeManager;
import org.apache.iotdb.tsfile.exception.NotCompatibleTsFileException;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
import org.apache.iotdb.tsfile.file.metadata.TimeseriesMetadata;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
import org.apache.iotdb.tsfile.utils.Pair;
@@ -289,7 +290,11 @@ public class TsFileRecoverPerformer {
for (Map.Entry<String, List<ChunkMetadata>> entry :
deviceChunkMetaDataMap.entrySet()) {
String deviceId = entry.getKey();
List<ChunkMetadata> chunkMetadataList = entry.getValue();
+ TSDataType dataType = entry.getValue().get(entry.getValue().size() -
1).getDataType();
for (ChunkMetadata chunkMetaData : chunkMetadataList) {
+ if (!chunkMetaData.getDataType().equals(dataType)) {
+ continue;
+ }
tsFileResource.updateStartTime(deviceId, chunkMetaData.getStartTime());
tsFileResource.updateEndTime(deviceId, chunkMetaData.getEndTime());
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBDeleteTimeseriesIT.java
b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBDeleteTimeseriesIT.java
index 67a451c..41e49d9 100644
---
a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBDeleteTimeseriesIT.java
+++
b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBDeleteTimeseriesIT.java
@@ -7,7 +7,7 @@
* "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
+ * 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
@@ -16,6 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.db.integration;
import java.sql.Connection;
@@ -23,6 +24,7 @@ import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.Statement;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.jdbc.Config;
import org.junit.After;
@@ -32,14 +34,23 @@ import org.junit.Test;
public class IoTDBDeleteTimeseriesIT {
+ private long memtableSizeThreshold;
+ private boolean enableVm;
+
@Before
public void setUp() throws Exception {
EnvironmentUtils.closeStatMonitor();
EnvironmentUtils.envSetUp();
+ memtableSizeThreshold =
IoTDBDescriptor.getInstance().getConfig().getMemtableSizeThreshold();
+ enableVm = IoTDBDescriptor.getInstance().getConfig().isEnableVm();
+ IoTDBDescriptor.getInstance().getConfig().setMemtableSizeThreshold(16);
+ IoTDBDescriptor.getInstance().getConfig().setEnableVm(false);
}
@After
public void tearDown() throws Exception {
+
IoTDBDescriptor.getInstance().getConfig().setMemtableSizeThreshold(memtableSizeThreshold);
+ IoTDBDescriptor.getInstance().getConfig().setEnableVm(enableVm);
EnvironmentUtils.cleanEnv();
}
@@ -47,20 +58,20 @@ public class IoTDBDeleteTimeseriesIT {
public void testDeleteTimeseries() throws Exception {
Class.forName(Config.JDBC_DRIVER_NAME);
String[] retArray = new String[]{
- "1,2,",
- "1,2.1,"
+ "1,1,",
+ "2,1.1,"
};
int cnt = 0;
try (Connection connection = DriverManager.
getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
- Statement statement = connection.createStatement();) {
+ Statement statement = connection.createStatement()) {
statement.execute(
- "create timeseries root.turbine1.d1.s2 with datatype=INT64,
encoding=PLAIN, compression=SNAPPY");
+ "create timeseries root.turbine1.d1.s1 with datatype=INT64,
encoding=PLAIN, compression=SNAPPY");
statement.execute(
- "create timeseries root.turbine1.d1.s3 with datatype=INT64,
encoding=PLAIN, compression=SNAPPY");
- statement.execute("INSERT INTO root.turbine1.d1(timestamp,s2,s3)
VALUES(1,2,3)");
- boolean hasResult = statement.execute("SELECT s2 FROM root.turbine1.d1");
+ "create timeseries root.turbine1.d1.s2 with datatype=INT64,
encoding=PLAIN, compression=SNAPPY");
+ statement.execute("INSERT INTO root.turbine1.d1(timestamp,s1,s2)
VALUES(1,1,2)");
+ boolean hasResult = statement.execute("SELECT s1 FROM root.turbine1.d1");
Assert.assertTrue(hasResult);
try (ResultSet resultSet = statement.getResultSet()) {
ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
@@ -73,11 +84,13 @@ public class IoTDBDeleteTimeseriesIT {
cnt++;
}
}
- statement.execute("DELETE timeseries root.turbine1.d1.s2");
+ statement.execute("DELETE timeseries root.turbine1.d1.s1");
statement.execute(
- "create timeseries root.turbine1.d1.s2 with datatype=FLOAT,
encoding=PLAIN, compression=SNAPPY");
- statement.execute("INSERT INTO root.turbine1.d1(timestamp,s2,s3)
VALUES(1,2.1,3)");
- hasResult = statement.execute("SELECT s2 FROM root.turbine1.d1");
+ "create timeseries root.turbine1.d1.s1 with datatype=DOUBLE,
encoding=PLAIN, compression=SNAPPY");
+ statement.execute("INSERT INTO root.turbine1.d1(timestamp,s1)
VALUES(2,1.1)");
+ statement.execute("FLUSH");
+
+ hasResult = statement.execute("SELECT s1 FROM root.turbine1.d1");
Assert.assertTrue(hasResult);
try (ResultSet resultSet = statement.getResultSet()) {
ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
@@ -91,5 +104,15 @@ public class IoTDBDeleteTimeseriesIT {
}
}
}
+
+ EnvironmentUtils.restartDaemon();
+
+ try(Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
+ "root");
+ Statement statement = connection.createStatement()){
+ boolean hasResult = statement.execute("SELECT * FROM root");
+ Assert.assertTrue(hasResult);
+ }
}
}
diff --git
a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileIOWriter.java
b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileIOWriter.java
index 17138aa..72586c0 100644
---
a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileIOWriter.java
+++
b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileIOWriter.java
@@ -291,13 +291,16 @@ public class TsFileIOWriter {
String device = path.getDevice();
// create TimeseriesMetaData
- TSDataType dataType = entry.getValue().get(0).getDataType();
+ TSDataType dataType = entry.getValue().get(entry.getValue().size() -
1).getDataType();
long offsetOfChunkMetadataList = out.getPosition();
Statistics seriesStatistics = Statistics.getStatsByType(dataType);
int chunkMetadataListLength = 0;
// flush chunkMetadataList one by one
for (ChunkMetadata chunkMetadata : entry.getValue()) {
+ if (!chunkMetadata.getDataType().equals(dataType)) {
+ continue;
+ }
chunkMetadataListLength +=
chunkMetadata.serializeTo(out.wrapAsStream());
seriesStatistics.mergeStatistics(chunkMetadata.getStatistics());
}
