You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by jk...@apache.org on 2017/04/21 23:56:18 UTC
[3/5] beam git commit: Replaces all usages of CountingInput with
GenerateSequence
Replaces all usages of CountingInput with GenerateSequence
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/6a9a24c0
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/6a9a24c0
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/6a9a24c0
Branch: refs/heads/master
Commit: 6a9a24c064518bb83a7383babdff9b263dc61346
Parents: 88c6612
Author: Eugene Kirpichov <ki...@google.com>
Authored: Wed Apr 19 15:32:08 2017 -0700
Committer: Eugene Kirpichov <ki...@google.com>
Committed: Fri Apr 21 16:53:50 2017 -0700
----------------------------------------------------------------------
.../translation/ReadUnboundTranslatorTest.java | 6 +-
.../EmptyFlattenAsCreateFactoryTest.java | 10 ++--
.../core/construction/PCollectionsTest.java | 6 +-
.../PTransformReplacementsTest.java | 4 +-
.../core/construction/PTransformsTest.java | 17 +++---
.../core/construction/SdkComponentsTest.java | 16 +++---
.../runners/direct/DirectGraphVisitorTest.java | 6 +-
.../beam/runners/direct/DirectRunnerTest.java | 10 ++--
.../runners/direct/EvaluationContextTest.java | 4 +-
.../beam/runners/flink/ReadSourceITCase.java | 4 +-
.../flink/ReadSourceStreamingITCase.java | 4 +-
.../streaming/StreamingSourceMetricsTest.java | 7 ++-
.../org/apache/beam/sdk/io/CountingSource.java | 43 ++++-----------
.../apache/beam/sdk/io/GenerateSequence.java | 5 ++
.../org/apache/beam/sdk/values/PCollection.java | 9 +--
.../java/org/apache/beam/sdk/PipelineTest.java | 58 +++++++++++---------
.../beam/sdk/io/GenerateSequenceTest.java | 18 +++---
.../apache/beam/sdk/metrics/MetricsTest.java | 12 ++--
.../sdk/runners/TransformHierarchyTest.java | 7 +--
.../beam/sdk/testing/GatherAllPanesTest.java | 8 +--
.../apache/beam/sdk/testing/PAssertTest.java | 12 ++--
.../apache/beam/sdk/transforms/FlattenTest.java | 8 +--
.../apache/beam/sdk/transforms/ParDoTest.java | 4 +-
.../sdk/transforms/windowing/WindowTest.java | 4 +-
.../beam/sdk/values/PCollectionListTest.java | 25 +++++----
.../beam/sdk/values/PCollectionTupleTest.java | 6 +-
.../sdk/io/gcp/bigquery/BigQueryIOTest.java | 6 +-
.../sdk/io/gcp/bigtable/BigtableWriteIT.java | 4 +-
.../beam/sdk/io/gcp/datastore/V1WriteIT.java | 4 +-
29 files changed, 162 insertions(+), 165 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/ReadUnboundTranslatorTest.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/ReadUnboundTranslatorTest.java b/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/ReadUnboundTranslatorTest.java
index 6d19bb9..e0cc251 100644
--- a/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/ReadUnboundTranslatorTest.java
+++ b/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/ReadUnboundTranslatorTest.java
@@ -35,7 +35,7 @@ import org.apache.beam.runners.apex.translation.operators.ApexReadUnboundedInput
import org.apache.beam.runners.apex.translation.utils.CollectionSource;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.StringUtf8Coder;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.Read;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
@@ -92,12 +92,12 @@ public class ReadUnboundTranslatorTest {
Pipeline p = Pipeline.create(options);
Set<Long> expected = ContiguousSet.create(Range.closedOpen(0L, 10L), DiscreteDomain.longs());
- p.apply(CountingInput.upTo(10))
+ p.apply(GenerateSequence.fromTo(0, 10))
.apply(ParDo.of(new EmbeddedCollector()));
ApexRunnerResult result = (ApexRunnerResult) p.run();
DAG dag = result.getApexDAG();
- String operatorName = "CountingInput.BoundedCountingInput/Read(BoundedCountingSource)";
+ String operatorName = "GenerateSequence/Read(BoundedCountingSource)";
DAG.OperatorMeta om = dag.getOperatorMeta(operatorName);
Assert.assertNotNull(om);
Assert.assertEquals(om.getOperator().getClass(), ApexReadUnboundedInputOperator.class);
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/EmptyFlattenAsCreateFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/EmptyFlattenAsCreateFactoryTest.java b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/EmptyFlattenAsCreateFactoryTest.java
index ae2d0a9..bfa3190 100644
--- a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/EmptyFlattenAsCreateFactoryTest.java
+++ b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/EmptyFlattenAsCreateFactoryTest.java
@@ -23,7 +23,7 @@ import static org.junit.Assert.assertThat;
import java.util.Collections;
import java.util.Map;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.runners.PTransformOverrideFactory.PTransformReplacement;
import org.apache.beam.sdk.runners.PTransformOverrideFactory.ReplacementOutput;
import org.apache.beam.sdk.testing.NeedsRunner;
@@ -71,8 +71,8 @@ public class EmptyFlattenAsCreateFactoryTest {
@Test
public void getInputNonEmptyThrows() {
PCollectionList<Long> nonEmpty =
- PCollectionList.of(pipeline.apply(CountingInput.unbounded()))
- .and(pipeline.apply(CountingInput.upTo(100L)));
+ PCollectionList.of(pipeline.apply("unbounded", GenerateSequence.from(0)))
+ .and(pipeline.apply("bounded", GenerateSequence.fromTo(0, 100)));
thrown.expect(IllegalArgumentException.class);
thrown.expectMessage(nonEmpty.expand().toString());
thrown.expectMessage(EmptyFlattenAsCreateFactory.class.getSimpleName());
@@ -87,8 +87,8 @@ public class EmptyFlattenAsCreateFactoryTest {
@Test
public void mapOutputsSucceeds() {
- PCollection<Long> original = pipeline.apply("Original", CountingInput.unbounded());
- PCollection<Long> replacement = pipeline.apply("Replacement", CountingInput.unbounded());
+ PCollection<Long> original = pipeline.apply("Original", GenerateSequence.from(0));
+ PCollection<Long> replacement = pipeline.apply("Replacement", GenerateSequence.from(0));
Map<PValue, ReplacementOutput> mapping = factory.mapOutputs(original.expand(), replacement);
assertThat(
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PCollectionsTest.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PCollectionsTest.java b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PCollectionsTest.java
index 636d245..48aa1f1 100644
--- a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PCollectionsTest.java
+++ b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PCollectionsTest.java
@@ -33,7 +33,7 @@ import org.apache.beam.sdk.coders.BigEndianLongCoder;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.CustomCoder;
import org.apache.beam.sdk.common.runner.v1.RunnerApi;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.GroupByKey;
@@ -71,7 +71,7 @@ public class PCollectionsTest {
public static Iterable<PCollection<?>> data() {
Pipeline pipeline = TestPipeline.create();
PCollection<Integer> ints = pipeline.apply("ints", Create.of(1, 2, 3));
- PCollection<Long> longs = pipeline.apply("unbounded longs", CountingInput.unbounded());
+ PCollection<Long> longs = pipeline.apply("unbounded longs", GenerateSequence.from(0));
PCollection<Long> windowedLongs =
longs.apply(
"into fixed windows",
@@ -83,7 +83,7 @@ public class PCollectionsTest {
.apply("group", GroupByKey.<String, String>create());
PCollection<Long> coderLongs =
pipeline
- .apply("counts with alternative coder", CountingInput.upTo(10L))
+ .apply("counts with alternative coder", GenerateSequence.fromTo(0, 10))
.setCoder(BigEndianLongCoder.of());
PCollection<Integer> allCustomInts =
pipeline
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformReplacementsTest.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformReplacementsTest.java b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformReplacementsTest.java
index b065617..c6c50f0 100644
--- a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformReplacementsTest.java
+++ b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformReplacementsTest.java
@@ -23,7 +23,7 @@ import static org.junit.Assert.assertThat;
import com.google.common.collect.ImmutableMap;
import java.util.Collections;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.AppliedPTransform;
import org.apache.beam.sdk.transforms.Create;
@@ -47,7 +47,7 @@ import org.junit.runners.JUnit4;
public class PTransformReplacementsTest {
@Rule public TestPipeline pipeline = TestPipeline.create().enableAbandonedNodeEnforcement(false);
@Rule public ExpectedException thrown = ExpectedException.none();
- private PCollection<Long> mainInput = pipeline.apply(CountingInput.unbounded());
+ private PCollection<Long> mainInput = pipeline.apply(GenerateSequence.from(0));
private PCollectionView<String> sideInput =
pipeline.apply(Create.of("foo")).apply(View.<String>asSingleton());
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformsTest.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformsTest.java b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformsTest.java
index 4e3cdb6..4ef70c0 100644
--- a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformsTest.java
+++ b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PTransformsTest.java
@@ -35,9 +35,8 @@ import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.common.runner.v1.RunnerApi;
import org.apache.beam.sdk.common.runner.v1.RunnerApi.Components;
import org.apache.beam.sdk.common.runner.v1.RunnerApi.PTransform;
-import org.apache.beam.sdk.io.CountingInput;
-import org.apache.beam.sdk.io.CountingInput.UnboundedCountingInput;
import org.apache.beam.sdk.io.CountingSource;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.Read;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.AppliedPTransform;
@@ -75,7 +74,7 @@ public class PTransformsTest {
TestPipeline compositeReadPipeline = TestPipeline.create();
ToAndFromProtoSpec compositeRead =
ToAndFromProtoSpec.composite(
- countingInput(compositeReadPipeline),
+ generateSequence(compositeReadPipeline),
ToAndFromProtoSpec.leaf(read(compositeReadPipeline)));
return ImmutableList.<ToAndFromProtoSpec>builder()
.add(readLeaf)
@@ -152,11 +151,11 @@ public class PTransformsTest {
@ProcessElement public void process(ProcessContext context) {}
}
- private static AppliedPTransform<?, ?, ?> countingInput(Pipeline pipeline) {
- UnboundedCountingInput input = CountingInput.unbounded();
- PCollection<Long> pcollection = pipeline.apply(input);
- return AppliedPTransform.<PBegin, PCollection<Long>, UnboundedCountingInput>of(
- "Count", pipeline.begin().expand(), pcollection.expand(), input, pipeline);
+ private static AppliedPTransform<?, ?, ?> generateSequence(Pipeline pipeline) {
+ GenerateSequence sequence = GenerateSequence.from(0);
+ PCollection<Long> pcollection = pipeline.apply(sequence);
+ return AppliedPTransform.<PBegin, PCollection<Long>, GenerateSequence>of(
+ "Count", pipeline.begin().expand(), pcollection.expand(), sequence, pipeline);
}
private static AppliedPTransform<?, ?, ?> read(Pipeline pipeline) {
@@ -169,7 +168,7 @@ public class PTransformsTest {
private static AppliedPTransform<?, ?, ?> multiMultiParDo(Pipeline pipeline) {
PCollectionView<String> view =
pipeline.apply(Create.of("foo")).apply(View.<String>asSingleton());
- PCollection<Long> input = pipeline.apply(CountingInput.unbounded());
+ PCollection<Long> input = pipeline.apply(GenerateSequence.from(0));
ParDo.MultiOutput<Long, KV<Long, String>> parDo =
ParDo.of(new TestDoFn())
.withSideInputs(view)
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/SdkComponentsTest.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/SdkComponentsTest.java b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/SdkComponentsTest.java
index 895aec4..021f19c 100644
--- a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/SdkComponentsTest.java
+++ b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/SdkComponentsTest.java
@@ -34,7 +34,7 @@ import org.apache.beam.sdk.coders.SetCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarLongCoder;
import org.apache.beam.sdk.common.runner.v1.RunnerApi.Components;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.AppliedPTransform;
import org.apache.beam.sdk.transforms.Create;
@@ -106,7 +106,7 @@ public class SdkComponentsTest {
@Test
public void registerTransformAfterChildren() throws IOException {
Create.Values<Long> create = Create.of(1L, 2L, 3L);
- CountingInput.UnboundedCountingInput createChild = CountingInput.unbounded();
+ GenerateSequence createChild = GenerateSequence.from(0);
PCollection<Long> pt = pipeline.apply(create);
String userName = "my_transform";
@@ -115,7 +115,7 @@ public class SdkComponentsTest {
AppliedPTransform.<PBegin, PCollection<Long>, Create.Values<Long>>of(
userName, pipeline.begin().expand(), pt.expand(), create, pipeline);
AppliedPTransform<?, ?, ?> childTransform =
- AppliedPTransform.<PBegin, PCollection<Long>, CountingInput.UnboundedCountingInput>of(
+ AppliedPTransform.<PBegin, PCollection<Long>, GenerateSequence>of(
childUserName, pipeline.begin().expand(), pt.expand(), createChild, pipeline);
String childId = components.registerPTransform(childTransform,
@@ -159,7 +159,7 @@ public class SdkComponentsTest {
@Test
public void registerTransformWithUnregisteredChildren() throws IOException {
Create.Values<Long> create = Create.of(1L, 2L, 3L);
- CountingInput.UnboundedCountingInput createChild = CountingInput.unbounded();
+ GenerateSequence createChild = GenerateSequence.from(0);
PCollection<Long> pt = pipeline.apply(create);
String userName = "my_transform";
@@ -168,7 +168,7 @@ public class SdkComponentsTest {
AppliedPTransform.<PBegin, PCollection<Long>, Create.Values<Long>>of(
userName, pipeline.begin().expand(), pt.expand(), create, pipeline);
AppliedPTransform<?, ?, ?> childTransform =
- AppliedPTransform.<PBegin, PCollection<Long>, CountingInput.UnboundedCountingInput>of(
+ AppliedPTransform.<PBegin, PCollection<Long>, GenerateSequence>of(
childUserName, pipeline.begin().expand(), pt.expand(), createChild, pipeline);
thrown.expect(IllegalArgumentException.class);
@@ -179,7 +179,7 @@ public class SdkComponentsTest {
@Test
public void registerPCollection() throws IOException {
- PCollection<Long> pCollection = pipeline.apply(CountingInput.unbounded()).setName("foo");
+ PCollection<Long> pCollection = pipeline.apply(GenerateSequence.from(0)).setName("foo");
String id = components.registerPCollection(pCollection);
assertThat(id, equalTo("foo"));
components.toComponents().getPcollectionsOrThrow(id);
@@ -188,10 +188,10 @@ public class SdkComponentsTest {
@Test
public void registerPCollectionExistingNameCollision() throws IOException {
PCollection<Long> pCollection =
- pipeline.apply("FirstCount", CountingInput.unbounded()).setName("foo");
+ pipeline.apply("FirstCount", GenerateSequence.from(0)).setName("foo");
String firstId = components.registerPCollection(pCollection);
PCollection<Long> duplicate =
- pipeline.apply("SecondCount", CountingInput.unbounded()).setName("foo");
+ pipeline.apply("SecondCount", GenerateSequence.from(0)).setName("foo");
String secondId = components.registerPCollection(duplicate);
assertThat(firstId, equalTo("foo"));
assertThat(secondId, containsString("foo"));
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/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 b44c890..2afb6c3 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
@@ -24,11 +24,10 @@ import static org.junit.Assert.assertThat;
import com.google.common.collect.Iterables;
import java.io.Serializable;
-import java.util.ArrayList;
import java.util.List;
import org.apache.beam.sdk.coders.StringUtf8Coder;
-import org.apache.beam.sdk.io.CountingInput;
import org.apache.beam.sdk.io.CountingSource;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.Read;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.AppliedPTransform;
@@ -89,11 +88,10 @@ public class DirectGraphVisitorTest implements Serializable {
public void getRootTransformsContainsRootTransforms() {
PCollection<String> created = p.apply(Create.of("foo", "bar"));
PCollection<Long> counted = p.apply(Read.from(CountingSource.upTo(1234L)));
- PCollection<Long> unCounted = p.apply(CountingInput.unbounded());
+ PCollection<Long> unCounted = p.apply(GenerateSequence.from(0));
p.traverseTopologically(visitor);
DirectGraph graph = visitor.getGraph();
assertThat(graph.getRootTransforms(), hasSize(3));
- List<PTransform<?, ?>> unapplied = new ArrayList<>();
assertThat(
graph.getRootTransforms(),
Matchers.<AppliedPTransform<?, ?, ?>>containsInAnyOrder(
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
index 246c111..f9063f0 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
@@ -52,8 +52,8 @@ import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.beam.sdk.coders.VarLongCoder;
import org.apache.beam.sdk.io.BoundedSource;
-import org.apache.beam.sdk.io.CountingInput;
import org.apache.beam.sdk.io.CountingSource;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.Read;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
@@ -246,7 +246,7 @@ public class DirectRunnerTest implements Serializable {
opts.setRunner(DirectRunner.class);
final Pipeline p = Pipeline.create(opts);
- p.apply(CountingInput.unbounded().withRate(1L, Duration.standardSeconds(1)));
+ p.apply(GenerateSequence.from(0).withRate(1L, Duration.standardSeconds(1)));
final BlockingQueue<PipelineResult> resultExchange = new ArrayBlockingQueue<>(1);
Runnable cancelRunnable = new Runnable() {
@@ -513,8 +513,7 @@ public class DirectRunnerTest implements Serializable {
@Test
public void testUnencodableOutputFromBoundedRead() throws Exception {
Pipeline p = getPipeline();
- PCollection<Long> pCollection =
- p.apply(CountingInput.upTo(10)).setCoder(new LongNoDecodeCoder());
+ p.apply(GenerateSequence.fromTo(0, 10)).setCoder(new LongNoDecodeCoder());
thrown.expectCause(isA(CoderException.class));
thrown.expectMessage("Cannot decode a long");
@@ -524,8 +523,7 @@ public class DirectRunnerTest implements Serializable {
@Test
public void testUnencodableOutputFromUnboundedRead() {
Pipeline p = getPipeline();
- PCollection<Long> pCollection =
- p.apply(CountingInput.unbounded()).setCoder(new LongNoDecodeCoder());
+ p.apply(GenerateSequence.from(0)).setCoder(new LongNoDecodeCoder());
thrown.expectCause(isA(CoderException.class));
thrown.expectMessage("Cannot decode a long");
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/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 7a65493..35b6709 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
@@ -44,7 +44,7 @@ import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate;
import org.apache.beam.sdk.coders.ByteArrayCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarIntCoder;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.AppliedPTransform;
@@ -103,7 +103,7 @@ public class EvaluationContextTest {
created = p.apply(Create.of(1, 2, 3));
downstream = created.apply(WithKeys.<String, Integer>of("foo"));
view = created.apply(View.<Integer>asIterable());
- unbounded = p.apply(CountingInput.unbounded());
+ unbounded = p.apply(GenerateSequence.from(0));
KeyedPValueTrackingVisitor keyedPValueTrackingVisitor = KeyedPValueTrackingVisitor.create();
p.traverseTopologically(keyedPValueTrackingVisitor);
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/runners/flink/src/test/java/org/apache/beam/runners/flink/ReadSourceITCase.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/ReadSourceITCase.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/ReadSourceITCase.java
index 44c9017..8302983 100644
--- a/runners/flink/src/test/java/org/apache/beam/runners/flink/ReadSourceITCase.java
+++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/ReadSourceITCase.java
@@ -21,7 +21,7 @@ import com.google.common.base.Joiner;
import java.io.File;
import java.net.URI;
import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
@@ -68,7 +68,7 @@ public class ReadSourceITCase extends JavaProgramTestBase {
Pipeline p = FlinkTestPipeline.createForBatch();
PCollection<String> result = p
- .apply(CountingInput.upTo(10))
+ .apply(GenerateSequence.fromTo(0, 10))
.apply(ParDo.of(new DoFn<Long, String>() {
@ProcessElement
public void processElement(ProcessContext c) throws Exception {
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/runners/flink/src/test/java/org/apache/beam/runners/flink/ReadSourceStreamingITCase.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/ReadSourceStreamingITCase.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/ReadSourceStreamingITCase.java
index 79b7882..c7a044e 100644
--- a/runners/flink/src/test/java/org/apache/beam/runners/flink/ReadSourceStreamingITCase.java
+++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/ReadSourceStreamingITCase.java
@@ -19,7 +19,7 @@ package org.apache.beam.runners.flink;
import com.google.common.base.Joiner;
import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
@@ -58,7 +58,7 @@ public class ReadSourceStreamingITCase extends StreamingProgramTestBase {
Pipeline p = FlinkTestPipeline.createForStreaming();
p
- .apply(CountingInput.upTo(10))
+ .apply(GenerateSequence.fromTo(0, 10))
.apply(ParDo.of(new DoFn<Long, String>() {
@ProcessElement
public void processElement(ProcessContext c) throws Exception {
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/StreamingSourceMetricsTest.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/StreamingSourceMetricsTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/StreamingSourceMetricsTest.java
index ea76d31..85fdf72 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/StreamingSourceMetricsTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/StreamingSourceMetricsTest.java
@@ -27,11 +27,12 @@ import org.apache.beam.runners.spark.PipelineRule;
import org.apache.beam.runners.spark.TestSparkPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.Source;
import org.apache.beam.sdk.metrics.MetricNameFilter;
import org.apache.beam.sdk.metrics.MetricQueryResults;
import org.apache.beam.sdk.metrics.MetricsFilter;
+import org.joda.time.Duration;
import org.junit.Rule;
import org.junit.Test;
@@ -53,7 +54,9 @@ public class StreamingSourceMetricsTest implements Serializable {
final long numElements = 1000;
- pipeline.apply(CountingInput.unbounded().withMaxNumRecords(numElements));
+ pipeline.apply(
+ // Use maxReadTime to force unbounded mode.
+ GenerateSequence.fromTo(0, numElements).withMaxReadTime(Duration.standardDays(1)));
PipelineResult pipelineResult = pipeline.run();
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java
index dd018f4..b66a8b2 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java
@@ -28,7 +28,6 @@ import org.apache.beam.sdk.coders.AvroCoder;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.DefaultCoder;
import org.apache.beam.sdk.coders.VarLongCoder;
-import org.apache.beam.sdk.io.CountingInput.UnboundedCountingInput;
import org.apache.beam.sdk.io.UnboundedSource.UnboundedReader;
import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.metrics.Metrics;
@@ -39,44 +38,26 @@ import org.joda.time.Duration;
import org.joda.time.Instant;
/**
- * A source that produces longs. When used as a {@link BoundedSource}, {@link CountingSource}
- * starts at {@code 0} and counts up to a specified maximum. When used as an
- * {@link UnboundedSource}, it counts up to {@link Long#MAX_VALUE} and then never produces more
- * output. (In practice, this limit should never be reached.)
+ * Most users should use {@link GenerateSequence} instead.
+ *
+ * <p>A source that produces longs. When used as a {@link BoundedSource}, {@link CountingSource}
+ * starts at {@code 0} and counts up to a specified maximum. When used as an {@link
+ * UnboundedSource}, it counts up to {@link Long#MAX_VALUE} and then never produces more output. (In
+ * practice, this limit should never be reached.)
*
* <p>The bounded {@link CountingSource} is implemented based on {@link OffsetBasedSource} and
* {@link OffsetBasedSource.OffsetBasedReader}, so it performs efficient initial splitting and it
* supports dynamic work rebalancing.
*
- * <p>To produce a bounded {@code PCollection<Long>}, use {@link CountingSource#upTo(long)}:
- *
- * <pre>{@code
- * Pipeline p = ...
- * PTransform<PBegin, PCollection<Long>> producer = CountingInput.upTo(1000);
- * PCollection<Long> bounded = p.apply(producer);
- * }</pre>
- *
- * <p>To produce an unbounded {@code PCollection<Long>}, use {@link CountingInput#unbounded()},
- * calling {@link UnboundedCountingInput#withTimestampFn(SerializableFunction)} to provide values
- * with timestamps other than {@link Instant#now}.
- *
- * <pre>{@code
- * Pipeline p = ...
- *
- * // To create an unbounded PCollection that uses processing time as the element timestamp.
- * PCollection<Long> unbounded = p.apply(CountingInput.unbounded());
- * // Or, to create an unbounded source that uses a provided function to set the element timestamp.
- * PCollection<Long> unboundedWithTimestamps =
- * p.apply(CountingInput.unbounded().withTimestampFn(someFn));
- *
- * }</pre>
+ * <p>To produce a bounded source, use {@link #createSourceForSubrange(long, long)}. To produce an
+ * unbounded source, use {@link #createUnboundedFrom(long)}.
*/
public class CountingSource {
/**
* Creates a {@link BoundedSource} that will produce the specified number of elements,
* from {@code 0} to {@code numElements - 1}.
*
- * @deprecated use {@link CountingInput#upTo(long)} instead
+ * @deprecated use {@link GenerateSequence} instead
*/
@Deprecated
public static BoundedSource<Long> upTo(long numElements) {
@@ -117,7 +98,7 @@ public class CountingSource {
* <p>Elements in the resulting {@link PCollection PCollection<Long>} will have timestamps
* corresponding to processing time at element generation, provided by {@link Instant#now}.
*
- * @deprecated use {@link CountingInput#unbounded()} instead
+ * @deprecated use {@link GenerateSequence} instead
*/
@Deprecated
public static UnboundedSource<Long, CounterMark> unbounded() {
@@ -133,8 +114,8 @@ public class CountingSource {
*
* <p>Note that the timestamps produced by {@code timestampFn} may not decrease.
*
- * @deprecated use {@link CountingInput#unbounded()} and call
- * {@link UnboundedCountingInput#withTimestampFn(SerializableFunction)} instead
+ * @deprecated use {@link GenerateSequence} and call
+ * {@link GenerateSequence#withTimestampFn(SerializableFunction)} instead
*/
@Deprecated
public static UnboundedSource<Long, CounterMark> unboundedWithTimestampFn(
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/sdks/java/core/src/main/java/org/apache/beam/sdk/io/GenerateSequence.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/GenerateSequence.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/GenerateSequence.java
index 189539f..8a83d39 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/GenerateSequence.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/GenerateSequence.java
@@ -109,6 +109,11 @@ public abstract class GenerateSequence extends PTransform<PBegin, PCollection<Lo
.build();
}
+ /** Specifies to generate the range [from, to). */
+ public static GenerateSequence fromTo(long from, long to) {
+ return from(from).to(to);
+ }
+
/** Specifies the maximum number to generate (exclusive). */
public GenerateSequence to(long to) {
checkArgument(
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollection.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollection.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollection.java
index b69185b..f89d2aa 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollection.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollection.java
@@ -20,7 +20,7 @@ package org.apache.beam.sdk.values;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.io.BoundedSource.BoundedReader;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.Read;
import org.apache.beam.sdk.io.UnboundedSource.UnboundedReader;
import org.apache.beam.sdk.transforms.Create;
@@ -39,9 +39,10 @@ import org.apache.beam.sdk.util.WindowingStrategy;
* be passed as the inputs of other PTransforms.
*
* <p>Some root transforms produce bounded {@code PCollections} and others
- * produce unbounded ones. For example, {@link CountingInput#upTo} produces a fixed set of integers,
- * so it produces a bounded {@link PCollection}. {@link CountingInput#unbounded} produces all
- * integers as an infinite stream, so it produces an unbounded {@link PCollection}.
+ * produce unbounded ones. For example, {@link GenerateSequence#fromTo} produces a fixed set of
+ * integers, so it produces a bounded {@link PCollection}. {@link GenerateSequence#from} without
+ * a {@link GenerateSequence#to} produces all integers as an infinite stream, so it produces an
+ * unbounded {@link PCollection}.
*
* <p>Each element in a {@link PCollection} has an associated timestamp. Readers assign timestamps
* to elements when they create {@link PCollection PCollections}, and other
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java
index 75cabf2..9604c3f 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java
@@ -36,9 +36,7 @@ import java.util.Map;
import java.util.Set;
import org.apache.beam.sdk.Pipeline.PipelineExecutionException;
import org.apache.beam.sdk.Pipeline.PipelineVisitor;
-import org.apache.beam.sdk.io.CountingInput;
-import org.apache.beam.sdk.io.CountingInput.BoundedCountingInput;
-import org.apache.beam.sdk.io.CountingInput.UnboundedCountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptions.CheckEnabled;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
@@ -317,8 +315,8 @@ public class PipelineTest {
@Test
public void testReplaceAll() {
pipeline.enableAbandonedNodeEnforcement(false);
- pipeline.apply(CountingInput.unbounded());
- pipeline.apply(CountingInput.upTo(100L));
+ pipeline.apply("unbounded", GenerateSequence.from(0));
+ pipeline.apply("bounded", GenerateSequence.fromTo(0, 100));
pipeline.replaceAll(
ImmutableList.of(
@@ -326,18 +324,18 @@ public class PipelineTest {
new PTransformMatcher() {
@Override
public boolean matches(AppliedPTransform<?, ?, ?> application) {
- return application.getTransform() instanceof UnboundedCountingInput;
+ return application.getTransform() instanceof GenerateSequence;
}
},
- new UnboundedCountingInputOverride()),
+ new GenerateSequenceToCreateOverride()),
PTransformOverride.of(
new PTransformMatcher() {
@Override
public boolean matches(AppliedPTransform<?, ?, ?> application) {
- return application.getTransform() instanceof BoundedCountingInput;
+ return application.getTransform() instanceof Create.Values;
}
},
- new BoundedCountingInputOverride())));
+ new CreateValuesToEmptyFlattenOverride())));
pipeline.traverseTopologically(
new PipelineVisitor.Defaults() {
@Override
@@ -348,9 +346,9 @@ public class PipelineTest {
not(
anyOf(
Matchers.<Class<? extends PTransform>>equalTo(
- UnboundedCountingInput.class),
+ GenerateSequence.class),
Matchers.<Class<? extends PTransform>>equalTo(
- BoundedCountingInput.class))));
+ Create.Values.class))));
}
return CompositeBehavior.ENTER_TRANSFORM;
}
@@ -364,7 +362,7 @@ public class PipelineTest {
@Test
public void testReplaceAllIncomplete() {
pipeline.enableAbandonedNodeEnforcement(false);
- pipeline.apply(CountingInput.unbounded());
+ pipeline.apply(GenerateSequence.from(0));
// The order is such that the output of the second will match the first, which is not permitted
thrown.expect(IllegalStateException.class);
@@ -374,18 +372,18 @@ public class PipelineTest {
new PTransformMatcher() {
@Override
public boolean matches(AppliedPTransform<?, ?, ?> application) {
- return application.getTransform() instanceof BoundedCountingInput;
+ return application.getTransform() instanceof Create.Values;
}
},
- new BoundedCountingInputOverride()),
+ new CreateValuesToEmptyFlattenOverride()),
PTransformOverride.of(
new PTransformMatcher() {
@Override
public boolean matches(AppliedPTransform<?, ?, ?> application) {
- return application.getTransform() instanceof UnboundedCountingInput;
+ return application.getTransform() instanceof GenerateSequence;
}
},
- new UnboundedCountingInputOverride())));
+ new GenerateSequenceToCreateOverride())));
}
@Test
@@ -455,11 +453,11 @@ public class PipelineTest {
assertThat(names, not(hasItem("original_application/custom_name2")));
}
- static class BoundedCountingInputOverride
- implements PTransformOverrideFactory<PBegin, PCollection<Long>, BoundedCountingInput> {
+ static class GenerateSequenceToCreateOverride
+ implements PTransformOverrideFactory<PBegin, PCollection<Long>, GenerateSequence> {
@Override
public PTransformReplacement<PBegin, PCollection<Long>> getReplacementTransform(
- AppliedPTransform<PBegin, PCollection<Long>, BoundedCountingInput> transform) {
+ AppliedPTransform<PBegin, PCollection<Long>, GenerateSequence> transform) {
return PTransformReplacement.of(transform.getPipeline().begin(), Create.of(0L));
}
@@ -476,18 +474,28 @@ public class PipelineTest {
TaggedPValue.of(replacement.getKey(), replacement.getValue())));
}
}
- static class UnboundedCountingInputOverride
- implements PTransformOverrideFactory<PBegin, PCollection<Long>, UnboundedCountingInput> {
+ private static class EmptyFlatten<T> extends PTransform<PBegin, PCollection<T>> {
@Override
- public PTransformReplacement<PBegin, PCollection<Long>> getReplacementTransform(
- AppliedPTransform<PBegin, PCollection<Long>, UnboundedCountingInput> transform) {
- return PTransformReplacement.of(transform.getPipeline().begin(), CountingInput.upTo(100L));
+ public PCollection<T> expand(PBegin input) {
+ PCollectionList<T> empty = PCollectionList.empty(input.getPipeline());
+ return empty.apply(Flatten.<T>pCollections());
+ }
+ }
+
+ static class CreateValuesToEmptyFlattenOverride<T>
+ implements PTransformOverrideFactory<PBegin, PCollection<T>, Create.Values<T>> {
+
+ @Override
+ public PTransformReplacement<PBegin, PCollection<T>> getReplacementTransform(
+ AppliedPTransform<PBegin, PCollection<T>, Create.Values<T>> transform) {
+ return PTransformReplacement.of(
+ transform.getPipeline().begin(), new EmptyFlatten<T>());
}
@Override
public Map<PValue, ReplacementOutput> mapOutputs(
- Map<TupleTag<?>, PValue> outputs, PCollection<Long> newOutput) {
+ Map<TupleTag<?>, PValue> outputs, PCollection<T> newOutput) {
Map.Entry<TupleTag<?>, PValue> original = Iterables.getOnlyElement(outputs.entrySet());
Map.Entry<TupleTag<?>, PValue> replacement =
Iterables.getOnlyElement(newOutput.expand().entrySet());
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/sdks/java/core/src/test/java/org/apache/beam/sdk/io/GenerateSequenceTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/GenerateSequenceTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/GenerateSequenceTest.java
index 49af479..e2c185b 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/GenerateSequenceTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/GenerateSequenceTest.java
@@ -65,7 +65,7 @@ public class GenerateSequenceTest {
@Category(ValidatesRunner.class)
public void testBoundedInput() {
long numElements = 1000;
- PCollection<Long> input = p.apply(GenerateSequence.from(0).to(numElements));
+ PCollection<Long> input = p.apply(GenerateSequence.fromTo(0, numElements));
addCountingAsserts(input, 0, numElements);
p.run();
@@ -74,7 +74,7 @@ public class GenerateSequenceTest {
@Test
@Category(ValidatesRunner.class)
public void testEmptyBoundedInput() {
- PCollection<Long> input = p.apply(GenerateSequence.from(0).to(0));
+ PCollection<Long> input = p.apply(GenerateSequence.fromTo(0, 0));
PAssert.that(input).empty();
p.run();
@@ -83,7 +83,7 @@ public class GenerateSequenceTest {
@Test
@Category(ValidatesRunner.class)
public void testEmptyBoundedInputSubrange() {
- PCollection<Long> input = p.apply(GenerateSequence.from(42).to(42));
+ PCollection<Long> input = p.apply(GenerateSequence.fromTo(42, 42));
PAssert.that(input).empty();
p.run();
@@ -94,7 +94,7 @@ public class GenerateSequenceTest {
public void testBoundedInputSubrange() {
long start = 10;
long end = 1000;
- PCollection<Long> input = p.apply(GenerateSequence.from(start).to(end));
+ PCollection<Long> input = p.apply(GenerateSequence.fromTo(start, end));
addCountingAsserts(input, start, end);
p.run();
@@ -102,7 +102,7 @@ public class GenerateSequenceTest {
@Test
public void testBoundedDisplayData() {
- PTransform<?, ?> input = GenerateSequence.from(0).to(1234);
+ PTransform<?, ?> input = GenerateSequence.fromTo(0, 1234);
DisplayData displayData = DisplayData.from(input);
assertThat(displayData, hasDisplayItem("from", 0));
assertThat(displayData, hasDisplayItem("to", 1234));
@@ -110,7 +110,7 @@ public class GenerateSequenceTest {
@Test
public void testBoundedDisplayDataSubrange() {
- PTransform<?, ?> input = GenerateSequence.from(12).to(1234);
+ PTransform<?, ?> input = GenerateSequence.fromTo(12, 1234);
DisplayData displayData = DisplayData.from(input);
assertThat(displayData, hasDisplayItem("from", 12));
assertThat(displayData, hasDisplayItem("to", 1234));
@@ -124,7 +124,7 @@ public class GenerateSequenceTest {
long elemsPerPeriod = 10L;
Duration periodLength = Duration.millis(8);
PCollection<Long> input =
- p.apply(GenerateSequence.from(0).to(numElements).withRate(elemsPerPeriod, periodLength));
+ p.apply(GenerateSequence.fromTo(0, numElements).withRate(elemsPerPeriod, periodLength));
addCountingAsserts(input, 0, numElements);
long expectedRuntimeMillis = (periodLength.getMillis() * numElements) / elemsPerPeriod;
@@ -147,7 +147,7 @@ public class GenerateSequenceTest {
long numElements = 1000;
PCollection<Long> input =
- p.apply(GenerateSequence.from(0).to(numElements).withTimestampFn(new ValueAsTimestampFn()));
+ p.apply(GenerateSequence.fromTo(0, numElements).withTimestampFn(new ValueAsTimestampFn()));
addCountingAsserts(input, 0, numElements);
PCollection<Long> diffs =
@@ -172,7 +172,7 @@ public class GenerateSequenceTest {
};
PTransform<?, ?> input =
- GenerateSequence.from(0).to(1234).withMaxReadTime(maxReadTime).withTimestampFn(timestampFn);
+ GenerateSequence.fromTo(0, 1234).withMaxReadTime(maxReadTime).withTimestampFn(timestampFn);
DisplayData displayData = DisplayData.from(input);
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java
index afe384d..312a433 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java
@@ -29,7 +29,7 @@ import static org.junit.Assert.assertThat;
import java.io.Serializable;
import org.apache.beam.sdk.PipelineResult;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.testing.UsesAttemptedMetrics;
import org.apache.beam.sdk.testing.UsesCommittedMetrics;
@@ -37,10 +37,10 @@ import org.apache.beam.sdk.testing.ValidatesRunner;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.TupleTagList;
import org.hamcrest.CoreMatchers;
+import org.joda.time.Duration;
import org.joda.time.Instant;
import org.junit.After;
import org.junit.Rule;
@@ -236,7 +236,7 @@ public class MetricsTest implements Serializable {
public void testBoundedSourceMetrics() {
long numElements = 1000;
- PCollection<Long> input = pipeline.apply(CountingInput.upTo(numElements));
+ pipeline.apply(GenerateSequence.fromTo(0, numElements));
PipelineResult pipelineResult = pipeline.run();
@@ -257,8 +257,10 @@ public class MetricsTest implements Serializable {
public void testUnboundedSourceMetrics() {
long numElements = 1000;
- PCollection<Long> input = pipeline
- .apply((CountingInput.unbounded()).withMaxNumRecords(numElements));
+
+ // Use withMaxReadTime to force unbounded mode.
+ pipeline.apply(
+ GenerateSequence.fromTo(0, numElements).withMaxReadTime(Duration.standardDays(1)));
PipelineResult pipelineResult = pipeline.run();
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/TransformHierarchyTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/TransformHierarchyTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/TransformHierarchyTest.java
index 3638fc8..1b884e2 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/TransformHierarchyTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/TransformHierarchyTest.java
@@ -32,9 +32,8 @@ import java.util.Map.Entry;
import java.util.Set;
import org.apache.beam.sdk.Pipeline.PipelineVisitor;
import org.apache.beam.sdk.Pipeline.PipelineVisitor.Defaults;
-import org.apache.beam.sdk.io.CountingInput;
-import org.apache.beam.sdk.io.CountingInput.UnboundedCountingInput;
import org.apache.beam.sdk.io.CountingSource;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.Read;
import org.apache.beam.sdk.runners.PTransformOverrideFactory.ReplacementOutput;
import org.apache.beam.sdk.runners.TransformHierarchy.Node;
@@ -254,7 +253,7 @@ public class TransformHierarchyTest implements Serializable {
}
});
- UnboundedCountingInput genUpstream = CountingInput.unbounded();
+ GenerateSequence genUpstream = GenerateSequence.from(0);
PCollection<Long> upstream = pipeline.apply(genUpstream);
PCollection<Long> output = upstream.apply("Original", originalParDo);
hierarchy.pushNode("Upstream", pipeline.begin(), genUpstream);
@@ -417,7 +416,7 @@ public class TransformHierarchyTest implements Serializable {
}
});
- UnboundedCountingInput genUpstream = CountingInput.unbounded();
+ GenerateSequence genUpstream = GenerateSequence.from(0);
PCollection<Long> upstream = pipeline.apply(genUpstream);
PCollection<Long> output = upstream.apply("Original", originalParDo);
Node upstreamNode = hierarchy.pushNode("Upstream", pipeline.begin(), genUpstream);
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/GatherAllPanesTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/GatherAllPanesTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/GatherAllPanesTest.java
index 969bbc4..b546158 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/GatherAllPanesTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/GatherAllPanesTest.java
@@ -21,7 +21,7 @@ import static org.junit.Assert.fail;
import com.google.common.collect.Iterables;
import java.io.Serializable;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.transforms.Flatten;
import org.apache.beam.sdk.transforms.GroupByKey;
import org.apache.beam.sdk.transforms.SerializableFunction;
@@ -54,7 +54,7 @@ public class GatherAllPanesTest implements Serializable {
@Category(NeedsRunner.class)
public void singlePaneSingleReifiedPane() {
PCollection<Iterable<ValueInSingleWindow<Iterable<Long>>>> accumulatedPanes =
- p.apply(CountingInput.upTo(20000))
+ p.apply(GenerateSequence.fromTo(0, 20000))
.apply(
WithTimestamps.of(
new SerializableFunction<Long, Instant>() {
@@ -95,8 +95,8 @@ public class GatherAllPanesTest implements Serializable {
@Test
@Category(NeedsRunner.class)
public void multiplePanesMultipleReifiedPane() {
- PCollection<Long> someElems = p.apply("someLongs", CountingInput.upTo(20000));
- PCollection<Long> otherElems = p.apply("otherLongs", CountingInput.upTo(20000));
+ PCollection<Long> someElems = p.apply("someLongs", GenerateSequence.fromTo(0, 20000));
+ PCollection<Long> otherElems = p.apply("otherLongs", GenerateSequence.fromTo(0, 20000));
PCollection<Iterable<ValueInSingleWindow<Iterable<Long>>>> accumulatedPanes =
PCollectionList.of(someElems)
.and(otherElems)
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PAssertTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PAssertTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PAssertTest.java
index 3528797..a5362b1 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PAssertTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PAssertTest.java
@@ -37,7 +37,7 @@ import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.AtomicCoder;
import org.apache.beam.sdk.coders.CoderException;
import org.apache.beam.sdk.coders.VarLongCoder;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.transforms.Sum;
@@ -425,7 +425,7 @@ public class PAssertTest implements Serializable {
@Test
@Category(ValidatesRunner.class)
public void testEmptyFalse() throws Exception {
- PCollection<Long> vals = pipeline.apply(CountingInput.upTo(5L));
+ PCollection<Long> vals = pipeline.apply(GenerateSequence.fromTo(0, 5));
PAssert.that("Vals should have been empty", vals).empty();
Throwable thrown = runExpectingAssertionFailure(pipeline);
@@ -439,7 +439,7 @@ public class PAssertTest implements Serializable {
@Test
@Category(ValidatesRunner.class)
public void testEmptyFalseDefaultReasonString() throws Exception {
- PCollection<Long> vals = pipeline.apply(CountingInput.upTo(5L));
+ PCollection<Long> vals = pipeline.apply(GenerateSequence.fromTo(0, 5));
PAssert.that(vals).empty();
Throwable thrown = runExpectingAssertionFailure(pipeline);
@@ -447,14 +447,14 @@ public class PAssertTest implements Serializable {
String message = thrown.getMessage();
assertThat(message,
- containsString("CountingInput.BoundedCountingInput/Read(BoundedCountingSource).out"));
+ containsString("GenerateSequence/Read(BoundedCountingSource).out"));
assertThat(message, containsString("Expected: iterable over [] in any order"));
}
@Test
@Category(ValidatesRunner.class)
public void testAssertionSiteIsCapturedWithMessage() throws Exception {
- PCollection<Long> vals = pipeline.apply(CountingInput.upTo(5L));
+ PCollection<Long> vals = pipeline.apply(GenerateSequence.fromTo(0, 5));
assertThatCollectionIsEmptyWithMessage(vals);
Throwable thrown = runExpectingAssertionFailure(pipeline);
@@ -473,7 +473,7 @@ public class PAssertTest implements Serializable {
@Test
@Category(ValidatesRunner.class)
public void testAssertionSiteIsCapturedWithoutMessage() throws Exception {
- PCollection<Long> vals = pipeline.apply(CountingInput.upTo(5L));
+ PCollection<Long> vals = pipeline.apply(GenerateSequence.fromTo(0, 5));
assertThatCollectionIsEmptyWithoutMessage(vals);
Throwable thrown = runExpectingAssertionFailure(pipeline);
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/FlattenTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/FlattenTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/FlattenTest.java
index a4f2545..fca804b 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/FlattenTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/FlattenTest.java
@@ -45,7 +45,7 @@ import org.apache.beam.sdk.coders.SetCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarLongCoder;
import org.apache.beam.sdk.coders.VoidCoder;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.testing.FlattenWithHeterogeneousCoders;
import org.apache.beam.sdk.testing.NeedsRunner;
import org.apache.beam.sdk.testing.PAssert;
@@ -138,9 +138,9 @@ public class FlattenTest implements Serializable {
@Category(ValidatesRunner.class)
public void testFlattenInputMultipleCopies() {
int count = 5;
- PCollection<Long> longs = p.apply("mkLines", CountingInput.upTo(count));
+ PCollection<Long> longs = p.apply("mkLines", GenerateSequence.fromTo(0, count));
PCollection<Long> biggerLongs =
- p.apply("mkOtherLines", CountingInput.upTo(count))
+ p.apply("mkOtherLines", GenerateSequence.fromTo(0, count))
.apply(
MapElements.via(
new SimpleFunction<Long, Long>() {
@@ -175,7 +175,7 @@ public class FlattenTest implements Serializable {
Create.of(0L, 1L, 2L, 3L, null, 4L, 5L, null, 6L, 7L, 8L, null, 9L)
.withCoder(NullableCoder.of(BigEndianLongCoder.of())));
PCollection<Long> varLongs =
- p.apply("VarLengthLongs", CountingInput.upTo(5L)).setCoder(VarLongCoder.of());
+ p.apply("VarLengthLongs", GenerateSequence.fromTo(0, 5)).setCoder(VarLongCoder.of());
PCollection<Long> flattened =
PCollectionList.of(bigEndianLongs)
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
index 589c744..3424f86 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
@@ -64,7 +64,7 @@ import org.apache.beam.sdk.coders.ListCoder;
import org.apache.beam.sdk.coders.SetCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarIntCoder;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.testing.NeedsRunner;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
@@ -889,7 +889,7 @@ public class ParDoTest implements Serializable {
@Test
public void testMultiOutputAppliedMultipleTimesDifferentOutputs() {
pipeline.enableAbandonedNodeEnforcement(false);
- PCollection<Long> longs = pipeline.apply(CountingInput.unbounded());
+ PCollection<Long> longs = pipeline.apply(GenerateSequence.from(0));
TupleTag<Long> mainOut = new TupleTag<>();
final TupleTag<String> valueAsString = new TupleTag<>();
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java
index 8bf022b..9fe7ec3 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java
@@ -39,7 +39,7 @@ import org.apache.beam.sdk.Pipeline.PipelineVisitor;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.Coder.NonDeterministicException;
import org.apache.beam.sdk.coders.StringUtf8Coder;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.runners.TransformHierarchy;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
@@ -319,7 +319,7 @@ public class WindowTest implements Serializable {
final PCollection<Long> initialWindows =
pipeline
- .apply(CountingInput.upTo(10L))
+ .apply(GenerateSequence.fromTo(0, 10))
.apply("AssignWindows", Window.into(new WindowOddEvenBuckets()));
// Sanity check the window assignment to demonstrate the baseline
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionListTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionListTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionListTest.java
index 76cba01..f687637 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionListTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionListTest.java
@@ -30,10 +30,11 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.Create;
import org.hamcrest.Matchers;
+import org.joda.time.Duration;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
@@ -61,34 +62,36 @@ public class PCollectionListTest {
public void testIterationOrder() {
Pipeline p = TestPipeline.create();
PCollection<Long> createOne = p.apply("CreateOne", Create.of(1L, 2L, 3L));
- PCollection<Long> boundedCount = p.apply("CountBounded", CountingInput.upTo(23L));
- PCollection<Long> unboundedCount = p.apply("CountUnbounded", CountingInput.unbounded());
+ PCollection<Long> boundedCount = p.apply("CountBounded", GenerateSequence.fromTo(0, 23));
+ PCollection<Long> unboundedCount = p.apply("CountUnbounded", GenerateSequence.from(0));
PCollection<Long> createTwo = p.apply("CreateTwo", Create.of(-1L, -2L));
- PCollection<Long> maxRecordsCount =
- p.apply("CountLimited", CountingInput.unbounded().withMaxNumRecords(22L));
+ PCollection<Long> maxReadTimeCount =
+ p.apply(
+ "CountLimited", GenerateSequence.from(0).withMaxReadTime(Duration.standardSeconds(5)));
ImmutableList<PCollection<Long>> counts =
- ImmutableList.of(boundedCount, maxRecordsCount, unboundedCount);
+ ImmutableList.of(boundedCount, maxReadTimeCount, unboundedCount);
// Build a PCollectionList from a list. This should have the same order as the input list.
PCollectionList<Long> pcList = PCollectionList.of(counts);
// Contains is the order-dependent matcher
assertThat(
pcList.getAll(),
- contains(boundedCount, maxRecordsCount, unboundedCount));
+ contains(boundedCount, maxReadTimeCount, unboundedCount));
// A list that is expanded with builder methods has the added value at the end
PCollectionList<Long> withOneCreate = pcList.and(createTwo);
assertThat(
- withOneCreate.getAll(), contains(boundedCount, maxRecordsCount, unboundedCount, createTwo));
+ withOneCreate.getAll(),
+ contains(boundedCount, maxReadTimeCount, unboundedCount, createTwo));
// Lists that are built entirely from the builder return outputs in the order they were added
PCollectionList<Long> fromEmpty =
PCollectionList.<Long>empty(p)
.and(unboundedCount)
.and(createOne)
- .and(ImmutableList.of(boundedCount, maxRecordsCount));
+ .and(ImmutableList.of(boundedCount, maxReadTimeCount));
assertThat(
- fromEmpty.getAll(), contains(unboundedCount, createOne, boundedCount, maxRecordsCount));
+ fromEmpty.getAll(), contains(unboundedCount, createOne, boundedCount, maxReadTimeCount));
Map<TupleTag<?>, PValue> expansion = fromEmpty.expand();
// Tag->PValue mappings are stable between expansions. They don't need to be stable across
@@ -96,7 +99,7 @@ public class PCollectionListTest {
assertThat(expansion, equalTo(fromEmpty.expand()));
List<PCollection<Long>> expectedList =
- ImmutableList.of(unboundedCount, createOne, boundedCount, maxRecordsCount);
+ ImmutableList.of(unboundedCount, createOne, boundedCount, maxReadTimeCount);
assertThat(expansion.values(), containsInAnyOrder(expectedList.toArray()));
}
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionTupleTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionTupleTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionTupleTest.java
index 9df0512..cc8761b 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionTupleTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionTupleTest.java
@@ -31,7 +31,7 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.testing.ValidatesRunner;
@@ -108,7 +108,7 @@ public final class PCollectionTupleTest implements Serializable {
public void testEquals() {
TestPipeline p = TestPipeline.create();
TupleTag<Long> longTag = new TupleTag<>();
- PCollection<Long> longs = p.apply(CountingInput.unbounded());
+ PCollection<Long> longs = p.apply(GenerateSequence.from(0));
TupleTag<String> strTag = new TupleTag<>();
PCollection<String> strs = p.apply(Create.of("foo", "bar"));
@@ -135,7 +135,7 @@ public final class PCollectionTupleTest implements Serializable {
TupleTag<Long> longTag = new TupleTag<>();
Pipeline p = TestPipeline.create();
- PCollection<Long> longs = p.apply(CountingInput.upTo(100L));
+ PCollection<Long> longs = p.apply(GenerateSequence.fromTo(0, 100));
PCollection<String> strs = p.apply(Create.of("foo", "bar", "baz"));
PCollection<Integer> ints = longs.apply(MapElements.via(new SimpleFunction<Long, Integer>() {
@Override
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
index 8e1632f..e11dd74 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
@@ -75,8 +75,8 @@ import org.apache.beam.sdk.coders.KvCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.beam.sdk.io.BoundedSource;
-import org.apache.beam.sdk.io.CountingInput;
import org.apache.beam.sdk.io.CountingSource;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers.JsonSchemaToTableSchema;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
@@ -1040,7 +1040,7 @@ public class BigQueryIOTest implements Serializable {
PCollection<TableRow> tableRows;
if (unbounded) {
tableRows =
- p.apply(CountingInput.unbounded())
+ p.apply(GenerateSequence.from(0))
.apply(
MapElements.via(
new SimpleFunction<Long, TableRow>() {
@@ -1090,7 +1090,7 @@ public class BigQueryIOTest implements Serializable {
tableRef.setTableId("sometable");
PCollection<TableRow> tableRows =
- p.apply(CountingInput.unbounded())
+ p.apply(GenerateSequence.from(0))
.apply(
MapElements.via(
new SimpleFunction<Long, TableRow>() {
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableWriteIT.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableWriteIT.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableWriteIT.java
index 240fb31..4ba77c9 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableWriteIT.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableWriteIT.java
@@ -43,7 +43,7 @@ import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.options.GcpOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.testing.TestPipeline;
@@ -106,7 +106,7 @@ public class BigtableWriteIT implements Serializable {
createEmptyTable(instanceName, tableId);
Pipeline p = Pipeline.create(options);
- p.apply(CountingInput.upTo(numRows))
+ p.apply(GenerateSequence.fromTo(0, numRows))
.apply(ParDo.of(new DoFn<Long, KV<ByteString, Iterable<Mutation>>>() {
@ProcessElement
public void processElement(ProcessContext c) {
http://git-wip-us.apache.org/repos/asf/beam/blob/6a9a24c0/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/datastore/V1WriteIT.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/datastore/V1WriteIT.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/datastore/V1WriteIT.java
index 0fcd9d3..322aecd 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/datastore/V1WriteIT.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/datastore/V1WriteIT.java
@@ -24,7 +24,7 @@ import static org.junit.Assert.assertEquals;
import java.util.UUID;
import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.gcp.datastore.V1TestUtil.CreateEntityFn;
import org.apache.beam.sdk.options.GcpOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
@@ -66,7 +66,7 @@ public class V1WriteIT {
Pipeline p = Pipeline.create(options);
// Write to datastore
- p.apply(CountingInput.upTo(numEntities))
+ p.apply(GenerateSequence.fromTo(0, numEntities))
.apply(ParDo.of(new CreateEntityFn(
options.getKind(), options.getNamespace(), ancestor)))
.apply(DatastoreIO.v1().write().withProjectId(project));