[
https://issues.apache.org/jira/browse/BEAM-4255?focusedWorklogId=105762&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-105762
]
ASF GitHub Bot logged work on BEAM-4255:
----------------------------------------
Author: ASF GitHub Bot
Created on: 24/May/18 22:36
Start Date: 24/May/18 22:36
Worklog Time Spent: 10m
Work Description: huygaa11 closed pull request #5311: [BEAM-4255] Set
singleton view writers' accumulation mode to discarding
URL: https://github.com/apache/beam/pull/5311
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java
index 6278d312d46..87e5b3db3b2 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java
@@ -1283,6 +1283,10 @@ private GloballyAsSingletonView(
insertDefault,
insertDefault ? fn.defaultValue() : null,
combined.getCoder());
+
+ // GBK's inside view writers for singleton view should discard previous
values.
+
materializationInput.setWindowingStrategyInternal(materializationInput.getWindowingStrategy()
+
.withMode(WindowingStrategy.AccumulationMode.DISCARDING_FIRED_PANES));
materializationInput.apply(CreatePCollectionView.of(view));
return view;
}
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 e34e98f1070..06a31cc6cb1 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
@@ -1299,6 +1299,38 @@ public void processElement(ProcessContext c) {
pipeline.run();
}
+ /**
+ * This test is added to catch an issue that singleton side input with
accumulating triggers
+ * were writing multiple values as singleton view. For more info:
+ * https://issues.apache.org/jira/browse/BEAM-4255
+ */
+ @Test
+ @Category({ ValidatesRunner.class })
+ public void testCombineGloballyAsSingletonViewWithTrigger() {
+ final PCollectionView<Integer> view = pipeline
+ .apply("CreateSideInput",
+ Create.timestamped(
+ TimestampedValue.of(1, new Instant(100)),
+ TimestampedValue.of(2, new Instant(100)),
+ TimestampedValue.of(3, new Instant(200)),
+ TimestampedValue.of(4, new Instant(200))))
+ .apply(Window.<Integer>into(new GlobalWindows())
+ .triggering(Repeatedly.forever(AfterPane.elementCountAtLeast(2)))
+ .accumulatingFiredPanes())
+ .apply(Sum.integersGlobally().asSingletonView());
+
+ PCollection<Integer> output = pipeline
+ .apply("CreateVoidMainInput", Create.of((Void) null))
+ .apply("OutputSideInput", ParDo.of(new DoFn<Void, Integer>() {
+ @ProcessElement
+ public void processElement(ProcessContext c) {
+ c.output(c.sideInput(view));
+ }
+ }).withSideInputs(view));
+
+ pipeline.run();
+ }
+
@Test
@Category(ValidatesRunner.class)
public void testWindowedCombineGloballyAsSingletonView() {
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 105762)
Time Spent: 1h (was: 50m)
> GloballyAsSingletonView is writing multiple values to the SingletonView
> -----------------------------------------------------------------------
>
> Key: BEAM-4255
> URL: https://issues.apache.org/jira/browse/BEAM-4255
> Project: Beam
> Issue Type: Bug
> Components: sdk-java-core
> Affects Versions: 2.4.0
> Reporter: Batkhuyag Batsaikhan
> Assignee: Batkhuyag Batsaikhan
> Priority: Major
> Fix For: 3.0.0
>
> Time Spent: 1h
> Remaining Estimate: 0h
>
> The code snippet below is writing multiple values to the sideInput. Resulting
> in exception: "PCollection with more than one element accessed as a singleton
> view."
> PCollectionView<Integer> globalView = input
> .apply(Window.<Integer>into(new GlobalWindows())
> .triggering(Repeatedly.forever(AfterPane.elementCountAtLeast(1)))
> .accumulatingFiredPanes())
> .apply(Sum.integersGlobally().asSingletonView())
>
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
