complete

apache · Nov 29, 2024 · 70c99e4 · 70c99e4
1 parent c5eacf0
commit 70c99e4
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 13 deletions.
diff --git a/.../java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaExecutionVisitor.java b/.../java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaExecutionVisitor.java
@@ -95,10 +95,11 @@ public class SchemaExecutionVisitor extends PlanVisitor<TSStatus, ISchemaRegion>
   private static final Logger logger = LoggerFactory.getLogger(SchemaExecutionVisitor.class);
 
   @Override
-  public TSStatus visitCreateTimeSeries(CreateTimeSeriesNode node, ISchemaRegion schemaRegion) {
+  public TSStatus visitCreateTimeSeries(
+      final CreateTimeSeriesNode node, final ISchemaRegion schemaRegion) {
     try {
       schemaRegion.createTimeSeries(node, -1);
-    } catch (MetadataException e) {
+    } catch (final MetadataException e) {
       logger.error("{}: MetaData error: ", IoTDBConstant.GLOBAL_DB_NAME, e);
       return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
     }
@@ -107,10 +108,10 @@ public TSStatus visitCreateTimeSeries(CreateTimeSeriesNode node, ISchemaRegion s
 
   @Override
   public TSStatus visitCreateAlignedTimeSeries(
-      CreateAlignedTimeSeriesNode node, ISchemaRegion schemaRegion) {
+      final CreateAlignedTimeSeriesNode node, final ISchemaRegion schemaRegion) {
     try {
       schemaRegion.createAlignedTimeSeries(node);
-    } catch (MetadataException e) {
+    } catch (final MetadataException e) {
       logger.error("{}: MetaData error: ", IoTDBConstant.GLOBAL_DB_NAME, e);
       return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
     }
@@ -119,13 +120,13 @@ public TSStatus visitCreateAlignedTimeSeries(
 
   @Override
   public TSStatus visitCreateMultiTimeSeries(
-      CreateMultiTimeSeriesNode node, ISchemaRegion schemaRegion) {
-    Map<PartialPath, MeasurementGroup> measurementGroupMap = node.getMeasurementGroupMap();
-    List<TSStatus> failingStatus = new ArrayList<>();
+      final CreateMultiTimeSeriesNode node, final ISchemaRegion schemaRegion) {
+    final Map<PartialPath, MeasurementGroup> measurementGroupMap = node.getMeasurementGroupMap();
+    final List<TSStatus> failingStatus = new ArrayList<>();
     PartialPath devicePath;
     MeasurementGroup measurementGroup;
     int size;
-    for (Map.Entry<PartialPath, MeasurementGroup> entry : measurementGroupMap.entrySet()) {
+    for (final Map.Entry<PartialPath, MeasurementGroup> entry : measurementGroupMap.entrySet()) {
       devicePath = entry.getKey();
       measurementGroup = entry.getValue();
       size = measurementGroup.getMeasurements().size();
@@ -148,7 +149,7 @@ public TSStatus visitCreateMultiTimeSeries(
   }
 
   private ICreateTimeSeriesPlan transformToCreateTimeSeriesPlan(
-      PartialPath devicePath, MeasurementGroup measurementGroup, int index) {
+      final PartialPath devicePath, final MeasurementGroup measurementGroup, final int index) {
     return SchemaRegionWritePlanFactory.getCreateTimeSeriesPlan(
         devicePath.concatAsMeasurementPath(measurementGroup.getMeasurements().get(index)),
         measurementGroup.getDataTypes().get(index),

diff --git a/.../main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java b/.../main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
@@ -49,6 +49,8 @@
 import org.apache.iotdb.db.queryengine.execution.relational.ColumnTransformerBuilder;
 import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
 import org.apache.iotdb.db.queryengine.plan.planner.LocalExecutionPlanner;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.CreateAlignedTimeSeriesNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.CreateTimeSeriesNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.InputLocation;
 import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
 import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceSchemaCache;
@@ -677,8 +679,10 @@ public void createTimeSeries(final ICreateTimeSeriesPlan plan, long offset)
               plan.getCompressor(),
               plan.getProps(),
               plan.getAlias(),
-              (plan instanceof CreateTimeSeriesPlanImpl
-                  && ((CreateTimeSeriesPlanImpl) plan).isWithMerge()));
+              plan instanceof CreateTimeSeriesPlanImpl
+                      && ((CreateTimeSeriesPlanImpl) plan).isWithMerge()
+                  || plan instanceof CreateTimeSeriesNode
+                      && ((CreateTimeSeriesNode) plan).isGeneratedByPipe());
 
       // Should merge
       if (Objects.isNull(leafMNode)) {
@@ -760,8 +764,10 @@ public void createAlignedTimeSeries(final ICreateAlignedTimeSeriesPlan plan)
               encodings,
               compressors,
               aliasList,
-              (plan instanceof CreateAlignedTimeSeriesPlanImpl
-                  && ((CreateAlignedTimeSeriesPlanImpl) plan).isWithMerge()),
+              plan instanceof CreateAlignedTimeSeriesPlanImpl
+                      && ((CreateAlignedTimeSeriesPlanImpl) plan).isWithMerge()
+                  || plan instanceof CreateAlignedTimeSeriesNode
+                      && ((CreateAlignedTimeSeriesNode) plan).isGeneratedByPipe(),
               existingMeasurementIndexes);
 
       // update statistics and schemaDataTypeNumMap