You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by xi...@apache.org on 2022/04/14 10:49:42 UTC
[iotdb] branch master updated: Implement basic SchemaFetcher (#5494)
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr 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 8c09a47c5d Implement basic SchemaFetcher (#5494)
8c09a47c5d is described below
commit 8c09a47c5de866484799ad4835420e88317b57ec
Author: Marcos_Zyk <38...@users.noreply.github.com>
AuthorDate: Thu Apr 14 18:49:38 2022 +0800
Implement basic SchemaFetcher (#5494)
---
.../statemachine/SchemaRegionStateMachine.java | 5 +-
.../exception/sql/StatementAnalyzeException.java | 13 +-
.../db/metadata/schemaregion/SchemaRegion.java | 3 +-
.../db/mpp/common/schematree/PathPatternNode.java | 13 ++
.../db/mpp/common/schematree/PathPatternTree.java | 43 +++++-
.../db/mpp/common/schematree/SchemaEntityNode.java | 24 ++++
.../mpp/common/schematree/SchemaInternalNode.java | 17 +++
.../common/schematree/SchemaMeasurementNode.java | 22 ++-
.../iotdb/db/mpp/common/schematree/SchemaNode.java | 8 +-
.../iotdb/db/mpp/common/schematree/SchemaTree.java | 129 +++++++++++++++---
.../mpp/operator/schema/SchemaFetchOperator.java | 129 ++++++++++++++++++
.../apache/iotdb/db/mpp/sql/analyze/Analyzer.java | 109 ++++++---------
.../mpp/sql/analyze/ClusterPartitionFetcher.java | 29 ++--
.../db/mpp/sql/analyze/ClusterSchemaFetcher.java | 65 ++++++++-
.../db/mpp/sql/analyze/IPartitionFetcher.java | 12 +-
.../db/mpp/sql/planner/DistributionPlanner.java | 22 +--
.../db/mpp/sql/planner/LocalExecutionPlanner.java | 22 ++-
.../iotdb/db/mpp/sql/planner/LogicalPlanner.java | 16 ++-
.../iotdb/db/mpp/sql/planner/QueryPlanBuilder.java | 20 ++-
.../db/mpp/sql/planner/plan/node/PlanNodeType.java | 6 +-
.../db/mpp/sql/planner/plan/node/PlanVisitor.java | 13 +-
.../{SchemaMergeNode.java => SchemaFetchNode.java} | 54 +++++---
.../plan/node/metedata/read/SchemaMergeNode.java | 2 +-
.../plan/node/metedata/read/SchemaScanNode.java | 11 +-
.../db/mpp/sql/statement/StatementVisitor.java | 5 +
.../statement/metadata/SchemaFetchStatement.java | 56 ++++++++
.../{ => schematree}/PathPatternTreeTest.java | 25 +++-
.../common/{ => schematree}/SchemaTreeTest.java | 93 +++++++++++--
.../operator/schema/SchemaFetchOperatorTest.java | 149 +++++++++++++++++++++
29 files changed, 930 insertions(+), 185 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/SchemaRegionStateMachine.java b/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/SchemaRegionStateMachine.java
index e5649f2733..f964522d29 100644
--- a/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/SchemaRegionStateMachine.java
+++ b/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/SchemaRegionStateMachine.java
@@ -35,6 +35,8 @@ public class SchemaRegionStateMachine extends BaseStateMachine {
private static final Logger logger = LoggerFactory.getLogger(SchemaRegionStateMachine.class);
private final ISchemaRegion schemaRegion;
+ private static final FragmentInstanceManager QUERY_INSTANCE_MANAGER =
+ FragmentInstanceManager.getInstance();
public SchemaRegionStateMachine(ISchemaRegion schemaRegion) {
this.schemaRegion = schemaRegion;
@@ -57,7 +59,6 @@ public class SchemaRegionStateMachine extends BaseStateMachine {
@Override
protected DataSet read(FragmentInstance fragmentInstance) {
logger.info("Execute read plan in SchemaRegionStateMachine");
- return FragmentInstanceManager.getInstance()
- .execSchemaQueryFragmentInstance(fragmentInstance, schemaRegion);
+ return QUERY_INSTANCE_MANAGER.execSchemaQueryFragmentInstance(fragmentInstance, schemaRegion);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/exception/sql/StatementAnalyzeException.java b/server/src/main/java/org/apache/iotdb/db/exception/sql/StatementAnalyzeException.java
index 4d4cea20cc..333c94e385 100644
--- a/server/src/main/java/org/apache/iotdb/db/exception/sql/StatementAnalyzeException.java
+++ b/server/src/main/java/org/apache/iotdb/db/exception/sql/StatementAnalyzeException.java
@@ -19,27 +19,22 @@
package org.apache.iotdb.db.exception.sql;
-import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.db.mpp.sql.constant.FilterConstant;
-import org.apache.iotdb.rpc.TSStatusCode;
-public class StatementAnalyzeException extends IoTDBException {
+public class StatementAnalyzeException extends RuntimeException {
public StatementAnalyzeException(String message) {
- super(message, TSStatusCode.LOGICAL_OPTIMIZE_ERROR.getStatusCode());
+ super(message);
}
public StatementAnalyzeException(String filterOperator, FilterConstant.FilterType filterType) {
super(
String.format(
"Unknown token in [%s]: [%s], [%s].",
- filterOperator, filterType, FilterConstant.filterNames.get(filterType)),
- TSStatusCode.LOGICAL_OPTIMIZE_ERROR.getStatusCode());
+ filterOperator, filterType, FilterConstant.filterNames.get(filterType)));
}
public StatementAnalyzeException(String type, String message) {
- super(
- String.format("Unsupported type: [%s]. %s", type, message),
- TSStatusCode.LOGICAL_OPTIMIZE_ERROR.getStatusCode());
+ super(String.format("Unsupported type: [%s]. %s", type, message));
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/SchemaRegion.java b/server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/SchemaRegion.java
index 38f1582fcd..7599a45942 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/SchemaRegion.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/SchemaRegion.java
@@ -51,7 +51,6 @@ import org.apache.iotdb.db.metadata.path.MeasurementPath;
import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.db.metadata.rescon.MemoryStatistics;
import org.apache.iotdb.db.metadata.rescon.TimeseriesStatistics;
-import org.apache.iotdb.db.metadata.storagegroup.StorageGroupSchemaManager;
import org.apache.iotdb.db.metadata.tag.TagManager;
import org.apache.iotdb.db.metadata.template.Template;
import org.apache.iotdb.db.metadata.template.TemplateManager;
@@ -148,7 +147,7 @@ import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.PATH_SEPARA
@SuppressWarnings("java:S1135") // ignore todos
public class SchemaRegion implements ISchemaRegion {
- private static final Logger logger = LoggerFactory.getLogger(StorageGroupSchemaManager.class);
+ private static final Logger logger = LoggerFactory.getLogger(SchemaRegion.class);
protected static IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternNode.java
index d780c52045..baf0a5a18e 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternNode.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
@@ -89,6 +90,18 @@ public class PathPatternNode {
return true;
}
+ public void serialize(ByteBuffer buffer) {
+ ReadWriteIOUtils.write(name, buffer);
+ ReadWriteIOUtils.write(children.size(), buffer);
+ serializeChildren(buffer);
+ }
+
+ void serializeChildren(ByteBuffer buffer) {
+ for (PathPatternNode childNode : children.values()) {
+ childNode.serialize(buffer);
+ }
+ }
+
public void serialize(PublicBAOS outputStream) throws IOException {
ReadWriteIOUtils.write(name, outputStream);
ReadWriteIOUtils.write(children.size(), outputStream);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternTree.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternTree.java
index 8384499b7d..f3b2eeeedf 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternTree.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternTree.java
@@ -29,8 +29,11 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Deque;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
@@ -45,10 +48,10 @@ public class PathPatternTree {
this.root = root;
}
- public PathPatternTree(PartialPath deivcePath, String[] measurements) {
+ public PathPatternTree(PartialPath devicePath, String[] measurements) {
this.root = new PathPatternNode(SQLConstant.ROOT);
this.pathList = new ArrayList<>();
- appendPaths(deivcePath, Arrays.asList(measurements));
+ appendPaths(devicePath, Arrays.asList(measurements));
}
public PathPatternTree(PartialPath deivcePath, List<String> measurements) {
@@ -186,6 +189,11 @@ public class PathPatternTree {
root.serialize(outputStream);
}
+ public void serialize(ByteBuffer buffer) {
+ constructTree();
+ root.serialize(buffer);
+ }
+
public static PathPatternTree deserialize(ByteBuffer buffer) {
PathPatternNode root = deserializeNode(buffer);
return new PathPatternTree(root);
@@ -202,6 +210,37 @@ public class PathPatternTree {
return node;
}
+ public List<PartialPath> splitToPathList() {
+ List<PartialPath> result = new ArrayList<>();
+ Deque<String> ancestors = new ArrayDeque<>();
+ searchFullPath(root, ancestors, result);
+ return result;
+ }
+
+ private void searchFullPath(
+ PathPatternNode node, Deque<String> ancestors, List<PartialPath> fullPaths) {
+ if (node.isLeaf()) {
+ fullPaths.add(constructFullPath(node, ancestors));
+ return;
+ }
+
+ ancestors.push(node.getName());
+ for (PathPatternNode child : node.getChildren().values()) {
+ searchFullPath(child, ancestors, fullPaths);
+ }
+ ancestors.pop();
+ }
+
+ private PartialPath constructFullPath(PathPatternNode node, Deque<String> ancestors) {
+ Iterator<String> iterator = ancestors.descendingIterator();
+ List<String> nodeList = new ArrayList<>(ancestors.size() + 1);
+ while (iterator.hasNext()) {
+ nodeList.add(iterator.next());
+ }
+ nodeList.add(node.getName());
+ return new PartialPath(nodeList.toArray(new String[0]));
+ }
+
@TestOnly
public boolean equalWith(PathPatternTree that) {
if (this == that) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaEntityNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaEntityNode.java
index 3360286d5d..d477c16194 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaEntityNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaEntityNode.java
@@ -59,6 +59,30 @@ public class SchemaEntityNode extends SchemaInternalNode {
isAligned = aligned;
}
+ @Override
+ public void replaceChild(String name, SchemaNode newChild) {
+ super.replaceChild(name, newChild);
+ if (newChild.isMeasurement()) {
+ SchemaMeasurementNode measurementNode = newChild.getAsMeasurementNode();
+ if (measurementNode.getAlias() != null) {
+ aliasChildren.replace(name, measurementNode);
+ }
+ }
+ }
+
+ @Override
+ public void copyDataTo(SchemaNode schemaNode) {
+ if (!schemaNode.isEntity()) {
+ return;
+ }
+ SchemaEntityNode entityNode = schemaNode.getAsEntityNode();
+ if (aliasChildren != null) {
+ for (SchemaMeasurementNode child : aliasChildren.values()) {
+ entityNode.addAliasChild(child.getAlias(), child);
+ }
+ }
+ }
+
@Override
public boolean isEntity() {
return true;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaInternalNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaInternalNode.java
index f93c5e4940..215ac84e98 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaInternalNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaInternalNode.java
@@ -46,6 +46,23 @@ public class SchemaInternalNode extends SchemaNode {
children.put(name, child);
}
+ @Override
+ public void replaceChild(String name, SchemaNode newChild) {
+ SchemaNode oldChild = children.get(name);
+ oldChild.copyDataTo(newChild);
+ children.replace(name, newChild);
+ }
+
+ @Override
+ public void copyDataTo(SchemaNode schemaNode) {
+ if (schemaNode.isMeasurement()) {
+ return;
+ }
+ for (SchemaNode child : children.values()) {
+ schemaNode.addChild(child.getName(), child);
+ }
+ }
+
@Override
public Map<String, SchemaNode> getChildren() {
return children;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaMeasurementNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaMeasurementNode.java
index a2c9f13c53..b8d76be279 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaMeasurementNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaMeasurementNode.java
@@ -27,7 +27,7 @@ import java.nio.ByteBuffer;
public class SchemaMeasurementNode extends SchemaNode {
private String alias;
- private final MeasurementSchema schema;
+ private MeasurementSchema schema;
public SchemaMeasurementNode(String name, MeasurementSchema schema) {
super(name);
@@ -46,6 +46,26 @@ public class SchemaMeasurementNode extends SchemaNode {
return schema;
}
+ @Override
+ public void replaceChild(String name, SchemaNode newChild) {
+ throw new UnsupportedOperationException(
+ "This operation is not supported in SchemaMeasurementNode.");
+ }
+
+ @Override
+ public void copyDataTo(SchemaNode schemaNode) {
+ if (!schemaNode.isMeasurement()) {
+ return;
+ }
+ SchemaMeasurementNode measurementNode = schemaNode.getAsMeasurementNode();
+ measurementNode.setSchema(schema);
+ measurementNode.setAlias(alias);
+ }
+
+ private void setSchema(MeasurementSchema schema) {
+ this.schema = schema;
+ }
+
@Override
public boolean isMeasurement() {
return true;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaNode.java
index 944bad3fd2..ed295fff42 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaNode.java
@@ -46,6 +46,10 @@ public abstract class SchemaNode {
public void addChild(String name, SchemaNode child) {}
+ public abstract void replaceChild(String name, SchemaNode newChild);
+
+ public abstract void copyDataTo(SchemaNode schemaNode);
+
public Map<String, SchemaNode> getChildren() {
return Collections.emptyMap();
}
@@ -63,11 +67,11 @@ public abstract class SchemaNode {
}
public SchemaEntityNode getAsEntityNode() {
- throw new UnsupportedOperationException("This not isn't instance of SchemaEntityNode.");
+ throw new UnsupportedOperationException("This node isn't instance of SchemaEntityNode.");
}
public SchemaMeasurementNode getAsMeasurementNode() {
- throw new UnsupportedOperationException("This not isn't instance of SchemaMeasurementNode.");
+ throw new UnsupportedOperationException("This node isn't instance of SchemaMeasurementNode.");
}
public abstract byte getType();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaTree.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaTree.java
index a2cb149657..0913a495a1 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaTree.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/SchemaTree.java
@@ -19,26 +19,33 @@
package org.apache.iotdb.db.mpp.common.schematree;
+import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.metadata.path.MeasurementPath;
import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
-import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Deque;
import java.util.List;
+import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_ROOT;
import static org.apache.iotdb.db.mpp.common.schematree.SchemaNode.SCHEMA_ENTITY_NODE;
import static org.apache.iotdb.db.mpp.common.schematree.SchemaNode.SCHEMA_MEASUREMENT_NODE;
public class SchemaTree {
+ private List<String> storageGroups;
+
private final SchemaNode root;
+ public SchemaTree() {
+ root = new SchemaInternalNode(PATH_ROOT);
+ }
+
public SchemaTree(SchemaNode root) {
this.root = root;
}
@@ -57,19 +64,6 @@ public class SchemaTree {
return new Pair<>(visitor.getAllResult(), visitor.getNextOffset());
}
- /**
- * Get storage group name by path
- *
- * <p>e.g., root.sg1 is a storage group and path = root.sg1.d1, return root.sg1
- *
- * @param path only full path, cannot be path pattern
- * @return storage group in the given path
- */
- public String getBelongedStorageGroup(PartialPath path) {
- // TODO: @zyk
- throw new UnsupportedOperationException();
- }
-
public DeviceSchemaInfo searchDeviceSchemaInfo(
PartialPath devicePath, List<String> measurements) {
@@ -88,7 +82,81 @@ public class SchemaTree {
devicePath, cur.getAsEntityNode().isAligned(), measurementSchemaList);
}
- public void serialize(ByteBuffer buffer) throws IOException {
+ public void appendMeasurementPaths(List<MeasurementPath> measurementPathList) {
+ for (MeasurementPath measurementPath : measurementPathList) {
+ appendSingleMeasurementPath(measurementPath);
+ }
+ }
+
+ private void appendSingleMeasurementPath(MeasurementPath measurementPath) {
+ String[] nodes = measurementPath.getNodes();
+ SchemaNode cur = root;
+ SchemaNode child;
+ for (int i = 1; i < nodes.length; i++) {
+ child = cur.getChild(nodes[i]);
+ if (child == null) {
+ if (i == nodes.length - 1) {
+ SchemaMeasurementNode measurementNode =
+ new SchemaMeasurementNode(
+ nodes[i], (MeasurementSchema) measurementPath.getMeasurementSchema());
+ if (measurementPath.isMeasurementAliasExists()) {
+ measurementNode.setAlias(measurementPath.getMeasurementAlias());
+ cur.getAsEntityNode()
+ .addAliasChild(measurementPath.getMeasurementAlias(), measurementNode);
+ }
+ child = measurementNode;
+ } else if (i == nodes.length - 2) {
+ SchemaEntityNode entityNode = new SchemaEntityNode(nodes[i]);
+ entityNode.setAligned(measurementPath.isUnderAlignedEntity());
+ child = entityNode;
+ } else {
+ child = new SchemaInternalNode(nodes[i]);
+ }
+ cur.addChild(nodes[i], child);
+ } else if (i == nodes.length - 2 && !child.isEntity()) {
+ SchemaEntityNode entityNode = new SchemaEntityNode(nodes[i]);
+ cur.replaceChild(nodes[i], entityNode);
+ child = entityNode;
+ }
+ cur = child;
+ }
+ }
+
+ public void mergeSchemaTree(SchemaTree schemaTree) {
+ traverseAndMerge(this.root, null, schemaTree.root);
+ }
+
+ private void traverseAndMerge(SchemaNode thisNode, SchemaNode thisParent, SchemaNode thatNode) {
+ SchemaNode thisChild;
+ for (SchemaNode thatChild : thatNode.getChildren().values()) {
+ thisChild = thisNode.getChild(thatChild.getName());
+ if (thisChild == null) {
+ thisNode.addChild(thatChild.getName(), thatChild);
+ if (thatChild.isMeasurement()) {
+ SchemaEntityNode entityNode;
+ if (thisNode.isEntity()) {
+ entityNode = thisNode.getAsEntityNode();
+ } else {
+ entityNode = new SchemaEntityNode(thisNode.getName());
+ thisParent.replaceChild(thisNode.getName(), entityNode);
+ thisNode = entityNode;
+ }
+
+ if (!entityNode.isAligned()) {
+ entityNode.setAligned(thatNode.getAsEntityNode().isAligned());
+ }
+ SchemaMeasurementNode measurementNode = thatChild.getAsMeasurementNode();
+ if (measurementNode.getAlias() != null) {
+ entityNode.addAliasChild(measurementNode.getAlias(), measurementNode);
+ }
+ }
+ } else {
+ traverseAndMerge(thisChild, thisNode, thatChild);
+ }
+ }
+ }
+
+ public void serialize(ByteBuffer buffer) {
root.serialize(buffer);
}
@@ -131,4 +199,35 @@ public class SchemaTree {
}
return new SchemaTree(stack.poll());
}
+
+ /**
+ * Get storage group name by path
+ *
+ * <p>e.g., root.sg1 is a storage group and path = root.sg1.d1, return root.sg1
+ *
+ * @param path only full path, cannot be path pattern
+ * @return storage group in the given path
+ */
+ public String getBelongedStorageGroup(PartialPath path) {
+ for (String storageGroup : storageGroups) {
+ if (path.getFullPath().startsWith(storageGroup)) {
+ return storageGroup;
+ }
+ }
+ throw new RuntimeException(
+ "No matched storage group. Please check the path " + path.getFullPath());
+ }
+
+ public List<String> getStorageGroups() {
+ return storageGroups;
+ }
+
+ public void setStorageGroups(List<String> storageGroups) {
+ this.storageGroups = storageGroups;
+ }
+
+ @TestOnly
+ SchemaNode getRoot() {
+ return root;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/schema/SchemaFetchOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/schema/SchemaFetchOperator.java
new file mode 100644
index 0000000000..003da76f2d
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/schema/SchemaFetchOperator.java
@@ -0,0 +1,129 @@
+/*
+ * 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.mpp.operator.schema;
+
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.metadata.schemaregion.ISchemaRegion;
+import org.apache.iotdb.db.mpp.common.schematree.PathPatternTree;
+import org.apache.iotdb.db.mpp.common.schematree.SchemaTree;
+import org.apache.iotdb.db.mpp.operator.OperatorContext;
+import org.apache.iotdb.db.mpp.operator.source.SourceOperator;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+import org.apache.iotdb.tsfile.read.common.block.column.BinaryColumn;
+import org.apache.iotdb.tsfile.read.common.block.column.TimeColumn;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.BufferOverflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+
+public class SchemaFetchOperator implements SourceOperator {
+
+ private static final Logger logger = LoggerFactory.getLogger(SchemaFetchOperator.class);
+ private static final int MAX_BINARY_SIZE = 1024 * 1024;
+
+ private final PlanNodeId sourceId;
+ private final OperatorContext operatorContext;
+ private final PathPatternTree patternTree;
+ private final ISchemaRegion schemaRegion;
+
+ private TsBlock tsBlock;
+ private boolean isFinished = false;
+
+ public SchemaFetchOperator(
+ PlanNodeId planNodeId,
+ OperatorContext context,
+ PathPatternTree patternTree,
+ ISchemaRegion schemaRegion) {
+ this.sourceId = planNodeId;
+ this.operatorContext = context;
+ this.patternTree = patternTree;
+ this.schemaRegion = schemaRegion;
+ }
+
+ @Override
+ public OperatorContext getOperatorContext() {
+ return operatorContext;
+ }
+
+ @Override
+ public TsBlock next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ isFinished = true;
+ try {
+ fetchSchema();
+ } catch (MetadataException e) {
+ logger.error("Error occurred during execute SchemaFetchOperator {}", sourceId, e);
+ throw new RuntimeException(e);
+ }
+ return tsBlock;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return !isFinished;
+ }
+
+ @Override
+ public boolean isFinished() throws IOException {
+ return isFinished;
+ }
+
+ @Override
+ public PlanNodeId getSourceId() {
+ return sourceId;
+ }
+
+ private void fetchSchema() throws MetadataException {
+ SchemaTree schemaTree = new SchemaTree();
+ List<PartialPath> partialPathList = patternTree.splitToPathList();
+ for (PartialPath path : partialPathList) {
+ schemaTree.appendMeasurementPaths(schemaRegion.getMeasurementPaths(path, false));
+ }
+ ByteBuffer bufferWithMaxSize = ByteBuffer.allocate(MAX_BINARY_SIZE);
+ try {
+ schemaTree.serialize(bufferWithMaxSize);
+ } catch (BufferOverflowException e) {
+ logger.error("The size of schemaTree's binary data is too large. {}", sourceId, e);
+ throw e;
+ }
+
+ bufferWithMaxSize.flip();
+ ByteBuffer byteBuffer = ByteBuffer.allocate(bufferWithMaxSize.limit());
+ byteBuffer.put(bufferWithMaxSize);
+ this.tsBlock =
+ new TsBlock(
+ new TimeColumn(1, new long[] {0}),
+ new BinaryColumn(
+ 1,
+ Optional.of(new boolean[] {false}),
+ new Binary[] {new Binary(byteBuffer.array())}));
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/analyze/Analyzer.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/analyze/Analyzer.java
index 3e205637c0..bb7114c01f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/analyze/Analyzer.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/analyze/Analyzer.java
@@ -42,7 +42,13 @@ import org.apache.iotdb.db.mpp.sql.statement.StatementVisitor;
import org.apache.iotdb.db.mpp.sql.statement.component.ResultColumn;
import org.apache.iotdb.db.mpp.sql.statement.component.WhereCondition;
import org.apache.iotdb.db.mpp.sql.statement.crud.*;
-import org.apache.iotdb.db.mpp.sql.statement.metadata.*;
+import org.apache.iotdb.db.mpp.sql.statement.metadata.AlterTimeSeriesStatement;
+import org.apache.iotdb.db.mpp.sql.statement.metadata.CreateAlignedTimeSeriesStatement;
+import org.apache.iotdb.db.mpp.sql.statement.metadata.CreateTimeSeriesStatement;
+import org.apache.iotdb.db.mpp.sql.statement.metadata.SchemaFetchStatement;
+import org.apache.iotdb.db.mpp.sql.statement.metadata.ShowDevicesStatement;
+import org.apache.iotdb.db.mpp.sql.statement.metadata.ShowStorageGroupStatement;
+import org.apache.iotdb.db.mpp.sql.statement.metadata.ShowTimeSeriesStatement;
import org.apache.iotdb.db.mpp.sql.statement.sys.AuthorStatement;
import org.apache.iotdb.db.qp.constant.SQLConstant;
import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
@@ -194,14 +200,9 @@ public class Analyzer {
Analysis analysis = new Analysis();
analysis.setStatement(createTimeSeriesStatement);
- SchemaPartition schemaPartitionInfo;
- try {
- schemaPartitionInfo =
- partitionFetcher.getSchemaPartition(
- new PathPatternTree(createTimeSeriesStatement.getPath()));
- } catch (StatementAnalyzeException e) {
- throw new SemanticException("An error occurred when fetching schema partition infos");
- }
+ SchemaPartition schemaPartitionInfo =
+ partitionFetcher.getSchemaPartition(
+ new PathPatternTree(createTimeSeriesStatement.getPath()));
analysis.setSchemaPartitionInfo(schemaPartitionInfo);
return analysis;
}
@@ -221,15 +222,11 @@ public class Analyzer {
analysis.setStatement(createAlignedTimeSeriesStatement);
SchemaPartition schemaPartitionInfo;
- try {
- schemaPartitionInfo =
- partitionFetcher.getSchemaPartition(
- new PathPatternTree(
- createAlignedTimeSeriesStatement.getDevicePath(),
- createAlignedTimeSeriesStatement.getMeasurements()));
- } catch (StatementAnalyzeException e) {
- throw new SemanticException("An error occurred when fetching schema partition infos");
- }
+ schemaPartitionInfo =
+ partitionFetcher.getSchemaPartition(
+ new PathPatternTree(
+ createAlignedTimeSeriesStatement.getDevicePath(),
+ createAlignedTimeSeriesStatement.getMeasurements()));
analysis.setSchemaPartitionInfo(schemaPartitionInfo);
return analysis;
}
@@ -241,13 +238,9 @@ public class Analyzer {
analysis.setStatement(alterTimeSeriesStatement);
SchemaPartition schemaPartitionInfo;
- try {
- schemaPartitionInfo =
- partitionFetcher.getSchemaPartition(
- new PathPatternTree(alterTimeSeriesStatement.getPath()));
- } catch (StatementAnalyzeException e) {
- throw new SemanticException("An error occurred when fetching schema partition infos");
- }
+ schemaPartitionInfo =
+ partitionFetcher.getSchemaPartition(
+ new PathPatternTree(alterTimeSeriesStatement.getPath()));
analysis.setSchemaPartitionInfo(schemaPartitionInfo);
return analysis;
}
@@ -275,11 +268,7 @@ public class Analyzer {
schemaTree.getBelongedStorageGroup(insertTabletStatement.getDevicePath()),
Collections.singletonList(dataPartitionQueryParam));
DataPartition dataPartition;
- try {
- dataPartition = partitionFetcher.getDataPartition(sgNameToQueryParamsMap);
- } catch (StatementAnalyzeException e) {
- throw new SemanticException("An error occurred when fetching data partition infos");
- }
+ dataPartition = partitionFetcher.getDataPartition(sgNameToQueryParamsMap);
Analysis analysis = new Analysis();
analysis.setSchemaTree(schemaTree);
@@ -294,14 +283,9 @@ public class Analyzer {
Analysis analysis = new Analysis();
analysis.setStatement(showTimeSeriesStatement);
- SchemaPartition schemaPartitionInfo;
- try {
- schemaPartitionInfo =
- partitionFetcher.getSchemaPartition(
- new PathPatternTree(showTimeSeriesStatement.getPathPattern()));
- } catch (StatementAnalyzeException e) {
- throw new SemanticException("An error occurred when fetching schema partition infos");
- }
+ SchemaPartition schemaPartitionInfo =
+ partitionFetcher.getSchemaPartition(
+ new PathPatternTree(showTimeSeriesStatement.getPathPattern()));
analysis.setSchemaPartitionInfo(schemaPartitionInfo);
analysis.setRespDatasetHeader(HeaderConstant.showTimeSeriesHeader);
return analysis;
@@ -322,14 +306,10 @@ public class Analyzer {
Analysis analysis = new Analysis();
analysis.setStatement(showDevicesStatement);
- SchemaPartition schemaPartitionInfo;
- try {
- schemaPartitionInfo =
- partitionFetcher.getSchemaPartition(
- new PathPatternTree(showDevicesStatement.getPathPattern().concatNode("*")));
- } catch (StatementAnalyzeException e) {
- throw new SemanticException("An error occurred when fetching schema partition infos");
- }
+ SchemaPartition schemaPartitionInfo =
+ partitionFetcher.getSchemaPartition(
+ new PathPatternTree(showDevicesStatement.getPathPattern().concatNode("*")));
+
analysis.setSchemaPartitionInfo(schemaPartitionInfo);
analysis.setRespDatasetHeader(
showDevicesStatement.hasSgCol()
@@ -505,12 +485,7 @@ public class Analyzer {
sgNameToQueryParamsMap.put(
schemaTree.getBelongedStorageGroup(insertRowStatement.getDevicePath()),
Collections.singletonList(dataPartitionQueryParam));
- DataPartition dataPartition;
- try {
- dataPartition = partitionFetcher.getDataPartition(sgNameToQueryParamsMap);
- } catch (StatementAnalyzeException e) {
- throw new SemanticException("An error occurred when fetching data partition infos");
- }
+ DataPartition dataPartition = partitionFetcher.getDataPartition(sgNameToQueryParamsMap);
Analysis analysis = new Analysis();
analysis.setSchemaTree(schemaTree);
@@ -554,12 +529,7 @@ public class Analyzer {
key -> new ArrayList<>())
.add(dataPartitionQueryParam);
}
- DataPartition dataPartition;
- try {
- dataPartition = partitionFetcher.getDataPartition(sgNameToQueryParamsMap);
- } catch (StatementAnalyzeException e) {
- throw new SemanticException("An error occurred when fetching data partition infos");
- }
+ DataPartition dataPartition = partitionFetcher.getDataPartition(sgNameToQueryParamsMap);
Analysis analysis = new Analysis();
analysis.setSchemaTree(schemaTree);
@@ -597,12 +567,7 @@ public class Analyzer {
key -> new ArrayList<>())
.add(dataPartitionQueryParam);
}
- DataPartition dataPartition;
- try {
- dataPartition = partitionFetcher.getDataPartition(sgNameToQueryParamsMap);
- } catch (StatementAnalyzeException e) {
- throw new SemanticException("An error occurred when fetching data partition infos");
- }
+ DataPartition dataPartition = partitionFetcher.getDataPartition(sgNameToQueryParamsMap);
Analysis analysis = new Analysis();
analysis.setSchemaTree(schemaTree);
@@ -642,12 +607,7 @@ public class Analyzer {
sgNameToQueryParamsMap.put(
schemaTree.getBelongedStorageGroup(insertRowsOfOneDeviceStatement.getDevicePath()),
Collections.singletonList(dataPartitionQueryParam));
- DataPartition dataPartition;
- try {
- dataPartition = partitionFetcher.getDataPartition(sgNameToQueryParamsMap);
- } catch (StatementAnalyzeException e) {
- throw new SemanticException("An error occurred when fetching data partition infos");
- }
+ DataPartition dataPartition = partitionFetcher.getDataPartition(sgNameToQueryParamsMap);
Analysis analysis = new Analysis();
analysis.setSchemaTree(schemaTree);
@@ -656,5 +616,14 @@ public class Analyzer {
return analysis;
}
+
+ @Override
+ public Analysis visitSchemaFetch(
+ SchemaFetchStatement schemaFetchStatement, MPPQueryContext context) {
+ Analysis analysis = new Analysis();
+ analysis.setStatement(schemaFetchStatement);
+ analysis.setSchemaPartitionInfo(schemaFetchStatement.getSchemaPartition());
+ return analysis;
+ }
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/analyze/ClusterPartitionFetcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/analyze/ClusterPartitionFetcher.java
index be16ac6615..67e3ca12eb 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/analyze/ClusterPartitionFetcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/analyze/ClusterPartitionFetcher.java
@@ -57,7 +57,17 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
private SeriesPartitionExecutor partitionExecutor;
- public ClusterPartitionFetcher() throws StatementAnalyzeException {
+ private static final class ClusterPartitionFetcherHolder {
+ private static final ClusterPartitionFetcher INSTANCE = new ClusterPartitionFetcher();
+
+ private ClusterPartitionFetcherHolder() {}
+ }
+
+ public static ClusterPartitionFetcher getInstance() {
+ return ClusterPartitionFetcherHolder.INSTANCE;
+ }
+
+ private ClusterPartitionFetcher() {
try {
client = new ConfigNodeClient();
} catch (IoTDBConnectionException | BadNodeUrlException e) {
@@ -66,7 +76,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
initPartitionExecutor();
}
- private void initPartitionExecutor() throws StatementAnalyzeException {
+ private void initPartitionExecutor() {
// TODO: @crz move to node-commons
IoTDBConfig conf = IoTDBDescriptor.getInstance().getConfig();
try {
@@ -88,8 +98,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
}
@Override
- public SchemaPartition getSchemaPartition(PathPatternTree patternTree)
- throws StatementAnalyzeException {
+ public SchemaPartition getSchemaPartition(PathPatternTree patternTree) {
try {
TSchemaPartitionResp schemaPartitionResp =
client.getSchemaPartition(constructSchemaPartitionReq(patternTree));
@@ -104,8 +113,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
}
@Override
- public SchemaPartition getOrCreateSchemaPartition(PathPatternTree patternTree)
- throws StatementAnalyzeException {
+ public SchemaPartition getOrCreateSchemaPartition(PathPatternTree patternTree) {
try {
TSchemaPartitionResp schemaPartitionResp =
client.getOrCreateSchemaPartition(constructSchemaPartitionReq(patternTree));
@@ -122,8 +130,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
@Override
public DataPartition getDataPartition(
- Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap)
- throws StatementAnalyzeException {
+ Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap) {
try {
TDataPartitionResp dataPartitionResp =
client.getDataPartition(constructDataPartitionReq(sgNameToQueryParamsMap));
@@ -138,8 +145,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
@Override
public DataPartition getOrCreateDataPartition(
- Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap)
- throws StatementAnalyzeException {
+ Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap) {
try {
TDataPartitionResp dataPartitionResp =
client.getOrCreateDataPartition(constructDataPartitionReq(sgNameToQueryParamsMap));
@@ -153,8 +159,7 @@ public class ClusterPartitionFetcher implements IPartitionFetcher {
return null;
}
- private TSchemaPartitionReq constructSchemaPartitionReq(PathPatternTree patternTree)
- throws StatementAnalyzeException {
+ private TSchemaPartitionReq constructSchemaPartitionReq(PathPatternTree patternTree) {
PublicBAOS baos = new PublicBAOS();
try {
patternTree.serialize(baos);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/analyze/ClusterSchemaFetcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/analyze/ClusterSchemaFetcher.java
index 5eb4a07991..9e47c755eb 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/analyze/ClusterSchemaFetcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/analyze/ClusterSchemaFetcher.java
@@ -18,24 +18,78 @@
*/
package org.apache.iotdb.db.mpp.sql.analyze;
+import org.apache.iotdb.commons.partition.RegionReplicaSet;
+import org.apache.iotdb.commons.partition.SchemaPartition;
+import org.apache.iotdb.commons.partition.SeriesPartitionSlot;
import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.mpp.common.QueryId;
import org.apache.iotdb.db.mpp.common.schematree.PathPatternTree;
import org.apache.iotdb.db.mpp.common.schematree.SchemaTree;
+import org.apache.iotdb.db.mpp.execution.Coordinator;
+import org.apache.iotdb.db.mpp.sql.statement.metadata.SchemaFetchStatement;
+import org.apache.iotdb.db.query.control.SessionManager;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+import org.apache.iotdb.tsfile.read.common.block.column.Column;
+import org.apache.iotdb.tsfile.utils.Binary;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
public class ClusterSchemaFetcher implements ISchemaFetcher {
+ private final Coordinator coordinator = Coordinator.getInstance();
+ private final IPartitionFetcher partitionFetcher = ClusterPartitionFetcher.getInstance();
+
+ private static final class ClusterSchemaFetcherHolder {
+ private static final ClusterSchemaFetcher INSTANCE = new ClusterSchemaFetcher();
+
+ private ClusterSchemaFetcherHolder() {}
+ }
+
+ public static ClusterSchemaFetcher getInstance() {
+ return ClusterSchemaFetcherHolder.INSTANCE;
+ }
+
+ private ClusterSchemaFetcher() {}
+
@Override
public SchemaTree fetchSchema(PathPatternTree patternTree) {
- return null;
+ SchemaPartition schemaPartition = partitionFetcher.getSchemaPartition(patternTree);
+ Map<String, Map<SeriesPartitionSlot, RegionReplicaSet>> schemaPartitionMap =
+ schemaPartition.getSchemaPartitionMap();
+ List<String> storageGroups = new ArrayList<>(schemaPartitionMap.keySet());
+
+ SchemaFetchStatement schemaFetchStatement = new SchemaFetchStatement(patternTree);
+ schemaFetchStatement.setSchemaPartition(schemaPartition);
+
+ QueryId queryId =
+ new QueryId(String.valueOf(SessionManager.getInstance().requestQueryId(false)));
+ coordinator.execute(schemaFetchStatement, queryId, QueryType.READ, null, "");
+ TsBlock tsBlock = coordinator.getResultSet(queryId);
+ SchemaTree result = new SchemaTree();
+ result.setStorageGroups(storageGroups);
+
+ Binary binary;
+ SchemaTree fetchedSchemaTree;
+ Column column = tsBlock.getColumn(0);
+ for (int i = 0; i < column.getPositionCount(); i++) {
+ binary = column.getBinary(i);
+ fetchedSchemaTree = SchemaTree.deserialize(ByteBuffer.wrap(binary.getValues()));
+ result.mergeSchemaTree(fetchedSchemaTree);
+ }
+ return result;
}
@Override
public SchemaTree fetchSchemaWithAutoCreate(
PartialPath devicePath, String[] measurements, TSDataType[] tsDataTypes, boolean aligned) {
- return null;
+ // todo implement auto create schema
+ return fetchSchema(new PathPatternTree(devicePath, measurements));
}
@Override
@@ -44,6 +98,11 @@ public class ClusterSchemaFetcher implements ISchemaFetcher {
List<String[]> measurements,
List<TSDataType[]> tsDataTypes,
List<Boolean> aligned) {
- return null;
+ Map<PartialPath, List<String>> deviceToMeasurementMap = new HashMap<>();
+ for (int i = 0; i < devicePath.size(); i++) {
+ deviceToMeasurementMap.put(devicePath.get(i), Arrays.asList(measurements.get(i)));
+ }
+ // todo implement auto create schema
+ return fetchSchema(new PathPatternTree(deviceToMeasurementMap));
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/analyze/IPartitionFetcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/analyze/IPartitionFetcher.java
index 51a051ef85..aa00f68f0b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/analyze/IPartitionFetcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/analyze/IPartitionFetcher.java
@@ -21,7 +21,6 @@ package org.apache.iotdb.db.mpp.sql.analyze;
import org.apache.iotdb.commons.partition.DataPartition;
import org.apache.iotdb.commons.partition.DataPartitionQueryParam;
import org.apache.iotdb.commons.partition.SchemaPartition;
-import org.apache.iotdb.db.exception.sql.StatementAnalyzeException;
import org.apache.iotdb.db.mpp.common.schematree.PathPatternTree;
import java.util.List;
@@ -29,15 +28,12 @@ import java.util.Map;
public interface IPartitionFetcher {
- SchemaPartition getSchemaPartition(PathPatternTree patternTree) throws StatementAnalyzeException;
+ SchemaPartition getSchemaPartition(PathPatternTree patternTree);
- SchemaPartition getOrCreateSchemaPartition(PathPatternTree patternTree)
- throws StatementAnalyzeException;
+ SchemaPartition getOrCreateSchemaPartition(PathPatternTree patternTree);
- DataPartition getDataPartition(Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap)
- throws StatementAnalyzeException;
+ DataPartition getDataPartition(Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap);
DataPartition getOrCreateDataPartition(
- Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap)
- throws StatementAnalyzeException;
+ Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/DistributionPlanner.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/DistributionPlanner.java
index e0d42e1bae..17961e5420 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/DistributionPlanner.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/DistributionPlanner.java
@@ -36,6 +36,7 @@ import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.SimplePlanNodeRewriter;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.WritePlanNode;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.SchemaFetchNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.SchemaMergeNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.SchemaScanNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.process.ExchangeNode;
@@ -144,7 +145,7 @@ public class DistributionPlanner {
}
@Override
- public PlanNode visitMetaMerge(SchemaMergeNode node, DistributionPlanContext context) {
+ public PlanNode visitSchemaMerge(SchemaMergeNode node, DistributionPlanContext context) {
SchemaMergeNode root = (SchemaMergeNode) node.clone();
SchemaScanNode seed = (SchemaScanNode) node.getChildren().get(0);
TreeSet<RegionReplicaSet> schemaRegions =
@@ -161,13 +162,13 @@ public class DistributionPlanner {
int count = schemaRegions.size();
schemaRegions.forEach(
region -> {
- SchemaScanNode metaScanNode = (SchemaScanNode) seed.clone();
- metaScanNode.setRegionReplicaSet(region);
+ SchemaScanNode schemaScanNode = (SchemaScanNode) seed.clone();
+ schemaScanNode.setRegionReplicaSet(region);
if (count > 1) {
- metaScanNode.setLimit(metaScanNode.getOffset() + metaScanNode.getLimit());
- metaScanNode.setOffset(0);
+ schemaScanNode.setLimit(schemaScanNode.getOffset() + schemaScanNode.getLimit());
+ schemaScanNode.setOffset(0);
}
- root.addChild(metaScanNode);
+ root.addChild(schemaScanNode);
});
return root;
}
@@ -271,7 +272,7 @@ public class DistributionPlanner {
}
@Override
- public PlanNode visitMetaMerge(SchemaMergeNode node, NodeGroupContext context) {
+ public PlanNode visitSchemaMerge(SchemaMergeNode node, NodeGroupContext context) {
node.getChildren()
.forEach(
child -> {
@@ -299,13 +300,18 @@ public class DistributionPlanner {
}
@Override
- public PlanNode visitMetaScan(SchemaScanNode node, NodeGroupContext context) {
+ public PlanNode visitSchemaScan(SchemaScanNode node, NodeGroupContext context) {
NodeDistribution nodeDistribution = new NodeDistribution(NodeDistributionType.NO_CHILD);
nodeDistribution.region = node.getRegionReplicaSet();
context.putNodeDistribution(node.getPlanNodeId(), nodeDistribution);
return node;
}
+ @Override
+ public PlanNode visitSchemaFetch(SchemaFetchNode node, NodeGroupContext context) {
+ return visitSchemaScan(node, context);
+ }
+
@Override
public PlanNode visitSeriesScan(SeriesScanNode node, NodeGroupContext context) {
context.putNodeDistribution(
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/LocalExecutionPlanner.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/LocalExecutionPlanner.java
index 74d796f8fc..77508888a2 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/LocalExecutionPlanner.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/LocalExecutionPlanner.java
@@ -37,6 +37,7 @@ import org.apache.iotdb.db.mpp.operator.OperatorContext;
import org.apache.iotdb.db.mpp.operator.process.LimitOperator;
import org.apache.iotdb.db.mpp.operator.process.TimeJoinOperator;
import org.apache.iotdb.db.mpp.operator.schema.DevicesSchemaScanOperator;
+import org.apache.iotdb.db.mpp.operator.schema.SchemaFetchOperator;
import org.apache.iotdb.db.mpp.operator.schema.SchemaMergeOperator;
import org.apache.iotdb.db.mpp.operator.schema.TimeSeriesSchemaScanOperator;
import org.apache.iotdb.db.mpp.operator.source.DataSourceOperator;
@@ -45,6 +46,7 @@ import org.apache.iotdb.db.mpp.operator.source.SeriesScanOperator;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.DevicesSchemaScanNode;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.SchemaFetchNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.SchemaMergeNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.TimeSeriesSchemaScanNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.process.AggregateNode;
@@ -157,7 +159,7 @@ public class LocalExecutionPlanner {
}
@Override
- public Operator visitTimeSeriesMetaScan(
+ public Operator visitTimeSeriesSchemaScan(
TimeSeriesSchemaScanNode node, LocalExecutionPlanContext context) {
OperatorContext operatorContext =
context.instanceContext.addOperatorContext(
@@ -178,7 +180,7 @@ public class LocalExecutionPlanner {
}
@Override
- public Operator visitDevicesMetaScan(
+ public Operator visitDevicesSchemaScan(
DevicesSchemaScanNode node, LocalExecutionPlanContext context) {
OperatorContext operatorContext =
context.instanceContext.addOperatorContext(
@@ -196,7 +198,7 @@ public class LocalExecutionPlanner {
}
@Override
- public Operator visitMetaMerge(SchemaMergeNode node, LocalExecutionPlanContext context) {
+ public Operator visitSchemaMerge(SchemaMergeNode node, LocalExecutionPlanContext context) {
List<Operator> children =
node.getChildren().stream()
.map(n -> n.accept(this, context))
@@ -332,6 +334,20 @@ public class LocalExecutionPlanner {
throw new RuntimeException("Error happened while creating sink handle", e);
}
}
+
+ @Override
+ public Operator visitSchemaFetch(SchemaFetchNode node, LocalExecutionPlanContext context) {
+ OperatorContext operatorContext =
+ context.instanceContext.addOperatorContext(
+ context.getNextOperatorId(),
+ node.getPlanNodeId(),
+ SchemaFetchOperator.class.getSimpleName());
+ return new SchemaFetchOperator(
+ node.getPlanNodeId(),
+ operatorContext,
+ node.getPatternTree(),
+ ((SchemaDriverContext) (context.instanceContext.getDriverContext())).getSchemaRegion());
+ }
}
private static class InstanceHolder {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/LogicalPlanner.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/LogicalPlanner.java
index 186ab2a845..cbaa275805 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/LogicalPlanner.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/LogicalPlanner.java
@@ -52,6 +52,7 @@ import org.apache.iotdb.db.mpp.sql.statement.crud.UDTFQueryStatement;
import org.apache.iotdb.db.mpp.sql.statement.metadata.AlterTimeSeriesStatement;
import org.apache.iotdb.db.mpp.sql.statement.metadata.CreateAlignedTimeSeriesStatement;
import org.apache.iotdb.db.mpp.sql.statement.metadata.CreateTimeSeriesStatement;
+import org.apache.iotdb.db.mpp.sql.statement.metadata.SchemaFetchStatement;
import org.apache.iotdb.db.mpp.sql.statement.metadata.ShowDevicesStatement;
import org.apache.iotdb.db.mpp.sql.statement.metadata.ShowTimeSeriesStatement;
import org.apache.iotdb.db.mpp.sql.statement.sys.AuthorStatement;
@@ -288,7 +289,7 @@ public class LogicalPlanner {
showTimeSeriesStatement.isOrderByHeat(),
showTimeSeriesStatement.isContains(),
showTimeSeriesStatement.isPrefixPath());
- planBuilder.planMetaMerge(showTimeSeriesStatement.isOrderByHeat());
+ planBuilder.planSchemaMerge(showTimeSeriesStatement.isOrderByHeat());
if (showTimeSeriesStatement.getLimit() > 0) {
planBuilder.planOffset(showTimeSeriesStatement.getOffset());
planBuilder.planLimit(showTimeSeriesStatement.getLimit());
@@ -300,13 +301,13 @@ public class LogicalPlanner {
public PlanNode visitShowDevices(
ShowDevicesStatement showDevicesStatement, MPPQueryContext context) {
QueryPlanBuilder planBuilder = new QueryPlanBuilder(context);
- planBuilder.planDeviceMetaSource(
+ planBuilder.planDeviceSchemaSource(
showDevicesStatement.getPathPattern(),
showDevicesStatement.getLimit(),
showDevicesStatement.getOffset(),
showDevicesStatement.isPrefixPath(),
showDevicesStatement.hasSgCol());
- planBuilder.planMetaMerge(false);
+ planBuilder.planSchemaMerge(false);
planBuilder.planOffset(showDevicesStatement.getOffset());
planBuilder.planLimit(showDevicesStatement.getLimit());
return planBuilder.getRoot();
@@ -523,5 +524,14 @@ public class LogicalPlanner {
}
return insertRowsNode;
}
+
+ @Override
+ public PlanNode visitSchemaFetch(
+ SchemaFetchStatement schemaFetchStatement, MPPQueryContext context) {
+ QueryPlanBuilder planBuilder = new QueryPlanBuilder(context);
+ planBuilder.planSchemaFetchSource(schemaFetchStatement.getPatternTree());
+ planBuilder.planSchemaMerge(false);
+ return planBuilder.getRoot();
+ }
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/QueryPlanBuilder.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/QueryPlanBuilder.java
index 2675206afb..fb1a5ef0fc 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/QueryPlanBuilder.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/QueryPlanBuilder.java
@@ -21,8 +21,10 @@ package org.apache.iotdb.db.mpp.sql.planner;
import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.db.mpp.common.MPPQueryContext;
+import org.apache.iotdb.db.mpp.common.schematree.PathPatternTree;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.DevicesSchemaScanNode;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.SchemaFetchNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.SchemaMergeNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.TimeSeriesSchemaScanNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.process.*;
@@ -302,18 +304,22 @@ public class QueryPlanBuilder {
this.root = timeSeriesMetaScanNode;
}
- public void planDeviceMetaSource(
+ public void planDeviceSchemaSource(
PartialPath pathPattern, int limit, int offset, boolean prefixPath, boolean hasSgCol) {
- DevicesSchemaScanNode devicesMetaScanNode =
+ DevicesSchemaScanNode devicesSchemaScanNode =
new DevicesSchemaScanNode(
context.getQueryId().genPlanNodeId(), pathPattern, limit, offset, prefixPath, hasSgCol);
- this.root = devicesMetaScanNode;
+ this.root = devicesSchemaScanNode;
}
- public void planMetaMerge(boolean orderByHeat) {
- SchemaMergeNode metaMergeNode =
+ public void planSchemaMerge(boolean orderByHeat) {
+ SchemaMergeNode schemaMergeNode =
new SchemaMergeNode(context.getQueryId().genPlanNodeId(), orderByHeat);
- metaMergeNode.addChild(this.getRoot());
- this.root = metaMergeNode;
+ schemaMergeNode.addChild(this.getRoot());
+ this.root = schemaMergeNode;
+ }
+
+ public void planSchemaFetchSource(PathPatternTree patternTree) {
+ this.root = new SchemaFetchNode(context.getQueryId().genPlanNodeId(), patternTree);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/PlanNodeType.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/PlanNodeType.java
index e5cd7ff8be..4e2400a836 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/PlanNodeType.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/PlanNodeType.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.mpp.sql.planner.plan.node;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.DevicesSchemaScanNode;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.SchemaFetchNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.ShowDevicesNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.TimeSeriesSchemaScanNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.write.AlterTimeSeriesNode;
@@ -74,7 +75,8 @@ public enum PlanNodeType {
CREATE_ALIGNED_TIME_SERIES((short) 23),
TIME_SERIES_SCHEMA_SCAN((short) 24),
// TODO @xinzhongtianxia remove this
- SHOW_DEVICES((short) 25);
+ SHOW_DEVICES((short) 25),
+ SCHEMA_FETCH((short) 26);
private final short nodeType;
@@ -141,6 +143,8 @@ public enum PlanNodeType {
return TimeSeriesSchemaScanNode.deserialize(buffer);
case 25:
return ShowDevicesNode.deserialize(buffer);
+ case 26:
+ return SchemaFetchNode.deserialize(buffer);
default:
throw new IllegalArgumentException("Invalid node type: " + nodeType);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/PlanVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/PlanVisitor.java
index e5a5dfe8c3..4b6a4f6758 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/PlanVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/PlanVisitor.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.mpp.sql.planner.plan.node;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.DevicesSchemaScanNode;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.SchemaFetchNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.SchemaMergeNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.SchemaScanNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read.TimeSeriesSchemaScanNode;
@@ -99,11 +100,11 @@ public abstract class PlanVisitor<R, C> {
return visitPlan(node, context);
}
- public R visitMetaMerge(SchemaMergeNode node, C context) {
+ public R visitSchemaMerge(SchemaMergeNode node, C context) {
return visitPlan(node, context);
};
- public R visitMetaScan(SchemaScanNode node, C context) {
+ public R visitSchemaScan(SchemaScanNode node, C context) {
return visitPlan(node, context);
}
@@ -111,11 +112,11 @@ public abstract class PlanVisitor<R, C> {
return visitPlan(node, context);
}
- public R visitTimeSeriesMetaScan(TimeSeriesSchemaScanNode node, C context) {
+ public R visitTimeSeriesSchemaScan(TimeSeriesSchemaScanNode node, C context) {
return visitPlan(node, context);
}
- public R visitDevicesMetaScan(DevicesSchemaScanNode node, C context) {
+ public R visitDevicesSchemaScan(DevicesSchemaScanNode node, C context) {
return visitPlan(node, context);
}
@@ -126,4 +127,8 @@ public abstract class PlanVisitor<R, C> {
public R visitCreateTimeSeries(CreateTimeSeriesNode node, C context) {
return visitPlan(node, context);
}
+
+ public R visitSchemaFetch(SchemaFetchNode node, C context) {
+ return visitPlan(node, context);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/metedata/read/SchemaMergeNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/metedata/read/SchemaFetchNode.java
similarity index 53%
copy from server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/metedata/read/SchemaMergeNode.java
copy to server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/metedata/read/SchemaFetchNode.java
index c315da6a7b..87888c9e1f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/metedata/read/SchemaMergeNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/metedata/read/SchemaFetchNode.java
@@ -16,61 +16,75 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.db.mpp.sql.planner.plan.node.metedata.read;
+import org.apache.iotdb.db.mpp.common.schematree.PathPatternTree;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanVisitor;
-import org.apache.iotdb.db.mpp.sql.planner.plan.node.process.ProcessNode;
import java.nio.ByteBuffer;
-import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
-public class SchemaMergeNode extends ProcessNode {
-
- private boolean orderByHeat;
+public class SchemaFetchNode extends SchemaScanNode {
- private List<PlanNode> children;
+ private final PathPatternTree patternTree;
- public SchemaMergeNode(PlanNodeId id) {
+ public SchemaFetchNode(PlanNodeId id, PathPatternTree patternTree) {
super(id);
- children = new ArrayList<>();
+ this.patternTree = patternTree;
}
- public SchemaMergeNode(PlanNodeId id, boolean orderByHeat) {
- this(id);
- this.orderByHeat = orderByHeat;
+ public PathPatternTree getPatternTree() {
+ return patternTree;
}
@Override
public List<PlanNode> getChildren() {
- return children;
+ return null;
}
@Override
- public void addChild(PlanNode child) {
- this.children.add(child);
- }
+ public void addChild(PlanNode child) {}
@Override
public PlanNode clone() {
- return new SchemaMergeNode(getPlanNodeId(), this.orderByHeat);
+ return new SchemaFetchNode(getPlanNodeId(), patternTree);
}
@Override
public int allowedChildCount() {
- return CHILD_COUNT_NO_LIMIT;
+ return 0;
}
@Override
- public void serialize(ByteBuffer byteBuffer) {}
+ protected void serializeAttributes(ByteBuffer byteBuffer) {
+ PlanNodeType.SCHEMA_FETCH.serialize(byteBuffer);
+ patternTree.serialize(byteBuffer);
+ }
+
+ public static SchemaFetchNode deserialize(ByteBuffer byteBuffer) {
+ PathPatternTree patternTree = PathPatternTree.deserialize(byteBuffer);
+ PlanNodeId id = PlanNodeId.deserialize(byteBuffer);
+ return new SchemaFetchNode(id, patternTree);
+ }
+
+ @Override
+ public void open() throws Exception {}
+
+ @Override
+ public List<String> getOutputColumnNames() {
+ return Collections.singletonList("SchemaTree");
+ }
@Override
- protected void serializeAttributes(ByteBuffer byteBuffer) {}
+ public void close() throws Exception {}
@Override
public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
- return visitor.visitMetaMerge(this, context);
+ return visitor.visitSchemaFetch(this, context);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/metedata/read/SchemaMergeNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/metedata/read/SchemaMergeNode.java
index c315da6a7b..7ae31a523e 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/metedata/read/SchemaMergeNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/metedata/read/SchemaMergeNode.java
@@ -71,6 +71,6 @@ public class SchemaMergeNode extends ProcessNode {
@Override
public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
- return visitor.visitMetaMerge(this, context);
+ return visitor.visitSchemaMerge(this, context);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/metedata/read/SchemaScanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/metedata/read/SchemaScanNode.java
index 04c2bdea11..72d32da4e4 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/metedata/read/SchemaScanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/metedata/read/SchemaScanNode.java
@@ -36,6 +36,15 @@ public abstract class SchemaScanNode extends SourceNode {
private RegionReplicaSet schemaRegionReplicaSet;
+ protected SchemaScanNode(PlanNodeId id) {
+ super(id);
+ limit = 0;
+ offset = 0;
+ path = null;
+ hasLimit = false;
+ isPrefixPath = false;
+ }
+
protected SchemaScanNode(
PlanNodeId id, PartialPath partialPath, int limit, int offset, boolean isPrefixPath) {
super(id);
@@ -114,6 +123,6 @@ public abstract class SchemaScanNode extends SourceNode {
@Override
public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
- return visitor.visitMetaScan(this, context);
+ return visitor.visitSchemaScan(this, context);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/statement/StatementVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/statement/StatementVisitor.java
index 5224ae4bde..42e88eaa23 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/statement/StatementVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/statement/StatementVisitor.java
@@ -36,6 +36,7 @@ import org.apache.iotdb.db.mpp.sql.statement.crud.UDTFQueryStatement;
import org.apache.iotdb.db.mpp.sql.statement.metadata.AlterTimeSeriesStatement;
import org.apache.iotdb.db.mpp.sql.statement.metadata.CreateAlignedTimeSeriesStatement;
import org.apache.iotdb.db.mpp.sql.statement.metadata.CreateTimeSeriesStatement;
+import org.apache.iotdb.db.mpp.sql.statement.metadata.SchemaFetchStatement;
import org.apache.iotdb.db.mpp.sql.statement.metadata.ShowDevicesStatement;
import org.apache.iotdb.db.mpp.sql.statement.metadata.ShowStorageGroupStatement;
import org.apache.iotdb.db.mpp.sql.statement.metadata.ShowTimeSeriesStatement;
@@ -252,4 +253,8 @@ public abstract class StatementVisitor<R, C> {
InsertRowsOfOneDeviceStatement insertRowsOfOneDeviceStatement, C context) {
return visitStatement(insertRowsOfOneDeviceStatement, context);
}
+
+ public R visitSchemaFetch(SchemaFetchStatement schemaFetchStatement, C context) {
+ return visitStatement(schemaFetchStatement, context);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/statement/metadata/SchemaFetchStatement.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/statement/metadata/SchemaFetchStatement.java
new file mode 100644
index 0000000000..dbe7ca53f3
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/statement/metadata/SchemaFetchStatement.java
@@ -0,0 +1,56 @@
+/*
+ * 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.mpp.sql.statement.metadata;
+
+import org.apache.iotdb.commons.partition.SchemaPartition;
+import org.apache.iotdb.db.mpp.common.schematree.PathPatternTree;
+import org.apache.iotdb.db.mpp.sql.constant.StatementType;
+import org.apache.iotdb.db.mpp.sql.statement.Statement;
+import org.apache.iotdb.db.mpp.sql.statement.StatementVisitor;
+
+public class SchemaFetchStatement extends Statement {
+
+ private PathPatternTree patternTree;
+
+ private SchemaPartition schemaPartition;
+
+ public SchemaFetchStatement(PathPatternTree patternTree) {
+ super();
+ this.patternTree = patternTree;
+ setType(StatementType.FETCH_SCHEMA);
+ }
+
+ public PathPatternTree getPatternTree() {
+ return patternTree;
+ }
+
+ public SchemaPartition getSchemaPartition() {
+ return schemaPartition;
+ }
+
+ public void setSchemaPartition(SchemaPartition schemaPartition) {
+ this.schemaPartition = schemaPartition;
+ }
+
+ @Override
+ public <R, C> R accept(StatementVisitor<R, C> visitor, C context) {
+ return visitor.visitSchemaFetch(this, context);
+ }
+}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/common/PathPatternTreeTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternTreeTest.java
similarity index 89%
rename from server/src/test/java/org/apache/iotdb/db/mpp/common/PathPatternTreeTest.java
rename to server/src/test/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternTreeTest.java
index 3d2d9e83c2..88f4050df3 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/common/PathPatternTreeTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternTreeTest.java
@@ -17,11 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.common;
+package org.apache.iotdb.db.mpp.common.schematree;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.metadata.path.PartialPath;
-import org.apache.iotdb.db.mpp.common.schematree.PathPatternTree;
import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.junit.Assert;
@@ -137,7 +136,7 @@ public class PathPatternTreeTest {
new PartialPath("root.sg1.d3.t1")));
}
- public void checkPathPatternTree(
+ private void checkPathPatternTree(
List<PartialPath> paths,
List<PartialPath> compressedPaths,
List<PartialPath> compressedDevicePaths)
@@ -171,4 +170,24 @@ public class PathPatternTreeTest {
PathPatternTree tmpPathPatternTree = PathPatternTree.deserialize(buffer);
Assert.assertTrue(resultPatternTree.equalWith(tmpPathPatternTree));
}
+
+ @Test
+ public void testPathPatternTreeSplit() throws Exception {
+ List<PartialPath> partialPathList =
+ Arrays.asList(
+ new PartialPath("root.sg1.d1.t1.s1"),
+ new PartialPath("root.sg1.d1.t2.s2"),
+ new PartialPath("root.sg1.*.t1.s1"),
+ new PartialPath("root.sg1.d2.t1.s1"));
+
+ PathPatternTree patternTree = new PathPatternTree();
+ for (PartialPath path : partialPathList) {
+ patternTree.appendPath(path);
+ }
+ patternTree.constructTree();
+
+ Assert.assertEquals(
+ Arrays.asList(new PartialPath("root.sg1.*.t1.s1"), new PartialPath("root.sg1.d1.t2.s2")),
+ patternTree.splitToPathList());
+ }
}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/common/SchemaTreeTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/common/schematree/SchemaTreeTest.java
similarity index 75%
rename from server/src/test/java/org/apache/iotdb/db/mpp/common/SchemaTreeTest.java
rename to server/src/test/java/org/apache/iotdb/db/mpp/common/schematree/SchemaTreeTest.java
index 3da83e3588..3e7f9cdd92 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/common/SchemaTreeTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/common/schematree/SchemaTreeTest.java
@@ -16,17 +16,10 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.mpp.common;
+package org.apache.iotdb.db.mpp.common.schematree;
import org.apache.iotdb.db.metadata.path.MeasurementPath;
import org.apache.iotdb.db.metadata.path.PartialPath;
-import org.apache.iotdb.db.mpp.common.schematree.DeviceSchemaInfo;
-import org.apache.iotdb.db.mpp.common.schematree.SchemaEntityNode;
-import org.apache.iotdb.db.mpp.common.schematree.SchemaInternalNode;
-import org.apache.iotdb.db.mpp.common.schematree.SchemaMeasurementNode;
-import org.apache.iotdb.db.mpp.common.schematree.SchemaNode;
-import org.apache.iotdb.db.mpp.common.schematree.SchemaTree;
-import org.apache.iotdb.db.mpp.common.schematree.SchemaTreeVisitor;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
@@ -46,6 +39,11 @@ public class SchemaTreeTest {
SchemaNode root = generateSchemaTree();
+ testSchemaTree(root);
+ }
+
+ private void testSchemaTree(SchemaNode root) throws Exception {
+
SchemaTreeVisitor visitor =
new SchemaTreeVisitor(root, new PartialPath("root.sg.d2.a.s1"), 0, 0, false);
checkVisitorResult(visitor, 1, new String[] {"root.sg.d2.a.s1"}, null, new boolean[] {true});
@@ -270,4 +268,83 @@ public class SchemaTreeTest {
testSearchDeviceInfo(schemaTree);
}
+
+ @Test
+ public void testAppendMeasurementPath() throws Exception {
+ SchemaTree schemaTree = new SchemaTree();
+ List<MeasurementPath> measurementPathList = generateMeasurementPathList();
+
+ schemaTree.appendMeasurementPaths(measurementPathList);
+
+ testSchemaTree(schemaTree.getRoot());
+ }
+
+ /**
+ * Generate the following tree: root.sg.d1.s1, root.sg.d1.s2(status) root.sg.d2.s1,
+ * root.sg.d2.s2(status) root.sg.d2.a.s1, root.sg.d2.a.s2(status)
+ */
+ private List<MeasurementPath> generateMeasurementPathList() throws Exception {
+ List<MeasurementPath> measurementPathList = new ArrayList<>();
+
+ MeasurementSchema schema1 = new MeasurementSchema("s1", TSDataType.INT32);
+ MeasurementSchema schema2 = new MeasurementSchema("s2", TSDataType.INT64);
+
+ MeasurementPath measurementPath = new MeasurementPath("root.sg.d1.s1");
+ measurementPath.setMeasurementSchema(schema1);
+ measurementPathList.add(measurementPath);
+
+ measurementPath = new MeasurementPath("root.sg.d1.s2");
+ measurementPath.setMeasurementSchema(schema2);
+ measurementPath.setMeasurementAlias("status");
+ measurementPathList.add(measurementPath);
+
+ measurementPath = new MeasurementPath("root.sg.d2.a.s1");
+ measurementPath.setMeasurementSchema(schema1);
+ measurementPath.setUnderAlignedEntity(true);
+ measurementPathList.add(measurementPath);
+
+ measurementPath = new MeasurementPath("root.sg.d2.a.s2");
+ measurementPath.setMeasurementSchema(schema2);
+ measurementPath.setMeasurementAlias("status");
+ measurementPath.setUnderAlignedEntity(true);
+ measurementPathList.add(measurementPath);
+
+ measurementPath = new MeasurementPath("root.sg.d2.s1");
+ measurementPath.setMeasurementSchema(schema1);
+ measurementPathList.add(measurementPath);
+
+ measurementPath = new MeasurementPath("root.sg.d2.s2");
+ measurementPath.setMeasurementSchema(schema2);
+ measurementPath.setMeasurementAlias("status");
+ measurementPathList.add(measurementPath);
+
+ return measurementPathList;
+ }
+
+ @Test
+ public void testMergeSchemaTree() throws Exception {
+ SchemaTree schemaTree = new SchemaTree();
+ for (SchemaTree tree : generateSchemaTrees()) {
+ schemaTree.mergeSchemaTree(tree);
+ }
+
+ testSchemaTree(schemaTree.getRoot());
+ }
+
+ private List<SchemaTree> generateSchemaTrees() throws Exception {
+ List<SchemaTree> schemaTreeList = new ArrayList<>();
+ SchemaTree schemaTree;
+ List<MeasurementPath> measurementPathList = generateMeasurementPathList();
+ List<MeasurementPath> list;
+ for (int i = 0; i < 6; i += 2) {
+ list = new ArrayList<>();
+ list.add(measurementPathList.get(i));
+ list.add(measurementPathList.get(i + 1));
+
+ schemaTree = new SchemaTree();
+ schemaTree.appendMeasurementPaths(list);
+ schemaTreeList.add(schemaTree);
+ }
+ return schemaTreeList;
+ }
}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/operator/schema/SchemaFetchOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/operator/schema/SchemaFetchOperatorTest.java
new file mode 100644
index 0000000000..2f5317e458
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/operator/schema/SchemaFetchOperatorTest.java
@@ -0,0 +1,149 @@
+/*
+ * 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.mpp.operator.schema;
+
+import org.apache.iotdb.commons.consensus.SchemaRegionId;
+import org.apache.iotdb.db.metadata.path.MeasurementPath;
+import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.metadata.schemaregion.ISchemaRegion;
+import org.apache.iotdb.db.metadata.schemaregion.SchemaEngine;
+import org.apache.iotdb.db.mpp.common.schematree.DeviceSchemaInfo;
+import org.apache.iotdb.db.mpp.common.schematree.PathPatternTree;
+import org.apache.iotdb.db.mpp.common.schematree.SchemaTree;
+import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class SchemaFetchOperatorTest {
+
+ @Before
+ public void setUp() {
+ EnvironmentUtils.envSetUp();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvironmentUtils.cleanEnv();
+ }
+
+ @Test
+ public void testSchemaFetchResult() throws Exception {
+ ISchemaRegion schemaRegion = prepareSchemaRegion();
+
+ PathPatternTree patternTree = new PathPatternTree();
+ patternTree.appendPath(new PartialPath("root.**.status"));
+ patternTree.appendPath(new PartialPath("root.**.s1"));
+ patternTree.constructTree();
+
+ SchemaFetchOperator schemaFetchOperator =
+ new SchemaFetchOperator(null, null, patternTree, schemaRegion);
+
+ Assert.assertTrue(schemaFetchOperator.hasNext());
+
+ TsBlock tsBlock = schemaFetchOperator.next();
+
+ Assert.assertFalse(schemaFetchOperator.hasNext());
+
+ Binary binary = tsBlock.getColumn(0).getBinary(0);
+ SchemaTree schemaTree = SchemaTree.deserialize(ByteBuffer.wrap(binary.getValues()));
+
+ DeviceSchemaInfo deviceSchemaInfo =
+ schemaTree.searchDeviceSchemaInfo(
+ new PartialPath("root.sg.d2.a"), Arrays.asList("s1", "status"));
+ Assert.assertTrue(deviceSchemaInfo.isAligned());
+ List<MeasurementSchema> measurementSchemaList = deviceSchemaInfo.getMeasurementSchemaList();
+ Assert.assertEquals(2, measurementSchemaList.size());
+ Assert.assertEquals(
+ Arrays.asList("s1", "s2"),
+ measurementSchemaList.stream()
+ .map(MeasurementSchema::getMeasurementId)
+ .sorted()
+ .collect(Collectors.toList()));
+
+ Pair<List<MeasurementPath>, Integer> pair =
+ schemaTree.searchMeasurementPaths(new PartialPath("root.sg.**.status"), 0, 0, false);
+ Assert.assertEquals(3, pair.left.size());
+ Assert.assertEquals(
+ Arrays.asList("root.sg.d1.s2", "root.sg.d2.s2", "root.sg.d2.a.s2"),
+ pair.left.stream().map(MeasurementPath::getFullPath).collect(Collectors.toList()));
+ }
+
+ private ISchemaRegion prepareSchemaRegion() throws Exception {
+ SchemaEngine schemaEngine = SchemaEngine.getInstance();
+ SchemaRegionId schemaRegionId = new SchemaRegionId(0);
+ schemaEngine.createSchemaRegion(new PartialPath("root.sg"), schemaRegionId);
+ ISchemaRegion schemaRegion = schemaEngine.getSchemaRegion(schemaRegionId);
+
+ CreateTimeSeriesPlan createTimeSeriesPlan =
+ new CreateTimeSeriesPlan(
+ new PartialPath("root.sg.d1.s1"),
+ TSDataType.INT32,
+ TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED,
+ null,
+ null,
+ null,
+ null);
+ schemaRegion.createTimeseries(createTimeSeriesPlan, -1);
+
+ createTimeSeriesPlan.setPath(new PartialPath("root.sg.d2.s1"));
+ schemaRegion.createTimeseries(createTimeSeriesPlan, -1);
+
+ createTimeSeriesPlan.setAlias("status");
+ createTimeSeriesPlan.setPath(new PartialPath("root.sg.d1.s2"));
+ schemaRegion.createTimeseries(createTimeSeriesPlan, -1);
+
+ createTimeSeriesPlan.setPath(new PartialPath("root.sg.d2.s2"));
+ schemaRegion.createTimeseries(createTimeSeriesPlan, -1);
+
+ CreateAlignedTimeSeriesPlan createAlignedTimeSeriesPlan =
+ new CreateAlignedTimeSeriesPlan(
+ new PartialPath("root.sg.d2.a"),
+ Arrays.asList("s1", "s2"),
+ Arrays.asList(TSDataType.INT32, TSDataType.INT32),
+ Arrays.asList(TSEncoding.PLAIN, TSEncoding.PLAIN),
+ Arrays.asList(CompressionType.UNCOMPRESSED, CompressionType.UNCOMPRESSED),
+ Arrays.asList(null, "status"),
+ Collections.emptyList(),
+ Collections.emptyList());
+
+ schemaRegion.createAlignedTimeSeries(createAlignedTimeSeriesPlan);
+
+ return schemaRegion;
+ }
+}