beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tg...@apache.org
Subject [2/2] beam git commit: Explicitly pass Pipeline in AppliedPTransform
Date Thu, 09 Feb 2017 21:22:12 GMT
Explicitly pass Pipeline in AppliedPTransform


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/74d977a7
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/74d977a7
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/74d977a7

Branch: refs/heads/master
Commit: 74d977a7d6e3b50b50c5e20f34ab17af5bd5dd91
Parents: 9ec22f1
Author: Thomas Groh <tgroh@google.com>
Authored: Tue Feb 7 09:58:16 2017 -0800
Committer: Thomas Groh <tgroh@google.com>
Committed: Thu Feb 9 13:21:53 2017 -0800

----------------------------------------------------------------------
 .../runners/core/PTransformMatchersTest.java    | 20 +++++++++++++-----
 .../runners/direct/CommittedResultTest.java     | 17 +++++++++------
 .../direct/WindowEvaluatorFactoryTest.java      |  7 ++++++-
 .../dataflow/DataflowPipelineJobTest.java       |  9 +++++++-
 .../beam/sdk/runners/TransformHierarchy.java    |  7 ++++++-
 .../beam/sdk/transforms/AppliedPTransform.java  | 22 +++++++++++---------
 6 files changed, 58 insertions(+), 24 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/74d977a7/runners/core-java/src/test/java/org/apache/beam/runners/core/PTransformMatchersTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/PTransformMatchersTest.java
b/runners/core-java/src/test/java/org/apache/beam/runners/core/PTransformMatchersTest.java
index c286a37..fe0c449 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/PTransformMatchersTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/PTransformMatchersTest.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.beam.runners.core.runnerapi;
+package org.apache.beam.runners.core;
 
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
@@ -24,7 +24,6 @@ import static org.hamcrest.Matchers.not;
 import static org.junit.Assert.assertThat;
 
 import java.io.Serializable;
-import org.apache.beam.runners.core.PTransformMatchers;
 import org.apache.beam.sdk.runners.PTransformMatcher;
 import org.apache.beam.sdk.testing.TestPipeline;
 import org.apache.beam.sdk.transforms.AppliedPTransform;
@@ -60,7 +59,11 @@ public class PTransformMatchersTest implements Serializable {
     });
     PCollection<Integer> output = input.apply(pardo);
 
-    AppliedPTransform<?, ?, ?> application = AppliedPTransform.of("DoStuff", input,
output, pardo);
+    AppliedPTransform<?, ?, ?> application =
+        AppliedPTransform
+            .<PCollection<Integer>, PCollection<Integer>,
+                PTransform<? super PCollection<Integer>, PCollection<Integer>>>
+                of("DoStuff", input.expand(), output.expand(), pardo, p);
 
     assertThat(matcher.matches(application), is(true));
   }
@@ -82,7 +85,10 @@ public class PTransformMatchersTest implements Serializable {
     PCollection<Integer> output = input.apply(subclass);
 
     AppliedPTransform<?, ?, ?> application =
-        AppliedPTransform.of("DoStuff", input, output, subclass);
+        AppliedPTransform
+            .<PCollection<Integer>, PCollection<Integer>,
+                PTransform<PCollection<Integer>, PCollection<Integer>>>
+                of("DoStuff", input.expand(), output.expand(), subclass, p);
 
     assertThat(matcher.matches(application), is(false));
   }
@@ -94,7 +100,11 @@ public class PTransformMatchersTest implements Serializable {
     Window.Bound<Integer> window = Window.into(new GlobalWindows());
     PCollection<Integer> output = input.apply(window);
 
-    AppliedPTransform<?, ?, ?> application = AppliedPTransform.of("DoStuff", input,
output, window);
+    AppliedPTransform<?, ?, ?> application =
+        AppliedPTransform
+            .<PCollection<Integer>, PCollection<Integer>,
+                PTransform<PCollection<Integer>, PCollection<Integer>>>
+                of("DoStuff", input.expand(), output.expand(), window, p);
 
     assertThat(matcher.matches(application), is(false));
   }

http://git-wip-us.apache.org/repos/asf/beam/blob/74d977a7/runners/direct-java/src/test/java/org/apache/beam/runners/direct/CommittedResultTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/CommittedResultTest.java
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/CommittedResultTest.java
index 736f554..68d6eba 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/CommittedResultTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/CommittedResultTest.java
@@ -54,12 +54,17 @@ public class CommittedResultTest implements Serializable {
 
   private transient PCollection<Integer> created = p.apply(Create.of(1, 2));
   private transient AppliedPTransform<?, ?, ?> transform =
-      AppliedPTransform.of("foo", p.begin(), PDone.in(p), new PTransform<PBegin, PDone>()
{
-        @Override
-        public PDone expand(PBegin begin) {
-          throw new IllegalArgumentException("Should never be applied");
-        }
-      });
+      AppliedPTransform.<PBegin, PDone, PTransform<PBegin, PDone>>of(
+          "foo",
+          p.begin().expand(),
+          PDone.in(p).expand(),
+          new PTransform<PBegin, PDone>() {
+            @Override
+            public PDone expand(PBegin begin) {
+              throw new IllegalArgumentException("Should never be applied");
+            }
+          },
+          p);
   private transient BundleFactory bundleFactory = ImmutableListBundleFactory.create();
 
   @Test

http://git-wip-us.apache.org/repos/asf/beam/blob/74d977a7/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WindowEvaluatorFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WindowEvaluatorFactoryTest.java
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WindowEvaluatorFactoryTest.java
index 9d0c68d..aa841ed 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WindowEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WindowEvaluatorFactoryTest.java
@@ -34,6 +34,7 @@ import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.testing.TestPipeline;
 import org.apache.beam.sdk.transforms.AppliedPTransform;
 import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.windowing.AfterPane;
 import org.apache.beam.sdk.transforms.windowing.AfterWatermark;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
@@ -311,7 +312,11 @@ public class WindowEvaluatorFactoryTest {
       throws Exception {
     TransformEvaluator<Long> evaluator =
         factory.forApplication(
-            AppliedPTransform.of("Window", input, windowed, windowTransform), inputBundle);
+            AppliedPTransform
+                .<PCollection<Long>, PCollection<Long>,
+                    PTransform<PCollection<Long>, PCollection<Long>>>
+                    of("Window", input.expand(), windowed.expand(), windowTransform, p),
+            inputBundle);
 
     evaluator.processElement(valueInGlobalWindow);
     evaluator.processElement(valueInGlobalAndTwoIntervalWindows);

http://git-wip-us.apache.org/repos/asf/beam/blob/74d977a7/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineJobTest.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineJobTest.java
b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineJobTest.java
index 36bf129..2690e71 100644
--- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineJobTest.java
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineJobTest.java
@@ -52,6 +52,7 @@ import com.google.common.collect.ImmutableSetMultimap;
 import java.io.IOException;
 import java.math.BigDecimal;
 import java.net.SocketTimeoutException;
+import java.util.Collections;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import org.apache.beam.runners.dataflow.testing.TestDataflowPipelineOptions;
@@ -71,6 +72,7 @@ import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.Sum;
 import org.apache.beam.sdk.values.PInput;
 import org.apache.beam.sdk.values.POutput;
+import org.apache.beam.sdk.values.TaggedPValue;
 import org.joda.time.Duration;
 import org.junit.Before;
 import org.junit.Rule;
@@ -671,7 +673,12 @@ public class DataflowPipelineJobTest {
       String fullName, PTransform<PInput, POutput> transform, Pipeline p) {
     PInput input = mock(PInput.class);
     when(input.getPipeline()).thenReturn(p);
-    return AppliedPTransform.of(fullName, input, mock(POutput.class), transform);
+    return AppliedPTransform.of(
+        fullName,
+        Collections.<TaggedPValue>emptyList(),
+        Collections.<TaggedPValue>emptyList(),
+        transform,
+        p);
   }
 
 

http://git-wip-us.apache.org/repos/asf/beam/blob/74d977a7/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/TransformHierarchy.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/TransformHierarchy.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/TransformHierarchy.java
index dc8f823..a4c28b8 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/TransformHierarchy.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/TransformHierarchy.java
@@ -323,7 +323,12 @@ public class TransformHierarchy {
      * Returns the {@link AppliedPTransform} representing this {@link Node}.
      */
     public AppliedPTransform<?, ?, ?> toAppliedPTransform() {
-      return AppliedPTransform.of(getFullName(), input, output, (PTransform) getTransform());
+      return AppliedPTransform.of(
+          getFullName(),
+          input.expand(),
+          output.expand(),
+          (PTransform) getTransform(),
+          input.getPipeline());
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/beam/blob/74d977a7/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/AppliedPTransform.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/AppliedPTransform.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/AppliedPTransform.java
index a6d8859..4de81ac 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/AppliedPTransform.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/AppliedPTransform.java
@@ -30,23 +30,25 @@ import org.apache.beam.sdk.values.TaggedPValue;
  *
  * <p>For internal use.
  *
- * @param <InputT> transform input type
- * @param <OutputT> transform output type
+ * @param <InputT>     transform input type
+ * @param <OutputT>    transform output type
  * @param <TransformT> transform type
  */
 @AutoValue
-public abstract class AppliedPTransform
-    <InputT extends PInput, OutputT extends POutput,
-     TransformT extends PTransform<? super InputT, OutputT>> {
+public abstract class AppliedPTransform<
+    InputT extends PInput, OutputT extends POutput,
+    TransformT extends PTransform<? super InputT, OutputT>> {
 
-  public static <
-          InputT extends PInput,
-          OutputT extends POutput,
+  public static <InputT extends PInput, OutputT extends POutput,
           TransformT extends PTransform<? super InputT, OutputT>>
       AppliedPTransform<InputT, OutputT, TransformT> of(
-          String fullName, InputT input, OutputT output, TransformT transform) {
+          String fullName,
+          List<TaggedPValue> input,
+          List<TaggedPValue> output,
+          TransformT transform,
+          Pipeline p) {
     return new AutoValue_AppliedPTransform<InputT, OutputT, TransformT>(
-        fullName, input.expand(), output.expand(), transform, input.getPipeline());
+        fullName, input, output, transform, p);
   }
 
   public abstract String getFullName();


Mime
View raw message