ivoson commented on code in PR #55891:
URL: https://github.com/apache/spark/pull/55891#discussion_r3264294240
##########
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:
thx, udpated
##########
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:
thx, updated
--
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]
