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 2016/12/06 18:47:04 UTC
[1/4] incubator-beam git commit: Remove getProducingTransformInternal
from DirectGraphVisitorTest
Repository: incubator-beam
Updated Branches:
refs/heads/master 077d9118d -> ca6ab6c68
Remove getProducingTransformInternal from DirectGraphVisitorTest
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/ec1eff38
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/ec1eff38
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/ec1eff38
Branch: refs/heads/master
Commit: ec1eff387a711039801289c8f59c4240b1f1d007
Parents: d6c6ad3
Author: Thomas Groh <tg...@google.com>
Authored: Fri Dec 2 14:26:04 2016 -0800
Committer: Thomas Groh <tg...@google.com>
Committed: Tue Dec 6 10:46:39 2016 -0800
----------------------------------------------------------------------
.../runners/direct/DirectGraphVisitorTest.java | 70 +++++++++++++-------
1 file changed, 47 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/ec1eff38/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectGraphVisitorTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectGraphVisitorTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectGraphVisitorTest.java
index fb84de8..5ad278b 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectGraphVisitorTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectGraphVisitorTest.java
@@ -19,25 +19,34 @@ package org.apache.beam.runners.direct;
import static org.hamcrest.Matchers.emptyIterable;
import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import com.google.common.collect.Iterables;
import java.io.Serializable;
+import java.util.ArrayList;
import java.util.List;
import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.CountingSource;
+import org.apache.beam.sdk.io.Read;
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.DoFn;
import org.apache.beam.sdk.transforms.Flatten;
+import org.apache.beam.sdk.transforms.Flatten.FlattenPCollectionList;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.View;
+import org.apache.beam.sdk.values.PBegin;
import org.apache.beam.sdk.values.PCollection;
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.PInput;
+import org.apache.beam.sdk.values.POutput;
import org.hamcrest.Matchers;
import org.junit.Rule;
import org.junit.Test;
@@ -48,7 +57,6 @@ import org.junit.runners.JUnit4;
/**
* Tests for {@link DirectGraphVisitor}.
*/
-// TODO: Replace uses of getProducing
@RunWith(JUnit4.class)
public class DirectGraphVisitorTest implements Serializable {
@Rule public transient ExpectedException thrown = ExpectedException.none();
@@ -81,26 +89,36 @@ public class DirectGraphVisitorTest implements Serializable {
@Test
public void getRootTransformsContainsPBegins() {
PCollection<String> created = p.apply(Create.of("foo", "bar"));
- PCollection<Long> counted = p.apply(CountingInput.upTo(1234L));
+ PCollection<Long> counted = p.apply(Read.from(CountingSource.upTo(1234L)));
PCollection<Long> unCounted = p.apply(CountingInput.unbounded());
p.traverseTopologically(visitor);
+ DirectGraph graph = visitor.getGraph();
+ assertThat(graph.getRootTransforms(), hasSize(3));
+ List<PTransform<?, ?>> unapplied = new ArrayList<>();
assertThat(
- visitor.getGraph().getRootTransforms(),
+ graph.getRootTransforms(),
Matchers.<AppliedPTransform<?, ?, ?>>containsInAnyOrder(
- created.getProducingTransformInternal(),
- counted.getProducingTransformInternal(),
- unCounted.getProducingTransformInternal()));
+ graph.getProducer(created), graph.getProducer(counted), graph.getProducer(unCounted)));
+ for (AppliedPTransform<?, ?, ?> root : graph.getRootTransforms()) {
+ assertTrue(root.getInput() instanceof PBegin);
+ assertThat(root.getOutput(), Matchers.<POutput>isOneOf(created, counted, unCounted));
+ }
}
@Test
public void getRootTransformsContainsEmptyFlatten() {
- PCollection<String> empty =
- PCollectionList.<String>empty(p).apply(Flatten.<String>pCollections());
+ FlattenPCollectionList<String> flatten = Flatten.pCollections();
+ PCollectionList<String> emptyList = PCollectionList.empty(p);
+ PCollection<String> empty = emptyList.apply(flatten);
p.traverseTopologically(visitor);
+ DirectGraph graph = visitor.getGraph();
assertThat(
- visitor.getGraph().getRootTransforms(),
- Matchers.<AppliedPTransform<?, ?, ?>>containsInAnyOrder(
- empty.getProducingTransformInternal()));
+ graph.getRootTransforms(),
+ Matchers.<AppliedPTransform<?, ?, ?>>containsInAnyOrder(graph.getProducer(empty)));
+ AppliedPTransform<?, ?, ?> onlyRoot = Iterables.getOnlyElement(graph.getRootTransforms());
+ assertThat(onlyRoot.getTransform(), Matchers.<PTransform<?, ?>>equalTo(flatten));
+ assertThat(onlyRoot.getInput(), Matchers.<PInput>equalTo(emptyList));
+ assertThat(onlyRoot.getOutput(), Matchers.<POutput>equalTo(empty));
}
@Test
@@ -122,16 +140,20 @@ public class DirectGraphVisitorTest implements Serializable {
p.traverseTopologically(visitor);
+ DirectGraph graph = visitor.getGraph();
+ AppliedPTransform<?, ?, ?> transformedProducer =
+ graph.getProducer(transformed);
+ AppliedPTransform<?, ?, ?> flattenedProducer =
+ graph.getProducer(flattened);
+
assertThat(
- visitor.getGraph().getPrimitiveConsumers(created),
+ graph.getPrimitiveConsumers(created),
Matchers.<AppliedPTransform<?, ?, ?>>containsInAnyOrder(
- transformed.getProducingTransformInternal(),
- flattened.getProducingTransformInternal()));
+ transformedProducer, flattenedProducer));
assertThat(
- visitor.getGraph().getPrimitiveConsumers(transformed),
- Matchers.<AppliedPTransform<?, ?, ?>>containsInAnyOrder(
- flattened.getProducingTransformInternal()));
- assertThat(visitor.getGraph().getPrimitiveConsumers(flattened), emptyIterable());
+ graph.getPrimitiveConsumers(transformed),
+ Matchers.<AppliedPTransform<?, ?, ?>>containsInAnyOrder(flattenedProducer));
+ assertThat(graph.getPrimitiveConsumers(flattened), emptyIterable());
}
@Test
@@ -143,12 +165,14 @@ public class DirectGraphVisitorTest implements Serializable {
p.traverseTopologically(visitor);
+ DirectGraph graph = visitor.getGraph();
+ AppliedPTransform<?, ?, ?> flattenedProducer = graph.getProducer(flattened);
+
assertThat(
- visitor.getGraph().getPrimitiveConsumers(created),
- Matchers.<AppliedPTransform<?, ?, ?>>containsInAnyOrder(
- flattened.getProducingTransformInternal(),
- flattened.getProducingTransformInternal()));
- assertThat(visitor.getGraph().getPrimitiveConsumers(flattened), emptyIterable());
+ graph.getPrimitiveConsumers(created),
+ Matchers.<AppliedPTransform<?, ?, ?>>containsInAnyOrder(flattenedProducer,
+ flattenedProducer));
+ assertThat(graph.getPrimitiveConsumers(flattened), emptyIterable());
}
@Test
[4/4] incubator-beam git commit: This closes #1495
Posted by tg...@apache.org.
This closes #1495
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/ca6ab6c6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/ca6ab6c6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/ca6ab6c6
Branch: refs/heads/master
Commit: ca6ab6c68d807765ab8e80273dd9d6f99aeaaf4a
Parents: 077d911 ec1eff3
Author: Thomas Groh <tg...@google.com>
Authored: Tue Dec 6 10:46:40 2016 -0800
Committer: Thomas Groh <tg...@google.com>
Committed: Tue Dec 6 10:46:40 2016 -0800
----------------------------------------------------------------------
.../direct/BoundedReadEvaluatorFactoryTest.java | 18 +-
.../runners/direct/DirectGraphVisitorTest.java | 69 ++++--
.../beam/runners/direct/DirectGraphs.java | 35 +++
.../runners/direct/EvaluationContextTest.java | 82 ++++---
.../direct/FlattenEvaluatorFactoryTest.java | 15 +-
.../direct/GroupByKeyEvaluatorFactoryTest.java | 2 +-
.../GroupByKeyOnlyEvaluatorFactoryTest.java | 3 +-
.../ImmutabilityEnforcementFactoryTest.java | 2 +-
.../beam/runners/direct/ParDoEvaluatorTest.java | 3 +-
.../StatefulParDoEvaluatorFactoryTest.java | 4 +-
.../runners/direct/StepTransformResultTest.java | 2 +-
.../direct/TestStreamEvaluatorFactoryTest.java | 14 +-
.../runners/direct/TransformExecutorTest.java | 9 +-
.../UnboundedReadEvaluatorFactoryTest.java | 24 +-
.../direct/ViewEvaluatorFactoryTest.java | 4 +-
.../direct/WatermarkCallbackExecutorTest.java | 6 +-
.../runners/direct/WatermarkManagerTest.java | 237 ++++++++-----------
17 files changed, 292 insertions(+), 237 deletions(-)
----------------------------------------------------------------------
[3/4] incubator-beam git commit: Add DirectGraphs to DirectRunner
Tests
Posted by tg...@apache.org.
Add DirectGraphs to DirectRunner Tests
Add getGraph(Pipeline) and getProducer(PValue), which use the
DirectGraphVisitor and DirectGraph methods to provide access to the
producing AppliedPTransform.
Remove getProducingTransformInternal from everywhere except
DirectGraphVisitorTest
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/d6c6ad37
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/d6c6ad37
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/d6c6ad37
Branch: refs/heads/master
Commit: d6c6ad37149622e4d35af39727cdf774e6263d1e
Parents: 077d911
Author: Thomas Groh <tg...@google.com>
Authored: Fri Dec 2 10:56:36 2016 -0800
Committer: Thomas Groh <tg...@google.com>
Committed: Tue Dec 6 10:46:39 2016 -0800
----------------------------------------------------------------------
.../direct/BoundedReadEvaluatorFactoryTest.java | 18 +-
.../runners/direct/DirectGraphVisitorTest.java | 1 +
.../beam/runners/direct/DirectGraphs.java | 35 +++
.../runners/direct/EvaluationContextTest.java | 82 ++++---
.../direct/FlattenEvaluatorFactoryTest.java | 15 +-
.../direct/GroupByKeyEvaluatorFactoryTest.java | 2 +-
.../GroupByKeyOnlyEvaluatorFactoryTest.java | 3 +-
.../ImmutabilityEnforcementFactoryTest.java | 2 +-
.../beam/runners/direct/ParDoEvaluatorTest.java | 3 +-
.../StatefulParDoEvaluatorFactoryTest.java | 4 +-
.../runners/direct/StepTransformResultTest.java | 2 +-
.../direct/TestStreamEvaluatorFactoryTest.java | 14 +-
.../runners/direct/TransformExecutorTest.java | 9 +-
.../UnboundedReadEvaluatorFactoryTest.java | 24 +-
.../direct/ViewEvaluatorFactoryTest.java | 4 +-
.../direct/WatermarkCallbackExecutorTest.java | 6 +-
.../runners/direct/WatermarkManagerTest.java | 237 ++++++++-----------
17 files changed, 246 insertions(+), 215 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d6c6ad37/runners/direct-java/src/test/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactoryTest.java
index b1ff689..acb1444 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactoryTest.java
@@ -80,6 +80,7 @@ public class BoundedReadEvaluatorFactoryTest {
private BoundedReadEvaluatorFactory factory;
@Mock private EvaluationContext context;
private BundleFactory bundleFactory;
+ private AppliedPTransform<?, ?, ?> longsProducer;
@Before
public void setup() {
@@ -92,6 +93,7 @@ public class BoundedReadEvaluatorFactoryTest {
new BoundedReadEvaluatorFactory(
context, Long.MAX_VALUE /* minimum size for dynamic splits */);
bundleFactory = ImmutableListBundleFactory.create();
+ longsProducer = DirectGraphs.getProducer(longs);
}
@Test
@@ -102,11 +104,11 @@ public class BoundedReadEvaluatorFactoryTest {
Collection<CommittedBundle<?>> initialInputs =
new BoundedReadEvaluatorFactory.InputProvider(context)
- .getInitialInputs(longs.getProducingTransformInternal(), 1);
+ .getInitialInputs(longsProducer, 1);
List<WindowedValue<?>> outputs = new ArrayList<>();
for (CommittedBundle<?> shardBundle : initialInputs) {
TransformEvaluator<?> evaluator =
- factory.forApplication(longs.getProducingTransformInternal(), null);
+ factory.forApplication(longsProducer, null);
for (WindowedValue<?> shard : shardBundle.getElements()) {
evaluator.processElement((WindowedValue) shard);
}
@@ -141,7 +143,7 @@ public class BoundedReadEvaluatorFactoryTest {
}
PCollection<Long> read =
TestPipeline.create().apply(Read.from(new TestSource<>(VarLongCoder.of(), 5, elems)));
- AppliedPTransform<?, ?, ?> transform = read.getProducingTransformInternal();
+ AppliedPTransform<?, ?, ?> transform = DirectGraphs.getProducer(read);
Collection<CommittedBundle<?>> unreadInputs =
new BoundedReadEvaluatorFactory.InputProvider(context).getInitialInputs(transform, 1);
@@ -191,7 +193,7 @@ public class BoundedReadEvaluatorFactoryTest {
PCollection<Long> read =
TestPipeline.create()
.apply(Read.from(SourceTestUtils.toUnsplittableSource(CountingSource.upTo(10L))));
- AppliedPTransform<?, ?, ?> transform = read.getProducingTransformInternal();
+ AppliedPTransform<?, ?, ?> transform = DirectGraphs.getProducer(read);
when(context.createRootBundle()).thenReturn(bundleFactory.createRootBundle());
when(context.createRootBundle()).thenReturn(bundleFactory.createRootBundle());
@@ -238,7 +240,7 @@ public class BoundedReadEvaluatorFactoryTest {
});
Collection<CommittedBundle<?>> initialInputs =
new BoundedReadEvaluatorFactory.InputProvider(context)
- .getInitialInputs(longs.getProducingTransformInternal(), 3);
+ .getInitialInputs(longsProducer, 3);
assertThat(initialInputs, hasSize(allOf(greaterThanOrEqualTo(3), lessThanOrEqualTo(4))));
@@ -271,7 +273,7 @@ public class BoundedReadEvaluatorFactoryTest {
CommittedBundle<BoundedSourceShard<Long>> shards = rootBundle.commit(Instant.now());
TransformEvaluator<BoundedSourceShard<Long>> evaluator =
- factory.forApplication(longs.getProducingTransformInternal(), shards);
+ factory.forApplication(longsProducer, shards);
for (WindowedValue<BoundedSourceShard<Long>> shard : shards.getElements()) {
UncommittedBundle<Long> outputBundle = bundleFactory.createBundle(longs);
when(context.createBundle(longs)).thenReturn(outputBundle);
@@ -299,7 +301,7 @@ public class BoundedReadEvaluatorFactoryTest {
TestPipeline p = TestPipeline.create();
PCollection<Long> pcollection = p.apply(Read.from(source));
- AppliedPTransform<?, ?, ?> sourceTransform = pcollection.getProducingTransformInternal();
+ AppliedPTransform<?, ?, ?> sourceTransform = DirectGraphs.getProducer(pcollection);
UncommittedBundle<Long> output = bundleFactory.createBundle(pcollection);
when(context.createBundle(pcollection)).thenReturn(output);
@@ -320,7 +322,7 @@ public class BoundedReadEvaluatorFactoryTest {
TestPipeline p = TestPipeline.create();
PCollection<Long> pcollection = p.apply(Read.from(source));
- AppliedPTransform<?, ?, ?> sourceTransform = pcollection.getProducingTransformInternal();
+ AppliedPTransform<?, ?, ?> sourceTransform = DirectGraphs.getProducer(pcollection);
UncommittedBundle<Long> output = bundleFactory.createBundle(pcollection);
when(context.createBundle(pcollection)).thenReturn(output);
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d6c6ad37/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectGraphVisitorTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectGraphVisitorTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectGraphVisitorTest.java
index d218a81..fb84de8 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectGraphVisitorTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectGraphVisitorTest.java
@@ -48,6 +48,7 @@ import org.junit.runners.JUnit4;
/**
* Tests for {@link DirectGraphVisitor}.
*/
+// TODO: Replace uses of getProducing
@RunWith(JUnit4.class)
public class DirectGraphVisitorTest implements Serializable {
@Rule public transient ExpectedException thrown = ExpectedException.none();
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d6c6ad37/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectGraphs.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectGraphs.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectGraphs.java
new file mode 100644
index 0000000..73ada19
--- /dev/null
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectGraphs.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.beam.runners.direct;
+
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.transforms.AppliedPTransform;
+import org.apache.beam.sdk.values.PValue;
+
+/** Test utilities for the {@link DirectRunner}. */
+final class DirectGraphs {
+ public static DirectGraph getGraph(Pipeline p) {
+ DirectGraphVisitor visitor = new DirectGraphVisitor();
+ p.traverseTopologically(visitor);
+ return visitor.getGraph();
+ }
+
+ public static AppliedPTransform<?, ?, ?> getProducer(PValue value) {
+ return getGraph(value.getPipeline()).getProducer(value);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d6c6ad37/runners/direct-java/src/test/java/org/apache/beam/runners/direct/EvaluationContextTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/EvaluationContextTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/EvaluationContextTest.java
index 17cdea1..a2bb15e 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/EvaluationContextTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/EvaluationContextTest.java
@@ -86,9 +86,13 @@ public class EvaluationContextTest {
private PCollectionView<Iterable<Integer>> view;
private PCollection<Long> unbounded;
- private BundleFactory bundleFactory;
private DirectGraph graph;
+ private AppliedPTransform<?, ?, ?> createdProducer;
+ private AppliedPTransform<?, ?, ?> downstreamProducer;
+ private AppliedPTransform<?, ?, ?> viewProducer;
+ private AppliedPTransform<?, ?, ?> unboundedProducer;
+
@Before
public void setup() {
DirectRunner runner =
@@ -101,14 +105,16 @@ public class EvaluationContextTest {
view = created.apply(View.<Integer>asIterable());
unbounded = p.apply(CountingInput.unbounded());
- DirectGraphVisitor graphVisitor = new DirectGraphVisitor();
- p.traverseTopologically(graphVisitor);
-
- bundleFactory = ImmutableListBundleFactory.create();
- graph = graphVisitor.getGraph();
+ BundleFactory bundleFactory = ImmutableListBundleFactory.create();
+ graph = DirectGraphs.getGraph(p);
context =
EvaluationContext.create(
runner.getPipelineOptions(), NanosOffsetClock.create(), bundleFactory, graph);
+
+ createdProducer = graph.getProducer(created);
+ downstreamProducer = graph.getProducer(downstream);
+ viewProducer = graph.getProducer(view);
+ unboundedProducer = graph.getProducer(unbounded);
}
@Test
@@ -146,7 +152,7 @@ public class EvaluationContextTest {
@Test
public void getExecutionContextSameStepSameKeyState() {
DirectExecutionContext fooContext =
- context.getExecutionContext(created.getProducingTransformInternal(),
+ context.getExecutionContext(createdProducer,
StructuralKey.of("foo", StringUtf8Coder.of()));
StateTag<Object, BagState<Integer>> intBag = StateTags.bag("myBag", VarIntCoder.of());
@@ -159,12 +165,12 @@ public class EvaluationContextTest {
.createKeyedBundle(StructuralKey.of("foo", StringUtf8Coder.of()), created)
.commit(Instant.now()),
ImmutableList.<TimerData>of(),
- StepTransformResult.withoutHold(created.getProducingTransformInternal())
+ StepTransformResult.withoutHold(createdProducer)
.withState(stepContext.commitState())
.build());
DirectExecutionContext secondFooContext =
- context.getExecutionContext(created.getProducingTransformInternal(),
+ context.getExecutionContext(createdProducer,
StructuralKey.of("foo", StringUtf8Coder.of()));
assertThat(
secondFooContext
@@ -179,7 +185,7 @@ public class EvaluationContextTest {
@Test
public void getExecutionContextDifferentKeysIndependentState() {
DirectExecutionContext fooContext =
- context.getExecutionContext(created.getProducingTransformInternal(),
+ context.getExecutionContext(createdProducer,
StructuralKey.of("foo", StringUtf8Coder.of()));
StateTag<Object, BagState<Integer>> intBag = StateTags.bag("myBag", VarIntCoder.of());
@@ -191,7 +197,7 @@ public class EvaluationContextTest {
.add(1);
DirectExecutionContext barContext =
- context.getExecutionContext(created.getProducingTransformInternal(),
+ context.getExecutionContext(createdProducer,
StructuralKey.of("bar", StringUtf8Coder.of()));
assertThat(barContext, not(equalTo(fooContext)));
assertThat(
@@ -207,7 +213,7 @@ public class EvaluationContextTest {
public void getExecutionContextDifferentStepsIndependentState() {
StructuralKey<?> myKey = StructuralKey.of("foo", StringUtf8Coder.of());
DirectExecutionContext fooContext =
- context.getExecutionContext(created.getProducingTransformInternal(), myKey);
+ context.getExecutionContext(createdProducer, myKey);
StateTag<Object, BagState<Integer>> intBag = StateTags.bag("myBag", VarIntCoder.of());
@@ -218,7 +224,7 @@ public class EvaluationContextTest {
.add(1);
DirectExecutionContext barContext =
- context.getExecutionContext(downstream.getProducingTransformInternal(), myKey);
+ context.getExecutionContext(downstreamProducer, myKey);
assertThat(
barContext
.getOrCreateStepContext("s1", "s1")
@@ -232,15 +238,15 @@ public class EvaluationContextTest {
public void handleResultCommitsAggregators() {
Class<?> fn = getClass();
DirectExecutionContext fooContext =
- context.getExecutionContext(created.getProducingTransformInternal(), null);
+ context.getExecutionContext(createdProducer, null);
DirectExecutionContext.StepContext stepContext = fooContext.createStepContext(
- "STEP", created.getProducingTransformInternal().getTransform().getName());
+ "STEP", createdProducer.getTransform().getName());
AggregatorContainer container = context.getAggregatorContainer();
AggregatorContainer.Mutator mutator = container.createMutator();
mutator.createAggregatorForDoFn(fn, stepContext, "foo", new SumLongFn()).addValue(4L);
TransformResult<?> result =
- StepTransformResult.withoutHold(created.getProducingTransformInternal())
+ StepTransformResult.withoutHold(createdProducer)
.withAggregatorChanges(mutator)
.build();
context.handleResult(null, ImmutableList.<TimerData>of(), result);
@@ -250,7 +256,7 @@ public class EvaluationContextTest {
mutatorAgain.createAggregatorForDoFn(fn, stepContext, "foo", new SumLongFn()).addValue(12L);
TransformResult<?> secondResult =
- StepTransformResult.withoutHold(downstream.getProducingTransformInternal())
+ StepTransformResult.withoutHold(downstreamProducer)
.withAggregatorChanges(mutatorAgain)
.build();
context.handleResult(
@@ -264,7 +270,7 @@ public class EvaluationContextTest {
public void handleResultStoresState() {
StructuralKey<?> myKey = StructuralKey.of("foo".getBytes(), ByteArrayCoder.of());
DirectExecutionContext fooContext =
- context.getExecutionContext(downstream.getProducingTransformInternal(), myKey);
+ context.getExecutionContext(downstreamProducer, myKey);
StateTag<Object, BagState<Integer>> intBag = StateTags.bag("myBag", VarIntCoder.of());
@@ -276,7 +282,7 @@ public class EvaluationContextTest {
bag.add(4);
TransformResult<?> stateResult =
- StepTransformResult.withoutHold(downstream.getProducingTransformInternal())
+ StepTransformResult.withoutHold(downstreamProducer)
.withState(state)
.build();
@@ -286,7 +292,7 @@ public class EvaluationContextTest {
stateResult);
DirectExecutionContext afterResultContext =
- context.getExecutionContext(downstream.getProducingTransformInternal(), myKey);
+ context.getExecutionContext(downstreamProducer, myKey);
CopyOnAccessInMemoryStateInternals<Object> afterResultState =
afterResultContext.getOrCreateStepContext("s1", "s1").stateInternals();
@@ -309,7 +315,7 @@ public class EvaluationContextTest {
downstream, GlobalWindow.INSTANCE, WindowingStrategy.globalDefault(), callback);
TransformResult<?> result =
- StepTransformResult.withHold(created.getProducingTransformInternal(), new Instant(0))
+ StepTransformResult.withHold(createdProducer, new Instant(0))
.build();
context.handleResult(null, ImmutableList.<TimerData>of(), result);
@@ -318,7 +324,7 @@ public class EvaluationContextTest {
assertThat(callLatch.await(500L, TimeUnit.MILLISECONDS), is(false));
TransformResult<?> finishedResult =
- StepTransformResult.withoutHold(created.getProducingTransformInternal()).build();
+ StepTransformResult.withoutHold(createdProducer).build();
context.handleResult(null, ImmutableList.<TimerData>of(), finishedResult);
context.forceRefresh();
// Obtain the value via blocking call
@@ -328,7 +334,7 @@ public class EvaluationContextTest {
@Test
public void callAfterOutputMustHaveBeenProducedAlreadyAfterCallsImmediately() throws Exception {
TransformResult<?> finishedResult =
- StepTransformResult.withoutHold(created.getProducingTransformInternal()).build();
+ StepTransformResult.withoutHold(createdProducer).build();
context.handleResult(null, ImmutableList.<TimerData>of(), finishedResult);
final CountDownLatch callLatch = new CountDownLatch(1);
@@ -348,7 +354,7 @@ public class EvaluationContextTest {
@Test
public void extractFiredTimersExtractsTimers() {
TransformResult<?> holdResult =
- StepTransformResult.withHold(created.getProducingTransformInternal(), new Instant(0))
+ StepTransformResult.withHold(createdProducer, new Instant(0))
.build();
context.handleResult(null, ImmutableList.<TimerData>of(), holdResult);
@@ -356,7 +362,7 @@ public class EvaluationContextTest {
TimerData toFire =
TimerData.of(StateNamespaces.global(), new Instant(100L), TimeDomain.EVENT_TIME);
TransformResult<?> timerResult =
- StepTransformResult.withoutHold(downstream.getProducingTransformInternal())
+ StepTransformResult.withoutHold(downstreamProducer)
.withState(CopyOnAccessInMemoryStateInternals.withUnderlying(key, null))
.withTimerUpdate(TimerUpdate.builder(key).setTimer(toFire).build())
.build();
@@ -372,7 +378,7 @@ public class EvaluationContextTest {
assertThat(context.extractFiredTimers(), emptyIterable());
TransformResult<?> advanceResult =
- StepTransformResult.withoutHold(created.getProducingTransformInternal()).build();
+ StepTransformResult.withoutHold(createdProducer).build();
// Should cause the downstream timer to fire
context.handleResult(null, ImmutableList.<TimerData>of(), advanceResult);
@@ -403,14 +409,14 @@ public class EvaluationContextTest {
@Test
public void isDoneWithUnboundedPCollectionAndShutdown() {
context.getPipelineOptions().setShutdownUnboundedProducersWithMaxWatermark(true);
- assertThat(context.isDone(unbounded.getProducingTransformInternal()), is(false));
+ assertThat(context.isDone(unboundedProducer), is(false));
context.handleResult(
null,
ImmutableList.<TimerData>of(),
- StepTransformResult.withoutHold(unbounded.getProducingTransformInternal()).build());
+ StepTransformResult.withoutHold(unboundedProducer).build());
context.extractFiredTimers();
- assertThat(context.isDone(unbounded.getProducingTransformInternal()), is(true));
+ assertThat(context.isDone(unboundedProducer), is(true));
}
@Test
@@ -428,14 +434,14 @@ public class EvaluationContextTest {
@Test
public void isDoneWithOnlyBoundedPCollections() {
context.getPipelineOptions().setShutdownUnboundedProducersWithMaxWatermark(false);
- assertThat(context.isDone(created.getProducingTransformInternal()), is(false));
+ assertThat(context.isDone(createdProducer), is(false));
context.handleResult(
null,
ImmutableList.<TimerData>of(),
- StepTransformResult.withoutHold(created.getProducingTransformInternal()).build());
+ StepTransformResult.withoutHold(createdProducer).build());
context.extractFiredTimers();
- assertThat(context.isDone(created.getProducingTransformInternal()), is(true));
+ assertThat(context.isDone(createdProducer), is(true));
}
@Test
@@ -449,7 +455,7 @@ public class EvaluationContextTest {
context.handleResult(
null,
ImmutableList.<TimerData>of(),
- StepTransformResult.<Integer>withoutHold(created.getProducingTransformInternal())
+ StepTransformResult.<Integer>withoutHold(createdProducer)
.addOutput(rootBundle)
.build());
@SuppressWarnings("unchecked")
@@ -458,7 +464,7 @@ public class EvaluationContextTest {
context.handleResult(
null,
ImmutableList.<TimerData>of(),
- StepTransformResult.withoutHold(unbounded.getProducingTransformInternal()).build());
+ StepTransformResult.withoutHold(unboundedProducer).build());
assertThat(context.isDone(), is(false));
for (AppliedPTransform<?, ?, ?> consumers : graph.getPrimitiveConsumers(created)) {
@@ -479,22 +485,22 @@ public class EvaluationContextTest {
context.handleResult(
null,
ImmutableList.<TimerData>of(),
- StepTransformResult.withoutHold(created.getProducingTransformInternal()).build());
+ StepTransformResult.withoutHold(createdProducer).build());
context.handleResult(
null,
ImmutableList.<TimerData>of(),
- StepTransformResult.withoutHold(unbounded.getProducingTransformInternal()).build());
+ StepTransformResult.withoutHold(unboundedProducer).build());
context.handleResult(
context.createBundle(created).commit(Instant.now()),
ImmutableList.<TimerData>of(),
- StepTransformResult.withoutHold(downstream.getProducingTransformInternal()).build());
+ StepTransformResult.withoutHold(downstreamProducer).build());
context.extractFiredTimers();
assertThat(context.isDone(), is(false));
context.handleResult(
context.createBundle(created).commit(Instant.now()),
ImmutableList.<TimerData>of(),
- StepTransformResult.withoutHold(view.getProducingTransformInternal()).build());
+ StepTransformResult.withoutHold(viewProducer).build());
context.extractFiredTimers();
assertThat(context.isDone(), is(false));
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d6c6ad37/runners/direct-java/src/test/java/org/apache/beam/runners/direct/FlattenEvaluatorFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/FlattenEvaluatorFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/FlattenEvaluatorFactoryTest.java
index cb27fbc..9e22c36 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/FlattenEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/FlattenEvaluatorFactoryTest.java
@@ -47,6 +47,7 @@ import org.junit.runners.JUnit4;
@RunWith(JUnit4.class)
public class FlattenEvaluatorFactoryTest {
private BundleFactory bundleFactory = ImmutableListBundleFactory.create();
+
@Test
public void testFlattenInMemoryEvaluator() throws Exception {
TestPipeline p = TestPipeline.create();
@@ -69,10 +70,11 @@ public class FlattenEvaluatorFactoryTest {
when(context.createBundle(flattened)).thenReturn(flattenedLeftBundle, flattenedRightBundle);
FlattenEvaluatorFactory factory = new FlattenEvaluatorFactory(context);
+ AppliedPTransform<?, ?, ?> flattenedProducer = DirectGraphs.getProducer(flattened);
TransformEvaluator<Integer> leftSideEvaluator =
- factory.forApplication(flattened.getProducingTransformInternal(), leftBundle);
+ factory.forApplication(flattenedProducer, leftBundle);
TransformEvaluator<Integer> rightSideEvaluator =
- factory.forApplication(flattened.getProducingTransformInternal(), rightBundle);
+ factory.forApplication(flattenedProducer, rightBundle);
leftSideEvaluator.processElement(WindowedValue.valueInGlobalWindow(1));
rightSideEvaluator.processElement(WindowedValue.valueInGlobalWindow(-1));
@@ -92,13 +94,13 @@ public class FlattenEvaluatorFactoryTest {
Matchers.<UncommittedBundle<?>>contains(flattenedRightBundle));
assertThat(
rightSideResult.getTransform(),
- Matchers.<AppliedPTransform<?, ?, ?>>equalTo(flattened.getProducingTransformInternal()));
+ Matchers.<AppliedPTransform<?, ?, ?>>equalTo(flattenedProducer));
assertThat(
leftSideResult.getOutputBundles(),
Matchers.<UncommittedBundle<?>>contains(flattenedLeftBundle));
assertThat(
leftSideResult.getTransform(),
- Matchers.<AppliedPTransform<?, ?, ?>>equalTo(flattened.getProducingTransformInternal()));
+ Matchers.<AppliedPTransform<?, ?, ?>>equalTo(flattenedProducer));
assertThat(
flattenedLeftBundle.commit(Instant.now()).getElements(),
@@ -126,9 +128,10 @@ public class FlattenEvaluatorFactoryTest {
.thenReturn(bundleFactory.createBundle(flattened));
FlattenEvaluatorFactory factory = new FlattenEvaluatorFactory(evaluationContext);
+ AppliedPTransform<?, ?, ?> flattendProducer = DirectGraphs.getProducer(flattened);
TransformEvaluator<Integer> emptyEvaluator =
factory.forApplication(
- flattened.getProducingTransformInternal(),
+ flattendProducer,
bundleFactory.createRootBundle().commit(BoundedWindow.TIMESTAMP_MAX_VALUE));
TransformResult<Integer> leftSideResult = emptyEvaluator.finishBundle();
@@ -138,7 +141,7 @@ public class FlattenEvaluatorFactoryTest {
assertThat(outputBundle.getElements(), emptyIterable());
assertThat(
leftSideResult.getTransform(),
- Matchers.<AppliedPTransform<?, ?, ?>>equalTo(flattened.getProducingTransformInternal()));
+ Matchers.<AppliedPTransform<?, ?, ?>>equalTo(flattendProducer));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d6c6ad37/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyEvaluatorFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyEvaluatorFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyEvaluatorFactoryTest.java
index 7ba38ce..f0b29f0 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyEvaluatorFactoryTest.java
@@ -97,7 +97,7 @@ public class GroupByKeyEvaluatorFactoryTest {
((KvCoder<String, Integer>) values.getCoder()).getKeyCoder();
TransformEvaluator<KV<String, Integer>> evaluator =
new GroupByKeyOnlyEvaluatorFactory(evaluationContext)
- .forApplication(groupedKvs.getProducingTransformInternal(), inputBundle);
+ .forApplication(DirectGraphs.getProducer(groupedKvs), inputBundle);
evaluator.processElement(WindowedValue.valueInGlobalWindow(firstFoo));
evaluator.processElement(WindowedValue.valueInGlobalWindow(secondFoo));
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d6c6ad37/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyOnlyEvaluatorFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyOnlyEvaluatorFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyOnlyEvaluatorFactoryTest.java
index 23340c6..7efdb3d 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyOnlyEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/GroupByKeyOnlyEvaluatorFactoryTest.java
@@ -90,8 +90,7 @@ public class GroupByKeyOnlyEvaluatorFactoryTest {
((KvCoder<String, Integer>) values.getCoder()).getKeyCoder();
TransformEvaluator<KV<String, Integer>> evaluator =
new GroupByKeyOnlyEvaluatorFactory(evaluationContext)
- .forApplication(
- groupedKvs.getProducingTransformInternal(), inputBundle);
+ .forApplication(DirectGraphs.getProducer(groupedKvs), inputBundle);
evaluator.processElement(WindowedValue.valueInGlobalWindow(firstFoo));
evaluator.processElement(WindowedValue.valueInGlobalWindow(secondFoo));
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d6c6ad37/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutabilityEnforcementFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutabilityEnforcementFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutabilityEnforcementFactoryTest.java
index a65cd30..1ad6ba6 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutabilityEnforcementFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutabilityEnforcementFactoryTest.java
@@ -64,7 +64,7 @@ public class ImmutabilityEnforcementFactoryTest implements Serializable {
c.element()[0] = 'b';
}
}));
- consumer = pcollection.apply(Count.<byte[]>globally()).getProducingTransformInternal();
+ consumer = DirectGraphs.getProducer(pcollection.apply(Count.<byte[]>globally()));
}
@Test
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d6c6ad37/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ParDoEvaluatorTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ParDoEvaluatorTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ParDoEvaluatorTest.java
index 85e99c5..d48ac14 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ParDoEvaluatorTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ParDoEvaluatorTest.java
@@ -152,8 +152,9 @@ public class ParDoEvaluatorTest {
when(evaluationContext.getAggregatorContainer()).thenReturn(container);
when(evaluationContext.getAggregatorMutator()).thenReturn(mutator);
+ @SuppressWarnings("unchecked")
AppliedPTransform<PCollection<Integer>, ?, ?> transform =
- (AppliedPTransform<PCollection<Integer>, ?, ?>) output.getProducingTransformInternal();
+ (AppliedPTransform<PCollection<Integer>, ?, ?>) DirectGraphs.getProducer(output);
return ParDoEvaluator.create(
evaluationContext,
stepContext,
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d6c6ad37/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java
index ecf11ed..06c85ef 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java
@@ -129,7 +129,7 @@ public class StatefulParDoEvaluatorFactoryTest implements Serializable {
AppliedPTransform<
PCollection<? extends KV<String, Iterable<Integer>>>, PCollectionTuple,
StatefulParDo<String, Integer, Integer>>
- producingTransform = (AppliedPTransform) produced.getProducingTransformInternal();
+ producingTransform = (AppliedPTransform) DirectGraphs.getProducer(produced);
// Then there will be a digging down to the step context to get the state internals
when(mockEvaluationContext.getExecutionContext(
@@ -239,7 +239,7 @@ public class StatefulParDoEvaluatorFactoryTest implements Serializable {
AppliedPTransform<
PCollection<KV<String, Iterable<Integer>>>, PCollectionTuple,
StatefulParDo<String, Integer, Integer>>
- producingTransform = (AppliedPTransform) produced.getProducingTransformInternal();
+ producingTransform = (AppliedPTransform) DirectGraphs.getProducer(produced);
// Then there will be a digging down to the step context to get the state internals
when(mockEvaluationContext.getExecutionContext(
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d6c6ad37/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StepTransformResultTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StepTransformResultTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StepTransformResultTest.java
index a21d8f7..d3a2cca 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StepTransformResultTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StepTransformResultTest.java
@@ -48,7 +48,7 @@ public class StepTransformResultTest {
public void setup() {
TestPipeline p = TestPipeline.create();
pc = p.apply(Create.of(1, 2, 3));
- transform = pc.getProducingTransformInternal();
+ transform = DirectGraphs.getGraph(p).getProducer(pc);
bundleFactory = ImmutableListBundleFactory.create();
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d6c6ad37/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TestStreamEvaluatorFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TestStreamEvaluatorFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TestStreamEvaluatorFactoryTest.java
index 3d31df6..6bb8623 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TestStreamEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TestStreamEvaluatorFactoryTest.java
@@ -32,6 +32,7 @@ import org.apache.beam.runners.direct.TestStreamEvaluatorFactory.TestStreamIndex
import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.testing.TestStream;
+import org.apache.beam.sdk.transforms.AppliedPTransform;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
@@ -80,15 +81,16 @@ public class TestStreamEvaluatorFactoryTest {
when(context.createBundle(streamVals))
.thenReturn(bundleFactory.createBundle(streamVals), bundleFactory.createBundle(streamVals));
+ AppliedPTransform<?, ?, ?> streamProducer = DirectGraphs.getProducer(streamVals);
Collection<CommittedBundle<?>> initialInputs =
new TestStreamEvaluatorFactory.InputProvider(context)
- .getInitialInputs(streamVals.getProducingTransformInternal(), 1);
+ .getInitialInputs(streamProducer, 1);
@SuppressWarnings("unchecked")
CommittedBundle<TestStreamIndex<Integer>> initialBundle =
(CommittedBundle<TestStreamIndex<Integer>>) Iterables.getOnlyElement(initialInputs);
TransformEvaluator<TestStreamIndex<Integer>> firstEvaluator =
- factory.forApplication(streamVals.getProducingTransformInternal(), initialBundle);
+ factory.forApplication(streamProducer, initialBundle);
firstEvaluator.processElement(Iterables.getOnlyElement(initialBundle.getElements()));
TransformResult<TestStreamIndex<Integer>> firstResult = firstEvaluator.finishBundle();
@@ -101,7 +103,7 @@ public class TestStreamEvaluatorFactoryTest {
CommittedBundle<TestStreamIndex<Integer>> secondBundle =
initialBundle.withElements(Collections.singleton(firstResidual));
TransformEvaluator<TestStreamIndex<Integer>> secondEvaluator =
- factory.forApplication(streamVals.getProducingTransformInternal(), secondBundle);
+ factory.forApplication(streamProducer, secondBundle);
secondEvaluator.processElement(firstResidual);
TransformResult<TestStreamIndex<Integer>> secondResult = secondEvaluator.finishBundle();
@@ -114,7 +116,7 @@ public class TestStreamEvaluatorFactoryTest {
CommittedBundle<TestStreamIndex<Integer>> thirdBundle =
secondBundle.withElements(Collections.singleton(secondResidual));
TransformEvaluator<TestStreamIndex<Integer>> thirdEvaluator =
- factory.forApplication(streamVals.getProducingTransformInternal(), thirdBundle);
+ factory.forApplication(streamProducer, thirdBundle);
thirdEvaluator.processElement(secondResidual);
TransformResult<TestStreamIndex<Integer>> thirdResult = thirdEvaluator.finishBundle();
@@ -128,7 +130,7 @@ public class TestStreamEvaluatorFactoryTest {
CommittedBundle<TestStreamIndex<Integer>> fourthBundle =
thirdBundle.withElements(Collections.singleton(thirdResidual));
TransformEvaluator<TestStreamIndex<Integer>> fourthEvaluator =
- factory.forApplication(streamVals.getProducingTransformInternal(), fourthBundle);
+ factory.forApplication(streamProducer, fourthBundle);
fourthEvaluator.processElement(thirdResidual);
TransformResult<TestStreamIndex<Integer>> fourthResult = fourthEvaluator.finishBundle();
@@ -142,7 +144,7 @@ public class TestStreamEvaluatorFactoryTest {
CommittedBundle<TestStreamIndex<Integer>> fifthBundle =
thirdBundle.withElements(Collections.singleton(fourthResidual));
TransformEvaluator<TestStreamIndex<Integer>> fifthEvaluator =
- factory.forApplication(streamVals.getProducingTransformInternal(), fifthBundle);
+ factory.forApplication(streamProducer, fifthBundle);
fifthEvaluator.processElement(fourthResidual);
TransformResult<TestStreamIndex<Integer>> fifthResult = fifthEvaluator.finishBundle();
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d6c6ad37/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java
index 08b1e18..4ad22bc 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java
@@ -89,8 +89,9 @@ public class TransformExecutorTest {
created = p.apply(Create.of("foo", "spam", "third"));
PCollection<KV<Integer, String>> downstream = created.apply(WithKeys.<Integer, String>of(3));
- createdProducer = created.getProducingTransformInternal();
- downstreamProducer = downstream.getProducingTransformInternal();
+ DirectGraph graph = DirectGraphs.getGraph(p);
+ createdProducer = graph.getProducer(created);
+ downstreamProducer = graph.getProducer(downstream);
when(evaluationContext.getMetrics()).thenReturn(metrics);
}
@@ -317,7 +318,7 @@ public class TransformExecutorTest {
@Test
public void callWithEnforcementThrowsOnFinishPropagates() throws Exception {
final TransformResult<Object> result =
- StepTransformResult.withoutHold(created.getProducingTransformInternal()).build();
+ StepTransformResult.withoutHold(createdProducer).build();
TransformEvaluator<Object> evaluator =
new TransformEvaluator<Object>() {
@@ -356,7 +357,7 @@ public class TransformExecutorTest {
@Test
public void callWithEnforcementThrowsOnElementPropagates() throws Exception {
final TransformResult<Object> result =
- StepTransformResult.withoutHold(created.getProducingTransformInternal()).build();
+ StepTransformResult.withoutHold(createdProducer).build();
TransformEvaluator<Object> evaluator =
new TransformEvaluator<Object>() {
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d6c6ad37/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java
index 5a10134..dd36a2f 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java
@@ -17,6 +17,7 @@
*/
package org.apache.beam.runners.direct;
+import static org.apache.beam.runners.direct.DirectGraphs.getProducer;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasSize;
@@ -90,6 +91,7 @@ public class UnboundedReadEvaluatorFactoryTest {
private BundleFactory bundleFactory = ImmutableListBundleFactory.create();
private UnboundedSource<Long, ?> source;
+ private DirectGraph graph;
@Before
public void setup() {
@@ -100,6 +102,7 @@ public class UnboundedReadEvaluatorFactoryTest {
context = mock(EvaluationContext.class);
factory = new UnboundedReadEvaluatorFactory(context);
output = bundleFactory.createBundle(longs);
+ graph = DirectGraphs.getGraph(p);
when(context.createBundle(longs)).thenReturn(output);
}
@@ -115,7 +118,7 @@ public class UnboundedReadEvaluatorFactoryTest {
int numSplits = 5;
Collection<CommittedBundle<?>> initialInputs =
new UnboundedReadEvaluatorFactory.InputProvider(context)
- .getInitialInputs(longs.getProducingTransformInternal(), numSplits);
+ .getInitialInputs(graph.getProducer(longs), numSplits);
// CountingSource.unbounded has very good splitting behavior
assertThat(initialInputs, hasSize(numSplits));
@@ -148,15 +151,14 @@ public class UnboundedReadEvaluatorFactoryTest {
Collection<CommittedBundle<?>> initialInputs =
new UnboundedReadEvaluatorFactory.InputProvider(context)
- .getInitialInputs(longs.getProducingTransformInternal(), 1);
+ .getInitialInputs(graph.getProducer(longs), 1);
CommittedBundle<?> inputShards = Iterables.getOnlyElement(initialInputs);
UnboundedSourceShard<Long, ?> inputShard =
(UnboundedSourceShard<Long, ?>)
Iterables.getOnlyElement(inputShards.getElements()).getValue();
TransformEvaluator<? super UnboundedSourceShard<Long, ?>> evaluator =
- factory.forApplication(
- longs.getProducingTransformInternal(), inputShards);
+ factory.forApplication(graph.getProducer(longs), inputShards);
evaluator.processElement((WindowedValue) Iterables.getOnlyElement(inputShards.getElements()));
TransformResult<? super UnboundedSourceShard<Long, ?>> result = evaluator.finishBundle();
@@ -190,7 +192,7 @@ public class UnboundedReadEvaluatorFactoryTest {
TestPipeline p = TestPipeline.create();
PCollection<Long> pcollection = p.apply(Read.from(source));
- AppliedPTransform<?, ?, ?> sourceTransform = pcollection.getProducingTransformInternal();
+ AppliedPTransform<?, ?, ?> sourceTransform = getProducer(pcollection);
when(context.createRootBundle()).thenReturn(bundleFactory.createRootBundle());
Collection<CommittedBundle<?>> initialInputs =
@@ -233,7 +235,7 @@ public class UnboundedReadEvaluatorFactoryTest {
// Read with a very slow rate so by the second read there are no more elements
PCollection<Long> pcollection =
p.apply(Read.from(new TestUnboundedSource<>(VarLongCoder.of(), 1L)));
- AppliedPTransform<?, ?, ?> sourceTransform = pcollection.getProducingTransformInternal();
+ AppliedPTransform<?, ?, ?> sourceTransform = DirectGraphs.getProducer(pcollection);
when(context.createRootBundle()).thenReturn(bundleFactory.createRootBundle());
Collection<CommittedBundle<?>> initialInputs =
@@ -291,7 +293,9 @@ public class UnboundedReadEvaluatorFactoryTest {
TestPipeline p = TestPipeline.create();
PCollection<Long> pcollection = p.apply(Read.from(source));
- AppliedPTransform<?, ?, ?> sourceTransform = pcollection.getProducingTransformInternal();
+ DirectGraph graph = DirectGraphs.getGraph(p);
+ AppliedPTransform<?, ?, ?> sourceTransform =
+ graph.getProducer(pcollection);
when(context.createRootBundle()).thenReturn(bundleFactory.createRootBundle());
UncommittedBundle<Long> output = bundleFactory.createBundle(pcollection);
@@ -307,8 +311,7 @@ public class UnboundedReadEvaluatorFactoryTest {
.commit(Instant.now());
UnboundedReadEvaluatorFactory factory =
new UnboundedReadEvaluatorFactory(context, 1.0 /* Always reuse */);
- new UnboundedReadEvaluatorFactory.InputProvider(context)
- .getInitialInputs(pcollection.getProducingTransformInternal(), 1);
+ new UnboundedReadEvaluatorFactory.InputProvider(context).getInitialInputs(sourceTransform, 1);
TransformEvaluator<UnboundedSourceShard<Long, TestCheckpointMark>> evaluator =
factory.forApplication(sourceTransform, inputBundle);
evaluator.processElement(shard);
@@ -336,7 +339,8 @@ public class UnboundedReadEvaluatorFactoryTest {
TestPipeline p = TestPipeline.create();
PCollection<Long> pcollection = p.apply(Read.from(source));
- AppliedPTransform<?, ?, ?> sourceTransform = pcollection.getProducingTransformInternal();
+ AppliedPTransform<?, ?, ?> sourceTransform =
+ DirectGraphs.getGraph(p).getProducer(pcollection);
when(context.createRootBundle()).thenReturn(bundleFactory.createRootBundle());
UncommittedBundle<Long> output = bundleFactory.createBundle(pcollection);
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d6c6ad37/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ViewEvaluatorFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ViewEvaluatorFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ViewEvaluatorFactoryTest.java
index 7d14020..7c08009 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ViewEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ViewEvaluatorFactoryTest.java
@@ -30,6 +30,7 @@ import org.apache.beam.sdk.coders.KvCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VoidCoder;
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.GroupByKey;
import org.apache.beam.sdk.transforms.Values;
@@ -73,9 +74,10 @@ public class ViewEvaluatorFactoryTest {
CommittedBundle<String> inputBundle =
bundleFactory.createBundle(input).commit(Instant.now());
+ AppliedPTransform<?, ?, ?> producer = DirectGraphs.getProducer(view);
TransformEvaluator<Iterable<String>> evaluator =
new ViewEvaluatorFactory(context)
- .forApplication(view.getProducingTransformInternal(), inputBundle);
+ .forApplication(producer, inputBundle);
evaluator.processElement(
WindowedValue.<Iterable<String>>valueInGlobalWindow(ImmutableList.of("foo", "bar")));
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d6c6ad37/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkCallbackExecutorTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkCallbackExecutorTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkCallbackExecutorTest.java
index 1be9a98..acdabb6 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkCallbackExecutorTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkCallbackExecutorTest.java
@@ -55,8 +55,10 @@ public class WatermarkCallbackExecutorTest {
public void setup() {
TestPipeline p = TestPipeline.create();
PCollection<Integer> created = p.apply(Create.of(1, 2, 3));
- create = created.getProducingTransformInternal();
- sum = created.apply(Sum.integersGlobally()).getProducingTransformInternal();
+ PCollection<Integer> summed = created.apply(Sum.integersGlobally());
+ DirectGraph graph = DirectGraphs.getGraph(p);
+ create = graph.getProducer(created);
+ sum = graph.getProducer(summed);
}
@Test
[2/4] incubator-beam git commit: Add DirectGraphs to DirectRunner
Tests
Posted by tg...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d6c6ad37/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java
index 076e0fb..eb4d0cd 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java
@@ -31,7 +31,6 @@ import java.io.Serializable;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
-import java.util.HashMap;
import java.util.Map;
import javax.annotation.Nullable;
import org.apache.beam.runners.direct.CommittedResult.OutputType;
@@ -63,7 +62,6 @@ import org.apache.beam.sdk.util.state.StateNamespaces;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionList;
-import org.apache.beam.sdk.values.PValue;
import org.apache.beam.sdk.values.TimestampedValue;
import org.hamcrest.BaseMatcher;
import org.hamcrest.Description;
@@ -116,33 +114,8 @@ public class WatermarkManagerTest implements Serializable {
PCollectionList<Integer> preFlatten = PCollectionList.of(createdInts).and(intsToFlatten);
flattened = preFlatten.apply("flattened", Flatten.<Integer>pCollections());
- Collection<AppliedPTransform<?, ?, ?>> rootTransforms =
- ImmutableList.<AppliedPTransform<?, ?, ?>>of(
- createdInts.getProducingTransformInternal(),
- intsToFlatten.getProducingTransformInternal());
-
- Map<PValue, Collection<AppliedPTransform<?, ?, ?>>> consumers = new HashMap<>();
- consumers.put(
- createdInts,
- ImmutableList.<AppliedPTransform<?, ?, ?>>of(filtered.getProducingTransformInternal(),
- keyed.getProducingTransformInternal(), flattened.getProducingTransformInternal()));
- consumers.put(
- filtered,
- Collections.<AppliedPTransform<?, ?, ?>>singleton(
- filteredTimesTwo.getProducingTransformInternal()));
- consumers.put(filteredTimesTwo, Collections.<AppliedPTransform<?, ?, ?>>emptyList());
- consumers.put(keyed, Collections.<AppliedPTransform<?, ?, ?>>emptyList());
-
- consumers.put(
- intsToFlatten,
- Collections.<AppliedPTransform<?, ?, ?>>singleton(
- flattened.getProducingTransformInternal()));
- consumers.put(flattened, Collections.<AppliedPTransform<?, ?, ?>>emptyList());
-
clock = MockClock.fromInstant(new Instant(1000));
- DirectGraphVisitor visitor = new DirectGraphVisitor();
- p.traverseTopologically(visitor);
- graph = visitor.getGraph();
+ graph = DirectGraphs.getGraph(p);
manager = WatermarkManager.create(clock, graph);
bundleFactory = ImmutableListBundleFactory.create();
@@ -155,7 +128,7 @@ public class WatermarkManagerTest implements Serializable {
@Test
public void getWatermarkForUntouchedTransform() {
TransformWatermarks watermarks =
- manager.getWatermarks(createdInts.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(createdInts));
assertThat(watermarks.getInputWatermark(), equalTo(BoundedWindow.TIMESTAMP_MIN_VALUE));
assertThat(watermarks.getOutputWatermark(), equalTo(BoundedWindow.TIMESTAMP_MIN_VALUE));
@@ -170,13 +143,13 @@ public class WatermarkManagerTest implements Serializable {
CommittedBundle<Integer> output = multiWindowedBundle(createdInts, 1);
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(output)),
new Instant(8000L));
manager.refreshAll();
TransformWatermarks updatedSourceWatermark =
- manager.getWatermarks(createdInts.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(createdInts));
assertThat(updatedSourceWatermark.getOutputWatermark(), equalTo(new Instant(8000L)));
}
@@ -191,7 +164,7 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(intsToFlatten.getProducingTransformInternal(),
+ result(graph.getProducer(intsToFlatten),
null,
Collections.<CommittedBundle<?>>singleton(secondPcollectionBundle)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
@@ -199,7 +172,7 @@ public class WatermarkManagerTest implements Serializable {
// We didn't do anything for the first source, so we shouldn't have progressed the watermark
TransformWatermarks firstSourceWatermark =
- manager.getWatermarks(createdInts.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(createdInts));
assertThat(
firstSourceWatermark.getOutputWatermark(),
not(laterThan(BoundedWindow.TIMESTAMP_MIN_VALUE)));
@@ -207,14 +180,14 @@ public class WatermarkManagerTest implements Serializable {
// the Second Source output all of the elements so it should be done (with a watermark at the
// end of time).
TransformWatermarks secondSourceWatermark =
- manager.getWatermarks(intsToFlatten.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(intsToFlatten));
assertThat(
secondSourceWatermark.getOutputWatermark(),
not(earlierThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
// We haven't consumed anything yet, so our watermark should be at the beginning of time
TransformWatermarks transformWatermark =
- manager.getWatermarks(flattened.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(flattened));
assertThat(
transformWatermark.getInputWatermark(), not(laterThan(BoundedWindow.TIMESTAMP_MIN_VALUE)));
assertThat(
@@ -225,15 +198,15 @@ public class WatermarkManagerTest implements Serializable {
// anything from the first PCollection yet; so our watermark shouldn't advance
manager.updateWatermarks(secondPcollectionBundle,
TimerUpdate.empty(),
- result(flattened.getProducingTransformInternal(),
+ result(graph.getProducer(flattened),
secondPcollectionBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(flattenedBundleSecondCreate)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
TransformWatermarks transformAfterProcessing =
- manager.getWatermarks(flattened.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(flattened));
manager.updateWatermarks(secondPcollectionBundle,
TimerUpdate.empty(),
- result(flattened.getProducingTransformInternal(),
+ result(graph.getProducer(flattened),
secondPcollectionBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(flattenedBundleSecondCreate)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
@@ -252,13 +225,13 @@ public class WatermarkManagerTest implements Serializable {
// past the end of the global window
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(firstPcollectionBundle)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
manager.refreshAll();
TransformWatermarks firstSourceWatermarks =
- manager.getWatermarks(createdInts.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(createdInts));
assertThat(
firstSourceWatermarks.getOutputWatermark(),
not(earlierThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
@@ -266,7 +239,7 @@ public class WatermarkManagerTest implements Serializable {
// We still haven't consumed any of the first source's input, so the watermark should still not
// progress
TransformWatermarks flattenAfterSourcesProduced =
- manager.getWatermarks(flattened.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(flattened));
assertThat(
flattenAfterSourcesProduced.getInputWatermark(), not(laterThan(firstCollectionTimestamp)));
assertThat(
@@ -276,7 +249,7 @@ public class WatermarkManagerTest implements Serializable {
// end of the global window), we should have a watermark equal to the min among buffered
// elements
TransformWatermarks withBufferedElements =
- manager.getWatermarks(flattened.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(flattened));
assertThat(withBufferedElements.getInputWatermark(), equalTo(firstCollectionTimestamp));
assertThat(withBufferedElements.getOutputWatermark(), equalTo(firstCollectionTimestamp));
@@ -284,13 +257,13 @@ public class WatermarkManagerTest implements Serializable {
bundleFactory.createBundle(flattened).commit(BoundedWindow.TIMESTAMP_MAX_VALUE);
manager.updateWatermarks(firstPcollectionBundle,
TimerUpdate.empty(),
- result(flattened.getProducingTransformInternal(),
+ result(graph.getProducer(flattened),
firstPcollectionBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(completedFlattenBundle)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
manager.refreshAll();
TransformWatermarks afterConsumingAllInput =
- manager.getWatermarks(flattened.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(flattened));
assertThat(
afterConsumingAllInput.getInputWatermark(),
not(earlierThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
@@ -329,8 +302,8 @@ public class WatermarkManagerTest implements Serializable {
Map<AppliedPTransform<?, ?, ?>, Collection<CommittedBundle<?>>> initialInputs =
ImmutableMap.<AppliedPTransform<?, ?, ?>, Collection<CommittedBundle<?>>>builder()
- .put(
- created.getProducingTransformInternal(),
+ .put(graph.getProducer(
+ created),
Collections.<CommittedBundle<?>>singleton(root))
.build();
tstMgr.initialize(initialInputs);
@@ -338,7 +311,7 @@ public class WatermarkManagerTest implements Serializable {
root,
TimerUpdate.empty(),
CommittedResult.create(
- StepTransformResult.withoutHold(created.getProducingTransformInternal()).build(),
+ StepTransformResult.withoutHold(graph.getProducer(created)).build(),
root.withElements(Collections.<WindowedValue<Void>>emptyList()),
Collections.singleton(createBundle),
EnumSet.allOf(OutputType.class)),
@@ -385,13 +358,13 @@ public class WatermarkManagerTest implements Serializable {
TimestampedValue.of(3, new Instant(-1000L)));
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(createdBundle)),
new Instant(Long.MAX_VALUE));
manager.refreshAll();
TransformWatermarks createdAfterProducing =
- manager.getWatermarks(createdInts.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(createdInts));
assertThat(
createdAfterProducing.getOutputWatermark(),
not(earlierThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
@@ -402,20 +375,20 @@ public class WatermarkManagerTest implements Serializable {
TimestampedValue.of(KV.of("MyKey", 3), new Instant(-1000L)));
manager.updateWatermarks(createdBundle,
TimerUpdate.empty(),
- result(keyed.getProducingTransformInternal(),
+ result(graph.getProducer(keyed),
createdBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(keyBundle)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
manager.refreshAll();
TransformWatermarks keyedWatermarks =
- manager.getWatermarks(keyed.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(keyed));
assertThat(
keyedWatermarks.getInputWatermark(), not(earlierThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
assertThat(
keyedWatermarks.getOutputWatermark(), not(earlierThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
TransformWatermarks filteredWatermarks =
- manager.getWatermarks(filtered.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(filtered));
assertThat(filteredWatermarks.getInputWatermark(), not(laterThan(new Instant(-1000L))));
assertThat(filteredWatermarks.getOutputWatermark(), not(laterThan(new Instant(-1000L))));
@@ -423,13 +396,13 @@ public class WatermarkManagerTest implements Serializable {
timestampedBundle(filtered, TimestampedValue.of(2, new Instant(1234L)));
manager.updateWatermarks(createdBundle,
TimerUpdate.empty(),
- result(filtered.getProducingTransformInternal(),
+ result(graph.getProducer(filtered),
createdBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(filteredBundle)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
manager.refreshAll();
TransformWatermarks filteredProcessedWatermarks =
- manager.getWatermarks(filtered.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(filtered));
assertThat(
filteredProcessedWatermarks.getInputWatermark(),
not(earlierThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
@@ -450,7 +423,7 @@ public class WatermarkManagerTest implements Serializable {
TimestampedValue.of(3, new Instant(-1000L)));
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(createdBundle)),
new Instant(Long.MAX_VALUE));
@@ -461,13 +434,13 @@ public class WatermarkManagerTest implements Serializable {
TimestampedValue.of(KV.of("MyKey", 3), new Instant(-1000L)));
manager.updateWatermarks(createdBundle,
TimerUpdate.empty(),
- result(keyed.getProducingTransformInternal(),
+ result(graph.getProducer(keyed),
createdBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(keyBundle)),
new Instant(500L));
manager.refreshAll();
TransformWatermarks keyedWatermarks =
- manager.getWatermarks(keyed.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(keyed));
assertThat(
keyedWatermarks.getInputWatermark(), not(earlierThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
assertThat(keyedWatermarks.getOutputWatermark(), not(laterThan(new Instant(500L))));
@@ -494,27 +467,27 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
ImmutableList.of(firstKeyBundle, secondKeyBundle)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
manager.updateWatermarks(firstKeyBundle,
TimerUpdate.empty(),
- result(filtered.getProducingTransformInternal(),
+ result(graph.getProducer(filtered),
firstKeyBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>emptyList()),
new Instant(-1000L));
manager.updateWatermarks(secondKeyBundle,
TimerUpdate.empty(),
- result(filtered.getProducingTransformInternal(),
+ result(graph.getProducer(filtered),
secondKeyBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>emptyList()),
new Instant(1234L));
manager.refreshAll();
TransformWatermarks filteredWatermarks =
- manager.getWatermarks(filtered.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(filtered));
assertThat(filteredWatermarks.getInputWatermark(),
not(earlierThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
assertThat(filteredWatermarks.getOutputWatermark(), not(laterThan(new Instant(-1000L))));
@@ -524,7 +497,7 @@ public class WatermarkManagerTest implements Serializable {
createdInts).commit(clock.now());
manager.updateWatermarks(fauxFirstKeyTimerBundle,
TimerUpdate.empty(),
- result(filtered.getProducingTransformInternal(),
+ result(graph.getProducer(filtered),
fauxFirstKeyTimerBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>emptyList()),
BoundedWindow.TIMESTAMP_MAX_VALUE);
@@ -537,7 +510,7 @@ public class WatermarkManagerTest implements Serializable {
createdInts).commit(clock.now());
manager.updateWatermarks(fauxSecondKeyTimerBundle,
TimerUpdate.empty(),
- result(filtered.getProducingTransformInternal(),
+ result(graph.getProducer(filtered),
fauxSecondKeyTimerBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>emptyList()),
new Instant(5678L));
@@ -546,7 +519,7 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(fauxSecondKeyTimerBundle,
TimerUpdate.empty(),
- result(filtered.getProducingTransformInternal(),
+ result(graph.getProducer(filtered),
fauxSecondKeyTimerBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>emptyList()),
BoundedWindow.TIMESTAMP_MAX_VALUE);
@@ -564,26 +537,26 @@ public class WatermarkManagerTest implements Serializable {
CommittedBundle<?> firstInput =
bundleFactory.createBundle(createdInts).commit(BoundedWindow.TIMESTAMP_MAX_VALUE);
manager.updateWatermarks(null, TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(firstInput)),
new Instant(0L));
manager.refreshAll();
TransformWatermarks firstWatermarks =
- manager.getWatermarks(createdInts.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(createdInts));
assertThat(firstWatermarks.getOutputWatermark(), equalTo(new Instant(0L)));
CommittedBundle<?> secondInput =
bundleFactory.createBundle(createdInts).commit(BoundedWindow.TIMESTAMP_MAX_VALUE);
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(secondInput)),
new Instant(-250L));
manager.refreshAll();
TransformWatermarks secondWatermarks =
- manager.getWatermarks(createdInts.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(createdInts));
assertThat(secondWatermarks.getOutputWatermark(), not(earlierThan(new Instant(0L))));
}
@@ -599,7 +572,7 @@ public class WatermarkManagerTest implements Serializable {
TimestampedValue.of(3, new Instant(-1000L)));
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(createdBundle)),
new Instant(Long.MAX_VALUE));
@@ -610,20 +583,20 @@ public class WatermarkManagerTest implements Serializable {
TimestampedValue.of(KV.of("MyKey", 3), new Instant(-1000L)));
manager.updateWatermarks(createdBundle,
TimerUpdate.empty(),
- result(keyed.getProducingTransformInternal(),
+ result(graph.getProducer(keyed),
createdBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(keyBundle)),
new Instant(500L));
manager.refreshAll();
TransformWatermarks keyedWatermarks =
- manager.getWatermarks(keyed.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(keyed));
assertThat(
keyedWatermarks.getInputWatermark(), not(earlierThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
assertThat(keyedWatermarks.getOutputWatermark(), not(laterThan(new Instant(500L))));
Instant oldOutputWatermark = keyedWatermarks.getOutputWatermark();
TransformWatermarks updatedWatermarks =
- manager.getWatermarks(keyed.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(keyed));
assertThat(
updatedWatermarks.getInputWatermark(), not(earlierThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
// We added a hold prior to the old watermark; we shouldn't progress (due to the earlier hold)
@@ -646,7 +619,7 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(createdBundle)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
@@ -655,12 +628,12 @@ public class WatermarkManagerTest implements Serializable {
TimestampedValue.of(KV.of("MyKey", 1), BoundedWindow.TIMESTAMP_MIN_VALUE));
manager.updateWatermarks(createdBundle,
TimerUpdate.empty(),
- result(keyed.getProducingTransformInternal(),
+ result(graph.getProducer(keyed),
createdBundle.withElements(ImmutableList.of(second, third)),
Collections.<CommittedBundle<?>>singleton(keyBundle)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
TransformWatermarks keyedWatermarks =
- manager.getWatermarks(keyed.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(keyed));
// the unprocessed second and third are readded to pending
assertThat(
keyedWatermarks.getInputWatermark(), not(laterThan(new Instant(-1000L))));
@@ -681,7 +654,7 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(createdBundle)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
@@ -689,15 +662,15 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(
neverCreatedBundle,
TimerUpdate.empty(),
- result(
- filtered.getProducingTransformInternal(),
+ result(graph.getProducer(
+ filtered),
neverCreatedBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>emptyList()),
BoundedWindow.TIMESTAMP_MAX_VALUE);
manager.refreshAll();
TransformWatermarks filteredWms =
- manager.getWatermarks(filtered.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(filtered));
assertThat(filteredWms.getInputWatermark(), equalTo(new Instant(22L)));
}
@@ -712,7 +685,7 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(createdBundle)),
sourceWatermark);
@@ -724,13 +697,13 @@ public class WatermarkManagerTest implements Serializable {
// Finish processing the on-time data. The watermarks should progress to be equal to the source
manager.updateWatermarks(createdBundle,
TimerUpdate.empty(),
- result(keyed.getProducingTransformInternal(),
+ result(graph.getProducer(keyed),
createdBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(keyBundle)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
manager.refreshAll();
TransformWatermarks onTimeWatermarks =
- manager.getWatermarks(keyed.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(keyed));
assertThat(onTimeWatermarks.getInputWatermark(), equalTo(sourceWatermark));
assertThat(onTimeWatermarks.getOutputWatermark(), equalTo(sourceWatermark));
@@ -740,19 +713,19 @@ public class WatermarkManagerTest implements Serializable {
// we don't advance the watermark past the current watermark until we've consumed the late data
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
createdBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(lateDataBundle)),
new Instant(2_000_000L));
manager.refreshAll();
TransformWatermarks bufferedLateWm =
- manager.getWatermarks(createdInts.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(createdInts));
assertThat(bufferedLateWm.getOutputWatermark(), equalTo(new Instant(2_000_000L)));
// The input watermark should be held to its previous value (not advanced due to late data; not
// moved backwards in the presence of watermarks due to monotonicity).
TransformWatermarks lateDataBufferedWatermark =
- manager.getWatermarks(keyed.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(keyed));
assertThat(lateDataBufferedWatermark.getInputWatermark(), not(earlierThan(sourceWatermark)));
assertThat(lateDataBufferedWatermark.getOutputWatermark(), not(earlierThan(sourceWatermark)));
@@ -760,7 +733,7 @@ public class WatermarkManagerTest implements Serializable {
timestampedBundle(keyed, TimestampedValue.of(KV.of("MyKey", 3), new Instant(-1000L)));
manager.updateWatermarks(lateDataBundle,
TimerUpdate.empty(),
- result(keyed.getProducingTransformInternal(),
+ result(graph.getProducer(keyed),
lateDataBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(lateKeyedBundle)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
@@ -770,7 +743,7 @@ public class WatermarkManagerTest implements Serializable {
public void updateWatermarkWithDifferentWindowedValueInstances() {
manager.updateWatermarks(
null,
- TimerUpdate.empty(), result(createdInts.getProducingTransformInternal(), null,
+ TimerUpdate.empty(), result(graph.getProducer(createdInts), null,
Collections.<CommittedBundle<?>>singleton(
bundleFactory
.createBundle(createdInts)
@@ -783,13 +756,13 @@ public class WatermarkManagerTest implements Serializable {
.commit(Instant.now());
manager.updateWatermarks(createdBundle,
TimerUpdate.empty(),
- result(keyed.getProducingTransformInternal(),
+ result(graph.getProducer(keyed),
createdBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>emptyList()),
null);
manager.refreshAll();
TransformWatermarks onTimeWatermarks =
- manager.getWatermarks(keyed.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(keyed));
assertThat(onTimeWatermarks.getInputWatermark(), equalTo(BoundedWindow.TIMESTAMP_MAX_VALUE));
}
@@ -802,20 +775,20 @@ public class WatermarkManagerTest implements Serializable {
CommittedBundle<Integer> emptyCreateOutput = multiWindowedBundle(createdInts);
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(emptyCreateOutput)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
manager.refreshAll();
TransformWatermarks updatedSourceWatermarks =
- manager.getWatermarks(createdInts.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(createdInts));
assertThat(
updatedSourceWatermarks.getOutputWatermark(),
not(earlierThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
TransformWatermarks finishedFilterWatermarks =
- manager.getWatermarks(filtered.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(filtered));
assertThat(
finishedFilterWatermarks.getInputWatermark(),
not(earlierThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
@@ -833,7 +806,7 @@ public class WatermarkManagerTest implements Serializable {
CommittedBundle<Integer> firstCreateOutput = multiWindowedBundle(createdInts, 1, 2);
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(firstCreateOutput)),
new Instant(12_000L));
@@ -841,33 +814,33 @@ public class WatermarkManagerTest implements Serializable {
CommittedBundle<Integer> firstFilterOutput = multiWindowedBundle(filtered);
manager.updateWatermarks(firstCreateOutput,
TimerUpdate.empty(),
- result(filtered.getProducingTransformInternal(),
+ result(graph.getProducer(filtered),
firstCreateOutput.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(firstFilterOutput)),
new Instant(10_000L));
manager.refreshAll();
TransformWatermarks firstFilterWatermarks =
- manager.getWatermarks(filtered.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(filtered));
assertThat(firstFilterWatermarks.getInputWatermark(), not(earlierThan(new Instant(12_000L))));
assertThat(firstFilterWatermarks.getOutputWatermark(), not(laterThan(new Instant(10_000L))));
CommittedBundle<Integer> emptyCreateOutput = multiWindowedBundle(createdInts);
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(emptyCreateOutput)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
manager.refreshAll();
TransformWatermarks updatedSourceWatermarks =
- manager.getWatermarks(createdInts.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(createdInts));
assertThat(
updatedSourceWatermarks.getOutputWatermark(),
not(earlierThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
TransformWatermarks finishedFilterWatermarks =
- manager.getWatermarks(filtered.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(filtered));
assertThat(
finishedFilterWatermarks.getInputWatermark(),
not(earlierThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
@@ -877,14 +850,14 @@ public class WatermarkManagerTest implements Serializable {
@Test
public void getSynchronizedProcessingTimeInputWatermarksHeldToPendingBundles() {
TransformWatermarks watermarks =
- manager.getWatermarks(createdInts.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(createdInts));
assertThat(watermarks.getSynchronizedProcessingInputTime(), equalTo(clock.now()));
assertThat(
watermarks.getSynchronizedProcessingOutputTime(),
equalTo(BoundedWindow.TIMESTAMP_MIN_VALUE));
TransformWatermarks filteredWatermarks =
- manager.getWatermarks(filtered.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(filtered));
// Non-root processing watermarks don't progress until data has been processed
assertThat(
filteredWatermarks.getSynchronizedProcessingInputTime(),
@@ -898,18 +871,18 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(createOutput)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
manager.refreshAll();
TransformWatermarks createAfterUpdate =
- manager.getWatermarks(createdInts.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(createdInts));
assertThat(createAfterUpdate.getSynchronizedProcessingInputTime(), equalTo(clock.now()));
assertThat(createAfterUpdate.getSynchronizedProcessingOutputTime(), equalTo(clock.now()));
TransformWatermarks filterAfterProduced =
- manager.getWatermarks(filtered.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(filtered));
assertThat(
filterAfterProduced.getSynchronizedProcessingInputTime(), not(laterThan(clock.now())));
assertThat(
@@ -929,13 +902,13 @@ public class WatermarkManagerTest implements Serializable {
bundleFactory.createBundle(intsToFlatten).commit(new Instant(1250L));
manager.updateWatermarks(createOutput,
TimerUpdate.empty(),
- result(filtered.getProducingTransformInternal(),
+ result(graph.getProducer(filtered),
createOutput.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(filterOutputBundle)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
manager.refreshAll();
TransformWatermarks filterAfterConsumed =
- manager.getWatermarks(filtered.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(filtered));
assertThat(
filterAfterConsumed.getSynchronizedProcessingInputTime(),
not(laterThan(createAfterUpdate.getSynchronizedProcessingOutputTime())));
@@ -955,16 +928,16 @@ public class WatermarkManagerTest implements Serializable {
CommittedBundle<Integer> createdBundle = multiWindowedBundle(createdInts, 1, 2, 4, 8);
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(createdBundle)),
new Instant(1248L));
manager.refreshAll();
TransformWatermarks filteredWms =
- manager.getWatermarks(filtered.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(filtered));
TransformWatermarks filteredDoubledWms =
- manager.getWatermarks(filteredTimesTwo.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(filteredTimesTwo));
Instant initialFilteredWm = filteredWms.getSynchronizedProcessingOutputTime();
Instant initialFilteredDoubledWm = filteredDoubledWms.getSynchronizedProcessingOutputTime();
@@ -978,7 +951,7 @@ public class WatermarkManagerTest implements Serializable {
TimerUpdate.builder(key).setTimer(pastTimer).setTimer(futureTimer).build();
manager.updateWatermarks(createdBundle,
timers,
- result(filtered.getProducingTransformInternal(),
+ result(graph.getProducer(filtered),
createdBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(filteredBundle)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
@@ -1014,7 +987,7 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(filteredTimerBundle,
TimerUpdate.builder(key)
.withCompletedTimers(Collections.<TimerData>singleton(pastTimer)).build(),
- result(filtered.getProducingTransformInternal(),
+ result(graph.getProducer(filtered),
filteredTimerBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(filteredTimerResult)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
@@ -1029,7 +1002,7 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(filteredTimerResult,
TimerUpdate.empty(),
- result(filteredTimesTwo.getProducingTransformInternal(),
+ result(graph.getProducer(filteredTimesTwo),
filteredTimerResult.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>emptyList()),
BoundedWindow.TIMESTAMP_MAX_VALUE);
@@ -1050,11 +1023,11 @@ public class WatermarkManagerTest implements Serializable {
public void getSynchronizedProcessingTimeOutputTimeIsMonotonic() {
Instant startTime = clock.now();
TransformWatermarks watermarks =
- manager.getWatermarks(createdInts.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(createdInts));
assertThat(watermarks.getSynchronizedProcessingInputTime(), equalTo(startTime));
TransformWatermarks filteredWatermarks =
- manager.getWatermarks(filtered.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(filtered));
// Non-root processing watermarks don't progress until data has been processed
assertThat(
filteredWatermarks.getSynchronizedProcessingInputTime(),
@@ -1068,13 +1041,13 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(createOutput)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
manager.refreshAll();
TransformWatermarks createAfterUpdate =
- manager.getWatermarks(createdInts.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(createdInts));
assertThat(createAfterUpdate.getSynchronizedProcessingInputTime(), not(laterThan(clock.now())));
assertThat(createAfterUpdate.getSynchronizedProcessingOutputTime(),
not(laterThan(clock.now())));
@@ -1083,7 +1056,7 @@ public class WatermarkManagerTest implements Serializable {
bundleFactory.createBundle(createdInts).commit(new Instant(750L));
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(createSecondOutput)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
@@ -1097,7 +1070,7 @@ public class WatermarkManagerTest implements Serializable {
CommittedBundle<Integer> created = multiWindowedBundle(createdInts, 1, 2, 3);
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(created)),
new Instant(40_900L));
@@ -1111,14 +1084,14 @@ public class WatermarkManagerTest implements Serializable {
TimerUpdate.builder(StructuralKey.of("key", StringUtf8Coder.of()))
.setTimer(upstreamProcessingTimer)
.build(),
- result(filtered.getProducingTransformInternal(),
+ result(graph.getProducer(filtered),
created.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(filteredBundle)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
manager.refreshAll();
TransformWatermarks downstreamWms =
- manager.getWatermarks(filteredTimesTwo.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(filteredTimesTwo));
assertThat(downstreamWms.getSynchronizedProcessingInputTime(), equalTo(clock.now()));
clock.set(BoundedWindow.TIMESTAMP_MAX_VALUE);
@@ -1133,7 +1106,7 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(otherCreated,
TimerUpdate.builder(StructuralKey.of("key", StringUtf8Coder.of()))
.withCompletedTimers(Collections.singleton(upstreamProcessingTimer)).build(),
- result(filtered.getProducingTransformInternal(),
+ result(graph.getProducer(filtered),
otherCreated.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>emptyList()),
BoundedWindow.TIMESTAMP_MAX_VALUE);
@@ -1148,7 +1121,7 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(
null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>singleton(created)),
new Instant(29_919_235L));
@@ -1160,14 +1133,14 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(
created,
TimerUpdate.empty(),
- result(filtered.getProducingTransformInternal(),
+ result(graph.getProducer(filtered),
created.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(filteredBundle)),
BoundedWindow.TIMESTAMP_MAX_VALUE);
manager.refreshAll();
TransformWatermarks downstreamWms =
- manager.getWatermarks(filteredTimesTwo.getProducingTransformInternal());
+ manager.getWatermarks(graph.getProducer(filteredTimesTwo));
assertThat(downstreamWms.getSynchronizedProcessingInputTime(), equalTo(clock.now()));
clock.set(BoundedWindow.TIMESTAMP_MAX_VALUE);
@@ -1184,7 +1157,7 @@ public class WatermarkManagerTest implements Serializable {
CommittedBundle<Integer> createdBundle = multiWindowedBundle(filtered);
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.singleton(createdBundle)),
new Instant(1500L));
@@ -1206,7 +1179,7 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(createdBundle,
update,
- result(filtered.getProducingTransformInternal(),
+ result(graph.getProducer(filtered),
createdBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(multiWindowedBundle(intsToFlatten))),
new Instant(1000L));
@@ -1220,7 +1193,7 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>emptyList()),
new Instant(50_000L));
@@ -1242,7 +1215,7 @@ public class WatermarkManagerTest implements Serializable {
CommittedBundle<Integer> createdBundle = multiWindowedBundle(filtered);
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.singleton(createdBundle)),
new Instant(1500L));
@@ -1264,7 +1237,7 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(
createdBundle,
update,
- result(filtered.getProducingTransformInternal(),
+ result(graph.getProducer(filtered),
createdBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(multiWindowedBundle(intsToFlatten))),
new Instant(1000L));
@@ -1278,7 +1251,7 @@ public class WatermarkManagerTest implements Serializable {
clock.set(new Instant(50_000L));
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>emptyList()),
new Instant(50_000L));
@@ -1301,7 +1274,7 @@ public class WatermarkManagerTest implements Serializable {
CommittedBundle<Integer> createdBundle = multiWindowedBundle(filtered);
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.singleton(createdBundle)),
new Instant(1500L));
@@ -1323,7 +1296,7 @@ public class WatermarkManagerTest implements Serializable {
manager.updateWatermarks(
createdBundle,
update,
- result(filtered.getProducingTransformInternal(),
+ result(graph.getProducer(filtered),
createdBundle.withElements(Collections.<WindowedValue<Integer>>emptyList()),
Collections.<CommittedBundle<?>>singleton(multiWindowedBundle(intsToFlatten))),
new Instant(1000L));
@@ -1338,7 +1311,7 @@ public class WatermarkManagerTest implements Serializable {
clock.set(new Instant(50_000L));
manager.updateWatermarks(null,
TimerUpdate.empty(),
- result(createdInts.getProducingTransformInternal(),
+ result(graph.getProducer(createdInts),
null,
Collections.<CommittedBundle<?>>emptyList()),
new Instant(50_000L));