This is an automated email from the ASF dual-hosted git repository. sunzesong pushed a commit to branch jira_836 in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit dff8a616aa76e671ffc763ff3145f2d6b26fb6a9 Author: samperson1997 <[email protected]> AuthorDate: Thu Aug 20 15:42:44 2020 +0800 [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 | 19 ++++++---- .../writelog/recover/TsFileRecoverPerformer.java | 5 +++ .../db/integration/IoTDBDeleteTimeseriesIT.java | 43 ++++++++++++++++------ .../iotdb/tsfile/write/writer/TsFileIOWriter.java | 5 ++- 6 files changed, 67 insertions(+), 20 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..a0f0a0c 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 @@ -1065,13 +1065,18 @@ public class PlanExecutor implements IPlanExecutor { * @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); + throws QueryProcessException, StorageEngineException { + for (Path path : pathList) { + try { + if (!mManager.isPathExist(path.getFullPath())) { + throw new QueryProcessException( + String.format("Time series %s does not exist.", path.getFullPath())); + } + mManager.getStorageGroupName(path.getFullPath()); + StorageEngine.getInstance().deleteTimeseries(path.getDevice(), path.getMeasurement()); + } catch (MetadataException | StorageEngineException e) { + throw new QueryProcessException(e); + } } } 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..1062c9a 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,19 @@ 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(); } @@ -47,20 +54,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 +80,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 +100,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()); }
