You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by tg...@apache.org on 2017/02/15 01:31:52 UTC
[1/3] beam git commit: Match Pane in WindowedValueMatcher
Repository: beam
Updated Branches:
refs/heads/master 2685ba528 -> 9061c65e3
Match Pane in WindowedValueMatcher
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/4e69a797
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/4e69a797
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/4e69a797
Branch: refs/heads/master
Commit: 4e69a7976faf32863eb796bfebf470332fcbd709
Parents: 2685ba5
Author: Thomas Groh <tg...@google.com>
Authored: Tue Feb 14 16:32:08 2017 -0800
Committer: Thomas Groh <tg...@google.com>
Committed: Tue Feb 14 17:31:33 2017 -0800
----------------------------------------------------------------------
.../test/java/org/apache/beam/runners/core/WindowMatchers.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/4e69a797/runners/core-java/src/test/java/org/apache/beam/runners/core/WindowMatchers.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/WindowMatchers.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/WindowMatchers.java
index 6c3a7e2..9769d10 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/WindowMatchers.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/WindowMatchers.java
@@ -198,7 +198,8 @@ public class WindowMatchers {
protected boolean matchesSafely(WindowedValue<? extends T> windowedValue) {
return valueMatcher.matches(windowedValue.getValue())
&& timestampMatcher.matches(windowedValue.getTimestamp())
- && windowsMatcher.matches(windowedValue.getWindows());
+ && windowsMatcher.matches(windowedValue.getWindows())
+ && paneInfoMatcher.matches(windowedValue.getPane());
}
}
}
[2/3] beam git commit: Propagate Pane in WindowEvaluatorFactory
Posted by tg...@apache.org.
Propagate Pane in WindowEvaluatorFactory
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/29fc84b2
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/29fc84b2
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/29fc84b2
Branch: refs/heads/master
Commit: 29fc84b2acd64b0028f189ff5350ed91e33854ad
Parents: 4e69a79
Author: Thomas Groh <tg...@google.com>
Authored: Tue Feb 14 16:19:54 2017 -0800
Committer: Thomas Groh <tg...@google.com>
Committed: Tue Feb 14 17:31:34 2017 -0800
----------------------------------------------------------------------
.../runners/direct/WindowEvaluatorFactory.java | 3 +-
.../direct/WindowEvaluatorFactoryTest.java | 41 ++++++++++----------
2 files changed, 22 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/29fc84b2/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WindowEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WindowEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WindowEvaluatorFactory.java
index 4ca556b..3cf178c 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WindowEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WindowEvaluatorFactory.java
@@ -25,7 +25,6 @@ import org.apache.beam.runners.direct.DirectRunner.UncommittedBundle;
import org.apache.beam.sdk.transforms.AppliedPTransform;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
-import org.apache.beam.sdk.transforms.windowing.PaneInfo;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.transforms.windowing.Window.Bound;
import org.apache.beam.sdk.transforms.windowing.WindowFn;
@@ -91,7 +90,7 @@ class WindowEvaluatorFactory implements TransformEvaluatorFactory {
Collection<? extends BoundedWindow> windows = assignWindows(windowFn, element);
outputBundle.add(
WindowedValue.<InputT>of(
- element.getValue(), element.getTimestamp(), windows, PaneInfo.NO_FIRING));
+ element.getValue(), element.getTimestamp(), windows, element.getPane()));
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/29fc84b2/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WindowEvaluatorFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WindowEvaluatorFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WindowEvaluatorFactoryTest.java
index aa841ed..7e6eb2f 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WindowEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WindowEvaluatorFactoryTest.java
@@ -19,6 +19,7 @@ package org.apache.beam.runners.direct;
import static org.apache.beam.runners.core.WindowMatchers.isSingleWindowedValue;
import static org.apache.beam.runners.core.WindowMatchers.isWindowedValue;
+import static org.apache.beam.sdk.transforms.windowing.PaneInfo.NO_FIRING;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.junit.Assert.assertThat;
import static org.mockito.Mockito.when;
@@ -43,6 +44,7 @@ import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
import org.apache.beam.sdk.transforms.windowing.NonMergingWindowFn;
import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo.Timing;
import org.apache.beam.sdk.transforms.windowing.SlidingWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.transforms.windowing.Window.Bound;
@@ -77,12 +79,12 @@ public class WindowEvaluatorFactoryTest {
private WindowedValue<Long> valueInGlobalWindow =
WindowedValue.timestampedValueInGlobalWindow(3L, new Instant(2L));
+ private final PaneInfo intervalWindowPane = PaneInfo.createPane(false, false, Timing.LATE, 3, 2);
private WindowedValue<Long> valueInIntervalWindow =
WindowedValue.of(
Long.valueOf(2L),
new Instant(-10L),
- new IntervalWindow(new Instant(-100), EPOCH),
- PaneInfo.NO_FIRING);
+ new IntervalWindow(new Instant(-100), EPOCH), intervalWindowPane);
private IntervalWindow intervalWindow1 =
new IntervalWindow(EPOCH, BoundedWindow.TIMESTAMP_MAX_VALUE);
@@ -91,12 +93,13 @@ public class WindowEvaluatorFactoryTest {
new IntervalWindow(
EPOCH.plus(Duration.standardDays(3)), EPOCH.plus(Duration.standardDays(6)));
+ private final PaneInfo multiWindowPane = PaneInfo.createPane(false, true, Timing.ON_TIME, 3, 0);
private WindowedValue<Long> valueInGlobalAndTwoIntervalWindows =
WindowedValue.of(
Long.valueOf(1L),
EPOCH.plus(Duration.standardDays(3)),
ImmutableList.of(GlobalWindow.INSTANCE, intervalWindow1, intervalWindow2),
- PaneInfo.NO_FIRING);
+ multiWindowPane);
@Rule
public TestPipeline p = TestPipeline.create().enableAbandonedNodeEnforcement(false);
@@ -158,18 +161,18 @@ public class WindowEvaluatorFactoryTest {
committed.getElements(),
containsInAnyOrder(
// value in global window
- isSingleWindowedValue(3L, new Instant(2L), firstSecondWindow, PaneInfo.NO_FIRING),
+ isSingleWindowedValue(3L, new Instant(2L), firstSecondWindow, NO_FIRING),
// value in just interval window
- isSingleWindowedValue(2L, new Instant(-10L), thirdWindow, PaneInfo.NO_FIRING),
+ isSingleWindowedValue(2L, new Instant(-10L), thirdWindow, intervalWindowPane),
// value in global window and two interval windows
isSingleWindowedValue(
- 1L, EPOCH.plus(Duration.standardDays(3)), firstSecondWindow, PaneInfo.NO_FIRING),
+ 1L, EPOCH.plus(Duration.standardDays(3)), firstSecondWindow, multiWindowPane),
isSingleWindowedValue(
- 1L, EPOCH.plus(Duration.standardDays(3)), firstSecondWindow, PaneInfo.NO_FIRING),
+ 1L, EPOCH.plus(Duration.standardDays(3)), firstSecondWindow, multiWindowPane),
isSingleWindowedValue(
- 1L, EPOCH.plus(Duration.standardDays(3)), firstSecondWindow, PaneInfo.NO_FIRING)));
+ 1L, EPOCH.plus(Duration.standardDays(3)), firstSecondWindow, multiWindowPane)));
}
@Test
@@ -204,31 +207,31 @@ public class WindowEvaluatorFactoryTest {
valueInGlobalWindow.getValue(),
valueInGlobalWindow.getTimestamp(),
ImmutableSet.of(w1, wMinusSlide),
- PaneInfo.NO_FIRING),
+ NO_FIRING),
// Value in interval window mapped to one windowed value in multiple windows
isWindowedValue(
valueInIntervalWindow.getValue(),
valueInIntervalWindow.getTimestamp(),
ImmutableSet.of(wMinus1, wMinusSlide),
- PaneInfo.NO_FIRING),
+ valueInIntervalWindow.getPane()),
// Value in three windows mapped to three windowed values in the same multiple windows
isWindowedValue(
valueInGlobalAndTwoIntervalWindows.getValue(),
valueInGlobalAndTwoIntervalWindows.getTimestamp(),
ImmutableSet.of(w1, w2),
- PaneInfo.NO_FIRING),
+ valueInGlobalAndTwoIntervalWindows.getPane()),
isWindowedValue(
valueInGlobalAndTwoIntervalWindows.getValue(),
valueInGlobalAndTwoIntervalWindows.getTimestamp(),
ImmutableSet.of(w1, w2),
- PaneInfo.NO_FIRING),
+ valueInGlobalAndTwoIntervalWindows.getPane()),
isWindowedValue(
valueInGlobalAndTwoIntervalWindows.getValue(),
valueInGlobalAndTwoIntervalWindows.getTimestamp(),
ImmutableSet.of(w1, w2),
- PaneInfo.NO_FIRING)));
+ valueInGlobalAndTwoIntervalWindows.getPane())));
}
@Test
@@ -256,14 +259,14 @@ public class WindowEvaluatorFactoryTest {
new IntervalWindow(
valueInGlobalWindow.getTimestamp(),
valueInGlobalWindow.getTimestamp().plus(1L)),
- PaneInfo.NO_FIRING),
+ valueInGlobalWindow.getPane()),
// Value in interval window mapped to the same window
isWindowedValue(
valueInIntervalWindow.getValue(),
valueInIntervalWindow.getTimestamp(),
valueInIntervalWindow.getWindows(),
- PaneInfo.NO_FIRING),
+ valueInIntervalWindow.getPane()),
// Value in global window and two interval windows exploded and mapped in both ways
isSingleWindowedValue(
@@ -272,19 +275,17 @@ public class WindowEvaluatorFactoryTest {
new IntervalWindow(
valueInGlobalAndTwoIntervalWindows.getTimestamp(),
valueInGlobalAndTwoIntervalWindows.getTimestamp().plus(1L)),
- PaneInfo.NO_FIRING),
-
+ valueInGlobalAndTwoIntervalWindows.getPane()),
isSingleWindowedValue(
valueInGlobalAndTwoIntervalWindows.getValue(),
valueInGlobalAndTwoIntervalWindows.getTimestamp(),
intervalWindow1,
- PaneInfo.NO_FIRING),
-
+ valueInGlobalAndTwoIntervalWindows.getPane()),
isSingleWindowedValue(
valueInGlobalAndTwoIntervalWindows.getValue(),
valueInGlobalAndTwoIntervalWindows.getTimestamp(),
intervalWindow2,
- PaneInfo.NO_FIRING)));
+ valueInGlobalAndTwoIntervalWindows.getPane())));
}
private CommittedBundle<Long> createInputBundle() {
[3/3] beam git commit: This closes #2008
Posted by tg...@apache.org.
This closes #2008
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/9061c65e
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/9061c65e
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/9061c65e
Branch: refs/heads/master
Commit: 9061c65e3dbab4391a5534558e6af652d8b30945
Parents: 2685ba5 29fc84b
Author: Thomas Groh <tg...@google.com>
Authored: Tue Feb 14 17:31:35 2017 -0800
Committer: Thomas Groh <tg...@google.com>
Committed: Tue Feb 14 17:31:35 2017 -0800
----------------------------------------------------------------------
.../beam/runners/core/WindowMatchers.java | 3 +-
.../runners/direct/WindowEvaluatorFactory.java | 3 +-
.../direct/WindowEvaluatorFactoryTest.java | 41 ++++++++++----------
3 files changed, 24 insertions(+), 23 deletions(-)
----------------------------------------------------------------------