Skip to content

[SPARK-21319][SQL] Fix memory leak in UnsafeExternalRowSorter.RowComparator #18543

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
j-baker wants to merge 4 commits into from
Closed

[SPARK-21319][SQL] Fix memory leak in UnsafeExternalRowSorter.RowComparator #18543

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b65c85b
SPARK-21319: Fix memory leak in UnsafeExternalRowSorter.RowComparator
j-baker Jul 5, 2017
5e7a935
Symmetric fix for UnsafeKVExternalSorter
j-baker Jul 5, 2017
05fc618
Use a different mechanism to prevent memory leaks
j-baker Jul 19, 2017
51ec9a0
Couple of cleanups
j-baker Jul 19, 2017
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
5 changes: 4 additions & 1 deletion 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 @@ -211,7 +211,10 @@ public int compare(Object baseObj1, long baseOff1, Object baseObj2, long baseOff
// TODO: Why are the sizes -1?
row1.pointTo(baseObj1, baseOff1, -1);
row2.pointTo(baseObj2, baseOff2, -1);
return ordering.compare(row1, row2);
int comparison = ordering.compare(row1, row2);
row1.pointTo(null, 0L, -1);
row2.pointTo(null, 0L, -1);
Copy link
Contributor

@cloud-fan cloud-fan Jul 11, 2017

Choose a reason for hiding this comment

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

can we avoid to do it per comparison? is there any places we can do a cleanup at the end?

Copy link
Member

@srowen srowen Jul 17, 2017

Choose a reason for hiding this comment

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

Good idea @cloud-fan . Looks like RowComparator and KVComparator could be cleaned up in UnsafeExternalSorter.cleanupResources and UnsafeInMemorySorter.free. @davies I know this is from a long while back, but does that make sense? Seems like reasonable places to simply 'flush' the references, and won't hurt anything.

I am not 100% sure there's not another path to take care of, or if this frees the ref soon enough to avoid the problem. Thoughts @j-baker ?

Copy link
Member

@kiszk kiszk Jul 17, 2017 •
edited
Loading

Choose a reason for hiding this comment

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

@cloud-fan @srowen It is good idea to do this cleanup only once at the end. Now, I am curious how to implement this cleanup.

While @srowen proposed to use nsafeExternalSorter.cleanupResources and UnsafeInMemorySorter.free that will be called when a task is finished, to do cleanup here does not seem to work in this case. This is because this issue occurs before completing a task since UnsafeExternalSorter instance is registered into the task taskContext at here. This cleanup approach will not be performed before an OOM occurs during execution of the task.

IIUC, the end of sort is here. This line calls this sort method. Either to do the cleanup at the first part or to do the cleanup after checking type of a given comparator at the second part could work.

What do you think?

Copy link
Author

@j-baker j-baker Jul 18, 2017

Choose a reason for hiding this comment

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

So I feel like if the thing ends up in memory, this is correct - but otherwise the comparator is used in the UnsafeSorterSpillMerger.

Since we're handing back an iterator, am I right in thinking that without some periodic cleanup task you always stand a risk from this kind of leak unless you clear after each comparison or have some kind of async cleanup task?

Copy link
Author

@j-baker j-baker Jul 19, 2017 •
edited
Loading

Choose a reason for hiding this comment

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

I suppose that if you assume that you can only use these sorters once, then you can probably null out the reference to the comparator in the UnsafeExternalSorter once you've constructed the UnsafeSorterSpillMerger using it, and that would also solve the problem I've been seeing (the callback keeping a strong reference).

It'd still feel weird that you have a comparator that could potentially be responsible for the liveness of hundreds of megabytes of memory, though.

Copy link
Author

@j-baker j-baker Jul 19, 2017

Choose a reason for hiding this comment

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

maybe it's worth having a clone method on the comparator and making sure we clone the comparator before passing it to anything?

Copy link
Author

@j-baker j-baker Jul 19, 2017

Choose a reason for hiding this comment

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

...I'll update this PR.

Copy link
Contributor

@cloud-fan cloud-fan Jul 19, 2017

Choose a reason for hiding this comment

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

The merger reads data from spill files lazily, so when the merging finishes, it's end of the task.

Copy link
Author

@j-baker j-baker Jul 19, 2017

Choose a reason for hiding this comment

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

Then why would we be seeing the OOMs? If at the end of the task the taskcompletionlistener fires and is removed, then the whole comparator becomes unreachable and we have no problem here.

My job looks something like:

dataset.sortWithinPartitions().coalescePartitions() - would not we potentially finish doing some merging before reaching the end of the task?

Copy link
Contributor

@cloud-fan cloud-fan Jul 19, 2017 •
edited
Loading

Choose a reason for hiding this comment

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

i see, some partition may finish merging but some do not, and the merger which is finished is not referred, but it changed the comparator and make it keep the input rows it compared last time.

return comparison;
}
}
}
5 changes: 4 additions & 1 deletion sql/core/src/main/java/org/apache/spark/sql/execution/UnsafeKVExternalSorter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -242,7 +242,10 @@ public int compare(Object baseObj1, long baseOff1, Object baseObj2, long baseOff
// into the row.
row1.pointTo(baseObj1, baseOff1 + 4, -1);
row2.pointTo(baseObj2, baseOff2 + 4, -1);
return ordering.compare(row1, row2);
int comparison = ordering.compare(row1, row2);
row1.pointTo(null, 0L, -1);
row2.pointTo(null, 0L, -1);
return comparison;
}
}

Expand Down