Allow output formatters to work in stream mode.

--
MOS_MIGRATED_REVID=108142169

13 files changed, 459 insertions, 226 deletions

diff --git a/src/main/java/com/google/devtools/build/lib/bazel/commands/FetchCommand.java b/src/main/java/com/google/devtools/build/lib/bazel/commands/FetchCommand.java
index c4a5dca0ef..6408276f99 100644
--- a/src/main/java/com/google/devtools/build/lib/bazel/commands/FetchCommand.java
+++ b/src/main/java/com/google/devtools/build/lib/bazel/commands/FetchCommand.java
@@ -22,6 +22,7 @@ import com.google.devtools.build.lib.events.Event;
 import com.google.devtools.build.lib.packages.Target;
 import com.google.devtools.build.lib.pkgcache.PackageCacheOptions;
 import com.google.devtools.build.lib.query2.AbstractBlazeQueryEnvironment;
+import com.google.devtools.build.lib.query2.engine.Callback;
 import com.google.devtools.build.lib.query2.engine.QueryEnvironment.Setting;
 import com.google.devtools.build.lib.query2.engine.QueryException;
 import com.google.devtools.build.lib.query2.engine.QueryExpression;
@@ -121,7 +122,13 @@ public final class FetchCommand implements BlazeCommand {
 
     // 2. Evaluate expression:
     try {
-      queryEnv.evaluateQuery(expr);
+      queryEnv.evaluateQuery(expr, new Callback<Target>() {
+        @Override
+        public void process(Iterable<Target> partialResult)
+            throws QueryException, InterruptedException {
+          // Throw away the result.
+        }
+      });
     } catch (QueryException | InterruptedException e) {
       // Keep consistent with reportBuildFileError()
       env.getReporter().handle(Event.error(e.getMessage()));
diff --git a/src/main/java/com/google/devtools/build/lib/query2/AbstractBlazeQueryEnvironment.java b/src/main/java/com/google/devtools/build/lib/query2/AbstractBlazeQueryEnvironment.java
index 11eef89f1d..b3e4341715 100644
--- a/src/main/java/com/google/devtools/build/lib/query2/AbstractBlazeQueryEnvironment.java
+++ b/src/main/java/com/google/devtools/build/lib/query2/AbstractBlazeQueryEnvironment.java
@@ -17,6 +17,7 @@ import com.google.common.base.Preconditions;
 import com.google.common.base.Predicate;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterables;
 import com.google.common.collect.Sets;
 import com.google.devtools.build.lib.cmdline.Label;
 import com.google.devtools.build.lib.cmdline.TargetParsingException;
@@ -31,11 +32,11 @@ import com.google.devtools.build.lib.pkgcache.PathPackageLocator;
 import com.google.devtools.build.lib.pkgcache.TargetPatternEvaluator;
 import com.google.devtools.build.lib.pkgcache.TransitivePackageLoader;
 import com.google.devtools.build.lib.profiler.AutoProfiler;
+import com.google.devtools.build.lib.query2.engine.Callback;
 import com.google.devtools.build.lib.query2.engine.QueryEnvironment;
 import com.google.devtools.build.lib.query2.engine.QueryEvalResult;
 import com.google.devtools.build.lib.query2.engine.QueryException;
 import com.google.devtools.build.lib.query2.engine.QueryExpression;
-import com.google.devtools.build.lib.query2.engine.QueryUtil;
 import com.google.devtools.build.lib.util.BinaryPredicate;
 import com.google.devtools.build.skyframe.WalkableGraph.WalkableGraphFactory;
 
@@ -45,6 +46,7 @@ import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Logger;
 
 import javax.annotation.Nullable;
@@ -136,10 +138,11 @@ public abstract class AbstractBlazeQueryEnvironment<T> implements QueryEnvironme
    * @throws QueryException if the evaluation failed and {@code --nokeep_going} was in
    *   effect
    */
-  public QueryEvalResult<T> evaluateQuery(QueryExpression expr)
+  public QueryEvalResult evaluateQuery(QueryExpression expr, final Callback<T> callback)
       throws QueryException, InterruptedException {
-    Set<T> resultNodes;
-    try (AutoProfiler p = AutoProfiler.logged("evaluating query", LOG)) {
+
+    final AtomicBoolean empty = new AtomicBoolean(true);
+    try (final AutoProfiler p = AutoProfiler.logged("evaluating query", LOG)) {
       resolvedTargetPatterns.clear();
 
       // In the --nokeep_going case, errors are reported in the order in which the patterns are
@@ -152,9 +155,15 @@ public abstract class AbstractBlazeQueryEnvironment<T> implements QueryEnvironme
         // Unfortunately, by evaluating the patterns in parallel, we lose some location information.
         throw new QueryException(expr, e.getMessage());
       }
-
       try {
-        resultNodes = QueryUtil.evalAll(this, expr);
+        this.eval(expr, new Callback<T>() {
+          @Override
+          public void process(Iterable<T> partialResult)
+              throws QueryException, InterruptedException {
+            empty.compareAndSet(true, Iterables.isEmpty(partialResult));
+            callback.process(partialResult);
+          }
+        });
       } catch (QueryException e) {
         throw new QueryException(e, expr);
       }
@@ -172,12 +181,12 @@ public abstract class AbstractBlazeQueryEnvironment<T> implements QueryEnvironme
       }
     }
 
-    return new QueryEvalResult<>(!eventHandler.hasErrors(), resultNodes);
+    return new QueryEvalResult(!eventHandler.hasErrors(), empty.get());
   }
 
-  public QueryEvalResult<T> evaluateQuery(String query)
+  public QueryEvalResult evaluateQuery(String query, Callback<T> callback)
       throws QueryException, InterruptedException {
-    return evaluateQuery(QueryExpression.parse(query, this));
+    return evaluateQuery(QueryExpression.parse(query, this), callback);
   }
 
   @Override
diff --git a/src/main/java/com/google/devtools/build/lib/query2/BlazeQueryEnvironment.java b/src/main/java/com/google/devtools/build/lib/query2/BlazeQueryEnvironment.java
index 55b1bc7be5..db006b540c 100644
--- a/src/main/java/com/google/devtools/build/lib/query2/BlazeQueryEnvironment.java
+++ b/src/main/java/com/google/devtools/build/lib/query2/BlazeQueryEnvironment.java
@@ -37,8 +37,8 @@ import com.google.devtools.build.lib.pkgcache.TargetEdgeObserver;
 import com.google.devtools.build.lib.pkgcache.TargetPatternEvaluator;
 import com.google.devtools.build.lib.pkgcache.TargetProvider;
 import com.google.devtools.build.lib.pkgcache.TransitivePackageLoader;
-import com.google.devtools.build.lib.query2.engine.BlazeQueryEvalResult;
 import com.google.devtools.build.lib.query2.engine.Callback;
+import com.google.devtools.build.lib.query2.engine.DigraphQueryEvalResult;
 import com.google.devtools.build.lib.query2.engine.QueryEvalResult;
 import com.google.devtools.build.lib.query2.engine.QueryException;
 import com.google.devtools.build.lib.query2.engine.QueryExpression;
@@ -54,6 +54,7 @@ import java.util.Iterator;
 import java.util.LinkedHashSet;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * The environment of a Blaze query. Not thread-safe.
@@ -102,15 +103,22 @@ public class BlazeQueryEnvironment extends AbstractBlazeQueryEnvironment<Target>
   }
 
   @Override
-  public BlazeQueryEvalResult<Target> evaluateQuery(QueryExpression expr)
-      throws QueryException, InterruptedException {
+  public DigraphQueryEvalResult<Target> evaluateQuery(QueryExpression expr,
+      final Callback<Target> callback) throws QueryException, InterruptedException {
     // Some errors are reported as QueryExceptions and others as ERROR events (if --keep_going). The
     // result is set to have an error iff there were errors emitted during the query, so we reset
     // errors here.
     eventHandler.resetErrors();
-    QueryEvalResult<Target> queryEvalResult = super.evaluateQuery(expr);
-    return new BlazeQueryEvalResult<>(queryEvalResult.getSuccess(), queryEvalResult.getResultSet(),
-        graph);
+    final AtomicBoolean empty = new AtomicBoolean(true);
+    QueryEvalResult queryEvalResult = super.evaluateQuery(expr, new Callback<Target>() {
+      @Override
+      public void process(Iterable<Target> partialResult)
+          throws QueryException, InterruptedException {
+        empty.compareAndSet(true, Iterables.isEmpty(partialResult));
+        callback.process(partialResult);
+      }
+    });
+    return new DigraphQueryEvalResult<>(queryEvalResult.getSuccess(), empty.get(), graph);
   }
 
   @Override
diff --git a/src/main/java/com/google/devtools/build/lib/query2/SkyQueryEnvironment.java b/src/main/java/com/google/devtools/build/lib/query2/SkyQueryEnvironment.java
index 54219cc3b1..21e999bc50 100644
--- a/src/main/java/com/google/devtools/build/lib/query2/SkyQueryEnvironment.java
+++ b/src/main/java/com/google/devtools/build/lib/query2/SkyQueryEnvironment.java
@@ -167,14 +167,14 @@ public class SkyQueryEnvironment extends AbstractBlazeQueryEnvironment<Target> {
   }
 
   @Override
-  public QueryEvalResult<Target> evaluateQuery(QueryExpression expr)
+  public QueryEvalResult evaluateQuery(QueryExpression expr, Callback<Target> callback)
       throws QueryException, InterruptedException {
     // Some errors are reported as QueryExceptions and others as ERROR events (if --keep_going). The
     // result is set to have an error iff there were errors emitted during the query, so we reset
     // errors here.
     eventHandler.resetErrors();
     init();
-    return super.evaluateQuery(expr);
+    return super.evaluateQuery(expr, callback);
   }
 
   private Map<Target, Collection<Target>> makeTargetsMap(Map<SkyKey, Iterable<SkyKey>> input) {
@@ -374,7 +374,7 @@ public class SkyQueryEnvironment extends AbstractBlazeQueryEnvironment<Target> {
     }
   }
 
-  private static Target getSubincludeTarget(final Label label, Package pkg) {
+  private static Target getSubincludeTarget(Label label, Package pkg) {
     return new FakeSubincludeTarget(label, pkg);
   }
 
diff --git a/src/main/java/com/google/devtools/build/lib/query2/engine/BlazeQueryEvalResult.java b/src/main/java/com/google/devtools/build/lib/query2/engine/DigraphQueryEvalResult.java
index 4ad13573f9..24612b25e3 100644
--- a/src/main/java/com/google/devtools/build/lib/query2/engine/BlazeQueryEvalResult.java
+++ b/src/main/java/com/google/devtools/build/lib/query2/engine/DigraphQueryEvalResult.java
@@ -17,20 +17,18 @@ package com.google.devtools.build.lib.query2.engine;
 import com.google.common.base.Preconditions;
 import com.google.devtools.build.lib.graph.Digraph;
 
-import java.util.Set;
-
 /** {@link QueryEvalResult} along with a digraph giving the structure of the results. */
-public class BlazeQueryEvalResult<T> extends QueryEvalResult<T> {
+public class DigraphQueryEvalResult<T> extends QueryEvalResult {
 
   private final Digraph<T> graph;
 
-  public BlazeQueryEvalResult(boolean success, Set<T> resultSet, Digraph<T> graph) {
-    super(success, resultSet);
+  public DigraphQueryEvalResult(boolean success, boolean isEmpty, Digraph<T> graph) {
+    super(success, isEmpty);
     this.graph = Preconditions.checkNotNull(graph);
   }
 
-  /** Returns the result as a directed graph over elements. */
-  public Digraph<T> getResultGraph() {
-    return graph.extractSubgraph(resultSet);
+  /** Returns the recorded graph */
+  public Digraph<T> getGraph() {
+    return graph;
   }
 }
diff --git a/src/main/java/com/google/devtools/build/lib/query2/engine/OutputFormatterCallback.java b/src/main/java/com/google/devtools/build/lib/query2/engine/OutputFormatterCallback.java
new file mode 100644
index 0000000000..546b0d51bb
--- /dev/null
+++ b/src/main/java/com/google/devtools/build/lib/query2/engine/OutputFormatterCallback.java
@@ -0,0 +1,93 @@
+// Copyright 2015 The Bazel Authors. All rights reserved.
+//
+// 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.devtools.build.lib.query2.engine;
+
+import com.google.devtools.build.lib.concurrent.ThreadSafety.ThreadCompatible;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import javax.annotation.Nullable;
+
+/** A callback that can receive a finish event when there are no more partial results */
+@ThreadCompatible
+public abstract class OutputFormatterCallback<T> implements Callback<T>, Closeable {
+
+  private IOException ioException;
+
+  /**
+   * This method will be called before any partial result are available.
+   *
+   * <p>It should be used for opening resources or sending a header to the output.
+   */
+  public void start() throws IOException { }
+
+  /**
+   * Same as start but for closing resources or writting a footer.
+   *
+   * <p>Will be called even in the case of an error.
+   */
+  @Override
+  public void close() throws IOException{}
+
+  /**
+   * Note that {@link Callback} interface does not throw IOExceptions. What this implementation does
+   * instead is throw {@code InterruptedException} and store the {@code IOException} in the {@code
+   * ioException} field. Users of this class should check on InterruptedException the field to
+   * disambiguate between real interruptions or IO Exceptions.
+   */
+  @Override
+  public final void process(Iterable<T> partialResult) throws QueryException, InterruptedException {
+    try {
+      processOutput(partialResult);
+    } catch (IOException e) {
+      ioException = e;
+      throw new InterruptedException("Interrupting due to a IOException in the OutputFormatter");
+    }
+  }
+
+  protected abstract void processOutput(Iterable<T> partialResult)
+      throws IOException, InterruptedException;
+
+  @Nullable
+  public IOException getIoException() {
+    return ioException;
+  }
+
+  /**
+   * Use an {@code OutputFormatterCallback} with an already computed set of targets. Note that this
+   * does not work in stream mode, as the {@code targets} would already be computed.
+   *
+   * <p>The intended usage of this method is to use {@code StreamedFormatter} formaters in non
+   * streaming contexts.
+   */
+  public static <T> void processAllTargets(OutputFormatterCallback<T> callback,
+      Iterable<T> targets) throws IOException, InterruptedException {
+    try {
+      callback.start();
+      callback.process(targets);
+    } catch (InterruptedException e) {
+      IOException ioException = callback.getIoException();
+      if (ioException != null) {
+        throw ioException;
+      }
+      throw e;
+    } catch (QueryException e) {
+      throw new IllegalStateException("This should not happen, as we are not running any query,"
+          + " only printing the results:" + e.getMessage(), e);
+    } finally {
+      callback.close();
+    }
+  }
+}
diff --git a/src/main/java/com/google/devtools/build/lib/query2/engine/QueryEvalResult.java b/src/main/java/com/google/devtools/build/lib/query2/engine/QueryEvalResult.java
index 23e77e6758..886cb439fe 100644
--- a/src/main/java/com/google/devtools/build/lib/query2/engine/QueryEvalResult.java
+++ b/src/main/java/com/google/devtools/build/lib/query2/engine/QueryEvalResult.java
@@ -14,24 +14,18 @@
 
 package com.google.devtools.build.lib.query2.engine;
 
-import com.google.common.base.Preconditions;
-
-import java.util.Set;
-
 /**
- * The result of a query evaluation, containing a set of elements.
- *
- * @param <T> the node type of the elements.
+ * Information about the query evaluation, like if it was successful and number of elements
+ * returned.
  */
-public class QueryEvalResult<T> {
+public class QueryEvalResult {
 
-  protected final boolean success;
-  protected final Set<T> resultSet;
+  private final boolean success;
+  private final boolean empty;
 
-  public QueryEvalResult(
-      boolean success, Set<T> resultSet) {
+  public QueryEvalResult(boolean success, boolean empty) {
     this.success = success;
-    this.resultSet = Preconditions.checkNotNull(resultSet);
+    this.empty = empty;
   }
 
   /**
@@ -42,16 +36,13 @@ public class QueryEvalResult<T> {
     return success;
   }
 
-  /**
-   * Returns the result as a set of targets.
-   */
-  public Set<T> getResultSet() {
-    return resultSet;
+  /** True if the query did not return any result; */
+  public boolean isEmpty() {
+    return empty;
   }
 
   @Override
   public String toString() {
-    return (getSuccess() ? "Successful" : "Unsuccessful") + ", result size = "
-        + getResultSet().size() + ", " + getResultSet();
+    return (getSuccess() ? "Successful" : "Unsuccessful") + ", empty = " + empty;
   }
 }
diff --git a/src/main/java/com/google/devtools/build/lib/query2/output/OutputFormatter.java b/src/main/java/com/google/devtools/build/lib/query2/output/OutputFormatter.java
index c773ed4003..1f7cbe9449 100644
--- a/src/main/java/com/google/devtools/build/lib/query2/output/OutputFormatter.java
+++ b/src/main/java/com/google/devtools/build/lib/query2/output/OutputFormatter.java
@@ -20,6 +20,7 @@ import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Sets;
 import com.google.devtools.build.lib.cmdline.Label;
+import com.google.devtools.build.lib.collect.CompactHashSet;
 import com.google.devtools.build.lib.events.Location;
 import com.google.devtools.build.lib.graph.Digraph;
 import com.google.devtools.build.lib.graph.Node;
@@ -27,6 +28,7 @@ import com.google.devtools.build.lib.packages.Attribute;
 import com.google.devtools.build.lib.packages.PackageSerializer;
 import com.google.devtools.build.lib.packages.Rule;
 import com.google.devtools.build.lib.packages.Target;
+import com.google.devtools.build.lib.query2.engine.OutputFormatterCallback;
 import com.google.devtools.build.lib.query2.output.QueryOptions.OrderOutput;
 import com.google.devtools.build.lib.syntax.EvalUtils;
 import com.google.devtools.build.lib.syntax.Printer;
@@ -153,17 +155,20 @@ public abstract class OutputFormatter implements Serializable {
       AspectResolver aspectProvider) throws IOException, InterruptedException;
 
   /**
-   * Unordered output formatter (wrt. dependency ordering).
+   * Unordered streamed output formatter (wrt. dependency ordering).
    *
-   * <p>Formatters that support unordered output may be used when only the set of query results is
+   * <p>Formatters that support streamed output may be used when only the set of query results is
    * requested but their ordering is irrelevant.
    *
-   * <p>The benefit of using a unordered formatter is that we can save the potentially expensive
-   * subgraph extraction step before presenting the query results.
+   * <p>The benefit of using a streamed formatter is that we can save the potentially expensive
+   * subgraph extraction step before presenting the query results and that depending on the query
+   * environment used, it can be more memory performant, as it does not aggregate all the data
+   * before writting in the output.
    */
-  public interface UnorderedFormatter {
-    void outputUnordered(QueryOptions options, Iterable<Target> result, PrintStream out,
-        AspectResolver aspectResolver) throws IOException, InterruptedException;
+  public interface StreamedFormatter {
+
+    OutputFormatterCallback<Target> createStreamCallback(QueryOptions options, PrintStream out,
+        AspectResolver aspectResolver);
   }
 
   /**
@@ -172,8 +177,8 @@ public abstract class OutputFormatter implements Serializable {
   public abstract String getName();
 
   abstract static class AbstractUnorderedFormatter extends OutputFormatter
-      implements UnorderedFormatter {
-    private static Iterable<Target> getOrderedTargets(
+      implements StreamedFormatter {
+    protected Iterable<Target> getOrderedTargets(
         Digraph<Target> result, QueryOptions options) {
       Iterable<Node<Target>> orderedResult =
           options.orderOutput == OrderOutput.DEPS
@@ -183,13 +188,11 @@ public abstract class OutputFormatter implements Serializable {
     }
 
     @Override
-    public void output(
-        QueryOptions options,
-        Digraph<Target> result,
-        PrintStream out,
-        AspectResolver aspectResolver)
-        throws IOException, InterruptedException {
-      outputUnordered(options, getOrderedTargets(result, options), out, aspectResolver);
+    public void output(QueryOptions options, Digraph<Target> result, PrintStream out,
+        AspectResolver aspectResolver) throws IOException, InterruptedException {
+      OutputFormatterCallback.processAllTargets(
+          createStreamCallback(options, out, aspectResolver),
+          getOrderedTargets(result, options));
     }
   }
 
@@ -201,7 +204,7 @@ public abstract class OutputFormatter implements Serializable {
 
     private final boolean showKind;
 
-    public LabelOutputFormatter(boolean showKind) {
+    private LabelOutputFormatter(boolean showKind) {
       this.showKind = showKind;
     }
 
@@ -211,15 +214,22 @@ public abstract class OutputFormatter implements Serializable {
     }
 
     @Override
-    public void outputUnordered(QueryOptions options, Iterable<Target> result, PrintStream out,
-        AspectResolver aspectResolver) {
-      for (Target target : result) {
-        if (showKind) {
-          out.print(target.getTargetKind());
-          out.print(' ');
+    public OutputFormatterCallback<Target> createStreamCallback(QueryOptions options,
+        final PrintStream out, AspectResolver aspectResolver) {
+      return new OutputFormatterCallback<Target>() {
+
+        @Override
+        protected void processOutput(Iterable<Target> partialResult)
+            throws IOException, InterruptedException {
+          for (Target target : partialResult) {
+            if (showKind) {
+              out.print(target.getTargetKind());
+              out.print(' ');
+            }
+            out.println(target.getLabel());
+          }
         }
-        out.println(target.getLabel());
-      }
+      };
     }
   }
 
@@ -245,15 +255,28 @@ public abstract class OutputFormatter implements Serializable {
     }
 
     @Override
-    public void outputUnordered(QueryOptions options, Iterable<Target> result, PrintStream out,
+    public OutputFormatterCallback<Target> createStreamCallback(QueryOptions options,
+        final PrintStream out,
         AspectResolver aspectResolver) {
-      Set<String> packageNames = Sets.newTreeSet();
-      for (Target target : result) {
-        packageNames.add(target.getLabel().getPackageName());
-      }
-      for (String packageName : packageNames) {
-        out.println(packageName);
-      }
+      return new OutputFormatterCallback<Target>() {
+        private final Set<String> packageNames = Sets.newTreeSet();
+
+        @Override
+        protected void processOutput(Iterable<Target> partialResult)
+            throws IOException, InterruptedException {
+
+          for (Target target : partialResult) {
+            packageNames.add(target.getLabel().getPackageName());
+          }
+        }
+
+        @Override
+        public void close() throws IOException {
+          for (String packageName : packageNames) {
+            out.println(packageName);
+          }
+        }
+      };
     }
   }
 
@@ -270,12 +293,20 @@ public abstract class OutputFormatter implements Serializable {
     }
 
     @Override
-    public void outputUnordered(QueryOptions options, Iterable<Target> result, PrintStream out,
+    public OutputFormatterCallback<Target> createStreamCallback(QueryOptions options,
+        final PrintStream out,
         AspectResolver aspectResolver) {
-      for (Target target : result) {
-        Location location = target.getLocation();
-        out.println(location.print()  + ": " + target.getTargetKind() + " " + target.getLabel());
-      }
+      return new OutputFormatterCallback<Target>() {
+
+        @Override
+        protected void processOutput(Iterable<Target> partialResult)
+            throws IOException, InterruptedException {
+          for (Target target : partialResult) {
+            Location location = target.getLocation();
+            out.println(location.print() + ": " + target.getTargetKind() + " " + target.getLabel());
+          }
+        }
+      };
     }
   }
 
@@ -290,48 +321,57 @@ public abstract class OutputFormatter implements Serializable {
       return "build";
     }
 
-    private void outputRule(Rule rule, PrintStream out) {
-      out.printf("# %s%n", rule.getLocation());
-      out.printf("%s(%n", rule.getRuleClass());
-      out.printf("  name = \"%s\",%n", rule.getName());
-
-      for (Attribute attr : rule.getAttributes()) {
-        Pair<Iterable<Object>, AttributeValueSource> values = getAttributeValues(rule, attr);
-        if (Iterables.size(values.first) != 1) {
-          continue;  // TODO(bazel-team): handle configurable attributes.
-        }
-        if (values.second != AttributeValueSource.RULE) {
-          continue;  // Don't print default values.
-        }
-        Object value = Iterables.getOnlyElement(values.first);
-        out.printf("  %s = ", attr.getPublicName());
-        if (value instanceof Label) {
-          value = value.toString();
-        } else if (value instanceof List<?> && EvalUtils.isImmutable(value)) {
-          // Display it as a list (and not as a tuple). Attributes can never be tuples.
-          value = new ArrayList<>((List<?>) value);
-        }
-        // It is *much* faster to write to a StringBuilder compared to the PrintStream object.
-        StringBuilder builder = new StringBuilder();
-        Printer.write(builder, value);
-        out.print(builder);
-        out.println(",");
-      }
-      out.printf(")\n%n");
-    }
-
     @Override
-    public void outputUnordered(QueryOptions options, Iterable<Target> result, PrintStream out,
+    public OutputFormatterCallback<Target> createStreamCallback(QueryOptions options,
+        final PrintStream out,
         AspectResolver aspectResolver) {
-      Set<Label> printed = new HashSet<>();
-      for (Target target : result) {
-        Rule rule = target.getAssociatedRule();
-        if (rule == null || printed.contains(rule.getLabel())) {
-          continue;
+      return new OutputFormatterCallback<Target>() {
+        private final Set<Label> printed = CompactHashSet.create();
+
+        private void outputRule(Rule rule, PrintStream out) {
+          out.printf("# %s%n", rule.getLocation());
+          out.printf("%s(%n", rule.getRuleClass());
+          out.printf("  name = \"%s\",%n", rule.getName());
+
+          for (Attribute attr : rule.getAttributes()) {
+            Pair<Iterable<Object>, AttributeValueSource> values = getAttributeValues(rule, attr);
+            if (Iterables.size(values.first) != 1) {
+              continue;  // TODO(bazel-team): handle configurable attributes.
+            }
+            if (values.second != AttributeValueSource.RULE) {
+              continue;  // Don't print default values.
+            }
+            Object value = Iterables.getOnlyElement(values.first);
+            out.printf("  %s = ", attr.getPublicName());
+            if (value instanceof Label) {
+              value = value.toString();
+            } else if (value instanceof List<?> && EvalUtils.isImmutable(value)) {
+              // Display it as a list (and not as a tuple). Attributes can never be tuples.
+              value = new ArrayList<>((List<?>) value);
+            }
+            // It is *much* faster to write to a StringBuilder compared to the PrintStream object.
+            StringBuilder builder = new StringBuilder();
+            Printer.write(builder, value);
+            out.print(builder);
+            out.println(",");
+          }
+          out.printf(")\n%n");
         }
-        outputRule(rule, out);
-        printed.add(rule.getLabel());
-      }
+
+        @Override
+        protected void processOutput(Iterable<Target> partialResult)
+            throws IOException, InterruptedException {
+
+          for (Target target : partialResult) {
+            Rule rule = target.getAssociatedRule();
+            if (rule == null || printed.contains(rule.getLabel())) {
+              continue;
+            }
+            outputRule(rule, out);
+            printed.add(rule.getLabel());
+          }
+        }
+      };
     }
   }
 
diff --git a/src/main/java/com/google/devtools/build/lib/query2/output/ProtoOutputFormatter.java b/src/main/java/com/google/devtools/build/lib/query2/output/ProtoOutputFormatter.java
index 66348b3be3..8bca7dfc1d 100644
--- a/src/main/java/com/google/devtools/build/lib/query2/output/ProtoOutputFormatter.java
+++ b/src/main/java/com/google/devtools/build/lib/query2/output/ProtoOutputFormatter.java
@@ -34,10 +34,12 @@ import com.google.devtools.build.lib.packages.ProtoUtils;
 import com.google.devtools.build.lib.packages.Rule;
 import com.google.devtools.build.lib.packages.Target;
 import com.google.devtools.build.lib.query2.FakeSubincludeTarget;
+import com.google.devtools.build.lib.query2.engine.OutputFormatterCallback;
 import com.google.devtools.build.lib.query2.output.AspectResolver.BuildFileDependencyMode;
-import com.google.devtools.build.lib.query2.output.OutputFormatter.UnorderedFormatter;
+import com.google.devtools.build.lib.query2.output.OutputFormatter.AbstractUnorderedFormatter;
 import com.google.devtools.build.lib.query2.output.QueryOptions.OrderOutput;
 import com.google.devtools.build.lib.query2.proto.proto2api.Build;
+import com.google.devtools.build.lib.query2.proto.proto2api.Build.QueryResult.Builder;
 import com.google.devtools.build.lib.syntax.Environment;
 import com.google.devtools.build.lib.util.BinaryPredicate;
 
@@ -54,7 +56,7 @@ import java.util.Set;
  * By taking the bytes and calling {@code mergeFrom()} on a
  * {@code Build.QueryResult} object the full result can be reconstructed.
  */
-public class ProtoOutputFormatter extends OutputFormatter implements UnorderedFormatter {
+public class ProtoOutputFormatter extends AbstractUnorderedFormatter {
 
   /**
    * A special attribute name for the rule implementation hash code.
@@ -77,19 +79,36 @@ public class ProtoOutputFormatter extends OutputFormatter implements UnorderedFo
   }
 
   @Override
-  public void outputUnordered(QueryOptions options, Iterable<Target> result, PrintStream out,
-      AspectResolver aspectResolver) throws IOException, InterruptedException {
+  public OutputFormatterCallback<Target> createStreamCallback(QueryOptions options,
+      final PrintStream out, AspectResolver aspectResolver) {
     relativeLocations = options.relativeLocations;
     this.aspectResolver = aspectResolver;
     this.includeDefaultValues = options.protoIncludeDefaultValues;
     setDependencyFilter(options);
 
-    Build.QueryResult.Builder queryResult = Build.QueryResult.newBuilder();
-    for (Target target : result) {
-      addTarget(queryResult, target);
-    }
+    return new OutputFormatterCallback<Target>() {
+
+      private Builder queryResult;
+
+      @Override
+      public void start() {
+        queryResult = Build.QueryResult.newBuilder();
+      }
+
+      @Override
+      protected void processOutput(Iterable<Target> partialResult)
+          throws IOException, InterruptedException {
+
+        for (Target target : partialResult) {
+          queryResult.addTarget(toTargetProtoBuffer(target));
+        }
+      }
 
-    queryResult.build().writeTo(out);
+      @Override
+      public void close() throws IOException {
+        queryResult.build().writeTo(out);
+      }
+    };
   }
 
   private static Iterable<Target> getSortedLabels(Digraph<Target> result) {
@@ -98,24 +117,8 @@ public class ProtoOutputFormatter extends OutputFormatter implements UnorderedFo
   }
 
   @Override
-  public void output(QueryOptions options, Digraph<Target> result, PrintStream out,
-      AspectResolver aspectResolver) throws IOException, InterruptedException {
-    outputUnordered(
-        options,
-        options.orderOutput == OrderOutput.FULL ? getSortedLabels(result) : result.getLabels(),
-        out,
-        aspectResolver);
-  }
-
-  /**
-   * Add the target to the query result.
-   * @param queryResult The query result that contains all rule, input and
-   *   output targets.
-   * @param target The query target being converted to a protocol buffer.
-   */
-  private void addTarget(Build.QueryResult.Builder queryResult, Target target)
-      throws InterruptedException {
-    queryResult.addTarget(toTargetProtoBuffer(target));
+  protected Iterable<Target> getOrderedTargets(Digraph<Target> result, QueryOptions options) {
+    return options.orderOutput == OrderOutput.FULL ? getSortedLabels(result) : result.getLabels();
   }
 
   /**
diff --git a/src/main/java/com/google/devtools/build/lib/query2/output/QueryOutputUtils.java b/src/main/java/com/google/devtools/build/lib/query2/output/QueryOutputUtils.java
index 36f5464461..1533b7bf43 100644
--- a/src/main/java/com/google/devtools/build/lib/query2/output/QueryOutputUtils.java
+++ b/src/main/java/com/google/devtools/build/lib/query2/output/QueryOutputUtils.java
@@ -14,32 +14,43 @@
 package com.google.devtools.build.lib.query2.output;
 
 import com.google.devtools.build.lib.packages.Target;
-import com.google.devtools.build.lib.query2.engine.BlazeQueryEvalResult;
+import com.google.devtools.build.lib.query2.engine.DigraphQueryEvalResult;
+import com.google.devtools.build.lib.query2.engine.OutputFormatterCallback;
 import com.google.devtools.build.lib.query2.engine.QueryEvalResult;
-import com.google.devtools.build.lib.query2.output.OutputFormatter.UnorderedFormatter;
+import com.google.devtools.build.lib.query2.output.OutputFormatter.StreamedFormatter;
 import com.google.devtools.build.lib.query2.output.QueryOptions.OrderOutput;
 
 import java.io.IOException;
 import java.io.PrintStream;
+import java.util.Set;
 
 /** Static utility methods for outputting a query. */
 public class QueryOutputUtils {
   // Utility class cannot be instantiated.
   private QueryOutputUtils() {}
 
-  public static boolean orderResults(QueryOptions queryOptions, OutputFormatter formatter) {
-    return queryOptions.orderOutput != OrderOutput.NO || !(formatter instanceof UnorderedFormatter);
+  public static boolean shouldStreamResults(QueryOptions queryOptions, OutputFormatter formatter) {
+    return queryOptions.orderOutput == OrderOutput.NO
+        && formatter instanceof StreamedFormatter;
   }
 
-  public static void output(QueryOptions queryOptions, QueryEvalResult<Target> result,
-      OutputFormatter formatter, PrintStream outputStream, AspectResolver aspectResolver)
+  public static void output(QueryOptions queryOptions, QueryEvalResult result,
+      Set<Target> targetsResult, OutputFormatter formatter, PrintStream outputStream,
+      AspectResolver aspectResolver)
       throws IOException, InterruptedException {
-    if (orderResults(queryOptions, formatter)) {
-      formatter.output(queryOptions, ((BlazeQueryEvalResult<Target>) result).getResultGraph(),
+    /*
+     * This is not really streaming, but we are using the streaming interface for writing into the
+     * output everything in one batch. This happens when the QueryEnvironment does not
+     * support streaming but we don't care about ordered results.
+     */
+    boolean orderedResults = !shouldStreamResults(queryOptions, formatter);
+    if (orderedResults) {
+      formatter.output(queryOptions,
+          ((DigraphQueryEvalResult<Target>) result).getGraph().extractSubgraph(targetsResult),
           outputStream, aspectResolver);
     } else {
-      ((UnorderedFormatter) formatter).outputUnordered(queryOptions, result.getResultSet(),
-          outputStream, aspectResolver);
+      OutputFormatterCallback.processAllTargets(((StreamedFormatter) formatter)
+          .createStreamCallback(queryOptions, outputStream, aspectResolver), targetsResult);
     }
   }
 }
diff --git a/src/main/java/com/google/devtools/build/lib/query2/output/XmlOutputFormatter.java b/src/main/java/com/google/devtools/build/lib/query2/output/XmlOutputFormatter.java
index b8b4466bad..2af71b7414 100644
--- a/src/main/java/com/google/devtools/build/lib/query2/output/XmlOutputFormatter.java
+++ b/src/main/java/com/google/devtools/build/lib/query2/output/XmlOutputFormatter.java
@@ -27,6 +27,7 @@ import com.google.devtools.build.lib.packages.PackageGroup;
 import com.google.devtools.build.lib.packages.Rule;
 import com.google.devtools.build.lib.packages.Target;
 import com.google.devtools.build.lib.query2.FakeSubincludeTarget;
+import com.google.devtools.build.lib.query2.engine.OutputFormatterCallback;
 import com.google.devtools.build.lib.query2.output.AspectResolver.BuildFileDependencyMode;
 import com.google.devtools.build.lib.query2.output.OutputFormatter.AbstractUnorderedFormatter;
 import com.google.devtools.build.lib.util.BinaryPredicate;
@@ -36,6 +37,7 @@ import org.w3c.dom.DOMException;
 import org.w3c.dom.Document;
 import org.w3c.dom.Element;
 
+import java.io.IOException;
 import java.io.PrintStream;
 import java.util.Collection;
 import java.util.HashSet;
@@ -57,11 +59,9 @@ import javax.xml.transform.stream.StreamResult;
  */
 class XmlOutputFormatter extends AbstractUnorderedFormatter {
 
-  private boolean xmlLineNumbers;
-  private boolean showDefaultValues;
-  private boolean relativeLocations;
-  private transient AspectResolver aspectResolver;
-  private transient BinaryPredicate<Rule, Attribute> dependencyFilter;
+  private QueryOptions options;
+  private AspectResolver aspectResolver;
+  private BinaryPredicate<Rule, Attribute> dependencyFilter;
 
   @Override
   public String getName() {
@@ -69,36 +69,52 @@ class XmlOutputFormatter extends AbstractUnorderedFormatter {
   }
 
   @Override
-  public void outputUnordered(QueryOptions options, Iterable<Target> result, PrintStream out,
-      AspectResolver aspectResolver) throws InterruptedException {
-    this.xmlLineNumbers = options.xmlLineNumbers;
-    this.showDefaultValues = options.xmlShowDefaultValues;
-    this.relativeLocations = options.relativeLocations;
-    this.dependencyFilter = OutputFormatter.getDependencyFilter(options);
+  public OutputFormatterCallback<Target> createStreamCallback(QueryOptions options,
+      final PrintStream out, AspectResolver aspectResolver) {
+    this.options = options;
     this.aspectResolver = aspectResolver;
-    Document doc;
-    try {
-      DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
-      doc = factory.newDocumentBuilder().newDocument();
-    } catch (ParserConfigurationException e) {
-      // This shouldn't be possible: all the configuration is hard-coded.
-      throw new IllegalStateException("XML output failed",  e);
-    }
-    doc.setXmlVersion("1.1");
-    Element queryElem = doc.createElement("query");
-    queryElem.setAttribute("version", "2");
-    doc.appendChild(queryElem);
-    for (Target target : result) {
-      queryElem.appendChild(createTargetElement(doc, target));
-    }
-    try {
-      Transformer transformer = TransformerFactory.newInstance().newTransformer();
-      transformer.setOutputProperty(OutputKeys.INDENT, "yes");
-      transformer.transform(new DOMSource(doc), new StreamResult(out));
-    } catch (TransformerFactoryConfigurationError | TransformerException e) {
-      // This shouldn't be possible: all the configuration is hard-coded.
-      throw new IllegalStateException("XML output failed",  e);
-    }
+    this.dependencyFilter = OutputFormatter.getDependencyFilter(options);
+    return new OutputFormatterCallback<Target>() {
+
+      private Document doc;
+      private Element queryElem;
+
+
+      @Override
+      public void start() {
+        try {
+          DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
+          doc = factory.newDocumentBuilder().newDocument();
+        } catch (ParserConfigurationException e) {
+          // This shouldn't be possible: all the configuration is hard-coded.
+          throw new IllegalStateException("XML output failed", e);
+        }
+        doc.setXmlVersion("1.1");
+        queryElem = doc.createElement("query");
+        queryElem.setAttribute("version", "2");
+        doc.appendChild(queryElem);
+      }
+
+      @Override
+      protected void processOutput(Iterable<Target> partialResult)
+          throws IOException, InterruptedException {
+        for (Target target : partialResult) {
+          queryElem.appendChild(createTargetElement(doc, target));
+        }
+      }
+
+      @Override
+      public void close() throws IOException {
+        try {
+          Transformer transformer = TransformerFactory.newInstance().newTransformer();
+          transformer.setOutputProperty(OutputKeys.INDENT, "yes");
+          transformer.transform(new DOMSource(doc), new StreamResult(out));
+        } catch (TransformerFactoryConfigurationError | TransformerException e) {
+          // This shouldn't be possible: all the configuration is hard-coded.
+          throw new IllegalStateException("XML output failed", e);
+        }
+      }
+    };
   }
 
   /**
@@ -120,9 +136,9 @@ class XmlOutputFormatter extends AbstractUnorderedFormatter {
       Rule rule = (Rule) target;
       elem = doc.createElement("rule");
       elem.setAttribute("class", rule.getRuleClass());
-      for (Attribute attr: rule.getAttributes()) {
+      for (Attribute attr : rule.getAttributes()) {
         Pair<Iterable<Object>, AttributeValueSource> values = getAttributeValues(rule, attr);
-        if (values.second == AttributeValueSource.RULE || showDefaultValues) {
+        if (values.second == AttributeValueSource.RULE || options.xmlShowDefaultValues) {
           Element attrElem = createValueElement(doc, attr.getType(), values.first);
           attrElem.setAttribute("name", attr.getName());
           elem.appendChild(attrElem);
@@ -205,8 +221,8 @@ class XmlOutputFormatter extends AbstractUnorderedFormatter {
     }
 
     elem.setAttribute("name", target.getLabel().toString());
-    String location = getLocation(target, relativeLocations);
-    if (!xmlLineNumbers) {
+    String location = getLocation(target, options.relativeLocations);
+    if (!options.xmlLineNumbers) {
       int firstColon = location.indexOf(':');
       if (firstColon != -1) {
         location = location.substring(0, firstColon);
diff --git a/src/main/java/com/google/devtools/build/lib/rules/genquery/GenQuery.java b/src/main/java/com/google/devtools/build/lib/rules/genquery/GenQuery.java
index 13d5be4dd8..76eac71bcf 100644
--- a/src/main/java/com/google/devtools/build/lib/rules/genquery/GenQuery.java
+++ b/src/main/java/com/google/devtools/build/lib/rules/genquery/GenQuery.java
@@ -49,10 +49,11 @@ import com.google.devtools.build.lib.pkgcache.FilteringPolicy;
 import com.google.devtools.build.lib.pkgcache.PackageProvider;
 import com.google.devtools.build.lib.pkgcache.TargetPatternEvaluator;
 import com.google.devtools.build.lib.query2.AbstractBlazeQueryEnvironment;
-import com.google.devtools.build.lib.query2.engine.BlazeQueryEvalResult;
+import com.google.devtools.build.lib.query2.engine.DigraphQueryEvalResult;
 import com.google.devtools.build.lib.query2.engine.QueryEnvironment.QueryFunction;
 import com.google.devtools.build.lib.query2.engine.QueryEnvironment.Setting;
 import com.google.devtools.build.lib.query2.engine.QueryException;
+import com.google.devtools.build.lib.query2.engine.QueryUtil.AggregateAllCallback;
 import com.google.devtools.build.lib.query2.engine.SkyframeRestartQueryException;
 import com.google.devtools.build.lib.query2.output.OutputFormatter;
 import com.google.devtools.build.lib.query2.output.QueryOptions;
@@ -260,8 +261,9 @@ public class GenQuery implements RuleConfiguredTargetFactory {
                          String query, RuleContext ruleContext)
       throws InterruptedException {
 
-    BlazeQueryEvalResult<Target> queryResult;
+    DigraphQueryEvalResult<Target> queryResult;
     OutputFormatter formatter;
+    AggregateAllCallback<Target> targets = new AggregateAllCallback<>();
     try {
       Set<Setting> settings = queryOptions.toSettings();
 
@@ -283,20 +285,20 @@ public class GenQuery implements RuleConfiguredTargetFactory {
       // All the packages are already loaded at this point, so there is no need
       // to start up many threads. 4 are started up to make good use of multiple
       // cores.
-      queryResult = (BlazeQueryEvalResult<Target>) AbstractBlazeQueryEnvironment
+      queryResult = (DigraphQueryEvalResult<Target>) AbstractBlazeQueryEnvironment
           .newQueryEnvironment(
               /*transitivePackageLoader=*/null, /*graph=*/null, packageProvider,
               evaluator,
               /*keepGoing=*/false,
               ruleContext.attributes().get("strict", Type.BOOLEAN),
-              /*orderedResults=*/QueryOutputUtils.orderResults(queryOptions, formatter),
+              /*orderedResults=*/!QueryOutputUtils.shouldStreamResults(queryOptions, formatter),
               /*universeScope=*/ImmutableList.<String>of(),
               /*loadingPhaseThreads=*/4,
               labelFilter,
               getEventHandler(ruleContext),
               settings,
               ImmutableList.<QueryFunction>of(),
-              /*packagePath=*/null).evaluateQuery(query);
+              /*packagePath=*/null).evaluateQuery(query, targets);
     } catch (SkyframeRestartQueryException e) {
       // Do not emit errors for skyframe restarts. They make output of the ConfiguredTargetFunction
       // inconsistent from run to run, and make detecting legitimate errors more difficult.
@@ -310,7 +312,8 @@ public class GenQuery implements RuleConfiguredTargetFactory {
     PrintStream printStream = new PrintStream(outputStream);
 
     try {
-      QueryOutputUtils.output(queryOptions, queryResult, formatter, printStream,
+      QueryOutputUtils
+          .output(queryOptions, queryResult, targets.getResult(), formatter, printStream,
           queryOptions.aspectDeps.createResolver(packageProvider, getEventHandler(ruleContext)));
     } catch (ClosedByInterruptException e) {
       throw new InterruptedException(e.getMessage());
diff --git a/src/main/java/com/google/devtools/build/lib/runtime/commands/QueryCommand.java b/src/main/java/com/google/devtools/build/lib/runtime/commands/QueryCommand.java
index 2f371c9dd1..b80be41938 100644
--- a/src/main/java/com/google/devtools/build/lib/runtime/commands/QueryCommand.java
+++ b/src/main/java/com/google/devtools/build/lib/runtime/commands/QueryCommand.java
@@ -16,17 +16,21 @@ package com.google.devtools.build.lib.runtime.commands;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Joiner;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
 import com.google.devtools.build.lib.Constants;
+import com.google.devtools.build.lib.collect.CompactHashSet;
 import com.google.devtools.build.lib.events.Event;
 import com.google.devtools.build.lib.packages.Target;
 import com.google.devtools.build.lib.pkgcache.PackageCacheOptions;
 import com.google.devtools.build.lib.query2.AbstractBlazeQueryEnvironment;
+import com.google.devtools.build.lib.query2.engine.OutputFormatterCallback;
 import com.google.devtools.build.lib.query2.engine.QueryEnvironment.QueryFunction;
 import com.google.devtools.build.lib.query2.engine.QueryEnvironment.Setting;
 import com.google.devtools.build.lib.query2.engine.QueryEvalResult;
 import com.google.devtools.build.lib.query2.engine.QueryException;
 import com.google.devtools.build.lib.query2.engine.QueryExpression;
 import com.google.devtools.build.lib.query2.output.OutputFormatter;
+import com.google.devtools.build.lib.query2.output.OutputFormatter.StreamedFormatter;
 import com.google.devtools.build.lib.query2.output.QueryOptions;
 import com.google.devtools.build.lib.query2.output.QueryOutputUtils;
 import com.google.devtools.build.lib.runtime.BlazeCommand;
@@ -107,10 +111,11 @@ public final class QueryCommand implements BlazeCommand {
     String query = Joiner.on(' ').join(options.getResidue());
 
     Set<Setting> settings = queryOptions.toSettings();
+    boolean streamResults = QueryOutputUtils.shouldStreamResults(queryOptions, formatter);
     AbstractBlazeQueryEnvironment<Target> queryEnv = newQueryEnvironment(
         env,
         queryOptions.keepGoing,
-        QueryOutputUtils.orderResults(queryOptions, formatter),
+        !streamResults,
         queryOptions.universeScope, queryOptions.loadingPhaseThreads,
         settings);
 
@@ -124,36 +129,70 @@ public final class QueryCommand implements BlazeCommand {
       return ExitCode.COMMAND_LINE_ERROR;
     }
 
-    // 2. Evaluate expression:
-    QueryEvalResult<Target> result;
+    QueryEvalResult result;
+    PrintStream output = new PrintStream(env.getReporter().getOutErr().getOutputStream());
+    OutputFormatterCallback<Target> callback;
+    if (streamResults) {
+      // 2. Evaluate expression:
+      callback = ((StreamedFormatter) formatter)
+          .createStreamCallback(queryOptions, output, queryOptions.aspectDeps.createResolver(
+              env.getPackageManager(), env.getReporter()));
+    } else {
+      callback = new AggregateAllOutputFormatterCallback<>();
+    }
     try {
-      result = queryEnv.evaluateQuery(expr);
-    } catch (QueryException | InterruptedException e) {
+      callback.start();
+      result = queryEnv.evaluateQuery(expr, callback);
+    } catch (QueryException e) {
       // Keep consistent with reportBuildFileError()
       env.getReporter()
-          // TODO(bazel-team): this is a kludge to fix a bug observed in the wild. We should make
-          // sure no null error messages ever get in.
-          .handle(Event.error(e.getMessage() == null ? e.toString() : e.getMessage()));
+         // TODO(bazel-team): this is a kludge to fix a bug observed in the wild. We should make
+         // sure no null error messages ever get in.
+         .handle(Event.error(e.getMessage() == null ? e.toString() : e.getMessage()));
       return ExitCode.ANALYSIS_FAILURE;
-    }
-
-    env.getReporter().switchToAnsiAllowingHandler();
-    // 3. Output results:
-    PrintStream output = new PrintStream(env.getReporter().getOutErr().getOutputStream());
-    try {
-      QueryOutputUtils.output(queryOptions, result, formatter, output,
-          queryOptions.aspectDeps.createResolver(
-              env.getPackageManager(), env.getReporter()));
-    } catch (ClosedByInterruptException | InterruptedException e) {
-      env.getReporter().handle(Event.error("query interrupted"));
-      return ExitCode.INTERRUPTED;
+    } catch (InterruptedException e) {
+      IOException ioException = callback.getIoException();
+      if (ioException == null || ioException instanceof ClosedByInterruptException) {
+        env.getReporter().handle(Event.error("query interrupted"));
+        return ExitCode.INTERRUPTED;
+      } else {
+        env.getReporter().handle(Event.error("I/O error: " + e.getMessage()));
+        return ExitCode.LOCAL_ENVIRONMENTAL_ERROR;
+      }
     } catch (IOException e) {
       env.getReporter().handle(Event.error("I/O error: " + e.getMessage()));
       return ExitCode.LOCAL_ENVIRONMENTAL_ERROR;
     } finally {
       output.flush();
+      try {
+        callback.close();
+      } catch (IOException e) {
+        env.getReporter().handle(Event.error("I/O error: " + e.getMessage()));
+        return ExitCode.LOCAL_ENVIRONMENTAL_ERROR;
+      }
+    }
+    if (!streamResults) {
+      // 3. Output results:
+      try {
+        Set<Target> targets = ((AggregateAllOutputFormatterCallback<Target>) callback).getOutput();
+        QueryOutputUtils.output(queryOptions, result,
+            targets, formatter, output,
+            queryOptions.aspectDeps.createResolver(
+                env.getPackageManager(), env.getReporter()));
+      } catch (ClosedByInterruptException | InterruptedException e) {
+        env.getReporter().handle(Event.error("query interrupted"));
+        return ExitCode.INTERRUPTED;
+      } catch (IOException e) {
+        env.getReporter().handle(Event.error("I/O error: " + e.getMessage()));
+        return ExitCode.LOCAL_ENVIRONMENTAL_ERROR;
+      } finally {
+        output.flush();
+      }
+
     }
-    if (result.getResultSet().isEmpty()) {
+    env.getReporter().switchToAnsiAllowingHandler();
+
+    if (result.isEmpty()) {
       env.getReporter().handle(Event.info("Empty results"));
     }
 
@@ -185,4 +224,19 @@ public final class QueryCommand implements BlazeCommand {
         functions.build(),
         env.getPackageManager().getPackagePath());
   }
+
+  private static class AggregateAllOutputFormatterCallback<T> extends OutputFormatterCallback<T> {
+
+    private Set<T> output = CompactHashSet.create();
+
+    @Override
+    protected final void processOutput(Iterable<T> partialResult)
+        throws IOException, InterruptedException {
+      Iterables.addAll(output, partialResult);
+    }
+
+    public Set<T> getOutput() {
+      return output;
+    }
+  }
 }