distributed-v2/src/main/java/com/orientechnologies/orient/distributed/impl/ODistributedStorage.java
protected boolean executeOnlyLocally(final String localNodeName, final ODistributedConfiguration dbCfg,
      final OCommandExecutor exec, final Collection<String> involvedClusters, final Collection<String> nodes) {
    boolean executeLocally = false;
    if (exec.isIdempotent()) {
      final int availableNodes = nodes.size();

      // IDEMPOTENT: CHECK IF CAN WORK LOCALLY ONLY
      int maxReadQuorum;
      if (involvedClusters.isEmpty())
        maxReadQuorum = dbCfg.getReadQuorum(null, availableNodes, localNodeName);
      else {
        maxReadQuorum = 0;
        for (String cl : involvedClusters)
          maxReadQuorum = Math.max(maxReadQuorum, dbCfg.getReadQuorum(cl, availableNodes, localNodeName));
      }

      if (nodes.contains(localNodeName) && maxReadQuorum <= 1)
        executeLocally = true;
    }

    return executeLocally;
  }

  public boolean isLocalEnv() {
    return localDistributedDatabase == null || dManager == null || distributedConfiguration == null || OScenarioThreadLocal.INSTANCE
        .isRunModeDistributed();
  }

  public OStorageOperationResult<ORawBuffer> readRecord(final ORecordId iRecordId, final String iFetchPlan,
      final boolean iIgnoreCache, final boolean prefetchRecords, final ORecordCallback<ORawBuffer> iCallback) {

    if (isLocalEnv()) {
      // ALREADY DISTRIBUTED
      return wrapped.readRecord(iRecordId, iFetchPlan, iIgnoreCache, prefetchRecords, iCallback);
    }

    final ORawBuffer memCopy = localDistributedDatabase.getRecordIfLocked(iRecordId);
    if (memCopy != null)
      return new OStorageOperationResult<ORawBuffer>(memCopy);

    try {
      final String clusterName = getClusterNameByRID(iRecordId);

      final ODistributedConfiguration dbCfg = distributedConfiguration;
      final List<String> nodes = dbCfg.getServers(clusterName, null);
      final int availableNodes = nodes.size();

      // CHECK IF LOCAL NODE OWNS THE DATA AND READ-QUORUM = 1: GET IT LOCALLY BECAUSE IT'S FASTER
      final String localNodeName = dManager.getLocalNodeName();

      if (nodes.isEmpty()
          || nodes.contains(dManager.getLocalNodeName()) && dbCfg.getReadQuorum(clusterName, availableNodes, localNodeName) <= 1) {
        // DON'T REPLICATE
        return (OStorageOperationResult<ORawBuffer>) OScenarioThreadLocal.executeAsDistributed(new Callable() {
          @Override
          public Object call() throws Exception {
            return wrapped.readRecord(iRecordId, iFetchPlan, iIgnoreCache, prefetchRecords, iCallback);
          }
        });
      }

      final OReadRecordTask task = ((OReadRecordTask) dManager.getTaskFactoryManager().getFactoryByServerNames(nodes)
          .createTask(OReadRecordTask.FACTORYID)).init(iRecordId);

      // DISTRIBUTE IT
      final ODistributedResponse response = dManager
          .sendRequest(getName(), Collections.singleton(clusterName), nodes, task, dManager.getNextMessageIdCounter(),
              EXECUTION_MODE.RESPONSE, null, null, null);
      final Object dResult = response != null ? response.getPayload() : null;

      if (dResult instanceof ONeedRetryException)
        throw (ONeedRetryException) dResult;
      else if (dResult instanceof Exception)
        throw OException
            .wrapException(new ODistributedException("Error on execution distributed read record"), (Exception) dResult);

      return new OStorageOperationResult<ORawBuffer>((ORawBuffer) dResult);

    } catch (ONeedRetryException e) {
      // PASS THROUGH
      throw e;
    } catch (Exception e) {
      handleDistributedException("Cannot route read record operation for %s to the distributed node", e, iRecordId);
      // UNREACHABLE
      return null;
    }
  }

  @Override
  public OStorageOperationResult<ORawBuffer> readRecordIfVersionIsNotLatest(final ORecordId rid, final String fetchPlan,
      final boolean ignoreCache, final int recordVersion) throws ORecordNotFoundException {
    if (isLocalEnv()) {
      return wrapped.readRecordIfVersionIsNotLatest(rid, fetchPlan, ignoreCache, recordVersion);
    }
    final ORawBuffer memCopy = localDistributedDatabase.getRecordIfLocked(rid);
    if (memCopy != null)
      return new OStorageOperationResult<ORawBuffer>(memCopy);

    try {
      final String clusterName = getClusterNameByRID(rid);

      final ODistributedConfiguration dbCfg = distributedConfiguration;
      final List<String> nodes = dbCfg.getServers(clusterName, null);
      final int availableNodes = nodes.size();

      // CHECK IF LOCAL NODE OWNS THE DATA AND READ-QUORUM = 1: GET IT LOCALLY BECAUSE IT'S FASTER
      final String localNodeName = dManager.getLocalNodeName();

      if (nodes.isEmpty()
          || nodes.contains(dManager.getLocalNodeName()) && dbCfg.getReadQuorum(clusterName, availableNodes, localNodeName) <= 1) {
        // DON'T REPLICATE
        return (OStorageOperationResult<ORawBuffer>) OScenarioThreadLocal.executeAsDistributed(new Callable() {
          @Override
          public Object call() throws Exception {
            return wrapped.readRecordIfVersionIsNotLatest(rid, fetchPlan, ignoreCache, recordVersion);
          }
        });
      }

      final OReadRecordIfNotLatestTask task = (OReadRecordIfNotLatestTask) dManager.getTaskFactoryManager()
          .getFactoryByServerNames(nodes).createTask(OReadRecordIfNotLatestTask.FACTORYID);
      task.init(rid, recordVersion);

      // DISTRIBUTE IT
      final Object result = dManager
          .sendRequest(getName(), Collections.singleton(clusterName), nodes, task, dManager.getNextMessageIdCounter(),
              EXECUTION_MODE.RESPONSE, null, null, null).getPayload();

      if (result instanceof ONeedRetryException)
        throw (ONeedRetryException) result;
      else if (result instanceof Exception)
        throw OException.wrapException(new ODistributedException("Error on execution distributed read record"), (Exception) result);

      return new OStorageOperationResult<ORawBuffer>((ORawBuffer) result);

    } catch (ONeedRetryException e) {
      // PASS THROUGH
      throw e;
    } catch (Exception e) {
      handleDistributedException("Cannot route read record operation for %s to the distributed node", e, rid);
      // UNREACHABLE
      return null;
    }
  }

  @Override
  public OStorageOperationResult<Boolean> deleteRecord(final ORecordId iRecordId, final int iVersion, final int iMode,
      final ORecordCallback<Boolean> iCallback) {
    // IF is a real delete should be with a tx
    return wrapped.deleteRecord(iRecordId, iVersion, iMode, iCallback);
  }
protected boolean executeOnlyLocally(final String localNodeName, final ODistributedConfiguration dbCfg,
      final OCommandExecutor exec, final Collection<String> involvedClusters, final Collection<String> nodes) {
    boolean executeLocally = false;
    if (exec.isIdempotent()) {
      final int availableNodes = nodes.size();

      // IDEMPOTENT: CHECK IF CAN WORK LOCALLY ONLY
      int maxReadQuorum;
      if (involvedClusters.isEmpty())
        maxReadQuorum = dbCfg.getReadQuorum(null, availableNodes, localNodeName);
      else {
        maxReadQuorum = 0;
        for (String cl : involvedClusters)
          maxReadQuorum = Math.max(maxReadQuorum, dbCfg.getReadQuorum(cl, availableNodes, localNodeName));
      }

      if (nodes.contains(localNodeName) && maxReadQuorum <= 1)
        executeLocally = true;
    }

    return executeLocally;
  }

  public boolean isLocalEnv() {
    return localDistributedDatabase == null || dManager == null || distributedConfiguration == null || OScenarioThreadLocal.INSTANCE
        .isRunModeDistributed();
  }

  public OStorageOperationResult<ORawBuffer> readRecord(final ORecordId iRecordId, final String iFetchPlan,
      final boolean iIgnoreCache, final boolean prefetchRecords, final ORecordCallback<ORawBuffer> iCallback) {

    if (isLocalEnv()) {
      // ALREADY DISTRIBUTED
      return wrapped.readRecord(iRecordId, iFetchPlan, iIgnoreCache, prefetchRecords, iCallback);
    }

    final ORawBuffer memCopy = localDistributedDatabase.getRecordIfLocked(iRecordId);
    if (memCopy != null)
      return new OStorageOperationResult<ORawBuffer>(memCopy);

    try {
      final String clusterName = getClusterNameByRID(iRecordId);

      final ODistributedConfiguration dbCfg = distributedConfiguration;
      final List<String> nodes = dbCfg.getServers(clusterName, null);
      final int availableNodes = nodes.size();

      // CHECK IF LOCAL NODE OWNS THE DATA AND READ-QUORUM = 1: GET IT LOCALLY BECAUSE IT'S FASTER
      final String localNodeName = dManager.getLocalNodeName();

      if (nodes.isEmpty()
          || nodes.contains(dManager.getLocalNodeName()) && dbCfg.getReadQuorum(clusterName, availableNodes, localNodeName) <= 1) {
        // DON'T REPLICATE
        return (OStorageOperationResult<ORawBuffer>) OScenarioThreadLocal.executeAsDistributed(new Callable() {
          @Override
          public Object call() throws Exception {
            return wrapped.readRecord(iRecordId, iFetchPlan, iIgnoreCache, prefetchRecords, iCallback);
          }
        });
      }

      final OReadRecordTask task = ((OReadRecordTask) dManager.getTaskFactoryManager().getFactoryByServerNames(nodes)
          .createTask(OReadRecordTask.FACTORYID)).init(iRecordId);

      // DISTRIBUTE IT
      final ODistributedResponse response = dManager
          .sendRequest(getName(), Collections.singleton(clusterName), nodes, task, dManager.getNextMessageIdCounter(),
              EXECUTION_MODE.RESPONSE, null, null, null);
      final Object dResult = response != null ? response.getPayload() : null;

      if (dResult instanceof ONeedRetryException)
        throw (ONeedRetryException) dResult;
      else if (dResult instanceof Exception)
        throw OException
            .wrapException(new ODistributedException("Error on execution distributed read record"), (Exception) dResult);

      return new OStorageOperationResult<ORawBuffer>((ORawBuffer) dResult);

    } catch (ONeedRetryException e) {
      // PASS THROUGH
      throw e;
    } catch (Exception e) {
      handleDistributedException("Cannot route read record operation for %s to the distributed node", e, iRecordId);
      // UNREACHABLE
      return null;
    }
  }

  @Override
  public OStorageOperationResult<ORawBuffer> readRecordIfVersionIsNotLatest(final ORecordId rid, final String fetchPlan,
      final boolean ignoreCache, final int recordVersion) throws ORecordNotFoundException {
    if (isLocalEnv()) {
      return wrapped.readRecordIfVersionIsNotLatest(rid, fetchPlan, ignoreCache, recordVersion);
    }
    final ORawBuffer memCopy = localDistributedDatabase.getRecordIfLocked(rid);
    if (memCopy != null)
      return new OStorageOperationResult<ORawBuffer>(memCopy);

    try {
      final String clusterName = getClusterNameByRID(rid);

      final ODistributedConfiguration dbCfg = distributedConfiguration;
      final List<String> nodes = dbCfg.getServers(clusterName, null);
      final int availableNodes = nodes.size();

      // CHECK IF LOCAL NODE OWNS THE DATA AND READ-QUORUM = 1: GET IT LOCALLY BECAUSE IT'S FASTER
      final String localNodeName = dManager.getLocalNodeName();

      if (nodes.isEmpty()
          || nodes.contains(dManager.getLocalNodeName()) && dbCfg.getReadQuorum(clusterName, availableNodes, localNodeName) <= 1) {
        // DON'T REPLICATE
        return (OStorageOperationResult<ORawBuffer>) OScenarioThreadLocal.executeAsDistributed(new Callable() {
          @Override
          public Object call() throws Exception {
            return wrapped.readRecordIfVersionIsNotLatest(rid, fetchPlan, ignoreCache, recordVersion);
          }
        });
      }

      final OReadRecordIfNotLatestTask task = (OReadRecordIfNotLatestTask) dManager.getTaskFactoryManager()
          .getFactoryByServerNames(nodes).createTask(OReadRecordIfNotLatestTask.FACTORYID);
      task.init(rid, recordVersion);

      // DISTRIBUTE IT
      final Object result = dManager
          .sendRequest(getName(), Collections.singleton(clusterName), nodes, task, dManager.getNextMessageIdCounter(),
              EXECUTION_MODE.RESPONSE, null, null, null).getPayload();

      if (result instanceof ONeedRetryException)
        throw (ONeedRetryException) result;
      else if (result instanceof Exception)
        throw OException.wrapException(new ODistributedException("Error on execution distributed read record"), (Exception) result);

      return new OStorageOperationResult<ORawBuffer>((ORawBuffer) result);

    } catch (ONeedRetryException e) {
      // PASS THROUGH
      throw e;
    } catch (Exception e) {
      handleDistributedException("Cannot route read record operation for %s to the distributed node", e, rid);
      // UNREACHABLE
      return null;
    }
  }

  @Override
  public OStorageOperationResult<Boolean> deleteRecord(final ORecordId iRecordId, final int iVersion, final int iMode,
      final ORecordCallback<Boolean> iCallback) {
    // IF is a real delete should be with a tx
    return wrapped.deleteRecord(iRecordId, iVersion, iMode, iCallback);
  }
Only idempotent commands that don't involve any other node can be executed locally.