sangkyoonnam opened a new pull request, #29545: URL: https://github.com/apache/flink/pull/29545
## What is the purpose of the change With auto-generated UIDs disabled, a PyFlink job can fail to submit even when every operator the user can reach has a uid. `key_by`, `partition_custom` and the calls that turn a `KeyedStream` back into a `DataStream` add helper operators without a uid. Users can't reach most of them, and `KeyedStream.uid()` raises. `key_by` right after a Java source, for example, fails with `IllegalStateException: Auto generated UIDs have been disabled but no UID or hash has been assigned to operator _stream_key_by_map_operator`. [FLINK-41016](https://issues.apache.org/jira/browse/FLINK-41016) has the reproduction. FLINK-40829 (#29312) fixed this for the operators `assign_timestamps_and_watermarks` hides; this PR derives uids for the `key_by`, `partition_custom` and values helpers when their inputs have uids. @florianvazelle, you mentioned the `key_by` side in #29312. Would you take a look? ## Brief change log - `PythonConfigUtil.configPythonOperator` gives each helper that has neither a uid nor a uid hash a uid derived from its input's uid, looking past partition transformations: `<input uid>-key-by`, `<input uid>-partition-custom` or `<input uid>-values`. A derived uid that is already taken gets `-2`, `-3` and so on. - An input with only a uid hash is just as valid with auto-generated UIDs disabled and hits the same failure, so its helpers get a uid hash derived from it, the same way #29312 derives one for the operators `assign_timestamps_and_watermarks` hides, from the input's hash and the same suffixes. - This runs in `PythonConfigUtil` when the job is configured rather than in the Python API as #29312 does, so it also sees input uids set after `key_by` or `partition_custom` was called. Unlike the stream #29312 returns, a `KeyedStream` rejects `.uid()`. - This only runs when auto-generated UIDs are disabled, so jobs that run with them enabled keep their operator IDs. #29312 instead derives identifiers whenever `.uid()` or `.set_uid_hash()` is called on the stream it returns, regardless of that setting, since the user asks for it there. Jobs that could already be submitted with them disabled keep theirs too: a helper without its own uid or uid hash could only get through by being fused into the Python operator before it, which supplies the uid or uid hash. A helper is left alone when its input, past partition transformations, has neither a uid nor a uid hash. After a `union`, the user can add an operator with a uid before `key_by`; `get_side_output()` already returns a Python `map` that accepts one. The numbered suffixes follow the order the helpers were created in, so adding a `key_by` on the same input earlier in the job can move an ID from one branch to another. That matters for the values helper after a `KeyedStream`, a keyed Python operator that registers a timer service: `--allowNonRestoredState` can drop state that no longer matches, but it can't fix state that now maps to the wrong helper. I haven't tested a restore. A helper keeps the uid it got when the job is configured again, so a user uid added later with the same value still collides. ## Verifying this change This change added tests and can be verified as follows: - Added `test_key_by_uid`, `test_key_by_uid_hash` and `test_partition_custom_uid` to `test_data_stream.py`: with auto-generated UIDs disabled, `get_execution_plan()` succeeds and the helpers get the expected uids, or the uid hashes `_derive_uid_hash` gives. All three fail on `master`. `test_key_by_uid_after_python_operator` checks that a key-by helper fused into a Python `map` keeps that map's uid, and passes with and without the change. - Added tests to `PythonConfigUtilTest` for the derived uids and uid hashes, collisions with existing uids and uid hashes (ignoring hex case) and with a branch added after the job was configured once, the enabled case, and helpers that already have a uid or whose input has none. The derivation and collision tests fail on `master`. - I built `flink-python` from `release-2.2` with an earlier version of this change, before the uid hash part, and used it with PyFlink 2.2.1. With auto-generated UIDs disabled, seven jobs fail to submit without the change and run to completion with it: `key_by` after a Java source, two `key_by`s on one source, `key_by` then `print`, `union` with a `KeyedStream`, a keyed `reduce` followed by another `key_by`, and `partition_custom` with and without a uid on the returned stream. - With `python.operator-chaining.enabled` set to `false`, `key_by` after a Python `map` fails the same way without the change and runs with it. With auto-generated UIDs enabled, all eight jobs run with and without the change. - With Python operator chaining enabled, the generated operator IDs in the job graph are the same with and without the change for the seven jobs with auto-generated UIDs enabled, and for `key_by` after a Python `map` with them disabled, which already ran before. A fused operator still takes the uid and uid hash of the first operator in the chain. A key-by helper fused into the Python `map` before it keeps that map's uid; when a values helper is fused into the Python operator after it, that operator's own uid is still ignored. Changing which operator supplies the uid would change the IDs of existing fused operators, so I'd handle it separately. ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): no - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no - The serializers: no - The runtime per-record code paths (performance sensitive): no - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes, it assigns operator uids when PyFlink builds the job graph with auto-generated UIDs disabled. I haven't tested a savepoint restore. - The S3 file system connector: no ## Documentation - Does this pull request introduce a new feature? no - If yes, how is the feature documented? not applicable --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) AI was used for the implementation, the verification runs and this description. Generated-by: Claude Code 2.1.302 (Claude Opus 5.5) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
