[BEAM-4394] Pull request review fixes.

seznam · May 21, 2018 · 013b078 · 013b078
1 parent 22810d1
commit 013b078
Show file tree

Hide file tree

Showing 6 changed files with 43 additions and 52 deletions.
diff --git a/...oria-beam/src/main/java/org/apache/beam/sdk/extensions/euphoria/beam/join/FullJoinFn.java b/...oria-beam/src/main/java/org/apache/beam/sdk/extensions/euphoria/beam/join/FullJoinFn.java
@@ -4,7 +4,6 @@
 import org.apache.beam.sdk.extensions.euphoria.core.client.functional.BinaryFunctor;
 import org.apache.beam.sdk.extensions.euphoria.core.client.util.Pair;
 import org.apache.beam.sdk.transforms.join.CoGbkResult;
-import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.TupleTag;
 
 /**
@@ -18,39 +17,34 @@ public FullJoinFn(BinaryFunctor<LeftT, RightT, OutputT> joiner, TupleTag<LeftT>
   }
 
   @Override
-  public void processElement(ProcessContext c) {
+  protected void doJoin(
+      ProcessContext c, K key, CoGbkResult value,
+      Iterable<LeftT> leftSideIter,
+      Iterable<RightT> rightSideIter) {
 
-    KV<K, CoGbkResult> element = c.element();
-    CoGbkResult value = element.getValue();
-    K key = element.getKey();
-
-    Iterable<LeftT> leftSideIter = value.getAll(leftTag);
-    Iterable<RightT> rightSIdeIter = value.getAll(rightTag);
+    boolean leftHasValues = leftSideIter.iterator().hasNext();
+    boolean rightHasValues = rightSideIter.iterator().hasNext();
 
     SingleValueCollector<OutputT> outCollector = new SingleValueCollector<>();
 
-    boolean leftHasValues = leftSideIter.iterator().hasNext();
-    boolean rightHasValues = rightSIdeIter.iterator().hasNext();
-
     if (leftHasValues && rightHasValues) {
-      for (RightT rightValue : rightSIdeIter) {
+      for (RightT rightValue : rightSideIter) {
         for (LeftT leftValue : leftSideIter) {
           joiner.apply(leftValue, rightValue, outCollector);
           c.output(Pair.of(key, outCollector.get()));
         }
       }
-    } else if (leftHasValues && !rightHasValues) {
+    } else if (leftHasValues) {
       for (LeftT leftValue : leftSideIter) {
         joiner.apply(leftValue, null, outCollector);
         c.output(Pair.of(key, outCollector.get()));
       }
-    } else if (!leftHasValues && rightHasValues) {
-      for (RightT rightValue : rightSIdeIter) {
+    } else if (rightHasValues) {
+      for (RightT rightValue : rightSideIter) {
         joiner.apply(null, rightValue, outCollector);
         c.output(Pair.of(key, outCollector.get()));
       }
     }
-
   }
 
   @Override

diff --git a/...ria-beam/src/main/java/org/apache/beam/sdk/extensions/euphoria/beam/join/InnerJoinFn.java b/...ria-beam/src/main/java/org/apache/beam/sdk/extensions/euphoria/beam/join/InnerJoinFn.java
@@ -4,7 +4,6 @@
 import org.apache.beam.sdk.extensions.euphoria.core.client.functional.BinaryFunctor;
 import org.apache.beam.sdk.extensions.euphoria.core.client.util.Pair;
 import org.apache.beam.sdk.transforms.join.CoGbkResult;
-import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.TupleTag;
 
 /**
@@ -20,14 +19,10 @@ public InnerJoinFn(
   }
 
   @Override
-  public void processElement(ProcessContext c) {
-
-    KV<K, CoGbkResult> element = c.element();
-    CoGbkResult value = element.getValue();
-    K key = element.getKey();
-
-    Iterable<LeftT> leftSideIter = value.getAll(leftTag);
-    Iterable<RightT> rightSideIter = value.getAll(rightTag);
+  protected void doJoin(
+      ProcessContext c, K key, CoGbkResult value,
+      Iterable<LeftT> leftSideIter,
+      Iterable<RightT> rightSideIter) {
 
     SingleValueCollector<OutputT> outCollector = new SingleValueCollector<>();
 

diff --git a/...euphoria-beam/src/main/java/org/apache/beam/sdk/extensions/euphoria/beam/join/JoinFn.java b/...euphoria-beam/src/main/java/org/apache/beam/sdk/extensions/euphoria/beam/join/JoinFn.java
@@ -31,7 +31,22 @@ protected JoinFn(
   }
 
   @ProcessElement
-  public abstract void processElement(ProcessContext c);
+  public final void processElement(ProcessContext c) {
+
+    KV<K, CoGbkResult> element = c.element();
+    CoGbkResult value = element.getValue();
+    K key = element.getKey();
+
+    Iterable<LeftT> leftSideIter = value.getAll(leftTag);
+    Iterable<RightT> rightSideIter = value.getAll(rightTag);
+
+    doJoin(c, key, value, leftSideIter, rightSideIter);
+  }
+
+  protected abstract void doJoin(
+      ProcessContext c, K key, CoGbkResult value,
+      Iterable<LeftT> leftSideIter,
+      Iterable<RightT> rightSideIter);
 
   public abstract String getFnName();
 }
diff --git a/...beam/src/main/java/org/apache/beam/sdk/extensions/euphoria/beam/join/LeftOuterJoinFn.java b/...beam/src/main/java/org/apache/beam/sdk/extensions/euphoria/beam/join/LeftOuterJoinFn.java
@@ -4,7 +4,6 @@
 import org.apache.beam.sdk.extensions.euphoria.core.client.functional.BinaryFunctor;
 import org.apache.beam.sdk.extensions.euphoria.core.client.util.Pair;
 import org.apache.beam.sdk.transforms.join.CoGbkResult;
-import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.TupleTag;
 
 
@@ -21,30 +20,24 @@ public LeftOuterJoinFn(
   }
 
   @Override
-  public void processElement(ProcessContext c) {
-
-    KV<K, CoGbkResult> element = c.element();
-    CoGbkResult value = element.getValue();
-    K key = element.getKey();
-
-    Iterable<LeftT> leftSideIter = value.getAll(leftTag);
-    Iterable<RightT> rightSIdeIter = value.getAll(rightTag);
+  protected void doJoin(
+      ProcessContext c, K key, CoGbkResult value,
+      Iterable<LeftT> leftSideIter,
+      Iterable<RightT> rightSideIter) {
 
     SingleValueCollector<OutputT> outCollector = new SingleValueCollector<>();
 
     for (LeftT leftValue : leftSideIter) {
-      if (rightSIdeIter.iterator().hasNext()) {
-        for (RightT rightValue : rightSIdeIter) {
+      if (rightSideIter.iterator().hasNext()) {
+        for (RightT rightValue : rightSideIter) {
           joiner.apply(leftValue, rightValue, outCollector);
           c.output(Pair.of(key, outCollector.get()));
         }
       } else {
-          joiner.apply(leftValue, null, outCollector);
-          c.output(Pair.of(key, outCollector.get()));
+        joiner.apply(leftValue, null, outCollector);
+        c.output(Pair.of(key, outCollector.get()));
       }
     }
-
-
   }
 
   @Override

diff --git a/...eam/src/main/java/org/apache/beam/sdk/extensions/euphoria/beam/join/RightOuterJoinFn.java b/...eam/src/main/java/org/apache/beam/sdk/extensions/euphoria/beam/join/RightOuterJoinFn.java
@@ -4,7 +4,6 @@
 import org.apache.beam.sdk.extensions.euphoria.core.client.functional.BinaryFunctor;
 import org.apache.beam.sdk.extensions.euphoria.core.client.util.Pair;
 import org.apache.beam.sdk.transforms.join.CoGbkResult;
-import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.TupleTag;
 
 /**
@@ -20,18 +19,12 @@ public RightOuterJoinFn(
   }
 
   @Override
-  public void processElement(ProcessContext c) {
-
-    KV<K, CoGbkResult> element = c.element();
-    CoGbkResult value = element.getValue();
-    K key = element.getKey();
-
-    Iterable<LeftT> leftSideIter = value.getAll(leftTag);
-    Iterable<RightT> rightSIdeIter = value.getAll(rightTag);
+  protected void doJoin(ProcessContext c, K key, CoGbkResult value, Iterable<LeftT> leftSideIter,
+      Iterable<RightT> rightSideIter) {
 
     SingleValueCollector<OutputT> outCollector = new SingleValueCollector<>();
 
-    for (RightT rightValue : rightSIdeIter) {
+    for (RightT rightValue : rightSideIter) {
       if (leftSideIter.iterator().hasNext()) {
         for (LeftT leftValue : leftSideIter) {
           joiner.apply(leftValue, rightValue, outCollector);

diff --git a/...ia-beam/src/main/java/org/apache/beam/sdk/extensions/euphoria/beam/join/package-info.java b/...ia-beam/src/main/java/org/apache/beam/sdk/extensions/euphoria/beam/join/package-info.java
@@ -17,6 +17,7 @@
  */
 
 /**
- * {@link cz.seznam.euphoria.core.client.operator.Join} translation centered classes.
+ * {@link org.apache.beam.sdk.extensions.euphoria.core.client.operator.Join} translation centered
+ * classes.
  */
 package org.apache.beam.sdk.extensions.euphoria.beam.join;