private TSStatus executeStatementAndClassifyExceptions(final Statement statement) {
    long estimatedMemory = 0L;
    final double pipeReceiverActualToEstimatedMemoryRatio =
        PIPE_CONFIG.getPipeReceiverActualToEstimatedMemoryRatio();
    try {
      if (statement instanceof InsertBaseStatement) {
        estimatedMemory = ((InsertBaseStatement) statement).ramBytesUsed();
        allocatedMemoryBlock =
            PipeDataNodeResourceManager.memory()
                .forceAllocate((long) (estimatedMemory * pipeReceiverActualToEstimatedMemoryRatio));
      }

      final TSStatus result =
          executeStatementWithPermissionCheckAndRetryOnDataTypeMismatch(statement);
      if (result.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
          || result.getCode() == TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) {
        return result;
      } else {
        LOGGER.warn(
            "Receiver id = {}: Failure status encountered while executing statement {}: {}",
            receiverId.get(),
            statement,
            result);
        return statement.accept(STATEMENT_STATUS_VISITOR, result);
      }
    } catch (final PipeRuntimeOutOfMemoryCriticalException e) {
      final String message =
          String.format(
              "Temporarily out of memory when executing statement %s, Requested memory: %s, "
                  + "used memory: %s, free memory: %s, total non-floating memory: %s",
              statement,
              estimatedMemory * pipeReceiverActualToEstimatedMemoryRatio,
              PipeDataNodeResourceManager.memory().getUsedMemorySizeInBytes(),
              PipeDataNodeResourceManager.memory().getFreeMemorySizeInBytes(),
              PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes());
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("Receiver id = {}: {}", receiverId.get(), message, e);
      }
      return new TSStatus(
              TSStatusCode.PIPE_RECEIVER_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode())
          .setMessage(message);
    } catch (final Exception e) {
      LOGGER.warn(
          "Receiver id = {}: Exception encountered while executing statement {}: ",
          receiverId.get(),
          statement,
          e);
      return statement.accept(STATEMENT_EXCEPTION_VISITOR, e);
    } finally {
      if (Objects.nonNull(allocatedMemoryBlock)) {
        allocatedMemoryBlock.close();
        allocatedMemoryBlock = null;
      }
    }
  }
For {@link InsertRowsStatement} and {@link InsertMultiTabletsStatement}, the returned {@link
TSStatus} will use sub-status to record the endpoint for redirection. Each sub-status records
the redirection endpoint for one device path, and the order is the same as the order of the
device paths in the statement. However, this order is not guaranteed to be the same as in the
request. So for each sub-status which needs to redirect, we record the device path using the
message field.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java