This is an automated email from the ASF dual-hosted git repository. xingtanzjr pushed a commit to branch xingtanzjr/fix_schema_fetch in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e13fef1f1cc733e3f9c564a9892487270dc2ee98 Author: Jinrui.Zhang <[email protected]> AuthorDate: Tue Jun 14 15:52:39 2022 +0800 fix the issue for fetch schema partition --- .../iotdb/confignode/manager/ConfigManager.java | 57 ++++++++++++++-------- .../apache/iotdb/confignode/manager/Manager.java | 2 +- .../thrift/ConfigNodeRPCServiceProcessor.java | 2 +- .../thrift/ConfigNodeRPCServiceProcessorTest.java | 4 +- .../org/apache/iotdb/commons/path/PartialPath.java | 26 ---------- .../apache/iotdb/db/mpp/plan/analyze/Analyzer.java | 18 ++++--- .../mpp/plan/analyze/ClusterPartitionFetcher.java | 11 ++--- .../db/mpp/plan/analyze/ClusterSchemaFetcher.java | 8 ++- .../mpp/plan/analyze/FakePartitionFetcherImpl.java | 2 +- .../db/mpp/plan/analyze/IPartitionFetcher.java | 2 +- .../plan/analyze/StandalonePartitionFetcher.java | 2 +- .../src/main/thrift/confignode.thrift | 1 - 12 files changed, 63 insertions(+), 72 deletions(-) diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java index 86aa821e05..23d62877c9 100644 --- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java +++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java @@ -22,6 +22,8 @@ package org.apache.iotdb.confignode.manager; import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot; import org.apache.iotdb.commons.conf.IoTDBConstant; +import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.commons.partition.SchemaPartition; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.utils.AuthUtils; import org.apache.iotdb.commons.utils.PathUtils; @@ -63,6 +65,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterResp; import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp; import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema; import org.apache.iotdb.consensus.common.DataSet; +import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.mpp.common.schematree.PathPatternTree; import org.apache.iotdb.rpc.TSStatusCode; @@ -79,6 +82,8 @@ import java.util.Map; import java.util.Set; import java.util.stream.Collectors; +import static org.apache.iotdb.commons.conf.IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD; + /** Entry of all management, AssignPartitionManager,AssignRegionManager. */ public class ConfigManager implements Manager { @@ -268,27 +273,24 @@ public class ConfigManager implements Manager { } private List<TSeriesPartitionSlot> calculateRelatedSlot( - PartialPath path, boolean matchTimeseries) { + PartialPath path, PartialPath storageGroup) { // The path contains `**` if (path.getFullPath().contains(IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD)) { return new ArrayList<>(); } - PartialPath innerPath; - if (!matchTimeseries - && path.getFullPath().contains(IoTDBConstant.ONE_LEVEL_PATH_WILDCARD)) { - return new ArrayList<>(); - } + List<PartialPath> innerPathList = path.alterPrefixPath(storageGroup); + PartialPath pathInnerStorageGroup = innerPathList.get(0); // The path contains `*` and the only `*` is not in last level - if (path.getDevice().contains(IoTDBConstant.ONE_LEVEL_PATH_WILDCARD)) { + if (pathInnerStorageGroup.getDevice().contains(IoTDBConstant.ONE_LEVEL_PATH_WILDCARD)) { return new ArrayList<>(); } return Collections.singletonList( - getPartitionManager().getSeriesPartitionSlot(path.getDevice())); + getPartitionManager().getSeriesPartitionSlot(pathInnerStorageGroup.getDevice())); } @Override - public DataSet getSchemaPartition(PathPatternTree patternTree, boolean matchTimeseries) { + public DataSet getSchemaPartition(PathPatternTree patternTree) { TSStatus status = confirmLeader(); if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { GetSchemaPartitionReq getSchemaPartitionReq = new GetSchemaPartitionReq(); @@ -298,21 +300,38 @@ public class ConfigManager implements Manager { Map<String, Boolean> scanAllRegions = new HashMap<>(); for (PartialPath path : relatedPaths) { for (String storageGroup : allStorageGroups) { - if (path.belongToStorageGroup(storageGroup) - && !scanAllRegions.containsKey(storageGroup)) { - List<TSeriesPartitionSlot> relatedSlot = calculateRelatedSlot(path, matchTimeseries); - if (relatedSlot.isEmpty()) { - scanAllRegions.put(storageGroup, true); - partitionSlotsMap.put(storageGroup, new HashSet<>()); - } else { - partitionSlotsMap - .computeIfAbsent(storageGroup, k -> new HashSet<>()) - .addAll(relatedSlot); + try { + PartialPath storageGroupPath = new PartialPath(storageGroup); + if (path.overlapWith(storageGroupPath.concatNode(MULTI_LEVEL_PATH_WILDCARD)) + && !scanAllRegions.containsKey(storageGroup)) { + List<TSeriesPartitionSlot> relatedSlot = calculateRelatedSlot(path, storageGroupPath); + if (relatedSlot.isEmpty()) { + scanAllRegions.put(storageGroup, true); + partitionSlotsMap.put(storageGroup, new HashSet<>()); + } else { + partitionSlotsMap + .computeIfAbsent(storageGroup, k -> new HashSet<>()) + .addAll(relatedSlot); + } } + } catch (IllegalPathException e) { + e.printStackTrace(); } } } + // return empty partition + if (partitionSlotsMap.isEmpty()) { + SchemaPartitionResp resp = new SchemaPartitionResp(); + resp.setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); + // TODO: (xingtanzjr) replace with new data structure + resp.setSchemaPartition( + new SchemaPartition( + new HashMap<>(), + IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionExecutorClass(), + IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionSlotNum())); + } + getSchemaPartitionReq.setPartitionSlotsMap( partitionSlotsMap.entrySet().stream() .collect(Collectors.toMap(Map.Entry::getKey, e -> new ArrayList<>(e.getValue())))); diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/Manager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/Manager.java index 32c9b2c8c4..50a931bb59 100644 --- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/Manager.java +++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/Manager.java @@ -152,7 +152,7 @@ public interface Manager { * * @return SchemaPartitionDataSet */ - DataSet getSchemaPartition(PathPatternTree patternTree, boolean matchTimeseries); + DataSet getSchemaPartition(PathPatternTree patternTree); /** * Get or create SchemaPartition diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java index a3f0ef3df3..2448d86b02 100644 --- a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java +++ b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java @@ -258,7 +258,7 @@ public class ConfigNodeRPCServiceProcessor implements ConfigIService.Iface { PathPatternTree patternTree = PathPatternTree.deserialize(ByteBuffer.wrap(req.getPathPatternTree())); SchemaPartitionResp schemaResp = - (SchemaPartitionResp) configManager.getSchemaPartition(patternTree, req.matchTimeseries); + (SchemaPartitionResp) configManager.getSchemaPartition(patternTree); TSchemaPartitionResp resp = new TSchemaPartitionResp(); schemaResp.convertToRpcSchemaPartitionResp(resp); diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessorTest.java b/confignode/src/test/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessorTest.java index 7b0d21f7a5..1b9cbb8357 100644 --- a/confignode/src/test/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessorTest.java +++ b/confignode/src/test/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessorTest.java @@ -382,7 +382,7 @@ public class ConfigNodeRPCServiceProcessorTest { // Test getOrCreateSchemaPartition, the result should be NOT_ENOUGH_DATANODE buffer = generatePatternTreeBuffer(new String[] {d00, d01, allSg1}); - schemaPartitionReq = new TSchemaPartitionReq(buffer, true); + schemaPartitionReq = new TSchemaPartitionReq(buffer); schemaPartitionResp = processor.getOrCreateSchemaPartition(schemaPartitionReq); Assert.assertEquals( TSStatusCode.NOT_ENOUGH_DATA_NODE.getStatusCode(), @@ -394,7 +394,7 @@ public class ConfigNodeRPCServiceProcessorTest { // Test getSchemaPartition, the result should be empty buffer = generatePatternTreeBuffer(new String[] {d00, d01, allSg1}); - schemaPartitionReq = new TSchemaPartitionReq(buffer, true); + schemaPartitionReq = new TSchemaPartitionReq(buffer); schemaPartitionResp = processor.getSchemaPartition(schemaPartitionReq); Assert.assertEquals( TSStatusCode.SUCCESS_STATUS.getStatusCode(), schemaPartitionResp.getStatus().getCode()); diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/path/PartialPath.java b/node-commons/src/main/java/org/apache/iotdb/commons/path/PartialPath.java index 68ff8f7638..e2d51f429f 100644 --- a/node-commons/src/main/java/org/apache/iotdb/commons/path/PartialPath.java +++ b/node-commons/src/main/java/org/apache/iotdb/commons/path/PartialPath.java @@ -353,32 +353,6 @@ public class PartialPath extends Path implements Comparable<Path>, Cloneable { return this.nodes.length == rNodes.length; } - /** - * Test whether the path belongs to given storage group or not. Example: [result]: [current path], - * [storage group path] 1) True: root.sg, root.sg 2) True: root.**, root.sg 3) True: root.sg.d1, - * root.sg 4) True: root.**.d1, root.sg 5) False: root.sg, root.sg2 6) False: root.sg.d1, root.sg2 - * - * @param storageGroup the storage group to be tested - */ - public boolean belongToStorageGroup(String storageGroup) { - try { - // We divide the judgement into two scenarios: - // 1) the size of nodes of the path is larger than size of storage groups nodes - // Then we can test the overlap of them - PartialPath storageGroupPath = - new PartialPath(storageGroup).concatNode(MULTI_LEVEL_PATH_WILDCARD); - if (overlapWith(storageGroupPath)) { - return true; - } - // 2) the size of nodes of the path is less or equal to size of storage groups nodes - // Then we test the prefix match - return storageGroup.startsWith(getFullPath()); - } catch (IllegalPathException e) { - // This code won't be reached in general - throw new RuntimeException(e); - } - } - @Override public String getFullPath() { if (fullPath == null) { diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analyzer.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analyzer.java index 58e3ae1c86..834ee25932 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analyzer.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analyzer.java @@ -1015,7 +1015,7 @@ public class Analyzer { SchemaPartition schemaPartitionInfo; schemaPartitionInfo = partitionFetcher.getSchemaPartition( - new PathPatternTree(alterTimeSeriesStatement.getPath()), true); + new PathPatternTree(alterTimeSeriesStatement.getPath())); analysis.setSchemaPartitionInfo(schemaPartitionInfo); return analysis; } @@ -1140,7 +1140,7 @@ public class Analyzer { SchemaPartition schemaPartitionInfo = partitionFetcher.getSchemaPartition( - new PathPatternTree(showTimeSeriesStatement.getPathPattern()), true); + new PathPatternTree(showTimeSeriesStatement.getPathPattern())); analysis.setSchemaPartitionInfo(schemaPartitionInfo); if (showTimeSeriesStatement.isOrderByHeat()) { @@ -1202,7 +1202,10 @@ public class Analyzer { SchemaPartition schemaPartitionInfo = partitionFetcher.getSchemaPartition( - new PathPatternTree(showDevicesStatement.getPathPattern()), false); + new PathPatternTree( + showDevicesStatement + .getPathPattern() + .concatNode(IoTDBConstant.ONE_LEVEL_PATH_WILDCARD))); analysis.setSchemaPartitionInfo(schemaPartitionInfo); analysis.setRespDatasetHeader( @@ -1250,8 +1253,7 @@ public class Analyzer { new PathPatternTree( countDevicesStatement .getPartialPath() - .concatNode(IoTDBConstant.ONE_LEVEL_PATH_WILDCARD)), - false); + .concatNode(IoTDBConstant.ONE_LEVEL_PATH_WILDCARD))); analysis.setSchemaPartitionInfo(schemaPartitionInfo); analysis.setRespDatasetHeader(HeaderConstant.countDevicesHeader); @@ -1266,7 +1268,7 @@ public class Analyzer { SchemaPartition schemaPartitionInfo = partitionFetcher.getSchemaPartition( - new PathPatternTree(countTimeSeriesStatement.getPartialPath()), true); + new PathPatternTree(countTimeSeriesStatement.getPartialPath())); analysis.setSchemaPartitionInfo(schemaPartitionInfo); analysis.setRespDatasetHeader(HeaderConstant.countTimeSeriesHeader); @@ -1281,7 +1283,7 @@ public class Analyzer { SchemaPartition schemaPartitionInfo = partitionFetcher.getSchemaPartition( - new PathPatternTree(countLevelTimeSeriesStatement.getPartialPath()), true); + new PathPatternTree(countLevelTimeSeriesStatement.getPartialPath())); analysis.setSchemaPartitionInfo(schemaPartitionInfo); analysis.setRespDatasetHeader(HeaderConstant.countLevelTimeSeriesHeader); @@ -1360,7 +1362,7 @@ public class Analyzer { PathPatternTree patternTree = new PathPatternTree(deleteDataStatement.getPathList()); - SchemaPartition schemaPartition = partitionFetcher.getSchemaPartition(patternTree, true); + SchemaPartition schemaPartition = partitionFetcher.getSchemaPartition(patternTree); SchemaTree schemaTree = schemaFetcher.fetchSchema(patternTree, schemaPartition); analysis.setSchemaTree(schemaTree); diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterPartitionFetcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterPartitionFetcher.java index 6f552f729c..29e0c6f526 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterPartitionFetcher.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterPartitionFetcher.java @@ -105,7 +105,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher { } @Override - public SchemaPartition getSchemaPartition(PathPatternTree patternTree, boolean matchTimeseries) { + public SchemaPartition getSchemaPartition(PathPatternTree patternTree) { try (DataNodeToConfigNodeClient client = configNodeClientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) { patternTree.constructTree(); @@ -114,7 +114,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher { SchemaPartition schemaPartition = partitionCache.getSchemaPartition(deviceToStorageGroupMap); if (null == schemaPartition) { TSchemaPartitionResp schemaPartitionResp = - client.getSchemaPartition(constructSchemaPartitionReq(patternTree, matchTimeseries)); + client.getSchemaPartition(constructSchemaPartitionReq(patternTree)); if (schemaPartitionResp.getStatus().getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { schemaPartition = parseSchemaPartitionResp(schemaPartitionResp); @@ -138,7 +138,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher { SchemaPartition schemaPartition = partitionCache.getSchemaPartition(deviceToStorageGroupMap); if (null == schemaPartition) { TSchemaPartitionResp schemaPartitionResp = - client.getOrCreateSchemaPartition(constructSchemaPartitionReq(patternTree, true)); + client.getOrCreateSchemaPartition(constructSchemaPartitionReq(patternTree)); if (schemaPartitionResp.getStatus().getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { schemaPartition = parseSchemaPartitionResp(schemaPartitionResp); @@ -348,15 +348,14 @@ public class ClusterPartitionFetcher implements IPartitionFetcher { return result; } - private TSchemaPartitionReq constructSchemaPartitionReq( - PathPatternTree patternTree, boolean matchTimeseries) { + private TSchemaPartitionReq constructSchemaPartitionReq(PathPatternTree patternTree) { PublicBAOS baos = new PublicBAOS(); try { patternTree.serialize(baos); ByteBuffer serializedPatternTree = ByteBuffer.allocate(baos.size()); serializedPatternTree.put(baos.getBuf(), 0, baos.size()); serializedPatternTree.flip(); - return new TSchemaPartitionReq(serializedPatternTree, matchTimeseries); + return new TSchemaPartitionReq(serializedPatternTree); } catch (IOException e) { throw new StatementAnalyzeException("An error occurred when serializing pattern tree"); } diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterSchemaFetcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterSchemaFetcher.java index cad131ee3e..cf7c6bb9af 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterSchemaFetcher.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ClusterSchemaFetcher.java @@ -78,7 +78,7 @@ public class ClusterSchemaFetcher implements ISchemaFetcher { @Override public SchemaTree fetchSchema(PathPatternTree patternTree) { - return fetchSchema(patternTree, partitionFetcher.getSchemaPartition(patternTree, true)); + return fetchSchema(patternTree, partitionFetcher.getSchemaPartition(patternTree)); } @Override @@ -144,8 +144,7 @@ public class ClusterSchemaFetcher implements ISchemaFetcher { SchemaTree remoteSchemaTree; if (!config.isAutoCreateSchemaEnabled()) { - remoteSchemaTree = - fetchSchema(patternTree, partitionFetcher.getSchemaPartition(patternTree, true)); + remoteSchemaTree = fetchSchema(patternTree, partitionFetcher.getSchemaPartition(patternTree)); schemaTree.mergeSchemaTree(remoteSchemaTree); schemaCache.put(remoteSchemaTree); return schemaTree; @@ -198,8 +197,7 @@ public class ClusterSchemaFetcher implements ISchemaFetcher { SchemaTree remoteSchemaTree; if (!config.isAutoCreateSchemaEnabled()) { - remoteSchemaTree = - fetchSchema(patternTree, partitionFetcher.getSchemaPartition(patternTree, true)); + remoteSchemaTree = fetchSchema(patternTree, partitionFetcher.getSchemaPartition(patternTree)); schemaTree.mergeSchemaTree(remoteSchemaTree); schemaCache.put(remoteSchemaTree); return schemaTree; diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/FakePartitionFetcherImpl.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/FakePartitionFetcherImpl.java index 38499227a0..8738ae9cd9 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/FakePartitionFetcherImpl.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/FakePartitionFetcherImpl.java @@ -42,7 +42,7 @@ import java.util.Map; public class FakePartitionFetcherImpl implements IPartitionFetcher { @Override - public SchemaPartition getSchemaPartition(PathPatternTree patternTree, boolean matchTimeseries) { + public SchemaPartition getSchemaPartition(PathPatternTree patternTree) { return null; } diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/IPartitionFetcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/IPartitionFetcher.java index 8755d0c65b..93f0b6a202 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/IPartitionFetcher.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/IPartitionFetcher.java @@ -29,7 +29,7 @@ import java.util.Map; public interface IPartitionFetcher { - SchemaPartition getSchemaPartition(PathPatternTree patternTree, boolean matchTimeseries); + SchemaPartition getSchemaPartition(PathPatternTree patternTree); SchemaPartition getOrCreateSchemaPartition(PathPatternTree patternTree); diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/StandalonePartitionFetcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/StandalonePartitionFetcher.java index 37175cee1b..d389e124f3 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/StandalonePartitionFetcher.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/StandalonePartitionFetcher.java @@ -67,7 +67,7 @@ public class StandalonePartitionFetcher implements IPartitionFetcher { } @Override - public SchemaPartition getSchemaPartition(PathPatternTree patternTree, boolean matchTimeseries) { + public SchemaPartition getSchemaPartition(PathPatternTree patternTree) { patternTree.constructTree(); return new SchemaPartition( localConfigNode.getSchemaPartition(patternTree), diff --git a/thrift-confignode/src/main/thrift/confignode.thrift b/thrift-confignode/src/main/thrift/confignode.thrift index fb280813bd..3f16b09684 100644 --- a/thrift-confignode/src/main/thrift/confignode.thrift +++ b/thrift-confignode/src/main/thrift/confignode.thrift @@ -108,7 +108,6 @@ struct TStorageGroupSchema { // SchemaPartition struct TSchemaPartitionReq { 1: required binary pathPatternTree - 2: required bool matchTimeseries } struct TSchemaPartitionResp {
