Skip to content

[SPARK-28699][SQL] Disable using radix sort for ShuffleExchangeExec in repartition case #25491

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
xuanyuanking wants to merge 6 commits into from
Closed

[SPARK-28699][SQL] Disable using radix sort for ShuffleExchangeExec in repartition case #25491

Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
1256c87
[SPARK-28699][Core] Disable using radix sort for ShuffleExchangeExec …
xuanyuanking Aug 19, 2019
398a891
More comments
xuanyuanking Aug 19, 2019
ec57677
comment address
xuanyuanking Aug 20, 2019
8bf4cc8
fix comment
xuanyuanking Aug 20, 2019
c692265
Optimization by using radix sort if possible
xuanyuanking Aug 20, 2019
2e26335
Revert "Optimization by using radix sort if possible"
xuanyuanking Aug 21, 2019
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
8 changes: 6 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchangeExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -242,7 +242,7 @@ object ShuffleExchangeExec {
}
// The comparator for comparing row hashcode, which should always be Integer.
val prefixComparator = PrefixComparators.LONG
val canUseRadixSort = SQLConf.get.enableRadixSort

// The prefix computer generates row hashcode as the prefix, so we may decrease the
// probability that the prefixes are equal when input rows choose column values from a
// limited range.
Expand All @@ -264,7 +264,11 @@ object ShuffleExchangeExec {
prefixComparator,
prefixComputer,
pageSize,
canUseRadixSort)
// As we need to compare the binary of UnsafeRow here, we can't make sure whether all
// the fields can sort fully with prefix like SortExec. So we disable radix sort here
// to avoid getting unstable sort, and result to a correctness bug.
// See more details in SPARK-28699.
Copy link
Contributor

@cloud-fan cloud-fan Aug 20, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

how about

we are comparing binary here, which does not support radix sort. See more details in SPARK-28699.

Copy link
Member Author

@xuanyuanking xuanyuanking Aug 20, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Copy, shorten it in 8bf4cc8.

Copy link
Member

@dongjoon-hyun dongjoon-hyun Aug 20, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi, @cloud-fan . Is this the only one for this topic? Or, do we have more TODO items against old branches (branch-2.4 and branch-2.3)?

Copy link
Contributor

@cloud-fan cloud-fan Aug 20, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I have one more: #25498 , which will be merged soon, after tests pass.

The current PR may need more time. We need to collect feedback about whether we should disable "sort before repartition" by default. cc @jiangxb1987

false /* canUseRadixSort */)
Copy link
Contributor

@hvanhovell hvanhovell Aug 19, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why not use the same check as used in SortExec? That would be something like:

canUseRadixSort && outputAttributes.length == 1 && SortPrefixUtils.canSortFullyWithPrefix(SortOrder(outputAttributes.head, Ascending))

Copy link
Contributor

@cloud-fan cloud-fan Aug 19, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We are comparing binary here, so SortPrefixUtils.canSortFullyWithPrefix always return false.

Copy link
Contributor

@hvanhovell hvanhovell Aug 19, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah ok, I did not know that we use binary comparisons. Is this because of Map not being comparable? If it is, then that might be problematic in itself, because you expect the retried stage to return map elements in the same order right?

Copy link
Contributor

@hvanhovell hvanhovell Aug 19, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Anyway, back to the code, the change looks good to me. It might help to add a more insightful comment here.

Copy link
Contributor

@cloud-fan cloud-fan Aug 19, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this because of Map not being comparable

Yes, and it's also because we don't have to do an expensive normal sort.

The only problem we want to fix here is inputs with random order. Here we just want the inputs to have a stable order, but don't really care what the order is. So comparing via the unsafe row binary format is good enough here.

Copy link
Contributor

@tgravescs tgravescs Aug 19, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agree a better comment would be nice here, specifically reference the jira or say don't enable because...

Copy link
Member Author

@xuanyuanking xuanyuanking Aug 19, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, we compare the UnsafeRow binary here, and we the sort here must have a stable result. Otherwise, it will cause the correctness bug.
Thanks for the advice, add more comments done.

sorter.sort(iter.asInstanceOf[Iterator[UnsafeRow]])
}
} else {
Expand Down