[IOTDB-3184] Implement Timeseries version and blacklist

schema-engine-rocksdb/src/main/java/org/apache/iotdb/db/metadata/schemaregion/rocksdb/RSchemaRegion.java

@@ -264,6 +264,13 @@ public void createTimeseries(CreateTimeSeriesPlan plan, long offset) throws Meta
     }
   }
 
+  @Override
+  public void createTimeseries(CreateTimeSeriesPlan plan, long offset, String version)
+      throws MetadataException {
+    throw new UnsupportedOperationException(
+        "RSchemaRegion currently doesn't support timeseries with version");
+  }
+
   @TestOnly
   protected void createTimeseries(
       PartialPath path,
@@ -527,6 +534,13 @@ public void createAlignedTimeSeries(CreateAlignedTimeSeriesPlan plan) throws Met
     }
   }
 
+  @Override
+  public void createAlignedTimeSeries(CreateAlignedTimeSeriesPlan plan, List<String> versionList)
+      throws MetadataException {
+    throw new UnsupportedOperationException(
+        "RSchemaRegion currently doesn't support timeseries with version");
+  }
+
   private void createEntityRecursively(String[] nodes, int start, int end, boolean aligned)
       throws RocksDBException, MetadataException, InterruptedException {
     if (start <= end) {

schema-engine-rocksdb/src/main/java/org/apache/iotdb/db/metadata/schemaregion/rocksdb/mnode/RMeasurementMNode.java

@@ -150,6 +150,16 @@ public void setLastCacheContainer(ILastCacheContainer lastCacheContainer) {
     throw new UnsupportedOperationException();
   }
 
+  @Override
+  public String getVersion() {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public void setVersion(String version) {
+    throw new UnsupportedOperationException();
+  }
+
   @Override
   public void serializeTo(MLogWriter logWriter) throws IOException {
     throw new UnsupportedOperationException();

server/src/main/java/org/apache/iotdb/db/metadata/cache/DataNodeSchemaCache.java

@@ -71,8 +71,9 @@ public SchemaTree get(PartialPath devicePath, String[] measurements) {
             devicePath.concatNode(
                 schemaCacheEntry.getSchemaEntryId()), // the cached path may be alias path
             schemaCacheEntry.getMeasurementSchema(),
-            schemaCacheEntry.getAlias(),
-            schemaCacheEntry.isAligned());
+            null,
+            schemaCacheEntry.isAligned(),
+            schemaCacheEntry.getVersion());
       }
     }
     return schemaTree;
@@ -83,17 +84,9 @@ public void put(SchemaTree schemaTree) {
       SchemaCacheEntry schemaCacheEntry =
           new SchemaCacheEntry(
               (MeasurementSchema) measurementPath.getMeasurementSchema(),
-              measurementPath.isMeasurementAliasExists()
-                  ? measurementPath.getMeasurementAlias()
-                  : null,
-              measurementPath.isUnderAlignedEntity());
+              measurementPath.isUnderAlignedEntity(),
+              measurementPath.getVersion());
       cache.put(new PartialPath(measurementPath.getNodes()), schemaCacheEntry);
-      if (measurementPath.isMeasurementAliasExists()) {
-        // cache alias path
-        cache.put(
-            measurementPath.getDevicePath().concatNode(measurementPath.getMeasurementAlias()),
-            schemaCacheEntry);
-      }
     }
   }
 

server/src/main/java/org/apache/iotdb/db/metadata/cache/SchemaCacheEntry.java

@@ -28,16 +28,16 @@ public class SchemaCacheEntry {
 
   private final MeasurementSchema measurementSchema;
 
-  private final String alias;
-
   private final boolean isAligned;
 
+  private final String version;
+
   private volatile ILastCacheContainer lastCacheContainer = null;
 
-  SchemaCacheEntry(MeasurementSchema measurementSchema, String alias, boolean isAligned) {
+  SchemaCacheEntry(MeasurementSchema measurementSchema, boolean isAligned, String version) {
     this.measurementSchema = measurementSchema;
-    this.alias = alias;
     this.isAligned = isAligned;
+    this.version = version;
   }
 
   public String getSchemaEntryId() {
@@ -52,14 +52,14 @@ public TSDataType getTsDataType() {
     return measurementSchema.getType();
   }
 
-  public String getAlias() {
-    return alias;
-  }
-
   public boolean isAligned() {
     return isAligned;
   }
 
+  public String getVersion() {
+    return version;
+  }
+
   public ILastCacheContainer getLastCacheContainer() {
     if (lastCacheContainer == null) {
       synchronized (this) {

server/src/main/java/org/apache/iotdb/db/metadata/idtable/entry/InsertMeasurementMNode.java

@@ -96,6 +96,16 @@ public ILastCacheContainer getLastCacheContainer() {
   @Override
   public void setLastCacheContainer(ILastCacheContainer lastCacheContainer) {}
 
+  @Override
+  public String getVersion() {
+    throw new UnsupportedOperationException("insert measurement mnode doesn't support this method");
+  }
+
+  @Override
+  public void setVersion(String version) {
+    throw new UnsupportedOperationException("insert measurement mnode doesn't support this method");
+  }
+
   @Override
   public IMeasurementSchema getSchema() {
     return schema;

server/src/main/java/org/apache/iotdb/db/metadata/mnode/IMeasurementMNode.java

@@ -51,4 +51,8 @@ public interface IMeasurementMNode extends IMNode {
   ILastCacheContainer getLastCacheContainer();
 
   void setLastCacheContainer(ILastCacheContainer lastCacheContainer);
+
+  String getVersion();
+
+  void setVersion(String version);
 }

server/src/main/java/org/apache/iotdb/db/metadata/mnode/MeasurementMNode.java

@@ -48,6 +48,8 @@ public class MeasurementMNode extends MNode implements IMeasurementMNode {
   /** last value cache */
   private volatile ILastCacheContainer lastCacheContainer = null;
 
+  private String version = null;
+
   /**
    * MeasurementMNode factory method. The type of returned MeasurementMNode is according to the
    * schema type. The default type is UnaryMeasurementMNode, which means if schema == null, an
@@ -148,6 +150,16 @@ public void setLastCacheContainer(ILastCacheContainer lastCacheContainer) {
     this.lastCacheContainer = lastCacheContainer;
   }
 
+  @Override
+  public String getVersion() {
+    return version;
+  }
+
+  @Override
+  public void setVersion(String version) {
+    this.version = version;
+  }
+
   @Override
   public void serializeTo(MLogWriter logWriter) throws IOException {
     logWriter.serializeMeasurementMNode(this);

server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTreeBelowSGMemoryImpl.java

@@ -610,6 +610,7 @@ protected void collectMeasurement(IMeasurementMNode node) {
               // only when user query with alias, the alias in path will be set
               path.setMeasurementAlias(node.getAlias());
             }
+            path.setVersion(node.getVersion());
             result.add(path);
           }
         };

server/src/main/java/org/apache/iotdb/db/metadata/path/MeasurementPath.java

@@ -32,6 +32,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.nio.ByteBuffer;
+import java.util.Objects;
 
 public class MeasurementPath extends PartialPath {
 
@@ -44,6 +45,8 @@ public class MeasurementPath extends PartialPath {
   // alias of measurement, null pointer cannot be serialized in thrift so empty string is instead
   private String measurementAlias = "";
 
+  private String version = null;
+
   public MeasurementPath() {}
 
   public MeasurementPath(String measurementPath) throws IllegalPathException {
@@ -123,6 +126,14 @@ public void setUnderAlignedEntity(boolean underAlignedEntity) {
     isUnderAlignedEntity = underAlignedEntity;
   }
 
+  public String getVersion() {
+    return version;
+  }
+
+  public void setVersion(String version) {
+    this.version = version;
+  }
+
   @Override
   public PartialPath copy() {
     MeasurementPath result = new MeasurementPath();
@@ -135,6 +146,15 @@ public PartialPath copy() {
     return result;
   }
 
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    if (!super.equals(o)) return false;
+    MeasurementPath that = (MeasurementPath) o;
+    return Objects.equals(version, that.version);
+  }
+
   /**
    * if isUnderAlignedEntity is true, return an AlignedPath with only one sub sensor otherwise,
    * return itself
@@ -173,6 +193,8 @@ public void serialize(ByteBuffer byteBuffer) {
     }
     ReadWriteIOUtils.write(isUnderAlignedEntity, byteBuffer);
     ReadWriteIOUtils.write(measurementAlias, byteBuffer);
+
+    ReadWriteIOUtils.write(version, byteBuffer);
   }
 
   public static MeasurementPath deserialize(ByteBuffer byteBuffer) {
@@ -189,6 +211,9 @@ public static MeasurementPath deserialize(ByteBuffer byteBuffer) {
     }
     measurementPath.isUnderAlignedEntity = ReadWriteIOUtils.readBool(byteBuffer);
     measurementPath.measurementAlias = ReadWriteIOUtils.readString(byteBuffer);
+
+    measurementPath.version = ReadWriteIOUtils.readString(byteBuffer);
+
     measurementPath.nodes = partialPath.getNodes();
     measurementPath.device = partialPath.getDevice();
     measurementPath.fullPath = partialPath.getFullPath();

server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/ISchemaRegion.java

@@ -100,8 +100,14 @@ public interface ISchemaRegion {
   // region Interfaces for Timeseries operation
   void createTimeseries(CreateTimeSeriesPlan plan, long offset) throws MetadataException;
 
+  void createTimeseries(CreateTimeSeriesPlan plan, long offset, String version)
+      throws MetadataException;
+
   void createAlignedTimeSeries(CreateAlignedTimeSeriesPlan plan) throws MetadataException;
 
+  void createAlignedTimeSeries(CreateAlignedTimeSeriesPlan plan, List<String> versionList)
+      throws MetadataException;
+
   /**
    * Delete all timeseries matching the given path pattern. If using prefix match, the path pattern
    * is used to match prefix path. All timeseries start with the matched prefix path will be

server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/SchemaRegionMemoryImpl.java

@@ -497,6 +497,12 @@ public void createTimeseries(CreateTimeSeriesPlan plan) throws MetadataException
 
   @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
   public void createTimeseries(CreateTimeSeriesPlan plan, long offset) throws MetadataException {
+    createTimeseries(plan, offset, null);
+  }
+
+  @Override
+  public void createTimeseries(CreateTimeSeriesPlan plan, long offset, String version)
+      throws MetadataException {
     if (!memoryStatistics.isAllowToCreateNewSeries()) {
       throw new SeriesOverflowException();
     }
@@ -516,6 +522,8 @@ public void createTimeseries(CreateTimeSeriesPlan plan, long offset) throws Meta
               plan.getProps(),
               plan.getAlias());
 
+      leafMNode.setVersion(version);
+
       // the cached mNode may be replaced by new entityMNode in mtree
       mNodeCache.invalidate(path.getDevicePath());
 
@@ -607,6 +615,12 @@ public void createAlignedTimeSeries(
    * @param plan CreateAlignedTimeSeriesPlan
    */
   public void createAlignedTimeSeries(CreateAlignedTimeSeriesPlan plan) throws MetadataException {
+    createAlignedTimeSeries(plan, null);
+  }
+
+  @Override
+  public void createAlignedTimeSeries(CreateAlignedTimeSeriesPlan plan, List<String> versionList)
+      throws MetadataException {
     if (!memoryStatistics.isAllowToCreateNewSeries()) {
       throw new SeriesOverflowException();
     }
@@ -633,6 +647,12 @@ public void createAlignedTimeSeries(CreateAlignedTimeSeriesPlan plan) throws Met
               plan.getCompressors(),
               plan.getAliasList());
 
+      if (versionList != null) {
+        for (int i = 0; i < measurements.size(); i++) {
+          measurementMNodeList.get(i).setVersion(versionList.get(i));
+        }
+      }
+
       // the cached mNode may be replaced by new entityMNode in mtree
       mNodeCache.invalidate(prefixPath);
 

server/src/main/java/org/apache/iotdb/db/metadata/schemaregion/SchemaRegionSchemaFileImpl.java

@@ -538,6 +538,13 @@ public void createTimeseries(CreateTimeSeriesPlan plan, long offset) throws Meta
     }
   }
 
+  @Override
+  public void createTimeseries(CreateTimeSeriesPlan plan, long offset, String version)
+      throws MetadataException {
+    throw new UnsupportedOperationException(
+        "SchemaRegion schema file mode currently doesn't support timeseries with version");
+  }
+
   /**
    * Add one timeseries to metadata tree, if the timeseries already exists, throw exception
    *
@@ -700,6 +707,13 @@ public void createAlignedTimeSeries(CreateAlignedTimeSeriesPlan plan) throws Met
     }
   }
 
+  @Override
+  public void createAlignedTimeSeries(CreateAlignedTimeSeriesPlan plan, List<String> versionList)
+      throws MetadataException {
+    throw new UnsupportedOperationException(
+        "SchemaRegion schema file mode currently doesn't support timeseries with version");
+  }
+
   /**
    * Delete all timeseries matching the given path pattern. If using prefix match, the path pattern
    * is used to match prefix path. All timeseries start with the matched prefix path will be

server/src/main/java/org/apache/iotdb/db/metadata/utils/TimeseriesVersionUtil.java

@@ -0,0 +1,28 @@
+/*
+ * 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.metadata.utils;
+
+import java.util.UUID;
+
+public class TimeseriesVersionUtil {
+
+  public static String generateVersion() {
+    return UUID.randomUUID().toString();
+  }
+}

server/src/main/java/org/apache/iotdb/db/metadata/visitor/SchemaExecutionVisitor.java

@@ -54,7 +54,7 @@ public class SchemaExecutionVisitor extends PlanVisitor<TSStatus, ISchemaRegion>
   public TSStatus visitCreateTimeSeries(CreateTimeSeriesNode node, ISchemaRegion schemaRegion) {
     try {
       PhysicalPlan plan = node.accept(new PhysicalPlanTransformer(), new TransformerContext());
-      schemaRegion.createTimeseries((CreateTimeSeriesPlan) plan, -1);
+      schemaRegion.createTimeseries((CreateTimeSeriesPlan) plan, -1, node.getVersion());
     } catch (MetadataException e) {
       logger.error("{}: MetaData error: ", IoTDBConstant.GLOBAL_DB_NAME, e);
       return RpcUtils.getStatus(TSStatusCode.METADATA_ERROR, e.getMessage());
@@ -67,7 +67,8 @@ public TSStatus visitCreateAlignedTimeSeries(
       CreateAlignedTimeSeriesNode node, ISchemaRegion schemaRegion) {
     try {
       PhysicalPlan plan = node.accept(new PhysicalPlanTransformer(), new TransformerContext());
-      schemaRegion.createAlignedTimeSeries((CreateAlignedTimeSeriesPlan) plan);
+      schemaRegion.createAlignedTimeSeries(
+          (CreateAlignedTimeSeriesPlan) plan, node.getVersionList());
     } catch (MetadataException e) {
       logger.error("{}: MetaData error: ", IoTDBConstant.GLOBAL_DB_NAME, e);
       return RpcUtils.getStatus(TSStatusCode.METADATA_ERROR, e.getMessage());