You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by tg...@apache.org on 2017/04/05 16:01:05 UTC
[1/2] beam git commit: This closes #2373
Repository: beam
Updated Branches:
refs/heads/master 646cbdb70 -> 645d0bba9
This closes #2373
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/645d0bba
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/645d0bba
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/645d0bba
Branch: refs/heads/master
Commit: 645d0bba936de8a3ed1d39fc3f7896767b3f9c0b
Parents: 646cbdb 496db38
Author: Thomas Groh <tg...@google.com>
Authored: Wed Apr 5 09:00:41 2017 -0700
Committer: Thomas Groh <tg...@google.com>
Committed: Wed Apr 5 09:00:41 2017 -0700
----------------------------------------------------------------------
.../core/construction/PTransformMatchers.java | 18 ++++++++
.../construction/PTransformMatchersTest.java | 47 ++++++++++++++++++++
2 files changed, 65 insertions(+)
----------------------------------------------------------------------
[2/2] beam git commit: Add CreateViewWithViewFn PTransformMatcher
Posted by tg...@apache.org.
Add CreateViewWithViewFn PTransformMatcher
This matcher is to stop matching the composite View.asFoo PTransforms,
as a PipelineRunner should support materialization of arbitrary
PTransforms created with the CreatePCollectionView transform.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/496db385
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/496db385
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/496db385
Branch: refs/heads/master
Commit: 496db3858f2ceaf037bdf7a63a6cbff8d62af4dc
Parents: 646cbdb
Author: Thomas Groh <tg...@google.com>
Authored: Thu Mar 30 11:21:19 2017 -0700
Committer: Thomas Groh <tg...@google.com>
Committed: Wed Apr 5 09:00:41 2017 -0700
----------------------------------------------------------------------
.../core/construction/PTransformMatchers.java | 18 ++++++++
.../construction/PTransformMatchersTest.java | 47 ++++++++++++++++++++
2 files changed, 65 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/496db385/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformMatchers.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformMatchers.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformMatchers.java
index f4ae577..945f3b2 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformMatchers.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformMatchers.java
@@ -29,9 +29,12 @@ import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.Flatten;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.View.CreatePCollectionView;
+import org.apache.beam.sdk.transforms.ViewFn;
import org.apache.beam.sdk.transforms.reflect.DoFnSignature;
import org.apache.beam.sdk.transforms.reflect.DoFnSignature.ProcessElementMethod;
import org.apache.beam.sdk.transforms.reflect.DoFnSignatures;
+import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PValue;
import org.apache.beam.sdk.values.TaggedPValue;
@@ -203,6 +206,21 @@ public class PTransformMatchers {
};
}
+ public static PTransformMatcher createViewWithViewFn(final Class<? extends ViewFn> viewFnType) {
+ return new PTransformMatcher() {
+ @Override
+ public boolean matches(AppliedPTransform<?, ?, ?> application) {
+ if (!(application.getTransform() instanceof CreatePCollectionView)) {
+ return false;
+ }
+ CreatePCollectionView<?, ?> createView =
+ (CreatePCollectionView<?, ?>) application.getTransform();
+ ViewFn<Iterable<WindowedValue<?>>, ?> viewFn = createView.getView().getViewFn();
+ return viewFn.getClass().equals(viewFnType);
+ }
+ };
+ }
+
/**
* A {@link PTransformMatcher} which matches a {@link Flatten.PCollections} which
* consumes no input {@link PCollection PCollections}.
http://git-wip-us.apache.org/repos/asf/beam/blob/496db385/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformMatchersTest.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformMatchersTest.java b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformMatchersTest.java
index 484aba4..1d12194 100644
--- a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformMatchersTest.java
+++ b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformMatchersTest.java
@@ -41,13 +41,18 @@ import org.apache.beam.sdk.transforms.Flatten;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Sum;
+import org.apache.beam.sdk.transforms.View;
+import org.apache.beam.sdk.transforms.View.CreatePCollectionView;
+import org.apache.beam.sdk.transforms.ViewFn;
import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
import org.apache.beam.sdk.transforms.windowing.GlobalWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.util.PCollectionViews;
import org.apache.beam.sdk.util.TimeDomain;
import org.apache.beam.sdk.util.Timer;
import org.apache.beam.sdk.util.TimerSpec;
import org.apache.beam.sdk.util.TimerSpecs;
+import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.util.WindowingStrategy;
import org.apache.beam.sdk.util.state.StateSpec;
import org.apache.beam.sdk.util.state.StateSpecs;
@@ -56,6 +61,7 @@ import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollection.IsBounded;
import org.apache.beam.sdk.values.PCollectionList;
+import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.PDone;
import org.apache.beam.sdk.values.TaggedPValue;
import org.apache.beam.sdk.values.TupleTag;
@@ -325,6 +331,47 @@ public class PTransformMatchersTest implements Serializable {
}
@Test
+ public void createViewWithViewFn() {
+ PCollection<Integer> input = p.apply(Create.of(1));
+ PCollectionView<Iterable<Integer>> view =
+ PCollectionViews.iterableView(input, input.getWindowingStrategy(), input.getCoder());
+ ViewFn<Iterable<WindowedValue<?>>, Iterable<Integer>> viewFn = view.getViewFn();
+ CreatePCollectionView<?, ?> createView = CreatePCollectionView.of(view);
+
+ PTransformMatcher matcher = PTransformMatchers.createViewWithViewFn(viewFn.getClass());
+ assertThat(matcher.matches(getAppliedTransform(createView)), is(true));
+ }
+
+ @Test
+ public void createViewWithViewFnDifferentViewFn() {
+ PCollection<Integer> input = p.apply(Create.of(1));
+ PCollectionView<Iterable<Integer>> view =
+ PCollectionViews.iterableView(input, input.getWindowingStrategy(), input.getCoder());
+ ViewFn<Iterable<WindowedValue<?>>, Iterable<Integer>> viewFn =
+ new ViewFn<Iterable<WindowedValue<?>>, Iterable<Integer>>() {
+ @Override
+ public Iterable<Integer> apply(Iterable<WindowedValue<?>> contents) {
+ return Collections.emptyList();
+ }
+ };
+ CreatePCollectionView<?, ?> createView = CreatePCollectionView.of(view);
+
+ PTransformMatcher matcher = PTransformMatchers.createViewWithViewFn(viewFn.getClass());
+ assertThat(matcher.matches(getAppliedTransform(createView)), is(false));
+ }
+
+ @Test
+ public void createViewWithViewFnNotCreatePCollectionView() {
+ PCollection<Integer> input = p.apply(Create.of(1));
+ PCollectionView<Iterable<Integer>> view =
+ PCollectionViews.iterableView(input, input.getWindowingStrategy(), input.getCoder());
+
+ PTransformMatcher matcher =
+ PTransformMatchers.createViewWithViewFn(view.getViewFn().getClass());
+ assertThat(matcher.matches(getAppliedTransform(View.asIterable())), is(false));
+ }
+
+ @Test
public void emptyFlattenWithEmptyFlatten() {
AppliedPTransform application =
AppliedPTransform