fixup! use non-sorting bucketmerger at MergeOperation

sql/src/main/java/io/crate/executor/transport/distributed/DownstreamOperationContext.java

@@ -46,7 +46,6 @@ public class DownstreamOperationContext {
     private final AtomicInteger mergeOperationsLeft;
     private final DownstreamOperation downstreamOperation;
     private final List<SettableFuture<Bucket>> upstreamFutures;
-    private final SettableFuture<Bucket> listener;
     private final Streamer<?>[] streamers;
     private final DistributedRequestContextManager.DoneCallback doneCallback;
     private final AtomicBoolean needsMoreRows;
@@ -64,7 +63,6 @@ public DownstreamOperationContext(DownstreamOperation downstreamOperation,
         }
         this.needsMoreRows = new AtomicBoolean(true);
         this.alreadyFinished = new AtomicBoolean(false);
-        this.listener = listener;
         Futures.addCallback(downstreamOperation.result(), new FutureCallback<Bucket>() {
             @Override
             public void onSuccess(Bucket result) {
@@ -107,11 +105,9 @@ public void addFailure(@Nullable Throwable failure) {
             failure = new UnknownUpstreamFailure();
         }
         if (bucketNum > 0 && !alreadyFinished.get()) {
+            // this will bubble the exception up to the result listener
             upstreamFutures.get(bucketNum).setException(failure);
         }
-        // TODO: needed?
-        boolean firstFailure = listener.setException(failure);
-        LOGGER.trace("addFailure first: {}", firstFailure);
     }
 
     public void add(Bucket bucket) {