Repository navigation
Conversation
dabla
force-pushed
the
feature/xcom-sequence-chunked-reads
branch
from
September 27, 2026 12:41
1c508ce to
5302e88
Compare
dabla
force-pushed
the
feature/xcom-sequence-chunked-reads
branch
from
September 27, 2026 17:32
5302e88 to
89b6cc1
Compare
Contributor
Author
dabla
force-pushed
the
feature/xcom-sequence-chunked-reads
branch
17 times, most recently
from
October 2, 2026 07:52
e7cfaef to
50d3d03
Compare
dabla
force-pushed
the
feature/xcom-sequence-chunked-reads
branch
2 times, most recently
from
October 8, 2026 15:13
ba13cca to
ea231c5
Compare
dabla
force-pushed
the
feature/xcom-sequence-chunked-reads
branch
from
October 9, 2026 14:18
9df820e to
8115f75
Compare
The success and skip callbacks fired from IndexedTaskRunner.__exit__, as soon as the operator returned and before the indexed task's checkpoint and result were written. A checkpoint write that failed had already announced a success, and the retry ran the indexed task again and announced it a second time; the callback also found no end_date on the task instance. A plain task fires its callback after the push and the state report, with end_date set. The exit now only notes a failure. IterableOperator._run_task calls the runner's report_success() or report_skip() once the matching checkpoint is written, where the indexed task's code ran (its worker thread for a sync operator, the loop for an async one); both set end_date and the state before the callback. A result push that fails after the checkpoint is replayed from it, so the callback fires once. The docs page and the class docstring say so. test_the_success_callback_waits_for_the_checkpoint and the two runner report tests fail before.
dabla
force-pushed
the
feature/xcom-sequence-chunked-reads
branch
from
October 9, 2026 14:48
8115f75 to
739ffb6
Compare
…valuating the policy again With several failed indexed tasks whose retry policy decisions were all the default, or whose evaluation raised, no exception outweighed the others, the group was handed to the runner and no decision was kept for it. The callbacks then evaluated the policy once more, on the BaseExceptionGroup rather than on any indexed task's exception: K+1 evaluations instead of K, and for a policy that answers differently per call a retry announced here that the runner's own evaluation of the group could refuse. The once-per-item test did not reach this path because its policy answers retry(), so an exception was always chosen. _failure_for_the_runner now keeps the default decision for the group when no decision outweighs it, so _task_will_retry follows it and falls back to retry eligibility without evaluating the policy again. test_an_undecided_group_is_not_evaluated_again_for_the_callbacks fails before with four evaluations.
…the iteration on_kill() reaches the sub-operators that have started, and the stop flag keeps the executor from pulling more. An indexed task instance is neither until its runner starts: on a retry attempt it first reads its checkpoint, and a kill that lands during that read found it unregistered, so it started afterwards, ran to completion unkilled, its remote work included, and the drain waited for it before the task could conclude as terminated. IterableOperator._run_task now asks the iteration state whether a stop was requested once the checkpoint replay is done and before the runner is built, and returns IndexedTaskInstanceNotStarted as the outcome: no code run, no checkpoint, no callback, and the next attempt runs it. IndexedTaskOutcomes counts these apart from the indexed tasks that ran and the kill message names them. A sync indexed task is still handed to the pool afterwards, whose threads are as many as the calls in flight, so only that pickup remains. The docs page and the class docstring list it with the other outcomes that fire no callback. test_an_item_pulled_before_the_kill_but_not_started_does_not_run fails before with two of three indexed tasks run; the outcomes unit test covers the count.
dabla
force-pushed
the
feature/xcom-sequence-chunked-reads
branch
from
October 9, 2026 16:34
739ffb6 to
8bcec79
Compare
The static checks fail on D401 for IndexedTaskInstance.context_for: its summary line named what the method returns instead of saying what it does.
dabla
force-pushed
the
feature/xcom-sequence-chunked-reads
branch
from
October 9, 2026 16:51
8bcec79 to
584fc0e
Compare
The message claimed "the rest never pulled" also when every item had been pulled, and said the input was being resolved for a kill that landed before the run started, where nothing was. IndexedTaskOutcomes._killed_message now names the items that ran, those pulled before the kill and never started and those never pulled, each only when the count is not zero, and a kill before the input was resolved says so. test_a_kill_after_every_item_was_pulled_claims_no_remainder fails before.
dabla
force-pushed
the
feature/xcom-sequence-chunked-reads
branch
from
October 9, 2026 17:35
584fc0e to
aa79f39
Compare
…puts compare by it
…nt's execution_timeout
… would leave taken
… task instance's identity
…per request LazyXComSequence fetched one item per GetXComSequenceItem request: iterating a mapped upstream's results, or an iterated task reading them by index, cost one supervisor round trip per item, which is the dominant cost for large inputs. Reads now come in chunks. An index outside the slice held fetches chunk_size items starting there with one GetXComSequenceSlice request, the message the slice path already used, and consecutive reads are served from it. Sequential access costs one request per chunk instead of one per item, both through __getitem__/__iter__ (send) and aget/__aiter__ (asend), and at most one chunk is held. A jump backwards fetches again from there. The chunk size is the new [core] xcom_sequence_chunk_size option, default 32, documented in config.yml and in the iteration docs; a LazyXComSequence can also be given one explicitly. A negative index no longer needs its own per-item request: it is mapped from the end through the cached length, one GetXComCount at most. The client no longer sends GetXComSequenceItem at all; the request handler keeps serving it.
dabla
force-pushed
the
feature/xcom-sequence-chunked-reads
branch
from
October 10, 2026 06:25
aa79f39 to
15c4964
Compare
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Stacked on #62922 (Task Iteration): only the last commit, b0859ce, belongs to this PR. It will be rebased once #62922 merges.
related: #62922
What
LazyXComSequence, the value a mapped upstream'sXComArgresolves to, fetched one item perGetXComSequenceItemrequest. Iterating it, or an iterated task reading it by index (see #62922, where.iterate()resolves its input by index), cost one supervisor round trip to the API server per item, which is the dominant cost for large inputs.Reads now come in chunks:
chunk_sizeitems starting there with oneGetXComSequenceSlicerequest, the message the slice path already used, and the consecutive reads that follow are served from it. Sequential access costs one request per chunk instead of one per item, on both the sync path (__getitem__/__iter__throughsend) and the async one (aget/__aiter__throughasend), and at most one chunk is held in memory. A jump backwards fetches again from there.GetXComCountat most.GetXComSequenceItemat all. The request handler and the API endpoint keep serving it, so nothing else changes.The sync and async paths share the request builder and the response parser, so they cannot drift.
Why
For a 17,000-item mapped upstream consumed by
.iterate(), this is about 530 requests instead of 17,000. It is the "XCom slice optimisation" left out of #62922 on purpose, so that PR stays a pure simplification with unchanged behaviour.Chunk size
A new option,
[core] xcom_sequence_chunk_size, default 32, described inconfig.ymlnext tomax_map_lengthand mentioned in the iteration docs. It bounds memory at that many values of whatever size the XComs are, and larger values mean fewer round trips.LazyXComSequencereads it as its default and also accepts an explicitchunk_size.Tests
test_lazy_sequence.pynow asserts slice requests, with the chunk size taken from the config option and given explicitly: one per chunk while iterating, none for a read within the held chunk, a new request on a jump backwards, and negative indices through the cached count on both paths. The mapped-operator fake supervisor honours slice bounds, which it previously ignored.Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code following the guidelines
🤖 Generated with Claude Code