Merge pull request apache#171 from seznam/158-extend-reduce-by-key-wi…

…th-value-sorting

158 extend reduce by key with value sorting

sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/io/Environment.java

@@ -61,4 +61,5 @@ public interface Environment {
    * @return Instance of a timer.
    */
   Timer getTimer(String name);
+
 }

sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/io/ExternalIterable.java

@@ -0,0 +1,29 @@
+/**
+ * Copyright 2016-2017 Seznam.cz, a.s.
+ *
+ * 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 cz.seznam.euphoria.core.client.io;
+
+import cz.seznam.euphoria.core.annotation.audience.Audience;
+import java.io.Closeable;
+
+/**
+ * An {@code Iterable} that is externalized to external storage
+ * (i.e. spilled to disk).
+ */
+@Audience(Audience.Type.EXECUTOR)
+public interface ExternalIterable<T> extends Iterable<T>, Closeable {
+
+
+}

sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/io/SpillTools.java

@@ -0,0 +1,85 @@
+/**
+ * Copyright 2016-2017 Seznam.cz, a.s.
+ *
+ * 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 cz.seznam.euphoria.core.client.io;
+
+import cz.seznam.euphoria.core.annotation.audience.Audience;
+import cz.seznam.euphoria.core.util.IOUtils;
+import cz.seznam.euphoria.shaded.guava.com.google.common.collect.Iterables;
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.Iterator;
+
+/**
+ * Tools that can be used to externalize a dataset to local storage.
+ */
+@Audience(Audience.Type.CLIENT)
+public interface SpillTools extends Serializable {
+
+  /**
+   * Simply externalize {@code Iterable}.
+   *
+   * @param <T> type of input
+   * @param what an {@code Iterable} that is to be externalized
+   * @return {@code ExternalIterable} that is backed by policy implemented
+   * by this interface
+   */
+  <T> ExternalIterable<T> externalize(Iterable<T> what);
+
+  /**
+   * Externalize and sort given {@code Iterable} to
+   * sorted parts. These parts can then be merge sorted.
+   *
+   * @param <T> type of input
+   * @param what the {@code Iterable} that is to be split and sorted
+   * @param comparator the {@code Comparator} to use for sorting
+   * @return collection of externalized iterables that are sorted
+   */
+  <T> Collection<ExternalIterable<T>> spillAndSortParts(
+      Iterable<T> what, Comparator<T> comparator);
+
+  /**
+   * Use external sort to return given {@code Iterable} sorted according
+   * to given comparator.
+   *
+   * @param <T> type of data in the {@code Iterable}.
+   * @param what the {@code Iterable} to external sort
+   * @param comparator the comparator to use when sorting
+   * @return the sorted {@code Iterable}
+   */
+  default <T> ExternalIterable<T> sorted(Iterable<T> what, Comparator<T> comparator) {
+
+    Collection<ExternalIterable<T>> parts = spillAndSortParts(what, comparator);
+    Iterable<T> ret = Iterables.mergeSorted(parts, comparator);
+
+    return new ExternalIterable<T>() {
+
+      @Override
+      public Iterator<T> iterator() {
+        return ret.iterator();
+      }
+
+      @Override
+      public void close() throws IOException {
+        IOUtils.forEach(parts, Closeable::close);
+      }
+
+    };
+  }
+
+}

sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/Join.java

@@ -30,6 +30,7 @@
 import cz.seznam.euphoria.core.client.operator.state.ListStorage;
 import cz.seznam.euphoria.core.client.operator.state.ListStorageDescriptor;
 import cz.seznam.euphoria.core.client.operator.state.State;
+import cz.seznam.euphoria.core.client.operator.state.StateContext;
 import cz.seznam.euphoria.core.client.operator.state.StorageProvider;
 import cz.seznam.euphoria.core.client.util.Either;
 import cz.seznam.euphoria.core.client.util.Pair;
@@ -531,11 +532,12 @@ public List<JoinHint> getHints() {
         keyExtractor,
         e -> e,
         getWindowing(),
-        (StorageProvider storages, Collector ctx) ->
-            ctx == null
-                ? new StableJoinState(storages)
-                : new EarlyEmittingJoinState(storages, ctx),
-        new StateSupport.MergeFromStateMerger<>());
+        (StateContext context, Collector ctx) -> {
+          StorageProvider storages = context.getStorageProvider();
+          return ctx == null
+              ? new StableJoinState(storages)
+              : new EarlyEmittingJoinState(storages, ctx);
+        }, new StateSupport.MergeFromStateMerger<>());
 
     DAG<Operator<?, ?>> dag = DAG.of(leftMap, rightMap);
     dag.add(union, leftMap, rightMap);