Skip to content

[SPARK-8357] Fix unsafe memory leak on empty inputs in GeneratedAggregate #7560

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
JoshRosen wants to merge 13 commits into from
Closed

[SPARK-8357] Fix unsafe memory leak on empty inputs in GeneratedAggregate #7560

Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
1b07556
[SPARK-8357] [SQL] Memory leakage on unsafe aggregation path with emp…
navis Jun 14, 2015
d396589
added comments
navis Jun 14, 2015
15c5afc
added a test as suggested by JoshRosen
navis Jun 22, 2015
4d326b9
fixed test fails
navis Jun 22, 2015
51178e8
addressed comments
navis Jun 23, 2015
1a02a55
Rolled-back test-conf cleanup & fixed possible CCE & added more tests
navis Jun 23, 2015
735972f
used new conf apis
navis Jun 23, 2015
143e1ef
fixed format & added test for CCE case
navis Jun 26, 2015
c5419b3
addressed comments
navis Jun 30, 2015
adc8239
Back out Projection changes.
JoshRosen Jul 21, 2015
3c7db0f
Merge remote-tracking branch 'origin/master' into SPARK-8357
JoshRosen Jul 21, 2015
c649310
Revert SparkPlan change:
JoshRosen Jul 21, 2015
3486ce4
Some minor cleanup
JoshRosen Jul 21, 2015
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
Next Next commit
fixed test fails
  • Loading branch information
@navis
navis committed Jun 29, 2015
commit 4d326b908952e1c67eb0abc1c8a5e87375e5f519
9 changes: 5 additions & 4 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/GeneratedAggregate.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -270,10 +270,8 @@ case class GeneratedAggregate(

val joinedRow = new JoinedRow3

if (!iter.hasNext) {
// unsafe aggregation buffer is not released if input is empty (see SPARK-8357)
Iterator[InternalRow]()
} else if (groupingExpressions.isEmpty) {
if (groupingExpressions.isEmpty) {
// even with the empty input, value of empty buffer should be forwarded
// TODO: Codegening anything other than the updateProjection is probably over kill.
val buffer = newAggregationBuffer(EmptyRow).asInstanceOf[MutableRow]
var currentRow: InternalRow = null
Expand All @@ -286,7 +284,10 @@ case class GeneratedAggregate(

val resultProjection = resultProjectionBuilder()
Iterator(resultProjection(buffer))
} else if (!iter.hasNext) {
Iterator[InternalRow]()
} else if (unsafeEnabled && schemaSupportsUnsafe) {
// unsafe aggregation buffer is not released if input is empty (see SPARK-8357)
assert(iter.hasNext, "There should be at least one row for this path")
log.info("Using Unsafe-based aggregator")
val aggregationMap = new UnsafeFixedWidthAggregationMap(
Expand Down