You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by zy...@apache.org on 2022/09/21 06:15:29 UTC
[iotdb] branch master updated: [IOTDB-4479] Add pattern overlap check when deleting timeseries (#7373)
This is an automated email from the ASF dual-hosted git repository.
zyk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 867eaf322f [IOTDB-4479] Add pattern overlap check when deleting timeseries (#7373)
867eaf322f is described below
commit 867eaf322f21010685d18333debc94d7590e7c3a
Author: Marcos_Zyk <38...@users.noreply.github.com>
AuthorDate: Wed Sep 21 14:15:23 2022 +0800
[IOTDB-4479] Add pattern overlap check when deleting timeseries (#7373)
[IOTDB-4479] Add pattern overlap check when deleting timeseries (#7373)
---
.../db/it/schema/IoTDBDeleteTimeseriesIT.java | 23 ++++++++
.../impl/DataNodeInternalRPCServiceImpl.java | 61 +++++++++++++++++++---
2 files changed, 77 insertions(+), 7 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBDeleteTimeseriesIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBDeleteTimeseriesIT.java
index ca75bd914c..e0b091e71b 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBDeleteTimeseriesIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBDeleteTimeseriesIT.java
@@ -331,6 +331,29 @@ public class IoTDBDeleteTimeseriesIT {
}
Assert.assertEquals(retArray1.length, cnt);
}
+
+ statement.execute("delete timeseries root.sg1.d1.s2, root.sg2.**");
+ try (ResultSet resultSet = statement.executeQuery("select s2 from root.sg1.*")) {
+ Assert.assertFalse(resultSet.next());
+ }
+
+ try (ResultSet resultSet = statement.executeQuery("show timeseries root.sg2.*.s2")) {
+ Assert.assertFalse(resultSet.next());
+ }
+
+ retArray1 = new String[] {"0,4,4"};
+ cnt = 0;
+ try (ResultSet resultSet = statement.executeQuery("select count(s2) from root.*.*")) {
+ while (resultSet.next()) {
+ StringBuilder ans = new StringBuilder(resultSet.getString(TIMESTAMP_STR));
+ for (int i = 3; i <= 4; i++) {
+ ans.append(",").append(resultSet.getString(count("root.sg" + i + ".d1.s2")));
+ }
+ Assert.assertEquals(retArray1[cnt], ans.toString());
+ cnt++;
+ }
+ Assert.assertEquals(retArray1.length, cnt);
+ }
}
@Test
diff --git a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java
index 2e1a9f7f4f..aaf807b3c5 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -31,6 +31,7 @@ import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.consensus.SchemaRegionId;
+import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
@@ -55,6 +56,7 @@ import org.apache.iotdb.db.engine.cache.TimeSeriesMetadataCache;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.cache.DataNodeSchemaCache;
+import org.apache.iotdb.db.metadata.schemaregion.ISchemaRegion;
import org.apache.iotdb.db.metadata.schemaregion.SchemaEngine;
import org.apache.iotdb.db.metadata.template.ClusterTemplateManager;
import org.apache.iotdb.db.metadata.template.TemplateInternalRPCUpdateType;
@@ -145,6 +147,8 @@ import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
+import static org.apache.iotdb.commons.conf.IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD;
+
public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface {
private static final Logger LOGGER =
@@ -341,10 +345,18 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
TSStatus status;
int preDeletedNum = 0;
for (TConsensusGroupId consensusGroupId : req.getSchemaRegionIdList()) {
+ String storageGroup =
+ schemaEngine
+ .getSchemaRegion(new SchemaRegionId(consensusGroupId.getId()))
+ .getStorageGroupFullPath();
+ PathPatternTree filteredPatternTree = filterPathPatternTree(patternTree, storageGroup);
+ if (filteredPatternTree.isEmpty()) {
+ continue;
+ }
status =
regionManager.executeSchemaPlanNode(
new SchemaRegionId(consensusGroupId.getId()),
- new ConstructSchemaBlackListNode(new PlanNodeId(""), patternTree));
+ new ConstructSchemaBlackListNode(new PlanNodeId(""), filteredPatternTree));
if (status.code == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
preDeletedNum += Integer.parseInt(status.getMessage());
} else {
@@ -366,10 +378,18 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
List<TSStatus> failureList = new ArrayList<>();
TSStatus status;
for (TConsensusGroupId consensusGroupId : req.getSchemaRegionIdList()) {
+ String storageGroup =
+ schemaEngine
+ .getSchemaRegion(new SchemaRegionId(consensusGroupId.getId()))
+ .getStorageGroupFullPath();
+ PathPatternTree filteredPatternTree = filterPathPatternTree(patternTree, storageGroup);
+ if (filteredPatternTree.isEmpty()) {
+ continue;
+ }
status =
regionManager.executeSchemaPlanNode(
new SchemaRegionId(consensusGroupId.getId()),
- new RollbackSchemaBlackListNode(new PlanNodeId(""), patternTree));
+ new RollbackSchemaBlackListNode(new PlanNodeId(""), filteredPatternTree));
if (status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
failureList.add(status);
}
@@ -408,10 +428,14 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
for (TConsensusGroupId consensusGroupId : req.getSchemaRegionIdList()) {
// todo implement as consensus layer read request
try {
- for (PartialPath path :
- schemaEngine
- .getSchemaRegion(new SchemaRegionId(consensusGroupId.getId()))
- .fetchSchemaBlackList(patternTree)) {
+ ISchemaRegion schemaRegion =
+ schemaEngine.getSchemaRegion(new SchemaRegionId(consensusGroupId.getId()));
+ PathPatternTree filteredPatternTree =
+ filterPathPatternTree(patternTree, schemaRegion.getStorageGroupFullPath());
+ if (filteredPatternTree.isEmpty()) {
+ continue;
+ }
+ for (PartialPath path : schemaRegion.fetchSchemaBlackList(filteredPatternTree)) {
result.appendFullPath(path);
}
} catch (MetadataException e) {
@@ -465,10 +489,18 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
List<TSStatus> failureList = new ArrayList<>();
TSStatus status;
for (TConsensusGroupId consensusGroupId : req.getSchemaRegionIdList()) {
+ String storageGroup =
+ schemaEngine
+ .getSchemaRegion(new SchemaRegionId(consensusGroupId.getId()))
+ .getStorageGroupFullPath();
+ PathPatternTree filteredPatternTree = filterPathPatternTree(patternTree, storageGroup);
+ if (filteredPatternTree.isEmpty()) {
+ continue;
+ }
status =
regionManager.executeSchemaPlanNode(
new SchemaRegionId(consensusGroupId.getId()),
- new DeleteTimeSeriesNode(new PlanNodeId(""), patternTree));
+ new DeleteTimeSeriesNode(new PlanNodeId(""), filteredPatternTree));
if (status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
failureList.add(status);
}
@@ -481,6 +513,21 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
return RpcUtils.SUCCESS_STATUS;
}
+ private PathPatternTree filterPathPatternTree(PathPatternTree patternTree, String storageGroup) {
+ PathPatternTree filteredPatternTree = new PathPatternTree();
+ try {
+ PartialPath storageGroupPattern =
+ new PartialPath(storageGroup).concatNode(MULTI_LEVEL_PATH_WILDCARD);
+ for (PartialPath pathPattern : patternTree.getOverlappedPathPatterns(storageGroupPattern)) {
+ filteredPatternTree.appendPathPattern(pathPattern);
+ }
+ filteredPatternTree.constructTree();
+ } catch (IllegalPathException e) {
+ // won't reach here
+ }
+ return filteredPatternTree;
+ }
+
@Override
public THeartbeatResp getDataNodeHeartBeat(THeartbeatReq req) throws TException {
THeartbeatResp resp = new THeartbeatResp();