[beam] 03/03: Add a test to check that CombineGlobally preserves windowing

[ec...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git

commit c6d2c5256dcff917e9a2f42da2ab23114bda5a8b
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Tue Jul 9 14:57:18 2019 +0200

    Add a test to check that CombineGlobally preserves windowing
---
 .../translation/batch/CombineTest.java                   | 16 +++++++++++++++-
 1 file changed, 15 insertions(+), 1 deletion(-)

diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
index 3505da5..cae46a6 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
@@ -25,6 +25,7 @@ import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.transforms.Combine;
 import org.apache.beam.sdk.transforms.Create;
 import org.apache.beam.sdk.transforms.Sum;
 import org.apache.beam.sdk.transforms.windowing.FixedWindows;
@@ -35,7 +36,6 @@ import org.apache.beam.sdk.values.TimestampedValue;
 import org.joda.time.Duration;
 import org.joda.time.Instant;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
@@ -62,6 +62,20 @@ public class CombineTest implements Serializable {
   }
 
   @Test
+  public void testCombineGloballyPreservesWindowing() {
+    PCollection<Integer> input = p.apply(Create.timestamped(
+        TimestampedValue.of(1, new Instant(1)),
+        TimestampedValue.of(2, new Instant(2)),
+        TimestampedValue.of(3, new Instant(11)),
+        TimestampedValue.of(4, new Instant(3)),
+        TimestampedValue.of(5, new Instant(11)),
+        TimestampedValue.of(6, new Instant(12))))
+        .apply(Window.into(FixedWindows.of(Duration.millis(10))))
+        .apply(Combine.globally(Sum.ofIntegers()).withoutDefaults());
+    PAssert.that(input).containsInAnyOrder(7, 14);
+  }
+
+  @Test
   public void testCombinePerKey() {
     List<KV<Integer, Integer>> elems = new ArrayList<>();
     elems.add(KV.of(1, 1));