Conversation
…its block resolver ## Summary ColumnarShuffleManager built its IndexShuffleBlockResolver with the single-arg constructor, so the resolver allocated its own taskIdMapsForShuffle while the manager kept a second, separate map. The resolver records blocks migrated in during executor decommissioning into its map, but unregisterShuffle reads the manager's map to delete map output. With two maps, migrated blocks were recorded where nothing read them, so their files were never deleted and leaked disk on decommissioned executors. Spark wires a single shared map in SortShuffleManager. taskIdMapsForShuffle is now declared before shuffleBlockResolver -- order matters, since Scala initializes vals in declaration order and the previous ordering would capture null -- and passed to the resolver. unregisterShuffle now iterates under mapTaskIds.synchronized, matching Spark, because the block-migration path mutates the same set under that lock. stop() now defers to super.stop(), which stops the resolver. Adds a ColumnarShuffleManagerSuite case that inserts an entry through the resolver's map and asserts unregisterShuffle clears it, which fails if the two maps are ever unshared again. ## Prompting Intent Follow-up to the ColumnarShuffleManager work merged in apache#13039, which extended SortShuffleManager to restore Spark's zero-copy row-based write path. That change landed only the first of two related parity fixes; the engineer asked to port the second one. Instructed to keep the change limited to what applies upstream. ## Rationale The resolver argument is passed positionally rather than by name. Spark 3.4 and 3.5 default both _blockManager and taskIdMapsForShuffle, so a named argument compiles there, but Spark 4.0 and 4.1 declare _blockManager with no default and a named-only call would not compile. The positional form satisfies every Spark version Gluten builds against (3.4, 3.5, 4.0, 4.1), so no shim is needed; ConcurrentHashMap conforms to both the java.util.Map parameter type used through 4.0 and the java.util.concurrent.ConcurrentMap type used in 4.1. Keeping the manager's own map and instead teaching unregisterShuffle to also drain the resolver's map was rejected: it would leave two sources of truth and diverge further from SortShuffleManager, which is the behaviour this class is meant to match. The test asserts the sharing through public behaviour rather than reference identity, so it still holds if the wiring is later expressed differently.
|
Run Gluten Clickhouse CI on x86 |
acvictor
marked this pull request as ready for review
September 21, 2026 13:12
Contributor
There was a problem hiding this comment.
Copilot encountered an error: Your billing is not configured or you have Copilot licenses from multiple standalone organizations or enterprises. To use premium requests, select a billing entity via the GitHub site, under Settings > Copilot > Features.
Contributor
There was a problem hiding this comment.
Copilot review overview
🟡 Changes recommended
The three-argument resolver constructor is unavailable in Spark 3.4/3.5 profiles, causing compilation failures.
Get a fresh assessment by requesting another Copilot review.
Review effort: Lite
Findings: 1
Open (1)
marin-ma
approved these changes
Sep 22, 2026
|
Run Gluten Clickhouse CI on x86 |
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.
What changes are proposed in this pull request?
ColumnarShuffleManager built its IndexShuffleBlockResolver with the single-arg constructor, so the resolver allocated its own taskIdMapsForShuffle while the manager kept a second, separate map. The resolver records blocks migrated in during executor decommissioning into its map, but unregisterShuffle reads the manager's map to delete map output. With two maps, migrated blocks were recorded where nothing read them, so their files were never deleted and leaked disk on decommissioned executors. Spark wires a single shared map in SortShuffleManager.
taskIdMapsForShuffle is now declared before shuffleBlockResolver -- order matters, since Scala initializes vals in declaration order and the previous ordering would capture null -- and passed to the resolver. unregisterShuffle now iterates under mapTaskIds.synchronized, matching Spark, because the block-migration path mutates the same set under that lock. stop() now defers to super.stop(), which stops the resolver.
Adds a ColumnarShuffleManagerSuite case that inserts an entry through the resolver's map and asserts unregisterShuffle clears it, which fails if the two maps are ever unshared again.
How was this patch tested?
UT
Was this patch authored or co-authored using generative AI tooling?
Co-authored by GitHub Copilot CLI 1.0.86