Implement Attribute Security Service & Improved semantics of "clear cache" SQL

iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4

@@ -1035,7 +1035,7 @@ flush
 
 // Clear Cache
 clearCache
-    : CLEAR CACHE (ON (LOCAL | CLUSTER))?
+    : CLEAR (SCHEMA | QUERY | ALL)? CACHE (ON (LOCAL | CLUSTER))?
     ;
 
 // Set Configuration

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java

@@ -95,6 +95,7 @@
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 /** Asynchronously send RPC requests to DataNodes. See queryengine.thrift for more details. */
@@ -104,6 +105,7 @@ public class CnToDnInternalServiceAsyncRequestManager
   private static final Logger LOGGER =
       LoggerFactory.getLogger(CnToDnInternalServiceAsyncRequestManager.class);
 
+  @SuppressWarnings("unchecked")
   @Override
   protected void initActionMapBuilder() {
     actionMapBuilder.put(
@@ -223,7 +225,8 @@ protected void initActionMapBuilder() {
             client.flush((TFlushReq) req, (DataNodeTSStatusRPCHandler) handler));
     actionMapBuilder.put(
         CnToDnAsyncRequestType.CLEAR_CACHE,
-        (req, client, handler) -> client.clearCache((DataNodeTSStatusRPCHandler) handler));
+        (req, client, handler) ->
+            client.clearCache((Set<Integer>) req, (DataNodeTSStatusRPCHandler) handler));
     actionMapBuilder.put(
         CnToDnAsyncRequestType.START_REPAIR_DATA,
         (req, client, handler) -> client.startRepairData((DataNodeTSStatusRPCHandler) handler));

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java

@@ -1614,18 +1614,18 @@ public TSStatus merge() {
   }
 
   @Override
-  public TSStatus flush(TFlushReq req) {
-    TSStatus status = confirmLeader();
+  public TSStatus flush(final TFlushReq req) {
+    final TSStatus status = confirmLeader();
     return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
         ? RpcUtils.squashResponseStatusList(nodeManager.flush(req))
         : status;
   }
 
   @Override
-  public TSStatus clearCache() {
-    TSStatus status = confirmLeader();
+  public TSStatus clearCache(final Set<Integer> clearCacheOptions) {
+    final TSStatus status = confirmLeader();
     return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
-        ? RpcUtils.squashResponseStatusList(nodeManager.clearCache())
+        ? RpcUtils.squashResponseStatusList(nodeManager.clearCache(clearCacheOptions))
         : status;
   }
 

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java

@@ -557,7 +557,7 @@ TDataPartitionTableResp getOrCreateDataPartition(
   TSStatus flush(TFlushReq req);
 
   /** Clear cache on all DataNodes. */
-  TSStatus clearCache();
+  TSStatus clearCache(final Set<Integer> clearCacheOptions);
 
   /** Set Configuration. */
   TSStatus setConfiguration(TSetConfigurationReq req);

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java

@@ -881,11 +881,12 @@ public List<TSStatus> flush(TFlushReq req) {
     return clientHandler.getResponseList();
   }
 
-  public List<TSStatus> clearCache() {
-    Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+  public List<TSStatus> clearCache(final Set<Integer> clearCacheOptions) {
+    final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
         configManager.getNodeManager().getRegisteredDataNodeLocations();
-    DataNodeAsyncRequestContext<Object, TSStatus> clientHandler =
-        new DataNodeAsyncRequestContext<>(CnToDnAsyncRequestType.CLEAR_CACHE, dataNodeLocationMap);
+    final DataNodeAsyncRequestContext<Set<Integer>, TSStatus> clientHandler =
+        new DataNodeAsyncRequestContext<>(
+            CnToDnAsyncRequestType.CLEAR_CACHE, clearCacheOptions, dataNodeLocationMap);
     CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
     return clientHandler.getResponseList();
   }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java

@@ -925,12 +925,12 @@ public TSStatus flush(TFlushReq req) throws TException {
   }
 
   @Override
-  public TSStatus clearCache() {
-    return configManager.clearCache();
+  public TSStatus clearCache(final Set<Integer> clearCacheOptions) {
+    return configManager.clearCache(clearCacheOptions);
   }
 
   @Override
-  public TSStatus setConfiguration(TSetConfigurationReq req) throws TException {
+  public TSStatus setConfiguration(TSetConfigurationReq req) {
     return configManager.setConfiguration(req);
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java

@@ -1044,6 +1044,14 @@ public class IoTDBConfig {
 
   private int dataNodeTableCacheSemaphorePermitNum = 5;
 
+  /** GRASS Service */
+  private long generalRegionAttributeSecurityServiceIntervalSeconds = 30L;
+
+  private long generalRegionAttributeSecurityServiceTimeoutSeconds = 20L;
+  private long generalRegionAttributeSecurityServiceFailureDurationSecondsToFetch = 600L;
+  private int generalRegionAttributeSecurityServiceFailureTimesToFetch = 20;
+  private long detailContainerMinDegradeMemoryInBytes = 1024 * 1024L;
+
   private String readConsistencyLevel = "strong";
 
   /** Maximum execution time of a DriverTask */
@@ -3465,6 +3473,55 @@ public void setDataNodeTableCacheSemaphorePermitNum(int dataNodeTableCacheSemaph
     this.dataNodeTableCacheSemaphorePermitNum = dataNodeTableCacheSemaphorePermitNum;
   }
 
+  public long getGeneralRegionAttributeSecurityServiceIntervalSeconds() {
+    return generalRegionAttributeSecurityServiceIntervalSeconds;
+  }
+
+  public void setGeneralRegionAttributeSecurityServiceIntervalSeconds(
+      long generalRegionAttributeSecurityServiceIntervalSeconds) {
+    this.generalRegionAttributeSecurityServiceIntervalSeconds =
+        generalRegionAttributeSecurityServiceIntervalSeconds;
+  }
+
+  public long getGeneralRegionAttributeSecurityServiceTimeoutSeconds() {
+    return generalRegionAttributeSecurityServiceTimeoutSeconds;
+  }
+
+  public void setGeneralRegionAttributeSecurityServiceTimeoutSeconds(
+      long generalRegionAttributeSecurityServiceTimeoutSeconds) {
+    this.generalRegionAttributeSecurityServiceTimeoutSeconds =
+        generalRegionAttributeSecurityServiceTimeoutSeconds;
+  }
+
+  public long getGeneralRegionAttributeSecurityServiceFailureDurationSecondsToFetch() {
+    return generalRegionAttributeSecurityServiceFailureDurationSecondsToFetch;
+  }
+
+  public void setGeneralRegionAttributeSecurityServiceFailureDurationSecondsToFetch(
+      long generalRegionAttributeSecurityServiceFailureDurationSecondsToFetch) {
+    this.generalRegionAttributeSecurityServiceFailureDurationSecondsToFetch =
+        generalRegionAttributeSecurityServiceFailureDurationSecondsToFetch;
+  }
+
+  public int getGeneralRegionAttributeSecurityServiceFailureTimesToFetch() {
+    return generalRegionAttributeSecurityServiceFailureTimesToFetch;
+  }
+
+  public void setGeneralRegionAttributeSecurityServiceFailureTimesToFetch(
+      int generalRegionAttributeSecurityServiceFailureTimesToFetch) {
+    this.generalRegionAttributeSecurityServiceFailureTimesToFetch =
+        generalRegionAttributeSecurityServiceFailureTimesToFetch;
+  }
+
+  public long getDetailContainerMinDegradeMemoryInBytes() {
+    return detailContainerMinDegradeMemoryInBytes;
+  }
+
+  public void setDetailContainerMinDegradeMemoryInBytes(
+      long detailContainerMinDegradeMemoryInBytes) {
+    this.detailContainerMinDegradeMemoryInBytes = detailContainerMinDegradeMemoryInBytes;
+  }
+
   public String getReadConsistencyLevel() {
     return readConsistencyLevel;
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java

@@ -1076,6 +1076,39 @@ public void loadProperties(Properties properties) throws BadNodeUrlException, IO
                 "datanode_table_cache_semaphore_permit_num",
                 String.valueOf(conf.getDataNodeTableCacheSemaphorePermitNum()))));
 
+    conf.setGeneralRegionAttributeSecurityServiceIntervalSeconds(
+        Long.parseLong(
+            properties.getProperty(
+                "general_region_attribute_security_service_interval_seconds",
+                String.valueOf(conf.getGeneralRegionAttributeSecurityServiceIntervalSeconds()))));
+
+    conf.setGeneralRegionAttributeSecurityServiceTimeoutSeconds(
+        Long.parseLong(
+            properties.getProperty(
+                "general_region_attribute_security_service_timeout_seconds",
+                String.valueOf(conf.getGeneralRegionAttributeSecurityServiceTimeoutSeconds()))));
+
+    conf.setGeneralRegionAttributeSecurityServiceFailureDurationSecondsToFetch(
+        Long.parseLong(
+            properties.getProperty(
+                "general_region_attribute_security_service_failure_duration_seconds_to_fetch",
+                String.valueOf(
+                    conf
+                        .getGeneralRegionAttributeSecurityServiceFailureDurationSecondsToFetch()))));
+
+    conf.setGeneralRegionAttributeSecurityServiceFailureTimesToFetch(
+        Integer.parseInt(
+            properties.getProperty(
+                "general_region_attribute_security_service_failure_times_to_fetch",
+                String.valueOf(
+                    conf.getGeneralRegionAttributeSecurityServiceFailureTimesToFetch()))));
+
+    conf.setDetailContainerMinDegradeMemoryInBytes(
+        Long.parseLong(
+            properties.getProperty(
+                "detail_container_min_degrade_memory_in_bytes",
+                String.valueOf(conf.getDetailContainerMinDegradeMemoryInBytes()))));
+
     loadIoTConsensusProps(properties);
     loadIoTConsensusV2Props(properties);
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaExecutionVisitor.java

@@ -33,7 +33,6 @@
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.WritePlanNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.read.TableDeviceAttributeUpdateNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.ActivateTemplateNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.AlterTimeSeriesNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.BatchActivateTemplateNode;
@@ -58,7 +57,10 @@
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedNonWritePlanNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedWritePlanNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeOperateSchemaQueueNode;
-import org.apache.iotdb.db.queryengine.plan.relational.planner.node.CreateOrUpdateTableDeviceNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeCommitUpdateNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeUpdateNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableNodeLocationAddNode;
 import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion;
 import org.apache.iotdb.db.schemaengine.schemaregion.write.req.ICreateAlignedTimeSeriesPlan;
 import org.apache.iotdb.db.schemaengine.schemaregion.write.req.ICreateTimeSeriesPlan;
@@ -580,7 +582,7 @@ public TSStatus visitDeleteLogicalView(DeleteLogicalViewNode node, ISchemaRegion
   }
 
   @Override
-  public TSStatus visitCreateTableDevice(
+  public TSStatus visitCreateOrUpdateTableDevice(
       final CreateOrUpdateTableDeviceNode node, final ISchemaRegion schemaRegion) {
     try {
       // todo implement storage for device of diverse data types
@@ -604,6 +606,30 @@ public TSStatus visitTableDeviceAttributeUpdate(
     }
   }
 
+  @Override
+  public TSStatus visitTableDeviceAttributeCommit(
+      final TableDeviceAttributeCommitUpdateNode node, final ISchemaRegion schemaRegion) {
+    try {
+      schemaRegion.commitUpdateAttribute(node);
+      return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
+    } catch (final MetadataException e) {
+      logger.error(e.getMessage(), e);
+      return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
+    }
+  }
+
+  @Override
+  public TSStatus visitTableNodeLocationAdd(
+      final TableNodeLocationAddNode node, final ISchemaRegion schemaRegion) {
+    try {
+      schemaRegion.addNodeLocation(node);
+      return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
+    } catch (final MetadataException e) {
+      logger.error(e.getMessage(), e);
+      return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
+    }
+  }
+
   @Override
   public TSStatus visitPipeEnrichedWritePlanNode(
       final PipeEnrichedWritePlanNode node, final ISchemaRegion schemaRegion) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java

@@ -33,6 +33,7 @@
 import org.apache.iotdb.db.queryengine.plan.planner.plan.FragmentInstance;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
 import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion;
+import org.apache.iotdb.db.schemaengine.schemaregion.attribute.update.GeneralRegionAttributeSecurityService;
 import org.apache.iotdb.db.tools.schema.SchemaRegionSnapshotParser;
 import org.apache.iotdb.rpc.TSStatusCode;
 
@@ -52,7 +53,7 @@ public class SchemaRegionStateMachine extends BaseStateMachine {
   private static final FragmentInstanceManager QUERY_INSTANCE_MANAGER =
       FragmentInstanceManager.getInstance();
 
-  public SchemaRegionStateMachine(ISchemaRegion schemaRegion) {
+  public SchemaRegionStateMachine(final ISchemaRegion schemaRegion) {
     this.schemaRegion = schemaRegion;
   }
 
@@ -65,10 +66,11 @@ public void start() {
   public void stop() {
     // Stop leader related service for schema pipe
     PipeDataNodeAgent.runtime().notifySchemaLeaderUnavailable(schemaRegion.getSchemaRegionId());
+    GeneralRegionAttributeSecurityService.getInstance().stopBroadcast(schemaRegion);
   }
 
   @Override
-  public void notifyLeaderChanged(ConsensusGroupId groupId, int newLeaderId) {
+  public void notifyLeaderChanged(final ConsensusGroupId groupId, final int newLeaderId) {
     if (newLeaderId != IoTDBDescriptor.getInstance().getConfig().getDataNodeId()) {
       logger.info(
           "Current node [nodeId: {}] is no longer the schema region leader [regionId: {}], "
@@ -81,7 +83,7 @@ public void notifyLeaderChanged(ConsensusGroupId groupId, int newLeaderId) {
 
   @Override
   public void notifyNotLeader() {
-    int dataNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
+    final int dataNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
     logger.info(
         "Current node [nodeId: {}] is no longer the schema region leader [regionId: {}], "
             + "start cleaning up related services.",
@@ -90,6 +92,7 @@ public void notifyNotLeader() {
 
     // Shutdown leader related service for schema pipe
     PipeDataNodeAgent.runtime().notifySchemaLeaderUnavailable(schemaRegion.getSchemaRegionId());
+    GeneralRegionAttributeSecurityService.getInstance().stopBroadcast(schemaRegion);
 
     logger.info(
         "Current node [nodeId: {}] is no longer the schema region leader [regionId: {}], "
@@ -107,6 +110,7 @@ public void notifyLeaderReady() {
 
     // Activate leader related service for schema pipe
     PipeDataNodeAgent.runtime().notifySchemaLeaderReady(schemaRegion.getSchemaRegionId());
+    GeneralRegionAttributeSecurityService.getInstance().startBroadcast(schemaRegion);
 
     logger.info(
         "Current node [nodeId: {}] as schema region leader [regionId: {}] is ready to work",
@@ -120,7 +124,7 @@ public boolean isReadOnly() {
   }
 
   @Override
-  public boolean takeSnapshot(File snapshotDir) {
+  public boolean takeSnapshot(final File snapshotDir) {
     if (schemaRegion.createSnapshot(snapshotDir)
         && PipeDataNodeAgent.runtime()
             .schemaListener(schemaRegion.getSchemaRegionId())
@@ -132,7 +136,7 @@ public boolean takeSnapshot(File snapshotDir) {
   }
 
   @Override
-  public void loadSnapshot(File latestSnapshotRootDir) {
+  public void loadSnapshot(final File latestSnapshotRootDir) {
     schemaRegion.loadSnapshot(latestSnapshotRootDir);
     PipeDataNodeAgent.runtime()
         .schemaListener(schemaRegion.getSchemaRegionId())
@@ -142,14 +146,14 @@ public void loadSnapshot(File latestSnapshotRootDir) {
     listen2Snapshot4PipeListener(false);
   }
 
-  public void listen2Snapshot4PipeListener(boolean isTmp) {
-    Pair<Path, Path> snapshotPaths =
+  public void listen2Snapshot4PipeListener(final boolean isTmp) {
+    final Pair<Path, Path> snapshotPaths =
         SchemaRegionSnapshotParser.getSnapshotPaths(
             Utils.fromConsensusGroupIdToRaftGroupId(schemaRegion.getSchemaRegionId())
                 .getUuid()
                 .toString(),
             isTmp);
-    SchemaRegionListeningQueue listener =
+    final SchemaRegionListeningQueue listener =
         PipeDataNodeAgent.runtime().schemaListener(schemaRegion.getSchemaRegionId());
     if (Objects.isNull(snapshotPaths) || Objects.isNull(snapshotPaths.getLeft())) {
       if (listener.isOpened()) {
@@ -169,27 +173,28 @@ public void listen2Snapshot4PipeListener(boolean isTmp) {
   }
 
   @Override
-  public TSStatus write(IConsensusRequest request) {
+  public TSStatus write(final IConsensusRequest request) {
     try {
-      TSStatus result = ((PlanNode) request).accept(new SchemaExecutionVisitor(), schemaRegion);
+      final TSStatus result =
+          ((PlanNode) request).accept(new SchemaExecutionVisitor(), schemaRegion);
       if (result.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
         PipeDataNodeAgent.runtime()
             .schemaListener(schemaRegion.getSchemaRegionId())
             .tryListenToNode((PlanNode) request);
       }
       return result;
-    } catch (IllegalArgumentException e) {
+    } catch (final IllegalArgumentException e) {
       logger.error(e.getMessage(), e);
       return new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
     }
   }
 
   @Override
-  public DataSet read(IConsensusRequest request) {
-    FragmentInstance fragmentInstance;
+  public DataSet read(final IConsensusRequest request) {
+    final FragmentInstance fragmentInstance;
     try {
       fragmentInstance = getFragmentInstance(request);
-    } catch (IllegalArgumentException e) {
+    } catch (final IllegalArgumentException e) {
       logger.error(e.getMessage());
       return null;
     }