Add extension point to allow custom merge super set computations

At Google we have performance issues with the Submitted Together REST
endpoint when topics span hundreds of projects. This extension point
allows us to distribute the load to compute the merge super set across
multiple server nodes and thus enhance the performance of the Submitted
Together REST endpoint.

In the multi-master setup at Google the load balancer distributes
requests to different nodes based on the project in the URL. This means
not all nodes have to load all repositories into cache which improves
performance a lot. However we still have performance problems with the
Submitted Together REST endpoint and for large topics which span several
hundreds of projects loading the list of changes that should be
submitted together can take several minutes. The problem is that the
request to the Submitted Together REST endpoint is not specific to one
project, but to include the predecessor change we have to open the
repository for all projects that are touched in the topics. This means
the one node node that handles the request has to open several hundreds
of repositories which is slow and also polutes the repository caches. To
fix this we want to implement the new extension point in such a way that
it distributes the load over multiple nodes that each have the necessary
repositories already cached. These nodes can do the computation in
parallel which should speed up the operation further.

Change-Id: I814498a7526cd5d818efed6cc000c03fc9bc5a09
Signed-off-by: Edwin Kempin <ekempin@google.com>

Documentation/dev-plugins.txt

@@ -483,6 +483,15 @@ plugins implementing this can modify commit message of the change being
 submitted by Rebase Always and Cherry Pick submit strategies as well as
 change being queried with COMMIT_FOOTERS option.
 
+[[merge-super-set-computation]]
+== Merge Super Set Computation
+
+The algorithm to compute the merge super set to detect changes that
+should be submitted together can be customized by implementing
+`com.google.gerrit.server.git.MergeSuperSetComputation`.
+MergeSuperSetComputation is a DynamicItem, so Gerrit may only have one
+implementation.
+
 [[receive-pack]]
 == Receive Pack Initializers
 

gerrit-pgm/src/main/java/com/google/gerrit/pgm/Daemon.java

@@ -66,6 +66,7 @@ import com.google.gerrit.server.config.GerritServerConfig;
 import com.google.gerrit.server.config.RestCacheAdminModule;
 import com.google.gerrit.server.events.StreamEventsApiListener;
 import com.google.gerrit.server.git.GarbageCollectionModule;
+import com.google.gerrit.server.git.LocalMergeSuperSetComputation;
 import com.google.gerrit.server.git.SearchingChangeCacheImpl;
 import com.google.gerrit.server.git.WorkQueue;
 import com.google.gerrit.server.git.receive.ReceiveCommitsExecutorModule;
@@ -472,6 +473,7 @@ public class Daemon extends SiteProgram {
     if (testSysModule != null) {
       modules.add(testSysModule);
     }
+    modules.add(new LocalMergeSuperSetComputation.Module());
     return cfgInjector.createChildInjector(modules);
   }
 

gerrit-server/src/main/java/com/google/gerrit/server/config/GerritGlobalModule.java

@@ -114,6 +114,7 @@ import com.google.gerrit.server.git.ChangeMessageModifier;
 import com.google.gerrit.server.git.EmailMerge;
 import com.google.gerrit.server.git.GitModule;
 import com.google.gerrit.server.git.GitModules;
+import com.google.gerrit.server.git.MergeSuperSetComputation;
 import com.google.gerrit.server.git.MergeUtil;
 import com.google.gerrit.server.git.MergedByPushOp;
 import com.google.gerrit.server.git.NotesBranchUtil;
@@ -377,6 +378,7 @@ public class GerritGlobalModule extends FactoryModule {
     DynamicItem.itemOf(binder(), AccountPatchReviewStore.class);
     DynamicSet.setOf(binder(), AssigneeValidationListener.class);
     DynamicSet.setOf(binder(), ActionVisitor.class);
+    DynamicItem.itemOf(binder(), MergeSuperSetComputation.class);
 
     DynamicMap.mapOf(binder(), MailFilter.class);
     bind(MailFilter.class).annotatedWith(Exports.named("ListMailFilter")).to(ListMailFilter.class);

gerrit-server/src/main/java/com/google/gerrit/server/git/LocalMergeSuperSetComputation.java

@@ -0,0 +1,260 @@
+// Copyright (C) 2017 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.gerrit.server.git;
+
+import static com.google.common.base.Preconditions.checkState;
+
+import com.google.auto.value.AutoValue;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableListMultimap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterables;
+import com.google.gerrit.common.data.SubmitTypeRecord;
+import com.google.gerrit.extensions.client.SubmitType;
+import com.google.gerrit.extensions.registration.DynamicItem;
+import com.google.gerrit.reviewdb.client.Branch;
+import com.google.gerrit.reviewdb.client.Project;
+import com.google.gerrit.reviewdb.server.ReviewDb;
+import com.google.gerrit.server.CurrentUser;
+import com.google.gerrit.server.git.MergeOpRepoManager.OpenRepo;
+import com.google.gerrit.server.permissions.ChangePermission;
+import com.google.gerrit.server.permissions.PermissionBackend;
+import com.google.gerrit.server.permissions.PermissionBackendException;
+import com.google.gerrit.server.project.NoSuchProjectException;
+import com.google.gerrit.server.query.change.ChangeData;
+import com.google.gerrit.server.query.change.InternalChangeQuery;
+import com.google.gwtorm.server.OrmException;
+import com.google.inject.AbstractModule;
+import com.google.inject.Inject;
+import com.google.inject.Provider;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import org.eclipse.jgit.lib.ObjectId;
+import org.eclipse.jgit.lib.Ref;
+import org.eclipse.jgit.revwalk.RevCommit;
+import org.eclipse.jgit.revwalk.RevSort;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Default implementation of MergeSuperSet that does the computation of the merge super set
+ * sequentially on the local Gerrit instance.
+ */
+public class LocalMergeSuperSetComputation implements MergeSuperSetComputation {
+  private static final Logger log = LoggerFactory.getLogger(LocalMergeSuperSetComputation.class);
+
+  public static class Module extends AbstractModule {
+    @Override
+    protected void configure() {
+      DynamicItem.bind(binder(), MergeSuperSetComputation.class)
+          .to(LocalMergeSuperSetComputation.class);
+    }
+  }
+
+  @AutoValue
+  abstract static class QueryKey {
+    private static QueryKey create(Branch.NameKey branch, Iterable<String> hashes) {
+      return new AutoValue_LocalMergeSuperSetComputation_QueryKey(
+          branch, ImmutableSet.copyOf(hashes));
+    }
+
+    abstract Branch.NameKey branch();
+
+    abstract ImmutableSet<String> hashes();
+  }
+
+  private final PermissionBackend permissionBackend;
+  private final Provider<InternalChangeQuery> queryProvider;
+  private final Map<QueryKey, List<ChangeData>> queryCache;
+  private final Map<Branch.NameKey, Optional<RevCommit>> heads;
+
+  @Inject
+  LocalMergeSuperSetComputation(
+      PermissionBackend permissionBackend, Provider<InternalChangeQuery> queryProvider) {
+    this.permissionBackend = permissionBackend;
+    this.queryProvider = queryProvider;
+    this.queryCache = new HashMap<>();
+    this.heads = new HashMap<>();
+  }
+
+  @Override
+  public ChangeSet completeWithoutTopic(
+      ReviewDb db, MergeOpRepoManager orm, ChangeSet changeSet, CurrentUser user)
+      throws OrmException, IOException, PermissionBackendException {
+    Collection<ChangeData> visibleChanges = new ArrayList<>();
+    Collection<ChangeData> nonVisibleChanges = new ArrayList<>();
+
+    // For each target branch we run a separate rev walk to find open changes
+    // reachable from changes already in the merge super set.
+    ImmutableListMultimap<Branch.NameKey, ChangeData> bc =
+        byBranch(Iterables.concat(changeSet.changes(), changeSet.nonVisibleChanges()));
+    for (Branch.NameKey b : bc.keySet()) {
+      OpenRepo or = getRepo(orm, b.getParentKey());
+      List<RevCommit> visibleCommits = new ArrayList<>();
+      List<RevCommit> nonVisibleCommits = new ArrayList<>();
+      for (ChangeData cd : bc.get(b)) {
+        boolean visible = isVisible(db, changeSet, cd, user);
+
+        if (submitType(cd) == SubmitType.CHERRY_PICK) {
+          if (visible) {
+            visibleChanges.add(cd);
+          } else {
+            nonVisibleChanges.add(cd);
+          }
+
+          continue;
+        }
+
+        // Get the underlying git commit object
+        String objIdStr = cd.currentPatchSet().getRevision().get();
+        RevCommit commit = or.rw.parseCommit(ObjectId.fromString(objIdStr));
+
+        // Always include the input, even if merged. This allows
+        // SubmitStrategyOp to correct the situation later, assuming it gets
+        // returned by byCommitsOnBranchNotMerged below.
+        if (visible) {
+          visibleCommits.add(commit);
+        } else {
+          nonVisibleCommits.add(commit);
+        }
+      }
+
+      Set<String> visibleHashes =
+          walkChangesByHashes(visibleCommits, Collections.emptySet(), or, b);
+      Iterables.addAll(visibleChanges, byCommitsOnBranchNotMerged(or, db, b, visibleHashes));
+
+      Set<String> nonVisibleHashes = walkChangesByHashes(nonVisibleCommits, visibleHashes, or, b);
+      Iterables.addAll(nonVisibleChanges, byCommitsOnBranchNotMerged(or, db, b, nonVisibleHashes));
+    }
+
+    return new ChangeSet(visibleChanges, nonVisibleChanges);
+  }
+
+  private static ImmutableListMultimap<Branch.NameKey, ChangeData> byBranch(
+      Iterable<ChangeData> changes) throws OrmException {
+    ImmutableListMultimap.Builder<Branch.NameKey, ChangeData> builder =
+        ImmutableListMultimap.builder();
+    for (ChangeData cd : changes) {
+      builder.put(cd.change().getDest(), cd);
+    }
+    return builder.build();
+  }
+
+  private OpenRepo getRepo(MergeOpRepoManager orm, Project.NameKey project) throws IOException {
+    try {
+      OpenRepo or = orm.getRepo(project);
+      checkState(or.rw.hasRevSort(RevSort.TOPO));
+      return or;
+    } catch (NoSuchProjectException e) {
+      throw new IOException(e);
+    }
+  }
+
+  private boolean isVisible(ReviewDb db, ChangeSet changeSet, ChangeData cd, CurrentUser user)
+      throws PermissionBackendException {
+    boolean visible = changeSet.ids().contains(cd.getId());
+    if (visible
+        && !permissionBackend.user(user).change(cd).database(db).test(ChangePermission.READ)) {
+      // We thought the change was visible, but it isn't.
+      // This can happen if the ACL changes during the
+      // completeChangeSet computation, for example.
+      visible = false;
+    }
+    return visible;
+  }
+
+  private SubmitType submitType(ChangeData cd) throws OrmException {
+    SubmitTypeRecord str = cd.submitTypeRecord();
+    if (!str.isOk()) {
+      logErrorAndThrow("Failed to get submit type for " + cd.getId() + ": " + str.errorMessage);
+    }
+    return str.type;
+  }
+
+  private List<ChangeData> byCommitsOnBranchNotMerged(
+      OpenRepo or, ReviewDb db, Branch.NameKey branch, Set<String> hashes)
+      throws OrmException, IOException {
+    if (hashes.isEmpty()) {
+      return ImmutableList.of();
+    }
+    QueryKey k = QueryKey.create(branch, hashes);
+    List<ChangeData> cached = queryCache.get(k);
+    if (cached != null) {
+      return cached;
+    }
+
+    List<ChangeData> result = new ArrayList<>();
+    Iterable<ChangeData> destChanges =
+        MergeSuperSet.query(queryProvider.get())
+            .byCommitsOnBranchNotMerged(or.repo, db, branch, hashes);
+    for (ChangeData chd : destChanges) {
+      result.add(chd);
+    }
+    queryCache.put(k, result);
+    return result;
+  }
+
+  private Set<String> walkChangesByHashes(
+      Collection<RevCommit> sourceCommits, Set<String> ignoreHashes, OpenRepo or, Branch.NameKey b)
+      throws IOException {
+    Set<String> destHashes = new HashSet<>();
+    or.rw.reset();
+    markHeadUninteresting(or, b);
+    for (RevCommit c : sourceCommits) {
+      String name = c.name();
+      if (ignoreHashes.contains(name)) {
+        continue;
+      }
+      destHashes.add(name);
+      or.rw.markStart(c);
+    }
+    for (RevCommit c : or.rw) {
+      String name = c.name();
+      if (ignoreHashes.contains(name)) {
+        continue;
+      }
+      destHashes.add(name);
+    }
+
+    return destHashes;
+  }
+
+  private void markHeadUninteresting(OpenRepo or, Branch.NameKey b) throws IOException {
+    Optional<RevCommit> head = heads.get(b);
+    if (head == null) {
+      Ref ref = or.repo.getRefDatabase().exactRef(b.get());
+      head = ref != null ? Optional.of(or.rw.parseCommit(ref.getObjectId())) : Optional.empty();
+      heads.put(b, head);
+    }
+    if (head.isPresent()) {
+      or.rw.markUninteresting(head.get());
+    }
+  }
+
+  private void logErrorAndThrow(String msg) throws OrmException {
+    if (log.isErrorEnabled()) {
+      log.error(msg);
+    }
+    throw new OrmException(msg);
+  }
+}

gerrit-server/src/main/java/com/google/gerrit/server/git/MergeSuperSet.java

@@ -17,27 +17,18 @@ package com.google.gerrit.server.git;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 
-import com.google.auto.value.AutoValue;
 import com.google.common.base.Strings;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableListMultimap;
 import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Iterables;
-import com.google.gerrit.common.data.SubmitTypeRecord;
-import com.google.gerrit.extensions.client.SubmitType;
-import com.google.gerrit.reviewdb.client.Branch;
+import com.google.gerrit.extensions.registration.DynamicItem;
 import com.google.gerrit.reviewdb.client.Change;
-import com.google.gerrit.reviewdb.client.Project;
 import com.google.gerrit.reviewdb.server.ReviewDb;
 import com.google.gerrit.server.CurrentUser;
 import com.google.gerrit.server.change.Submit;
 import com.google.gerrit.server.config.GerritServerConfig;
-import com.google.gerrit.server.git.MergeOpRepoManager.OpenRepo;
 import com.google.gerrit.server.index.change.ChangeField;
 import com.google.gerrit.server.permissions.ChangePermission;
 import com.google.gerrit.server.permissions.PermissionBackend;
 import com.google.gerrit.server.permissions.PermissionBackendException;
-import com.google.gerrit.server.project.NoSuchProjectException;
 import com.google.gerrit.server.query.change.ChangeData;
 import com.google.gerrit.server.query.change.InternalChangeQuery;
 import com.google.gwtorm.server.OrmException;
@@ -45,21 +36,10 @@ import com.google.inject.Inject;
 import com.google.inject.Provider;
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
-import java.util.Optional;
 import java.util.Set;
 import org.eclipse.jgit.lib.Config;
-import org.eclipse.jgit.lib.ObjectId;
-import org.eclipse.jgit.lib.Ref;
-import org.eclipse.jgit.revwalk.RevCommit;
-import org.eclipse.jgit.revwalk.RevSort;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * Calculates the minimal superset of changes required to be merged.
@@ -72,10 +52,9 @@ import org.slf4j.LoggerFactory;
  * included.
  */
 public class MergeSuperSet {
-  private static final Logger log = LoggerFactory.getLogger(MergeSuperSet.class);
 
   public static void reloadChanges(ChangeSet changeSet) throws OrmException {
-    // Clear exactly the fields requested by query() below.
+    // Clear exactly the fields requested by query(InternalChangeQuery) below.
     for (ChangeData cd : changeSet.changes()) {
       cd.reloadChange();
       cd.setPatchSets(null);
@@ -83,24 +62,26 @@ public class MergeSuperSet {
     }
   }
 
-  @AutoValue
-  abstract static class QueryKey {
-    private static QueryKey create(Branch.NameKey branch, Iterable<String> hashes) {
-      return new AutoValue_MergeSuperSet_QueryKey(branch, ImmutableSet.copyOf(hashes));
-    }
-
-    abstract Branch.NameKey branch();
-
-    abstract ImmutableSet<String> hashes();
+  public static InternalChangeQuery query(InternalChangeQuery q) {
+    // Request fields required for completing the ChangeSet and converting to
+    // ChangeInfo without having to touch the database or opening the repository
+    // more than necessary. This provides reasonable performance when loading
+    // the change screen; callers that care about reading the latest value of
+    // these fields should clear them explicitly using reloadChanges().
+    Set<String> fields =
+        ImmutableSet.of(
+            ChangeField.CHANGE.getName(),
+            ChangeField.PATCH_SET.getName(),
+            ChangeField.MERGEABLE.getName());
+    return q.setRequestedFields(fields);
   }
 
   private final ChangeData.Factory changeDataFactory;
   private final Provider<InternalChangeQuery> queryProvider;
   private final Provider<MergeOpRepoManager> repoManagerProvider;
+  private final DynamicItem<MergeSuperSetComputation> mergeSuperSetComputation;
   private final PermissionBackend permissionBackend;
   private final Config cfg;
-  private final Map<QueryKey, List<ChangeData>> queryCache;
-  private final Map<Branch.NameKey, Optional<RevCommit>> heads;
 
   private MergeOpRepoManager orm;
   private boolean closeOrm;
@@ -111,14 +92,14 @@ public class MergeSuperSet {
       ChangeData.Factory changeDataFactory,
       Provider<InternalChangeQuery> queryProvider,
       Provider<MergeOpRepoManager> repoManagerProvider,
+      DynamicItem<MergeSuperSetComputation> mergeSuperSetComputation,
       PermissionBackend permissionBackend) {
     this.cfg = cfg;
     this.changeDataFactory = changeDataFactory;
     this.queryProvider = queryProvider;
     this.repoManagerProvider = repoManagerProvider;
+    this.mergeSuperSetComputation = mergeSuperSetComputation;
     this.permissionBackend = permissionBackend;
-    queryCache = new HashMap<>();
-    heads = new HashMap<>();
   }
 
   public MergeSuperSet setMergeOpRepoManager(MergeOpRepoManager orm) {
@@ -131,6 +112,11 @@ public class MergeSuperSet {
   public ChangeSet completeChangeSet(ReviewDb db, Change change, CurrentUser user)
       throws IOException, OrmException, PermissionBackendException {
     try {
+      if (orm == null) {
+        orm = repoManagerProvider.get();
+        closeOrm = true;
+      }
+
       ChangeData cd = changeDataFactory.create(db, change.getProject(), change.getId());
       ChangeSet changeSet =
           new ChangeSet(
@@ -138,7 +124,7 @@ public class MergeSuperSet {
       if (Submit.wholeTopicEnabled(cfg)) {
         return completeChangeSetIncludingTopics(db, changeSet, user);
       }
-      return completeChangeSetWithoutTopic(db, changeSet, user);
+      return mergeSuperSetComputation.get().completeWithoutTopic(db, orm, changeSet, user);
     } finally {
       if (closeOrm && orm != null) {
         orm.close();
@@ -147,161 +133,13 @@ public class MergeSuperSet {
     }
   }
 
-  private SubmitType submitType(ChangeData cd) throws OrmException {
-    SubmitTypeRecord str = cd.submitTypeRecord();
-    if (!str.isOk()) {
-      logErrorAndThrow("Failed to get submit type for " + cd.getId() + ": " + str.errorMessage);
-    }
-    return str.type;
-  }
-
-  private static ImmutableListMultimap<Branch.NameKey, ChangeData> byBranch(
-      Iterable<ChangeData> changes) throws OrmException {
-    ImmutableListMultimap.Builder<Branch.NameKey, ChangeData> builder =
-        ImmutableListMultimap.builder();
-    for (ChangeData cd : changes) {
-      builder.put(cd.change().getDest(), cd);
-    }
-    return builder.build();
-  }
-
-  private Set<String> walkChangesByHashes(
-      Collection<RevCommit> sourceCommits, Set<String> ignoreHashes, OpenRepo or, Branch.NameKey b)
-      throws IOException {
-    Set<String> destHashes = new HashSet<>();
-    or.rw.reset();
-    markHeadUninteresting(or, b);
-    for (RevCommit c : sourceCommits) {
-      String name = c.name();
-      if (ignoreHashes.contains(name)) {
-        continue;
-      }
-      destHashes.add(name);
-      or.rw.markStart(c);
-    }
-    for (RevCommit c : or.rw) {
-      String name = c.name();
-      if (ignoreHashes.contains(name)) {
-        continue;
-      }
-      destHashes.add(name);
-    }
-
-    return destHashes;
-  }
-
-  private ChangeSet completeChangeSetWithoutTopic(
-      ReviewDb db, ChangeSet changeSet, CurrentUser user)
-      throws IOException, OrmException, PermissionBackendException {
-    Collection<ChangeData> visibleChanges = new ArrayList<>();
-    Collection<ChangeData> nonVisibleChanges = new ArrayList<>();
-
-    // For each target branch we run a separate rev walk to find open changes
-    // reachable from changes already in the merge super set.
-    ImmutableListMultimap<Branch.NameKey, ChangeData> bc =
-        byBranch(Iterables.concat(changeSet.changes(), changeSet.nonVisibleChanges()));
-    for (Branch.NameKey b : bc.keySet()) {
-      OpenRepo or = getRepo(b.getParentKey());
-      List<RevCommit> visibleCommits = new ArrayList<>();
-      List<RevCommit> nonVisibleCommits = new ArrayList<>();
-      for (ChangeData cd : bc.get(b)) {
-        boolean visible = changeSet.ids().contains(cd.getId());
-        if (visible && !canRead(db, user, cd)) {
-          // We thought the change was visible, but it isn't.
-          // This can happen if the ACL changes during the
-          // completeChangeSet computation, for example.
-          visible = false;
-        }
-
-        if (submitType(cd) == SubmitType.CHERRY_PICK) {
-          if (visible) {
-            visibleChanges.add(cd);
-          } else {
-            nonVisibleChanges.add(cd);
-          }
-
-          continue;
-        }
-
-        // Get the underlying git commit object
-        String objIdStr = cd.currentPatchSet().getRevision().get();
-        RevCommit commit = or.rw.parseCommit(ObjectId.fromString(objIdStr));
-
-        // Always include the input, even if merged. This allows
-        // SubmitStrategyOp to correct the situation later, assuming it gets
-        // returned by byCommitsOnBranchNotMerged below.
-        if (visible) {
-          visibleCommits.add(commit);
-        } else {
-          nonVisibleCommits.add(commit);
-        }
-      }
-
-      Set<String> visibleHashes =
-          walkChangesByHashes(visibleCommits, Collections.emptySet(), or, b);
-      Iterables.addAll(visibleChanges, byCommitsOnBranchNotMerged(or, db, b, visibleHashes));
-
-      Set<String> nonVisibleHashes = walkChangesByHashes(nonVisibleCommits, visibleHashes, or, b);
-      Iterables.addAll(nonVisibleChanges, byCommitsOnBranchNotMerged(or, db, b, nonVisibleHashes));
-    }
-
-    return new ChangeSet(visibleChanges, nonVisibleChanges);
-  }
-
-  private OpenRepo getRepo(Project.NameKey project) throws IOException {
-    if (orm == null) {
-      orm = repoManagerProvider.get();
-      closeOrm = true;
-    }
-    try {
-      OpenRepo or = orm.getRepo(project);
-      checkState(or.rw.hasRevSort(RevSort.TOPO));
-      return or;
-    } catch (NoSuchProjectException e) {
-      throw new IOException(e);
-    }
-  }
-
-  private void markHeadUninteresting(OpenRepo or, Branch.NameKey b) throws IOException {
-    Optional<RevCommit> head = heads.get(b);
-    if (head == null) {
-      Ref ref = or.repo.getRefDatabase().exactRef(b.get());
-      head = ref != null ? Optional.of(or.rw.parseCommit(ref.getObjectId())) : Optional.empty();
-      heads.put(b, head);
-    }
-    if (head.isPresent()) {
-      or.rw.markUninteresting(head.get());
-    }
-  }
-
-  private List<ChangeData> byCommitsOnBranchNotMerged(
-      OpenRepo or, ReviewDb db, Branch.NameKey branch, Set<String> hashes)
-      throws OrmException, IOException {
-    if (hashes.isEmpty()) {
-      return ImmutableList.of();
-    }
-    QueryKey k = QueryKey.create(branch, hashes);
-    List<ChangeData> cached = queryCache.get(k);
-    if (cached != null) {
-      return cached;
-    }
-
-    List<ChangeData> result = new ArrayList<>();
-    Iterable<ChangeData> destChanges =
-        query().byCommitsOnBranchNotMerged(or.repo, db, branch, hashes);
-    for (ChangeData chd : destChanges) {
-      result.add(chd);
-    }
-    queryCache.put(k, result);
-    return result;
-  }
-
   /**
    * Completes {@code changeSet} with any additional changes from its topics
    *
    * <p>{@link #completeChangeSetIncludingTopics} calls this repeatedly, alternating with {@link
-   * #completeChangeSetWithoutTopic}, to discover what additional changes should be submitted with a
-   * change until the set stops growing.
+   * MergeSuperSetComputation#completeWithoutTopic(ReviewDb, MergeOpRepoManager, ChangeSet,
+   * CurrentUser)}, to discover what additional changes should be submitted with a change until the
+   * set stops growing.
    *
    * <p>{@code topicsSeen} and {@code visibleTopicsSeen} keep track of topics already explored to
    * avoid wasted work.
@@ -324,7 +162,7 @@ public class MergeSuperSet {
       if (Strings.isNullOrEmpty(topic) || visibleTopicsSeen.contains(topic)) {
         continue;
       }
-      for (ChangeData topicCd : query().byTopicOpen(topic)) {
+      for (ChangeData topicCd : byTopicOpen(topic)) {
         if (canRead(db, user, topicCd)) {
           visibleChanges.add(topicCd);
         } else {
@@ -340,7 +178,7 @@ public class MergeSuperSet {
       if (Strings.isNullOrEmpty(topic) || topicsSeen.contains(topic)) {
         continue;
       }
-      for (ChangeData topicCd : query().byTopicOpen(topic)) {
+      for (ChangeData topicCd : byTopicOpen(topic)) {
         nonVisibleChanges.add(topicCd);
       }
       topicsSeen.add(topic);
@@ -360,36 +198,15 @@ public class MergeSuperSet {
       oldSeen = seen;
 
       changeSet = topicClosure(db, changeSet, user, topicsSeen, visibleTopicsSeen);
-      changeSet = completeChangeSetWithoutTopic(db, changeSet, user);
+      changeSet = mergeSuperSetComputation.get().completeWithoutTopic(db, orm, changeSet, user);
 
       seen = topicsSeen.size() + visibleTopicsSeen.size();
     } while (seen != oldSeen);
     return changeSet;
   }
 
-  private InternalChangeQuery query() {
-    // Request fields required for completing the ChangeSet and converting to
-    // ChangeInfo without having to touch the database or opening the repository
-    // more than necessary. This provides reasonable performance when loading
-    // the change screen; callers that care about reading the latest value of
-    // these fields should clear them explicitly using reloadChanges().
-    Set<String> fields =
-        ImmutableSet.of(
-            ChangeField.CHANGE.getName(),
-            ChangeField.PATCH_SET.getName(),
-            ChangeField.MERGEABLE.getName());
-    return queryProvider.get().setRequestedFields(fields);
-  }
-
-  private void logError(String msg) {
-    if (log.isErrorEnabled()) {
-      log.error(msg);
-    }
-  }
-
-  private void logErrorAndThrow(String msg) throws OrmException {
-    logError(msg);
-    throw new OrmException(msg);
+  private List<ChangeData> byTopicOpen(String topic) throws OrmException {
+    return query(queryProvider.get()).byTopicOpen(topic);
   }
 
   private boolean canRead(ReviewDb db, CurrentUser user, ChangeData cd)

gerrit-server/src/main/java/com/google/gerrit/server/git/MergeSuperSetComputation.java

@@ -0,0 +1,53 @@
+// Copyright (C) 2017 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.gerrit.server.git;
+
+import com.google.gerrit.extensions.annotations.ExtensionPoint;
+import com.google.gerrit.reviewdb.server.ReviewDb;
+import com.google.gerrit.server.CurrentUser;
+import com.google.gerrit.server.permissions.PermissionBackendException;
+import com.google.gwtorm.server.OrmException;
+import java.io.IOException;
+
+/**
+ * Interface to compute the merge super set to detect changes that should be submitted together.
+ *
+ * <p>E.g. to speed up performance implementations could decide to do the computation in batches in
+ * parallel on different server nodes.
+ */
+@ExtensionPoint
+public interface MergeSuperSetComputation {
+
+  /**
+   * Compute the set of changes that should be submitted together. As input a set of changes is
+   * provided for which it is known that they should be submitted together. This method should
+   * complete the set by including open predecessor changes that need to be submitted as well. To
+   * decide whether open predecessor changes should be included the method must take the submit type
+   * into account (e.g. for changes with submit type "Cherry-Pick" open predecessor changes must not
+   * be included).
+   *
+   * <p>This method is invoked iteratively while new changes to be submitted together are discovered
+   * by expanding the topics of the changes. This method must not do any topic expansion on its own.
+   *
+   * @param db {@link ReviewDb} instance
+   * @param orm {@link MergeOpRepoManager} that should be used to access repositories
+   * @param changeSet A set of changes for which it is known that they should be submitted together
+   * @param user The user for which the visibility checks should be performed
+   * @return the completed set of changes that should be submitted together
+   */
+  ChangeSet completeWithoutTopic(
+      ReviewDb db, MergeOpRepoManager orm, ChangeSet changeSet, CurrentUser user)
+      throws OrmException, IOException, PermissionBackendException;
+}

gerrit-server/src/test/java/com/google/gerrit/testutil/InMemoryModule.java

@@ -47,6 +47,7 @@ import com.google.gerrit.server.config.TrackingFooters;
 import com.google.gerrit.server.config.TrackingFootersProvider;
 import com.google.gerrit.server.git.GarbageCollection;
 import com.google.gerrit.server.git.GitRepositoryManager;
+import com.google.gerrit.server.git.LocalMergeSuperSetComputation;
 import com.google.gerrit.server.git.PerThreadRequestScope;
 import com.google.gerrit.server.git.SearchingChangeCacheImpl;
 import com.google.gerrit.server.git.SendEmailExecutor;
@@ -202,7 +203,7 @@ public class InMemoryModule extends FactoryModule {
             return CanonicalWebUrlProvider.class;
           }
         });
-    //Replacement of DiffExecutorModule to not use thread pool in the tests
+    // Replacement of DiffExecutorModule to not use thread pool in the tests
     install(
         new AbstractModule() {
           @Override
@@ -220,6 +221,7 @@ public class InMemoryModule extends FactoryModule {
     install(new SignedTokenEmailTokenVerifier.Module());
     install(new GpgModule(cfg));
     install(new InMemoryAccountPatchReviewStore.Module());
+    install(new LocalMergeSuperSetComputation.Module());
 
     bind(AllAccountsIndexer.class).toProvider(Providers.of(null));
     bind(AllChangesIndexer.class).toProvider(Providers.of(null));

gerrit-war/src/main/java/com/google/gerrit/httpd/WebAppInitializer.java

@@ -50,6 +50,7 @@ import com.google.gerrit.server.config.SitePath;
 import com.google.gerrit.server.events.StreamEventsApiListener;
 import com.google.gerrit.server.git.GarbageCollectionModule;
 import com.google.gerrit.server.git.GitRepositoryManagerModule;
+import com.google.gerrit.server.git.LocalMergeSuperSetComputation;
 import com.google.gerrit.server.git.SearchingChangeCacheImpl;
 import com.google.gerrit.server.git.WorkQueue;
 import com.google.gerrit.server.git.receive.ReceiveCommitsExecutorModule;
@@ -325,6 +326,7 @@ public class WebAppInitializer extends GuiceServletContextListener implements Fi
     modules.add(cfgInjector.getInstance(MailReceiver.Module.class));
     modules.add(new SmtpEmailSender.Module());
     modules.add(new SignedTokenEmailTokenVerifier.Module());
+    modules.add(new LocalMergeSuperSetComputation.Module());
 
     // Plugin module needs to be inserted *before* the index module.
     // There is the concept of LifecycleModule, in Gerrit's own extension