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 2020/06/29 06:02:20 UTC
[incubator-iotdb] branch master updated: Support partition removal
(#1409)
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 fe913c9 Support partition removal (#1409)
fe913c9 is described below
commit fe913c9bcd62aae2b14db7f8aa8cd647b9143e17
Author: Jiang Tian <jt...@163.com>
AuthorDate: Mon Jun 29 01:02:08 2020 -0500
Support partition removal (#1409)
* support partition removal
---
.../DML Data Manipulation Language.md | 15 ++
docs/UserGuide/Operation Manual/SQL Reference.md | 8 +
.../DML Data Manipulation Language.md | 12 ++
.../zh/UserGuide/Operation Manual/SQL Reference.md | 8 +
.../org/apache/iotdb/db/qp/strategy/SqlBase.g4 | 5 +
.../org/apache/iotdb/db/engine/StorageEngine.java | 17 ++
.../engine/storagegroup/StorageGroupProcessor.java | 54 ++++++
.../db/engine/storagegroup/TsFileResource.java | 3 +
.../main/java/org/apache/iotdb/db/qp/Planner.java | 1 +
.../apache/iotdb/db/qp/constant/SQLConstant.java | 4 +-
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 9 +
.../org/apache/iotdb/db/qp/logical/Operator.java | 2 +-
.../db/qp/logical/sys/DeletePartitionOperator.java | 50 ++++++
.../db/qp/physical/crud/DeletePartitionPlan.java | 51 ++++++
.../iotdb/db/qp/strategy/LogicalGenerator.java | 16 ++
.../iotdb/db/qp/strategy/PhysicalGenerator.java | 5 +
.../db/integration/IoTDBRemovePartitionIT.java | 193 +++++++++++++++++++++
17 files changed, 451 insertions(+), 2 deletions(-)
diff --git a/docs/UserGuide/Operation Manual/DML Data Manipulation Language.md b/docs/UserGuide/Operation Manual/DML Data Manipulation Language.md
index 1f5551f..6746033 100644
--- a/docs/UserGuide/Operation Manual/DML Data Manipulation Language.md
+++ b/docs/UserGuide/Operation Manual/DML Data Manipulation Language.md
@@ -800,3 +800,18 @@ It should be noted that when the deleted path does not exist, IoTDB will give th
IoTDB> delete from root.ln.wf03.wt02.status where time < now()
Msg: TimeSeries does not exist and its data cannot be deleted
```
+
+## Delete Time Partition (experimental)
+You may delete all data in a time partition of a storage group using the following grammar:
+
+```
+DELETE PARTITION root.ln 0,1,2
+```
+
+The `0,1,2` above is the id of the partition that is to be deleted, you can find it from the IoTDB
+data folders or convert a timestamp manually to an id using `timestamp / partitionInterval
+` (flooring), and the `partitionInterval` should be in your config (if time-partitioning is
+supported in your version).
+
+Please notice that this function is experimental and mainly for development, please use it with
+extreme care.
\ No newline at end of file
diff --git a/docs/UserGuide/Operation Manual/SQL Reference.md b/docs/UserGuide/Operation Manual/SQL Reference.md
index 359273f..a2e7d63 100644
--- a/docs/UserGuide/Operation Manual/SQL Reference.md
+++ b/docs/UserGuide/Operation Manual/SQL Reference.md
@@ -942,6 +942,14 @@ atomicity of data deletion is not guaranteed for efficiency concerns. So we reco
not change the TTL once it is set or at least do not reset it frequently, unless you are determined
to suffer the unpredictability.
+* Delete Partition (experimental)
+```
+DELETE PARTITION StorageGroupName INT(COMMA INT)*
+Eg DELETE PARTITION root.sg1 0,1,2
+This example will delete the first 3 time partitions of storage group root.sg1.
+```
+The partitionId can be found in data folders or converted using `timestamp / partitionInterval`.
+
# Reference
## Keywords
diff --git a/docs/zh/UserGuide/Operation Manual/DML Data Manipulation Language.md b/docs/zh/UserGuide/Operation Manual/DML Data Manipulation Language.md
index 21a9828..fb370c5 100644
--- a/docs/zh/UserGuide/Operation Manual/DML Data Manipulation Language.md
+++ b/docs/zh/UserGuide/Operation Manual/DML Data Manipulation Language.md
@@ -863,3 +863,15 @@ delete from root.ln.wf02.wt02.* where time <= 2017-11-01T16:26:00;
IoTDB> delete from root.ln.wf03.wt02.status where time < now()
Msg: TimeSeries does not exist and its data cannot be deleted
```
+
+## 删除时间分区 (实验性功能)
+您可以通过如下语句来删除某一个存储组下的指定时间分区:
+
+```
+DELETE PARTITION root.ln 0,1,2
+```
+
+上例中的0,1,2为待删除时间分区的id，您可以通过查看IoTDB的数据文件夹找到它，或者可以通过计算`timestamp / partitionInterval`(向下取整),
+手动地将一个时间戳转换为对应的id，其中的`partitionInterval`可以在IoTDB的配置文件中找到（如果您使用的版本支持时间分区）。
+
+请注意该功能目前只是实验性的，如果您不是开发者，使用时请务必谨慎。
\ No newline at end of file
diff --git a/docs/zh/UserGuide/Operation Manual/SQL Reference.md b/docs/zh/UserGuide/Operation Manual/SQL Reference.md
index 2655ecb..60806b9 100644
--- a/docs/zh/UserGuide/Operation Manual/SQL Reference.md
+++ b/docs/zh/UserGuide/Operation Manual/SQL Reference.md
@@ -903,6 +903,14 @@ Eg.2 SHOW TTL ON root.group1,root.group2,root.group3
一部分之前不可见的数据可能重新可见，而那些已经被物理删除的数据则将永久丢失。也就是说，TTL操作不会原子性地删除
对应的数据。因此我们不推荐您频繁修改TTL，除非您能接受该操作带来的一定程度的不可预知性。
+* 删除时间分区 (实验性功能)
+```
+DELETE PARTITION StorageGroupName INT(COMMA INT)*
+Eg DELETE PARTITION root.sg1 0,1,2
+该例子将删除存储组root.sg1的前三个时间分区
+```
+partitionId 可以通过查看数据文件夹获取，或者是计算 `timestamp / partitionInterval`得到。
+
# 参考
## 关键字
diff --git a/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4 b/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
index e6d6329..e9ae14e 100644
--- a/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
+++ b/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
@@ -82,6 +82,7 @@ statement
| LOAD STRING_LITERAL autoCreateSchema? #loadFiles
| REMOVE STRING_LITERAL #removeFile
| MOVE STRING_LITERAL STRING_LITERAL #moveFile
+ | DELETE PARTITION prefixPath INT(COMMA INT)* #deletePartition
| CREATE SNAPSHOT FOR SCHEMA #createSnapshot
| SELECT INDEX func=ID //not support yet
LR_BRACKET
@@ -934,6 +935,10 @@ LATEST
: L A T E S T
;
+PARTITION
+ : P A R T I T I O N
+ ;
+
SNAPSHOT
: S N A P S H O T
;
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 60eda4b..95a854f 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
@@ -44,6 +44,7 @@ import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy;
import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy.DirectFlushPolicy;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor;
+import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.TimePartitionFilter;
import org.apache.iotdb.db.engine.storagegroup.TsFileProcessor;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.BatchInsertionException;
@@ -66,6 +67,7 @@ import org.apache.iotdb.db.query.control.QueryFileManager;
import org.apache.iotdb.db.service.IService;
import org.apache.iotdb.db.service.ServiceType;
import org.apache.iotdb.db.utils.FilePathUtils;
+import org.apache.iotdb.db.utils.TestOnly;
import org.apache.iotdb.db.utils.UpgradeUtils;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.expression.impl.SingleSeriesExpression;
@@ -612,4 +614,19 @@ public class StorageEngine implements IService {
throws StorageEngineException {
getProcessor(storageGroup).setPartitionFileVersionToMax(partitionId, newMaxVersion);
}
+
+ public void removePartitions(String storageGroupName, TimePartitionFilter filter)
+ throws StorageEngineException {
+ getProcessor(storageGroupName).removePartitions(filter);
+ }
+
+ @TestOnly
+ public static void setTimePartitionInterval(long timePartitionInterval) {
+ StorageEngine.timePartitionInterval = timePartitionInterval;
+ }
+
+ @TestOnly
+ public static void setEnablePartition(boolean enablePartition) {
+ StorageEngine.enablePartition = enablePartition;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
index d53b7e8..f8dd135 100755
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
@@ -2410,4 +2410,58 @@ public class StorageGroupProcessor {
void call(TsFileResource caller);
}
+
+ /**
+ * remove all partitions that satisfy a filter.
+ * @param filter
+ */
+ public void removePartitions(TimePartitionFilter filter) {
+ // this requires blocking all other activities
+ insertLock.writeLock().lock();
+ mergeLock.writeLock().lock();
+ try {
+ // abort ongoing merges
+ MergeManager.getINSTANCE().abortMerge(storageGroupName);
+ // close all working files that should be removed
+ removePartitions(filter, workSequenceTsFileProcessors.entrySet());
+ removePartitions(filter, workUnsequenceTsFileProcessors.entrySet());
+
+ // remove data files
+ removePartitions(filter, sequenceFileTreeSet.iterator());
+ removePartitions(filter, unSequenceFileList.iterator());
+
+ } finally {
+ insertLock.writeLock().unlock();
+ mergeLock.writeLock().unlock();
+ }
+ }
+
+ //may remove the processorEntrys
+ private void removePartitions(TimePartitionFilter filter, Set<Entry<Long, TsFileProcessor>> processorEntrys) {
+ for (Iterator<Entry<Long, TsFileProcessor>> iterator = processorEntrys.iterator(); iterator.hasNext(); ) {
+ Entry<Long, TsFileProcessor> longTsFileProcessorEntry = iterator.next();
+ long partitionId = longTsFileProcessorEntry.getKey();
+ TsFileProcessor processor = longTsFileProcessorEntry.getValue();
+ if (filter.satisfy(storageGroupName, partitionId)) {
+ processor.syncClose();
+ iterator.remove();
+ }
+ }
+ }
+
+ //may remove the iterator's data
+ private void removePartitions(TimePartitionFilter filter, Iterator<TsFileResource> iterator) {
+ while ( iterator.hasNext()) {
+ TsFileResource tsFileResource = iterator.next();
+ if (filter.satisfy(storageGroupName, tsFileResource.getTimePartition())) {
+ tsFileResource.remove();
+ iterator.remove();
+ }
+ }
+ }
+
+ @FunctionalInterface
+ public interface TimePartitionFilter {
+ boolean satisfy(String storageGroupName, long timePartitionId);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
index 2a93991..c8a0bca 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
@@ -539,6 +539,9 @@ public class TsFileResource {
modFile = null;
}
+ /**
+ * Remove the data file, its resource file, and its modification file physically.
+ */
public void remove() {
file.delete();
fsFactory.getFile(file.getPath() + RESOURCE_SUFFIX).delete();
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/Planner.java b/server/src/main/java/org/apache/iotdb/db/qp/Planner.java
index 1601102..05a41c1 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/Planner.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/Planner.java
@@ -101,6 +101,7 @@ public class Planner {
case CLEAR_CACHE:
case NULL:
case SHOW_MERGE_STATUS:
+ case DELETE_PARTITION:
case CREATE_SCHEMA_SNAPSHOT:
return operator;
case QUERY:
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java b/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
index 3ae7b21..c2aea65 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
@@ -151,8 +151,9 @@ public class SQLConstant {
public static final int TOK_LOAD_CONFIGURATION_LOCAL = 86;
public static final int TOK_SHOW_MERGE_STATUS = 87;
+ public static final int TOK_DELETE_PARTITION = 88;
- public static final int TOK_CREATE_SCHEMA_SNAPSHOT = 88;
+ public static final int TOK_CREATE_SCHEMA_SNAPSHOT = 89;
public static final int TOK_TRACING = 91;
public static final Map<Integer, String> tokenSymbol = new HashMap<>();
@@ -224,6 +225,7 @@ public class SQLConstant {
tokenNames.put(TOK_MOVE_FILE, "TOK_MOVE_FILE");
tokenNames.put(TOK_SHOW_MERGE_STATUS, "TOK_SHOW_MERGE_STATUS");
+ tokenNames.put(TOK_DELETE_PARTITION, "TOK_DELETE_PARTITION");
tokenNames.put(TOK_TRACING, "TOK_TRACING");
}
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 adb0b04..761d27b 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
@@ -35,6 +35,7 @@ import org.apache.iotdb.db.engine.cache.TimeSeriesMetadataCache;
import org.apache.iotdb.db.engine.flush.pool.FlushTaskPoolManager;
import org.apache.iotdb.db.engine.merge.manage.MergeManager;
import org.apache.iotdb.db.engine.merge.manage.MergeManager.TaskStatus;
+import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.TimePartitionFilter;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.DeleteFailedException;
@@ -198,6 +199,14 @@ public class PlanExecutor implements IPlanExecutor {
case CLEAR_CACHE:
operateClearCache((ClearCachePlan) plan);
return true;
+ case DELETE_PARTITION:
+ DeletePartitionPlan p = (DeletePartitionPlan) plan;
+ TimePartitionFilter filter =
+ (storageGroupName, partitionId) ->
+ storageGroupName.equals(((DeletePartitionPlan) plan).getStorageGroupName())
+ && p.getPartitionId().contains(partitionId);
+ StorageEngine.getInstance().removePartitions(((DeletePartitionPlan) plan).getStorageGroupName(), filter);
+ return true;
case CREATE_SCHEMA_SNAPSHOT:
operateCreateSnapshot();
return true;
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
index 42f5345..90cd5d2 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
@@ -76,6 +76,6 @@ public abstract class Operator {
GRANT_WATERMARK_EMBEDDING, REVOKE_WATERMARK_EMBEDDING,
TTL, DELETE_STORAGE_GROUP, LOAD_CONFIGURATION, SHOW, LOAD_FILES, REMOVE_FILE, MOVE_FILE, LAST, GROUP_BY_FILL,
ALTER_TIMESERIES, FLUSH, MERGE, FULL_MERGE, CLEAR_CACHE,
- SHOW_MERGE_STATUS, CREATE_SCHEMA_SNAPSHOT, TRACING
+ SHOW_MERGE_STATUS, CREATE_SCHEMA_SNAPSHOT, TRACING, DELETE_PARTITION
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/DeletePartitionOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/DeletePartitionOperator.java
new file mode 100644
index 0000000..33e2345
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/DeletePartitionOperator.java
@@ -0,0 +1,50 @@
+/*
+ * 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.qp.logical.sys;
+
+import java.util.Set;
+import org.apache.iotdb.db.qp.logical.RootOperator;
+
+public class DeletePartitionOperator extends RootOperator {
+
+ private String storageGroupName;
+ private Set<Long> partitionIds;
+
+ public DeletePartitionOperator(int tokenIntType) {
+ super(tokenIntType);
+ operatorType = OperatorType.DELETE_PARTITION;
+ }
+
+ public void setStorageGroupName(String storageGroupName) {
+ this.storageGroupName = storageGroupName;
+ }
+
+ public void setPartitionIds(Set<Long> partitionIds) {
+ this.partitionIds = partitionIds;
+ }
+
+ public String getStorageGroupName() {
+ return storageGroupName;
+ }
+
+ public Set<Long> getPartitionId() {
+ return partitionIds;
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePartitionPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePartitionPlan.java
new file mode 100644
index 0000000..c720d27
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePartitionPlan.java
@@ -0,0 +1,51 @@
+/*
+ * 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.qp.physical.crud;
+
+import java.util.List;
+import java.util.Set;
+import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.tsfile.read.common.Path;
+
+public class DeletePartitionPlan extends PhysicalPlan {
+
+ private String storageGroupName;
+ private Set<Long> partitionId;
+
+ public DeletePartitionPlan(String storageGroupName, Set<Long> partitionId) {
+ super(false, OperatorType.DELETE_PARTITION);
+ this.storageGroupName = storageGroupName;
+ this.partitionId = partitionId;
+ }
+
+ @Override
+ public List<Path> getPaths() {
+ return null;
+ }
+
+ public String getStorageGroupName() {
+ return storageGroupName;
+ }
+
+ public Set<Long> getPartitionId() {
+ return partitionId;
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java b/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
index 6b14f26..a9fd1c0 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
@@ -51,6 +51,7 @@ import org.apache.iotdb.db.qp.logical.sys.CountOperator;
import org.apache.iotdb.db.qp.logical.sys.CreateSnapshotOperator;
import org.apache.iotdb.db.qp.logical.sys.CreateTimeSeriesOperator;
import org.apache.iotdb.db.qp.logical.sys.DataAuthOperator;
+import org.apache.iotdb.db.qp.logical.sys.DeletePartitionOperator;
import org.apache.iotdb.db.qp.logical.sys.DeleteStorageGroupOperator;
import org.apache.iotdb.db.qp.logical.sys.DeleteTimeSeriesOperator;
import org.apache.iotdb.db.qp.logical.sys.FlushOperator;
@@ -84,6 +85,7 @@ import org.apache.iotdb.db.qp.strategy.SqlBaseParser.CreateSnapshotContext;
import org.apache.iotdb.db.qp.strategy.SqlBaseParser.CreateTimeseriesContext;
import org.apache.iotdb.db.qp.strategy.SqlBaseParser.CreateUserContext;
import org.apache.iotdb.db.qp.strategy.SqlBaseParser.DateExpressionContext;
+import org.apache.iotdb.db.qp.strategy.SqlBaseParser.DeletePartitionContext;
import org.apache.iotdb.db.qp.strategy.SqlBaseParser.DeleteStatementContext;
import org.apache.iotdb.db.qp.strategy.SqlBaseParser.DeleteStorageGroupContext;
import org.apache.iotdb.db.qp.strategy.SqlBaseParser.DeleteTimeseriesContext;
@@ -1569,6 +1571,20 @@ public class LogicalGenerator extends SqlBaseBaseListener {
}
@Override
+ public void enterDeletePartition(DeletePartitionContext ctx) {
+ super.enterDeletePartition(ctx);
+ DeletePartitionOperator deletePartitionOperator = new DeletePartitionOperator(
+ SQLConstant.TOK_DELETE_PARTITION);
+ deletePartitionOperator.setStorageGroupName(ctx.prefixPath().getText());
+ Set<Long> idSet = new HashSet<>();
+ for (TerminalNode terminalNode : ctx.INT()) {
+ idSet.add(Long.parseLong(terminalNode.getText()));
+ }
+ deletePartitionOperator.setPartitionIds(idSet);
+ initializedOperator = deletePartitionOperator;
+ }
+
+ @Override
public void enterCreateSnapshot(CreateSnapshotContext ctx) {
super.enterCreateSnapshot(ctx);
initializedOperator = new CreateSnapshotOperator(SQLConstant.TOK_CREATE_SCHEMA_SNAPSHOT);
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
index a72fcb0..6b5dc43 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
@@ -45,6 +45,7 @@ import org.apache.iotdb.db.qp.logical.sys.AuthorOperator;
import org.apache.iotdb.db.qp.logical.sys.CountOperator;
import org.apache.iotdb.db.qp.logical.sys.CreateTimeSeriesOperator;
import org.apache.iotdb.db.qp.logical.sys.DataAuthOperator;
+import org.apache.iotdb.db.qp.logical.sys.DeletePartitionOperator;
import org.apache.iotdb.db.qp.logical.sys.DeleteStorageGroupOperator;
import org.apache.iotdb.db.qp.logical.sys.DeleteTimeSeriesOperator;
import org.apache.iotdb.db.qp.logical.sys.FlushOperator;
@@ -65,6 +66,7 @@ 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.AlignByDevicePlan;
import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan.MeasurementType;
+import org.apache.iotdb.db.qp.physical.crud.DeletePartitionPlan;
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.GroupByTimeFillPlan;
@@ -264,6 +266,9 @@ public class PhysicalGenerator {
return new ClearCachePlan();
case SHOW_MERGE_STATUS:
return new ShowMergeStatusPlan();
+ case DELETE_PARTITION:
+ DeletePartitionOperator op = (DeletePartitionOperator) operator;
+ return new DeletePartitionPlan(op.getStorageGroupName(), op.getPartitionId());
case CREATE_SCHEMA_SNAPSHOT:
return new CreateSnapshotPlan();
default:
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBRemovePartitionIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBRemovePartitionIT.java
new file mode 100644
index 0000000..c8db426
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBRemovePartitionIT.java
@@ -0,0 +1,193 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.iotdb.db.engine.StorageEngine;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.jdbc.Config;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class IoTDBRemovePartitionIT {
+
+ private static int partitionInterval = 100;
+
+ @Before
+ public void setUp() throws Exception {
+ EnvironmentUtils.closeStatMonitor();
+ EnvironmentUtils.envSetUp();
+ StorageEngine.setEnablePartition(true);
+ StorageEngine.setTimePartitionInterval(partitionInterval);
+ insertData();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvironmentUtils.cleanEnv();
+ StorageEngine.setEnablePartition(false);
+ StorageEngine.setTimePartitionInterval(-1);
+ }
+
+ @Test
+ public void testRemoveNoPartition() throws StorageEngineException {
+ StorageEngine.getInstance().removePartitions("root.test1",
+ (storageGroupName, timePartitionId) -> false);
+
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM root.test1")) {
+ int count = 0;
+ while (resultSet.next()) {
+ assertEquals(count / 2 * 100 + count % 2 * 50, resultSet.getLong(1));
+ assertEquals(count / 2 * 100 + count % 2 * 50, resultSet.getLong(2));
+ count ++;
+ }
+ assertEquals(20, count);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ @Test
+ public void testRemovePartialPartition() throws StorageEngineException {
+ StorageEngine.getInstance().removePartitions("root.test1",
+ (storageGroupName, timePartitionId) -> timePartitionId >= 5);
+ StorageEngine.getInstance().removePartitions("root.test2",
+ (storageGroupName, timePartitionId) -> timePartitionId < 5);
+
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM root.test1")) {
+ int count = 0;
+ while (resultSet.next()) {
+ assertEquals(count / 2 * 100 + count % 2 * 50, resultSet.getLong(1));
+ assertEquals(count / 2 * 100 + count % 2 * 50, resultSet.getLong(2));
+ count ++;
+ }
+ assertEquals(10, count);
+ }
+
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM root.test2")) {
+ int count = 0;
+ while (resultSet.next()) {
+ assertEquals(count / 2 * 100 + count % 2 * 50 + 500, resultSet.getLong(1));
+ assertEquals(count / 2 * 100 + count % 2 * 50 + 500, resultSet.getLong(2));
+ count ++;
+ }
+ assertEquals(10, count);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ @Test
+ public void testRemoveAllPartition() throws StorageEngineException {
+ StorageEngine.getInstance().removePartitions("root.test1",
+ (storageGroupName, timePartitionId) -> true);
+
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM root.test1")) {
+ assertFalse(resultSet.next());
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ @Test
+ public void testSQLRemovePartition() {
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("DELETE PARTITION root.test2 0,1,2,3,4");
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM root.test2")) {
+ int count = 0;
+ while (resultSet.next()) {
+ assertEquals(count / 2 * 100 + count % 2 * 50 + 500, resultSet.getLong(1));
+ assertEquals(count / 2 * 100 + count % 2 * 50 + 500, resultSet.getLong(2));
+ count ++;
+ }
+ assertEquals(10, count);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ private static void insertData() throws ClassNotFoundException {
+ List<String> sqls = new ArrayList<>(Arrays.asList(
+ "SET STORAGE GROUP TO root.test1",
+ "SET STORAGE GROUP TO root.test2",
+ "CREATE TIMESERIES root.test1.s0 WITH DATATYPE=INT64,ENCODING=PLAIN",
+ "CREATE TIMESERIES root.test2.s0 WITH DATATYPE=INT64,ENCODING=PLAIN"
+ ));
+ // 10 partitions, each one with one seq file and one unseq file
+ for (int i = 0; i < 10; i++) {
+ // seq files
+ for (int j = 1; j <= 2; j++) {
+ sqls.add(String.format("INSERT INTO root.test%d(timestamp, s0) VALUES (%d, %d)", j,
+ i * partitionInterval + 50, i * partitionInterval + 50));
+ }
+ // last file is unclosed
+ if (i < 9) {
+ sqls.add("FLUSH");
+ }
+ // unseq files
+ for (int j = 1; j <= 2; j++) {
+ sqls.add(String.format("INSERT INTO root.test%d(timestamp, s0) VALUES (%d, %d)", j,
+ i * partitionInterval, i * partitionInterval));
+ }
+ // last file is unclosed
+ if (i < 9) {
+ sqls.add("FLUSH");
+ }
+ }
+ sqls.add("MERGE");
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ for (String sql : sqls) {
+ statement.execute(sql);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+}