[SPARK-23376][SQL] creating UnsafeKVExternalSorter with BytesToBytesMap may fail #20561

cloud-fan · 2018-02-09T15:19:28Z

What changes were proposed in this pull request?

This is a long-standing bug in UnsafeKVExternalSorter and was reported in the dev list multiple times.

When creating UnsafeKVExternalSorter with BytesToBytesMap, we need to create a UnsafeInMemorySorter to sort the data in BytesToBytesMap. The data format of the sorter and the map is same, so no data movement is required. However, both the sorter and the map need a point array for some bookkeeping work.

There is an optimization in UnsafeKVExternalSorter: reuse the point array between the sorter and the map, to avoid an extra memory allocation. This sounds like a reasonable optimization, the length of the BytesToBytesMap point array is at least 4 times larger than the number of keys(to avoid hash collision, the hash table size should be at least 2 times larger than the number of keys, and each key occupies 2 slots). UnsafeInMemorySorter needs the pointer array size to be 4 times of the number of entries, so we are safe to reuse the point array.

However, the number of keys of the map doesn't equal to the number of entries in the map, because BytesToBytesMap supports duplicated keys. This breaks the assumption of the above optimization and we may run out of space when inserting data into the sorter, and hit error

java.lang.IllegalStateException: There is no space for new record
   at org.apache.spark.util.collection.unsafe.sort.UnsafeInMemorySorter.insertRecord(UnsafeInMemorySorter.java:239)
   at org.apache.spark.sql.execution.UnsafeKVExternalSorter.<init>(UnsafeKVExternalSorter.java:149)
...

This PR fixes this bug by creating a new point array if the existing one is not big enough.

How was this patch tested?

a new test

cloud-fan · 2018-02-09T15:19:59Z

cc @JoshRosen @davies @viirya @jiangxb1987

davies · 2018-02-09T17:54:16Z

sql/core/src/main/java/org/apache/spark/sql/execution/UnsafeKVExternalSorter.java

-      // another is the key prefix.
-      assert(map.numKeys() * 2 <= map.getArray().size() / 2);
+      // `BytesToBytesMap`'s point array is only guaranteed to hold all the distinct keys, but
+      // `UnsafeInMemorySorter`'s point array need to hold all the entries. Since `BytesToBytesMap`


It's possible to change UnsafeInMemorySorter to have multiple entries with same key.

yea, but it's not trivial, I'd like to do it later. The required change I can think of: BytesToBytesMap is actually a key -> list[value], and we need to provide a way to iterate key -> list[value] instead of key -> value.

davies · 2018-02-09T17:54:38Z

sql/core/src/main/java/org/apache/spark/sql/execution/UnsafeKVExternalSorter.java

+      // empty. Note: each record in the map takes two entries in the point array, one is record
+      // pointer, another is the key prefix.
+      if (map.numValues() > map.getArray().size() / 4) {
+        pointArray = map.allocateArray(map.numValues() * 4);


The allocation may fail.

Since overflow may occur (e.g. 0x70000000 * 4), should we use * 4L instead of * 4?

map.allocateArray will trigger other consumers to spill if memory is not enough. If the allocation still fails, there is nothing we can do, just let the execution fail.

davies · 2018-02-09T17:56:41Z

sql/core/src/test/scala/org/apache/spark/sql/execution/UnsafeKVExternalSorterSuite.scala

+    }
+
+    // Make sure we can successfully create a UnsafeKVExternalSorter with a `BytesToBytesMap`
+    // which has duplicated keys and the number of entries exceeds its capacity.


For aggregation, there are no multiple entries for same key, that only happen for hash join (Don't remember the details)

yes, we use BytesToBytesMap to build the broadcast join hash relation, which may have duplicated keys. I only create a new pointer array if the existing one is not big enough, so we won't have performance regression for aggregate.

SparkQA · 2018-02-09T18:29:28Z

Test build #87263 has finished for PR 20561 at commit 51d381f.

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

SparkQA · 2018-02-10T05:35:09Z

Test build #87277 has finished for PR 20561 at commit 8dab79a.

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

kiszk · 2018-02-10T07:07:59Z

retest this please

SparkQA · 2018-02-10T08:05:01Z

Test build #87279 has finished for PR 20561 at commit 8dab79a.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

kiszk · 2018-02-10T08:51:22Z

retest this please

SparkQA · 2018-02-10T11:57:13Z

Test build #87282 has finished for PR 20561 at commit 8dab79a.

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

davies · 2018-02-10T17:05:09Z

lgtm

viirya · 2018-02-11T05:42:31Z

sql/core/src/main/java/org/apache/spark/sql/execution/UnsafeKVExternalSorter.java

+        // to spill, if the memory is not enough.
+        pointArray = map.allocateArray(map.numValues() * 4L);
+      }
+
      // During spilling, the array in map will not be used, so we can borrow that and use it
      // as the underlying array for in-memory sorter (it's always large enough).


Shall we update the comment here too?

viirya

LGTM

SparkQA · 2018-02-11T06:11:40Z

Test build #87294 has finished for PR 20561 at commit 151a92d.

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

SparkQA · 2018-02-11T08:05:01Z

Test build #87300 has finished for PR 20561 at commit 2e7a5ad.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-02-11T08:05:01Z

Test build #87298 has finished for PR 20561 at commit 5e93313.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2018-02-11T08:12:20Z

retest this please.

SparkQA · 2018-02-11T10:35:31Z

Test build #87303 has finished for PR 20561 at commit 2e7a5ad.

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

viirya · 2018-02-11T12:00:51Z

retest this please.

SparkQA · 2018-02-11T15:06:32Z

Test build #87310 has finished for PR 20561 at commit 2e7a5ad.

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

…ap may fail ## What changes were proposed in this pull request? This is a long-standing bug in `UnsafeKVExternalSorter` and was reported in the dev list multiple times. When creating `UnsafeKVExternalSorter` with `BytesToBytesMap`, we need to create a `UnsafeInMemorySorter` to sort the data in `BytesToBytesMap`. The data format of the sorter and the map is same, so no data movement is required. However, both the sorter and the map need a point array for some bookkeeping work. There is an optimization in `UnsafeKVExternalSorter`: reuse the point array between the sorter and the map, to avoid an extra memory allocation. This sounds like a reasonable optimization, the length of the `BytesToBytesMap` point array is at least 4 times larger than the number of keys(to avoid hash collision, the hash table size should be at least 2 times larger than the number of keys, and each key occupies 2 slots). `UnsafeInMemorySorter` needs the pointer array size to be 4 times of the number of entries, so we are safe to reuse the point array. However, the number of keys of the map doesn't equal to the number of entries in the map, because `BytesToBytesMap` supports duplicated keys. This breaks the assumption of the above optimization and we may run out of space when inserting data into the sorter, and hit error ``` java.lang.IllegalStateException: There is no space for new record at org.apache.spark.util.collection.unsafe.sort.UnsafeInMemorySorter.insertRecord(UnsafeInMemorySorter.java:239) at org.apache.spark.sql.execution.UnsafeKVExternalSorter.<init>(UnsafeKVExternalSorter.java:149) ... ``` This PR fixes this bug by creating a new point array if the existing one is not big enough. ## How was this patch tested? a new test Author: Wenchen Fan <[email protected]> Closes #20561 from cloud-fan/bug. (cherry picked from commit 4bbd744) Signed-off-by: Wenchen Fan <[email protected]>

cloud-fan · 2018-02-11T16:15:26Z

thanks, merging to master/2.3/2.2!

…ap may fail This is a long-standing bug in `UnsafeKVExternalSorter` and was reported in the dev list multiple times. When creating `UnsafeKVExternalSorter` with `BytesToBytesMap`, we need to create a `UnsafeInMemorySorter` to sort the data in `BytesToBytesMap`. The data format of the sorter and the map is same, so no data movement is required. However, both the sorter and the map need a point array for some bookkeeping work. There is an optimization in `UnsafeKVExternalSorter`: reuse the point array between the sorter and the map, to avoid an extra memory allocation. This sounds like a reasonable optimization, the length of the `BytesToBytesMap` point array is at least 4 times larger than the number of keys(to avoid hash collision, the hash table size should be at least 2 times larger than the number of keys, and each key occupies 2 slots). `UnsafeInMemorySorter` needs the pointer array size to be 4 times of the number of entries, so we are safe to reuse the point array. However, the number of keys of the map doesn't equal to the number of entries in the map, because `BytesToBytesMap` supports duplicated keys. This breaks the assumption of the above optimization and we may run out of space when inserting data into the sorter, and hit error ``` java.lang.IllegalStateException: There is no space for new record at org.apache.spark.util.collection.unsafe.sort.UnsafeInMemorySorter.insertRecord(UnsafeInMemorySorter.java:239) at org.apache.spark.sql.execution.UnsafeKVExternalSorter.<init>(UnsafeKVExternalSorter.java:149) ... ``` This PR fixes this bug by creating a new point array if the existing one is not big enough. a new test Author: Wenchen Fan <[email protected]> Closes #20561 from cloud-fan/bug. (cherry picked from commit 4bbd744) Signed-off-by: Wenchen Fan <[email protected]>

…ap may fail ## What changes were proposed in this pull request? This is a long-standing bug in `UnsafeKVExternalSorter` and was reported in the dev list multiple times. When creating `UnsafeKVExternalSorter` with `BytesToBytesMap`, we need to create a `UnsafeInMemorySorter` to sort the data in `BytesToBytesMap`. The data format of the sorter and the map is same, so no data movement is required. However, both the sorter and the map need a point array for some bookkeeping work. There is an optimization in `UnsafeKVExternalSorter`: reuse the point array between the sorter and the map, to avoid an extra memory allocation. This sounds like a reasonable optimization, the length of the `BytesToBytesMap` point array is at least 4 times larger than the number of keys(to avoid hash collision, the hash table size should be at least 2 times larger than the number of keys, and each key occupies 2 slots). `UnsafeInMemorySorter` needs the pointer array size to be 4 times of the number of entries, so we are safe to reuse the point array. However, the number of keys of the map doesn't equal to the number of entries in the map, because `BytesToBytesMap` supports duplicated keys. This breaks the assumption of the above optimization and we may run out of space when inserting data into the sorter, and hit error ``` java.lang.IllegalStateException: There is no space for new record at org.apache.spark.util.collection.unsafe.sort.UnsafeInMemorySorter.insertRecord(UnsafeInMemorySorter.java:239) at org.apache.spark.sql.execution.UnsafeKVExternalSorter.<init>(UnsafeKVExternalSorter.java:149) ... ``` This PR fixes this bug by creating a new point array if the existing one is not big enough. ## How was this patch tested? a new test Author: Wenchen Fan <[email protected]> Closes apache#20561 from cloud-fan/bug.

…ap may fail This is a long-standing bug in `UnsafeKVExternalSorter` and was reported in the dev list multiple times. When creating `UnsafeKVExternalSorter` with `BytesToBytesMap`, we need to create a `UnsafeInMemorySorter` to sort the data in `BytesToBytesMap`. The data format of the sorter and the map is same, so no data movement is required. However, both the sorter and the map need a point array for some bookkeeping work. There is an optimization in `UnsafeKVExternalSorter`: reuse the point array between the sorter and the map, to avoid an extra memory allocation. This sounds like a reasonable optimization, the length of the `BytesToBytesMap` point array is at least 4 times larger than the number of keys(to avoid hash collision, the hash table size should be at least 2 times larger than the number of keys, and each key occupies 2 slots). `UnsafeInMemorySorter` needs the pointer array size to be 4 times of the number of entries, so we are safe to reuse the point array. However, the number of keys of the map doesn't equal to the number of entries in the map, because `BytesToBytesMap` supports duplicated keys. This breaks the assumption of the above optimization and we may run out of space when inserting data into the sorter, and hit error ``` java.lang.IllegalStateException: There is no space for new record at org.apache.spark.util.collection.unsafe.sort.UnsafeInMemorySorter.insertRecord(UnsafeInMemorySorter.java:239) at org.apache.spark.sql.execution.UnsafeKVExternalSorter.<init>(UnsafeKVExternalSorter.java:149) ... ``` This PR fixes this bug by creating a new point array if the existing one is not big enough. a new test Author: Wenchen Fan <[email protected]> Closes apache#20561 from cloud-fan/bug. (cherry picked from commit 4bbd744) Signed-off-by: Wenchen Fan <[email protected]>

creating UnsafeKVExternalSorter with BytesToBytesMap may fail

51d381f

davies reviewed Feb 9, 2018

View reviewed changes

address comments

8dab79a

more comments

151a92d

viirya reviewed Feb 11, 2018

View reviewed changes

viirya approved these changes Feb 11, 2018

View reviewed changes

address comment

2e7a5ad

cloud-fan force-pushed the bug branch from 5e93313 to 2e7a5ad Compare February 11, 2018 06:22

asfgit closed this in 4bbd744 Feb 11, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-23376][SQL] creating UnsafeKVExternalSorter with BytesToBytesMap may fail #20561

[SPARK-23376][SQL] creating UnsafeKVExternalSorter with BytesToBytesMap may fail #20561

cloud-fan commented Feb 9, 2018

cloud-fan commented Feb 9, 2018

davies Feb 9, 2018

cloud-fan Feb 10, 2018 •

edited

Loading

davies Feb 9, 2018

kiszk Feb 9, 2018 •

edited

Loading

cloud-fan Feb 10, 2018 •

edited

Loading

davies Feb 9, 2018

cloud-fan Feb 10, 2018

SparkQA commented Feb 9, 2018

SparkQA commented Feb 10, 2018

kiszk commented Feb 10, 2018

SparkQA commented Feb 10, 2018

kiszk commented Feb 10, 2018

SparkQA commented Feb 10, 2018

davies commented Feb 10, 2018

viirya Feb 11, 2018

viirya left a comment

SparkQA commented Feb 11, 2018

SparkQA commented Feb 11, 2018

SparkQA commented Feb 11, 2018

viirya commented Feb 11, 2018

SparkQA commented Feb 11, 2018

viirya commented Feb 11, 2018

SparkQA commented Feb 11, 2018

cloud-fan commented Feb 11, 2018

[SPARK-23376][SQL] creating UnsafeKVExternalSorter with BytesToBytesMap may fail #20561

[SPARK-23376][SQL] creating UnsafeKVExternalSorter with BytesToBytesMap may fail #20561

Conversation

cloud-fan commented Feb 9, 2018

What changes were proposed in this pull request?

How was this patch tested?

cloud-fan commented Feb 9, 2018

davies Feb 9, 2018

Choose a reason for hiding this comment

cloud-fan Feb 10, 2018 • edited Loading

Choose a reason for hiding this comment

davies Feb 9, 2018

Choose a reason for hiding this comment

kiszk Feb 9, 2018 • edited Loading

Choose a reason for hiding this comment

cloud-fan Feb 10, 2018 • edited Loading

Choose a reason for hiding this comment

davies Feb 9, 2018

Choose a reason for hiding this comment

cloud-fan Feb 10, 2018

Choose a reason for hiding this comment

SparkQA commented Feb 9, 2018

SparkQA commented Feb 10, 2018

kiszk commented Feb 10, 2018

SparkQA commented Feb 10, 2018

kiszk commented Feb 10, 2018

SparkQA commented Feb 10, 2018

davies commented Feb 10, 2018

viirya Feb 11, 2018

Choose a reason for hiding this comment

viirya left a comment

Choose a reason for hiding this comment

SparkQA commented Feb 11, 2018

SparkQA commented Feb 11, 2018

SparkQA commented Feb 11, 2018

viirya commented Feb 11, 2018

SparkQA commented Feb 11, 2018

viirya commented Feb 11, 2018

SparkQA commented Feb 11, 2018

cloud-fan commented Feb 11, 2018

cloud-fan Feb 10, 2018 •

edited

Loading

kiszk Feb 9, 2018 •

edited

Loading

cloud-fan Feb 10, 2018 •

edited

Loading