Fix the status code of fetching schema when memory is not enough

Author: shuwenwei

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/SchemaFetchScanOperator.java

@@ -222,11 +222,17 @@ private void prepareSchemaNodeIteratorForSerialize() {
       schemaNodeIteratorForSerialize = schemaTree.getIteratorForSerialize();
       baos = new PublicBAOS(DEFAULT_MAX_TSBLOCK_SIZE_IN_BYTES + EXTRA_SIZE_TO_AVOID_GROW);
       if (operatorContext != null) {
-        schemaTreeMemCost = schemaTree.ramBytesUsed();
+        long ramBytesUsed = schemaTree.ramBytesUsed();
         operatorContext
             .getInstanceContext()
             .getMemoryReservationContext()
-            .reserveMemoryCumulatively(schemaTreeMemCost);
+            .reserveMemoryCumulatively(ramBytesUsed);
+        // For temporary and independently counted memory, we need process it immediately
+        operatorContext
+            .getInstanceContext()
+            .getMemoryReservationContext()
+            .reserveMemoryImmediately();
+        this.schemaTreeMemCost = ramBytesUsed;
       }
     } catch (MetadataException e) {
       throw new SchemaExecutionException(e);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analyzer.java

@@ -45,7 +45,12 @@ public Analysis analyze(Statement statement) {
     long startTime = System.nanoTime();
     AnalyzeVisitor visitor = new AnalyzeVisitor(partitionFetcher, schemaFetcher);
     Analysis analysis = null;
-    context.setReserveMemoryForSchemaTreeFunc(context::reserveMemoryForFrontEnd);
+    context.setReserveMemoryForSchemaTreeFunc(
+        mem -> {
+          context.reserveMemoryForFrontEnd(mem);
+          // For temporary and independently counted memory, we need process it immediately
+          context.reserveMemoryForFrontEndImmediately();
+        });
     try {
       analysis = visitor.process(statement, context);
     } finally {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ClusterSchemaFetchExecutor.java

@@ -31,6 +31,7 @@
 import org.apache.iotdb.db.protocol.session.SessionManager;
 import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
 import org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree;
+import org.apache.iotdb.db.queryengine.exception.MemoryNotEnoughException;
 import org.apache.iotdb.db.queryengine.plan.Coordinator;
 import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
 import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
@@ -326,6 +327,8 @@ private void parseFetchedData(
         throw new RuntimeException(
             new MetadataException("Failed to fetch schema because of unrecognized data"));
       }
+    } catch (MemoryNotEnoughException e) {
+      throw e;
     } catch (Exception e) {
       throw new RuntimeException(e);
     }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ErrorHandlingUtils.java

@@ -163,7 +163,9 @@ private static TSStatus tryCatchQueryException(Exception e) {
           TSStatusCode.QUERY_NOT_ALLOWED, INFO_NOT_ALLOWED_IN_BATCH_ERROR + rootCause.getMessage());
     } else if (t instanceof RootFIPlacementException
         || t instanceof ReplicaSetUnreachableException
-        || t instanceof QuerySchemaFetchFailedException) {
+        || (t instanceof QuerySchemaFetchFailedException
+            && ((QuerySchemaFetchFailedException) t).getErrorCode()
+                != TSStatusCode.QUERY_EXECUTION_MEMORY_NOT_ENOUGH.getStatusCode())) {
       return RpcUtils.getStatus(TSStatusCode.PLAN_FAILED_NETWORK_PARTITION, rootCause.getMessage());
     } else if (t instanceof IoTDBException) {
       return Objects.nonNull(((IoTDBException) t).getStatus())