[To dev/1.3] Upgraded the schema cache implementation

integration-test/src/test/java/org/apache/iotdb/confignode/it/removeconfignode/IoTDBRemoveConfigNodeITFramework.java

@@ -99,7 +99,7 @@ public void testRemoveConfigNode(
         SyncConfigNodeIServiceClient client =
             (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
 
-      // Insert data in tree model
+      // Insert data
       statement.execute(TREE_MODEL_INSERTION);
 
       Map<Integer, Set<Integer>> regionMap = getDataRegionMap(statement);

integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeITFramework.java

@@ -154,7 +154,7 @@ public void testRemoveDataNode(
         SyncConfigNodeIServiceClient client =
             (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
 
-      // Insert data in tree model
+      // Insert data
       statement.execute(TREE_MODEL_INSERTION);
 
       Map<Integer, Set<Integer>> regionMap = getDataRegionMap(statement);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java

@@ -715,8 +715,7 @@ private TSStatus executeStatementWithRetryOnDataTypeMismatch(final Statement sta
 
     final TSStatus status = executeStatement(statement);
 
-    // Try to convert data type if the statement is a tree model statement
-    // and the status code is not success
+    // Try to convert the data type and the status code is not success
     return shouldConvertDataTypeOnTypeMismatch
             && ((statement instanceof InsertBaseStatement
                     && ((InsertBaseStatement) statement).hasFailedMeasurements())

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java

@@ -46,6 +46,7 @@
 import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
 import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
 import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeSchemaCache;
+import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DeviceLastCache;
 import org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
 import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
 import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
@@ -110,7 +111,7 @@ public Response executeFastLastQueryStatement(
 
       PartialPath prefixPath =
           new PartialPath(prefixPathList.getPrefixPaths().toArray(new String[0]));
-      Map<PartialPath, Map<String, TimeValuePair>> resultMap = new HashMap<>();
+      final Map<String, Map<PartialPath, Map<String, TimeValuePair>>> resultMap = new HashMap<>();
 
       final String prefixString = prefixPath.toString();
       for (ISchemaRegion region : SchemaEngine.getInstance().getAllSchemaRegions()) {
@@ -121,7 +122,7 @@ public Response executeFastLastQueryStatement(
         region.fillLastQueryMap(prefixPath, resultMap);
       }
       // Check cache first
-      if (!DataNodeSchemaCache.getInstance().getLastCache(resultMap)) {
+      if (!DataNodeSchemaCache.getInstance().getDeviceSchemaCache().getLastCache(resultMap)) {
         IClientSession clientSession = SESSION_MANAGER.getCurrSessionAndUpdateIdleTime();
         TSLastDataQueryReq tsLastDataQueryReq =
             FastLastHandler.createTSLastDataQueryReq(clientSession, prefixPathList);
@@ -169,14 +170,22 @@ public Response executeFastLastQueryStatement(
       List<Object> timeseries = new ArrayList<>();
       List<Object> valueList = new ArrayList<>();
       List<Object> dataTypeList = new ArrayList<>();
-      for (Map.Entry<PartialPath, Map<String, TimeValuePair>> entry : resultMap.entrySet()) {
-        final String deviceWithPrefix = entry.getKey() + TsFileConstant.PATH_SEPARATOR;
-        for (Map.Entry<String, TimeValuePair> measurementEntry : entry.getValue().entrySet()) {
-          final TimeValuePair tvPair = measurementEntry.getValue();
-          valueList.add(tvPair.getValue().getStringValue());
-          dataTypeList.add(tvPair.getValue().getDataType().name());
-          targetDataSet.addTimestampsItem(tvPair.getTimestamp());
-          timeseries.add(deviceWithPrefix + measurementEntry.getKey());
+      for (Map.Entry<String, Map<PartialPath, Map<String, TimeValuePair>>> entry :
+          resultMap.entrySet()) {
+        for (final Map.Entry<PartialPath, Map<String, TimeValuePair>> device2MeasurementLastEntry :
+            entry.getValue().entrySet()) {
+          final String deviceWithSeparator =
+              device2MeasurementLastEntry.getKey() + TsFileConstant.PATH_SEPARATOR;
+          for (Map.Entry<String, TimeValuePair> measurementEntry :
+              device2MeasurementLastEntry.getValue().entrySet()) {
+            final TimeValuePair tvPair = measurementEntry.getValue();
+            if (tvPair != DeviceLastCache.EMPTY_TIME_VALUE_PAIR) {
+              valueList.add(tvPair.getValue().getStringValue());
+              dataTypeList.add(tvPair.getValue().getDataType().name());
+              targetDataSet.addTimestampsItem(tvPair.getTimestamp());
+              timeseries.add(deviceWithSeparator + measurementEntry.getKey());
+            }
+          }
         }
       }
       if (!timeseries.isEmpty()) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java

@@ -70,6 +70,7 @@
 import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
 import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
 import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeSchemaCache;
+import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DeviceLastCache;
 import org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
 import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
 import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
@@ -817,7 +818,7 @@ public TSExecuteStatementResp executeFastLastDataQueryForOnePrefixPath(
                     "The \"executeFastLastDataQueryForOnePrefixPath\" dos not support wildcards."));
       }
 
-      final Map<PartialPath, Map<String, TimeValuePair>> resultMap = new HashMap<>();
+      final Map<String, Map<PartialPath, Map<String, TimeValuePair>>> resultMap = new HashMap<>();
       int sensorNum = 0;
 
       final String prefixString = prefixPath.toString();
@@ -830,26 +831,34 @@ public TSExecuteStatementResp executeFastLastDataQueryForOnePrefixPath(
       }
 
       // 2.DATA_NODE_SCHEMA_CACHE.getLastCache()
-      if (!DataNodeSchemaCache.getInstance().getLastCache(resultMap)) {
+      if (!DataNodeSchemaCache.getInstance().getDeviceSchemaCache().getLastCache(resultMap)) {
         // 2.1 any sensor miss cache, construct last query sql, then return
         return executeLastDataQueryInternal(convert(req), SELECT_RESULT);
       }
 
       // 2.2 all sensors hit cache, return response ~= 20ms
       final TsBlockBuilder builder = LastQueryUtil.createTsBlockBuilder(sensorNum);
 
-      for (final Map.Entry<PartialPath, Map<String, TimeValuePair>> result : resultMap.entrySet()) {
-        final String deviceWithPrefix = result.getKey() + TsFileConstant.PATH_SEPARATOR;
-        for (final Map.Entry<String, TimeValuePair> measurementLastEntry :
+      for (final Map.Entry<String, Map<PartialPath, Map<String, TimeValuePair>>> result :
+          resultMap.entrySet()) {
+        for (final Map.Entry<PartialPath, Map<String, TimeValuePair>> device2MeasurementLastEntry :
             result.getValue().entrySet()) {
-          final TimeValuePair tvPair = measurementLastEntry.getValue();
-          LastQueryUtil.appendLastValue(
-              builder,
-              tvPair.getTimestamp(),
-              new Binary(
-                  deviceWithPrefix + measurementLastEntry.getKey(), TSFileConfig.STRING_CHARSET),
-              tvPair.getValue().getStringValue(),
-              tvPair.getValue().getDataType().name());
+          final String deviceWithSeparator =
+              device2MeasurementLastEntry.getKey() + TsFileConstant.PATH_SEPARATOR;
+          for (final Map.Entry<String, TimeValuePair> measurementLastEntry :
+              device2MeasurementLastEntry.getValue().entrySet()) {
+            final TimeValuePair tvPair = measurementLastEntry.getValue();
+            if (tvPair != DeviceLastCache.EMPTY_TIME_VALUE_PAIR) {
+              LastQueryUtil.appendLastValue(
+                  builder,
+                  tvPair.getTimestamp(),
+                  new Binary(
+                      deviceWithSeparator + measurementLastEntry.getKey(),
+                      TSFileConfig.STRING_CHARSET),
+                  tvPair.getValue().getStringValue(),
+                  tvPair.getValue().getDataType().name());
+            }
+          }
         }
       }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java

@@ -537,7 +537,7 @@ public TSStatus invalidatePartitionCache(final TInvalidateCacheReq req) {
 
   @Override
   public TSStatus invalidateLastCache(final String database) {
-    DataNodeSchemaCache.getInstance().invalidateLastCacheInDataRegion(database);
+    DataNodeSchemaCache.getInstance().invalidateDatabaseLastCache(database);
     return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
   }
 
@@ -546,7 +546,7 @@ public TSStatus invalidateSchemaCache(final TInvalidateCacheReq req) {
     DataNodeSchemaCache.getInstance().takeWriteLock();
     try {
       // req.getFullPath() is a database path
-      DataNodeSchemaCache.getInstance().invalidate(req.getFullPath());
+      DataNodeSchemaCache.getInstance().getDeviceSchemaCache().invalidate(req.getFullPath());
       ClusterTemplateManager.getInstance().invalid(req.getFullPath());
       LOGGER.info("Schema cache of {} has been invalidated", req.getFullPath());
       return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
@@ -1974,12 +1974,16 @@ public TSStatus clearCacheImpl(final Set<CacheClearOptions> options) {
           || options.contains(CacheClearOptions.QUERY)) {
         storageEngine.clearCache();
       }
-      if (options.contains(CacheClearOptions.TREE_SCHEMA)) {
-        schemaCache.invalidateAll();
+      if (options.contains(CacheClearOptions.QUERY)
+          && options.contains(CacheClearOptions.TREE_SCHEMA)) {
+        schemaCache.getDeviceSchemaCache().invalidateAll();
         return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
       }
       if (options.contains(CacheClearOptions.QUERY)) {
-        schemaCache.invalidateLastCache(new PartialPath("root.**"));
+        schemaCache.getDeviceSchemaCache().invalidateLastCache();
+      }
+      if (options.contains(CacheClearOptions.TREE_SCHEMA)) {
+        schemaCache.getDeviceSchemaCache().invalidateTreeSchema();
       }
     } catch (final Exception e) {
       return RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, e.getMessage());

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/schematree/DeviceSchemaInfo.java

@@ -30,16 +30,12 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
-import static org.apache.iotdb.commons.schema.SchemaConstant.NON_TEMPLATE;
-
 public class DeviceSchemaInfo {
 
-  private PartialPath devicePath;
-  private boolean isAligned;
-  private List<IMeasurementSchemaInfo> measurementSchemaInfoList;
-  private int templateId = NON_TEMPLATE;
-
-  private DeviceSchemaInfo() {}
+  private final PartialPath devicePath;
+  private final boolean isAligned;
+  private final List<IMeasurementSchemaInfo> measurementSchemaInfoList;
+  private final int templateId;
 
   public DeviceSchemaInfo(
       PartialPath devicePath,

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/last/AbstractUpdateLastCacheOperator.java

@@ -26,6 +26,7 @@
 import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
 import org.apache.iotdb.db.queryengine.execution.operator.process.ProcessOperator;
 import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeSchemaCache;
+import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DeviceLastCache;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.ListenableFuture;
@@ -35,9 +36,11 @@
 import org.apache.tsfile.read.common.block.TsBlockBuilder;
 import org.apache.tsfile.utils.Pair;
 import org.apache.tsfile.utils.TsPrimitiveType;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
 
 import javax.annotation.Nullable;
 
+import java.util.Objects;
 import java.util.concurrent.atomic.AtomicInteger;
 
 public abstract class AbstractUpdateLastCacheOperator implements ProcessOperator {
@@ -117,18 +120,36 @@ protected void mayUpdateLastCache(
       }
 
       if (!deviceInMultiRegion) {
-        lastCache.updateLastCache(
-            getDatabaseName(), fullPath, new TimeValuePair(time, value), false, Long.MIN_VALUE);
+        lastCache.updateLastCacheIfExists(
+            getDatabaseName(),
+            fullPath.getDevicePath(),
+            new String[] {fullPath.getMeasurement()},
+            new TimeValuePair[] {
+              Objects.nonNull(value)
+                  ? new TimeValuePair(time, value)
+                  : needUpdateNullEntry ? DeviceLastCache.EMPTY_TIME_VALUE_PAIR : null
+            },
+            fullPath.isUnderAlignedEntity(),
+            new IMeasurementSchema[] {fullPath.getMeasurementSchema()});
         return;
       }
       // update cache in DataNodeQueryContext
       if (seriesScanInfo.right == null || time > seriesScanInfo.right.getTimestamp()) {
-        seriesScanInfo.right = new TimeValuePair(time, value);
+        if (Objects.nonNull(value)) {
+          seriesScanInfo.right = new TimeValuePair(time, value);
+        } else {
+          seriesScanInfo.right = needUpdateNullEntry ? DeviceLastCache.EMPTY_TIME_VALUE_PAIR : null;
+        }
       }
 
       if (seriesScanInfo.left.decrementAndGet() == 0) {
-        lastCache.updateLastCache(
-            getDatabaseName(), fullPath, seriesScanInfo.right, false, Long.MIN_VALUE);
+        lastCache.updateLastCacheIfExists(
+            getDatabaseName(),
+            fullPath.getDevicePath(),
+            new String[] {fullPath.getMeasurement()},
+            new TimeValuePair[] {seriesScanInfo.right},
+            fullPath.isUnderAlignedEntity(),
+            new IMeasurementSchema[] {fullPath.getMeasurementSchema()});
       }
     } finally {
       dataNodeQueryContext.unLock(deviceInMultiRegion);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/last/LastQueryUtil.java

@@ -166,8 +166,8 @@ public static List<Aggregator> createAggregators(TSDataType dataType, int valueC
 
   public static boolean needUpdateCache(Filter timeFilter) {
     // Update the cache only when, the filter is gt (greater than) or ge (greater than or equal to)
-    return CACHE_ENABLED && (timeFilter == null || timeFilter instanceof TimeGtEq)
-        || (timeFilter instanceof TimeGt);
+    return CACHE_ENABLED
+        && (timeFilter == null || timeFilter instanceof TimeGtEq || timeFilter instanceof TimeGt);
   }
 
   public static boolean needUpdateNullEntry(Filter timeFilter) {