Repository navigation
fix(scheduler): restrict a leaf a task runs none of to an empty leaf - #2852
Draft
NoahKusaba wants to merge 3 commits into
Draft
NoahKusaba wants to merge 3 commits into
NoahKusaba wants to merge 3 commits into
Conversation
A task that owns none of a leaf's partitions, such as one running another branch of a UnionExec, never executes it, so the leaf can be replaced by a zero-partition EmptyExec whatever it is. Leaves the scheduler doesn't know, such as a DataSourceExec over a custom source, no longer fail such a task, and need no resolver for it, which the standalone scheduler can't take. The static planner's dry run now empties these branches too, so a union branch over an UnresolvedShuffleExec stays in the union's stage instead of getting one of its own.
…unts The leaf a task runs none of was an EmptyExec, which DataFusion's proto codec decodes with one partition, so the union on the executor ran the wrong branch. Use a zero-partition MemorySourceConfig, as the EmptyExec and PlaceholderRowExec case already did, through one helper. Also refuse a UnionExec branch that keeps another partition count after restriction, such as one that collapses its input through a SortExec under the adaptive planner, instead of running the wrong branch.
github-actions
Bot
added
the
documentation
Check for a leaf the task runs none of at the top of the walk, for every leaf, instead of after the resolver for unknown leaves and separately for EmptyExec and PlaceholderRowExec. A broadcast reader in a union branch another task owns is now emptied too, rather than kept whole. Also: document the union branch check on restrict_plan_to_partitions, word its error without a plural for one partition, and round-trip the serialization test through Ballista's codec.
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
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.
Which issue does this PR close?
Follows up #2679 / #2721. Found through andygrove's
UNION ALLrepro on #2638 (#2638 (comment)).Rationale for this change
When a stage is split between tasks, a task that owns one
UnionExecbranch restricts the others to no partitions ([]). Since #2721, a leaf the scheduler can't restrict fails the job unless the task runs all of it, so a union over a custom source such as Iceberg fails whenever its stage is split.A task never executes a leaf it runs none of, so any such leaf can be replaced by an empty one, whatever it is. That needs no resolver, so it also works in standalone mode.
What changes are included in this PR?
MemorySourceConfig: anEmptyExecwould decode on the executor with one partition. This replaces the separate[]case forEmptyExec/PlaceholderRowExec.UnionExecbranch that keeps another partition count after restriction is refused, rather than letting the union run the wrong branch. This catches a branch that collapses its input, such as anORDER BY'sSortExec.[]now succeeds for a branch over anUnresolvedShuffleExec, so the branch stays in the union's stage. One test that hardcoded stage numbers is updated.Are these changes tested?
Yes. New unit tests cover the empty leaf, a union branch the task doesn't own, a round trip through
BallistaPhysicalExtensionCodec(which fails with anEmptyExec), and the union branch check. Scheduler and client tests pass.End to end, standalone with
ballista.scheduler.max_partitions_per_task = 1, throughballista-iceberg(#2638) pointed at this branch:UNION ALLUNION ALL … ORDER BY idThe
ORDER BYcase is a separate adaptive-planner bug, Parquet included: the static planner gives a collapsing union branch its own stage (can_stay_inline, #2184), the adaptive planner doesn't. This PR makes it fail instead of returning wrong rows; I'll open an issue for the fix.Are there any user-facing changes?
UNION ALL … ORDER BYunder the adaptive planner now fails instead of returning duplicates.The 56.0.0 upgrade guide and
extending-components.mdare updated.