LuciferYang commented on code in PR #55891:
URL: https://github.com/apache/spark/pull/55891#discussion_r3260219046
##########
core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java:
##########
@@ -633,6 +614,56 @@ public UnsafeSorterIterator getSortedIterator() throws
IOException {
}
}
+ @VisibleForTesting
+ static final class BoundedMergerContext {
+ final List<UnsafeSorterSpillWriter> snapshot;
+ @Nullable final SpillableIterator inMemIter;
+ final UnsafeSorterBoundedSpillMerger merger;
+
+ BoundedMergerContext(
+ List<UnsafeSorterSpillWriter> snapshot,
+ @Nullable SpillableIterator inMemIter,
+ UnsafeSorterBoundedSpillMerger merger) {
+ this.snapshot = snapshot;
+ this.inMemIter = inMemIter;
+ this.merger = merger;
+ }
+ }
+
+ @VisibleForTesting
+ BoundedMergerContext prepareBoundedMerge() {
+ // Snapshot MUST precede readingIterator publication. Once readingIterator
is
+ // non-null, a sibling MemoryConsumer's spill request is routed via
+ // readingIterator.spill(), which appends a new writer to spillWriters AND
rebinds
+ // readingIterator.upstream to that same file. A post-publication snapshot
would
+ // then feed that file to BOTH the snapshot path and readingIterator --
duplicate
+ // records in the merged output.
+ final List<UnsafeSorterSpillWriter> snapshot = new
ArrayList<>(spillWriters);
Review Comment:
nit: ownership chain is clean after `1e6de55` — single snapshot point in
`prepareBoundedMerge()` (`UnsafeExternalSorter.java:641`), `BSM.merge()` no
longer mutates its input. One remaining tightening:
`BoundedMergerContext.snapshot` (`UnsafeExternalSorter.java:619`) is still
typed as a mutable `List`. Switching `prepareBoundedMerge()` to
`List.copyOf(spillWriters)` makes the snapshot unmodifiable at runtime — any
future code that mutates it (or aliases the live `spillWriters` field into the
context and adds to it) fails with `UnsupportedOperationException` rather than
relying on convention. Complements the new test, which asserts size-level
isolation of `ctx.snapshot` but not element-level immutability.
##########
core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java:
##########
@@ -633,6 +614,56 @@ public UnsafeSorterIterator getSortedIterator() throws
IOException {
}
}
+ @VisibleForTesting
+ static final class BoundedMergerContext {
+ final List<UnsafeSorterSpillWriter> snapshot;
+ @Nullable final SpillableIterator inMemIter;
+ final UnsafeSorterBoundedSpillMerger merger;
+
+ BoundedMergerContext(
+ List<UnsafeSorterSpillWriter> snapshot,
+ @Nullable SpillableIterator inMemIter,
+ UnsafeSorterBoundedSpillMerger merger) {
+ this.snapshot = snapshot;
+ this.inMemIter = inMemIter;
+ this.merger = merger;
+ }
+ }
+
+ @VisibleForTesting
+ BoundedMergerContext prepareBoundedMerge() {
+ // Snapshot MUST precede readingIterator publication. Once readingIterator
is
+ // non-null, a sibling MemoryConsumer's spill request is routed via
+ // readingIterator.spill(), which appends a new writer to spillWriters AND
rebinds
+ // readingIterator.upstream to that same file. A post-publication snapshot
would
+ // then feed that file to BOTH the snapshot path and readingIterator --
duplicate
+ // records in the merged output.
+ final List<UnsafeSorterSpillWriter> snapshot = new
ArrayList<>(spillWriters);
+
+ // This assignment is not inside synchronized(this), unlike the read in
Review Comment:
nit: the `synchronized(this)` rationale at
`UnsafeExternalSorter.java:643-649` only mentions `boundedMerger`, but
`prepareBoundedMerge()` also publishes `readingIterator` at its tail
(`UnsafeExternalSorter.java:661`). Both fields are volatile and rely on the
same task-thread-serialization argument. Rewrite to cover both, or hoist to the
method top.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
