diff --git a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Dataflow_V2_Streaming.json b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Dataflow_V2_Streaming.json index 7dab8be7160a..c4edaa85a89d 100644 --- a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Dataflow_V2_Streaming.json +++ b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Dataflow_V2_Streaming.json @@ -1,7 +1,3 @@ { - "https://github.com/apache/beam/pull/34902": "Introducing OutputBuilder", - "comment": "Modify this file in a trivial way to cause this test suite to run", - "https://github.com/apache/beam/pull/31156": "noting that PR #31156 should run this test", - "https://github.com/apache/beam/pull/31268": "noting that PR #31268 should run this test", - "https://github.com/apache/beam/pull/31490": "noting that PR #31490 should run this test" + "comment": "Modify this file in a trivial way to cause this test suite to run" } diff --git a/runners/google-cloud-dataflow-java/build.gradle b/runners/google-cloud-dataflow-java/build.gradle index bafba869b3c1..17594ded5ed7 100644 --- a/runners/google-cloud-dataflow-java/build.gradle +++ b/runners/google-cloud-dataflow-java/build.gradle @@ -276,7 +276,7 @@ def createRunnerV2ValidatesRunnerTest = { Map args -> testClassesDirs = files(project(":sdks:java:core").sourceSets.test.output.classesDirs) + files(project(project.path).sourceSets.test.output.classesDirs) useJUnit { - includeCategories 'org.apache.beam.sdk.testing.ValidatesRunner' + includeCategories 'org.apache.beam.sdk.testing.RerunTest' commonRunnerV2ExcludeCategories.each { excludeCategories it } @@ -440,6 +440,7 @@ task validatesRunnerStreaming { name: 'validatesRunnerLegacyWorkerTestStreaming', pipelineOptions: legacyPipelineOptions + ['--streaming'], excludedCategories: [ + 'org.apache.beam.sdk.testing.BatchOnly', 'org.apache.beam.sdk.testing.UsesCommittedMetrics', 'org.apache.beam.sdk.testing.UsesMapState', 'org.apache.beam.sdk.testing.UsesRequiresTimeSortedInput', @@ -550,6 +551,7 @@ task validatesRunnerV2Streaming { name: 'validatesRunnerV2TestStreaming', pipelineOptions: runnerV2PipelineOptions + ['--streaming', '--experiments=enable_streaming_engine'], excludedCategories: [ + 'org.apache.beam.sdk.testing.BatchOnly', 'org.apache.beam.sdk.testing.LargeKeys$Above10KB', 'org.apache.beam.sdk.testing.UsesBoundedSplittableParDo', 'org.apache.beam.sdk.testing.UsesCommittedMetrics', diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/BatchOnly.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/BatchOnly.java new file mode 100644 index 000000000000..80f85722e123 --- /dev/null +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/BatchOnly.java @@ -0,0 +1,29 @@ +/* + * 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.sdk.testing; + +import org.apache.beam.sdk.annotations.Internal; + +/** + * Category tag for validation tests run on batch mode only. + * + *

Used to dedup tests from streaming test suites as long as its batch counterpart running the + * same test is considered sufficient. + */ +@Internal +public interface BatchOnly {} diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/RerunTest.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/RerunTest.java new file mode 100644 index 000000000000..5e78e61e8a0e --- /dev/null +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/RerunTest.java @@ -0,0 +1,29 @@ +/* + * 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.sdk.testing; + +import org.apache.beam.sdk.annotations.Internal; + +/** + * Category tag for validation tests run on batch mode only. + * + *

Used to dedup tests from streaming test suites as long as its batch counterpart running the + * same test is considered sufficient. + */ +@Internal +public interface RerunTest {} diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CombineTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CombineTest.java index 993b84a528d7..8ee23947c31d 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CombineTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CombineTest.java @@ -53,6 +53,7 @@ import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.coders.VarIntCoder; import org.apache.beam.sdk.coders.VoidCoder; +import org.apache.beam.sdk.testing.BatchOnly; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -665,13 +666,13 @@ public void testSimpleCombine() { } @Test - @Category({ValidatesRunner.class, UsesSideInputs.class}) + @Category({ValidatesRunner.class, UsesSideInputs.class, BatchOnly.class}) public void testSimpleCombineEmpty() { runTestSimpleCombine(EMPTY_TABLE, 0, Collections.emptyList()); } @Test - @Category({ValidatesRunner.class, UsesSideInputs.class}) + @Category({ValidatesRunner.class, UsesSideInputs.class, BatchOnly.class}) public void testBasicCombine() { runTestBasicCombine( Arrays.asList(KV.of("a", 1), KV.of("a", 1), KV.of("a", 4), KV.of("b", 1), KV.of("b", 13)), @@ -682,7 +683,7 @@ public void testBasicCombine() { } @Test - @Category({ValidatesRunner.class, UsesSideInputs.class}) + @Category({ValidatesRunner.class, UsesSideInputs.class, BatchOnly.class}) public void testBasicCombineEmpty() { runTestBasicCombine(EMPTY_TABLE, ImmutableSet.of(), Collections.emptyList()); } @@ -702,7 +703,7 @@ public void testCombinerNames() { } @Test - @Category({ValidatesRunner.class}) + @Category({ValidatesRunner.class, BatchOnly.class}) public void testHotKeyCombining() { PCollection> input = copy( @@ -839,7 +840,7 @@ public void populateDisplayData(DisplayData.Builder builder) { } @Test - @Category(ValidatesRunner.class) + @Category({ValidatesRunner.class, BatchOnly.class}) public void testCombinePerKeyPrimitiveDisplayData() { DisplayDataEvaluator evaluator = DisplayDataEvaluator.create(); @@ -909,7 +910,7 @@ public void testCombinePerKeyLambda() { /** Tests creation of a per-key binary {@link Combine} via a Java 8 lambda. */ @Test - @Category(ValidatesRunner.class) + @Category({ValidatesRunner.class, BatchOnly.class}) public void testBinaryCombinePerKeyLambda() { PCollection> output = @@ -923,7 +924,7 @@ public void testBinaryCombinePerKeyLambda() { /** Tests creation of a per-key {@link Combine} via a Java 8 method reference. */ @Test - @Category(ValidatesRunner.class) + @Category({ValidatesRunner.class, BatchOnly.class}) public void testCombinePerKeyInstanceMethodReference() { PCollection> output = @@ -937,7 +938,7 @@ public void testCombinePerKeyInstanceMethodReference() { /** Tests creation of a per-key binary {@link Combine} via a Java 8 method reference. */ @Test - @Category(ValidatesRunner.class) + @Category({ValidatesRunner.class, BatchOnly.class}) public void testBinaryCombinePerKeyInstanceMethodReference() { PCollection> output = @@ -1001,7 +1002,7 @@ public void testSimpleCombineWithContext() { } @Test - @Category({ValidatesRunner.class, UsesSideInputs.class}) + @Category({ValidatesRunner.class, UsesSideInputs.class, BatchOnly.class}) public void testSimpleCombineWithContextEmpty() { runTestSimpleCombineWithContext(EMPTY_TABLE, 0, Collections.emptyList(), new String[] {}); } @@ -1031,7 +1032,7 @@ public void testWithFanoutPreservesSideInputs() { } @Test - @Category({ValidatesRunner.class, UsesSideInputs.class}) + @Category({ValidatesRunner.class, UsesSideInputs.class, BatchOnly.class}) public void testHotKeyCombineWithSideInputs() { PCollection> input = createInput( diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CreateTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CreateTest.java index 03f4f1c67c67..90ea453ab4e3 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CreateTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CreateTest.java @@ -52,6 +52,7 @@ import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider; import org.apache.beam.sdk.schemas.Schema; import org.apache.beam.sdk.schemas.SchemaCoder; +import org.apache.beam.sdk.testing.BatchOnly; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.SourceTestUtils; @@ -101,7 +102,7 @@ public void testCreate() { } @Test - @Category(ValidatesRunner.class) + @Category({ValidatesRunner.class, BatchOnly.class}) public void testCreateEmpty() { PCollection output = p.apply(Create.empty(StringUtf8Coder.of())); @@ -125,7 +126,7 @@ public void testCreateEmptyIterableRequiresCoder() { } @Test - @Category(NeedsRunner.class) + @Category({NeedsRunner.class, BatchOnly.class}) public void testCreateEmptyIterableWithCoder() { PCollection output = p.apply(Create.of(Collections.emptyList()).withCoder(VoidCoder.of())); @@ -161,7 +162,7 @@ public void testPolymorphicType() throws Exception { } @Test - @Category(ValidatesRunner.class) + @Category({ValidatesRunner.class, BatchOnly.class}) public void testCreateWithNullsAndValues() throws Exception { PCollection output = p.apply( diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MapViewTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MapViewTest.java new file mode 100644 index 000000000000..0fd7f269a9a7 --- /dev/null +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MapViewTest.java @@ -0,0 +1,1017 @@ +/* + * 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.sdk.transforms; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.Serializable; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Map; +import java.util.Map.Entry; +import org.apache.beam.sdk.coders.AtomicCoder; +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.coders.CoderException; +import org.apache.beam.sdk.coders.KvCoder; +import org.apache.beam.sdk.coders.NullableCoder; +import org.apache.beam.sdk.coders.StringUtf8Coder; +import org.apache.beam.sdk.coders.VarIntCoder; +import org.apache.beam.sdk.testing.NeedsRunner; +import org.apache.beam.sdk.testing.PAssert; +import org.apache.beam.sdk.testing.RerunTest; +import org.apache.beam.sdk.testing.TestPipeline; +import org.apache.beam.sdk.testing.UsesSideInputs; +import org.apache.beam.sdk.testing.ValidatesRunner; +import org.apache.beam.sdk.transforms.windowing.FixedWindows; +import org.apache.beam.sdk.transforms.windowing.Window; +import org.apache.beam.sdk.values.KV; +import org.apache.beam.sdk.values.PCollection; +import org.apache.beam.sdk.values.PCollectionView; +import org.apache.beam.sdk.values.TimestampedValue; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; +import org.joda.time.Duration; +import org.joda.time.Instant; +import org.junit.Rule; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.rules.ExpectedException; +import org.junit.rules.Timeout; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** + * Tests for (multi)map {@link View}. See also {@link ParDoTest}, which provides additional coverage + * since views can only be observed via {@link ParDo}. + */ +@RunWith(JUnit4.class) +@Category({UsesSideInputs.class, RerunTest.class}) +public class MapViewTest implements Serializable { + // This test is Serializable, just so that it's easy to have + // anonymous inner classes inside the non-static test methods. + + @Rule public final transient TestPipeline pipeline = TestPipeline.create(); + + @Rule public transient ExpectedException thrown = ExpectedException.none(); + + @Rule public transient Timeout globalTimeout = Timeout.seconds(1200); + + @Test + @Category(ValidatesRunner.class) + public void testMultimapSideInput() { + + final PCollectionView>> view = + pipeline + .apply( + "CreateSideInput", + Create.of(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3))) + .apply(View.asMultimap()); + + PCollection> output = + pipeline + .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { + c.output(KV.of(c.element(), v)); + } + } + }) + .withSideInputs(view)); + + PAssert.that(output) + .containsInAnyOrder( + KV.of("apple", 1), + KV.of("apple", 1), + KV.of("apple", 2), + KV.of("banana", 3), + KV.of("blackberry", 3)); + + pipeline.run(); + } + + @Test + @Category(ValidatesRunner.class) + public void testMultimapAsEntrySetSideInput() { + + final PCollectionView>> view = + pipeline + .apply( + "CreateSideInput", + Create.of(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3))) + .apply(View.asMultimap()); + + PCollection> output = + pipeline + .apply("CreateMainInput", Create.of(2 /* size */)) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + assertEquals((int) c.element(), c.sideInput(view).size()); + assertEquals((int) c.element(), c.sideInput(view).entrySet().size()); + for (Entry> entry : + c.sideInput(view).entrySet()) { + for (Integer value : entry.getValue()) { + c.output(KV.of(entry.getKey(), value)); + } + } + } + }) + .withSideInputs(view)); + + PAssert.that(output) + .containsInAnyOrder(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3)); + + pipeline.run(); + } + + @Test + @Category(NeedsRunner.class) + public void testMultimapInMemorySideInput() { + + final PCollectionView>> view = + pipeline + .apply( + "CreateSideInput", + Create.of(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3))) + .apply(View.asMultimap().inMemory()); + + PCollection> output = + pipeline + .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { + c.output(KV.of(c.element(), v)); + } + } + }) + .withSideInputs(view)); + + PAssert.that(output) + .containsInAnyOrder( + KV.of("apple", 1), + KV.of("apple", 1), + KV.of("apple", 2), + KV.of("banana", 3), + KV.of("blackberry", 3)); + + pipeline.run(); + } + + private static class NonDeterministicStringCoder extends AtomicCoder { + @Override + public void encode(String value, OutputStream outStream) throws CoderException, IOException { + encode(value, outStream, Coder.Context.NESTED); + } + + @Override + public void encode(String value, OutputStream outStream, Coder.Context context) + throws CoderException, IOException { + StringUtf8Coder.of().encode(value, outStream, context); + } + + @Override + public String decode(InputStream inStream) throws CoderException, IOException { + return decode(inStream, Coder.Context.NESTED); + } + + @Override + public String decode(InputStream inStream, Coder.Context context) + throws CoderException, IOException { + return StringUtf8Coder.of().decode(inStream, context); + } + + @Override + public void verifyDeterministic() + throws org.apache.beam.sdk.coders.Coder.NonDeterministicException { + throw new NonDeterministicException(this, "Test coder is not deterministic on purpose."); + } + } + + @Test + @Category({ValidatesRunner.class}) + public void testMultimapSideInputWithNonDeterministicKeyCoder() { + + final PCollectionView>> view = + pipeline + .apply( + "CreateSideInput", + Create.of(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3)) + .withCoder(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) + .apply(View.asMultimap()); + + PCollection> output = + pipeline + .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { + c.output(KV.of(c.element(), v)); + } + } + }) + .withSideInputs(view)); + + PAssert.that(output) + .containsInAnyOrder( + KV.of("apple", 1), + KV.of("apple", 1), + KV.of("apple", 2), + KV.of("banana", 3), + KV.of("blackberry", 3)); + + pipeline.run(); + } + + @Test + @Category(ValidatesRunner.class) + public void testWindowedMultimapSideInput() { + + final PCollectionView>> view = + pipeline + .apply( + "CreateSideInput", + Create.timestamped( + TimestampedValue.of(KV.of("a", 1), new Instant(1)), + TimestampedValue.of(KV.of("a", 1), new Instant(2)), + TimestampedValue.of(KV.of("a", 2), new Instant(7)), + TimestampedValue.of(KV.of("b", 3), new Instant(14)))) + .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) + .apply(View.asMultimap()); + + PCollection> output = + pipeline + .apply( + "CreateMainInput", + Create.timestamped( + TimestampedValue.of("apple", new Instant(5)), + TimestampedValue.of("banana", new Instant(13)), + TimestampedValue.of("blackberry", new Instant(16)))) + .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { + c.output(KV.of(c.element(), v)); + } + } + }) + .withSideInputs(view)); + + PAssert.that(output) + .containsInAnyOrder( + KV.of("apple", 1), + KV.of("apple", 1), + KV.of("apple", 2), + KV.of("banana", 3), + KV.of("blackberry", 3)); + + pipeline.run(); + } + + @Test + @Category(ValidatesRunner.class) + public void testWindowedMultimapAsEntrySetSideInput() { + + final PCollectionView>> view = + pipeline + .apply( + "CreateSideInput", + Create.timestamped( + TimestampedValue.of(KV.of("a", 1), new Instant(1)), + TimestampedValue.of(KV.of("a", 1), new Instant(2)), + TimestampedValue.of(KV.of("a", 2), new Instant(7)), + TimestampedValue.of(KV.of("b", 3), new Instant(14)))) + .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) + .apply(View.asMultimap()); + + PCollection> output = + pipeline + .apply( + "CreateMainInput", + Create.timestamped( + TimestampedValue.of(1 /* size */, new Instant(5)), + TimestampedValue.of(1 /* size */, new Instant(16)))) + .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + assertEquals((int) c.element(), c.sideInput(view).size()); + assertEquals((int) c.element(), c.sideInput(view).entrySet().size()); + for (Entry> entry : + c.sideInput(view).entrySet()) { + for (Integer value : entry.getValue()) { + c.output(KV.of(entry.getKey(), value)); + } + } + } + }) + .withSideInputs(view)); + + PAssert.that(output) + .containsInAnyOrder(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3)); + + pipeline.run(); + } + + @Test + @Category({ValidatesRunner.class}) + public void testWindowedMultimapSideInputWithNonDeterministicKeyCoder() { + + final PCollectionView>> view = + pipeline + .apply( + "CreateSideInput", + Create.timestamped( + TimestampedValue.of(KV.of("a", 1), new Instant(1)), + TimestampedValue.of(KV.of("a", 1), new Instant(2)), + TimestampedValue.of(KV.of("a", 2), new Instant(7)), + TimestampedValue.of(KV.of("b", 3), new Instant(14))) + .withCoder(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) + .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) + .apply(View.asMultimap()); + + PCollection> output = + pipeline + .apply( + "CreateMainInput", + Create.timestamped( + TimestampedValue.of("apple", new Instant(5)), + TimestampedValue.of("banana", new Instant(13)), + TimestampedValue.of("blackberry", new Instant(16)))) + .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { + c.output(KV.of(c.element(), v)); + } + } + }) + .withSideInputs(view)); + + PAssert.that(output) + .containsInAnyOrder( + KV.of("apple", 1), + KV.of("apple", 1), + KV.of("apple", 2), + KV.of("banana", 3), + KV.of("blackberry", 3)); + + pipeline.run(); + } + + @Test + @Category(ValidatesRunner.class) + public void testEmptyMultimapSideInput() throws Exception { + + final PCollectionView>> view = + pipeline + .apply( + "CreateEmptyView", Create.empty(KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of()))) + .apply(View.asMultimap()); + + PCollection results = + pipeline + .apply("Create1", Create.of(1)) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn() { + @ProcessElement + public void processElement(ProcessContext c) { + assertTrue(c.sideInput(view).isEmpty()); + assertTrue(c.sideInput(view).entrySet().isEmpty()); + assertFalse(c.sideInput(view).entrySet().iterator().hasNext()); + c.output(c.element()); + } + }) + .withSideInputs(view)); + + // Pass at least one value through to guarantee that DoFn executes. + PAssert.that(results).containsInAnyOrder(1); + + pipeline.run(); + } + + @Test + @Category({ValidatesRunner.class}) + public void testEmptyMultimapSideInputWithNonDeterministicKeyCoder() throws Exception { + + final PCollectionView>> view = + pipeline + .apply( + "CreateEmptyView", + Create.empty(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) + .apply(View.asMultimap()); + + PCollection results = + pipeline + .apply("Create1", Create.of(1)) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn() { + @ProcessElement + public void processElement(ProcessContext c) { + assertTrue(c.sideInput(view).isEmpty()); + assertTrue(c.sideInput(view).entrySet().isEmpty()); + assertFalse(c.sideInput(view).entrySet().iterator().hasNext()); + c.output(c.element()); + } + }) + .withSideInputs(view)); + + // Pass at least one value through to guarantee that DoFn executes. + PAssert.that(results).containsInAnyOrder(1); + + pipeline.run(); + } + + @Test + @Category(ValidatesRunner.class) + public void testMultimapSideInputIsImmutable() { + + final PCollectionView>> view = + pipeline.apply("CreateSideInput", Create.of(KV.of("a", 1))).apply(View.asMultimap()); + + PCollection> output = + pipeline + .apply("CreateMainInput", Create.of("apple")) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + try { + c.sideInput(view).clear(); + fail("Expected UnsupportedOperationException on clear()"); + } catch (UnsupportedOperationException expected) { + } + try { + c.sideInput(view).put("c", ImmutableList.of(3)); + fail("Expected UnsupportedOperationException on put()"); + } catch (UnsupportedOperationException expected) { + } + try { + c.sideInput(view).remove("c"); + fail("Expected UnsupportedOperationException on remove()"); + } catch (UnsupportedOperationException expected) { + } + try { + c.sideInput(view).putAll(new HashMap<>()); + fail("Expected UnsupportedOperationException on putAll()"); + } catch (UnsupportedOperationException expected) { + } + for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { + c.output(KV.of(c.element(), v)); + } + } + }) + .withSideInputs(view)); + + // Pass at least one value through to guarantee that DoFn executes. + PAssert.that(output).containsInAnyOrder(KV.of("apple", 1)); + + pipeline.run(); + } + + @Test + @Category(ValidatesRunner.class) + public void testMapSideInput() { + + final PCollectionView> view = + pipeline + .apply("CreateSideInput", Create.of(KV.of("a", 1), KV.of("b", 3))) + .apply(View.asMap()); + + PCollection> output = + pipeline + .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + c.output( + KV.of( + c.element(), + c.sideInput(view).get(c.element().substring(0, 1)))); + } + }) + .withSideInputs(view)); + + PAssert.that(output) + .containsInAnyOrder(KV.of("apple", 1), KV.of("banana", 3), KV.of("blackberry", 3)); + + pipeline.run(); + } + + @Test + @Category(ValidatesRunner.class) + public void testMapAsEntrySetSideInput() { + + final PCollectionView> view = + pipeline + .apply("CreateSideInput", Create.of(KV.of("a", 1), KV.of("b", 3))) + .apply(View.asMap()); + + PCollection> output = + pipeline + .apply("CreateMainInput", Create.of(2 /* size */)) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + assertEquals((int) c.element(), c.sideInput(view).size()); + assertEquals((int) c.element(), c.sideInput(view).entrySet().size()); + for (Entry entry : c.sideInput(view).entrySet()) { + c.output(KV.of(entry.getKey(), entry.getValue())); + } + } + }) + .withSideInputs(view)); + + PAssert.that(output).containsInAnyOrder(KV.of("a", 1), KV.of("b", 3)); + + pipeline.run(); + } + + @Test + @Category(NeedsRunner.class) + public void testMapInMemorySideInput() { + + final PCollectionView> view = + pipeline + .apply("CreateSideInput", Create.of(KV.of("a", 1), KV.of("b", 3))) + .apply(View.asMap().inMemory()); + + PCollection> output = + pipeline + .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + c.output( + KV.of( + c.element(), + c.sideInput(view).get(c.element().substring(0, 1)))); + } + }) + .withSideInputs(view)); + + PAssert.that(output) + .containsInAnyOrder(KV.of("apple", 1), KV.of("banana", 3), KV.of("blackberry", 3)); + + pipeline.run(); + } + + @Test + @Category(NeedsRunner.class) + public void testMapInMemorySideInputWithNonStructuralKey() { + + final PCollectionView> view = + pipeline + .apply( + "CreateSideInput", + Create.of( + KV.of("a".getBytes(StandardCharsets.UTF_8), 1), + KV.of("b".getBytes(StandardCharsets.UTF_8), 3))) + .apply(View.asMap().inMemory()); + + PCollection> output = + pipeline + .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + c.output( + KV.of( + c.element(), + c.sideInput(view) + .get( + c.element() + .substring(0, 1) + .getBytes(StandardCharsets.UTF_8)))); + } + }) + .withSideInputs(view)); + + PAssert.that(output) + .containsInAnyOrder(KV.of("apple", 1), KV.of("banana", 3), KV.of("blackberry", 3)); + + pipeline.run(); + } + + @Test + @Category({ValidatesRunner.class}) + public void testMapSideInputWithNonDeterministicKeyCoder() { + + final PCollectionView> view = + pipeline + .apply( + "CreateSideInput", + Create.of(KV.of("a", 1), KV.of("b", 3)) + .withCoder(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) + .apply(View.asMap()); + + PCollection> output = + pipeline + .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + c.output( + KV.of( + c.element(), + c.sideInput(view).get(c.element().substring(0, 1)))); + } + }) + .withSideInputs(view)); + + PAssert.that(output) + .containsInAnyOrder(KV.of("apple", 1), KV.of("banana", 3), KV.of("blackberry", 3)); + + pipeline.run(); + } + + @Test + @Category(ValidatesRunner.class) + public void testWindowedMapSideInput() { + + final PCollectionView> view = + pipeline + .apply( + "CreateSideInput", + Create.timestamped( + TimestampedValue.of(KV.of("a", 1), new Instant(1)), + TimestampedValue.of(KV.of("b", 2), new Instant(4)), + TimestampedValue.of(KV.of("b", 3), new Instant(18)))) + .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) + .apply(View.asMap()); + + PCollection> output = + pipeline + .apply( + "CreateMainInput", + Create.timestamped( + TimestampedValue.of("apple", new Instant(5)), + TimestampedValue.of("banana", new Instant(4)), + TimestampedValue.of("blackberry", new Instant(16)))) + .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + c.output( + KV.of( + c.element(), + c.sideInput(view).get(c.element().substring(0, 1)))); + } + }) + .withSideInputs(view)); + + PAssert.that(output) + .containsInAnyOrder(KV.of("apple", 1), KV.of("banana", 2), KV.of("blackberry", 3)); + + pipeline.run(); + } + + @Test + @Category(ValidatesRunner.class) + public void testWindowedMapAsEntrySetSideInput() { + + final PCollectionView> view = + pipeline + .apply( + "CreateSideInput", + Create.timestamped( + TimestampedValue.of(KV.of("a", 1), new Instant(1)), + TimestampedValue.of(KV.of("b", 2), new Instant(4)), + TimestampedValue.of(KV.of("b", 3), new Instant(18)))) + .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) + .apply(View.asMap()); + + PCollection> output = + pipeline + .apply( + "CreateMainInput", + Create.timestamped( + TimestampedValue.of(2 /* size */, new Instant(5)), + TimestampedValue.of(1 /* size */, new Instant(16)))) + .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + assertEquals((int) c.element(), c.sideInput(view).size()); + assertEquals((int) c.element(), c.sideInput(view).entrySet().size()); + for (Entry entry : c.sideInput(view).entrySet()) { + c.output(KV.of(entry.getKey(), entry.getValue())); + } + } + }) + .withSideInputs(view)); + + PAssert.that(output).containsInAnyOrder(KV.of("a", 1), KV.of("b", 2), KV.of("b", 3)); + + pipeline.run(); + } + + @Test + @Category({ValidatesRunner.class}) + public void testWindowedMapSideInputWithNonDeterministicKeyCoder() { + + final PCollectionView> view = + pipeline + .apply( + "CreateSideInput", + Create.timestamped( + TimestampedValue.of(KV.of("a", 1), new Instant(1)), + TimestampedValue.of(KV.of("b", 2), new Instant(4)), + TimestampedValue.of(KV.of("b", 3), new Instant(18))) + .withCoder(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) + .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) + .apply(View.asMap()); + + PCollection> output = + pipeline + .apply( + "CreateMainInput", + Create.timestamped( + TimestampedValue.of("apple", new Instant(5)), + TimestampedValue.of("banana", new Instant(4)), + TimestampedValue.of("blackberry", new Instant(16)))) + .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + c.output( + KV.of( + c.element(), + c.sideInput(view).get(c.element().substring(0, 1)))); + } + }) + .withSideInputs(view)); + + PAssert.that(output) + .containsInAnyOrder(KV.of("apple", 1), KV.of("banana", 2), KV.of("blackberry", 3)); + + pipeline.run(); + } + + @Test + @Category(ValidatesRunner.class) + public void testEmptyMapSideInput() throws Exception { + + final PCollectionView> view = + pipeline + .apply( + "CreateEmptyView", Create.empty(KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of()))) + .apply(View.asMap()); + + PCollection results = + pipeline + .apply("Create1", Create.of(1)) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn() { + @ProcessElement + public void processElement(ProcessContext c) { + assertTrue(c.sideInput(view).isEmpty()); + assertTrue(c.sideInput(view).entrySet().isEmpty()); + assertFalse(c.sideInput(view).entrySet().iterator().hasNext()); + c.output(c.element()); + } + }) + .withSideInputs(view)); + + // Pass at least one value through to guarantee that DoFn executes. + PAssert.that(results).containsInAnyOrder(1); + + pipeline.run(); + } + + @Test + @Category({ValidatesRunner.class}) + public void testEmptyMapSideInputWithNonDeterministicKeyCoder() throws Exception { + + final PCollectionView> view = + pipeline + .apply( + "CreateEmptyView", + Create.empty(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) + .apply(View.asMap()); + + PCollection results = + pipeline + .apply("Create1", Create.of(1)) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn() { + @ProcessElement + public void processElement(ProcessContext c) { + assertTrue(c.sideInput(view).isEmpty()); + assertTrue(c.sideInput(view).entrySet().isEmpty()); + assertFalse(c.sideInput(view).entrySet().iterator().hasNext()); + c.output(c.element()); + } + }) + .withSideInputs(view)); + + // Pass at least one value through to guarantee that DoFn executes. + PAssert.that(results).containsInAnyOrder(1); + + pipeline.run(); + } + + @Test + @Category(ValidatesRunner.class) + public void testMapSideInputWithNullValuesCatchesDuplicates() { + + final PCollectionView> view = + pipeline + .apply( + "CreateSideInput", + Create.of(KV.of("a", (Integer) null), KV.of("a", (Integer) null)) + .withCoder( + KvCoder.of(StringUtf8Coder.of(), NullableCoder.of(VarIntCoder.of())))) + .apply(View.asMap()); + + PCollection> output = + pipeline + .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + c.output( + KV.of( + c.element(), + c.sideInput(view) + .getOrDefault(c.element().substring(0, 1), 0))); + } + }) + .withSideInputs(view)); + + PAssert.that(output) + .containsInAnyOrder(KV.of("apple", 1), KV.of("banana", 3), KV.of("blackberry", 3)); + + // As long as we get an error, be flexible with how a runner surfaces it + thrown.expect(Exception.class); + + pipeline.run(); + } + + @Test + @Category(ValidatesRunner.class) + public void testMapSideInputIsImmutable() { + + final PCollectionView> view = + pipeline.apply("CreateSideInput", Create.of(KV.of("a", 1))).apply(View.asMap()); + + PCollection> output = + pipeline + .apply("CreateMainInput", Create.of("apple")) + .apply( + "OutputSideInputs", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + try { + c.sideInput(view).clear(); + fail("Expected UnsupportedOperationException on clear()"); + } catch (UnsupportedOperationException expected) { + } + try { + c.sideInput(view).put("c", 3); + fail("Expected UnsupportedOperationException on put()"); + } catch (UnsupportedOperationException expected) { + } + try { + c.sideInput(view).remove("c"); + fail("Expected UnsupportedOperationException on remove()"); + } catch (UnsupportedOperationException expected) { + } + try { + c.sideInput(view).putAll(new HashMap<>()); + fail("Expected UnsupportedOperationException on putAll()"); + } catch (UnsupportedOperationException expected) { + } + c.output( + KV.of( + c.element(), + c.sideInput(view).get(c.element().substring(0, 1)))); + } + }) + .withSideInputs(view)); + + // Pass at least one value through to guarantee that DoFn executes. + PAssert.that(output).containsInAnyOrder(KV.of("apple", 1)); + + pipeline.run(); + } + + @Test + @Category(ValidatesRunner.class) + public void testCombinedMapSideInput() { + + final PCollectionView> view = + pipeline + .apply("CreateSideInput", Create.of(KV.of("a", 1), KV.of("a", 20), KV.of("b", 3))) + .apply("SumIntegers", Combine.perKey(Sum.ofIntegers())) + .apply(View.asMap()); + + PCollection> output = + pipeline + .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) + .apply( + "Output", + ParDo.of( + new DoFn>() { + @ProcessElement + public void processElement(ProcessContext c) { + c.output( + KV.of( + c.element(), + c.sideInput(view).get(c.element().substring(0, 1)))); + } + }) + .withSideInputs(view)); + + PAssert.that(output) + .containsInAnyOrder(KV.of("apple", 21), KV.of("banana", 3), KV.of("blackberry", 3)); + + pipeline.run(); + } +} 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 8409133772eb..71428949a7fd 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 @@ -98,6 +98,7 @@ import org.apache.beam.sdk.state.TimerSpec; import org.apache.beam.sdk.state.TimerSpecs; import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.testing.BatchOnly; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -405,7 +406,7 @@ public void testParDo() { } @Test - @Category(ValidatesRunner.class) + @Category({ValidatesRunner.class, BatchOnly.class}) public void testParDoEmpty() { List inputs = Arrays.asList(); @@ -421,7 +422,7 @@ public void testParDoEmpty() { } @Test - @Category(ValidatesRunner.class) + @Category({ValidatesRunner.class, BatchOnly.class}) public void testParDoEmptyOutputs() { List inputs = Arrays.asList(); @@ -442,7 +443,7 @@ public void testParDoTransformNameBasedDoFnWithTrimmedSuffix() { } @Test - @Category(ValidatesRunner.class) + @Category({ValidatesRunner.class, BatchOnly.class}) public void testParDoInCustomTransform() { List inputs = Arrays.asList(3, -42, 666); @@ -629,7 +630,7 @@ public void onTimer(BoundedWindow w) {} } @Test - @Category(ValidatesRunner.class) + @Category({ValidatesRunner.class, BatchOnly.class}) public void testPipelineOptionsParameter() { PCollection results = pipeline @@ -728,7 +729,7 @@ public void testParDoWithTaggedOutput() { } @Test - @Category(ValidatesRunner.class) + @Category({ValidatesRunner.class, BatchOnly.class}) public void testParDoEmptyWithTaggedOutput() { TupleTag mainOutputTag = new TupleTag("main") {}; TupleTag additionalOutputTag1 = new TupleTag("additional1") {}; @@ -770,7 +771,7 @@ public void testParDoEmptyWithTaggedOutput() { } @Test - @Category(ValidatesRunner.class) + @Category({ValidatesRunner.class, BatchOnly.class}) public void testParDoWithEmptyTaggedOutput() { TupleTag mainOutputTag = new TupleTag("main") {}; TupleTag additionalOutputTag1 = new TupleTag("additional1") {}; @@ -794,7 +795,7 @@ public void testParDoWithEmptyTaggedOutput() { } @Test - @Category(ValidatesRunner.class) + @Category({ValidatesRunner.class, BatchOnly.class}) public void testParDoWithOnlyTaggedOutput() { List inputs = Arrays.asList(3, -42, 666); @@ -823,7 +824,7 @@ public void processElement( } @Test - @Category(NeedsRunner.class) + @Category({NeedsRunner.class, BatchOnly.class}) public void testParDoWritingToUndeclaredTag() { List inputs = Arrays.asList(3, -42, 666); @@ -906,7 +907,7 @@ public void testSameSideInputReadTwice() { } @Test - @Category({NeedsRunner.class, UsesSideInputs.class}) + @Category({NeedsRunner.class, UsesSideInputs.class, BatchOnly.class}) public void testSideInputAnnotationFailedValidationMissing() { // SideInput tag id final String sideInputTag1 = "tag1"; @@ -923,7 +924,7 @@ public void processElement(@SideInput(sideInputTag1) String tag1) {} } @Test - @Category({NeedsRunner.class, UsesSideInputs.class}) + @Category({NeedsRunner.class, UsesSideInputs.class, BatchOnly.class}) public void testSideInputAnnotationFailedValidationSingletonType() { final PCollectionView sideInput1 = @@ -948,7 +949,7 @@ public void processElement(@SideInput(sideInputTag1) String tag1) {} } @Test - @Category({NeedsRunner.class, UsesSideInputs.class}) + @Category({NeedsRunner.class, UsesSideInputs.class, BatchOnly.class}) public void testSideInputAnnotationFailedValidationListType() { final PCollectionView> sideInput1 = @@ -973,7 +974,7 @@ public void processElement(@SideInput(sideInputTag1) List tag1) {} } @Test - @Category({NeedsRunner.class, UsesSideInputs.class}) + @Category({NeedsRunner.class, UsesSideInputs.class, BatchOnly.class}) public void testSideInputAnnotationFailedValidationIterableType() { final PCollectionView> sideInput1 = @@ -998,7 +999,7 @@ public void processElement(@SideInput(sideInputTag1) List tag1) {} } @Test - @Category({NeedsRunner.class, UsesSideInputs.class}) + @Category({NeedsRunner.class, UsesSideInputs.class, BatchOnly.class}) public void testSideInputAnnotationFailedValidationMapType() { final PCollectionView> sideInput1 = @@ -1048,7 +1049,7 @@ public void processElement(@SideInput(sideInputTag1) Map tag1) } @Test - @Category({ValidatesRunner.class, UsesSideInputs.class}) + @Category({ValidatesRunner.class, UsesSideInputs.class, BatchOnly.class}) public void testSideInputAnnotation() { final PCollectionView> sideInput1 = @@ -1184,7 +1185,7 @@ public void processElement( } @Test - @Category({ValidatesRunner.class, UsesSideInputs.class}) + @Category({ValidatesRunner.class, UsesSideInputs.class, BatchOnly.class}) public void testParDoWithSideInputsIsCumulative() { List inputs = Arrays.asList(3, -42, 666); @@ -1294,7 +1295,7 @@ public void testMultiOutputParDoWithSideInputsIsCumulative() { } @Test - @Category(NeedsRunner.class) + @Category({NeedsRunner.class, BatchOnly.class}) public void testParDoReadingFromUnknownSideInput() { List inputs = Arrays.asList(3, -42, 666); @@ -1506,7 +1507,7 @@ public void testTaggedOutputUnregisteredExplicitCoder() throws Exception { } @Test - @Category(NeedsRunner.class) + @Category({NeedsRunner.class, BatchOnly.class}) public void testMainOutputUnregisteredExplicitCoder() { PCollection input = pipeline.apply(Create.of(Arrays.asList(1, 2, 3))); @@ -1524,7 +1525,7 @@ public void testMainOutputUnregisteredExplicitCoder() { } @Test - @Category(NeedsRunner.class) + @Category({NeedsRunner.class, BatchOnly.class}) public void testMainOutputApplyTaggedOutputNoCoder() { // Regression test: applying a transform to the main output // should not cause a crash based on lack of a coder for the diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java index 2bdc9061e23c..36664d66ec4b 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java @@ -19,35 +19,22 @@ import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; import static org.hamcrest.MatcherAssert.assertThat; -import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; import java.io.Serializable; -import java.nio.charset.StandardCharsets; import java.util.ArrayList; -import java.util.HashMap; import java.util.Iterator; import java.util.List; -import java.util.Map; -import java.util.Map.Entry; -import org.apache.beam.sdk.Pipeline; -import org.apache.beam.sdk.coders.AtomicCoder; -import org.apache.beam.sdk.coders.Coder; -import org.apache.beam.sdk.coders.CoderException; import org.apache.beam.sdk.coders.KvCoder; -import org.apache.beam.sdk.coders.NullableCoder; -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.coders.VoidCoder; 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.RerunTest; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testing.TestStream; import org.apache.beam.sdk.testing.UsesSideInputs; @@ -61,19 +48,15 @@ import org.apache.beam.sdk.transforms.windowing.Repeatedly; import org.apache.beam.sdk.transforms.windowing.Window; import org.apache.beam.sdk.values.KV; -import org.apache.beam.sdk.values.PBegin; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PCollectionView; import org.apache.beam.sdk.values.TimestampedValue; -import org.apache.beam.sdk.values.WindowingStrategy; -import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; import org.hamcrest.Matchers; import org.joda.time.Duration; import org.joda.time.Instant; import org.junit.Rule; import org.junit.Test; import org.junit.experimental.categories.Category; -import org.junit.internal.matchers.ThrowableMessageMatcher; import org.junit.rules.ExpectedException; import org.junit.rules.Timeout; import org.junit.runner.RunWith; @@ -84,7 +67,7 @@ * views can only be observed via {@link ParDo}. */ @RunWith(JUnit4.class) -@Category(UsesSideInputs.class) +@Category({UsesSideInputs.class, RerunTest.class}) public class ViewTest implements Serializable { // This test is Serializable, just so that it's easy to have // anonymous inner classes inside the non-static test methods. @@ -693,945 +676,6 @@ public void processElement(ProcessContext c) { pipeline.run(); } - @Test - @Category(ValidatesRunner.class) - public void testMultimapSideInput() { - - final PCollectionView>> view = - pipeline - .apply( - "CreateSideInput", - Create.of(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3))) - .apply(View.asMultimap()); - - PCollection> output = - pipeline - .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { - c.output(KV.of(c.element(), v)); - } - } - }) - .withSideInputs(view)); - - PAssert.that(output) - .containsInAnyOrder( - KV.of("apple", 1), - KV.of("apple", 1), - KV.of("apple", 2), - KV.of("banana", 3), - KV.of("blackberry", 3)); - - pipeline.run(); - } - - @Test - @Category(ValidatesRunner.class) - public void testMultimapAsEntrySetSideInput() { - - final PCollectionView>> view = - pipeline - .apply( - "CreateSideInput", - Create.of(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3))) - .apply(View.asMultimap()); - - PCollection> output = - pipeline - .apply("CreateMainInput", Create.of(2 /* size */)) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - assertEquals((int) c.element(), c.sideInput(view).size()); - assertEquals((int) c.element(), c.sideInput(view).entrySet().size()); - for (Entry> entry : - c.sideInput(view).entrySet()) { - for (Integer value : entry.getValue()) { - c.output(KV.of(entry.getKey(), value)); - } - } - } - }) - .withSideInputs(view)); - - PAssert.that(output) - .containsInAnyOrder(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3)); - - pipeline.run(); - } - - @Test - @Category(NeedsRunner.class) - public void testMultimapInMemorySideInput() { - - final PCollectionView>> view = - pipeline - .apply( - "CreateSideInput", - Create.of(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3))) - .apply(View.asMultimap().inMemory()); - - PCollection> output = - pipeline - .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { - c.output(KV.of(c.element(), v)); - } - } - }) - .withSideInputs(view)); - - PAssert.that(output) - .containsInAnyOrder( - KV.of("apple", 1), - KV.of("apple", 1), - KV.of("apple", 2), - KV.of("banana", 3), - KV.of("blackberry", 3)); - - pipeline.run(); - } - - private static class NonDeterministicStringCoder extends AtomicCoder { - @Override - public void encode(String value, OutputStream outStream) throws CoderException, IOException { - encode(value, outStream, Coder.Context.NESTED); - } - - @Override - public void encode(String value, OutputStream outStream, Coder.Context context) - throws CoderException, IOException { - StringUtf8Coder.of().encode(value, outStream, context); - } - - @Override - public String decode(InputStream inStream) throws CoderException, IOException { - return decode(inStream, Coder.Context.NESTED); - } - - @Override - public String decode(InputStream inStream, Coder.Context context) - throws CoderException, IOException { - return StringUtf8Coder.of().decode(inStream, context); - } - - @Override - public void verifyDeterministic() - throws org.apache.beam.sdk.coders.Coder.NonDeterministicException { - throw new NonDeterministicException(this, "Test coder is not deterministic on purpose."); - } - } - - @Test - @Category({ValidatesRunner.class}) - public void testMultimapSideInputWithNonDeterministicKeyCoder() { - - final PCollectionView>> view = - pipeline - .apply( - "CreateSideInput", - Create.of(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3)) - .withCoder(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) - .apply(View.asMultimap()); - - PCollection> output = - pipeline - .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { - c.output(KV.of(c.element(), v)); - } - } - }) - .withSideInputs(view)); - - PAssert.that(output) - .containsInAnyOrder( - KV.of("apple", 1), - KV.of("apple", 1), - KV.of("apple", 2), - KV.of("banana", 3), - KV.of("blackberry", 3)); - - pipeline.run(); - } - - @Test - @Category(ValidatesRunner.class) - public void testWindowedMultimapSideInput() { - - final PCollectionView>> view = - pipeline - .apply( - "CreateSideInput", - Create.timestamped( - TimestampedValue.of(KV.of("a", 1), new Instant(1)), - TimestampedValue.of(KV.of("a", 1), new Instant(2)), - TimestampedValue.of(KV.of("a", 2), new Instant(7)), - TimestampedValue.of(KV.of("b", 3), new Instant(14)))) - .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) - .apply(View.asMultimap()); - - PCollection> output = - pipeline - .apply( - "CreateMainInput", - Create.timestamped( - TimestampedValue.of("apple", new Instant(5)), - TimestampedValue.of("banana", new Instant(13)), - TimestampedValue.of("blackberry", new Instant(16)))) - .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { - c.output(KV.of(c.element(), v)); - } - } - }) - .withSideInputs(view)); - - PAssert.that(output) - .containsInAnyOrder( - KV.of("apple", 1), - KV.of("apple", 1), - KV.of("apple", 2), - KV.of("banana", 3), - KV.of("blackberry", 3)); - - pipeline.run(); - } - - @Test - @Category(ValidatesRunner.class) - public void testWindowedMultimapAsEntrySetSideInput() { - - final PCollectionView>> view = - pipeline - .apply( - "CreateSideInput", - Create.timestamped( - TimestampedValue.of(KV.of("a", 1), new Instant(1)), - TimestampedValue.of(KV.of("a", 1), new Instant(2)), - TimestampedValue.of(KV.of("a", 2), new Instant(7)), - TimestampedValue.of(KV.of("b", 3), new Instant(14)))) - .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) - .apply(View.asMultimap()); - - PCollection> output = - pipeline - .apply( - "CreateMainInput", - Create.timestamped( - TimestampedValue.of(1 /* size */, new Instant(5)), - TimestampedValue.of(1 /* size */, new Instant(16)))) - .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - assertEquals((int) c.element(), c.sideInput(view).size()); - assertEquals((int) c.element(), c.sideInput(view).entrySet().size()); - for (Entry> entry : - c.sideInput(view).entrySet()) { - for (Integer value : entry.getValue()) { - c.output(KV.of(entry.getKey(), value)); - } - } - } - }) - .withSideInputs(view)); - - PAssert.that(output) - .containsInAnyOrder(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3)); - - pipeline.run(); - } - - @Test - @Category({ValidatesRunner.class}) - public void testWindowedMultimapSideInputWithNonDeterministicKeyCoder() { - - final PCollectionView>> view = - pipeline - .apply( - "CreateSideInput", - Create.timestamped( - TimestampedValue.of(KV.of("a", 1), new Instant(1)), - TimestampedValue.of(KV.of("a", 1), new Instant(2)), - TimestampedValue.of(KV.of("a", 2), new Instant(7)), - TimestampedValue.of(KV.of("b", 3), new Instant(14))) - .withCoder(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) - .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) - .apply(View.asMultimap()); - - PCollection> output = - pipeline - .apply( - "CreateMainInput", - Create.timestamped( - TimestampedValue.of("apple", new Instant(5)), - TimestampedValue.of("banana", new Instant(13)), - TimestampedValue.of("blackberry", new Instant(16)))) - .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { - c.output(KV.of(c.element(), v)); - } - } - }) - .withSideInputs(view)); - - PAssert.that(output) - .containsInAnyOrder( - KV.of("apple", 1), - KV.of("apple", 1), - KV.of("apple", 2), - KV.of("banana", 3), - KV.of("blackberry", 3)); - - pipeline.run(); - } - - @Test - @Category(ValidatesRunner.class) - public void testEmptyMultimapSideInput() throws Exception { - - final PCollectionView>> view = - pipeline - .apply( - "CreateEmptyView", Create.empty(KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of()))) - .apply(View.asMultimap()); - - PCollection results = - pipeline - .apply("Create1", Create.of(1)) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn() { - @ProcessElement - public void processElement(ProcessContext c) { - assertTrue(c.sideInput(view).isEmpty()); - assertTrue(c.sideInput(view).entrySet().isEmpty()); - assertFalse(c.sideInput(view).entrySet().iterator().hasNext()); - c.output(c.element()); - } - }) - .withSideInputs(view)); - - // Pass at least one value through to guarantee that DoFn executes. - PAssert.that(results).containsInAnyOrder(1); - - pipeline.run(); - } - - @Test - @Category({ValidatesRunner.class}) - public void testEmptyMultimapSideInputWithNonDeterministicKeyCoder() throws Exception { - - final PCollectionView>> view = - pipeline - .apply( - "CreateEmptyView", - Create.empty(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) - .apply(View.asMultimap()); - - PCollection results = - pipeline - .apply("Create1", Create.of(1)) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn() { - @ProcessElement - public void processElement(ProcessContext c) { - assertTrue(c.sideInput(view).isEmpty()); - assertTrue(c.sideInput(view).entrySet().isEmpty()); - assertFalse(c.sideInput(view).entrySet().iterator().hasNext()); - c.output(c.element()); - } - }) - .withSideInputs(view)); - - // Pass at least one value through to guarantee that DoFn executes. - PAssert.that(results).containsInAnyOrder(1); - - pipeline.run(); - } - - @Test - @Category(ValidatesRunner.class) - public void testMultimapSideInputIsImmutable() { - - final PCollectionView>> view = - pipeline.apply("CreateSideInput", Create.of(KV.of("a", 1))).apply(View.asMultimap()); - - PCollection> output = - pipeline - .apply("CreateMainInput", Create.of("apple")) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - try { - c.sideInput(view).clear(); - fail("Expected UnsupportedOperationException on clear()"); - } catch (UnsupportedOperationException expected) { - } - try { - c.sideInput(view).put("c", ImmutableList.of(3)); - fail("Expected UnsupportedOperationException on put()"); - } catch (UnsupportedOperationException expected) { - } - try { - c.sideInput(view).remove("c"); - fail("Expected UnsupportedOperationException on remove()"); - } catch (UnsupportedOperationException expected) { - } - try { - c.sideInput(view).putAll(new HashMap<>()); - fail("Expected UnsupportedOperationException on putAll()"); - } catch (UnsupportedOperationException expected) { - } - for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { - c.output(KV.of(c.element(), v)); - } - } - }) - .withSideInputs(view)); - - // Pass at least one value through to guarantee that DoFn executes. - PAssert.that(output).containsInAnyOrder(KV.of("apple", 1)); - - pipeline.run(); - } - - @Test - @Category(ValidatesRunner.class) - public void testMapSideInput() { - - final PCollectionView> view = - pipeline - .apply("CreateSideInput", Create.of(KV.of("a", 1), KV.of("b", 3))) - .apply(View.asMap()); - - PCollection> output = - pipeline - .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - c.output( - KV.of( - c.element(), - c.sideInput(view).get(c.element().substring(0, 1)))); - } - }) - .withSideInputs(view)); - - PAssert.that(output) - .containsInAnyOrder(KV.of("apple", 1), KV.of("banana", 3), KV.of("blackberry", 3)); - - pipeline.run(); - } - - @Test - @Category(ValidatesRunner.class) - public void testMapAsEntrySetSideInput() { - - final PCollectionView> view = - pipeline - .apply("CreateSideInput", Create.of(KV.of("a", 1), KV.of("b", 3))) - .apply(View.asMap()); - - PCollection> output = - pipeline - .apply("CreateMainInput", Create.of(2 /* size */)) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - assertEquals((int) c.element(), c.sideInput(view).size()); - assertEquals((int) c.element(), c.sideInput(view).entrySet().size()); - for (Entry entry : c.sideInput(view).entrySet()) { - c.output(KV.of(entry.getKey(), entry.getValue())); - } - } - }) - .withSideInputs(view)); - - PAssert.that(output).containsInAnyOrder(KV.of("a", 1), KV.of("b", 3)); - - pipeline.run(); - } - - @Test - @Category(NeedsRunner.class) - public void testMapInMemorySideInput() { - - final PCollectionView> view = - pipeline - .apply("CreateSideInput", Create.of(KV.of("a", 1), KV.of("b", 3))) - .apply(View.asMap().inMemory()); - - PCollection> output = - pipeline - .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - c.output( - KV.of( - c.element(), - c.sideInput(view).get(c.element().substring(0, 1)))); - } - }) - .withSideInputs(view)); - - PAssert.that(output) - .containsInAnyOrder(KV.of("apple", 1), KV.of("banana", 3), KV.of("blackberry", 3)); - - pipeline.run(); - } - - @Test - @Category(NeedsRunner.class) - public void testMapInMemorySideInputWithNonStructuralKey() { - - final PCollectionView> view = - pipeline - .apply( - "CreateSideInput", - Create.of( - KV.of("a".getBytes(StandardCharsets.UTF_8), 1), - KV.of("b".getBytes(StandardCharsets.UTF_8), 3))) - .apply(View.asMap().inMemory()); - - PCollection> output = - pipeline - .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - c.output( - KV.of( - c.element(), - c.sideInput(view) - .get( - c.element() - .substring(0, 1) - .getBytes(StandardCharsets.UTF_8)))); - } - }) - .withSideInputs(view)); - - PAssert.that(output) - .containsInAnyOrder(KV.of("apple", 1), KV.of("banana", 3), KV.of("blackberry", 3)); - - pipeline.run(); - } - - @Test - @Category({ValidatesRunner.class}) - public void testMapSideInputWithNonDeterministicKeyCoder() { - - final PCollectionView> view = - pipeline - .apply( - "CreateSideInput", - Create.of(KV.of("a", 1), KV.of("b", 3)) - .withCoder(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) - .apply(View.asMap()); - - PCollection> output = - pipeline - .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - c.output( - KV.of( - c.element(), - c.sideInput(view).get(c.element().substring(0, 1)))); - } - }) - .withSideInputs(view)); - - PAssert.that(output) - .containsInAnyOrder(KV.of("apple", 1), KV.of("banana", 3), KV.of("blackberry", 3)); - - pipeline.run(); - } - - @Test - @Category(ValidatesRunner.class) - public void testWindowedMapSideInput() { - - final PCollectionView> view = - pipeline - .apply( - "CreateSideInput", - Create.timestamped( - TimestampedValue.of(KV.of("a", 1), new Instant(1)), - TimestampedValue.of(KV.of("b", 2), new Instant(4)), - TimestampedValue.of(KV.of("b", 3), new Instant(18)))) - .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) - .apply(View.asMap()); - - PCollection> output = - pipeline - .apply( - "CreateMainInput", - Create.timestamped( - TimestampedValue.of("apple", new Instant(5)), - TimestampedValue.of("banana", new Instant(4)), - TimestampedValue.of("blackberry", new Instant(16)))) - .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - c.output( - KV.of( - c.element(), - c.sideInput(view).get(c.element().substring(0, 1)))); - } - }) - .withSideInputs(view)); - - PAssert.that(output) - .containsInAnyOrder(KV.of("apple", 1), KV.of("banana", 2), KV.of("blackberry", 3)); - - pipeline.run(); - } - - @Test - @Category(ValidatesRunner.class) - public void testWindowedMapAsEntrySetSideInput() { - - final PCollectionView> view = - pipeline - .apply( - "CreateSideInput", - Create.timestamped( - TimestampedValue.of(KV.of("a", 1), new Instant(1)), - TimestampedValue.of(KV.of("b", 2), new Instant(4)), - TimestampedValue.of(KV.of("b", 3), new Instant(18)))) - .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) - .apply(View.asMap()); - - PCollection> output = - pipeline - .apply( - "CreateMainInput", - Create.timestamped( - TimestampedValue.of(2 /* size */, new Instant(5)), - TimestampedValue.of(1 /* size */, new Instant(16)))) - .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - assertEquals((int) c.element(), c.sideInput(view).size()); - assertEquals((int) c.element(), c.sideInput(view).entrySet().size()); - for (Entry entry : c.sideInput(view).entrySet()) { - c.output(KV.of(entry.getKey(), entry.getValue())); - } - } - }) - .withSideInputs(view)); - - PAssert.that(output).containsInAnyOrder(KV.of("a", 1), KV.of("b", 2), KV.of("b", 3)); - - pipeline.run(); - } - - @Test - @Category({ValidatesRunner.class}) - public void testWindowedMapSideInputWithNonDeterministicKeyCoder() { - - final PCollectionView> view = - pipeline - .apply( - "CreateSideInput", - Create.timestamped( - TimestampedValue.of(KV.of("a", 1), new Instant(1)), - TimestampedValue.of(KV.of("b", 2), new Instant(4)), - TimestampedValue.of(KV.of("b", 3), new Instant(18))) - .withCoder(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) - .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) - .apply(View.asMap()); - - PCollection> output = - pipeline - .apply( - "CreateMainInput", - Create.timestamped( - TimestampedValue.of("apple", new Instant(5)), - TimestampedValue.of("banana", new Instant(4)), - TimestampedValue.of("blackberry", new Instant(16)))) - .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - c.output( - KV.of( - c.element(), - c.sideInput(view).get(c.element().substring(0, 1)))); - } - }) - .withSideInputs(view)); - - PAssert.that(output) - .containsInAnyOrder(KV.of("apple", 1), KV.of("banana", 2), KV.of("blackberry", 3)); - - pipeline.run(); - } - - @Test - @Category(ValidatesRunner.class) - public void testEmptyMapSideInput() throws Exception { - - final PCollectionView> view = - pipeline - .apply( - "CreateEmptyView", Create.empty(KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of()))) - .apply(View.asMap()); - - PCollection results = - pipeline - .apply("Create1", Create.of(1)) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn() { - @ProcessElement - public void processElement(ProcessContext c) { - assertTrue(c.sideInput(view).isEmpty()); - assertTrue(c.sideInput(view).entrySet().isEmpty()); - assertFalse(c.sideInput(view).entrySet().iterator().hasNext()); - c.output(c.element()); - } - }) - .withSideInputs(view)); - - // Pass at least one value through to guarantee that DoFn executes. - PAssert.that(results).containsInAnyOrder(1); - - pipeline.run(); - } - - @Test - @Category({ValidatesRunner.class}) - public void testEmptyMapSideInputWithNonDeterministicKeyCoder() throws Exception { - - final PCollectionView> view = - pipeline - .apply( - "CreateEmptyView", - Create.empty(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) - .apply(View.asMap()); - - PCollection results = - pipeline - .apply("Create1", Create.of(1)) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn() { - @ProcessElement - public void processElement(ProcessContext c) { - assertTrue(c.sideInput(view).isEmpty()); - assertTrue(c.sideInput(view).entrySet().isEmpty()); - assertFalse(c.sideInput(view).entrySet().iterator().hasNext()); - c.output(c.element()); - } - }) - .withSideInputs(view)); - - // Pass at least one value through to guarantee that DoFn executes. - PAssert.that(results).containsInAnyOrder(1); - - pipeline.run(); - } - - @Test - @Category(ValidatesRunner.class) - public void testMapSideInputWithNullValuesCatchesDuplicates() { - - final PCollectionView> view = - pipeline - .apply( - "CreateSideInput", - Create.of(KV.of("a", (Integer) null), KV.of("a", (Integer) null)) - .withCoder( - KvCoder.of(StringUtf8Coder.of(), NullableCoder.of(VarIntCoder.of())))) - .apply(View.asMap()); - - PCollection> output = - pipeline - .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - c.output( - KV.of( - c.element(), - c.sideInput(view) - .getOrDefault(c.element().substring(0, 1), 0))); - } - }) - .withSideInputs(view)); - - PAssert.that(output) - .containsInAnyOrder(KV.of("apple", 1), KV.of("banana", 3), KV.of("blackberry", 3)); - - // As long as we get an error, be flexible with how a runner surfaces it - thrown.expect(Exception.class); - - pipeline.run(); - } - - @Test - @Category(ValidatesRunner.class) - public void testMapSideInputIsImmutable() { - - final PCollectionView> view = - pipeline.apply("CreateSideInput", Create.of(KV.of("a", 1))).apply(View.asMap()); - - PCollection> output = - pipeline - .apply("CreateMainInput", Create.of("apple")) - .apply( - "OutputSideInputs", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - try { - c.sideInput(view).clear(); - fail("Expected UnsupportedOperationException on clear()"); - } catch (UnsupportedOperationException expected) { - } - try { - c.sideInput(view).put("c", 3); - fail("Expected UnsupportedOperationException on put()"); - } catch (UnsupportedOperationException expected) { - } - try { - c.sideInput(view).remove("c"); - fail("Expected UnsupportedOperationException on remove()"); - } catch (UnsupportedOperationException expected) { - } - try { - c.sideInput(view).putAll(new HashMap<>()); - fail("Expected UnsupportedOperationException on putAll()"); - } catch (UnsupportedOperationException expected) { - } - c.output( - KV.of( - c.element(), - c.sideInput(view).get(c.element().substring(0, 1)))); - } - }) - .withSideInputs(view)); - - // Pass at least one value through to guarantee that DoFn executes. - PAssert.that(output).containsInAnyOrder(KV.of("apple", 1)); - - pipeline.run(); - } - - @Test - @Category(ValidatesRunner.class) - public void testCombinedMapSideInput() { - - final PCollectionView> view = - pipeline - .apply("CreateSideInput", Create.of(KV.of("a", 1), KV.of("a", 20), KV.of("b", 3))) - .apply("SumIntegers", Combine.perKey(Sum.ofIntegers())) - .apply(View.asMap()); - - PCollection> output = - pipeline - .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) - .apply( - "Output", - ParDo.of( - new DoFn>() { - @ProcessElement - public void processElement(ProcessContext c) { - c.output( - KV.of( - c.element(), - c.sideInput(view).get(c.element().substring(0, 1)))); - } - }) - .withSideInputs(view)); - - PAssert.that(output) - .containsInAnyOrder(KV.of("apple", 21), KV.of("banana", 3), KV.of("blackberry", 3)); - - pipeline.run(); - } - @Test @Category(ValidatesRunner.class) public void testWindowedSideInputFixedToFixed() { @@ -1836,59 +880,4 @@ public void processElement(ProcessContext c) { pipeline.run(); } - - @Test - public void testViewGetName() { - assertEquals("View.AsSingleton", View.asSingleton().getName()); - assertEquals("View.AsIterable", View.asIterable().getName()); - assertEquals("View.AsMap", View.asMap().getName()); - assertEquals("View.AsMultimap", View.asMultimap().getName()); - } - - private void testViewUnbounded( - Pipeline pipeline, - PTransform>, ? extends PCollectionView> view) { - thrown.expect(IllegalStateException.class); - thrown.expectMessage("Unable to create a side-input view from input"); - thrown.expectCause( - ThrowableMessageMatcher.hasMessage(Matchers.containsString("non-bounded PCollection"))); - pipeline - .apply( - new PTransform>>() { - @Override - public PCollection> expand(PBegin input) { - return PCollection.createPrimitiveOutputInternal( - input.getPipeline(), - WindowingStrategy.globalDefault(), - PCollection.IsBounded.UNBOUNDED, - KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of())); - } - }) - .apply(view); - } - - @Test - public void testViewUnboundedAsSingletonDirect() { - testViewUnbounded(pipeline, View.asSingleton()); - } - - @Test - public void testViewUnboundedAsIterableDirect() { - testViewUnbounded(pipeline, View.asIterable()); - } - - @Test - public void testViewUnboundedAsListDirect() { - testViewUnbounded(pipeline, View.asList()); - } - - @Test - public void testViewUnboundedAsMapDirect() { - testViewUnbounded(pipeline, View.asMap()); - } - - @Test - public void testViewUnboundedAsMultimapDirect() { - testViewUnbounded(pipeline, View.asMultimap()); - } }