[
https://issues.apache.org/jira/browse/FLINK-20491?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17246439#comment-17246439
]
Aljoscha Krettek commented on FLINK-20491:
------------------------------------------
Thanks, those are very thoughtful comments! I'll try and answer individually.
I think we don't need to sort the broadcast side. My assumption is that the
broadcast side is for building up some state that the other side is then
"streamed by" (even though it's not really streamed for BATCH execution).
Ordering by time should not be required for that case. If we see that it
becomes a requirement in the future we could still do it.
The above resolves questions about sorting non-keyed inputs and the {{byte[]}}
array trick for now, by sidestepping them.
I would like to completely separate the different types of input, yes. What I
would do next is change {{MultiInputSortingDataInput.wrapInputs()}} to pass in
separately: inputs that need sorting and pass-through inputs.
This leaves the problem of the {{BatchExecutionKeyedStateBackend}}. It's true
that we also cannot support iterating over keys from the keyed side. However, I
think we need to silently ignore it because real-word programs will have to use
this from the broadcast side for {{STREAMING}} execution. I'll think about what
the implications are some more. Maybe it's fine to make it a no-op for both
sides. Or maybe we'll disallow iterating from the keyed side for all execution
modes. Although we then have to figure out how we determine "on which side" we
are.
> Support Broadcast State in BATCH execution mode
> -----------------------------------------------
>
> Key: FLINK-20491
> URL: https://issues.apache.org/jira/browse/FLINK-20491
> Project: Flink
> Issue Type: Improvement
> Components: API / DataStream
> Reporter: Aljoscha Krettek
> Assignee: Aljoscha Krettek
> Priority: Major
> Labels: pull-request-available
>
> Right now, we don't support {{DataStream.connect(BroadcastStream)}} in
> {{BATCH}} execution mode. I believe we can add support for this with not too
> much work.
> The key insight is that we can process the broadcast side before the
> non-broadcast side. Initially, we were shying away from this because of
> concerns about {{ctx.applyToKeyedState()}} which allows the broadcast side of
> the user function to access/iterate over state from the keyed side. We
> thought that we couldn't support this. However, since we know that we process
> the broadcast side first we know that the keyed side will always be empty
> when doing so. We can thus just make this "keyed iteration" call a no-op,
> instead of throwing an exception as we do now.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
