You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2019/06/23 02:17:42 UTC
[incubator-iotdb] branch feature_async_close_tsfile updated: add
delete half
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch feature_async_close_tsfile
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/feature_async_close_tsfile by this push:
new 5305a37 add delete half
5305a37 is described below
commit 5305a37e938b333ca18d3afc64589ab6d2dc0026
Author: qiaojialin <64...@qq.com>
AuthorDate: Sun Jun 23 10:17:20 2019 +0800
add delete half
---
.../db/engine/filenodeV2/FileNodeManagerV2.java | 18 +++--
.../db/engine/filenodeV2/FileNodeProcessorV2.java | 92 +++++++++++++++++++++-
.../db/engine/filenodeV2/TsFileResourceV2.java | 9 ++-
.../filenodeV2/UnsealedTsFileProcessorV2.java | 17 ++--
.../db/qp/executor/IQueryProcessExecutor.java | 6 +-
.../iotdb/db/qp/executor/OverflowQPExecutor.java | 7 +-
.../iotdb/db/qp/executor/QueryProcessExecutor.java | 23 +++---
7 files changed, 137 insertions(+), 35 deletions(-)
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeManagerV2.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeManagerV2.java
index a33d548..8321cb2 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeManagerV2.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeManagerV2.java
@@ -28,6 +28,7 @@ import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.io.FileUtils;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.bufferwrite.BufferWriteProcessor;
import org.apache.iotdb.db.engine.filenode.TsFileResource;
import org.apache.iotdb.db.engine.querycontext.QueryDataSourceV2;
import org.apache.iotdb.db.exception.FileNodeManagerException;
@@ -36,6 +37,7 @@ import org.apache.iotdb.db.exception.PathErrorException;
import org.apache.iotdb.db.exception.ProcessorException;
import org.apache.iotdb.db.exception.StartupException;
import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.service.IService;
@@ -115,7 +117,7 @@ public class FileNodeManagerV2 implements IService {
private FileNodeProcessorV2 getProcessor(String devicePath)
- throws FileNodeManagerException, ProcessorException {
+ throws FileNodeManagerException {
String storageGroup = "";
try {
// return the storage group name
@@ -137,11 +139,8 @@ public class FileNodeManagerV2 implements IService {
}
}
return processor;
- } catch (PathErrorException e) {
- LOGGER.error("MManager get storage group name error, seriesPath is {}", devicePath);
- throw new FileNodeManagerException(e);
- } catch (FileNodeProcessorException e) {
- LOGGER.error("Fail to init simple file version controller of file node {}", storageGroup, e);
+ } catch (PathErrorException | ProcessorException e) {
+ LOGGER.error("Fail to get FileNodeProcessor {}", storageGroup, e);
throw new FileNodeManagerException(e);
}
}
@@ -213,7 +212,12 @@ public class FileNodeManagerV2 implements IService {
*/
public void delete(String deviceId, String measurementId, long timestamp)
throws FileNodeManagerException {
- // TODO
+ FileNodeProcessorV2 fileNodeProcessor = getProcessor(deviceId);
+ try {
+ fileNodeProcessor.delete(deviceId, measurementId, timestamp);
+ } catch (IOException e) {
+ throw new FileNodeManagerException(e);
+ }
}
private void delete(String processorName,
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeProcessorV2.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeProcessorV2.java
index 47e19ab..8895e6c 100755
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeProcessorV2.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeProcessorV2.java
@@ -29,11 +29,15 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Supplier;
+import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.conf.directories.DirectoryManager;
import org.apache.iotdb.db.engine.filenode.CopyOnReadLinkedList;
+import org.apache.iotdb.db.engine.modification.Deletion;
+import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.querycontext.QueryDataSourceV2;
import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
import org.apache.iotdb.db.engine.version.SimpleFileVersionController;
@@ -41,6 +45,7 @@ import org.apache.iotdb.db.engine.version.VersionController;
import org.apache.iotdb.db.exception.FileNodeProcessorException;
import org.apache.iotdb.db.exception.ProcessorException;
import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.writelog.recover.SeqTsFileRecoverPerformer;
import org.apache.iotdb.db.writelog.recover.UnSeqTsFileRecoverPerformer;
@@ -97,6 +102,13 @@ public class FileNodeProcessorV2 {
private VersionController versionController;
+ private ReentrantLock mergeDeleteLock = new ReentrantLock();
+
+ /**
+ * This is the modification file of the result of the current merge.
+ */
+ private ModificationFile mergingModification;
+
public FileNodeProcessorV2(String baseDir, String storageGroupName) throws ProcessorException {
this.storageGroupName = storageGroupName;
lock = new ReentrantReadWriteLock();
@@ -290,7 +302,6 @@ public class FileNodeProcessorV2 {
// TODO need a read lock, please consider the concurrency with flush manager threads.
public QueryDataSourceV2 query(String deviceId, String measurementId) {
-
lock.readLock().lock();
try {
List<TsFileResourceV2> seqResources = getFileReSourceListForQuery(sequenceFileList,
@@ -303,7 +314,6 @@ public class FileNodeProcessorV2 {
}
}
-
/**
* @param tsFileResources includes sealed and unsealed tsfile resources
* @return fill unsealed tsfile resources with memory data and ChunkMetadataList of data in disk
@@ -335,6 +345,84 @@ public class FileNodeProcessorV2 {
/**
+ * Delete data whose timestamp <= 'timestamp' and belong to timeseries deviceId.measurementId.
+ *
+ * @param deviceId the deviceId of the timeseries to be deleted.
+ * @param measurementId the measurementId of the timeseries to be deleted.
+ * @param timestamp the delete range is (0, timestamp].
+ */
+ public void delete(String deviceId, String measurementId, long timestamp) throws IOException {
+ // TODO: how to avoid partial deletion?
+ mergeDeleteLock.lock();
+ long lastUpdateTime = latestTimeForEachDevice.get(deviceId);
+ // no tsfile data, the delete operation is invalid
+ if (lastUpdateTime == Long.MIN_VALUE) {
+ LOGGER.warn("The last update time is -1, delete overflow is invalid, "
+ + "the storage group is {}", storageGroupName);
+ return;
+ }
+
+ // write log
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
+ workSequenceTsFileProcessor.getLogNode().write(new DeletePlan(timestamp, new Path(deviceId, measurementId)));
+ workUnSequenceTsFileProcessor.getLogNode().write(new DeletePlan(timestamp, new Path(deviceId, measurementId)));
+ }
+
+ long version = versionController.nextVersion();
+
+ // record what files are updated so we can roll back them in case of exception
+ List<ModificationFile> updatedModFiles = new ArrayList<>();
+
+ try {
+ String fullPath = deviceId + IoTDBConstant.PATH_SEPARATOR + measurementId;
+ Deletion deletion = new Deletion(fullPath, version, timestamp);
+ if (mergingModification != null) {
+ mergingModification.write(deletion);
+ updatedModFiles.add(mergingModification);
+ }
+
+ // delete sequence files
+ for (TsFileResourceV2 tsFileResource : sequenceFileList) {
+ if (!tsFileResource.containsDevice(deviceId) ||
+ tsFileResource.getStartTimeMap().get(deviceId) <= deletion.getTimestamp()) {
+ continue;
+ }
+ if (tsFileResource.isClosed()) {
+ tsFileResource.getModFile().write(deletion);
+ updatedModFiles.add(tsFileResource.getModFile());
+ } else {
+ // TODO delete unsealed file
+ }
+ }
+
+ // delete unSequence files
+ for (TsFileResourceV2 tsFileResource : unSequenceFileList) {
+ if (!tsFileResource.containsDevice(deviceId) ||
+ tsFileResource.getStartTimeMap().get(deviceId) <= deletion.getTimestamp()) {
+ continue;
+ }
+ if (tsFileResource.isClosed()) {
+ tsFileResource.getModFile().write(deletion);
+ updatedModFiles.add(tsFileResource.getModFile());
+ } else {
+ // TODO delete unsealed file
+ }
+ }
+
+
+ } catch (Exception e) {
+ // roll back
+ for (ModificationFile modFile : updatedModFiles) {
+ modFile.abort();
+ }
+ throw new IOException(e);
+ } finally {
+ mergeDeleteLock.unlock();
+ }
+ }
+
+
+ /**
* ensure there must be a flush thread submitted after setCloseMark() is called, therefore the setCloseMark task
* will be executed by a flush thread.
*
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/TsFileResourceV2.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/TsFileResourceV2.java
index 0fd2e31..f35bed3 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/TsFileResourceV2.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/TsFileResourceV2.java
@@ -97,10 +97,17 @@ public class TsFileResourceV2 {
return readOnlyMemChunk;
}
- public ModificationFile getModFile() {
+ public synchronized ModificationFile getModFile() {
+ if (modFile == null) {
+ modFile = new ModificationFile(file.getAbsolutePath() + ModificationFile.FILE_SUFFIX);
+ }
return modFile;
}
+ public boolean containsDevice(String deviceId) {
+ return startTimeMap.containsKey(deviceId);
+ }
+
public File getFile() {
return file;
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/UnsealedTsFileProcessorV2.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/UnsealedTsFileProcessorV2.java
index 2a72e07..016e9b1 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/UnsealedTsFileProcessorV2.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/UnsealedTsFileProcessorV2.java
@@ -120,13 +120,14 @@ public class UnsealedTsFileProcessorV2 {
}
}
- try {
- logNode.write(insertPlan);
- } catch (IOException e) {
- LOGGER.error("write WAL failed", e);
- return false;
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
+ try {
+ logNode.write(insertPlan);
+ } catch (IOException e) {
+ LOGGER.error("write WAL failed", e);
+ return false;
+ }
}
-
// update start time of this memtable
tsFileResource.updateStartTime(insertPlan.getDeviceId(), insertPlan.getTime());
@@ -297,6 +298,10 @@ public class UnsealedTsFileProcessorV2 {
return managedByFlushManager;
}
+ public WriteLogNode getLogNode() {
+ return logNode;
+ }
+
public void setManagedByFlushManager(boolean managedByFlushManager) {
this.managedByFlushManager = managedByFlushManager;
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/IQueryProcessExecutor.java b/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/IQueryProcessExecutor.java
index 8cf82af..2ffab93 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/IQueryProcessExecutor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/IQueryProcessExecutor.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.db.exception.FileNodeManagerException;
import org.apache.iotdb.db.exception.PathErrorException;
import org.apache.iotdb.db.exception.ProcessorException;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.fill.IFill;
@@ -91,11 +92,10 @@ public interface IQueryProcessExecutor {
/**
* execute delete command and return whether the operator is successful.
*
- * @param paths : delete series paths
- * @param deleteTime end time in delete command
+ * @param deletePlan physical delete plan
* @return - whether the operator is successful.
*/
- boolean delete(List<Path> paths, long deleteTime) throws ProcessorException;
+ boolean delete(DeletePlan deletePlan) throws ProcessorException;
/**
* execute delete command and return whether the operator is successful.
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/OverflowQPExecutor.java b/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/OverflowQPExecutor.java
index 9735319..5ce697a 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/OverflowQPExecutor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/OverflowQPExecutor.java
@@ -113,8 +113,7 @@ public class OverflowQPExecutor extends QueryProcessExecutor {
public boolean processNonQuery(PhysicalPlan plan) throws ProcessorException {
switch (plan.getOperatorType()) {
case DELETE:
- DeletePlan delete = (DeletePlan) plan;
- return delete(delete.getPaths(), delete.getDeleteTime());
+ return delete((DeletePlan) plan);
case UPDATE:
UpdatePlan update = (UpdatePlan) plan;
boolean flag = true;
@@ -216,7 +215,7 @@ public class OverflowQPExecutor extends QueryProcessExecutor {
try {
String fullPath = deviceId + "." + measurementId;
if (!mManager.pathExist(fullPath)) {
- throw new ProcessorException(String.format("Timeseries %s does not exist.", fullPath));
+ throw new ProcessorException(String.format("Time series %s does not exist.", fullPath));
}
mManager.getFileNameByPath(fullPath);
TSDataType dataType = mManager.getSeriesType(fullPath);
@@ -235,7 +234,7 @@ public class OverflowQPExecutor extends QueryProcessExecutor {
try {
if (!mManager.pathExist(path.getFullPath())) {
throw new ProcessorException(
- String.format("Timeseries %s does not exist.", path.getFullPath()));
+ String.format("Time series %s does not exist.", path.getFullPath()));
}
mManager.getFileNameByPath(path.getFullPath());
fileNodeManager.delete(deviceId, measurementId, timestamp);
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/QueryProcessExecutor.java b/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/QueryProcessExecutor.java
index 94939e9..6ad83ef 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/QueryProcessExecutor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/qp/executor/QueryProcessExecutor.java
@@ -29,6 +29,7 @@ import org.apache.iotdb.db.exception.ProcessorException;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.qp.physical.crud.AggregationPlan;
+import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.FillQueryPlan;
import org.apache.iotdb.db.qp.physical.crud.GroupByPlan;
import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
@@ -101,27 +102,25 @@ public abstract class QueryProcessExecutor implements IQueryProcessExecutor {
}
@Override
- public boolean delete(List<Path> paths, long deleteTime) throws ProcessorException {
+ public boolean delete(DeletePlan deletePlan) throws ProcessorException {
try {
boolean result = true;
MManager mManager = MManager.getInstance();
- Set<String> pathSet = new HashSet<>();
- for (Path p : paths) {
- pathSet.addAll(mManager.getPaths(p.getFullPath()));
+ Set<String> existingPaths = new HashSet<>();
+ for (Path p : deletePlan.getPaths()) {
+ existingPaths.addAll(mManager.getPaths(p.getFullPath()));
}
- if (pathSet.isEmpty()) {
- throw new ProcessorException("TimeSeries does not exist and cannot be delete data");
+ if (existingPaths.isEmpty()) {
+ throw new ProcessorException("TimeSeries does not exist and its data cannot be deleted");
}
- for (String onePath : pathSet) {
+ for (String onePath : existingPaths) {
if (!mManager.pathExist(onePath)) {
throw new ProcessorException(
- String.format("TimeSeries %s does not exist and cannot be delete its data", onePath));
+ String.format("TimeSeries %s does not exist and its data cannot be deleted", onePath));
}
}
- List<String> fullPath = new ArrayList<>();
- fullPath.addAll(pathSet);
- for (String path : fullPath) {
- result &= delete(new Path(path), deleteTime);
+ for (String path : existingPaths) {
+ result &= delete(new Path(path), deletePlan.getDeleteTime());
}
return result;
} catch (PathErrorException e) {