[incubator-nemo] branch master updated: [NEMO-217] Immediately emits elements in FlattenTransform (#120)

[johnyangk]

This is an automated email from the ASF dual-hosted git repository.

johnyangk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git


The following commit(s) were added to refs/heads/master by this push:
     new 223caf2  [NEMO-217] Immediately emits elements in FlattenTransform 
(#120)
223caf2 is described below

commit 223caf26298c820548b59a0a96eb409c9a829e86
Author: Taegeon Um <taegeo...@gmail.com>
AuthorDate: Thu Sep 13 14:52:39 2018 +0900

    [NEMO-217] Immediately emits elements in FlattenTransform (#120)
    
    JIRA: [NEMO-217: Immediately emits elements in 
FlattenTransform](https://issues.apache.org/jira/projects/NEMO/issues/NEMO-217)
    
    **Major changes:**
    - remove `ArrayList` that collects elements in `FlattenTransform` and 
directly emit the elements
    
    **Minor changes to note:**
    -
    
    **Tests for the changes:**
    -
    
    **Other comments:**
    -
    
    Closes #120
---
 .../nemo/compiler/frontend/beam/transform/FlattenTransform.java   | 8 +-------
 1 file changed, 1 insertion(+), 7 deletions(-)

diff --git 
a/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/transform/FlattenTransform.java
 
b/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/transform/FlattenTransform.java
index dd23c9b..2dd0e61 100644
--- 
a/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/transform/FlattenTransform.java
+++ 
b/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/transform/FlattenTransform.java
@@ -18,21 +18,17 @@ package org.apache.nemo.compiler.frontend.beam.transform;
 import org.apache.nemo.common.ir.OutputCollector;
 import org.apache.nemo.common.ir.vertex.transform.Transform;
 
-import java.util.ArrayList;
-
 /**
  * Flatten transform implementation.
  * @param <T> input/output type.
  */
 public final class FlattenTransform<T> implements Transform<T, T> {
-  private final ArrayList<T> collectedElements;
   private OutputCollector<T> outputCollector;
 
   /**
    * FlattenTransform Constructor.
    */
   public FlattenTransform() {
-    this.collectedElements = new ArrayList<>();
   }
 
   @Override
@@ -42,13 +38,11 @@ public final class FlattenTransform<T> implements 
Transform<T, T> {
 
   @Override
   public void onData(final T element) {
-    collectedElements.add(element);
+    outputCollector.emit(element);
   }
 
   @Override
   public void close() {
-    collectedElements.forEach(outputCollector::emit);
-    collectedElements.clear();
   }
 
   @Override