You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by zy...@apache.org on 2023/01/13 05:08:22 UTC
[iotdb] branch master updated: [IOTDB-5399] Implement batch auto create schema (#8826)
This is an automated email from the ASF dual-hosted git repository.
zyk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 280ea4e034 [IOTDB-5399] Implement batch auto create schema (#8826)
280ea4e034 is described below
commit 280ea4e0345670acf8a63b9fff44980330cf9201
Author: Marcos_Zyk <38...@users.noreply.github.com>
AuthorDate: Fri Jan 13 13:08:16 2023 +0800
[IOTDB-5399] Implement batch auto create schema (#8826)
---
.../db/metadata/mtree/MTreeBelowSGCachedImpl.java | 7 +-
.../db/metadata/mtree/MTreeBelowSGMemoryImpl.java | 7 +-
.../metadata/visitor/SchemaExecutionVisitor.java | 61 +++++
.../common/schematree/DeviceGroupSchemaTree.java | 94 -------
.../db/mpp/common/schematree/DeviceSchemaInfo.java | 71 ------
.../execution/executor/RegionWriteExecutor.java | 137 ++++++++++
.../iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java | 47 +++-
.../analyze/schema/AutoCreateSchemaExecutor.java | 276 +++++++++++++++++----
.../analyze/schema/ClusterSchemaFetchExecutor.java | 10 +-
.../plan/analyze/schema/ClusterSchemaFetcher.java | 143 ++++++-----
.../db/mpp/plan/planner/LogicalPlanVisitor.java | 32 +++
.../mpp/plan/planner/plan/node/PlanNodeType.java | 10 +-
.../db/mpp/plan/planner/plan/node/PlanVisitor.java | 10 +
.../write/InternalBatchActivateTemplateNode.java | 162 ++++++++++++
.../write/InternalCreateMultiTimeSeriesNode.java | 164 ++++++++++++
.../iotdb/db/mpp/plan/statement/StatementType.java | 3 +
.../db/mpp/plan/statement/StatementVisitor.java | 12 +
.../InternalBatchActivateTemplateStatement.java | 59 +++++
.../InternalCreateMultiTimeSeriesStatement.java | 57 +++++
19 files changed, 1075 insertions(+), 287 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGCachedImpl.java b/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGCachedImpl.java
index 3ef1001231..1d7bc22b42 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGCachedImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGCachedImpl.java
@@ -30,6 +30,7 @@ import org.apache.iotdb.db.exception.metadata.MeasurementAlreadyExistException;
import org.apache.iotdb.db.exception.metadata.MeasurementInBlackListException;
import org.apache.iotdb.db.exception.metadata.PathAlreadyExistException;
import org.apache.iotdb.db.exception.metadata.PathNotExistException;
+import org.apache.iotdb.db.exception.metadata.template.DifferentTemplateException;
import org.apache.iotdb.db.exception.metadata.template.TemplateImcompatibeException;
import org.apache.iotdb.db.exception.metadata.template.TemplateIsInUseException;
import org.apache.iotdb.db.metadata.MetadataConstant;
@@ -846,7 +847,11 @@ public class MTreeBelowSGCachedImpl implements IMTreeBelowSG {
}
if (cur.isUseTemplate()) {
- throw new TemplateIsInUseException(cur.getFullPath());
+ if (template.getId() == cur.getSchemaTemplateId()) {
+ throw new TemplateIsInUseException(cur.getFullPath());
+ } else {
+ throw new DifferentTemplateException(activatePath.getFullPath(), template.getName());
+ }
}
if (cur.isEntity()) {
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGMemoryImpl.java b/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGMemoryImpl.java
index c0970fa144..dd2d4feae4 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGMemoryImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGMemoryImpl.java
@@ -30,6 +30,7 @@ import org.apache.iotdb.db.exception.metadata.MeasurementAlreadyExistException;
import org.apache.iotdb.db.exception.metadata.MeasurementInBlackListException;
import org.apache.iotdb.db.exception.metadata.PathAlreadyExistException;
import org.apache.iotdb.db.exception.metadata.PathNotExistException;
+import org.apache.iotdb.db.exception.metadata.template.DifferentTemplateException;
import org.apache.iotdb.db.exception.metadata.template.TemplateImcompatibeException;
import org.apache.iotdb.db.exception.metadata.template.TemplateIsInUseException;
import org.apache.iotdb.db.metadata.MetadataConstant;
@@ -736,7 +737,11 @@ public class MTreeBelowSGMemoryImpl implements IMTreeBelowSG {
}
if (cur.isUseTemplate()) {
- throw new TemplateIsInUseException(cur.getFullPath());
+ if (template.getId() == cur.getSchemaTemplateId()) {
+ throw new TemplateIsInUseException(cur.getFullPath());
+ } else {
+ throw new DifferentTemplateException(activatePath.getFullPath(), template.getName());
+ }
}
if (cur.isEntity()) {
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/visitor/SchemaExecutionVisitor.java b/server/src/main/java/org/apache/iotdb/db/metadata/visitor/SchemaExecutionVisitor.java
index e59b116826..d69597fca8 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/visitor/SchemaExecutionVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/visitor/SchemaExecutionVisitor.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.exception.metadata.MeasurementAlreadyExistException;
+import org.apache.iotdb.db.exception.metadata.template.TemplateIsInUseException;
import org.apache.iotdb.db.metadata.plan.schemaregion.impl.write.SchemaRegionWritePlanFactory;
import org.apache.iotdb.db.metadata.plan.schemaregion.write.ICreateAlignedTimeSeriesPlan;
import org.apache.iotdb.db.metadata.plan.schemaregion.write.ICreateTimeSeriesPlan;
@@ -41,6 +42,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateMulti
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.DeactivateTemplateNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.DeleteTimeSeriesNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.InternalBatchActivateTemplateNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.InternalCreateMultiTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.InternalCreateTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.MeasurementGroup;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.PreDeactivateTemplateNode;
@@ -51,6 +54,7 @@ import org.apache.iotdb.rpc.TSStatusCode;
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.utils.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -162,6 +166,39 @@ public class SchemaExecutionVisitor extends PlanVisitor<TSStatus, ISchemaRegion>
return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS, "Execute successfully");
}
+ @Override
+ public TSStatus visitInternalCreateMultiTimeSeries(
+ InternalCreateMultiTimeSeriesNode node, ISchemaRegion schemaRegion) {
+ PartialPath devicePath;
+ MeasurementGroup measurementGroup;
+
+ List<TSStatus> alreadyExistingTimeseries = new ArrayList<>();
+ List<TSStatus> failingStatus = new ArrayList<>();
+
+ for (Map.Entry<PartialPath, Pair<Boolean, MeasurementGroup>> deviceEntry :
+ node.getDeviceMap().entrySet()) {
+ devicePath = deviceEntry.getKey();
+ measurementGroup = deviceEntry.getValue().right;
+ if (deviceEntry.getValue().left) {
+ executeInternalCreateAlignedTimeseries(
+ devicePath, measurementGroup, schemaRegion, alreadyExistingTimeseries, failingStatus);
+ } else {
+ executeInternalCreateTimeseries(
+ devicePath, measurementGroup, schemaRegion, alreadyExistingTimeseries, failingStatus);
+ }
+ }
+
+ if (!failingStatus.isEmpty()) {
+ return RpcUtils.getStatus(failingStatus);
+ }
+
+ if (!alreadyExistingTimeseries.isEmpty()) {
+ return RpcUtils.getStatus(alreadyExistingTimeseries);
+ }
+
+ return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS, "Execute successfully");
+ }
+
private void executeInternalCreateTimeseries(
PartialPath devicePath,
MeasurementGroup measurementGroup,
@@ -289,6 +326,30 @@ public class SchemaExecutionVisitor extends PlanVisitor<TSStatus, ISchemaRegion>
}
}
+ @Override
+ public TSStatus visitInternalBatchActivateTemplate(
+ InternalBatchActivateTemplateNode node, ISchemaRegion schemaRegion) {
+ for (Map.Entry<PartialPath, Pair<Integer, Integer>> entry :
+ node.getTemplateActivationMap().entrySet()) {
+ Template template = ClusterTemplateManager.getInstance().getTemplate(entry.getValue().left);
+ try {
+ schemaRegion.activateSchemaTemplate(
+ SchemaRegionWritePlanFactory.getActivateTemplateInClusterPlan(
+ entry.getKey(), entry.getValue().right, entry.getValue().left),
+ template);
+ } catch (TemplateIsInUseException e) {
+ logger.info(
+ String.format(
+ "Schema template has already been activated on path %s, there's no need to activate again.",
+ entry.getKey()));
+ } catch (MetadataException e) {
+ logger.error(e.getMessage(), e);
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
+ }
+ }
+ return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
+ }
+
@Override
public TSStatus visitConstructSchemaBlackList(
ConstructSchemaBlackListNode node, ISchemaRegion schemaRegion) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/DeviceGroupSchemaTree.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/DeviceGroupSchemaTree.java
deleted file mode 100644
index 389c38f4d4..0000000000
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/DeviceGroupSchemaTree.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
- * 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.common.schematree;
-
-import org.apache.iotdb.commons.path.MeasurementPath;
-import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.tsfile.utils.Pair;
-
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-/**
- * This class is specifically for standalone schema validation during data insertion. Since the
- * schema fetch is mainly based on device path, the schema is directly grouped by device rater than
- * organized as a trie.
- */
-public class DeviceGroupSchemaTree implements ISchemaTree {
-
- private final Map<PartialPath, DeviceSchemaInfo> deviceSchemaInfoMap = new HashMap<>();
-
- @Override
- public DeviceSchemaInfo searchDeviceSchemaInfo(
- PartialPath devicePath, List<String> measurements) {
- return deviceSchemaInfoMap.get(devicePath).getSubDeviceSchemaInfo(measurements);
- }
-
- @Override
- public boolean isEmpty() {
- return deviceSchemaInfoMap.isEmpty();
- }
-
- public void addDeviceInfo(DeviceSchemaInfo deviceSchemaInfo) {
- deviceSchemaInfoMap.put(deviceSchemaInfo.getDevicePath(), deviceSchemaInfo);
- }
-
- public void merge(DeviceGroupSchemaTree schemaTree) {
- deviceSchemaInfoMap.putAll(schemaTree.deviceSchemaInfoMap);
- }
-
- @Override
- public Pair<List<MeasurementPath>, Integer> searchMeasurementPaths(
- PartialPath pathPattern, int slimit, int soffset, boolean isPrefixMatch) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public Pair<List<MeasurementPath>, Integer> searchMeasurementPaths(PartialPath pathPattern) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public List<DeviceSchemaInfo> getMatchedDevices(PartialPath pathPattern, boolean isPrefixMatch) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public List<DeviceSchemaInfo> getMatchedDevices(PartialPath pathPattern) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public String getBelongedDatabase(String pathName) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public String getBelongedDatabase(PartialPath path) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public Set<String> getDatabases() {
- throw new UnsupportedOperationException();
- }
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/DeviceSchemaInfo.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/DeviceSchemaInfo.java
index 73b0b63a62..e1d988e55e 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/DeviceSchemaInfo.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/DeviceSchemaInfo.java
@@ -22,13 +22,10 @@ package org.apache.iotdb.db.mpp.common.schematree;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
@@ -38,9 +35,6 @@ public class DeviceSchemaInfo {
private boolean isAligned;
private List<MeasurementSchemaInfo> measurementSchemaInfoList;
- private Map<String, MeasurementSchemaInfo> measurementSchemaInfoMap;
- private Map<String, MeasurementSchemaInfo> aliasMap;
-
private DeviceSchemaInfo() {}
public DeviceSchemaInfo(
@@ -60,50 +54,6 @@ public class DeviceSchemaInfo {
return isAligned;
}
- public DeviceSchemaInfo getSubDeviceSchemaInfo(List<String> measurements) {
- DeviceSchemaInfo result = new DeviceSchemaInfo();
- result.devicePath = devicePath;
- result.isAligned = isAligned;
- List<MeasurementSchemaInfo> desiredMeasurementSchemaInfoList =
- new ArrayList<>(measurements.size());
-
- if (measurementSchemaInfoMap == null) {
- constructMap();
- }
-
- MeasurementSchemaInfo measurementSchemaInfo;
- for (String measurement : measurements) {
- measurementSchemaInfo = measurementSchemaInfoMap.get(measurement);
- if (measurementSchemaInfo == null) {
- measurementSchemaInfo = aliasMap.get(measurement);
- }
-
- if (measurementSchemaInfo == null) {
- desiredMeasurementSchemaInfoList.add(null);
- }
-
- desiredMeasurementSchemaInfoList.add(measurementSchemaInfo);
- }
-
- result.measurementSchemaInfoList = desiredMeasurementSchemaInfoList;
- return result;
- }
-
- private void constructMap() {
- measurementSchemaInfoMap = new HashMap<>();
- aliasMap = new HashMap<>();
- measurementSchemaInfoList.forEach(
- measurementSchemaInfo -> {
- if (measurementSchemaInfo == null) {
- return;
- }
- measurementSchemaInfoMap.put(measurementSchemaInfo.getName(), measurementSchemaInfo);
- if (measurementSchemaInfo.getAlias() != null) {
- aliasMap.put(measurementSchemaInfo.getAlias(), measurementSchemaInfo);
- }
- });
- }
-
public List<MeasurementSchema> getMeasurementSchemaList() {
return measurementSchemaInfoList.stream()
.map(
@@ -149,25 +99,4 @@ public class DeviceSchemaInfo {
}
return measurementPaths;
}
-
- public MeasurementPath getPathByMeasurement(String measurementName) {
- for (MeasurementSchemaInfo measurementSchemaInfo : measurementSchemaInfoList) {
- MeasurementPath measurementPath =
- new MeasurementPath(
- devicePath.concatNode(measurementSchemaInfo.getName()),
- measurementSchemaInfo.getSchema());
- measurementPath.setUnderAlignedEntity(isAligned);
- if (measurementSchemaInfo.getName().equals(measurementName)) {
- return measurementPath;
- } else if (measurementSchemaInfo.getAlias() != null
- && measurementSchemaInfo.getAlias().equals(measurementName)) {
- measurementPath.setMeasurementAlias(measurementSchemaInfo.getAlias());
- return measurementPath;
- }
- }
- throw new SemanticException(
- String.format(
- "ALIGN BY DEVICE: measurement '%s' does not exist in device '%s'",
- measurementName, getDevicePath()));
- }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/executor/RegionWriteExecutor.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/executor/RegionWriteExecutor.java
index 73df88fd94..b237707483 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/executor/RegionWriteExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/executor/RegionWriteExecutor.java
@@ -46,6 +46,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.ActivateTem
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateAlignedTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateMultiTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateTimeSeriesNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.InternalBatchActivateTemplateNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.InternalCreateMultiTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.InternalCreateTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.MeasurementGroup;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
@@ -67,6 +69,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -501,6 +504,107 @@ public class RegionWriteExecutor {
}
}
+ @Override
+ public RegionExecutionResult visitInternalCreateMultiTimeSeries(
+ InternalCreateMultiTimeSeriesNode node, WritePlanNodeExecutionContext context) {
+ ISchemaRegion schemaRegion =
+ SchemaEngine.getInstance().getSchemaRegion((SchemaRegionId) context.getRegionId());
+ if (config.getSchemaRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS)) {
+ context.getRegionWriteValidationRWLock().writeLock().lock();
+ try {
+ List<TSStatus> failingStatus = new ArrayList<>();
+ List<TSStatus> alreadyExistingStatus = new ArrayList<>();
+
+ MeasurementGroup measurementGroup;
+ Map<Integer, MetadataException> failingMeasurementMap;
+ MetadataException metadataException;
+ for (Map.Entry<PartialPath, Pair<Boolean, MeasurementGroup>> deviceEntry :
+ node.getDeviceMap().entrySet()) {
+ measurementGroup = deviceEntry.getValue().right;
+ failingMeasurementMap =
+ schemaRegion.checkMeasurementExistence(
+ deviceEntry.getKey(),
+ measurementGroup.getMeasurements(),
+ measurementGroup.getAliasList());
+ // filter failed measurement and keep the rest for execution
+ for (Map.Entry<Integer, MetadataException> failingMeasurement :
+ failingMeasurementMap.entrySet()) {
+ metadataException = failingMeasurement.getValue();
+ if (metadataException.getErrorCode()
+ == TSStatusCode.TIMESERIES_ALREADY_EXIST.getStatusCode()) {
+ LOGGER.info(
+ "There's no need to internal create timeseries. {}",
+ failingMeasurement.getValue().getMessage());
+ alreadyExistingStatus.add(
+ RpcUtils.getStatus(
+ metadataException.getErrorCode(),
+ MeasurementPath.transformDataToString(
+ ((MeasurementAlreadyExistException) metadataException)
+ .getMeasurementPath())));
+ } else {
+ LOGGER.error("Metadata error: ", metadataException);
+ failingStatus.add(
+ RpcUtils.getStatus(
+ metadataException.getErrorCode(), metadataException.getMessage()));
+ }
+ }
+ measurementGroup.removeMeasurements(failingMeasurementMap.keySet());
+ }
+
+ RegionExecutionResult executionResult =
+ super.visitInternalCreateMultiTimeSeries(node, context);
+
+ if (failingStatus.isEmpty() && alreadyExistingStatus.isEmpty()) {
+ return executionResult;
+ }
+
+ TSStatus executionStatus = executionResult.getStatus();
+
+ // separate the measurement_already_exist exception and other exceptions process,
+ // measurement_already_exist exception is acceptable due to concurrent timeseries creation
+ if (failingStatus.isEmpty()) {
+ if (executionStatus.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()) {
+ if (executionStatus.getSubStatus().get(0).getCode()
+ == TSStatusCode.TIMESERIES_ALREADY_EXIST.getStatusCode()) {
+ // there's only measurement_already_exist exception
+ alreadyExistingStatus.addAll(executionStatus.getSubStatus());
+ } else {
+ failingStatus.addAll(executionStatus.getSubStatus());
+ }
+ } else if (executionStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ failingStatus.add(executionStatus);
+ }
+ } else {
+ if (executionStatus.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()) {
+ if (executionStatus.getSubStatus().get(0).getCode()
+ != TSStatusCode.TIMESERIES_ALREADY_EXIST.getStatusCode()) {
+ failingStatus.addAll(executionStatus.getSubStatus());
+ }
+ } else if (executionStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ failingStatus.add(executionStatus);
+ }
+ }
+
+ TSStatus status;
+ if (failingStatus.isEmpty()) {
+ status = RpcUtils.getStatus(alreadyExistingStatus);
+ } else {
+ status = RpcUtils.getStatus(failingStatus);
+ }
+
+ RegionExecutionResult result = new RegionExecutionResult();
+ result.setAccepted(false);
+ result.setMessage(status.getMessage());
+ result.setStatus(status);
+ return result;
+ } finally {
+ context.getRegionWriteValidationRWLock().writeLock().unlock();
+ }
+ } else {
+ return super.visitInternalCreateMultiTimeSeries(node, context);
+ }
+ }
+
@Override
public RegionExecutionResult visitActivateTemplate(
ActivateTemplateNode node, WritePlanNodeExecutionContext context) {
@@ -527,6 +631,39 @@ public class RegionWriteExecutor {
context.getRegionWriteValidationRWLock().readLock().unlock();
}
}
+
+ @Override
+ public RegionExecutionResult visitInternalBatchActivateTemplate(
+ InternalBatchActivateTemplateNode node, WritePlanNodeExecutionContext context) {
+ // activate template operation shall be blocked by unset template check
+ context.getRegionWriteValidationRWLock().readLock().lock();
+ try {
+ for (Map.Entry<PartialPath, Pair<Integer, Integer>> entry :
+ node.getTemplateActivationMap().entrySet()) {
+ Pair<Template, PartialPath> templateSetInfo =
+ ClusterTemplateManager.getInstance().checkTemplateSetInfo(entry.getKey());
+ if (templateSetInfo == null) {
+ // The activation has already been validated during analyzing.
+ // That means the template is being unset during the activation plan transport.
+ RegionExecutionResult result = new RegionExecutionResult();
+ result.setAccepted(false);
+ String message =
+ String.format(
+ "Template is being unsetting from prefix path of %s. Please try activating later.",
+ new PartialPath(
+ Arrays.copyOf(entry.getKey().getNodes(), entry.getValue().right + 1))
+ .getFullPath());
+ result.setMessage(message);
+ result.setStatus(RpcUtils.getStatus(TSStatusCode.METADATA_ERROR, message));
+ return result;
+ }
+ }
+
+ return super.visitInternalBatchActivateTemplate(node, context);
+ } finally {
+ context.getRegionWriteValidationRWLock().readLock().unlock();
+ }
+ }
}
private static class WritePlanNodeExecutionContext {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java
index 36324cac78..4dd8e2cb23 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/AnalyzeVisitor.java
@@ -87,6 +87,8 @@ import org.apache.iotdb.db.mpp.plan.statement.crud.InsertStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.InsertTabletStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.LoadTsFileStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.QueryStatement;
+import org.apache.iotdb.db.mpp.plan.statement.internal.InternalBatchActivateTemplateStatement;
+import org.apache.iotdb.db.mpp.plan.statement.internal.InternalCreateMultiTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.internal.InternalCreateTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.internal.SchemaFetchStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.AlterTimeSeriesStatement;
@@ -1563,11 +1565,6 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
Analysis analysis = new Analysis();
analysis.setStatement(internalCreateTimeSeriesStatement);
- checkIsTemplateCompatible(
- internalCreateTimeSeriesStatement.getDevicePath(),
- internalCreateTimeSeriesStatement.getMeasurements(),
- null);
-
PathPatternTree pathPatternTree = new PathPatternTree();
for (String measurement : internalCreateTimeSeriesStatement.getMeasurements()) {
pathPatternTree.appendFullPath(
@@ -1580,6 +1577,26 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
return analysis;
}
+ @Override
+ public Analysis visitInternalCreateMultiTimeSeries(
+ InternalCreateMultiTimeSeriesStatement internalCreateMultiTimeSeriesStatement,
+ MPPQueryContext context) {
+ context.setQueryType(QueryType.WRITE);
+
+ Analysis analysis = new Analysis();
+ analysis.setStatement(internalCreateMultiTimeSeriesStatement);
+
+ PathPatternTree pathPatternTree = new PathPatternTree();
+ for (PartialPath devicePath : internalCreateMultiTimeSeriesStatement.getDeviceMap().keySet()) {
+ pathPatternTree.appendFullPath(devicePath.concatNode(ONE_LEVEL_PATH_WILDCARD));
+ }
+
+ SchemaPartition schemaPartitionInfo;
+ schemaPartitionInfo = partitionFetcher.getOrCreateSchemaPartition(pathPatternTree);
+ analysis.setSchemaPartitionInfo(schemaPartitionInfo);
+ return analysis;
+ }
+
@Override
public Analysis visitCreateMultiTimeseries(
CreateMultiTimeSeriesStatement createMultiTimeSeriesStatement, MPPQueryContext context) {
@@ -2450,6 +2467,26 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
return analysis;
}
+ @Override
+ public Analysis visitInternalBatchActivateTemplate(
+ InternalBatchActivateTemplateStatement internalBatchActivateTemplateStatement,
+ MPPQueryContext context) {
+ context.setQueryType(QueryType.WRITE);
+ Analysis analysis = new Analysis();
+ analysis.setStatement(internalBatchActivateTemplateStatement);
+
+ PathPatternTree patternTree = new PathPatternTree();
+ for (PartialPath activatePath :
+ internalBatchActivateTemplateStatement.getDeviceMap().keySet()) {
+ patternTree.appendPathPattern(activatePath.concatNode(ONE_LEVEL_PATH_WILDCARD));
+ }
+ SchemaPartition partition = partitionFetcher.getOrCreateSchemaPartition(patternTree);
+
+ analysis.setSchemaPartitionInfo(partition);
+
+ return analysis;
+ }
+
@Override
public Analysis visitShowPathsUsingTemplate(
ShowPathsUsingTemplateStatement showPathsUsingTemplateStatement, MPPQueryContext context) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/schema/AutoCreateSchemaExecutor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/schema/AutoCreateSchemaExecutor.java
index 638e0206e0..4b4ef0f795 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/schema/AutoCreateSchemaExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/schema/AutoCreateSchemaExecutor.java
@@ -29,7 +29,10 @@ import org.apache.iotdb.db.metadata.template.ITemplateManager;
import org.apache.iotdb.db.metadata.template.Template;
import org.apache.iotdb.db.mpp.common.schematree.ClusterSchemaTree;
import org.apache.iotdb.db.mpp.plan.execution.ExecutionResult;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.MeasurementGroup;
import org.apache.iotdb.db.mpp.plan.statement.Statement;
+import org.apache.iotdb.db.mpp.plan.statement.internal.InternalBatchActivateTemplateStatement;
+import org.apache.iotdb.db.mpp.plan.statement.internal.InternalCreateMultiTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.internal.InternalCreateTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.ActivateTemplateStatement;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -43,6 +46,7 @@ import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -63,36 +67,24 @@ class AutoCreateSchemaExecutor {
this.statementExecutor = statementExecutor;
}
+ // auto create the missing measurements and merge them into given schemaTree
void autoCreateMissingMeasurements(
ClusterSchemaTree schemaTree,
PartialPath devicePath,
List<Integer> indexOfTargetMeasurements,
String[] measurements,
Function<Integer, TSDataType> getDataType,
- TSEncoding[] encodings,
- CompressionType[] compressionTypes,
boolean isAligned) {
// check whether there is template should be activated
Pair<Template, PartialPath> templateInfo = templateManager.checkTemplateSetInfo(devicePath);
if (templateInfo != null) {
Template template = templateInfo.left;
- boolean shouldActivateTemplate = false;
- for (int index : indexOfTargetMeasurements) {
- if (template.hasSchema(measurements[index])) {
- shouldActivateTemplate = true;
- break;
- }
- }
-
- if (shouldActivateTemplate) {
+ List<Integer> indexOfMeasurementsNotInTemplate =
+ checkMeasurementsInSchemaTemplate(
+ devicePath, indexOfTargetMeasurements, measurements, isAligned, template);
+ if (indexOfMeasurementsNotInTemplate.size() < indexOfTargetMeasurements.size()) {
+ // there are measurements in schema template
internalActivateTemplate(devicePath);
- List<Integer> recheckedIndexOfMissingMeasurements = new ArrayList<>();
- for (int i = 0; i < indexOfTargetMeasurements.size(); i++) {
- if (!template.hasSchema(measurements[i])) {
- recheckedIndexOfMissingMeasurements.add(indexOfTargetMeasurements.get(i));
- }
- }
- indexOfTargetMeasurements = recheckedIndexOfMissingMeasurements;
for (Map.Entry<String, IMeasurementSchema> entry : template.getSchemaMap().entrySet()) {
schemaTree.appendSingleMeasurement(
devicePath.concatNode(entry.getKey()),
@@ -101,11 +93,12 @@ class AutoCreateSchemaExecutor {
null,
template.isDirectAligned());
}
-
- if (indexOfTargetMeasurements.isEmpty()) {
- return;
- }
}
+ if (indexOfMeasurementsNotInTemplate.isEmpty()) {
+ return;
+ }
+ // there are measurements need to be created as normal timeseries
+ indexOfTargetMeasurements = indexOfMeasurementsNotInTemplate;
}
// auto create the rest missing timeseries
@@ -124,30 +117,162 @@ class AutoCreateSchemaExecutor {
if (tsDataType != null) {
missingMeasurements.add(measurements[index]);
dataTypesOfMissingMeasurement.add(tsDataType);
- encodingsOfMissingMeasurement.add(
- encodings == null ? getDefaultEncoding(tsDataType) : encodings[index]);
+ encodingsOfMissingMeasurement.add(getDefaultEncoding(tsDataType));
compressionTypesOfMissingMeasurement.add(
- compressionTypes == null
- ? TSFileDescriptor.getInstance().getConfig().getCompressor()
- : compressionTypes[index]);
+ TSFileDescriptor.getInstance().getConfig().getCompressor());
}
});
if (!missingMeasurements.isEmpty()) {
- schemaTree.mergeSchemaTree(
- internalCreateTimeseries(
- devicePath,
- missingMeasurements,
- dataTypesOfMissingMeasurement,
- encodingsOfMissingMeasurement,
- compressionTypesOfMissingMeasurement,
- isAligned));
+ internalCreateTimeSeries(
+ schemaTree,
+ devicePath,
+ missingMeasurements,
+ dataTypesOfMissingMeasurement,
+ encodingsOfMissingMeasurement,
+ compressionTypesOfMissingMeasurement,
+ isAligned);
+ }
+ }
+
+ void autoCreateMissingMeasurements(
+ ClusterSchemaTree schemaTree,
+ List<PartialPath> devicePathList,
+ List<Integer> indexOfTargetDevices,
+ List<List<Integer>> indexOfTargetMeasurementsList,
+ List<String[]> measurementsList,
+ List<TSDataType[]> tsDataTypesList,
+ List<TSEncoding[]> encodingsList,
+ List<CompressionType[]> compressionTypesList,
+ List<Boolean> isAlignedList) {
+ // check whether there is template should be activated
+
+ Map<PartialPath, Pair<Template, PartialPath>> devicesNeedActivateTemplate = new HashMap<>();
+ Map<PartialPath, Pair<Boolean, MeasurementGroup>> devicesNeedAutoCreateTimeSeries =
+ new HashMap<>();
+ int deviceIndex;
+ PartialPath devicePath;
+ List<Integer> indexOfTargetMeasurements;
+ Pair<Template, PartialPath> templateInfo;
+ Template template;
+ List<Integer> indexOfMeasurementsNotInTemplate;
+ for (int i = 0, size = indexOfTargetDevices.size(); i < size; i++) {
+ deviceIndex = indexOfTargetDevices.get(i);
+ devicePath = devicePathList.get(deviceIndex);
+ indexOfTargetMeasurements = indexOfTargetMeasurementsList.get(i);
+
+ templateInfo = devicesNeedActivateTemplate.get(devicePath);
+ if (templateInfo == null) {
+ templateInfo = templateManager.checkTemplateSetInfo(devicePath);
+ }
+
+ if (templateInfo == null) {
+ indexOfMeasurementsNotInTemplate = indexOfTargetMeasurements;
+ } else {
+ template = templateInfo.left;
+ indexOfMeasurementsNotInTemplate =
+ checkMeasurementsInSchemaTemplate(
+ devicePath,
+ indexOfTargetMeasurements,
+ measurementsList.get(deviceIndex),
+ isAlignedList.get(deviceIndex),
+ template);
+ if (indexOfMeasurementsNotInTemplate.size() < indexOfTargetMeasurements.size()) {
+ // there are measurements in schema template
+ devicesNeedActivateTemplate.putIfAbsent(devicePath, templateInfo);
+ }
+ }
+
+ if (!indexOfMeasurementsNotInTemplate.isEmpty()) {
+ // there are measurements need to be created as normal timeseries
+ int finalDeviceIndex = deviceIndex;
+ List<Integer> finalIndexOfMeasurementsNotInTemplate = indexOfMeasurementsNotInTemplate;
+ devicesNeedAutoCreateTimeSeries.compute(
+ devicePath,
+ (k, v) -> {
+ if (v == null) {
+ v = new Pair<>(isAlignedList.get(finalDeviceIndex), new MeasurementGroup());
+ }
+ MeasurementGroup measurementGroup = v.right;
+ String[] measurements = measurementsList.get(finalDeviceIndex);
+ TSDataType[] tsDataTypes = tsDataTypesList.get(finalDeviceIndex);
+ TSEncoding[] encodings =
+ encodingsList == null ? null : encodingsList.get(finalDeviceIndex);
+ CompressionType[] compressionTypes =
+ compressionTypesList == null ? null : compressionTypesList.get(finalDeviceIndex);
+ for (int measurementIndex : finalIndexOfMeasurementsNotInTemplate) {
+ if (tsDataTypes[measurementIndex] == null) {
+ continue;
+ }
+ measurementGroup.addMeasurement(
+ measurements[measurementIndex],
+ tsDataTypes[measurementIndex],
+ encodings == null
+ ? getDefaultEncoding(tsDataTypes[measurementIndex])
+ : encodings[measurementIndex],
+ compressionTypes == null
+ ? TSFileDescriptor.getInstance().getConfig().getCompressor()
+ : compressionTypes[measurementIndex]);
+ }
+ return v;
+ });
+ }
+ }
+
+ if (!devicesNeedActivateTemplate.isEmpty()) {
+ internalActivateTemplate(devicesNeedActivateTemplate);
+ for (Map.Entry<PartialPath, Pair<Template, PartialPath>> entry :
+ devicesNeedActivateTemplate.entrySet()) {
+ devicePath = entry.getKey();
+ template = entry.getValue().left;
+ for (Map.Entry<String, IMeasurementSchema> measurementEntry :
+ template.getSchemaMap().entrySet()) {
+ schemaTree.appendSingleMeasurement(
+ devicePath.concatNode(measurementEntry.getKey()),
+ (MeasurementSchema) measurementEntry.getValue(),
+ null,
+ null,
+ template.isDirectAligned());
+ }
+ }
+ }
+
+ if (!devicesNeedAutoCreateTimeSeries.isEmpty()) {
+ internalCreateTimeSeries(schemaTree, devicesNeedAutoCreateTimeSeries);
}
}
- // try to create the target timeseries and return schemaTree involving successfully created
- // timeseries and existing timeseries
- private ClusterSchemaTree internalCreateTimeseries(
+ private List<Integer> checkMeasurementsInSchemaTemplate(
+ PartialPath devicePath,
+ List<Integer> indexOfTargetMeasurements,
+ String[] measurements,
+ boolean isAligned,
+ Template template) {
+ // check whether there is template should be activated
+ boolean shouldActivateTemplate = false;
+ for (int index : indexOfTargetMeasurements) {
+ if (template.hasSchema(measurements[index])) {
+ shouldActivateTemplate = true;
+ break;
+ }
+ }
+ if (shouldActivateTemplate) {
+ List<Integer> recheckedIndexOfMissingMeasurements = new ArrayList<>();
+ for (int index : indexOfTargetMeasurements) {
+ if (!template.hasSchema(measurements[index])) {
+ recheckedIndexOfMissingMeasurements.add(index);
+ }
+ }
+ return recheckedIndexOfMissingMeasurements;
+ } else {
+ return indexOfTargetMeasurements;
+ }
+ }
+
+ // try to create the target timeseries and merge schema of successfully created
+ // timeseries and existing timeseries into given schemaTree
+ private void internalCreateTimeSeries(
+ ClusterSchemaTree schemaTree,
PartialPath devicePath,
List<String> measurements,
List<TSDataType> tsDataTypes,
@@ -164,7 +289,6 @@ class AutoCreateSchemaExecutor {
.map(o -> measurements.indexOf(o.getMeasurement()))
.collect(Collectors.toSet());
- ClusterSchemaTree schemaTree = new ClusterSchemaTree();
schemaTree.appendMeasurementPaths(measurementPathList);
for (int i = 0, size = measurements.size(); i < size; i++) {
@@ -180,13 +304,10 @@ class AutoCreateSchemaExecutor {
null,
isAligned);
}
-
- return schemaTree;
}
// auto create timeseries and return the existing timeseries info
- private List<MeasurementPath> executeInternalCreateTimeseriesStatement(
- InternalCreateTimeSeriesStatement statement) {
+ private List<MeasurementPath> executeInternalCreateTimeseriesStatement(Statement statement) {
ExecutionResult executionResult = statementExecutor.apply(statement);
@@ -224,7 +345,74 @@ class AutoCreateSchemaExecutor {
TSStatus status = executionResult.status;
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()
&& status.getCode() != TSStatusCode.TEMPLATE_IS_IN_USE.getStatusCode()) {
- throw new RuntimeException(new IoTDBException(status.getMessage(), status.getCode()));
+ throw new SemanticException(new IoTDBException(status.getMessage(), status.getCode()));
+ }
+ }
+
+ private void internalActivateTemplate(
+ Map<PartialPath, Pair<Template, PartialPath>> devicesNeedActivateTemplate) {
+ ExecutionResult executionResult =
+ statementExecutor.apply(
+ new InternalBatchActivateTemplateStatement(devicesNeedActivateTemplate));
+ TSStatus status = executionResult.status;
+ if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
+ || status.getCode() == TSStatusCode.TEMPLATE_IS_IN_USE.getStatusCode()) {
+ return;
+ }
+ if (status.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()) {
+ Set<String> failedActivationSet = new HashSet<>();
+ for (TSStatus subStatus : status.subStatus) {
+ if (subStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()
+ && subStatus.getCode() != TSStatusCode.TEMPLATE_IS_IN_USE.getStatusCode()) {
+ failedActivationSet.add(subStatus.message);
+ }
+ }
+ if (!failedActivationSet.isEmpty()) {
+ throw new SemanticException(new MetadataException(String.join("; ", failedActivationSet)));
+ }
+ } else {
+ throw new SemanticException(new IoTDBException(status.getMessage(), status.getCode()));
+ }
+ }
+
+ private void internalCreateTimeSeries(
+ ClusterSchemaTree schemaTree,
+ Map<PartialPath, Pair<Boolean, MeasurementGroup>> devicesNeedAutoCreateTimeSeries) {
+
+ List<MeasurementPath> measurementPathList =
+ executeInternalCreateTimeseriesStatement(
+ new InternalCreateMultiTimeSeriesStatement(devicesNeedAutoCreateTimeSeries));
+
+ schemaTree.appendMeasurementPaths(measurementPathList);
+
+ Map<PartialPath, Set<String>> alreadyExistingMeasurementMap = new HashMap<>();
+ for (MeasurementPath measurementPath : measurementPathList) {
+ alreadyExistingMeasurementMap
+ .computeIfAbsent(measurementPath.getDevicePath(), k -> new HashSet<>())
+ .add(measurementPath.getMeasurement());
+ }
+ Set<String> measurementSet;
+ MeasurementGroup measurementGroup;
+ for (Map.Entry<PartialPath, Pair<Boolean, MeasurementGroup>> entry :
+ devicesNeedAutoCreateTimeSeries.entrySet()) {
+ measurementSet = alreadyExistingMeasurementMap.get(entry.getKey());
+ measurementGroup = entry.getValue().right;
+ for (int i = 0, size = measurementGroup.size(); i < size; i++) {
+ if (measurementSet != null
+ && measurementSet.contains(measurementGroup.getMeasurements().get(i))) {
+ continue;
+ }
+ schemaTree.appendSingleMeasurement(
+ entry.getKey().concatNode(measurementGroup.getMeasurements().get(i)),
+ new MeasurementSchema(
+ measurementGroup.getMeasurements().get(i),
+ measurementGroup.getDataTypes().get(i),
+ measurementGroup.getEncodings().get(i),
+ measurementGroup.getCompressors().get(i)),
+ null,
+ null,
+ entry.getValue().left);
+ }
}
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/schema/ClusterSchemaFetchExecutor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/schema/ClusterSchemaFetchExecutor.java
index 3f8e609604..9e8576a18c 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/schema/ClusterSchemaFetchExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/schema/ClusterSchemaFetchExecutor.java
@@ -116,11 +116,15 @@ class ClusterSchemaFetchExecutor {
ClusterSchemaTree fetchSchemaOfMultiDevices(
List<PartialPath> devicePathList,
List<String[]> measurementsList,
+ List<Integer> indexOfTargetDevices,
List<List<Integer>> indexOfTargetMeasurementsList) {
PathPatternTree patternTree = new PathPatternTree();
- for (int i = 0; i < devicePathList.size(); i++) {
- for (int index : indexOfTargetMeasurementsList.get(i)) {
- patternTree.appendFullPath(devicePathList.get(i), measurementsList.get(i)[index]);
+ int deviceIndex;
+ for (int i = 0, size = indexOfTargetDevices.size(); i < size; i++) {
+ deviceIndex = indexOfTargetDevices.get(i);
+ for (int measurementIndex : indexOfTargetMeasurementsList.get(i)) {
+ patternTree.appendFullPath(
+ devicePathList.get(deviceIndex), measurementsList.get(deviceIndex)[measurementIndex]);
}
}
patternTree.constructTree();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/schema/ClusterSchemaFetcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/schema/ClusterSchemaFetcher.java
index fb835ff76f..09868e6b9a 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/schema/ClusterSchemaFetcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/schema/ClusterSchemaFetcher.java
@@ -183,15 +183,18 @@ public class ClusterSchemaFetcher implements ISchemaFetcher {
}
// auto create the still missing schema and merge them into schemaTree
- checkAndAutoCreateMissingMeasurements(
- schemaTree,
- devicePath,
- indexOfMissingMeasurements,
- measurements,
- getDataType,
- null,
- null,
- isAligned);
+ indexOfMissingMeasurements =
+ checkMissingMeasurementsAfterSchemaFetch(
+ schemaTree, devicePath, indexOfMissingMeasurements, measurements);
+ if (!indexOfMissingMeasurements.isEmpty()) {
+ autoCreateSchemaExecutor.autoCreateMissingMeasurements(
+ schemaTree,
+ devicePath,
+ indexOfMissingMeasurements,
+ measurements,
+ getDataType,
+ isAligned);
+ }
return schemaTree;
} finally {
@@ -223,26 +226,29 @@ public class ClusterSchemaFetcher implements ISchemaFetcher {
try {
ClusterSchemaTree schemaTree = new ClusterSchemaTree();
List<List<Integer>> indexOfMissingMeasurementsList = new ArrayList<>(devicePathList.size());
- boolean hasMissingMeasurement = false;
+ List<Integer> indexOfDevicesWithMissingMeasurements = new ArrayList<>();
for (int i = 0; i < devicePathList.size(); i++) {
schemaTree.mergeSchemaTree(schemaCache.get(devicePathList.get(i), measurementsList.get(i)));
List<Integer> indexOfMissingMeasurements =
checkMissingMeasurements(schemaTree, devicePathList.get(i), measurementsList.get(i));
- indexOfMissingMeasurementsList.add(indexOfMissingMeasurements);
if (!indexOfMissingMeasurements.isEmpty()) {
- hasMissingMeasurement = true;
+ indexOfDevicesWithMissingMeasurements.add(i);
+ indexOfMissingMeasurementsList.add(indexOfMissingMeasurements);
}
}
// all schema can be taken from cache
- if (!hasMissingMeasurement) {
+ if (indexOfDevicesWithMissingMeasurements.isEmpty()) {
return schemaTree;
}
// try fetch the missing schema from remote and cache fetched schema
ClusterSchemaTree remoteSchemaTree =
clusterSchemaFetchExecutor.fetchSchemaOfMultiDevices(
- devicePathList, measurementsList, indexOfMissingMeasurementsList);
+ devicePathList,
+ measurementsList,
+ indexOfDevicesWithMissingMeasurements,
+ indexOfMissingMeasurementsList);
if (!remoteSchemaTree.isEmpty()) {
schemaTree.mergeSchemaTree(remoteSchemaTree);
}
@@ -252,18 +258,38 @@ public class ClusterSchemaFetcher implements ISchemaFetcher {
}
// auto create the still missing schema and merge them into schemaTree
- for (int i = 0; i < devicePathList.size(); i++) {
- int finalI = i;
- checkAndAutoCreateMissingMeasurements(
+ List<Integer> indexOfDevicesNeedAutoCreateSchema = new ArrayList<>();
+ List<List<Integer>> indexOfMeasurementsNeedAutoCreate = new ArrayList<>();
+ List<Integer> indexOfMissingMeasurements;
+ int deviceIndex;
+ for (int i = 0, size = indexOfDevicesWithMissingMeasurements.size(); i < size; i++) {
+ deviceIndex = indexOfDevicesWithMissingMeasurements.get(i);
+ indexOfMissingMeasurements = indexOfMissingMeasurementsList.get(i);
+ indexOfMissingMeasurements =
+ checkMissingMeasurementsAfterSchemaFetch(
+ schemaTree,
+ devicePathList.get(deviceIndex),
+ indexOfMissingMeasurements,
+ measurementsList.get(deviceIndex));
+ if (!indexOfMissingMeasurements.isEmpty()) {
+ indexOfDevicesNeedAutoCreateSchema.add(deviceIndex);
+ indexOfMeasurementsNeedAutoCreate.add(indexOfMissingMeasurements);
+ }
+ }
+
+ if (!indexOfDevicesNeedAutoCreateSchema.isEmpty()) {
+ autoCreateSchemaExecutor.autoCreateMissingMeasurements(
schemaTree,
- devicePathList.get(i),
- indexOfMissingMeasurementsList.get(i),
- measurementsList.get(i),
- index -> tsDataTypesList.get(finalI)[index],
- encodingsList == null ? null : encodingsList.get(i),
- compressionTypesList == null ? null : compressionTypesList.get(i),
- isAlignedList.get(i));
+ devicePathList,
+ indexOfDevicesNeedAutoCreateSchema,
+ indexOfMeasurementsNeedAutoCreate,
+ measurementsList,
+ tsDataTypesList,
+ encodingsList,
+ compressionTypesList,
+ isAlignedList);
}
+
return schemaTree;
} finally {
schemaCache.releaseReadLock();
@@ -285,49 +311,6 @@ public class ClusterSchemaFetcher implements ISchemaFetcher {
return templateManager.getAllPathsSetTemplate(templateName);
}
- // check which measurements are missing and auto create the missing measurements and merge them
- // into given schemaTree
- private void checkAndAutoCreateMissingMeasurements(
- ClusterSchemaTree schemaTree,
- PartialPath devicePath,
- List<Integer> indexOfMissingMeasurements,
- String[] measurements,
- Function<Integer, TSDataType> getDataType,
- TSEncoding[] encodings,
- CompressionType[] compressionTypes,
- boolean isAligned) {
- // check missing measurements
- DeviceSchemaInfo deviceSchemaInfo =
- schemaTree.searchDeviceSchemaInfo(
- devicePath,
- indexOfMissingMeasurements.stream()
- .map(index -> measurements[index])
- .collect(Collectors.toList()));
- if (deviceSchemaInfo != null) {
- List<MeasurementSchema> schemaList = deviceSchemaInfo.getMeasurementSchemaList();
- int removedCount = 0;
- for (int i = 0, size = schemaList.size(); i < size; i++) {
- if (schemaList.get(i) != null) {
- indexOfMissingMeasurements.remove(i - removedCount);
- removedCount++;
- }
- }
- }
- if (indexOfMissingMeasurements.isEmpty()) {
- return;
- }
-
- autoCreateSchemaExecutor.autoCreateMissingMeasurements(
- schemaTree,
- devicePath,
- indexOfMissingMeasurements,
- measurements,
- getDataType,
- encodings,
- compressionTypes,
- isAligned);
- }
-
private List<Integer> checkMissingMeasurements(
ISchemaTree schemaTree, PartialPath devicePath, String[] measurements) {
DeviceSchemaInfo deviceSchemaInfo =
@@ -347,6 +330,32 @@ public class ClusterSchemaFetcher implements ISchemaFetcher {
return indexOfMissingMeasurements;
}
+ private List<Integer> checkMissingMeasurementsAfterSchemaFetch(
+ ClusterSchemaTree schemaTree,
+ PartialPath devicePath,
+ List<Integer> indexOfTargetMeasurements,
+ String[] measurements) {
+ DeviceSchemaInfo deviceSchemaInfo =
+ schemaTree.searchDeviceSchemaInfo(
+ devicePath,
+ indexOfTargetMeasurements.stream()
+ .map(index -> measurements[index])
+ .collect(Collectors.toList()));
+ if (deviceSchemaInfo == null) {
+ return indexOfTargetMeasurements;
+ }
+
+ List<Integer> indexOfMissingMeasurements = new ArrayList<>();
+ List<MeasurementSchema> schemaList = deviceSchemaInfo.getMeasurementSchemaList();
+ for (int i = 0, size = schemaList.size(); i < size; i++) {
+ if (schemaList.get(i) == null) {
+ indexOfMissingMeasurements.add(indexOfTargetMeasurements.get(i));
+ }
+ }
+
+ return indexOfMissingMeasurements;
+ }
+
@Override
public void invalidAllCache() {
DataNodeSchemaCache.getInstance().cleanUp();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanVisitor.java
index 1b78534118..81ab22aa87 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanVisitor.java
@@ -18,6 +18,8 @@
*/
package org.apache.iotdb.db.mpp.plan.planner;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.metadata.template.Template;
import org.apache.iotdb.db.mpp.common.MPPQueryContext;
import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
import org.apache.iotdb.db.mpp.plan.analyze.ExpressionAnalyzer;
@@ -29,6 +31,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.AlterTimeSe
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateAlignedTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateMultiTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateTimeSeriesNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.InternalBatchActivateTemplateNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.InternalCreateMultiTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.InternalCreateTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.MeasurementGroup;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
@@ -50,6 +54,8 @@ import org.apache.iotdb.db.mpp.plan.statement.crud.InsertRowsStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.InsertTabletStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.LoadTsFileStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.QueryStatement;
+import org.apache.iotdb.db.mpp.plan.statement.internal.InternalBatchActivateTemplateStatement;
+import org.apache.iotdb.db.mpp.plan.statement.internal.InternalCreateMultiTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.internal.InternalCreateTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.internal.SchemaFetchStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.AlterTimeSeriesStatement;
@@ -67,9 +73,11 @@ import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.ActivateTemplateStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.ShowPathsUsingTemplateStatement;
import org.apache.iotdb.db.mpp.plan.statement.sys.ShowQueriesStatement;
+import org.apache.iotdb.tsfile.utils.Pair;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -360,6 +368,15 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
createMultiTimeSeriesStatement.getAttributesList());
}
+ @Override
+ public PlanNode visitInternalCreateMultiTimeSeries(
+ InternalCreateMultiTimeSeriesStatement internalCreateMultiTimeSeriesStatement,
+ MPPQueryContext context) {
+ return new InternalCreateMultiTimeSeriesNode(
+ context.getQueryId().genPlanNodeId(),
+ internalCreateMultiTimeSeriesStatement.getDeviceMap());
+ }
+
@Override
public PlanNode visitAlterTimeseries(
AlterTimeSeriesStatement alterTimeSeriesStatement, MPPQueryContext context) {
@@ -695,6 +712,21 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
analysis.getTemplateSetInfo().left.getId());
}
+ @Override
+ public PlanNode visitInternalBatchActivateTemplate(
+ InternalBatchActivateTemplateStatement internalBatchActivateTemplateStatement,
+ MPPQueryContext context) {
+ Map<PartialPath, Pair<Integer, Integer>> templateActivationMap = new HashMap<>();
+ for (Map.Entry<PartialPath, Pair<Template, PartialPath>> entry :
+ internalBatchActivateTemplateStatement.getDeviceMap().entrySet()) {
+ templateActivationMap.put(
+ entry.getKey(),
+ new Pair<>(entry.getValue().left.getId(), entry.getValue().right.getNodeLength() - 1));
+ }
+ return new InternalBatchActivateTemplateNode(
+ context.getQueryId().genPlanNodeId(), templateActivationMap);
+ }
+
@Override
public PlanNode visitShowPathsUsingTemplate(
ShowPathsUsingTemplateStatement showPathsUsingTemplateStatement, MPPQueryContext context) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeType.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeType.java
index a289b1145b..a3db5a9db5 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeType.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeType.java
@@ -42,6 +42,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateMulti
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.DeactivateTemplateNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.DeleteTimeSeriesNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.InternalBatchActivateTemplateNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.InternalCreateMultiTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.InternalCreateTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.InvalidateSchemaCacheNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.PreDeactivateTemplateNode;
@@ -158,7 +160,9 @@ public enum PlanNodeType {
VERTICALLY_CONCAT((short) 64),
SINGLE_DEVICE_VIEW((short) 65),
MERGE_SORT((short) 66),
- SHOW_QUERIES((short) 67);
+ SHOW_QUERIES((short) 67),
+ INTERNAL_BATCH_ACTIVATE_TEMPLATE((short) 68),
+ INTERNAL_CREATE_MULTI_TIMESERIES((short) 69);
public static final int BYTES = Short.BYTES;
@@ -343,6 +347,10 @@ public enum PlanNodeType {
return MergeSortNode.deserialize(buffer);
case 67:
return ShowQueriesNode.deserialize(buffer);
+ case 68:
+ return InternalBatchActivateTemplateNode.deserialize(buffer);
+ case 69:
+ return InternalCreateMultiTimeSeriesNode.deserialize(buffer);
default:
throw new IllegalArgumentException("Invalid node type: " + nodeType);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java
index 96422f5074..2911971423 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java
@@ -41,6 +41,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateMulti
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.DeactivateTemplateNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.DeleteTimeSeriesNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.InternalBatchActivateTemplateNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.InternalCreateMultiTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.InternalCreateTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.PreDeactivateTemplateNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.RollbackPreDeactivateTemplateNode;
@@ -346,4 +348,12 @@ public abstract class PlanVisitor<R, C> {
public R visitShowQueries(ShowQueriesNode node, C context) {
return visitPlan(node, context);
}
+
+ public R visitInternalBatchActivateTemplate(InternalBatchActivateTemplateNode node, C context) {
+ return visitPlan(node, context);
+ }
+
+ public R visitInternalCreateMultiTimeSeries(InternalCreateMultiTimeSeriesNode node, C context) {
+ return visitPlan(node, context);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/InternalBatchActivateTemplateNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/InternalBatchActivateTemplateNode.java
new file mode 100644
index 0000000000..d705ba08fa
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/InternalBatchActivateTemplateNode.java
@@ -0,0 +1,162 @@
+/*
+ * 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.plan.planner.plan.node.metedata.write;
+
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.path.PathDeserializeUtil;
+import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.WritePlanNode;
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class InternalBatchActivateTemplateNode extends WritePlanNode {
+
+ // devicePath -> <templateId, templateSetLevel>
+ private Map<PartialPath, Pair<Integer, Integer>> templateActivationMap;
+
+ private TRegionReplicaSet regionReplicaSet;
+
+ public InternalBatchActivateTemplateNode(
+ PlanNodeId id, Map<PartialPath, Pair<Integer, Integer>> templateActivationMap) {
+ super(id);
+ this.templateActivationMap = templateActivationMap;
+ }
+
+ private InternalBatchActivateTemplateNode(
+ PlanNodeId id,
+ Map<PartialPath, Pair<Integer, Integer>> templateActivationMap,
+ TRegionReplicaSet regionReplicaSet) {
+ super(id);
+ this.templateActivationMap = templateActivationMap;
+ this.regionReplicaSet = regionReplicaSet;
+ }
+
+ public Map<PartialPath, Pair<Integer, Integer>> getTemplateActivationMap() {
+ return templateActivationMap;
+ }
+
+ @Override
+ public TRegionReplicaSet getRegionReplicaSet() {
+ return regionReplicaSet;
+ }
+
+ @Override
+ public List<PlanNode> getChildren() {
+ return new ArrayList<>();
+ }
+
+ @Override
+ public void addChild(PlanNode child) {}
+
+ @Override
+ public PlanNode clone() {
+ return new InternalBatchActivateTemplateNode(
+ getPlanNodeId(), templateActivationMap, regionReplicaSet);
+ }
+
+ @Override
+ public int allowedChildCount() {
+ return 0;
+ }
+
+ @Override
+ public List<String> getOutputColumnNames() {
+ return null;
+ }
+
+ @Override
+ protected void serializeAttributes(ByteBuffer byteBuffer) {
+ PlanNodeType.INTERNAL_BATCH_ACTIVATE_TEMPLATE.serialize(byteBuffer);
+
+ int size = templateActivationMap.size();
+ ReadWriteIOUtils.write(size, byteBuffer);
+ for (Map.Entry<PartialPath, Pair<Integer, Integer>> entry : templateActivationMap.entrySet()) {
+ entry.getKey().serialize(byteBuffer);
+ ReadWriteIOUtils.write(entry.getValue().left, byteBuffer);
+ ReadWriteIOUtils.write(entry.getValue().right, byteBuffer);
+ }
+ }
+
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.INTERNAL_BATCH_ACTIVATE_TEMPLATE.serialize(stream);
+
+ int size = templateActivationMap.size();
+ ReadWriteIOUtils.write(size, stream);
+ for (Map.Entry<PartialPath, Pair<Integer, Integer>> entry : templateActivationMap.entrySet()) {
+ entry.getKey().serialize(stream);
+ ReadWriteIOUtils.write(entry.getValue().left, stream);
+ ReadWriteIOUtils.write(entry.getValue().right, stream);
+ }
+ }
+
+ public static InternalBatchActivateTemplateNode deserialize(ByteBuffer byteBuffer) {
+ int size = ReadWriteIOUtils.readInt(byteBuffer);
+ Map<PartialPath, Pair<Integer, Integer>> templateActivationMap = new HashMap<>(size);
+ for (int i = 0; i < size; i++) {
+ templateActivationMap.put(
+ (PartialPath) PathDeserializeUtil.deserialize(byteBuffer),
+ new Pair<>(ReadWriteIOUtils.readInt(byteBuffer), ReadWriteIOUtils.readInt(byteBuffer)));
+ }
+
+ PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
+ return new InternalBatchActivateTemplateNode(planNodeId, templateActivationMap);
+ }
+
+ @Override
+ public List<WritePlanNode> splitByPartition(Analysis analysis) {
+ // gather devices to same target region
+ Map<TRegionReplicaSet, Map<PartialPath, Pair<Integer, Integer>>> splitMap = new HashMap<>();
+ for (Map.Entry<PartialPath, Pair<Integer, Integer>> entry : templateActivationMap.entrySet()) {
+ TRegionReplicaSet regionReplicaSet =
+ analysis.getSchemaPartitionInfo().getSchemaRegionReplicaSet(entry.getKey().getFullPath());
+ splitMap
+ .computeIfAbsent(regionReplicaSet, k -> new HashMap<>())
+ .put(entry.getKey(), entry.getValue());
+ }
+
+ List<WritePlanNode> result = new ArrayList<>();
+ for (Map.Entry<TRegionReplicaSet, Map<PartialPath, Pair<Integer, Integer>>> entry :
+ splitMap.entrySet()) {
+ result.add(
+ new InternalBatchActivateTemplateNode(getPlanNodeId(), entry.getValue(), entry.getKey()));
+ }
+
+ return result;
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitInternalBatchActivateTemplate(this, context);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/InternalCreateMultiTimeSeriesNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/InternalCreateMultiTimeSeriesNode.java
new file mode 100644
index 0000000000..40cd224e2f
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/InternalCreateMultiTimeSeriesNode.java
@@ -0,0 +1,164 @@
+/*
+ * 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.plan.planner.plan.node.metedata.write;
+
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.path.PathDeserializeUtil;
+import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.WritePlanNode;
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class InternalCreateMultiTimeSeriesNode extends WritePlanNode {
+
+ private Map<PartialPath, Pair<Boolean, MeasurementGroup>> deviceMap;
+
+ private TRegionReplicaSet regionReplicaSet;
+
+ public InternalCreateMultiTimeSeriesNode(
+ PlanNodeId id, Map<PartialPath, Pair<Boolean, MeasurementGroup>> deviceMap) {
+ super(id);
+ this.deviceMap = deviceMap;
+ }
+
+ private InternalCreateMultiTimeSeriesNode(
+ PlanNodeId id,
+ Map<PartialPath, Pair<Boolean, MeasurementGroup>> deviceMap,
+ TRegionReplicaSet regionReplicaSet) {
+ super(id);
+ this.deviceMap = deviceMap;
+ this.regionReplicaSet = regionReplicaSet;
+ }
+
+ public Map<PartialPath, Pair<Boolean, MeasurementGroup>> getDeviceMap() {
+ return deviceMap;
+ }
+
+ @Override
+ public TRegionReplicaSet getRegionReplicaSet() {
+ return regionReplicaSet;
+ }
+
+ @Override
+ public List<PlanNode> getChildren() {
+ return new ArrayList<>();
+ }
+
+ @Override
+ public void addChild(PlanNode child) {}
+
+ @Override
+ public PlanNode clone() {
+ return new InternalCreateMultiTimeSeriesNode(getPlanNodeId(), deviceMap, regionReplicaSet);
+ }
+
+ @Override
+ public int allowedChildCount() {
+ return 0;
+ }
+
+ @Override
+ public List<String> getOutputColumnNames() {
+ return null;
+ }
+
+ @Override
+ protected void serializeAttributes(ByteBuffer byteBuffer) {
+ PlanNodeType.INTERNAL_CREATE_MULTI_TIMESERIES.serialize(byteBuffer);
+
+ ReadWriteIOUtils.write(deviceMap.size(), byteBuffer);
+ for (Map.Entry<PartialPath, Pair<Boolean, MeasurementGroup>> entry : deviceMap.entrySet()) {
+ entry.getKey().serialize(byteBuffer);
+ ReadWriteIOUtils.write(entry.getValue().left, byteBuffer);
+ entry.getValue().right.serialize(byteBuffer);
+ }
+ }
+
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.INTERNAL_CREATE_MULTI_TIMESERIES.serialize(stream);
+
+ ReadWriteIOUtils.write(deviceMap.size(), stream);
+ for (Map.Entry<PartialPath, Pair<Boolean, MeasurementGroup>> entry : deviceMap.entrySet()) {
+ entry.getKey().serialize(stream);
+ ReadWriteIOUtils.write(entry.getValue().left, stream);
+ entry.getValue().right.serialize(stream);
+ }
+ }
+
+ public static InternalCreateMultiTimeSeriesNode deserialize(ByteBuffer byteBuffer) {
+ int size = ReadWriteIOUtils.readInt(byteBuffer);
+ Map<PartialPath, Pair<Boolean, MeasurementGroup>> deviceMap = new HashMap<>(size);
+ PartialPath devicePath;
+ boolean isAligned;
+ MeasurementGroup measurementGroup;
+ for (int i = 0; i < size; i++) {
+ devicePath = (PartialPath) PathDeserializeUtil.deserialize(byteBuffer);
+ isAligned = ReadWriteIOUtils.readBool(byteBuffer);
+ measurementGroup = new MeasurementGroup();
+ measurementGroup.deserialize(byteBuffer);
+ deviceMap.put(devicePath, new Pair<>(isAligned, measurementGroup));
+ }
+
+ PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
+ return new InternalCreateMultiTimeSeriesNode(planNodeId, deviceMap);
+ }
+
+ @Override
+ public List<WritePlanNode> splitByPartition(Analysis analysis) {
+ // gather devices to same target region
+ Map<TRegionReplicaSet, Map<PartialPath, Pair<Boolean, MeasurementGroup>>> splitMap =
+ new HashMap<>();
+ for (Map.Entry<PartialPath, Pair<Boolean, MeasurementGroup>> entry : deviceMap.entrySet()) {
+ TRegionReplicaSet regionReplicaSet =
+ analysis.getSchemaPartitionInfo().getSchemaRegionReplicaSet(entry.getKey().getFullPath());
+ splitMap
+ .computeIfAbsent(regionReplicaSet, k -> new HashMap<>())
+ .put(entry.getKey(), entry.getValue());
+ }
+
+ List<WritePlanNode> result = new ArrayList<>();
+ for (Map.Entry<TRegionReplicaSet, Map<PartialPath, Pair<Boolean, MeasurementGroup>>> entry :
+ splitMap.entrySet()) {
+ result.add(
+ new InternalCreateMultiTimeSeriesNode(getPlanNodeId(), entry.getValue(), entry.getKey()));
+ }
+
+ return result;
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitInternalCreateMultiTimeSeries(this, context);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementType.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementType.java
index 96eb0ee5f1..96ef577f18 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementType.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementType.java
@@ -150,4 +150,7 @@ public enum StatementType {
SHOW_TRIGGERS,
DEACTIVATE_TEMPLATE,
+
+ INTERNAL_BATCH_ACTIVATE_TEMPLATE,
+ INTERNAL_CREATE_MULTI_TIMESERIES
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementVisitor.java
index 280a0cf0cb..c78db0b7af 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementVisitor.java
@@ -28,6 +28,8 @@ import org.apache.iotdb.db.mpp.plan.statement.crud.InsertStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.InsertTabletStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.LoadTsFileStatement;
import org.apache.iotdb.db.mpp.plan.statement.crud.QueryStatement;
+import org.apache.iotdb.db.mpp.plan.statement.internal.InternalBatchActivateTemplateStatement;
+import org.apache.iotdb.db.mpp.plan.statement.internal.InternalCreateMultiTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.internal.InternalCreateTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.internal.SchemaFetchStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.AlterTimeSeriesStatement;
@@ -465,4 +467,14 @@ public abstract class StatementVisitor<R, C> {
DropSchemaTemplateStatement dropSchemaTemplateStatement, C context) {
return visitStatement(dropSchemaTemplateStatement, context);
}
+
+ public R visitInternalBatchActivateTemplate(
+ InternalBatchActivateTemplateStatement internalBatchActivateTemplateStatement, C context) {
+ return visitStatement(internalBatchActivateTemplateStatement, context);
+ }
+
+ public R visitInternalCreateMultiTimeSeries(
+ InternalCreateMultiTimeSeriesStatement internalCreateMultiTimeSeriesStatement, C context) {
+ return visitStatement(internalCreateMultiTimeSeriesStatement, context);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/internal/InternalBatchActivateTemplateStatement.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/internal/InternalBatchActivateTemplateStatement.java
new file mode 100644
index 0000000000..b1d3cf8bd1
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/internal/InternalBatchActivateTemplateStatement.java
@@ -0,0 +1,59 @@
+/*
+ * 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.plan.statement.internal;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.metadata.template.Template;
+import org.apache.iotdb.db.mpp.plan.statement.Statement;
+import org.apache.iotdb.db.mpp.plan.statement.StatementType;
+import org.apache.iotdb.db.mpp.plan.statement.StatementVisitor;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+// This is only used for auto activate template on multi devices while inserting data
+public class InternalBatchActivateTemplateStatement extends Statement {
+
+ // devicePath -> <Template, TemplateSetPath>
+ private Map<PartialPath, Pair<Template, PartialPath>> deviceMap;
+
+ public InternalBatchActivateTemplateStatement(
+ Map<PartialPath, Pair<Template, PartialPath>> deviceMap) {
+ super();
+ setType(StatementType.INTERNAL_BATCH_ACTIVATE_TEMPLATE);
+ this.deviceMap = deviceMap;
+ }
+
+ public Map<PartialPath, Pair<Template, PartialPath>> getDeviceMap() {
+ return deviceMap;
+ }
+
+ @Override
+ public List<? extends PartialPath> getPaths() {
+ return new ArrayList<>(deviceMap.keySet());
+ }
+
+ @Override
+ public <R, C> R accept(StatementVisitor<R, C> visitor, C context) {
+ return visitor.visitInternalBatchActivateTemplate(this, context);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/internal/InternalCreateMultiTimeSeriesStatement.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/internal/InternalCreateMultiTimeSeriesStatement.java
new file mode 100644
index 0000000000..42d0b0f935
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/internal/InternalCreateMultiTimeSeriesStatement.java
@@ -0,0 +1,57 @@
+/*
+ * 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.plan.statement.internal;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.MeasurementGroup;
+import org.apache.iotdb.db.mpp.plan.statement.Statement;
+import org.apache.iotdb.db.mpp.plan.statement.StatementType;
+import org.apache.iotdb.db.mpp.plan.statement.StatementVisitor;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+public class InternalCreateMultiTimeSeriesStatement extends Statement {
+
+ private Map<PartialPath, Pair<Boolean, MeasurementGroup>> deviceMap;
+
+ public InternalCreateMultiTimeSeriesStatement(
+ Map<PartialPath, Pair<Boolean, MeasurementGroup>> deviceMap) {
+ super();
+ statementType = StatementType.INTERNAL_CREATE_MULTI_TIMESERIES;
+ this.deviceMap = deviceMap;
+ }
+
+ public Map<PartialPath, Pair<Boolean, MeasurementGroup>> getDeviceMap() {
+ return deviceMap;
+ }
+
+ @Override
+ public List<? extends PartialPath> getPaths() {
+ return new ArrayList<>(deviceMap.keySet());
+ }
+
+ @Override
+ public <R, C> R accept(StatementVisitor<R, C> visitor, C context) {
+ return visitor.visitInternalCreateMultiTimeSeries(this, context);
+ }
+}