[
https://issues.apache.org/jira/browse/FLINK-41016?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18129536#comment-18129536
]
Dian Fu commented on FLINK-41016:
---------------------------------
[~sangkyoonnam] Sounds reasonable for me. Thanks for working on this issue!
> Missing uids on the operators PyFlink adds for key_by, partition_custom and
> KeyedStream can prevent submission with auto-generated UIDs disabled
> ------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41016
> URL: https://issues.apache.org/jira/browse/FLINK-41016
> Project: Flink
> Issue Type: Bug
> Components: API / Python
> Affects Versions: 2.2.0, 2.3.0, 2.2.1
> Reporter: Sangkyoon Nam
> Priority: Major
>
> The Python DataStream API adds helper operators that get no uid unless the
> user can reach them. {{key_by}} adds {{_stream_key_by_map_operator}} before
> the partitioning, and {{partition_custom}} adds
> {{_partition_custom_map_operator}} before it. {{KeyedStream.print}},
> {{add_sink}}, {{union}} and {{project}} call {{_values()}}, which adds
> {{_keyed_stream_values_operator}}, and {{DataStream.union}} does the same for
> each {{KeyedStream}} argument. {{partition_custom}} also adds a values helper
> after the partitioning, but that one is the stream it returns and can be
> given a uid. FLINK-40829 covers the same problem for the operators
> {{assign_timestamps_and_watermarks}} hides.
> *With auto-generated UIDs disabled, a job can fail to submit even with a uid
> on every operator the user can reach.* Reproduced with PyFlink 2.2.1, with a
> uid on the source, every Python {{map}} and the sink; row 4 also gives the
> stream {{partition_custom}} returns a uid. The helper creation, uid check and
> Python fusion code is the same in 2.2.0, 2.3.0 and master:
> || Pipeline || Result ||
> | Java source, {{key_by}}, keyed {{map}} | {{IllegalStateException}} for
> {{_stream_key_by_map_operator}} |
> | One Java source, two branches each with {{key_by}} and a keyed {{map}} |
> {{IllegalStateException}} for {{_stream_key_by_map_operator}} |
> | Java source, {{key_by}}, {{print}} | {{IllegalStateException}} for
> {{_stream_key_by_map_operator}} |
> | Java source, {{partition_custom}} with a uid, {{map}} |
> {{IllegalStateException}} for {{_partition_custom_map_operator}} |
> | Java source, Python {{map}}, {{key_by}}, keyed {{map}} | submits |
> The last row passes because {{PythonOperatorChainingOptimizer}} fuses the
> key-by helper into the preceding Python {{map}}, and the fused operator keeps
> that map's uid.
> *Proposal.* In {{PythonConfigUtil.configPythonOperator}}, derive a uid for
> each helper that has neither a uid nor a hash: {{<input uid>-key-by}},
> {{<input uid>-partition-custom}} and {{<input uid>-values}}, with {{-2}},
> {{-3}} for further helpers of the same kind on the same input. The input is
> the upstream transformation past any {{PartitionTransformation}}; a helper
> whose input has no uid is left alone. Do it only when auto-generated UIDs are
> disabled, so jobs that run with them enabled keep their operator IDs. I tried
> this on master with unit tests. With a {{flink-python}} jar built from
> {{release-2.2}} plus this change, used with PyFlink 2.2.1, all five rows
> submit, and so does the {{partition_custom}} pipeline without a uid on the
> stream it returns.
> *Separate decision: ignored downstream uids.* When the optimizer fuses a
> values helper with a following Python operator, it gives the fused operator
> the helper's uid, even when that is null, and drops the downstream operator's
> uid. With stock PyFlink 2.2.1 and auto-generated UIDs enabled, changing {{x}}
> from {{"map"}} to {{"other"}} in
> {{source.partition_custom(...).map(...).uid(x)}}, without a uid on the
> {{partition_custom}} stream, left every generated operator ID unchanged; with
> {{key_by}} instead, the keyed {{map}}'s ID changed. I compared
> {{getGeneratedOperatorID()}} values in the job graphs and didn't test a
> restore. The proposal doesn't fix this: with auto-generated UIDs disabled,
> the fused operator gets the derived values uid in that example, or the
> returned stream's uid in row 4, and with them enabled nothing changes.
> Keeping the downstream operator's own uid when a values helper is fused into
> it would fix it, but it changes the generated IDs of existing fused
> operators, so savepoint compatibility needs its own decision.
> [~dianfu], does the derived-uid approach work for you for the first part? I'd
> like to work on it, and I'd file the ignored-uid part as its own ticket
> unless you'd rather handle both here.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
