@Override
  public SettableFuture<ConfigTaskResult> alterLogicalView(
      final AlterLogicalViewStatement alterLogicalViewStatement, final MPPQueryContext context) {
    final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
    final CreateLogicalViewStatement createLogicalViewStatement = new CreateLogicalViewStatement();
    createLogicalViewStatement.setTargetPaths(alterLogicalViewStatement.getTargetPaths());
    createLogicalViewStatement.setSourcePaths(alterLogicalViewStatement.getSourcePaths());
    createLogicalViewStatement.setQueryStatement(alterLogicalViewStatement.getQueryStatement());

    final Analysis analysis = Analyzer.analyze(createLogicalViewStatement, context);
    analysis.setDatabaseName(context.getDatabaseName().orElse(null));
    if (analysis.isFailed()) {
      future.setException(
          new IoTDBException(
              analysis.getFailStatus().getMessage(), analysis.getFailStatus().getCode()));
      return future;
    }

    // Transform all Expressions into ViewExpressions.
    final TransformToViewExpressionVisitor transformToViewExpressionVisitor =
        new TransformToViewExpressionVisitor();
    final List<Expression> expressionList = alterLogicalViewStatement.getSourceExpressionList();
    final List<ViewExpression> viewExpressionList = new ArrayList<>();
    for (final Expression expression : expressionList) {
      viewExpressionList.add(transformToViewExpressionVisitor.process(expression, null));
    }

    final List<PartialPath> viewPathList = alterLogicalViewStatement.getTargetPathList();

    final ByteArrayOutputStream stream = new ByteArrayOutputStream();
    try {
      ReadWriteIOUtils.write(viewPathList.size(), stream);
      for (int i = 0; i < viewPathList.size(); i++) {
        viewPathList.get(i).serialize(stream);
        ViewExpression.serialize(viewExpressionList.get(i), stream);
      }
    } catch (final IOException e) {
      throw new RuntimeException(e);
    }

    final TAlterLogicalViewReq req =
        new TAlterLogicalViewReq(
            context.getQueryId().getId(), ByteBuffer.wrap(stream.toByteArray()));
    try (final ConfigNodeClient client =
        CLUSTER_DELETION_CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
      TSStatus tsStatus;
      do {
        try {
          tsStatus = client.alterLogicalView(req);
        } catch (final TTransportException e) {
          if (e.getType() == TTransportException.TIMED_OUT
              || e.getCause() instanceof SocketTimeoutException) {
            // time out mainly caused by slow execution, wait until
            tsStatus = RpcUtils.getStatus(TSStatusCode.OVERLAP_WITH_EXISTING_TASK);
          } else {
            throw e;
          }
        }
        // keep waiting until task ends
      } while (TSStatusCode.OVERLAP_WITH_EXISTING_TASK.getStatusCode() == tsStatus.getCode());

      if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
        if (tsStatus.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()) {
          future.setException(
              new BatchProcessException(tsStatus.subStatus.toArray(new TSStatus[0])));
        } else {
          future.setException(new IoTDBException(tsStatus.getMessage(), tsStatus.getCode()));
        }
      } else {
        future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
      }
      return future;
    } catch (final ClientManagerException | TException e) {
      future.setException(e);
      return future;
    }
  }
FIXME Consolidate this clientManager with the upper one.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java