Skip to content

[SPARK-23207][SPARK-22905][SPARK-24564][SPARK-25114][SQL][BACKPORT-2.1] Shuffle+Repartition on a DataFrame could lead to incorrect answers #22211

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
henryr wants to merge 2 commits into from
Closed

[SPARK-23207][SPARK-22905][SPARK-24564][SPARK-25114][SQL][BACKPORT-2.1] Shuffle+Repartition on a DataFrame could lead to incorrect answers #22211

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5bdd9e3
[SPARK-23207][SPARK-22905][SPARK-24564][SPARK-25114][SQL][BACKPORT-2.…
jiangxb1987 Aug 23, 2018
d269015
Remove dependency on java 8's Supplier entirely
henryr Aug 23, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Remove dependency on java 8's Supplier entirely
  • Loading branch information
@henryr
henryr committed Aug 23, 2018
commit d26901516f414d16f8eb38183053d22dc3b056e3
11 changes: 7 additions & 4 deletions sql/catalyst/src/main/java/org/apache/spark/sql/execution/UnsafeExternalRowSorter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@
package org.apache.spark.sql.execution;

import java.io.IOException;
import java.util.function.Supplier;

import scala.collection.Iterator;
import scala.math.Ordering;
Expand Down Expand Up @@ -52,6 +51,10 @@ public final class UnsafeExternalRowSorter {
private final PrefixComputer prefixComputer;
private final UnsafeExternalSorter sorter;

public static interface RecordComparatorSupplier {
public RecordComparator get();
}

public abstract static class PrefixComputer {

public static class Prefix {
Expand All @@ -71,7 +74,7 @@ public static class Prefix {

public static UnsafeExternalRowSorter createWithRecordComparator(
StructType schema,
Supplier<RecordComparator> recordComparatorSupplier,
RecordComparatorSupplier recordComparatorSupplier,
PrefixComparator prefixComparator,
PrefixComputer prefixComputer,
long pageSizeBytes,
Expand All @@ -87,7 +90,7 @@ public static UnsafeExternalRowSorter create(
PrefixComputer prefixComputer,
long pageSizeBytes,
boolean canUseRadixSort) throws IOException {
Supplier<RecordComparator> recordComparatorSupplier = new Supplier<RecordComparator>() {
RecordComparatorSupplier recordComparatorSupplier = new RecordComparatorSupplier() {
public RecordComparator get() { return new RowComparator(ordering, schema.length()); }
};
return new UnsafeExternalRowSorter(schema, recordComparatorSupplier, prefixComparator,
Expand All @@ -96,7 +99,7 @@ public static UnsafeExternalRowSorter create(

private UnsafeExternalRowSorter(
StructType schema,
Supplier<RecordComparator> recordComparatorSupplier,
RecordComparatorSupplier recordComparatorSupplier,
PrefixComparator prefixComparator,
PrefixComputer prefixComputer,
long pageSizeBytes,
Expand Down
3 changes: 1 addition & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchange.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@
package org.apache.spark.sql.execution.exchange

import java.util.Random
import java.util.function.Supplier

import org.apache.spark._
import org.apache.spark.rdd.RDD
Expand Down Expand Up @@ -257,7 +256,7 @@ object ShuffleExchange {
newPartitioning.numPartitions > 1 &&
newPartitioning.isInstanceOf[RoundRobinPartitioning]) {
rdd.mapPartitionsInternal { iter =>
val recordComparatorSupplier = new Supplier[RecordComparator] {
val recordComparatorSupplier = new UnsafeExternalRowSorter.RecordComparatorSupplier {
override def get: RecordComparator = new RecordBinaryComparator()
}
// The comparator for comparing row hashcode, which should always be Integer.
Expand Down