[SPARK-18403][SQL] Fix unsafe data false sharing issue in ObjectHashAggregateExec #15976

liancheng · 2016-11-22T06:40:21Z

What changes were proposed in this pull request?

This PR fixes a random OOM issue occurred while running ObjectHashAggregateSuite.

This issue can be steadily reproduced under the following conditions:

The aggregation must be evaluated using ObjectHashAggregateExec;
There must be an input column whose data type involves ArrayType (an input column of MapType may even cause SIGSEGV);
Sort-based aggregation fallback must be triggered during evaluation.

The root cause is that while falling back to sort-based aggregation, we must sort and feed already evaluated partial aggregation buffers living in the hash map to the sort-based aggregator using an external sorter. However, the underlying mutable byte buffer of UnsafeRows produced by the iterator of the external sorter is reused and may get overwritten when the iterator steps forward. After the last entry is consumed, the byte buffer points to a block of uninitialized memory filled by 5a. Therefore, while reading an UnsafeArrayData out of the UnsafeRow, 5a5a5a5a is treated as array size and triggers a memory allocation for a ridiculously large array and immediately blows up the JVM with an OOM.

To fix this issue, we only need to add .copy() accordingly.

How was this patch tested?

New regression test case added in ObjectHashAggregateSuite.

SparkQA · 2016-11-22T06:42:36Z

Test build #68982 has started for PR 15976 at commit 4b88eed.

liancheng · 2016-11-22T18:12:03Z

The last build failure was caused by a logical conflict with #15703. We don't really have any aggregate functions that don't support partial aggregation now after merging #15703, while the re-enabled test cases still check for that condition.

liancheng · 2016-11-22T18:22:51Z

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/ObjectHashAggregateSuite.scala

              }
+
+              doubleSafeCheckRows(actual1, expected, 1e-4)
+              doubleSafeCheckRows(actual2, expected, 1e-4)


All the changes made above in this file are used to resolve a logical conflict with PR #15703. We don't really have any aggregate functions that don't support partial aggregation now after merging #15703, must update the tests to reflect that.

liancheng · 2016-11-22T18:26:16Z

cc @yhuai @cloud-fan

dongjoon-hyun · 2016-11-22T18:45:58Z

Thank you for fixing this, @liancheng !

SparkQA · 2016-11-22T19:32:40Z

Test build #69016 has finished for PR 15976 at commit 6db5af9.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

liancheng · 2016-11-22T21:07:05Z

retest this please

liancheng · 2016-11-22T21:08:24Z

The last build failure was caused by irrelevant YARN tests.

SparkQA · 2016-11-22T23:53:12Z

Test build #69027 has finished for PR 15976 at commit 6db5af9.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

liancheng · 2016-11-22T23:58:00Z

Also cc @davies and @sameeragarwal.

liancheng · 2016-11-23T00:01:15Z

A similar alternative fix @yhuai proposed is to convert the underlying UnsafeRow into a safe row (i.e. GenericInternalRow in this case) using a projection instead of simply adding a .copy(). In this way, we prevent adding unsafe data into a safe row, which is in general safer. This approach may further affects performance, though.

cloud-fan · 2016-11-23T05:05:22Z

...core/src/main/scala/org/apache/spark/sql/execution/aggregate/ObjectAggregationIterator.scala

-            processRow(result.aggregationBuffer, inputIterator.getValue)
+            // Since `inputIterator.getValue` is an `UnsafeRow` whose underlying buffer will be
+            // overwritten when `inputIterator` steps forward, we need to do a deep copy here.
+            processRow(result.aggregationBuffer, inputIterator.getValue.copy())


So the problem is, during processRow we cache the input row somehow?

I think it's caused by MutableProjection? As MutableProjection may keep an "pointer" that points to a memory region of an unsafe row. Maybe we can fix this bug by #15082?

nvm, #15082 needs some significant refactor, we should get this fix in 2.1 first.

cloud-fan · 2016-11-23T15:43:21Z

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/ObjectHashAggregateSuite.scala

+    //  3. Sort-based aggregation fallback must be triggered during evaluation.
+    withSQLConf(
+      SQLConf.USE_OBJECT_HASH_AGG.key -> "true",
+      SQLConf.OBJECT_AGG_SORT_BASED_FALLBACK_THRESHOLD.key -> "1"


not related to this PR, but the config name looks weird, how about OBJECT_AGG_FALLBACK_TO_SORT_THRESHOLD

cloud-fan · 2016-11-28T06:14:05Z

retest this please

SparkQA · 2016-11-28T06:17:37Z

Test build #69224 has started for PR 15976 at commit 6db5af9.

dongjoon-hyun · 2016-11-28T08:47:33Z

Retest this please

SparkQA · 2016-11-28T11:09:30Z

Test build #69234 has finished for PR 15976 at commit 6db5af9.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2016-11-29T00:58:51Z

LGTM

cloud-fan · 2016-11-29T01:03:10Z

thanks, merging to master!

liancheng · 2016-11-29T19:18:44Z

@cloud-fan @dongjoon-hyun Thanks for the review!

…ggregateExec ## What changes were proposed in this pull request? This PR fixes a random OOM issue occurred while running `ObjectHashAggregateSuite`. This issue can be steadily reproduced under the following conditions: 1. The aggregation must be evaluated using `ObjectHashAggregateExec`; 2. There must be an input column whose data type involves `ArrayType` (an input column of `MapType` may even cause SIGSEGV); 3. Sort-based aggregation fallback must be triggered during evaluation. The root cause is that while falling back to sort-based aggregation, we must sort and feed already evaluated partial aggregation buffers living in the hash map to the sort-based aggregator using an external sorter. However, the underlying mutable byte buffer of `UnsafeRow`s produced by the iterator of the external sorter is reused and may get overwritten when the iterator steps forward. After the last entry is consumed, the byte buffer points to a block of uninitialized memory filled by `5a`. Therefore, while reading an `UnsafeArrayData` out of the `UnsafeRow`, `5a5a5a5a` is treated as array size and triggers a memory allocation for a ridiculously large array and immediately blows up the JVM with an OOM. To fix this issue, we only need to add `.copy()` accordingly. ## How was this patch tested? New regression test case added in `ObjectHashAggregateSuite`. Author: Cheng Lian <[email protected]> Closes apache#15976 from liancheng/investigate-oom.

Fix unsafe data false sharing issue in ObjectHashAggregateExec

4b88eed

Fix logical conflict with PR apache#15703

6db5af9

liancheng commented Nov 22, 2016

View reviewed changes

cloud-fan reviewed Nov 23, 2016

View reviewed changes

asfgit closed this in 2e80990 Nov 29, 2016

liancheng deleted the investigate-oom branch November 29, 2016 19:18

[SPARK-18403][SQL] Fix unsafe data false sharing issue in ObjectHashAggregateExec #15976

[SPARK-18403][SQL] Fix unsafe data false sharing issue in ObjectHashAggregateExec #15976

Uh oh!

Conversation

liancheng commented Nov 22, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

SparkQA commented Nov 22, 2016

Uh oh!

liancheng commented Nov 22, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

liancheng Nov 22, 2016

Choose a reason for hiding this comment

Uh oh!

liancheng commented Nov 22, 2016

Uh oh!

dongjoon-hyun commented Nov 22, 2016

Uh oh!

SparkQA commented Nov 22, 2016

Uh oh!

liancheng commented Nov 22, 2016

Uh oh!

liancheng commented Nov 22, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SparkQA commented Nov 22, 2016

Uh oh!

liancheng commented Nov 22, 2016

Uh oh!

liancheng commented Nov 23, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cloud-fan Nov 23, 2016

Choose a reason for hiding this comment

Uh oh!

cloud-fan Nov 23, 2016

Choose a reason for hiding this comment

Uh oh!

cloud-fan Nov 23, 2016

Choose a reason for hiding this comment

Uh oh!

cloud-fan Nov 23, 2016

Choose a reason for hiding this comment

Uh oh!

cloud-fan commented Nov 28, 2016

Uh oh!

SparkQA commented Nov 28, 2016

Uh oh!

dongjoon-hyun commented Nov 28, 2016

Uh oh!

SparkQA commented Nov 28, 2016

Uh oh!

cloud-fan commented Nov 29, 2016

Uh oh!

cloud-fan commented Nov 29, 2016

Uh oh!

liancheng commented Nov 29, 2016

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

liancheng commented Nov 22, 2016 •

edited

Loading

liancheng commented Nov 22, 2016 •

edited

Loading

liancheng commented Nov 22, 2016 •

edited

Loading

liancheng commented Nov 23, 2016 •

edited

Loading