You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2020/06/23 04:22:53 UTC
[incubator-iotdb] branch add_partition_removal created (now 6a5c090)
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a change to branch add_partition_removal
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git.
at 6a5c090 support partition removal
This branch includes the following new commits:
new 6a5c090 support partition removal
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[incubator-iotdb] 01/01: support partition removal
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch add_partition_removal
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 6a5c090c58fe52572aadb054447ac34541b8c2a4
Author: jt2594838 <jt...@163.com>
AuthorDate: Tue Jun 23 12:22:36 2020 +0800
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 | 7 +
.../org/apache/iotdb/db/engine/StorageEngine.java | 17 ++
.../engine/storagegroup/StorageGroupProcessor.java | 60 +++++++
.../db/engine/storagegroup/TsFileResource.java | 3 +
.../main/java/org/apache/iotdb/db/qp/Planner.java | 1 +
.../apache/iotdb/db/qp/constant/SQLConstant.java | 2 +
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 10 ++
.../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 | 18 ++
.../iotdb/db/qp/strategy/PhysicalGenerator.java | 5 +
.../db/integration/IoTDBRemovePartitionIT.java | 193 +++++++++++++++++++++
17 files changed, 461 insertions(+), 1 deletion(-)
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 b327d45..2465d41 100644
--- a/docs/UserGuide/Operation Manual/SQL Reference.md
+++ b/docs/UserGuide/Operation Manual/SQL Reference.md
@@ -937,6 +937,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 154da9e..f945b5c 100644
--- a/docs/zh/UserGuide/Operation Manual/SQL Reference.md
+++ b/docs/zh/UserGuide/Operation Manual/SQL Reference.md
@@ -898,6 +898,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 8170130..e716d11 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
@@ -80,6 +80,7 @@ statement
| LOAD STRING_LITERAL autoCreateSchema? #loadFiles
| REMOVE STRING_LITERAL #removeFile
| MOVE STRING_LITERAL STRING_LITERAL #moveFile
+ | DELETE PARTITION prefixPath INT_LIST #deletePartition
| SELECT INDEX func=ID //not support yet
LR_BRACKET
p1=fullPath COMMA p2=fullPath COMMA n1=timeValue COMMA n2=timeValue COMMA
@@ -883,6 +884,10 @@ LATEST
: L A T E S T
;
+PARTITION
+ : P A R T I T I O N
+ ;
+
//============================
// End of the keywords list
//============================
@@ -970,6 +975,8 @@ DATETIME
/** Allow unicode rule/token names */
ID : FIRST_NAME_CHAR NAME_CHAR*;
+INT_LIST : INT(COMMA INT)*;
+
fragment
NAME_CHAR
: 'A'..'Z'
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 009acac..a8704e3 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,64 @@ 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
+ for (Iterator<Entry<Long, TsFileProcessor>> iterator = workSequenceTsFileProcessors
+ .entrySet().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();
+ }
+ }
+ for (Iterator<Entry<Long, TsFileProcessor>> iterator = workUnsequenceTsFileProcessors
+ .entrySet().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();
+ }
+ }
+ // remove data files
+ for (Iterator<TsFileResource> iterator = sequenceFileTreeSet.iterator();
+ iterator.hasNext(); ) {
+ TsFileResource tsFileResource = iterator.next();
+ if (filter.satisfy(storageGroupName, tsFileResource.getTimePartition())) {
+ tsFileResource.remove();
+ iterator.remove();
+ }
+ }
+ for (Iterator<TsFileResource> iterator = unSequenceFileList.iterator();
+ iterator.hasNext(); ) {
+ TsFileResource tsFileResource = iterator.next();
+ if (filter.satisfy(storageGroupName, tsFileResource.getTimePartition())) {
+ tsFileResource.remove();
+ iterator.remove();
+ }
+ }
+ } finally {
+ insertLock.writeLock().unlock();
+ mergeLock.writeLock().unlock();
+ }
+ }
+
+ @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 3b31bbe..e6e5629 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
@@ -100,6 +100,7 @@ public class Planner {
case CLEAR_CACHE:
case NULL:
case SHOW_MERGE_STATUS:
+ case DELETE_PARTITION:
return operator;
case QUERY:
case UPDATE:
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 1acf1a7..db7ff5f 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,6 +151,7 @@ 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 = 87;
public static final Map<Integer, String> tokenSymbol = new HashMap<>();
public static final Map<Integer, String> tokenNames = new HashMap<>();
@@ -221,6 +222,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");
}
static {
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 3cf570e..00408ce 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
@@ -74,6 +74,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;
@@ -92,6 +93,7 @@ import org.apache.iotdb.db.qp.logical.sys.AuthorOperator.AuthorType;
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.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;
@@ -261,6 +263,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;
default:
throw new UnsupportedOperationException(
String.format("operation %s is not supported", plan.getOperatorType()));
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 746326c..d1c7c3b 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
+ SHOW_MERGE_STATUS, 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 ebd41f0..ba17676 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
@@ -50,6 +50,7 @@ import org.apache.iotdb.db.qp.logical.sys.ClearCacheOperator;
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;
@@ -81,6 +82,7 @@ import org.apache.iotdb.db.qp.strategy.SqlBaseParser.CreateRoleContext;
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;
@@ -1622,4 +1624,20 @@ public class LogicalGenerator extends SqlBaseBaseListener {
super.enterShowMergeStatus(ctx);
initializedOperator = new ShowMergeStatusOperator(SQLConstant.TOK_SHOW_MERGE_STATUS);
}
+
+ @Override
+ public void enterDeletePartition(DeletePartitionContext ctx) {
+ super.enterDeletePartition(ctx);
+ DeletePartitionOperator deletePartitionOperator = new DeletePartitionOperator(
+ SQLConstant.TOK_DELETE_PARTITION);
+ deletePartitionOperator.setStorageGroupName(ctx.prefixPath().getText());
+ String intListStr = ctx.INT_LIST().getText();
+ String[] split = intListStr.split(",");
+ Set<Long> idSet = new HashSet<>();
+ for (String s : split) {
+ idSet.add(Long.parseLong(s));
+ }
+ deletePartitionOperator.setPartitionIds(idSet);
+ initializedOperator = deletePartitionOperator;
+ }
}
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 d050e5d..f4f95db 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;
@@ -64,6 +65,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;
@@ -258,6 +260,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());
default:
throw new LogicalOperatorException(operator.getType().toString(), "");
}
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();
+ }
+ }
+}