This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch rel/0.10
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/rel/0.10 by this push:
new d766679 [IOTDB-836] [To rel/0.10] Fix merging statistics failed when
endFile (#1624)
d766679 is described below
commit d766679cd23c67ce780a0eb2f6f29bdc6820a343
Author: Zesong Sun <[email protected]>
AuthorDate: Sat Aug 22 16:41:26 2020 +0800
[IOTDB-836] [To rel/0.10] Fix merging statistics failed when endFile (#1624)
---
.../org/apache/iotdb/db/engine/StorageEngine.java | 13 +++
.../iotdb/db/engine/memtable/AbstractMemTable.java | 1 +
.../db/engine/storagegroup/TsFileProcessor.java | 1 +
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 17 +--
.../writelog/recover/TsFileRecoverPerformer.java | 5 +
.../db/integration/IoTDBDeleteTimeseriesIT.java | 115 +++++++++++++++++++++
.../iotdb/tsfile/write/writer/TsFileIOWriter.java | 8 +-
7 files changed, 142 insertions(+), 18 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 65f99d4..d59744f 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
@@ -361,6 +361,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.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 bf4fc9c..00dbb55 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
@@ -230,6 +230,7 @@ public abstract class AbstractMemTable implements IMemTable
{
if (chunk == null) {
return;
}
+ // If timestamp == Long.MAX_VALUE, it means that the whole timeseries is
deleted
if (timestamp == Long.MAX_VALUE) {
deviceMap.remove(measurementId);
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index 6d90ed8..7241c08 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -244,6 +244,7 @@ public class TsFileProcessor {
workMemTable
.delete(deletion.getDevice(), deletion.getMeasurement(),
deletion.getTimestamp());
}
+
// flushing memTables are immutable, only record this deletion in these
memTables for query
for (IMemTable memTable : flushingMemTables) {
memTable.delete(deletion);
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 973057a..875d339 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
@@ -1157,9 +1157,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);
@@ -1239,21 +1239,6 @@ public class PlanExecutor implements IPlanExecutor {
return true;
}
- /**
- * Delete all data of time series in pathList.
- *
- * @param pathList deleted paths
- */
- private void deleteDataOfTimeSeries(List<Path> pathList)
- throws QueryProcessException, StorageGroupNotSetException,
StorageEngineException {
- for (Path p : pathList) {
- DeletePlan deletePlan = new DeletePlan();
- deletePlan.addPath(p);
- deletePlan.setDeleteTime(Long.MAX_VALUE);
- processNonQuery(deletePlan);
- }
- }
-
private QueryDataSet processAuthorQuery(AuthorPlan plan) throws
QueryProcessException {
AuthorType authorType = plan.getAuthorType();
String userName = plan.getUserName();
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 38ab3c2..707a403 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
@@ -41,6 +41,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.write.writer.RestorableTsFileIOWriter;
@@ -177,7 +178,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;
+ }
resource.updateStartTime(deviceId, chunkMetaData.getStartTime());
resource.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
new file mode 100644
index 0000000..a60a066
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBDeleteTimeseriesIT.java
@@ -0,0 +1,115 @@
+/*
+ * 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;
+
+import java.sql.Connection;
+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;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class IoTDBDeleteTimeseriesIT {
+
+ private long memtableSizeThreshold;
+
+ @Before
+ public void setUp() throws Exception {
+ EnvironmentUtils.closeStatMonitor();
+ EnvironmentUtils.envSetUp();
+ memtableSizeThreshold =
IoTDBDescriptor.getInstance().getConfig().getMemtableSizeThreshold();
+ IoTDBDescriptor.getInstance().getConfig().setMemtableSizeThreshold(16);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+
IoTDBDescriptor.getInstance().getConfig().setMemtableSizeThreshold(memtableSizeThreshold);
+ EnvironmentUtils.cleanEnv();
+ }
+
+ @Test
+ public void testDeleteTimeseries() throws Exception {
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ String[] retArray = new String[]{
+ "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.execute(
+ "create timeseries root.turbine1.d1.s1 with datatype=INT64,
encoding=PLAIN, compression=SNAPPY");
+ statement.execute(
+ "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();
+ while (resultSet.next()) {
+ StringBuilder builder = new StringBuilder();
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ builder.append(resultSet.getString(i)).append(",");
+ }
+ Assert.assertEquals(retArray[cnt], builder.toString());
+ cnt++;
+ }
+ }
+ statement.execute("DELETE timeseries root.turbine1.d1.s1");
+ statement.execute(
+ "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();
+ while (resultSet.next()) {
+ StringBuilder builder = new StringBuilder();
+ for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+ builder.append(resultSet.getString(i)).append(",");
+ }
+ Assert.assertEquals(retArray[cnt], builder.toString());
+ cnt++;
+ }
+ }
+ }
+
+ 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);
+ }
+ }
+
+}
\ No newline at end of file
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 c6b1726..abd69f9 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
@@ -87,7 +87,7 @@ public class TsFileIOWriter {
private String currentChunkGroupDeviceId;
private long currentChunkGroupStartOffset;
protected List<Pair<Long, Long>> versionInfo = new ArrayList<>();
-
+
// for upgrade tool
Map<String, List<TimeseriesMetadata>> deviceTimeseriesMetadataMap;
@@ -289,10 +289,11 @@ public class TsFileIOWriter {
for (Map.Entry<Path, List<ChunkMetadata>> entry :
chunkMetadataListMap.entrySet()) {
Path path = entry.getKey();
String device = path.getDevice();
+
// create TimeseriesMetaData
TimeseriesMetadata timeseriesMetaData = new TimeseriesMetadata();
timeseriesMetaData.setMeasurementId(path.getMeasurement());
- TSDataType dataType = entry.getValue().get(0).getDataType();
+ TSDataType dataType = entry.getValue().get(entry.getValue().size() -
1).getDataType();
timeseriesMetaData.setTSDataType(dataType);
timeseriesMetaData.setOffsetOfChunkMetaDataList(out.getPosition());
@@ -300,6 +301,9 @@ public class TsFileIOWriter {
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());
}
