apache / apache/beam

SDF should properly support windowed side inputs

Open
#18,366 0 comments 0 reactions 0 assignees View on GitHub
apex bug core dataflow direct flink java P3 runners
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
2d 2h
Merged PRs (30d)
205

Description

Currently there is no test coverage for Splittable DoFn **** windowed side inputs, especially when not all of the side input windows are ready.

Moreover, current implementation of SDF in the direct runner is definitely wrong: it uses a ParDoEvaluator to run the ProcessFn, and this ParDoEvaluator looks at the wrong windows to decide which windows are ready and which are not: https://github.com/apache/beam/blob/master/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluator.java#L134 - the WindowedValue in question is a KeyedWorkItem, and they are always in the global window, but the important windows are windows of elements inside this KWI's elementsIterable().

The Flink implementation is also wrong in the same way.

This JIRA is to:
1) add test coverage for this case
2) implement proper support in all runners

I believe the easiest way to do 2) is to:
- make SplittableParDo, in case the DoFn has side inputs, pre-explode windows before feeding them into GroupByKeyIntoKeyedWorkItems , so that the resulting KWI's have elements only in a single window
- tweak runners to look at the proper window, and assert that there's only one window, while evaluating ProcessFn, in case the DoFn uses side inputs

Imported from Jira [BEAM-1983](https://issues.apache.org/jira/browse/BEAM-1983). Original Jira may contain additional context.
Reported by: jkff.

Contributor guide

Open the contributing guide

Research direction

Start with runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluator.java at the referenced readiness check, then trace SplittableParDo and GroupByKeyIntoKeyedWorkItems. Add coverage for windowed side inputs when some windows are not ready, and inspect the corresponding Flink implementation. Done means the test passes and all affected runners evaluate the proper side-input window.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
data-engineering, distributed-systems
Issue type
Bug
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.