made bucketmerge impls asynchronous

and inherit from PageDownstream and RowUpstream to officially bridge between Pages and Rows, Projectors

core/src/main/java/io/crate/core/collections/BucketPage.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed to CRATE Technology GmbH ("Crate") under one or more contributor
+ * license agreements.  See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership.  Crate licenses
+ * this file to you 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.
+ *
+ * However, if you have executed another commercial license agreement
+ * with Crate these terms will supersede the license and you may use the
+ * software solely pursuant to the terms of the relevant commercial agreement.
+ */
+
+package io.crate.core.collections;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ *
+ * Abstraction on a set of buckets that belong together
+ * e.g. sent from different nodes but belonging to the same relation
+ *
+ * TODO: rename to Page once we are rid of io.crate.executor.Page
+ */
+public class BucketPage {
+
+    private final List<ListenableFuture<Bucket>> buckets;
+
+    public BucketPage(List<ListenableFuture<Bucket>> buckets) {
+        this.buckets = buckets;
+    }
+
+    public List<ListenableFuture<Bucket>> buckets() {
+        return buckets;
+    }
+
+    @SafeVarargs
+    public static BucketPage from(ListenableFuture<Bucket> ... futures) {
+        return new BucketPage(Arrays.asList(futures));
+    }
+}

sql/src/main/java/io/crate/operation/PageConsumeListener.java

@@ -0,0 +1,43 @@
+/*
+ * Licensed to CRATE Technology GmbH ("Crate") under one or more contributor
+ * license agreements.  See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership.  Crate licenses
+ * this file to you 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.
+ *
+ * However, if you have executed another commercial license agreement
+ * with Crate these terms will supersede the license and you may use the
+ * software solely pursuant to the terms of the relevant commercial agreement.
+ */
+
+package io.crate.operation;
+
+/**
+ * Listener that is called when a Page is consumed to signal how to carry on
+ * e.g. getting another page or shutting down.
+ *
+ */
+public interface PageConsumeListener {
+
+    /**
+     * called by the consumer of a {@link io.crate.core.collections.BucketPage}
+     * to signal that he needs another page.
+     */
+    public void needMore();
+
+    /**
+     * called by the consumer of a {@link io.crate.core.collections.BucketPage}
+     * to signal that he does not need any more pages.
+     */
+    public void finished();
+
+}

sql/src/main/java/io/crate/operation/PageDownstream.java

@@ -0,0 +1,65 @@
+/*
+ * Licensed to CRATE Technology GmbH ("Crate") under one or more contributor
+ * license agreements.  See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership.  Crate licenses
+ * this file to you 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.
+ *
+ * However, if you have executed another commercial license agreement
+ * with Crate these terms will supersede the license and you may use the
+ * software solely pursuant to the terms of the relevant commercial agreement.
+ */
+
+package io.crate.operation;
+
+import io.crate.core.collections.BucketPage;
+
+/**
+ * Receiving and consuming {@linkplain io.crate.core.collections.BucketPage}s.
+ *
+ * The contract is that {@link #nextPage(io.crate.core.collections.BucketPage, PageConsumeListener)} is
+ * called as long as this downstream does not call {@link PageConsumeListener#finished()} and the page source
+ * is not exceeded.
+ * <p>
+ * One of {@link #fail(Throwable)} and {@link #finish()} must be called exactly
+ * once at last.
+ *
+ */
+public interface PageDownstream {
+
+    /**
+     * Feed a page to this downstream.
+     * This method may not be called when this downstream has already called
+     * {@link PageConsumeListener#finished()}.
+     * @param page the page with the buckets to consume
+     * @param listener to be called when the page is consumed signalling if more pages are needed.
+     */
+    public void nextPage(BucketPage page, PageConsumeListener listener);
+
+    /**
+     * Must be called by the ones feeding pages
+     * when producing and feeding pages to this downstream
+     * has finished due to calling {@link PageConsumeListener#finished()} or due to
+     * the page source being exceeded.
+     * <p>
+     * Must be called exactly once.
+     */
+    public void finish();
+
+    /**
+     * Must be called when the one feeding pages to this downstream somehow failed.
+     * <p>
+     * Must be called exactly once.
+     */
+    public void fail(Throwable e);
+
+}

sql/src/main/java/io/crate/operation/merge/BucketMerger.java

@@ -21,11 +21,9 @@
 
 package io.crate.operation.merge;
 
-import com.google.common.util.concurrent.ListenableFuture;
-import io.crate.core.collections.Bucket;
-import io.crate.operation.ProjectorUpstream;
-
-import java.util.List;
+import io.crate.operation.PageDownstream;
+import io.crate.operation.RowDownstream;
+import io.crate.operation.RowUpstream;
 
 /**
  * Bridge between Pages transmitted over the network containing buckets and Projectors
@@ -34,9 +32,12 @@
  * Merges rows from a list of buckets to a single stream of rows which are fed to
  * the downstream projector.
  */
-public interface BucketMerger extends ProjectorUpstream {
-    public void merge(List<ListenableFuture<Bucket>> buckets);
-
-    public void finish();
+public interface BucketMerger extends PageDownstream, RowUpstream {
+    /**
+     * sets the downstream of this merger
+     *
+     * @param downstream the downstream to be used
+     */
+    public void downstream(RowDownstream downstream);
 
 }

sql/src/main/java/io/crate/operation/merge/NonSortingBucketMerger.java

@@ -25,15 +25,16 @@
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import io.crate.core.collections.Bucket;
+import io.crate.core.collections.BucketPage;
 import io.crate.core.collections.Row;
+import io.crate.operation.PageConsumeListener;
+import io.crate.operation.RowDownstream;
+import io.crate.operation.RowDownstreamHandle;
 import io.crate.operation.projectors.NoOpProjector;
-import io.crate.operation.projectors.Projector;
 
 import javax.annotation.Nullable;
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * BucketMerger implementation that does not care about sorting
@@ -42,73 +43,80 @@
  */
 public class NonSortingBucketMerger implements BucketMerger {
 
-    private Projector downstream;
+    private RowDownstreamHandle downstream;
     private AtomicBoolean wantMore;
+    private AtomicBoolean alreadyFinished;
 
     public NonSortingBucketMerger() {
         this.downstream = NoOpProjector.INSTANCE;
         this.wantMore = new AtomicBoolean(true);
+        this.alreadyFinished = new AtomicBoolean(false);
     }
 
     @Override
-    public void merge(List<ListenableFuture<Bucket>> buckets) {
-        final AtomicReference<Throwable> exception = new AtomicReference<>();
-        final CountDownLatch countDownLatch = new CountDownLatch(buckets.size());
+    public void nextPage(BucketPage page, final PageConsumeListener listener) {
+        final AtomicInteger pendingBuckets = new AtomicInteger(page.buckets().size());
+        final AtomicBoolean alreadyNotified = new AtomicBoolean(false);
         FutureCallback<Bucket> callback = new FutureCallback<Bucket>() {
             @Override
             public void onSuccess(@Nullable Bucket result) {
-                if (result != null && wantMore.get()) {
+                if (result != null && wantMore.get() && !alreadyFinished.get()) {
                     for (Row row : result) {
                         if (!emitRow(row)) {
-                            stop();
+                            wantMore.set(false);
+                            notifyListener(); // shortcut finish
                             break;
                         }
                     }
                 }
-                countDownLatch.countDown();
+                if (pendingBuckets.decrementAndGet() == 0) {
+                    notifyListener();
+                }
             }
 
             @Override
             public void onFailure(Throwable t) {
-                exception.set(t);
-                stop();
+                wantMore.set(false);
+                fail(t);
+                notifyListener();
             }
 
-            private void stop() {
-                wantMore.set(false);
-                // exceed countdownlatch
-                while(countDownLatch.getCount() > 0) {
-                    countDownLatch.countDown();
+            private void notifyListener() {
+                if (!alreadyNotified.getAndSet(true)) {
+                    if (wantMore.get()) {
+                        listener.needMore();
+                    } else {
+                        listener.finished();
+                    }
                 }
             }
         };
-        for (ListenableFuture<Bucket> bucketFuture : buckets) {
+        for (ListenableFuture<Bucket> bucketFuture : page.buckets()) {
             Futures.addCallback(bucketFuture, callback);
         }
-        try {
-            countDownLatch.await();
-            Throwable caught = exception.get();
-            if (caught != null) {
-                downstream.upstreamFailed(caught);
-            }
-        } catch (InterruptedException e) {
-            downstream.upstreamFailed(e);
-        }
     }
 
     @Override
     public void finish() {
-        downstream.upstreamFinished();
+        if (!alreadyFinished.getAndSet(true)) {
+            downstream.finish();
+        }
+    }
+
+    @Override
+    public void fail(Throwable e) {
+        if (!alreadyFinished.getAndSet(true)) {
+            downstream.fail(e);
+        }
     }
 
     private boolean emitRow(Row row) {
         return downstream.setNextRow(row);
     }
 
-    @Override
-    public void downstream(Projector downstream) {
+    public void downstream(RowDownstream downstream) {
         assert downstream != null : "downstream must not be null";
-        downstream.registerUpstream(this);
-        this.downstream = downstream;
+        this.downstream = downstream.registerUpstream(this);
     }
+
 }