[BEAM-22] Allow InProcess Evaluators to check Side Input completion #220
Conversation
This checks to ensure that the PCollectionView in the SideInputWindow for the provided window either has elements available or is empty. Schedule a future to ensure that the SideInputWindows are appropriately filled with an empty iterable after retreiving the element.
tgroh
force-pushed
the
ippr_side_input_is_ready
branch
from
ffc47d7 to
ab26b17
Compare
| for (PCollectionView<?> view : readerViews) { | ||
| try { | ||
| BoundedWindow viewWindow = | ||
| view.getWindowingStrategyInternal().getWindowFn().getSideInputWindow(elementWindow); |
There was a problem hiding this comment.
I feel the conversion from the main input window to the side input window doesn't belong to the side input reader.
allViewsReadyInWindow(mainWindow) is not consistent with get(...) method.
There was a problem hiding this comment.
isReady(view, sideInputWindow) is symmetric with get(view, sideInputWindow). Changed to that.
| try { | ||
| Future<Iterable<? extends WindowedValue<?>>> viewContents = | ||
| getViewFuture(view, window); | ||
| if (!viewContents.isDone()) { |
There was a problem hiding this comment.
return viewContents.isDone();
and remove return true at the very end
| * contents if necessary. | ||
| */ | ||
| private <T> Future<Iterable<? extends WindowedValue<?>>> getViewFuture( | ||
| final PCollectionView<T> view, final BoundedWindow window) throws ExecutionException { |
There was a problem hiding this comment.
If you wrap ExecutionException into a RuntimeException here, you can avoid to do it twice outside.
There was a problem hiding this comment.
Using getUnchecked - our CacheLoader will never throw an exception.
| + "Contained views; %s", | ||
| view, | ||
| readerViews); | ||
| Future<Iterable<? extends WindowedValue<?>>> viewContents = getViewFuture(view, window); |
There was a problem hiding this comment.
inline?
return getViewFuture(view, window).isDone();
|
LGTM |
|
@kennknowles for committer |
| } | ||
|
|
||
| @Test | ||
| public void isReadyForSomeNotReadyViewsFalseUntilElements() { |
There was a problem hiding this comment.
javadoc this a little bit so one has context when parsing the test
|
LGTM, merging. |
Co-authored-by: Tres Seaver <tseaver@palladion.com> Co-authored-by: Christopher Wilcox <crwilcox@google.com>
3 participants
Be sure to do all of the following to help us incorporate your contribution
quickly and easily:
[BEAM-<Jira issue #>] Description of pull requestmvn clean verify. (Even better, enableTravis-CI on your fork and ensure the whole test matrix passes).
<Jira issue #>in the title with the actual Jira issuenumber, if there is one.
Individual Contributor License Agreement.
This checks to ensure that the PCollectionView in the SideInputWindow
for the provided window either has elements available or is empty.
Schedule a future to ensure that the SideInputWindows are appropriately
filled with an empty iterable after retreiving the element.
This is allows the ParDoEvaluator to not attempt to process elements
that cannot currently be completed.