Sangkyoon Nam created FLINK-41016:
-------------------------------------
Summary: 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.1, 2.2.0, 2.3.0
Reporter: Sangkyoon Nam
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)
