dabla commented on PR #62922: URL: https://github.com/apache/airflow/pull/62922#issuecomment-6165326907
## Why `iterableoperator.py` changed shape in the last three commits A short guide for reviewers, and for any agent asked to review the PR: the last three commits restructure `iterableoperator.py` **without changing behaviour**. Here is why, and how to check it. ### Why Each review round fixed a real problem (kill paths, checkpoint ordering, thread affinity, params, the input fingerprint). Each fix landed where the code already was: as another method, parameter or flag on `IterableOperator`. By the approval round the class was **1,033 lines with 33 methods**. Each method was sound on its own, but the class no longer had one responsibility. You could see it in the code: - one run's executor, context, template environment, checkpoint flags and outcomes were passed from method to method through six methods (`_run_index` took seven parameters); - `IterationState` held two jobs behind two locks; - static methods used no state of their class. The structure was the consequence of the review rounds, so we fixed it in this PR rather than leaving it for a follow-up. It's the core of the feature, and the next fix would otherwise land on the same pile. ### What | Commit | Change | |---|---| | `e57eb9a931` | The input fingerprint moves out of the operator into `definitions/_internal/fingerprint.py`. | | `9726e07e08` | The fingerprint module is split by what each part holds: `InputFingerprint` (one index), `_UpstreamInputs` and `_LazyDigests` (one per-run cache each), canonical JSON as a module function. | | `d6a69c59f0` | One run and one indexed task get classes of their own. | `iterableoperator.py` now reads top-down, with `IterableOperator` last, after everything it uses: | Class | Responsibility | Was | |---|---|---| | `Checkpoints` | whether this attempt may resume, and whether one index's checkpoint may be trusted | + the trust check from `_replay_checkpoint` | | `IterationState` | what a kill must reach: indexed operators in flight, the stop flag, the kill threads; kills them | minus the run's data and its lock | | `IndexedTaskOutcomes` | what the indexed tasks ended with, and the task's outcome | `_retryable` / `_fail_fast_for` became module functions (no outcome state) | | `IndexedTaskUnmapper` | builds one index's indexed task: unmap, params, render, fingerprint | `_create_indexed_task`, `_render_unmapped_operator` | | `IndexedTaskLifecycle` | one indexed task from its checkpoint to its outcome: replay, run, checkpoint, publish / skip / fail, reports on the right thread | `_run_indexed_task`, `_replay_checkpoint`, `_publish`, `_record_skip`, `_record_failure`, `_report_indexed_task` (static), `axcom_push` | | `IterationRun` | one `execute()`: reads the input, drives every index to an outcome, kills what is in flight when the loop ends early | `_run_tasks`, `_run_index`, the result built in `execute` | | `IterableOperator` | the Dag node: forwards the wrapped operator, `execute`, `on_kill`, the partial skip | **1,033 → 399 lines (about 160 of them the class docstring), 33 → 12 methods** | ### Principles applied - **Single responsibility**, checked on the whole class after the fixes, not on each diff (Martin, [*SRP*](https://blog.cleancoder.com/uncle-bob/2014/05/08/SingleReponsibilityPrinciple.html)). - **Introduce Parameter Object**: values passed together from method to method are an object that is missing ([Fowler](https://refactoring.com/catalog/introduceParameterObject.html)). - **Extract Class**: methods that use fields the rest of the class never touches are a class of their own ([Fowler](https://refactoring.com/catalog/extractClass.html)). - **Move Function**: a static method that uses no state of its class moves to where its data is, or becomes a module function ([Fowler](https://refactoring.com/catalog/moveFunction.html)). - **Composition over inheritance**: the new classes are collaborators the operator makes and holds, never bases. `IterableOperator` subclasses `BaseOperator` only because the Dag needs one. Two choices were deliberate: - **`IterationState` stays on the operator** and is renewed in `execute`. `on_kill` arrives on the operator, possibly before the run exists or after it ended. The renewal is one attribute assignment, so the SIGTERM handler, which takes no lock, never sees a half-reset state. - **The run hands its collaborators in rather than building them inside** (`IterationRun(operator=..., unmapper=..., state=..., context=...)`), so tests build a run exactly as `execute` does. ### How to check that nothing changed - **Tests:** every test function of `test_iterableoperator.py` is kept under its name. Only call targets changed, through a `run_of()` helper that builds a run as `execute()` does. Two tests are new: `IterationState.kill` reaching every operator whatever one raises, and the unmapper's `do_xcom_push`. - **Coverage:** line and branch coverage of the module is unchanged: 17 lines and 4 branches missed, against 18 and 4 on the base. - **Lint:** ruff's opt-in design rules (`PLR0904, PLR0913, PLR0917, PLR6301`) flagged `_run_index`, `_run_indexed_task`, `_publish` and `axcom_push` before, and pass now. - **Performance:** CodSpeed simulation (instruction counts), base `9726e07e08` vs head, two rounds each, plus peak memory: | Benchmark | base `9726e07e08` | head | mean | |---|---|---|---| | sync items | 156.42M / 154.50M | 152.44M / 153.99M | −1.4% | | async items | 152.87M / 152.56M | 152.38M / 152.14M | −0.3% | | iterated upstream | 106.81M / 107.18M | 106.94M / 106.61M | −0.2% | | mapped upstream | 107.53M / 105.84M | 106.51M / 105.40M | −0.7% | | plain upstream | 189.08M / 187.85M | 188.34M / 186.83M | −0.5% | | control | 0.517M / 0.519M | 0.518M / 0.519M | 0 | | peak memory (plain upstream) | 1.02 / 0.86 MiB | 0.95 / 0.86 MiB | | Every case is within its round-to-round spread: no measurable cost in time or memory. Drafted-by: Claude Opus 5.5; reviewed by @dabla before posting -- 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]
