This is an automated email from the ASF dual-hosted git repository. jiangtian pushed a commit to branch jira_1450 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e6b7ea01d32b3fde0572d6bf61238a8644254c99 Author: jt2594838 <[email protected]> AuthorDate: Wed Jun 23 12:10:49 2021 +0800 fix deletion may delete other data groups --- .../iotdb/cluster/log/applier/BaseApplier.java | 6 ++ .../iotdb/cluster/log/applier/DataLogApplier.java | 8 ++ .../partition/slot/SlotTimePartitionFilter.java | 55 +++++++++++ .../iotdb/cluster/query/ClusterPlanExecutor.java | 13 ++- .../cluster/server/member/DataGroupMember.java | 13 +++ .../cluster/log/applier/DataLogApplierTest.java | 29 +++++- .../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 2 +- .../org/apache/iotdb/db/engine/StorageEngine.java | 17 +++- .../engine/storagegroup/StorageGroupProcessor.java | 54 ++++++++-- .../virtualSg/VirtualStorageGroupManager.java | 9 +- .../apache/iotdb/db/qp/executor/IPlanExecutor.java | 9 +- .../apache/iotdb/db/qp/executor/PlanExecutor.java | 16 ++- .../iotdb/db/qp/physical/crud/DeletePlan.java | 15 +++ .../db/qp/physical/sys/DeleteTimeSeriesPlan.java | 15 +++ .../engine/modification/DeletionFileNodeTest.java | 109 ++++++++++++++++----- .../db/engine/modification/DeletionQueryTest.java | 77 ++++++++++----- .../storagegroup/StorageGroupProcessorTest.java | 2 +- 17 files changed, 372 insertions(+), 77 deletions(-) diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/BaseApplier.java b/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/BaseApplier.java index 52e8baa..1f626d2 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/BaseApplier.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/BaseApplier.java @@ -39,6 +39,7 @@ import org.apache.iotdb.db.qp.physical.BatchPlan; import org.apache.iotdb.db.qp.physical.PhysicalPlan; import org.apache.iotdb.db.qp.physical.crud.InsertPlan; import org.apache.iotdb.db.utils.SchemaUtils; +import org.apache.iotdb.db.utils.TestOnly; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.service.rpc.thrift.TSStatus; @@ -185,4 +186,9 @@ abstract class BaseApplier implements LogApplier { } return queryExecutor; } + + @TestOnly + public void setQueryExecutor(PlanExecutor queryExecutor) { + this.queryExecutor = queryExecutor; + } } diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/DataLogApplier.java b/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/DataLogApplier.java index 542e422..0da4252 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/DataLogApplier.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/DataLogApplier.java @@ -35,11 +35,13 @@ import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException; import org.apache.iotdb.db.exception.query.QueryProcessException; import org.apache.iotdb.db.metadata.PartialPath; 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.InsertMultiTabletPlan; import org.apache.iotdb.db.qp.physical.crud.InsertPlan; import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan; import org.apache.iotdb.db.qp.physical.crud.InsertRowsPlan; import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan; +import org.apache.iotdb.db.qp.physical.sys.DeleteTimeSeriesPlan; import org.apache.iotdb.db.service.IoTDB; import org.slf4j.Logger; @@ -67,6 +69,12 @@ public class DataLogApplier extends BaseApplier { if (log instanceof PhysicalPlanLog) { PhysicalPlanLog physicalPlanLog = (PhysicalPlanLog) log; PhysicalPlan plan = physicalPlanLog.getPlan(); + if (plan instanceof DeletePlan) { + ((DeletePlan) plan).setPartitionFilter(dataGroupMember.getTimePartitionFilter()); + } else if (plan instanceof DeleteTimeSeriesPlan) { + ((DeleteTimeSeriesPlan) plan) + .setPartitionFilter(dataGroupMember.getTimePartitionFilter()); + } if (plan instanceof InsertMultiTabletPlan) { applyInsert((InsertMultiTabletPlan) plan); } else if (plan instanceof InsertRowsPlan) { diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/partition/slot/SlotTimePartitionFilter.java b/cluster/src/main/java/org/apache/iotdb/cluster/partition/slot/SlotTimePartitionFilter.java new file mode 100644 index 0000000..1296179 --- /dev/null +++ b/cluster/src/main/java/org/apache/iotdb/cluster/partition/slot/SlotTimePartitionFilter.java @@ -0,0 +1,55 @@ +/* + * 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.cluster.partition.slot; + +import org.apache.iotdb.cluster.config.ClusterConstant; +import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.TimePartitionFilter; + +import java.util.Objects; +import java.util.Set; + +public class SlotTimePartitionFilter implements TimePartitionFilter { + + private Set<Integer> slotSet; + + public SlotTimePartitionFilter(Set<Integer> slotSet) { + this.slotSet = slotSet; + } + + @Override + public boolean satisfy(String storageGroupName, long timePartitionId) { + int slot = + SlotPartitionTable.getSlotStrategy() + .calculateSlotByPartitionNum( + storageGroupName, timePartitionId, ClusterConstant.SLOT_NUM); + return slotSet.contains(slot); + } + + @Override + public boolean equals(Object obj) { + return obj instanceof SlotTimePartitionFilter + && Objects.equals(((SlotTimePartitionFilter) obj).slotSet, slotSet); + } + + @Override + public int hashCode() { + return Objects.hash(slotSet); + } +} diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPlanExecutor.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPlanExecutor.java index ca16cfb..aeefa44 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPlanExecutor.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPlanExecutor.java @@ -36,6 +36,7 @@ import org.apache.iotdb.cluster.server.member.DataGroupMember; import org.apache.iotdb.cluster.server.member.MetaGroupMember; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.engine.StorageEngine; +import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.TimePartitionFilter; import org.apache.iotdb.db.exception.StorageEngineException; import org.apache.iotdb.db.exception.metadata.MetadataException; import org.apache.iotdb.db.exception.metadata.PathNotExistException; @@ -623,15 +624,21 @@ public class ClusterPlanExecutor extends PlanExecutor { path, deletePlan.getDeleteStartTime(), deletePlan.getDeleteEndTime(), - deletePlan.getIndex()); + deletePlan.getIndex(), + deletePlan.getPartitionFilter()); } } @Override - public void delete(PartialPath path, long startTime, long endTime, long planIndex) + public void delete( + PartialPath path, + long startTime, + long endTime, + long planIndex, + TimePartitionFilter timePartitionFilter) throws QueryProcessException { try { - StorageEngine.getInstance().delete(path, startTime, endTime, planIndex); + StorageEngine.getInstance().delete(path, startTime, endTime, planIndex, timePartitionFilter); } catch (StorageEngineException e) { throw new QueryProcessException(e); } diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java index 6f6936a..0283665 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java @@ -50,6 +50,7 @@ import org.apache.iotdb.cluster.partition.slot.SlotManager.SlotStatus; import org.apache.iotdb.cluster.partition.slot.SlotNodeAdditionResult; import org.apache.iotdb.cluster.partition.slot.SlotNodeRemovalResult; import org.apache.iotdb.cluster.partition.slot.SlotPartitionTable; +import org.apache.iotdb.cluster.partition.slot.SlotTimePartitionFilter; import org.apache.iotdb.cluster.query.LocalQueryExecutor; import org.apache.iotdb.cluster.query.manage.ClusterQueryManager; import org.apache.iotdb.cluster.rpc.thrift.ElectionRequest; @@ -787,6 +788,18 @@ public class DataGroupMember extends RaftMember { } /** + * Return a TimePartitionFilter that tells whether a time partition is managed by this member. + * + * @return a TimePartitionFilter that tells whether a time partition is managed by this member. + */ + public TimePartitionFilter getTimePartitionFilter() { + Set<Integer> slotSet = + new HashSet<>( + ((SlotPartitionTable) metaGroupMember.getPartitionTable()).getNodeSlots(getHeader())); + return new SlotTimePartitionFilter(slotSet); + } + + /** * When a node is removed and IT IS NOT THE HEADER of the group, the member should take over some * slots from the removed group, and add a new node to the group the removed node was in the * group. diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/DataLogApplierTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/DataLogApplierTest.java index 5f8f43f..0d8920a 100644 --- a/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/DataLogApplierTest.java +++ b/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/DataLogApplierTest.java @@ -27,7 +27,6 @@ import org.apache.iotdb.cluster.common.TestDataGroupMember; import org.apache.iotdb.cluster.common.TestMetaGroupMember; import org.apache.iotdb.cluster.common.TestUtils; import org.apache.iotdb.cluster.coordinator.Coordinator; -import org.apache.iotdb.cluster.log.LogApplier; import org.apache.iotdb.cluster.log.logtypes.CloseFileLog; import org.apache.iotdb.cluster.log.logtypes.PhysicalPlanLog; import org.apache.iotdb.cluster.metadata.CMManager; @@ -49,6 +48,7 @@ import org.apache.iotdb.cluster.server.service.MetaAsyncService; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.engine.StorageEngine; import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor; +import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.TimePartitionFilter; import org.apache.iotdb.db.exception.StorageEngineException; import org.apache.iotdb.db.exception.metadata.IllegalPathException; import org.apache.iotdb.db.exception.metadata.MetadataException; @@ -56,6 +56,8 @@ import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException; import org.apache.iotdb.db.exception.query.QueryProcessException; import org.apache.iotdb.db.metadata.PartialPath; import org.apache.iotdb.db.metadata.mnode.MeasurementMNode; +import org.apache.iotdb.db.qp.executor.PlanExecutor; +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.InsertRowPlan; import org.apache.iotdb.db.qp.physical.sys.CreateMultiTimeSeriesPlan; @@ -140,7 +142,7 @@ public class DataLogApplierTest extends IoTDBTest { private TestDataGroupMember testDataGroupMember = new TestDataGroupMember(); - private LogApplier applier = new DataLogApplier(testMetaGroupMember, testDataGroupMember); + private DataLogApplier applier; @Override @Before @@ -234,6 +236,8 @@ public class DataLogApplierTest extends IoTDBTest { } }); ((CMManager) IoTDB.metaManager).setMetaGroupMember(testMetaGroupMember); + testDataGroupMember.setMetaGroupMember(testMetaGroupMember); + applier = new DataLogApplier(testMetaGroupMember, testDataGroupMember); } @Override @@ -373,4 +377,25 @@ public class DataLogApplierTest extends IoTDBTest { assertTrue(IoTDB.metaManager.getAllStorageGroupPaths().contains(new PartialPath("root.sg2"))); assertNull(log.getException()); } + + @Test + public void testApplyDelete() throws QueryProcessException { + applier.setQueryExecutor( + new PlanExecutor() { + @Override + public boolean processNonQuery(PhysicalPlan plan) { + assertTrue(plan instanceof DeletePlan); + DeletePlan deletePlan = (DeletePlan) plan; + TimePartitionFilter planFilter = deletePlan.getPartitionFilter(); + TimePartitionFilter memberFilter = testDataGroupMember.getTimePartitionFilter(); + assertEquals(planFilter, memberFilter); + return true; + } + }); + + DeletePlan deletePlan = new DeletePlan(); + PhysicalPlanLog log = new PhysicalPlanLog(deletePlan); + applier.apply(log); + assertNull(log.getException()); + } } diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java index 9f11f41..4cb910d 100644 --- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java +++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java @@ -736,7 +736,7 @@ public class IoTDBConfig { return concurrentWritingTimePartition; } - void setConcurrentWritingTimePartition(int concurrentWritingTimePartition) { + public void setConcurrentWritingTimePartition(int concurrentWritingTimePartition) { this.concurrentWritingTimePartition = concurrentWritingTimePartition; } 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 f5d0bff..2332938 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 @@ -669,7 +669,12 @@ public class StorageEngine implements IService { virtualStorageGroupManager.closeStorageGroupProcessor(partitionId, isSeq, isSync); } - public void delete(PartialPath path, long startTime, long endTime, long planIndex) + public void delete( + PartialPath path, + long startTime, + long endTime, + long planIndex, + TimePartitionFilter timePartitionFilter) throws StorageEngineException { try { List<PartialPath> sgPaths = IoTDB.metaManager.searchAllRelatedStorageGroups(path); @@ -680,7 +685,9 @@ public class StorageEngine implements IService { } PartialPath newPath = path.alterPrefixPath(storageGroupPath); - processorMap.get(storageGroupPath).delete(newPath, startTime, endTime, planIndex); + processorMap + .get(storageGroupPath) + .delete(newPath, startTime, endTime, planIndex, timePartitionFilter); } } catch (IOException | MetadataException e) { throw new StorageEngineException(e.getMessage()); @@ -688,7 +695,9 @@ public class StorageEngine implements IService { } /** delete data of timeseries "{deviceId}.{measurementId}" */ - public void deleteTimeseries(PartialPath path, long planIndex) throws StorageEngineException { + public void deleteTimeseries( + PartialPath path, long planIndex, TimePartitionFilter timePartitionFilter) + throws StorageEngineException { try { List<PartialPath> sgPaths = IoTDB.metaManager.searchAllRelatedStorageGroups(path); for (PartialPath storageGroupPath : sgPaths) { @@ -700,7 +709,7 @@ public class StorageEngine implements IService { PartialPath newPath = path.alterPrefixPath(storageGroupPath); processorMap .get(storageGroupPath) - .delete(newPath, Long.MIN_VALUE, Long.MAX_VALUE, planIndex); + .delete(newPath, Long.MIN_VALUE, Long.MAX_VALUE, planIndex, timePartitionFilter); } } catch (IOException | MetadataException e) { throw new StorageEngineException(e.getMessage()); 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 2b2185d..afb463a 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 @@ -1766,8 +1766,14 @@ public class StorageGroupProcessor { * @param path the timeseries path of the to be deleted. * @param startTime the startTime of delete range. * @param endTime the endTime of delete range. + * @param timePartitionFilter */ - public void delete(PartialPath path, long startTime, long endTime, long planIndex) + public void delete( + PartialPath path, + long startTime, + long endTime, + long planIndex, + TimePartitionFilter timePartitionFilter) throws IOException { // If there are still some old version tsfiles, the delete won't succeeded. if (upgradeFileCount.get() != 0) { @@ -1798,7 +1804,7 @@ public class StorageGroupProcessor { } // write log to impacted working TsFileProcessors - logDeletion(startTime, endTime, path); + logDeletion(startTime, endTime, path, timePartitionFilter); Deletion deletion = new Deletion(path, MERGE_MOD_START_VERSION_NUM, startTime, endTime); if (tsFileManagement.mergingModification != null) { @@ -1807,9 +1813,19 @@ public class StorageGroupProcessor { } deleteDataInFiles( - tsFileManagement.getTsFileList(true), deletion, devicePaths, updatedModFiles, planIndex); + tsFileManagement.getTsFileList(true), + deletion, + devicePaths, + updatedModFiles, + planIndex, + timePartitionFilter); deleteDataInFiles( - tsFileManagement.getTsFileList(false), deletion, devicePaths, updatedModFiles, planIndex); + tsFileManagement.getTsFileList(false), + deletion, + devicePaths, + updatedModFiles, + planIndex, + timePartitionFilter); } catch (Exception e) { // roll back @@ -1822,19 +1838,27 @@ public class StorageGroupProcessor { } } - private void logDeletion(long startTime, long endTime, PartialPath path) throws IOException { + private void logDeletion( + long startTime, long endTime, PartialPath path, TimePartitionFilter timePartitionFilter) + throws IOException { long timePartitionStartId = StorageEngine.getTimePartition(startTime); long timePartitionEndId = StorageEngine.getTimePartition(endTime); if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) { DeletePlan deletionPlan = new DeletePlan(startTime, endTime, path); for (Map.Entry<Long, TsFileProcessor> entry : workSequenceTsFileProcessors.entrySet()) { - if (timePartitionStartId <= entry.getKey() && entry.getKey() <= timePartitionEndId) { + if (timePartitionStartId <= entry.getKey() + && entry.getKey() <= timePartitionEndId + && (timePartitionFilter == null + || timePartitionFilter.satisfy(logicalStorageGroupName, entry.getKey()))) { entry.getValue().getLogNode().write(deletionPlan); } } for (Map.Entry<Long, TsFileProcessor> entry : workUnsequenceTsFileProcessors.entrySet()) { - if (timePartitionStartId <= entry.getKey() && entry.getKey() <= timePartitionEndId) { + if (timePartitionStartId <= entry.getKey() + && entry.getKey() <= timePartitionEndId + && (timePartitionFilter == null + || timePartitionFilter.satisfy(logicalStorageGroupName, entry.getKey()))) { entry.getValue().getLogNode().write(deletionPlan); } } @@ -1845,7 +1869,12 @@ public class StorageGroupProcessor { TsFileResource tsFileResource, Set<PartialPath> devicePaths, long deleteStart, - long deleteEnd) { + long deleteEnd, + TimePartitionFilter timePartitionFilter) { + if (timePartitionFilter != null + && !timePartitionFilter.satisfy(virtualStorageGroupId, tsFileResource.getTimePartition())) { + return true; + } for (PartialPath device : devicePaths) { String deviceId = device.getFullPath(); long endTime = tsFileResource.getEndTime(deviceId); @@ -1866,11 +1895,16 @@ public class StorageGroupProcessor { Deletion deletion, Set<PartialPath> devicePaths, List<ModificationFile> updatedModFiles, - long planIndex) + long planIndex, + TimePartitionFilter timePartitionFilter) throws IOException { for (TsFileResource tsFileResource : tsFileResourceList) { if (canSkipDelete( - tsFileResource, devicePaths, deletion.getStartTime(), deletion.getEndTime())) { + tsFileResource, + devicePaths, + deletion.getStartTime(), + deletion.getEndTime(), + timePartitionFilter)) { continue; } diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/VirtualStorageGroupManager.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/VirtualStorageGroupManager.java index e4faadb..be9e1ac 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/VirtualStorageGroupManager.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/VirtualStorageGroupManager.java @@ -265,11 +265,16 @@ public class VirtualStorageGroupManager { } /** push delete operation down to all virtual storage group processors */ - public void delete(PartialPath path, long startTime, long endTime, long planIndex) + public void delete( + PartialPath path, + long startTime, + long endTime, + long planIndex, + TimePartitionFilter timePartitionFilter) throws IOException { for (StorageGroupProcessor storageGroupProcessor : virtualStorageGroupProcessor) { if (storageGroupProcessor != null) { - storageGroupProcessor.delete(path, startTime, endTime, planIndex); + storageGroupProcessor.delete(path, startTime, endTime, planIndex, timePartitionFilter); } } } diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java index af24983..57ab564 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java @@ -18,6 +18,7 @@ */ package org.apache.iotdb.db.qp.executor; +import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.TimePartitionFilter; import org.apache.iotdb.db.exception.BatchProcessException; import org.apache.iotdb.db.exception.StorageEngineException; import org.apache.iotdb.db.exception.metadata.MetadataException; @@ -86,8 +87,14 @@ public interface IPlanExecutor { * @param startTime start time in delete command * @param endTime end time in delete command * @param planIndex index of the deletion plan + * @param partitionFilter specify involving time partitions, if null, all partitions are involved */ - void delete(PartialPath path, long startTime, long endTime, long planIndex) + void delete( + PartialPath path, + long startTime, + long endTime, + long planIndex, + TimePartitionFilter partitionFilter) throws QueryProcessException; /** 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 4122240..c22ce34 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 @@ -984,7 +984,8 @@ public class PlanExecutor implements IPlanExecutor { path, deletePlan.getDeleteStartTime(), deletePlan.getDeleteEndTime(), - deletePlan.getIndex()); + deletePlan.getIndex(), + deletePlan.getPartitionFilter()); } } @@ -1171,10 +1172,15 @@ public class PlanExecutor implements IPlanExecutor { } @Override - public void delete(PartialPath path, long startTime, long endTime, long planIndex) + public void delete( + PartialPath path, + long startTime, + long endTime, + long planIndex, + TimePartitionFilter timePartitionFilter) throws QueryProcessException { try { - StorageEngine.getInstance().delete(path, startTime, endTime, planIndex); + StorageEngine.getInstance().delete(path, startTime, endTime, planIndex, timePartitionFilter); } catch (StorageEngineException e) { throw new QueryProcessException(e); } @@ -1509,7 +1515,9 @@ public class PlanExecutor implements IPlanExecutor { for (int i = 0; i < deletePathList.size(); i++) { PartialPath path = deletePathList.get(i); try { - StorageEngine.getInstance().deleteTimeseries(path, deleteTimeSeriesPlan.getIndex()); + StorageEngine.getInstance() + .deleteTimeseries( + path, deleteTimeSeriesPlan.getIndex(), deleteTimeSeriesPlan.getPartitionFilter()); String failed = IoTDB.metaManager.deleteTimeseries(path); if (failed != null) { deleteTimeSeriesPlan diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePlan.java index f3c1b43..761a85f 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePlan.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePlan.java @@ -18,6 +18,7 @@ */ package org.apache.iotdb.db.qp.physical.crud; +import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.TimePartitionFilter; import org.apache.iotdb.db.exception.metadata.IllegalPathException; import org.apache.iotdb.db.metadata.PartialPath; import org.apache.iotdb.db.qp.logical.Operator; @@ -35,6 +36,12 @@ public class DeletePlan extends PhysicalPlan { private long deleteStartTime; private long deleteEndTime; private List<PartialPath> paths = new ArrayList<>(); + /** + * This deletion only affects those time partitions that evaluate true by the filter. If the + * filter is null, all partitions are processed. This is to avoid redundant data deletions when + * one timeseries deletion is split and executed into different replication groups. + */ + private TimePartitionFilter partitionFilter; public DeletePlan() { super(false, Operator.OperatorType.DELETE); @@ -102,6 +109,14 @@ public class DeletePlan extends PhysicalPlan { this.paths = paths; } + public TimePartitionFilter getPartitionFilter() { + return partitionFilter; + } + + public void setPartitionFilter(TimePartitionFilter partitionFilter) { + this.partitionFilter = partitionFilter; + } + @Override public int hashCode() { return Objects.hash(deleteStartTime, deleteEndTime, paths); diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DeleteTimeSeriesPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DeleteTimeSeriesPlan.java index d0b1677..23e3187 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DeleteTimeSeriesPlan.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DeleteTimeSeriesPlan.java @@ -18,6 +18,7 @@ */ package org.apache.iotdb.db.qp.physical.sys; +import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.TimePartitionFilter; import org.apache.iotdb.db.exception.metadata.IllegalPathException; import org.apache.iotdb.db.metadata.PartialPath; import org.apache.iotdb.db.qp.logical.Operator; @@ -37,6 +38,12 @@ public class DeleteTimeSeriesPlan extends PhysicalPlan { private List<PartialPath> deletePathList; private Map<Integer, TSStatus> results = new TreeMap<>(); + /** + * This deletion only affects those time partitions that evaluate true by the filter. If the + * filter is null, all partitions are processed. This is to avoid redundant data deletions when + * one timeseries deletion is split and executed into different replication groups. + */ + private TimePartitionFilter partitionFilter; public DeleteTimeSeriesPlan(List<PartialPath> deletePathList) { super(false, Operator.OperatorType.DELETE_TIMESERIES); @@ -56,6 +63,14 @@ public class DeleteTimeSeriesPlan extends PhysicalPlan { this.deletePathList = deletePathList; } + public TimePartitionFilter getPartitionFilter() { + return partitionFilter; + } + + public void setPartitionFilter(TimePartitionFilter partitionFilter) { + this.partitionFilter = partitionFilter; + } + @Override public void serialize(DataOutputStream stream) throws IOException { int type = PhysicalPlanType.DELETE_TIMESERIES.ordinal(); diff --git a/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionFileNodeTest.java b/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionFileNodeTest.java index bfe9522..973da63 100644 --- a/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionFileNodeTest.java +++ b/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionFileNodeTest.java @@ -26,6 +26,7 @@ import org.apache.iotdb.db.engine.modification.io.LocalTextModificationAccessor; import org.apache.iotdb.db.engine.querycontext.QueryDataSource; import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk; import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor; +import org.apache.iotdb.db.engine.storagegroup.TsFileResource; import org.apache.iotdb.db.exception.StorageEngineException; import org.apache.iotdb.db.exception.metadata.IllegalPathException; import org.apache.iotdb.db.exception.metadata.MetadataException; @@ -116,14 +117,61 @@ public class DeletionFileNodeTest { StorageEngine.getInstance().insert(new InsertRowPlan(record)); } - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[3]), 0, 50, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[4]), 0, 50, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[5]), 0, 30, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[5]), 30, 50, -1); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[3]), 0, 50, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[4]), 0, 50, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 0, 30, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 30, 50, -1, null); + checkSeriesPointCount(5, 50); + } + + @Test + public void testDeleteWithTimePartitionFilter() + throws IllegalPathException, StorageEngineException, QueryProcessException, IOException { + boolean prevEnablePartition = StorageEngine.isEnablePartition(); + long prevPartitionInterval = StorageEngine.getTimePartitionInterval(); + int prevConcurrentTimePartition = + IoTDBDescriptor.getInstance().getConfig().getConcurrentWritingTimePartition(); + try { + StorageEngine.setEnablePartition(true); + IoTDBDescriptor.getInstance().getConfig().setConcurrentWritingTimePartition(10); + long newPartitionInterval = 100; + StorageEngine.setTimePartitionInterval(newPartitionInterval); + // generate 10 time partitions + for (int i = 0; i < 10; i++) { + TSRecord record = new TSRecord(i * newPartitionInterval, processorName); + record.addTuple(new DoubleDataPoint(measurements[0], i * 1.0)); + StorageEngine.getInstance().insert(new InsertRowPlan(record)); + } + // the filter only allows to delete the first 5 partitions + StorageEngine.getInstance() + .delete( + new PartialPath(processorName, measurements[0]), + 0, + Long.MAX_VALUE, + -1, + (storageGroupName, timePartitionId) -> timePartitionId < 5); + checkSeriesPointCount(0, 5); + } finally { + StorageEngine.setEnablePartition(prevEnablePartition); + StorageEngine.setTimePartitionInterval(prevPartitionInterval); + IoTDBDescriptor.getInstance() + .getConfig() + .setConcurrentWritingTimePartition(prevConcurrentTimePartition); + } + } + + private void checkSeriesPointCount(int measurementIdx, int expectedCount) + throws IllegalPathException, StorageEngineException, QueryProcessException, IOException { SingleSeriesExpression expression = new SingleSeriesExpression( - new PartialPath(processorName + TsFileConstant.PATH_SEPARATOR + measurements[5]), null); + new PartialPath( + processorName + TsFileConstant.PATH_SEPARATOR + measurements[measurementIdx]), + null); List<StorageGroupProcessor> list = StorageEngine.getInstance() .mergeLock(Collections.singletonList((PartialPath) expression.getSeriesPath())); @@ -132,17 +180,22 @@ public class DeletionFileNodeTest { QueryResourceManager.getInstance() .getQueryDataSource( (PartialPath) expression.getSeriesPath(), TEST_QUERY_CONTEXT, null); - List<ReadOnlyMemChunk> timeValuePairs = - dataSource.getSeqResources().get(0).getReadOnlyMemChunk(); + int count = 0; - for (ReadOnlyMemChunk chunk : timeValuePairs) { - IPointReader iterator = chunk.getPointReader(); - while (iterator.hasNextTimeValuePair()) { - iterator.nextTimeValuePair(); - count++; + for (TsFileResource seqResource : dataSource.getSeqResources()) { + List<ReadOnlyMemChunk> timeValuePairs = seqResource.getReadOnlyMemChunk(); + if (timeValuePairs == null) { + continue; + } + for (ReadOnlyMemChunk chunk : timeValuePairs) { + IPointReader iterator = chunk.getPointReader(); + while (iterator.hasNextTimeValuePair()) { + iterator.nextTimeValuePair(); + count++; + } } } - assertEquals(50, count); + assertEquals(expectedCount, count); QueryResourceManager.getInstance().endQuery(TEST_QUERY_JOB_ID); } finally { StorageEngine.getInstance().mergeUnLock(list); @@ -161,9 +214,12 @@ public class DeletionFileNodeTest { } StorageEngine.getInstance().syncCloseAllProcessor(); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[5]), 0, 50, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[4]), 0, 40, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[3]), 0, 30, -1); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 0, 50, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[4]), 0, 40, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[3]), 0, 30, -1, null); Modification[] realModifications = new Modification[] { @@ -239,10 +295,14 @@ public class DeletionFileNodeTest { StorageEngine.getInstance().insert(new InsertRowPlan(record)); } - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[3]), 0, 50, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[4]), 0, 50, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[5]), 0, 30, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[5]), 30, 50, -1); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[3]), 0, 50, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[4]), 0, 50, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 0, 30, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 30, 50, -1, null); SingleSeriesExpression expression = new SingleSeriesExpression( @@ -298,9 +358,12 @@ public class DeletionFileNodeTest { } StorageEngine.getInstance().syncCloseAllProcessor(); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[5]), 0, 50, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[4]), 0, 40, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[3]), 0, 30, -1); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 0, 50, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[4]), 0, 40, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[3]), 0, 30, -1, null); Modification[] realModifications = new Modification[] { diff --git a/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionQueryTest.java b/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionQueryTest.java index 7375045..a11e8ec 100644 --- a/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionQueryTest.java +++ b/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionQueryTest.java @@ -108,10 +108,14 @@ public class DeletionQueryTest { StorageEngine.getInstance().insert(new InsertRowPlan(record)); } - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[3]), 0, 50, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[4]), 0, 50, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[5]), 0, 30, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[5]), 30, 50, -1); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[3]), 0, 50, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[4]), 0, 50, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 0, 30, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 30, 50, -1, null); List<PartialPath> pathList = new ArrayList<>(); pathList.add(new PartialPath(processorName + TsFileConstant.PATH_SEPARATOR + measurements[3])); @@ -147,9 +151,12 @@ public class DeletionQueryTest { } StorageEngine.getInstance().syncCloseAllProcessor(); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[5]), 0, 50, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[4]), 0, 40, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[3]), 0, 30, -1); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 0, 50, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[4]), 0, 40, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[3]), 0, 30, -1, null); List<PartialPath> pathList = new ArrayList<>(); pathList.add(new PartialPath(processorName + TsFileConstant.PATH_SEPARATOR + measurements[3])); @@ -196,10 +203,14 @@ public class DeletionQueryTest { StorageEngine.getInstance().insert(new InsertRowPlan(record)); } - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[3]), 0, 50, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[4]), 0, 50, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[5]), 0, 30, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[5]), 30, 50, -1); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[3]), 0, 50, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[4]), 0, 50, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 0, 30, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 30, 50, -1, null); List<PartialPath> pathList = new ArrayList<>(); pathList.add(new PartialPath(processorName + TsFileConstant.PATH_SEPARATOR + measurements[3])); @@ -246,9 +257,12 @@ public class DeletionQueryTest { } StorageEngine.getInstance().syncCloseAllProcessor(); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[5]), 0, 50, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[4]), 0, 40, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[3]), 0, 30, -1); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 0, 50, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[4]), 0, 40, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[3]), 0, 30, -1, null); List<PartialPath> pathList = new ArrayList<>(); pathList.add(new PartialPath(processorName + TsFileConstant.PATH_SEPARATOR + measurements[3])); @@ -284,10 +298,14 @@ public class DeletionQueryTest { StorageEngine.getInstance().insert(new InsertRowPlan(record)); } - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[3]), 0, 50, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[4]), 0, 50, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[5]), 0, 30, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[5]), 30, 50, -1); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[3]), 0, 50, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[4]), 0, 50, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 0, 30, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 30, 50, -1, null); StorageEngine.getInstance().syncCloseAllProcessor(); @@ -299,11 +317,14 @@ public class DeletionQueryTest { StorageEngine.getInstance().insert(new InsertRowPlan(record)); } - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[3]), 0, 250, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[4]), 0, 250, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[5]), 0, 230, -1); StorageEngine.getInstance() - .delete(new PartialPath(processorName, measurements[5]), 230, 250, -1); + .delete(new PartialPath(processorName, measurements[3]), 0, 250, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[4]), 0, 250, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 0, 230, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 230, 250, -1, null); StorageEngine.getInstance().syncCloseAllProcessor(); @@ -315,10 +336,14 @@ public class DeletionQueryTest { StorageEngine.getInstance().insert(new InsertRowPlan(record)); } - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[3]), 0, 50, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[4]), 0, 50, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[5]), 0, 30, -1); - StorageEngine.getInstance().delete(new PartialPath(processorName, measurements[5]), 30, 50, -1); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[3]), 0, 50, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[4]), 0, 50, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 0, 30, -1, null); + StorageEngine.getInstance() + .delete(new PartialPath(processorName, measurements[5]), 30, 50, -1, null); StorageEngine.getInstance().syncCloseAllProcessor(); diff --git a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java index 3d7e2b6..9cfd805 100644 --- a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java +++ b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java @@ -120,7 +120,7 @@ public class StorageGroupProcessorTest { processor.insert(new InsertRowPlan(record)); } - processor.delete(new PartialPath(deviceId, measurementId), 0, 15L, -1); + processor.delete(new PartialPath(deviceId, measurementId), 0, 15L, -1, null); List<TsFileResource> tsfileResourcesForQuery = new ArrayList<>(); for (TsFileProcessor tsfileProcessor : processor.getWorkUnsequenceTsFileProcessors()) {
